Skip to content

Services Reference

Services are the orchestration layer between UI interactions, database access, workflow execution, and operational controls.

Package entrypoint

Purpose: service package exports.

init module.

Analytics

Purpose: analytics event collection and reporting helpers.

Operational analytics and KPI aggregation services.

AnalyticsService

Aggregates operational QA and production KPIs.

Source code in accessibility_mgr/services/analytics.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class AnalyticsService:
    """Aggregates operational QA and production KPIs."""

    def __init__(self) -> None:
        self._records: list[KPIRecord] = []

    def record_metric(
        self,
        *,
        metric_name: str,
        metric_value: float,
        category: str,
        metadata: dict[str, Any] | None = None,
    ) -> KPIRecord:
        record = KPIRecord(
            metric_name=metric_name,
            metric_value=metric_value,
            category=category,
            recorded_at=datetime.now(timezone.utc).isoformat(),
            metadata=metadata or {},
        )

        self._records.append(record)
        return record

    def summarize(self) -> dict[str, Any]:
        total = len(self._records)

        if not total:
            return {
                "total_metrics": 0,
                "average_score": 0,
                "categories": {},
            }

        avg = sum(r.metric_value for r in self._records) / total

        categories: dict[str, int] = {}

        for record in self._records:
            categories[record.category] = (
                categories.get(record.category, 0) + 1
            )

        return {
            "total_metrics": total,
            "average_score": round(avg, 2),
            "categories": categories,
        }

    def list_metrics(self) -> list[dict[str, Any]]:
        return [asdict(record) for record in self._records]

Artifact retention

Purpose: retention and cleanup policy handling for generated artifacts.

Artifact retention lifecycle management — SQLite-backed.

ArtifactRetentionService

SQLite-backed retention lifecycle management for generated artifacts.

Source code in accessibility_mgr/services/artifact_retention.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class ArtifactRetentionService:
    """SQLite-backed retention lifecycle management for generated artifacts."""

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(self.database_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _initialize(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS artifact_retention_record (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    artifact_path TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    retention_days INTEGER NOT NULL DEFAULT 30,
                    status TEXT NOT NULL DEFAULT 'active'
                )"""
            )

    def register_artifact(
        self,
        artifact_path: str,
        *,
        retention_days: int = 30,
    ) -> dict[str, Any]:
        created_at = datetime.now(UTC).isoformat()
        with self._connect() as conn:
            conn.execute(
                "INSERT INTO artifact_retention_record (artifact_path, created_at, retention_days, status) "
                "VALUES (?, ?, ?, 'active')",
                (artifact_path, created_at, retention_days),
            )
        return {
            "artifact_path": artifact_path,
            "created_at": created_at,
            "retention_days": retention_days,
            "status": "active",
        }

    def evaluate_retention(self) -> list[dict[str, Any]]:
        now = datetime.now(UTC)
        with self._connect() as conn:
            for row in conn.execute(
                "SELECT * FROM artifact_retention_record WHERE status = 'active'"
            ).fetchall():
                created = datetime.fromisoformat(row["created_at"])
                expiry = created + timedelta(days=row["retention_days"])
                if now > expiry:
                    conn.execute(
                        "UPDATE artifact_retention_record SET status = 'expired' WHERE id = ?",
                        (row["id"],),
                    )
            return [
                {
                    "artifact_path": r["artifact_path"],
                    "created_at": r["created_at"],
                    "retention_days": r["retention_days"],
                    "status": r["status"],
                }
                for r in conn.execute(
                    "SELECT * FROM artifact_retention_record"
                ).fetchall()
            ]

    def cleanup_expired(self) -> list[str]:
        removed: list[str] = []
        with self._connect() as conn:
            for row in conn.execute(
                "SELECT * FROM artifact_retention_record WHERE status = 'expired'"
            ).fetchall():
                path = Path(row["artifact_path"])
                if path.exists():
                    path.unlink()
                removed.append(row["artifact_path"])
                conn.execute(
                    "UPDATE artifact_retention_record SET status = 'deleted' WHERE id = ?",
                    (row["id"],),
                )
        return removed

Audit log

Purpose: audit stream normalization and retrieval.

Audit-grade operational event logging — SQLite-backed.

AuditLogService

SQLite-backed immutable-style audit event logging service.

Source code in accessibility_mgr/services/audit_log.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class AuditLogService:
    """SQLite-backed immutable-style audit event logging service."""

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(self.database_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _initialize(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS audit_event (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    event_type TEXT NOT NULL,
                    actor TEXT NOT NULL,
                    payload_json TEXT NOT NULL DEFAULT '{}',
                    event_hash TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )"""
            )

    def record_event(
        self,
        *,
        event_type: str,
        actor: str,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        created_at = datetime.now(UTC).isoformat()
        payload_json = json.dumps(payload, sort_keys=True)

        digest_payload = json.dumps(
            {"event_type": event_type, "actor": actor, "payload": payload, "created_at": created_at},
            sort_keys=True,
        )
        event_hash = hashlib.sha256(digest_payload.encode("utf-8")).hexdigest()

        with self._connect() as conn:
            conn.execute(
                "INSERT INTO audit_event (event_type, actor, payload_json, event_hash, created_at) "
                "VALUES (?, ?, ?, ?, ?)",
                (event_type, actor, payload_json, event_hash, created_at),
            )

        return {
            "event_type": event_type,
            "actor": actor,
            "payload": payload,
            "event_hash": event_hash,
            "created_at": created_at,
        }

    def list_events(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "event_type": r["event_type"],
                    "actor": r["actor"],
                    "payload": json.loads(r["payload_json"]),
                    "event_hash": r["event_hash"],
                    "created_at": r["created_at"],
                }
                for r in conn.execute("SELECT * FROM audit_event ORDER BY id").fetchall()
            ]

    def verify_integrity(self) -> bool:
        with self._connect() as conn:
            for event in conn.execute("SELECT * FROM audit_event ORDER BY id").fetchall():
                payload = json.loads(event["payload_json"])
                digest_payload = json.dumps(
                    {"event_type": event["event_type"], "actor": event["actor"],
                     "payload": payload, "created_at": event["created_at"]},
                    sort_keys=True,
                )
                calculated = hashlib.sha256(digest_payload.encode("utf-8")).hexdigest()
                if calculated != event["event_hash"]:
                    return False
            return True

Authentication

Purpose: user/session authentication workflows and credential validation.

Authentication and API token infrastructure.

GEN-009 / FUN-031: Tokens are currently stored in-memory only and are lost on restart. Revocation and token rotation are not persistent. A future migration should back this with the database (api_token table). All validation attempts are now logged for audit purposes.

AuthenticationService

Authentication and credential lifecycle service.

Source code in accessibility_mgr/services/authentication.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class AuthenticationService:
    """Authentication and credential lifecycle service."""

    def __init__(self) -> None:
        self._tokens: list[APIToken] = []
        self._sessions: list[AuthSession] = []

    def create_api_token(
        self,
        *,
        owner: str,
        expiration_hours: int = 24,
    ) -> dict:
        raw_token = secrets.token_urlsafe(32)
        token_hash = hashlib.sha256(
            raw_token.encode("utf-8")
        ).hexdigest()

        now = datetime.now(timezone.utc)

        token = APIToken(
            token_id=secrets.token_hex(8),
            token_hash=token_hash,
            owner=owner,
            created_at=now.isoformat(),
            expires_at=(
                now + timedelta(hours=expiration_hours)
            ).isoformat(),
            active=True,
        )

        self._tokens.append(token)

        return {
            "token": raw_token,
            "token_id": token.token_id,
            "expires_at": token.expires_at,
        }

    def register_api_token(
        self,
        *,
        owner: str,
        raw_token: str,
        expiration_hours: int = 24,
    ) -> dict:
        """Register a caller-provided raw API token for validation."""
        token_hash = hashlib.sha256(
            raw_token.encode("utf-8")
        ).hexdigest()
        now = datetime.now(timezone.utc)

        token = APIToken(
            token_id=secrets.token_hex(8),
            token_hash=token_hash,
            owner=owner,
            created_at=now.isoformat(),
            expires_at=(
                now + timedelta(hours=expiration_hours)
            ).isoformat(),
            active=True,
        )

        self._tokens.append(token)
        return {
            "token_id": token.token_id,
            "expires_at": token.expires_at,
        }

    def validate_token(self, raw_token: str, *, caller_ip: str = "unknown") -> bool:
        """Validate *raw_token* and log the attempt.

        FUN-031: every call is logged with timestamp and caller_ip so
        operators can detect brute-force attempts.  Token values are never
        logged — only token_ids.
        """
        hashed = hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
        now = datetime.now(timezone.utc)

        for token in self._tokens:
            if not token.active:
                continue
            if token.token_hash != hashed:
                continue
            if datetime.fromisoformat(token.expires_at) < now:
                log.warning(
                    "validate_token: expired token presented [id=%s owner=%s ip=%s]",
                    token.token_id, token.owner, caller_ip,
                )
                return False
            log.info(
                "validate_token: accepted [id=%s owner=%s ip=%s]",
                token.token_id, token.owner, caller_ip,
            )
            return True

        log.warning(
            "validate_token: rejected unknown/invalid token [ip=%s at=%s]",
            caller_ip, now.isoformat(),
        )
        return False

    def create_session(
        self,
        *,
        username: str,
        session_hours: int = 8,
    ) -> AuthSession:
        now = datetime.now(timezone.utc)

        session = AuthSession(
            username=username,
            session_id=secrets.token_hex(16),
            created_at=now.isoformat(),
            expires_at=(
                now + timedelta(hours=session_hours)
            ).isoformat(),
        )

        self._sessions.append(session)
        return session

    def list_tokens(self) -> list[dict]:
        return [
            {
                "token_id": token.token_id,
                "owner": token.owner,
                "expires_at": token.expires_at,
                "active": token.active,
            }
            for token in self._tokens
        ]

register_api_token(*, owner, raw_token, expiration_hours=24)

Register a caller-provided raw API token for validation.

Source code in accessibility_mgr/services/authentication.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def register_api_token(
    self,
    *,
    owner: str,
    raw_token: str,
    expiration_hours: int = 24,
) -> dict:
    """Register a caller-provided raw API token for validation."""
    token_hash = hashlib.sha256(
        raw_token.encode("utf-8")
    ).hexdigest()
    now = datetime.now(timezone.utc)

    token = APIToken(
        token_id=secrets.token_hex(8),
        token_hash=token_hash,
        owner=owner,
        created_at=now.isoformat(),
        expires_at=(
            now + timedelta(hours=expiration_hours)
        ).isoformat(),
        active=True,
    )

    self._tokens.append(token)
    return {
        "token_id": token.token_id,
        "expires_at": token.expires_at,
    }

validate_token(raw_token, *, caller_ip='unknown')

Validate raw_token and log the attempt.

FUN-031: every call is logged with timestamp and caller_ip so operators can detect brute-force attempts. Token values are never logged — only token_ids.

Source code in accessibility_mgr/services/authentication.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def validate_token(self, raw_token: str, *, caller_ip: str = "unknown") -> bool:
    """Validate *raw_token* and log the attempt.

    FUN-031: every call is logged with timestamp and caller_ip so
    operators can detect brute-force attempts.  Token values are never
    logged — only token_ids.
    """
    hashed = hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
    now = datetime.now(timezone.utc)

    for token in self._tokens:
        if not token.active:
            continue
        if token.token_hash != hashed:
            continue
        if datetime.fromisoformat(token.expires_at) < now:
            log.warning(
                "validate_token: expired token presented [id=%s owner=%s ip=%s]",
                token.token_id, token.owner, caller_ip,
            )
            return False
        log.info(
            "validate_token: accepted [id=%s owner=%s ip=%s]",
            token.token_id, token.owner, caller_ip,
        )
        return True

    log.warning(
        "validate_token: rejected unknown/invalid token [ip=%s at=%s]",
        caller_ip, now.isoformat(),
    )
    return False

Backup

Purpose: scheduled and on-demand database backups with retention.

Backup service — automated weekly SQLite database backups.

Performs a hot backup using SQLite's built-in VACUUM INTO (or sqlite3.connect().backup()) so the WAL is fully checkpointed and the copy is a clean, consistent database file.

Rotation keeps the most recent 10 backups in backups/ and purges older ones automatically.

Usage

Call BackupService.start() once at application startup. It schedules a background thread that fires immediately (to ensure at least one backup exists) and then every 7 days.

Manual backup

from accessibility_mgr.services.backup_service import BackupService
path = BackupService.run_backup(trigger="manual")

BackupService

Automated weekly SQLite database backups service.

Source code in accessibility_mgr/services/backup_service.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class BackupService:
    """Automated weekly SQLite database backups service."""

    # Populated lazily so the service can be imported before init_db() runs.
    _db_path: Path | None = None
    _backups_dir: Path | None = None

    _KEEP_BACKUPS = 10          # number of most-recent backups to retain
    _INTERVAL_SECONDS = 7 * 24 * 60 * 60   # one week

    _timer: threading.Timer | None = None
    _lock = threading.Lock()
    _run_lock = threading.Lock()
    _scheduler_enabled = False

    @staticmethod
    def _paths() -> tuple[Path, Path]:
        """Return (DB_PATH, BACKUPS_DIR) importing lazily to avoid circular imports."""
        if BackupService._db_path is None:
            from ..db.schema import BACKUPS_DIR, DB_PATH
            BackupService._db_path = DB_PATH
            BackupService._backups_dir = BACKUPS_DIR
        return BackupService._db_path, BackupService._backups_dir  # type: ignore[return-value]

    # ── Core backup logic ─────────────────────────────────────────────────────────

    @staticmethod
    def run_backup(trigger: str = "scheduled") -> str:
        """
        Copy the live database to ``backups/`` and return the backup file path.

        The copy is done via ``sqlite3.Connection.backup()``, which performs a
        WAL checkpoint and produces a consistent snapshot even while the app is
        running.  Old backups beyond the retention limit are pruned afterwards.

        Returns the absolute path of the new backup file as a string.
        Raises ``RuntimeError`` if the source database does not yet exist.
        """
        if not BackupService._run_lock.acquire(blocking=False):
            raise RuntimeError("Backup already in progress")

        try:
            db_path, backups_dir = BackupService._paths()
            backups_dir.mkdir(parents=True, exist_ok=True)

            if not db_path.exists():
                raise RuntimeError(f"Database not found at {db_path}; cannot back up.")

            # FUN-012: abort early if there is not enough free disk space
            db_size = db_path.stat().st_size
            free = shutil.disk_usage(backups_dir).free
            required = db_size * 2  # headroom for WAL journal during backup
            if free < required:
                raise RuntimeError(
                    f"Insufficient disk space for backup: {free:,} bytes free, "
                    f"{required:,} bytes needed (2× source size {db_size:,})."
                )

            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            dest = backups_dir / f"accessibility_manager_{timestamp}.db"

            # Use sqlite3.Connection.backup() — checkpoints WAL, works on a live DB.
            src_conn = sqlite3.connect(str(db_path))
            dst_conn = sqlite3.connect(str(dest))
            try:
                src_conn.backup(dst_conn)
            finally:
                dst_conn.close()
                src_conn.close()

            size = dest.stat().st_size
            log.info("Backup created: %s (%d bytes)", dest, size)

            # Record in DB (best-effort — don't crash if table not ready yet)
            try:
                from ..db.queries import log_backup
                log_backup(str(dest), size, trigger=trigger, status="ok")
            except Exception as exc:  # noqa: BLE001
                log.warning("Could not write backup_log entry: %s", exc)

            BackupService._rotate(backups_dir)
            return str(dest)
        finally:
            BackupService._run_lock.release()

    @staticmethod
    def _schedule_next_locked(delay_seconds: int) -> None:
        """Schedule the next timer while holding ``_lock``."""
        BackupService._timer = threading.Timer(delay_seconds, BackupService._scheduled_run)
        BackupService._timer.daemon = True
        BackupService._timer.name = "db-backup-timer"
        BackupService._timer.start()

    @staticmethod
    def _rotate(backups_dir: Path) -> None:
        """Delete oldest backup files beyond the retention limit."""
        backups = sorted(backups_dir.glob("accessibility_manager_*.db"))
        for old in backups[:-BackupService._KEEP_BACKUPS]:
            try:
                old.unlink()
                log.debug("Pruned old backup: %s", old)
            except OSError as exc:
                log.warning("Could not prune backup %s: %s", old, exc)

    # ── Scheduler ─────────────────────────────────────────────────────────────────

    @staticmethod
    def _scheduled_run() -> None:
        """Execute a backup then reschedule the next one."""
        try:
            path = BackupService.run_backup(trigger="scheduled")
            log.info("Scheduled backup completed: %s", path)
        except Exception as exc:  # noqa: BLE001
            log.error("Scheduled backup FAILED: %s", exc)
        finally:
            # Always reschedule so one failure doesn't stop all future backups.
            with BackupService._lock:
                if BackupService._scheduler_enabled:
                    BackupService._schedule_next_locked(BackupService._INTERVAL_SECONDS)
                else:
                    BackupService._timer = None

    @staticmethod
    def start() -> None:
        """
        Start the weekly backup scheduler.

        Safe to call multiple times — subsequent calls are no-ops.
        The first backup fires after a short delay (30 s) so startup is not
        blocked, and then repeats every 7 days.
        """
        with BackupService._lock:
            if BackupService._scheduler_enabled:
                return  # already running

            BackupService._scheduler_enabled = True
            # Small initial delay so init_db() has fully committed before first backup.
            BackupService._schedule_next_locked(30)
            log.info(
                "Database backup scheduler started — first backup in 30 s, "
                "then every 7 days.  Backups directory: %s",
                BackupService._paths()[1],
            )

    @staticmethod
    def stop() -> None:
        """Cancel the scheduled backup timer (called on app shutdown)."""
        with BackupService._lock:
            BackupService._scheduler_enabled = False
            if BackupService._timer is not None:
                BackupService._timer.cancel()
                BackupService._timer = None
                log.info("Database backup scheduler stopped.")

    # ── Restore ───────────────────────────────────────────────────────────────────

    @staticmethod
    def restore_backup(backup_path: str) -> None:
        """Restore the live database from *backup_path*.

        Performs a verified restore:
        1. Validates the backup file is a readable SQLite database.
        2. Creates a safety snapshot of the current live DB before overwriting.
        3. Copies the backup over the live DB path using sqlite3.backup().
        4. Verifies the restored file can be opened and queried.

        Raises ``RuntimeError`` on any validation or IO failure so the caller
        can surface the error to the user without crashing the app.
        """
        db_path, backups_dir = BackupService._paths()
        src = Path(backup_path)

        if not src.exists():
            raise RuntimeError(f"Backup file not found: {backup_path}")

        # Validate backup is an intact SQLite database
        try:
            test_conn = sqlite3.connect(str(src))
            test_conn.execute("PRAGMA integrity_check").fetchone()
            test_conn.close()
        except sqlite3.DatabaseError as exc:
            raise RuntimeError(
                f"Backup file '{backup_path}' is not a valid SQLite database: {exc}"
            ) from exc

        # Safety snapshot of the current live DB
        if db_path.exists():
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            safety = backups_dir / f"pre_restore_{timestamp}.db"
            backups_dir.mkdir(parents=True, exist_ok=True)
            try:
                live = sqlite3.connect(str(db_path))
                snap = sqlite3.connect(str(safety))
                live.backup(snap)
                snap.close()
                live.close()
                log.info("Pre-restore safety snapshot saved to %s", safety)
            except Exception as exc:  # noqa: BLE001
                log.warning("Could not create safety snapshot before restore: %s", exc)

        # Perform the restore
        try:
            src_conn = sqlite3.connect(str(src))
            dst_conn = sqlite3.connect(str(db_path))
            src_conn.backup(dst_conn)
            dst_conn.close()
            src_conn.close()
        except Exception as exc:
            raise RuntimeError(f"Restore failed while copying backup to live DB: {exc}") from exc

        # Verify restored DB is queryable
        try:
            verify = sqlite3.connect(str(db_path))
            verify.execute("PRAGMA integrity_check").fetchone()
            verify.close()
        except sqlite3.DatabaseError as exc:
            raise RuntimeError(
                f"Restored database failed integrity check: {exc}. "
                "The safety snapshot (if created) can be used to recover."
            ) from exc

        log.info("Database successfully restored from %s", backup_path)

    # ── Status helper (for Admin UI) ──────────────────────────────────────────────

    @staticmethod
    def status() -> dict[str, object]:
        """Return a summary dict for display in the Admin panel."""
        _, backups_dir = BackupService._paths()
        backups = sorted(backups_dir.glob("accessibility_manager_*.db"))
        last = backups[-1] if backups else None
        return {
            "backup_count": len(backups),
            "latest_backup": str(last) if last else "none",
            "latest_size_bytes": last.stat().st_size if last else 0,
            "backups_dir": str(backups_dir),
            "retention_limit": BackupService._KEEP_BACKUPS,
            "interval_days": BackupService._INTERVAL_SECONDS // 86400,
            "scheduler_active": BackupService._scheduler_enabled,
        }

restore_backup(backup_path) staticmethod

Restore the live database from backup_path.

Performs a verified restore: 1. Validates the backup file is a readable SQLite database. 2. Creates a safety snapshot of the current live DB before overwriting. 3. Copies the backup over the live DB path using sqlite3.backup(). 4. Verifies the restored file can be opened and queried.

Raises RuntimeError on any validation or IO failure so the caller can surface the error to the user without crashing the app.

Source code in accessibility_mgr/services/backup_service.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
@staticmethod
def restore_backup(backup_path: str) -> None:
    """Restore the live database from *backup_path*.

    Performs a verified restore:
    1. Validates the backup file is a readable SQLite database.
    2. Creates a safety snapshot of the current live DB before overwriting.
    3. Copies the backup over the live DB path using sqlite3.backup().
    4. Verifies the restored file can be opened and queried.

    Raises ``RuntimeError`` on any validation or IO failure so the caller
    can surface the error to the user without crashing the app.
    """
    db_path, backups_dir = BackupService._paths()
    src = Path(backup_path)

    if not src.exists():
        raise RuntimeError(f"Backup file not found: {backup_path}")

    # Validate backup is an intact SQLite database
    try:
        test_conn = sqlite3.connect(str(src))
        test_conn.execute("PRAGMA integrity_check").fetchone()
        test_conn.close()
    except sqlite3.DatabaseError as exc:
        raise RuntimeError(
            f"Backup file '{backup_path}' is not a valid SQLite database: {exc}"
        ) from exc

    # Safety snapshot of the current live DB
    if db_path.exists():
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        safety = backups_dir / f"pre_restore_{timestamp}.db"
        backups_dir.mkdir(parents=True, exist_ok=True)
        try:
            live = sqlite3.connect(str(db_path))
            snap = sqlite3.connect(str(safety))
            live.backup(snap)
            snap.close()
            live.close()
            log.info("Pre-restore safety snapshot saved to %s", safety)
        except Exception as exc:  # noqa: BLE001
            log.warning("Could not create safety snapshot before restore: %s", exc)

    # Perform the restore
    try:
        src_conn = sqlite3.connect(str(src))
        dst_conn = sqlite3.connect(str(db_path))
        src_conn.backup(dst_conn)
        dst_conn.close()
        src_conn.close()
    except Exception as exc:
        raise RuntimeError(f"Restore failed while copying backup to live DB: {exc}") from exc

    # Verify restored DB is queryable
    try:
        verify = sqlite3.connect(str(db_path))
        verify.execute("PRAGMA integrity_check").fetchone()
        verify.close()
    except sqlite3.DatabaseError as exc:
        raise RuntimeError(
            f"Restored database failed integrity check: {exc}. "
            "The safety snapshot (if created) can be used to recover."
        ) from exc

    log.info("Database successfully restored from %s", backup_path)

run_backup(trigger='scheduled') staticmethod

Copy the live database to backups/ and return the backup file path.

The copy is done via sqlite3.Connection.backup(), which performs a WAL checkpoint and produces a consistent snapshot even while the app is running. Old backups beyond the retention limit are pruned afterwards.

Returns the absolute path of the new backup file as a string. Raises RuntimeError if the source database does not yet exist.

Source code in accessibility_mgr/services/backup_service.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@staticmethod
def run_backup(trigger: str = "scheduled") -> str:
    """
    Copy the live database to ``backups/`` and return the backup file path.

    The copy is done via ``sqlite3.Connection.backup()``, which performs a
    WAL checkpoint and produces a consistent snapshot even while the app is
    running.  Old backups beyond the retention limit are pruned afterwards.

    Returns the absolute path of the new backup file as a string.
    Raises ``RuntimeError`` if the source database does not yet exist.
    """
    if not BackupService._run_lock.acquire(blocking=False):
        raise RuntimeError("Backup already in progress")

    try:
        db_path, backups_dir = BackupService._paths()
        backups_dir.mkdir(parents=True, exist_ok=True)

        if not db_path.exists():
            raise RuntimeError(f"Database not found at {db_path}; cannot back up.")

        # FUN-012: abort early if there is not enough free disk space
        db_size = db_path.stat().st_size
        free = shutil.disk_usage(backups_dir).free
        required = db_size * 2  # headroom for WAL journal during backup
        if free < required:
            raise RuntimeError(
                f"Insufficient disk space for backup: {free:,} bytes free, "
                f"{required:,} bytes needed (2× source size {db_size:,})."
            )

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        dest = backups_dir / f"accessibility_manager_{timestamp}.db"

        # Use sqlite3.Connection.backup() — checkpoints WAL, works on a live DB.
        src_conn = sqlite3.connect(str(db_path))
        dst_conn = sqlite3.connect(str(dest))
        try:
            src_conn.backup(dst_conn)
        finally:
            dst_conn.close()
            src_conn.close()

        size = dest.stat().st_size
        log.info("Backup created: %s (%d bytes)", dest, size)

        # Record in DB (best-effort — don't crash if table not ready yet)
        try:
            from ..db.queries import log_backup
            log_backup(str(dest), size, trigger=trigger, status="ok")
        except Exception as exc:  # noqa: BLE001
            log.warning("Could not write backup_log entry: %s", exc)

        BackupService._rotate(backups_dir)
        return str(dest)
    finally:
        BackupService._run_lock.release()

start() staticmethod

Start the weekly backup scheduler.

Safe to call multiple times — subsequent calls are no-ops. The first backup fires after a short delay (30 s) so startup is not blocked, and then repeats every 7 days.

Source code in accessibility_mgr/services/backup_service.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@staticmethod
def start() -> None:
    """
    Start the weekly backup scheduler.

    Safe to call multiple times — subsequent calls are no-ops.
    The first backup fires after a short delay (30 s) so startup is not
    blocked, and then repeats every 7 days.
    """
    with BackupService._lock:
        if BackupService._scheduler_enabled:
            return  # already running

        BackupService._scheduler_enabled = True
        # Small initial delay so init_db() has fully committed before first backup.
        BackupService._schedule_next_locked(30)
        log.info(
            "Database backup scheduler started — first backup in 30 s, "
            "then every 7 days.  Backups directory: %s",
            BackupService._paths()[1],
        )

status() staticmethod

Return a summary dict for display in the Admin panel.

Source code in accessibility_mgr/services/backup_service.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@staticmethod
def status() -> dict[str, object]:
    """Return a summary dict for display in the Admin panel."""
    _, backups_dir = BackupService._paths()
    backups = sorted(backups_dir.glob("accessibility_manager_*.db"))
    last = backups[-1] if backups else None
    return {
        "backup_count": len(backups),
        "latest_backup": str(last) if last else "none",
        "latest_size_bytes": last.stat().st_size if last else 0,
        "backups_dir": str(backups_dir),
        "retention_limit": BackupService._KEEP_BACKUPS,
        "interval_days": BackupService._INTERVAL_SECONDS // 86400,
        "scheduler_active": BackupService._scheduler_enabled,
    }

stop() staticmethod

Cancel the scheduled backup timer (called on app shutdown).

Source code in accessibility_mgr/services/backup_service.py 
179
180
181
182
183
184
185
186
187
@staticmethod
def stop() -> None:
    """Cancel the scheduled backup timer (called on app shutdown)."""
    with BackupService._lock:
        BackupService._scheduler_enabled = False
        if BackupService._timer is not None:
            BackupService._timer.cancel()
            BackupService._timer = None
            log.info("Database backup scheduler stopped.")

Compliance reporting

Purpose: compliance-oriented summaries and control evidence output.

Compliance reporting and signed provenance exports.

ComplianceReportingService

Governance and compliance export service.

AUDIT-FIX-005: exports are now signed with HMAC-SHA256 using a server-held key (ACCESSMAN_SIGNING_KEY) when one is configured, which proves both integrity and that the export was produced by a holder of that key. Previously this used a bare SHA-256 hash of the payload — that only detects accidental corruption; anyone can recompute the same hash for fabricated data, so it provided no real authenticity guarantee despite being called a "signature".

If no signing key is configured, exports fall back to a SHA-256 checksum and are labeled "SHA256-CHECKSUM-UNSIGNED" rather than silently claiming to be signed.

Source code in accessibility_mgr/services/compliance_reporting.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class ComplianceReportingService:
    """Governance and compliance export service.

    AUDIT-FIX-005: exports are now signed with HMAC-SHA256 using a
    server-held key (ACCESSMAN_SIGNING_KEY) when one is configured, which
    proves both integrity and that the export was produced by a holder of
    that key. Previously this used a bare SHA-256 hash of the payload —
    that only detects accidental corruption; anyone can recompute the same
    hash for fabricated data, so it provided no real authenticity
    guarantee despite being called a "signature".

    If no signing key is configured, exports fall back to a SHA-256
    checksum and are labeled "SHA256-CHECKSUM-UNSIGNED" rather than
    silently claiming to be signed.
    """

    def __init__(self) -> None:
        self.audit_log = AuditLogService()
        self.provenance = PersistentProvenanceRegistry()
        signing_key = os.getenv("ACCESSMAN_SIGNING_KEY", "").strip()
        self._signing_key: bytes | None = (
            signing_key.encode("utf-8") if signing_key else None
        )

    def generate_provenance_export(self) -> dict[str, Any]:
        payload = {
            "events": self.provenance.list_events(),
            "generated_at": datetime.now(timezone.utc).isoformat(),
        }

        signature, algorithm = self._sign_payload(payload)

        export = ComplianceExport(
            export_type="provenance",
            generated_at=datetime.now(timezone.utc).isoformat(),
            signature=signature,
            signature_algorithm=algorithm,
            payload=payload,
        )

        self.audit_log.record_event(
            event_type="compliance_export_generated",
            actor="system",
            payload={
                "export_type": "provenance",
                "signature": signature,
                "signature_algorithm": algorithm,
            },
        )

        return asdict(export)

    def generate_governance_report(self) -> dict[str, Any]:
        payload = {
            "audit_events": self.audit_log.list_events(),
            "generated_at": datetime.now(timezone.utc).isoformat(),
        }

        signature, algorithm = self._sign_payload(payload)

        export = ComplianceExport(
            export_type="governance",
            generated_at=datetime.now(timezone.utc).isoformat(),
            signature=signature,
            signature_algorithm=algorithm,
            payload=payload,
        )

        return asdict(export)

    def verify_signature(
        self,
        payload: dict[str, Any],
        signature: str,
        *,
        algorithm: str,
    ) -> bool:
        """Re-derive a signature for *payload* and compare it to *signature*.

        Only meaningful for algorithm == "HMAC-SHA256" — a checksum
        ("SHA256-CHECKSUM-UNSIGNED") can be reproduced by anyone and
        verifying it proves nothing about who generated the export.
        """
        expected_signature, expected_algorithm = self._sign_payload(payload)
        if algorithm != expected_algorithm:
            return False
        return hmac.compare_digest(signature, expected_signature)

    def _sign_payload(self, payload: dict[str, Any]) -> tuple[str, str]:
        serialized = json.dumps(payload, sort_keys=True).encode("utf-8")

        if self._signing_key:
            digest = hmac.new(
                self._signing_key, serialized, hashlib.sha256
            ).hexdigest()
            return digest, "HMAC-SHA256"

        # No ACCESSMAN_SIGNING_KEY configured: fall back to an
        # integrity-only checksum and label it honestly rather than
        # calling it a signature.
        digest = hashlib.sha256(serialized).hexdigest()
        return digest, "SHA256-CHECKSUM-UNSIGNED"

verify_signature(payload, signature, *, algorithm)

Re-derive a signature for payload and compare it to signature.

Only meaningful for algorithm == "HMAC-SHA256" — a checksum ("SHA256-CHECKSUM-UNSIGNED") can be reproduced by anyone and verifying it proves nothing about who generated the export.

Source code in accessibility_mgr/services/compliance_reporting.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def verify_signature(
    self,
    payload: dict[str, Any],
    signature: str,
    *,
    algorithm: str,
) -> bool:
    """Re-derive a signature for *payload* and compare it to *signature*.

    Only meaningful for algorithm == "HMAC-SHA256" — a checksum
    ("SHA256-CHECKSUM-UNSIGNED") can be reproduced by anyone and
    verifying it proves nothing about who generated the export.
    """
    expected_signature, expected_algorithm = self._sign_payload(payload)
    if algorithm != expected_algorithm:
        return False
    return hmac.compare_digest(signature, expected_signature)

Distributed workers

Purpose: worker distribution primitives for asynchronous operations.

Distributed worker coordination primitives — SQLite-backed.

DistributedWorkerRegistry

SQLite-backed registry for distributed orchestration workers.

Source code in accessibility_mgr/services/distributed_workers.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class DistributedWorkerRegistry:
    """SQLite-backed registry for distributed orchestration workers."""

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(self.database_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _initialize(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS distributed_worker_node (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    node_id TEXT NOT NULL UNIQUE,
                    hostname TEXT NOT NULL,
                    status TEXT NOT NULL DEFAULT 'online',
                    registered_at TEXT NOT NULL
                )"""
            )

    def register_node(
        self,
        *,
        node_id: str,
        hostname: str,
    ) -> dict[str, Any]:
        registered_at = datetime.now(UTC).isoformat()
        with self._connect() as conn:
            conn.execute(
                "INSERT OR REPLACE INTO distributed_worker_node (node_id, hostname, status, registered_at) "
                "VALUES (?, ?, 'online', ?)",
                (node_id, hostname, registered_at),
            )
        return {
            "node_id": node_id,
            "hostname": hostname,
            "status": "online",
            "registered_at": registered_at,
        }

    def set_status(self, node_id: str, status: str) -> None:
        with self._connect() as conn:
            conn.execute(
                "UPDATE distributed_worker_node SET status = ? WHERE node_id = ?",
                (status, node_id),
            )

    def list_nodes(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "node_id": r["node_id"],
                    "hostname": r["hostname"],
                    "status": r["status"],
                    "registered_at": r["registered_at"],
                }
                for r in conn.execute("SELECT * FROM distributed_worker_node").fetchall()
            ]

EPUB QA

Purpose: EPUB-specific QA checks and result handling.

EPUB QA automation services.

Provides orchestration primitives for automated EPUB accessibility quality assurance workflows and pipeline execution tracking.

EPUBQAService

Accessibility QA orchestration service.

Source code in accessibility_mgr/services/epub_qa.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
class EPUBQAService:
    """Accessibility QA orchestration service."""

    def __init__(
        self,
        binary_service: AccessibilityBinaryIntegrationService | None = None,
    ) -> None:
        self._runs: list[PipelineRun] = []
        self._reports: dict[int, QAResult] = {}
        self.binary_service = binary_service or AccessibilityBinaryIntegrationService()

    def start_pipeline(
        self,
        *,
        pipeline_name: str,
        asset_id: int,
    ) -> PipelineRun:
        run = PipelineRun(
            pipeline_name=pipeline_name,
            asset_id=asset_id,
            status="running",
            started_at=datetime.now(timezone.utc).isoformat(),
        )

        self._runs.append(run)
        return run

    def append_log(
        self,
        run: PipelineRun,
        message: str,
    ) -> None:
        timestamp = datetime.now(timezone.utc).isoformat()
        run.logs.append(f"{timestamp} {message}")

    def complete_pipeline(
        self,
        run: PipelineRun,
        *,
        success: bool,
    ) -> None:
        run.status = "completed" if success else "failed"
        run.completed_at = datetime.now(timezone.utc).isoformat()

    def retry_pipeline(self, run: PipelineRun) -> None:
        run.retry_count += 1
        run.status = "retrying"
        self.append_log(run, "Pipeline retry requested")

    def run_ace_check(
        self,
        *,
        asset_id: int,
        epub_path: str,
    ) -> QAResult:
        """Run a real DAISY Ace accessibility audit against *epub_path*.

        AUDIT-FIX-002: this previously fabricated a score from the file
        extension and the word "draft" in the filename and never invoked
        Ace at all. It now calls AccessibilityBinaryIntegrationService,
        which runs the real `ace` CLI, and parses the JSON report Ace
        writes to its output directory. If Ace is not installed, or the
        report can't be parsed, that is reported honestly — the result is
        never silently marked as passed.
        """

        path = Path(epub_path)
        issues: list[QAIssue] = []

        if path.suffix.lower() != ".epub":
            issues.append(
                QAIssue(
                    severity="error",
                    code="INVALID_FORMAT",
                    message="Input file is not an EPUB package",
                )
            )
            result = QAResult(
                passed=False,
                score=0,
                engine="DAISY Ace",
                checked_at=datetime.now(timezone.utc).isoformat(),
                issues=issues,
            )
            self._reports[asset_id] = result
            return result

        if not path.exists():
            issues.append(
                QAIssue(
                    severity="error",
                    code="FILE_NOT_FOUND",
                    message=f"EPUB file not found: {epub_path}",
                )
            )
            result = QAResult(
                passed=False,
                score=0,
                engine="DAISY Ace",
                checked_at=datetime.now(timezone.utc).isoformat(),
                issues=issues,
            )
            self._reports[asset_id] = result
            return result

        outcome = self.binary_service.run_daisy_ace(epub_path)

        if outcome.get("status") == "unavailable":
            issues.append(
                QAIssue(
                    severity="error",
                    code="TOOL_UNAVAILABLE",
                    message=outcome.get(
                        "reason", "DAISY Ace CLI is not installed"
                    ),
                )
            )
            result = QAResult(
                passed=False,
                score=0,
                engine="DAISY Ace (unavailable)",
                checked_at=datetime.now(timezone.utc).isoformat(),
                issues=issues,
            )
            self._reports[asset_id] = result
            return result

        execution = outcome.get("execution", {})
        exit_code = execution.get("exit_code", 1)
        output_dir = outcome.get("output_directory")
        report_issues, report_parsed = self._parse_ace_report(output_dir)
        issues.extend(report_issues)

        if exit_code != 0 and not report_parsed:
            # Ace ran and reported failure, and we have no structured
            # report to explain why — surface the raw stderr instead of
            # guessing.
            stderr = (execution.get("stderr") or "").strip()
            issues.append(
                QAIssue(
                    severity="error",
                    code="ACE_EXECUTION_FAILED",
                    message=stderr[:500] if stderr else (
                        f"DAISY Ace exited with code {exit_code}"
                    ),
                )
            )

        error_count = sum(1 for i in issues if i.severity == "error")
        warning_count = sum(1 for i in issues if i.severity == "warning")
        score = max(0, 100 - (error_count * 15) - (warning_count * 5))

        result = QAResult(
            passed=exit_code == 0 and error_count == 0,
            score=score,
            engine="DAISY Ace",
            checked_at=datetime.now(timezone.utc).isoformat(),
            issues=issues,
        )

        self._reports[asset_id] = result
        return result

    @staticmethod
    def _parse_ace_report(
        output_dir: str | None,
    ) -> tuple[list[QAIssue], bool]:
        """Best-effort parse of Ace's report.json.

        Ace's report schema has changed across versions, so this looks for
        the most common shapes (an 'assertions'/'violations' list with
        'severity' or 'earl:result' entries) rather than assuming one exact
        structure. Returns (issues, parsed_successfully).
        """
        if not output_dir:
            return [], False

        report_path = Path(output_dir) / "report.json"
        if not report_path.exists():
            return [], False

        try:
            data = json.loads(report_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return [], False

        raw_entries: list[dict[str, Any]] = []
        for key in ("assertions", "violations", "issues"):
            value = data.get(key) if isinstance(data, dict) else None
            if isinstance(value, list):
                raw_entries = value
                break

        issues: list[QAIssue] = []
        for entry in raw_entries:
            if not isinstance(entry, dict):
                continue

            earl_result = entry.get("earl:result")
            if entry.get("severity"):
                severity = str(entry["severity"])
            elif isinstance(earl_result, dict) and earl_result.get("earl:outcome"):
                severity = str(earl_result["earl:outcome"])
            else:
                severity = "warning"
            severity = severity.lower()
            if severity not in {"error", "warning", "info"}:
                severity = "warning"

            issues.append(
                QAIssue(
                    severity=severity,
                    code=str(entry.get("code") or entry.get("rule") or "ACE_RULE"),
                    message=str(
                        entry.get("message")
                        or entry.get("description")
                        or "Accessibility rule violation reported by Ace"
                    ),
                    location=str(entry.get("location"))
                    if entry.get("location")
                    else None,
                )
            )

        return issues, True

    def get_report(self, asset_id: int) -> QAResult | None:
        return self._reports.get(asset_id)

    def list_pipeline_runs(self) -> list[dict[str, Any]]:
        return [
            {
                "pipeline_name": run.pipeline_name,
                "asset_id": run.asset_id,
                "status": run.status,
                "started_at": run.started_at,
                "completed_at": run.completed_at,
                "retry_count": run.retry_count,
                "logs": run.logs,
            }
            for run in self._runs
        ]

run_ace_check(*, asset_id, epub_path)

Run a real DAISY Ace accessibility audit against epub_path.

AUDIT-FIX-002: this previously fabricated a score from the file extension and the word "draft" in the filename and never invoked Ace at all. It now calls AccessibilityBinaryIntegrationService, which runs the real ace CLI, and parses the JSON report Ace writes to its output directory. If Ace is not installed, or the report can't be parsed, that is reported honestly — the result is never silently marked as passed.

Source code in accessibility_mgr/services/epub_qa.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def run_ace_check(
    self,
    *,
    asset_id: int,
    epub_path: str,
) -> QAResult:
    """Run a real DAISY Ace accessibility audit against *epub_path*.

    AUDIT-FIX-002: this previously fabricated a score from the file
    extension and the word "draft" in the filename and never invoked
    Ace at all. It now calls AccessibilityBinaryIntegrationService,
    which runs the real `ace` CLI, and parses the JSON report Ace
    writes to its output directory. If Ace is not installed, or the
    report can't be parsed, that is reported honestly — the result is
    never silently marked as passed.
    """

    path = Path(epub_path)
    issues: list[QAIssue] = []

    if path.suffix.lower() != ".epub":
        issues.append(
            QAIssue(
                severity="error",
                code="INVALID_FORMAT",
                message="Input file is not an EPUB package",
            )
        )
        result = QAResult(
            passed=False,
            score=0,
            engine="DAISY Ace",
            checked_at=datetime.now(timezone.utc).isoformat(),
            issues=issues,
        )
        self._reports[asset_id] = result
        return result

    if not path.exists():
        issues.append(
            QAIssue(
                severity="error",
                code="FILE_NOT_FOUND",
                message=f"EPUB file not found: {epub_path}",
            )
        )
        result = QAResult(
            passed=False,
            score=0,
            engine="DAISY Ace",
            checked_at=datetime.now(timezone.utc).isoformat(),
            issues=issues,
        )
        self._reports[asset_id] = result
        return result

    outcome = self.binary_service.run_daisy_ace(epub_path)

    if outcome.get("status") == "unavailable":
        issues.append(
            QAIssue(
                severity="error",
                code="TOOL_UNAVAILABLE",
                message=outcome.get(
                    "reason", "DAISY Ace CLI is not installed"
                ),
            )
        )
        result = QAResult(
            passed=False,
            score=0,
            engine="DAISY Ace (unavailable)",
            checked_at=datetime.now(timezone.utc).isoformat(),
            issues=issues,
        )
        self._reports[asset_id] = result
        return result

    execution = outcome.get("execution", {})
    exit_code = execution.get("exit_code", 1)
    output_dir = outcome.get("output_directory")
    report_issues, report_parsed = self._parse_ace_report(output_dir)
    issues.extend(report_issues)

    if exit_code != 0 and not report_parsed:
        # Ace ran and reported failure, and we have no structured
        # report to explain why — surface the raw stderr instead of
        # guessing.
        stderr = (execution.get("stderr") or "").strip()
        issues.append(
            QAIssue(
                severity="error",
                code="ACE_EXECUTION_FAILED",
                message=stderr[:500] if stderr else (
                    f"DAISY Ace exited with code {exit_code}"
                ),
            )
        )

    error_count = sum(1 for i in issues if i.severity == "error")
    warning_count = sum(1 for i in issues if i.severity == "warning")
    score = max(0, 100 - (error_count * 15) - (warning_count * 5))

    result = QAResult(
        passed=exit_code == 0 and error_count == 0,
        score=score,
        engine="DAISY Ace",
        checked_at=datetime.now(timezone.utc).isoformat(),
        issues=issues,
    )

    self._reports[asset_id] = result
    return result

Event stream

Purpose: event bus style stream publication and consumption.

Event streaming and webhook infrastructure — SQLite-backed.

EventStreamService

SQLite-backed internal event stream.

Future targets: - webhook delivery - Kafka/NATS adapters - distributed event streaming - audit event propagation

Source code in accessibility_mgr/services/event_stream.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class EventStreamService:
    """SQLite-backed internal event stream.

    Future targets:
    - webhook delivery
    - Kafka/NATS adapters
    - distributed event streaming
    - audit event propagation
    """

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(self.database_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _initialize(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS event_subscription (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    event_type TEXT NOT NULL,
                    callback_url TEXT NOT NULL,
                    active INTEGER NOT NULL DEFAULT 1
                )"""
            )
            conn.execute(
                """CREATE TABLE IF NOT EXISTS platform_event (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    event_type TEXT NOT NULL,
                    payload_json TEXT NOT NULL DEFAULT '{}',
                    created_at TEXT NOT NULL
                )"""
            )

    def subscribe(
        self,
        *,
        event_type: str,
        callback_url: str,
    ) -> None:
        with self._connect() as conn:
            conn.execute(
                "INSERT INTO event_subscription (event_type, callback_url, active) VALUES (?, ?, 1)",
                (event_type, callback_url),
            )

    def publish(
        self,
        *,
        event_type: str,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        created_at = datetime.now(UTC).isoformat()
        payload_json = json.dumps(payload, sort_keys=True)
        with self._connect() as conn:
            conn.execute(
                "INSERT INTO platform_event (event_type, payload_json, created_at) VALUES (?, ?, ?)",
                (event_type, payload_json, created_at),
            )
        return {
            "event_type": event_type,
            "payload": payload,
            "created_at": created_at,
        }

    def list_events(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "event_type": r["event_type"],
                    "payload": json.loads(r["payload_json"]),
                    "created_at": r["created_at"],
                }
                for r in conn.execute(
                    "SELECT * FROM platform_event ORDER BY id"
                ).fetchall()
            ]

    def list_subscriptions(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "event_type": r["event_type"],
                    "callback_url": r["callback_url"],
                    "active": bool(r["active"]),
                }
                for r in conn.execute("SELECT * FROM event_subscription").fetchall()
            ]

Execution service

Purpose: controlled command/process execution used by pipeline tasks.

Execution service — runs external tool commands via subprocess.

Used by QA tooling and pipeline orchestration.

ExecutionResult dataclass

Result of a subprocess command execution.

Source code in accessibility_mgr/services/execution_service.py 
31
32
33
34
35
36
37
@dataclass
class ExecutionResult:
    """Result of a subprocess command execution."""
    command: str
    success: bool
    output: str
    return_code: int

ExecutionService

Controlled subprocess execution with allowlist and timeout enforcement.

Source code in accessibility_mgr/services/execution_service.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class ExecutionService:
    """Controlled subprocess execution with allowlist and timeout enforcement."""
    DEFAULT_TIMEOUT: int = 120  # seconds

    @staticmethod
    def run_command(
        command: list[str],
        timeout: int = 120,
        cwd: Optional[str] = None,
    ) -> ExecutionResult:
        """Run a shell command and return the full result."""
        cmd_str = " ".join(command)
        exe_name = Path(command[0]).name
        if exe_name not in ALLOWED_EXECUTABLES:
            return ExecutionResult(
                command=cmd_str,
                success=False,
                output=(
                    f"Executable '{exe_name}' is not in the permitted allowlist. "
                    "Add it to ALLOWED_EXECUTABLES in execution_service.py if required."
                ),
                return_code=-4,
            )
        try:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=timeout,
                check=False,
                cwd=cwd,
            )
            output = result.stdout or ""
            if result.stderr:
                output = output + ("\n" if output else "") + result.stderr
            return ExecutionResult(
                command=cmd_str,
                success=result.returncode == 0,
                output=output.strip(),
                return_code=result.returncode,
            )
        except FileNotFoundError:
            return ExecutionResult(
                command=cmd_str,
                success=False,
                output=(
                    f"Command not found: '{command[0]}'. "
                    "Is the tool installed and on PATH?"
                ),
                return_code=-1,
            )
        except subprocess.TimeoutExpired:
            return ExecutionResult(
                command=cmd_str,
                success=False,
                output=f"Command timed out after {timeout} seconds.",
                return_code=-2,
            )
        except Exception as exc:
            return ExecutionResult(
                command=cmd_str,
                success=False,
                output=f"Unexpected error: {exc}",
                return_code=-3,
            )

    @staticmethod
    def check_tool_available(tool_name: str) -> bool:
        """Return True if the named executable can be found on PATH."""
        result = ExecutionService.run_command(["which", tool_name], timeout=5)
        if not result.success:
            result = ExecutionService.run_command(["where", tool_name], timeout=5)
        return result.success

check_tool_available(tool_name) staticmethod

Return True if the named executable can be found on PATH.

Source code in accessibility_mgr/services/execution_service.py 
106
107
108
109
110
111
112
@staticmethod
def check_tool_available(tool_name: str) -> bool:
    """Return True if the named executable can be found on PATH."""
    result = ExecutionService.run_command(["which", tool_name], timeout=5)
    if not result.success:
        result = ExecutionService.run_command(["where", tool_name], timeout=5)
    return result.success

run_command(command, timeout=120, cwd=None) staticmethod

Run a shell command and return the full result.

Source code in accessibility_mgr/services/execution_service.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@staticmethod
def run_command(
    command: list[str],
    timeout: int = 120,
    cwd: Optional[str] = None,
) -> ExecutionResult:
    """Run a shell command and return the full result."""
    cmd_str = " ".join(command)
    exe_name = Path(command[0]).name
    if exe_name not in ALLOWED_EXECUTABLES:
        return ExecutionResult(
            command=cmd_str,
            success=False,
            output=(
                f"Executable '{exe_name}' is not in the permitted allowlist. "
                "Add it to ALLOWED_EXECUTABLES in execution_service.py if required."
            ),
            return_code=-4,
        )
    try:
        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
            cwd=cwd,
        )
        output = result.stdout or ""
        if result.stderr:
            output = output + ("\n" if output else "") + result.stderr
        return ExecutionResult(
            command=cmd_str,
            success=result.returncode == 0,
            output=output.strip(),
            return_code=result.returncode,
        )
    except FileNotFoundError:
        return ExecutionResult(
            command=cmd_str,
            success=False,
            output=(
                f"Command not found: '{command[0]}'. "
                "Is the tool installed and on PATH?"
            ),
            return_code=-1,
        )
    except subprocess.TimeoutExpired:
        return ExecutionResult(
            command=cmd_str,
            success=False,
            output=f"Command timed out after {timeout} seconds.",
            return_code=-2,
        )
    except Exception as exc:
        return ExecutionResult(
            command=cmd_str,
            success=False,
            output=f"Unexpected error: {exc}",
            return_code=-3,
        )

Metadata validation

Purpose: validation rules for metadata payloads and constraints.

Metadata governance and validation services.

This module centralizes metadata normalization, controlled vocabulary validation, and accessibility metadata checks for EPUB-oriented workflows.

MetadataValidationService

Validate metadata against governance and accessibility rules.

Source code in accessibility_mgr/services/metadata_validation.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class MetadataValidationService:
    """Validate metadata against governance and accessibility rules."""

    def validate_dublin_core(
        self,
        metadata: dict,
    ) -> ValidationResult:
        issues: list[ValidationIssue] = []

        for field_name in sorted(DC_REQUIRED_FIELDS):
            value = metadata.get(field_name)

            if value:
                continue

            issues.append(
                ValidationIssue(
                    field_name=field_name,
                    severity="error",
                    message=f"Required Dublin Core field missing: {field_name}",
                )
            )

        language = metadata.get("language")

        if language and language not in SUPPORTED_LANGUAGES:
            issues.append(
                ValidationIssue(
                    field_name="language",
                    severity="warning",
                    message="Language code is not in approved vocabulary",
                    suggested_value="en",
                )
            )

        return ValidationResult(valid=not issues, issues=issues)

    def validate_epub_accessibility(
        self,
        metadata: dict,
    ) -> ValidationResult:
        issues: list[ValidationIssue] = []

        access_mode = metadata.get("schema:accessMode")

        if access_mode and access_mode not in ACCESS_MODE_VALUES:
            issues.append(
                ValidationIssue(
                    field_name="schema:accessMode",
                    severity="error",
                    message="Invalid accessibility accessMode value",
                    suggested_value="textual",
                )
            )

        access_hazard = metadata.get("schema:accessHazard")

        if access_hazard and access_hazard not in ACCESS_HAZARD_VALUES:
            issues.append(
                ValidationIssue(
                    field_name="schema:accessHazard",
                    severity="error",
                    message="Invalid accessibility hazard declaration",
                    suggested_value="none",
                )
            )

        if not metadata.get("schema:accessibilitySummary"):
            issues.append(
                ValidationIssue(
                    field_name="schema:accessibilitySummary",
                    severity="warning",
                    message="Accessibility summary should be provided",
                )
            )

        return ValidationResult(valid=not issues, issues=issues)

    def validate_all(self, metadata: dict) -> ValidationResult:
        dc_result = self.validate_dublin_core(metadata)
        epub_result = self.validate_epub_accessibility(metadata)

        issues = dc_result.issues + epub_result.issues

        has_errors = any(issue.severity == "error" for issue in issues)

        return ValidationResult(
            valid=not has_errors,
            issues=issues,
        )

Multi-tenant

Purpose: tenant isolation and tenant-scoped helper operations.

Multi-tenant organization infrastructure — SQLite-backed.

MultiTenantService

SQLite-backed organization and tenant isolation service.

Source code in accessibility_mgr/services/multi_tenant.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class MultiTenantService:
    """SQLite-backed organization and tenant isolation service."""

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(self.database_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _initialize(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS organization (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    organization_id TEXT NOT NULL UNIQUE,
                    name TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )"""
            )
            conn.execute(
                """CREATE TABLE IF NOT EXISTS tenant_membership (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    username TEXT NOT NULL,
                    organization_id TEXT NOT NULL,
                    role TEXT NOT NULL
                )"""
            )

    def create_organization(self, name: str) -> dict[str, Any]:
        created_at = datetime.now(UTC).isoformat()
        with self._connect() as conn:
            count = conn.execute("SELECT COUNT(*) FROM organization").fetchone()[0]
            org_id = f"org-{count + 1}"
            conn.execute(
                "INSERT INTO organization (organization_id, name, created_at) VALUES (?, ?, ?)",
                (org_id, name, created_at),
            )
        return {
            "organization_id": org_id,
            "name": name,
            "created_at": created_at,
        }

    def add_member(
        self,
        *,
        username: str,
        organization_id: str,
        role: str,
    ) -> dict[str, Any]:
        with self._connect() as conn:
            conn.execute(
                "INSERT INTO tenant_membership (username, organization_id, role) VALUES (?, ?, ?)",
                (username, organization_id, role),
            )
        return {
            "username": username,
            "organization_id": organization_id,
            "role": role,
        }

    def list_organizations(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "organization_id": r["organization_id"],
                    "name": r["name"],
                    "created_at": r["created_at"],
                }
                for r in conn.execute("SELECT * FROM organization").fetchall()
            ]

    def list_memberships(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "username": r["username"],
                    "organization_id": r["organization_id"],
                    "role": r["role"],
                }
                for r in conn.execute("SELECT * FROM tenant_membership").fetchall()
            ]

Persistent analytics

Purpose: durable analytics storage and retrieval.

Persistent analytics service — SQLite-backed KPI storage.

AUDIT-FIX-007: this was previously an empty subclass of AnalyticsService with no override at all:

class PersistentAnalyticsService(AnalyticsService):
    '''Compatibility wrapper for API-facing analytics access.'''

Despite the name, every metric was stored in a plain Python list and lost on every restart. This now persists metrics to a real SQLite table in the same database the rest of the application uses.

PersistentAnalyticsService

Bases: AnalyticsService

SQLite-backed analytics service.

Source code in accessibility_mgr/services/persistent_analytics.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class PersistentAnalyticsService(AnalyticsService):
    """SQLite-backed analytics service."""

    def __init__(self, database_path: Path | None = None) -> None:
        super().__init__()
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        return sqlite3.connect(self.database_path)

    def _initialize(self) -> None:
        with self._connect() as connection:
            connection.execute(
                """
                CREATE TABLE IF NOT EXISTS analytics_metric (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    metric_name TEXT NOT NULL,
                    metric_value REAL NOT NULL,
                    category TEXT NOT NULL,
                    recorded_at TEXT NOT NULL,
                    metadata TEXT
                )
                """
            )

    def record_metric(
        self,
        *,
        metric_name: str,
        metric_value: float,
        category: str,
        metadata: dict[str, Any] | None = None,
    ) -> KPIRecord:
        record = KPIRecord(
            metric_name=metric_name,
            metric_value=metric_value,
            category=category,
            recorded_at=datetime.now(timezone.utc).isoformat(),
            metadata=metadata or {},
        )

        with self._connect() as connection:
            connection.execute(
                "INSERT INTO analytics_metric "
                "(metric_name, metric_value, category, recorded_at, metadata) "
                "VALUES (?, ?, ?, ?, ?)",
                (
                    record.metric_name,
                    record.metric_value,
                    record.category,
                    record.recorded_at,
                    json.dumps(record.metadata),
                ),
            )

        return record

    def summarize(self) -> dict[str, Any]:
        with self._connect() as connection:
            rows = connection.execute(
                "SELECT metric_value, category FROM analytics_metric"
            ).fetchall()

        total = len(rows)
        if not total:
            return {"total_metrics": 0, "average_score": 0, "categories": {}}

        avg = sum(value for value, _ in rows) / total

        categories: dict[str, int] = {}
        for _, category in rows:
            categories[category] = categories.get(category, 0) + 1

        return {
            "total_metrics": total,
            "average_score": round(avg, 2),
            "categories": categories,
        }

    def list_metrics(self) -> list[dict[str, Any]]:
        with self._connect() as connection:
            rows = connection.execute(
                "SELECT metric_name, metric_value, category, recorded_at, metadata "
                "FROM analytics_metric ORDER BY id DESC"
            ).fetchall()

        results = []
        for name, value, category, recorded_at, metadata in rows:
            results.append(
                {
                    "metric_name": name,
                    "metric_value": value,
                    "category": category,
                    "recorded_at": recorded_at,
                    "metadata": json.loads(metadata) if metadata else {},
                }
            )
        return results

Persistent provenance

Purpose: durable provenance registry back-end.

Persistent provenance registry — SQLite-backed provenance events.

AUDIT-FIX-007: this was previously an empty subclass of ProvenanceRegistry with no override at all, so despite the name every provenance event was stored in a plain Python list and lost on every restart. It now persists events to a real SQLite table in the same database the rest of the application uses.

PersistentProvenanceRegistry

Bases: ProvenanceRegistry

SQLite-backed provenance registry.

Source code in accessibility_mgr/services/persistent_provenance.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class PersistentProvenanceRegistry(ProvenanceRegistry):
    """SQLite-backed provenance registry."""

    def __init__(self, database_path: Path | None = None) -> None:
        super().__init__()
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        return sqlite3.connect(self.database_path)

    def _initialize(self) -> None:
        with self._connect() as connection:
            connection.execute(
                """
                CREATE TABLE IF NOT EXISTS provenance_event (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    asset_id INTEGER NOT NULL,
                    event_type TEXT NOT NULL,
                    summary TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    metadata TEXT
                )
                """
            )

    def register_event(
        self,
        *,
        asset_id: int,
        event_type: str,
        summary: str,
        metadata: dict[str, Any] | None = None,
    ) -> ProvenanceEvent:
        event = ProvenanceEvent(
            asset_id=asset_id,
            event_type=event_type,
            summary=summary,
            created_at=datetime.now(timezone.utc).isoformat(),
            metadata=metadata or {},
        )

        with self._connect() as connection:
            connection.execute(
                "INSERT INTO provenance_event "
                "(asset_id, event_type, summary, created_at, metadata) "
                "VALUES (?, ?, ?, ?, ?)",
                (
                    event.asset_id,
                    event.event_type,
                    event.summary,
                    event.created_at,
                    json.dumps(event.metadata),
                ),
            )

        return event

    def list_events(
        self,
        *,
        asset_id: int | None = None,
    ) -> list[dict[str, Any]]:
        with self._connect() as connection:
            if asset_id is not None:
                rows = connection.execute(
                    "SELECT asset_id, event_type, summary, created_at, metadata "
                    "FROM provenance_event WHERE asset_id = ? ORDER BY id ASC",
                    (asset_id,),
                ).fetchall()
            else:
                rows = connection.execute(
                    "SELECT asset_id, event_type, summary, created_at, metadata "
                    "FROM provenance_event ORDER BY id ASC"
                ).fetchall()

        results = []
        for row_asset_id, event_type, summary, created_at, metadata in rows:
            results.append(
                {
                    "asset_id": row_asset_id,
                    "event_type": event_type,
                    "summary": summary,
                    "created_at": created_at,
                    "metadata": json.loads(metadata) if metadata else {},
                }
            )
        return results

Persistent queue

Purpose: durable workflow queue persistence and replay.

Persistent distributed workflow queue backend.

AUDIT-FIX-004/006: this module was fully implemented (a genuine SQLite-backed queue) but was never imported anywhere else in the codebase — the live app used the in-memory WorkflowQueueService instead, so "Persistent SQLite-backed workflow queue" was true of this file in isolation but not true of anything a user could actually reach. It now has full method parity with WorkflowQueueService (next_job / complete_job / fail_job) so it can be used as a drop-in replacement, and services/singletons.py has been updated to use it.

PersistentWorkflowQueue

SQLite-backed distributed workflow queue.

Method names intentionally mirror WorkflowQueueService (enqueue / next_job / complete_job / fail_job / list_jobs) so this can be used as a drop-in replacement wherever that class is used.

Source code in accessibility_mgr/services/persistent_queue.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class PersistentWorkflowQueue:
    """SQLite-backed distributed workflow queue.

    Method names intentionally mirror WorkflowQueueService
    (enqueue / next_job / complete_job / fail_job / list_jobs) so this can
    be used as a drop-in replacement wherever that class is used.
    """

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        return sqlite3.connect(self.database_path)

    def _initialize(self) -> None:
        with self._connect() as connection:
            connection.execute(
                """
                CREATE TABLE IF NOT EXISTS workflow_queue (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    workflow_name TEXT NOT NULL,
                    asset_id INTEGER NOT NULL,
                    priority INTEGER NOT NULL,
                    status TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )
                """
            )

    def enqueue(
        self,
        *,
        workflow_name: str,
        asset_id: int,
        priority: int = 5,
    ) -> PersistentWorkflowJob:
        created_at = datetime.now(timezone.utc).isoformat()

        with self._connect() as connection:
            cursor = connection.execute(
                """
                INSERT INTO workflow_queue (
                    workflow_name, asset_id, priority, status, created_at
                ) VALUES (?, ?, ?, ?, ?)
                """,
                (workflow_name, asset_id, priority, "queued", created_at),
            )
            job_id = cursor.lastrowid

        return PersistentWorkflowJob(
            id=job_id,
            workflow_name=workflow_name,
            asset_id=asset_id,
            priority=priority,
            status="queued",
            created_at=created_at,
        )

    def next_job(self) -> PersistentWorkflowJob | None:
        with self._connect() as connection:
            row = connection.execute(
                """
                SELECT id, workflow_name, asset_id, priority, created_at
                FROM workflow_queue
                WHERE status = 'queued'
                ORDER BY priority ASC, created_at ASC
                LIMIT 1
                """
            ).fetchone()

            if not row:
                return None

            connection.execute(
                "UPDATE workflow_queue SET status = 'running' WHERE id = ?",
                (row[0],),
            )

        return PersistentWorkflowJob(
            id=row[0],
            workflow_name=row[1],
            asset_id=row[2],
            priority=row[3],
            status="running",
            created_at=row[4],
        )

    def complete_job(self, job: PersistentWorkflowJob) -> None:
        with self._connect() as connection:
            connection.execute(
                "UPDATE workflow_queue SET status = 'completed' WHERE id = ?",
                (job.id,),
            )
        job.status = "completed"

    def fail_job(self, job: PersistentWorkflowJob) -> None:
        with self._connect() as connection:
            connection.execute(
                "UPDATE workflow_queue SET status = 'failed' WHERE id = ?",
                (job.id,),
            )
        job.status = "failed"

    def list_jobs(self) -> list[dict[str, Any]]:
        with self._connect() as connection:
            rows = connection.execute(
                """
                SELECT id, workflow_name, asset_id, priority,
                       status, created_at
                FROM workflow_queue
                ORDER BY priority ASC, created_at ASC
                """
            ).fetchall()

        return [
            {
                "id": row[0],
                "workflow_name": row[1],
                "asset_id": row[2],
                "priority": row[3],
                "status": row[4],
                "created_at": row[5],
            }
            for row in rows
        ]

Pipeline service

Purpose: pipeline orchestration, step flow, and status lifecycle.

Pipeline service — multi-stage accessibility production workflow automation.

Each pipeline definition carries ordered steps with tool names and commands. Execution runs each step via ExecutionService and persists run records to DB.

PipelineRunResult dataclass

Outcome of a complete pipeline execution, including per-step results.

Source code in accessibility_mgr/services/pipeline_service.py 
141
142
143
144
145
146
147
@dataclass
class PipelineRunResult:
    """Outcome of a complete pipeline execution, including per-step results."""
    pipeline_name: str
    run_id: int
    step_results: list[ExecutionResult]
    overall_success: bool

PipelineService

Service for listing and executing multi-step workflow pipelines.

Source code in accessibility_mgr/services/pipeline_service.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
class PipelineService:
    """Service for listing and executing multi-step workflow pipelines."""

    @staticmethod
    def list_pipelines() -> list[WorkflowPipeline]:
        """Return all registered workflow pipelines."""
        return PIPELINES

    @staticmethod
    def get_pipeline(name: str) -> Optional[WorkflowPipeline]:
        """Look up a pipeline by name, or return None if not found."""
        return _PIPELINE_MAP.get(name)

    @staticmethod
    def run_pipeline(name: str, input_path: str = "") -> PipelineRunResult:
        """Execute all steps of a named pipeline, persisting results to DB.

        Each step's ``required_binary`` is checked via shutil.which before
        execution.  Missing binaries produce an explicit FAIL result with
        installation guidance rather than silently returning success.
        """
        pipeline = _PIPELINE_MAP.get(name)
        if pipeline is None:
            return PipelineRunResult(
                pipeline_name=name,
                run_id=-1,
                step_results=[
                    ExecutionResult(
                        command=name,
                        success=False,
                        output=f"Unknown pipeline: '{name}'",
                        return_code=-1,
                    )
                ],
                overall_success=False,
            )

        run_id = Q.start_pipeline_run(name)
        step_results: list[ExecutionResult] = []
        overall_success = True

        for step in pipeline.steps:
            # Explicit binary pre-check with install guidance
            if step.required_binary and not shutil.which(step.required_binary):
                missing_result = ExecutionResult(
                    command=step.required_binary,
                    success=False,
                    output=(
                        f"Required binary '{step.required_binary}' not found on PATH.  "
                        f"Install the tool and ensure it is accessible, or configure its "
                        f"path in tools.ini before running this pipeline."
                    ),
                    return_code=-5,
                )
                Q.log_pipeline_step(
                    pipeline_run_id=run_id,
                    step_name=step.name,
                    tool=step.tool,
                    command=step.required_binary,
                    success=False,
                    output=missing_result.output,
                )
                step_results.append(missing_result)
                overall_success = False
                continue

            command = step.build_command(input_path)
            result = ExecutionService.run_command(command, timeout=step.timeout)

            Q.log_pipeline_step(
                pipeline_run_id=run_id,
                step_name=step.name,
                tool=step.tool,
                command=result.command,
                success=result.success,
                output=result.output,
            )

            step_results.append(result)

            if not result.success:
                overall_success = False
                # Continue remaining steps so all results are recorded

        Q.finish_pipeline_run(run_id, status="completed" if overall_success else "failed")

        return PipelineRunResult(
            pipeline_name=name,
            run_id=run_id,
            step_results=step_results,
            overall_success=overall_success,
        )

get_pipeline(name) staticmethod

Look up a pipeline by name, or return None if not found.

Source code in accessibility_mgr/services/pipeline_service.py 
158
159
160
161
@staticmethod
def get_pipeline(name: str) -> Optional[WorkflowPipeline]:
    """Look up a pipeline by name, or return None if not found."""
    return _PIPELINE_MAP.get(name)

list_pipelines() staticmethod

Return all registered workflow pipelines.

Source code in accessibility_mgr/services/pipeline_service.py 
153
154
155
156
@staticmethod
def list_pipelines() -> list[WorkflowPipeline]:
    """Return all registered workflow pipelines."""
    return PIPELINES

run_pipeline(name, input_path='') staticmethod

Execute all steps of a named pipeline, persisting results to DB.

Each step's required_binary is checked via shutil.which before execution. Missing binaries produce an explicit FAIL result with installation guidance rather than silently returning success.

Source code in accessibility_mgr/services/pipeline_service.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@staticmethod
def run_pipeline(name: str, input_path: str = "") -> PipelineRunResult:
    """Execute all steps of a named pipeline, persisting results to DB.

    Each step's ``required_binary`` is checked via shutil.which before
    execution.  Missing binaries produce an explicit FAIL result with
    installation guidance rather than silently returning success.
    """
    pipeline = _PIPELINE_MAP.get(name)
    if pipeline is None:
        return PipelineRunResult(
            pipeline_name=name,
            run_id=-1,
            step_results=[
                ExecutionResult(
                    command=name,
                    success=False,
                    output=f"Unknown pipeline: '{name}'",
                    return_code=-1,
                )
            ],
            overall_success=False,
        )

    run_id = Q.start_pipeline_run(name)
    step_results: list[ExecutionResult] = []
    overall_success = True

    for step in pipeline.steps:
        # Explicit binary pre-check with install guidance
        if step.required_binary and not shutil.which(step.required_binary):
            missing_result = ExecutionResult(
                command=step.required_binary,
                success=False,
                output=(
                    f"Required binary '{step.required_binary}' not found on PATH.  "
                    f"Install the tool and ensure it is accessible, or configure its "
                    f"path in tools.ini before running this pipeline."
                ),
                return_code=-5,
            )
            Q.log_pipeline_step(
                pipeline_run_id=run_id,
                step_name=step.name,
                tool=step.tool,
                command=step.required_binary,
                success=False,
                output=missing_result.output,
            )
            step_results.append(missing_result)
            overall_success = False
            continue

        command = step.build_command(input_path)
        result = ExecutionService.run_command(command, timeout=step.timeout)

        Q.log_pipeline_step(
            pipeline_run_id=run_id,
            step_name=step.name,
            tool=step.tool,
            command=result.command,
            success=result.success,
            output=result.output,
        )

        step_results.append(result)

        if not result.success:
            overall_success = False
            # Continue remaining steps so all results are recorded

    Q.finish_pipeline_run(run_id, status="completed" if overall_success else "failed")

    return PipelineRunResult(
        pipeline_name=name,
        run_id=run_id,
        step_results=step_results,
        overall_success=overall_success,
    )

PipelineStep dataclass

A single step in a multi-stage workflow pipeline.

Source code in accessibility_mgr/services/pipeline_service.py 
19
20
21
22
23
24
25
26
27
28
29
30
@dataclass
class PipelineStep:
    """A single step in a multi-stage workflow pipeline."""
    name: str
    tool: str
    command_template: str   # {input} replaced at runtime
    timeout: int = 120
    required_binary: str = ""  # Binary to check before running

    def build_command(self, input_path: str = "") -> list[str]:
        cmd = self.command_template.replace("{input}", input_path)
        return shlex.split(cmd)

WorkflowPipeline dataclass

An ordered collection of PipelineSteps representing a production workflow.

Source code in accessibility_mgr/services/pipeline_service.py 
33
34
35
36
37
38
@dataclass
class WorkflowPipeline:
    """An ordered collection of PipelineSteps representing a production workflow."""
    name: str
    description: str
    steps: list[PipelineStep] = field(default_factory=list)

Provenance registry

Purpose: provenance event registration and lookup.

Unified provenance registry.

Aggregates metadata audit events, QA artifacts, and QA execution history into a normalized provenance timeline abstraction.

ProvenanceRegistry

Central provenance aggregation service.

Source code in accessibility_mgr/services/provenance_registry.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class ProvenanceRegistry:
    """Central provenance aggregation service."""

    def __init__(self) -> None:
        self._events: list[ProvenanceEvent] = []

    def register_event(
        self,
        *,
        asset_id: int,
        event_type: str,
        summary: str,
        metadata: dict[str, Any] | None = None,
    ) -> ProvenanceEvent:
        event = ProvenanceEvent(
            asset_id=asset_id,
            event_type=event_type,
            summary=summary,
            created_at=datetime.now(timezone.utc).isoformat(),
            metadata=metadata or {},
        )

        self._events.append(event)
        return event

    def list_events(
        self,
        *,
        asset_id: int | None = None,
    ) -> list[dict[str, Any]]:
        events = self._events

        if asset_id is not None:
            events = [
                event for event in events
                if event.asset_id == asset_id
            ]

        return [asdict(event) for event in events]

QA service

Purpose: QA workflow execution, scoring, and result emission.

QA service — accessibility validation tool registry and execution.

Changes applied (see fix_specs.json): FIX-012 When job_type and job_id are provided, a QA_RUN event is written to the job's metadata_event record in addition to qa_run table.

QAService

Service for listing and executing QA tooling commands.

Source code in accessibility_mgr/services/qa_service.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class QAService:
    """Service for listing and executing QA tooling commands."""

    @staticmethod
    def list_tools() -> list[QATool]:
        return QA_TOOLS

    @staticmethod
    def get_tool(name: str) -> Optional[QATool]:
        return _TOOL_MAP.get(name)

    @staticmethod
    def run_tool(
        name: str,
        input_path: str = "",
        job_type: Optional[str] = None,
        job_id: Optional[int] = None,
    ) -> ExecutionResult:
        """Execute a QA tool, persist the result, and return it.

        FIX-012: When job_type and job_id are provided, a QA_RUN event is
        also written to the job's metadata_event record so the result appears
        in the job's audit trail.
        """
        tool = _TOOL_MAP.get(name)
        if tool is None:
            return ExecutionResult(
                command=name,
                success=False,
                output=f"Unknown QA tool: '{name}'",
                return_code=-1,
            )

        if tool.manual_review:
            # Manual-review tools have no CLI — the UI must route them to
            # the review form (qa.py _run_tool_dialog → manual branch).
            # If run_tool is called for one anyway, surface an honest error
            # rather than running echo and recording a fake SUCCESS.
            return ExecutionResult(
                command="(manual review — no CLI)",
                success=False,
                output=(
                    f"'{name}' is a manual-review workflow with no CLI tool. "
                    "Use the 'Record Manual Review' form to submit findings."
                ),
                return_code=-2,
            )

        command = tool.build_command(input_path)
        result = ExecutionService.run_command(command, timeout=tool.timeout)

        # Persist to qa_run table
        Q.log_qa_run(
            tool_name=name,
            command=result.command,
            success=result.success,
            output=result.output,
            job_type=job_type,
            job_id=job_id,
        )

        # FIX-012: also write to the job's event log when linked to a job
        if job_type and job_id:
            Q.log_event(
                job_type, job_id,
                "QA_RUN",
                "SUCCESS" if result.success else "FAILURE",
                agent="system",
                detail=f"{name}: {'PASS' if result.success else 'FAIL'}",
                extra_metadata={
                    "tool": name,
                    "command": result.command,
                    "output_preview": result.output[:500] if result.output else "",
                },
            )

        return result

    @staticmethod
    def log_manual_qa_review(
        tool_name: str,
        asset_path: str,
        passed: bool,
        reviewer: str,
        notes: str,
        job_type: Optional[str] = None,
        job_id: Optional[int] = None,
    ) -> None:
        """Persist a manual QA review finding to qa_run and optionally to a job's event log.

        Used by the 'Record Manual Review' form for tools where no CLI
        exists (manual_review=True), such as ANZAGG Validation. Writing a
        real record here replaces the previous behavior of running `echo`
        and fabricating a SUCCESS result.
        """
        outcome = "PASS" if passed else "FAIL"
        summary = f"Manual review by {reviewer or 'unknown'}: {outcome}. {notes}".strip()

        Q.log_qa_run(
            tool_name=tool_name,
            command="(manual review)",
            success=passed,
            output=summary,
            job_type=job_type,
            job_id=job_id,
        )

        if job_type and job_id:
            Q.log_event(
                job_type, job_id,
                "MANUAL_QA_REVIEW",
                "SUCCESS" if passed else "FAILURE",
                agent=reviewer or "reviewer",
                detail=f"{tool_name} manual review: {outcome}",
                extra_metadata={
                    "tool": tool_name,
                    "asset_path": asset_path,
                    "reviewer": reviewer,
                    "notes": notes,
                },
            )

log_manual_qa_review(tool_name, asset_path, passed, reviewer, notes, job_type=None, job_id=None) staticmethod

Persist a manual QA review finding to qa_run and optionally to a job's event log.

Used by the 'Record Manual Review' form for tools where no CLI exists (manual_review=True), such as ANZAGG Validation. Writing a real record here replaces the previous behavior of running echo and fabricating a SUCCESS result.

Source code in accessibility_mgr/services/qa_service.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
@staticmethod
def log_manual_qa_review(
    tool_name: str,
    asset_path: str,
    passed: bool,
    reviewer: str,
    notes: str,
    job_type: Optional[str] = None,
    job_id: Optional[int] = None,
) -> None:
    """Persist a manual QA review finding to qa_run and optionally to a job's event log.

    Used by the 'Record Manual Review' form for tools where no CLI
    exists (manual_review=True), such as ANZAGG Validation. Writing a
    real record here replaces the previous behavior of running `echo`
    and fabricating a SUCCESS result.
    """
    outcome = "PASS" if passed else "FAIL"
    summary = f"Manual review by {reviewer or 'unknown'}: {outcome}. {notes}".strip()

    Q.log_qa_run(
        tool_name=tool_name,
        command="(manual review)",
        success=passed,
        output=summary,
        job_type=job_type,
        job_id=job_id,
    )

    if job_type and job_id:
        Q.log_event(
            job_type, job_id,
            "MANUAL_QA_REVIEW",
            "SUCCESS" if passed else "FAILURE",
            agent=reviewer or "reviewer",
            detail=f"{tool_name} manual review: {outcome}",
            extra_metadata={
                "tool": tool_name,
                "asset_path": asset_path,
                "reviewer": reviewer,
                "notes": notes,
            },
        )

run_tool(name, input_path='', job_type=None, job_id=None) staticmethod

Execute a QA tool, persist the result, and return it.

FIX-012: When job_type and job_id are provided, a QA_RUN event is also written to the job's metadata_event record so the result appears in the job's audit trail.

Source code in accessibility_mgr/services/qa_service.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
@staticmethod
def run_tool(
    name: str,
    input_path: str = "",
    job_type: Optional[str] = None,
    job_id: Optional[int] = None,
) -> ExecutionResult:
    """Execute a QA tool, persist the result, and return it.

    FIX-012: When job_type and job_id are provided, a QA_RUN event is
    also written to the job's metadata_event record so the result appears
    in the job's audit trail.
    """
    tool = _TOOL_MAP.get(name)
    if tool is None:
        return ExecutionResult(
            command=name,
            success=False,
            output=f"Unknown QA tool: '{name}'",
            return_code=-1,
        )

    if tool.manual_review:
        # Manual-review tools have no CLI — the UI must route them to
        # the review form (qa.py _run_tool_dialog → manual branch).
        # If run_tool is called for one anyway, surface an honest error
        # rather than running echo and recording a fake SUCCESS.
        return ExecutionResult(
            command="(manual review — no CLI)",
            success=False,
            output=(
                f"'{name}' is a manual-review workflow with no CLI tool. "
                "Use the 'Record Manual Review' form to submit findings."
            ),
            return_code=-2,
        )

    command = tool.build_command(input_path)
    result = ExecutionService.run_command(command, timeout=tool.timeout)

    # Persist to qa_run table
    Q.log_qa_run(
        tool_name=name,
        command=result.command,
        success=result.success,
        output=result.output,
        job_type=job_type,
        job_id=job_id,
    )

    # FIX-012: also write to the job's event log when linked to a job
    if job_type and job_id:
        Q.log_event(
            job_type, job_id,
            "QA_RUN",
            "SUCCESS" if result.success else "FAILURE",
            agent="system",
            detail=f"{name}: {'PASS' if result.success else 'FAIL'}",
            extra_metadata={
                "tool": name,
                "command": result.command,
                "output_preview": result.output[:500] if result.output else "",
            },
        )

    return result

RBAC

Purpose: role-based access control logic and permission checks.

Role-based access control infrastructure.

RBACService

Central authorization service.

Source code in accessibility_mgr/services/rbac.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class RBACService:
    """Central authorization service."""

    def __init__(self) -> None:
        self._roles: dict[str, Role] = {}
        self._seed_roles()

    def _seed_roles(self) -> None:
        self.register_role(
            Role(
                name="administrator",
                permissions={
                    "qa.execute",
                    "qa.review",
                    "governance.manage",
                    "workflow.manage",
                    "analytics.view",
                    "rbac.manage",
                },
            )
        )

        self.register_role(
            Role(
                name="operator",
                permissions={
                    "qa.execute",
                    "workflow.manage",
                    "analytics.view",
                    "governance.manage",
                },
            )
        )

        self.register_role(
            Role(
                name="reviewer",
                permissions={
                    "qa.review",
                    "analytics.view",
                },
            )
        )

    def register_role(self, role: Role) -> None:
        self._roles[role.name] = role

    def get_role(self, role_name: str) -> Role | None:
        return self._roles.get(role_name)

    def authorize(
        self,
        user: UserIdentity,
        permission: str,
    ) -> bool:
        for role in user.roles:
            if permission in role.permissions:
                return True

        return False

    def list_roles(self) -> list[dict]:
        return [
            {
                "name": role.name,
                "permissions": sorted(role.permissions),
            }
            for role in self._roles.values()
        ]

SLA monitoring

Purpose: SLA tracking and breach detection helpers.

SLA monitoring and operational escalation infrastructure — SQLite-backed.

SLAMonitoringService

SQLite-backed SLA and escalation monitoring.

Source code in accessibility_mgr/services/sla_monitoring.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class SLAMonitoringService:
    """SQLite-backed SLA and escalation monitoring."""

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(self.database_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _initialize(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS sla_record (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    workflow_name TEXT NOT NULL,
                    asset_id INTEGER NOT NULL,
                    started_at TEXT NOT NULL,
                    sla_minutes INTEGER NOT NULL DEFAULT 30,
                    breached INTEGER NOT NULL DEFAULT 0
                )"""
            )
            conn.execute(
                """CREATE TABLE IF NOT EXISTS escalation_event (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    workflow_name TEXT NOT NULL,
                    asset_id INTEGER NOT NULL,
                    severity TEXT NOT NULL,
                    summary TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )"""
            )

    def register_workflow(
        self,
        *,
        workflow_name: str,
        asset_id: int,
        sla_minutes: int = 30,
    ) -> dict[str, Any]:
        started_at = datetime.now(UTC).isoformat()
        with self._connect() as conn:
            conn.execute(
                "INSERT INTO sla_record (workflow_name, asset_id, started_at, sla_minutes, breached) "
                "VALUES (?, ?, ?, ?, 0)",
                (workflow_name, asset_id, started_at, sla_minutes),
            )
        return {
            "workflow_name": workflow_name,
            "asset_id": asset_id,
            "started_at": started_at,
            "sla_minutes": sla_minutes,
            "breached": False,
        }

    def evaluate_slas(self) -> list[dict[str, Any]]:
        now = datetime.now(UTC)
        with self._connect() as conn:
            rows = conn.execute("SELECT * FROM sla_record").fetchall()
            for row in rows:
                started = datetime.fromisoformat(row["started_at"])
                deadline = started + timedelta(minutes=row["sla_minutes"])
                if now > deadline and not row["breached"]:
                    conn.execute(
                        "UPDATE sla_record SET breached = 1 WHERE id = ?", (row["id"],)
                    )
                    conn.execute(
                        "INSERT INTO escalation_event (workflow_name, asset_id, severity, summary, created_at) "
                        "VALUES (?, ?, 'high', 'Workflow SLA breached', ?)",
                        (row["workflow_name"], row["asset_id"], now.isoformat()),
                    )
            return [
                {
                    "workflow_name": r["workflow_name"],
                    "asset_id": r["asset_id"],
                    "started_at": r["started_at"],
                    "sla_minutes": r["sla_minutes"],
                    "breached": bool(r["breached"]),
                }
                for r in conn.execute("SELECT * FROM sla_record").fetchall()
            ]

    def list_escalations(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "workflow_name": r["workflow_name"],
                    "asset_id": r["asset_id"],
                    "severity": r["severity"],
                    "summary": r["summary"],
                    "created_at": r["created_at"],
                }
                for r in conn.execute(
                    "SELECT * FROM escalation_event ORDER BY id"
                ).fetchall()
            ]

    def health_summary(self) -> dict[str, Any]:
        with self._connect() as conn:
            total = conn.execute("SELECT COUNT(*) FROM sla_record").fetchone()[0]
            breached = conn.execute(
                "SELECT COUNT(*) FROM sla_record WHERE breached = 1"
            ).fetchone()[0]
            escalations = conn.execute("SELECT COUNT(*) FROM escalation_event").fetchone()[0]
            return {
                "tracked_workflows": total,
                "sla_breaches": breached,
                "healthy_workflows": total - breached,
                "escalations": escalations,
            }

Toolchain core

Purpose: shared toolchain runtime helpers.

Accessibility toolchain integration layer.

Provides subprocess execution wrappers for: - DAISY Ace - EPUBCheck - DAISY Pipeline

This layer standardizes: - timeout handling - artifact capture - execution isolation - structured results

AccessibilityToolchainService

Accessibility subprocess execution service.

Source code in accessibility_mgr/services/toolchain.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class AccessibilityToolchainService:
    """Accessibility subprocess execution service."""

    def __init__(self, *, timeout_seconds: int = 60) -> None:
        self.timeout_seconds = timeout_seconds

    def execute(
        self,
        *,
        tool_name: str,
        command: list[str],
        artifact_paths: list[str] | None = None,
    ) -> ToolExecutionResult:
        try:
            completed = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=self.timeout_seconds,
                check=False,
            )

            status = (
                "completed"
                if completed.returncode == 0
                else "failed"
            )

            return ToolExecutionResult(
                tool_name=tool_name,
                command=command,
                status=status,
                exit_code=completed.returncode,
                stdout=completed.stdout,
                stderr=completed.stderr,
                artifacts=artifact_paths or [],
                executed_at=datetime.now(timezone.utc).isoformat(),
            )

        except subprocess.TimeoutExpired as exc:
            return ToolExecutionResult(
                tool_name=tool_name,
                command=command,
                status="timeout",
                exit_code=-1,
                stdout=exc.stdout or "",
                stderr=exc.stderr or "Execution timed out",
                artifacts=[],
                executed_at=datetime.now(timezone.utc).isoformat(),
            )

Toolchain binaries

Purpose: discovery and management of external binary dependencies.

Production accessibility binary integrations.

Provides executable wrappers for: - DAISY Ace CLI - EPUBCheck - Liblouis braille translation (lou_translate / file2brl) - GLOW (ACB Large Print Toolkit)

This layer extends the existing subprocess abstraction with: - binary discovery - secure invocation - artifact output directories - structured execution contracts

AccessibilityBinaryIntegrationService

Production binary integration service.

Source code in accessibility_mgr/services/toolchain_binaries.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class AccessibilityBinaryIntegrationService:
    """Production binary integration service."""

    def __init__(
        self,
        toolchain: AccessibilityToolchainService | None = None,
    ) -> None:
        self.toolchain = toolchain or AccessibilityToolchainService(
            timeout_seconds=300,
        )

    def discover_binary(self, binary_name: str) -> str | None:
        return shutil.which(binary_name)

    def run_daisy_ace(
        self,
        epub_path: str,
    ) -> dict:
        binary = self.discover_binary("ace")

        if not binary:
            return {
                "status": "unavailable",
                "reason": "DAISY Ace CLI binary not installed",
            }

        output_dir = tempfile.mkdtemp(prefix="ace-output-")

        command = [
            binary,
            epub_path,
            "--outdir",
            output_dir,
            "--force",
        ]

        result: ToolExecutionResult = self.toolchain.execute(
            tool_name="DAISY Ace",
            command=command,
            artifact_paths=[output_dir],
        )

        return {
            "execution": asdict(result),
            "output_directory": output_dir,
        }

    def run_epubcheck(
        self,
        epub_path: str,
    ) -> dict:
        binary = self.discover_binary("epubcheck")

        if not binary:
            return {
                "status": "unavailable",
                "reason": "EPUBCheck binary not installed",
            }

        report_path = Path(tempfile.mktemp(suffix=".xml"))

        command = [
            binary,
            epub_path,
            "-out",
            str(report_path),
        ]

        result: ToolExecutionResult = self.toolchain.execute(
            tool_name="EPUBCheck",
            command=command,
            artifact_paths=[str(report_path)],
        )

        return {
            "execution": asdict(result),
            "report_path": str(report_path),
        }

    def run_glow_audit(
        self,
        source_path: str,
    ) -> dict:
        """Audit a document with the GLOW (ACB Large Print Toolkit) CLI."""
        binary = self.discover_binary("acb-large-print")

        if not binary:
            return {
                "status": "unavailable",
                "reason": "GLOW (acb-large-print) CLI binary not installed",
            }

        report_path = Path(tempfile.mktemp(suffix=".txt"))

        command = [
            binary,
            "audit",
            source_path,
            "--format",
            "json",
            "-o",
            str(report_path),
        ]

        result: ToolExecutionResult = self.toolchain.execute(
            tool_name="GLOW (ACB Large Print)",
            command=command,
            artifact_paths=[str(report_path)],
        )

        return {
            "execution": asdict(result),
            "report_path": str(report_path),
        }

    def run_liblouis_translation(
        self,
        source_path: str,
        *,
        table: str = "en-ueb-g2.ctb",
        use_file2brl: bool = True,
    ) -> dict:
        """Translate *source_path* to BRF using Liblouis.

        Uses the real ``file2brl`` (preferred) or ``lou_translate`` binary.

        Parameters
        ----------
        source_path:
            Path to the plain-text or formatted source document.
        table:
            Liblouis braille table name (default: en-ueb-g2.ctb for UEB grade 2).
        use_file2brl:
            When True (default), use ``file2brl`` which handles formatting.
            When False, fall back to ``lou_translate`` for raw cell output.

        Returns a dict with keys ``status``, ``output_path``, and
        ``execution`` (the raw ToolExecutionResult dict).
        """
        # Prefer file2brl for full-document translation; fall back to lou_translate
        binary_name = "file2brl" if use_file2brl else "lou_translate"
        binary = self.discover_binary(binary_name)
        if not binary and use_file2brl:
            binary_name = "lou_translate"
            binary = self.discover_binary("lou_translate")

        if not binary:
            return {
                "status": "unavailable",
                "reason": (
                    "Neither 'file2brl' nor 'lou_translate' found on PATH. "
                    "Install liblouis (https://liblouis.io) and ensure the "
                    "binaries are accessible."
                ),
            }

        src  = Path(source_path)
        brf  = Path(tempfile.mkdtemp(prefix="liblouis-")) / (src.stem + ".brf")

        if binary_name == "file2brl":
            command = [binary, "-t", table, str(src), str(brf)]
        else:
            # lou_translate writes to stdout; redirect in the command template
            command = [binary, "-f", table, str(src)]

        result: ToolExecutionResult = self.toolchain.execute(
            tool_name="Liblouis",
            command=command,
            artifact_paths=[str(brf)],
        )

        # For lou_translate, stdout IS the BRF content
        if binary_name == "lou_translate" and result.status == "completed" and result.stdout:
            brf.write_text(result.stdout, encoding="utf-8")

        return {
            "status": result.status,
            "output_path": str(brf) if brf.exists() else None,
            "execution": asdict(result),
        }

run_glow_audit(source_path)

Audit a document with the GLOW (ACB Large Print Toolkit) CLI.

Source code in accessibility_mgr/services/toolchain_binaries.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def run_glow_audit(
    self,
    source_path: str,
) -> dict:
    """Audit a document with the GLOW (ACB Large Print Toolkit) CLI."""
    binary = self.discover_binary("acb-large-print")

    if not binary:
        return {
            "status": "unavailable",
            "reason": "GLOW (acb-large-print) CLI binary not installed",
        }

    report_path = Path(tempfile.mktemp(suffix=".txt"))

    command = [
        binary,
        "audit",
        source_path,
        "--format",
        "json",
        "-o",
        str(report_path),
    ]

    result: ToolExecutionResult = self.toolchain.execute(
        tool_name="GLOW (ACB Large Print)",
        command=command,
        artifact_paths=[str(report_path)],
    )

    return {
        "execution": asdict(result),
        "report_path": str(report_path),
    }

run_liblouis_translation(source_path, *, table='en-ueb-g2.ctb', use_file2brl=True)

Translate source_path to BRF using Liblouis.

Uses the real file2brl (preferred) or lou_translate binary.

Parameters

source_path: Path to the plain-text or formatted source document. table: Liblouis braille table name (default: en-ueb-g2.ctb for UEB grade 2). use_file2brl: When True (default), use file2brl which handles formatting. When False, fall back to lou_translate for raw cell output.

Returns a dict with keys status, output_path, and execution (the raw ToolExecutionResult dict).

Source code in accessibility_mgr/services/toolchain_binaries.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def run_liblouis_translation(
    self,
    source_path: str,
    *,
    table: str = "en-ueb-g2.ctb",
    use_file2brl: bool = True,
) -> dict:
    """Translate *source_path* to BRF using Liblouis.

    Uses the real ``file2brl`` (preferred) or ``lou_translate`` binary.

    Parameters
    ----------
    source_path:
        Path to the plain-text or formatted source document.
    table:
        Liblouis braille table name (default: en-ueb-g2.ctb for UEB grade 2).
    use_file2brl:
        When True (default), use ``file2brl`` which handles formatting.
        When False, fall back to ``lou_translate`` for raw cell output.

    Returns a dict with keys ``status``, ``output_path``, and
    ``execution`` (the raw ToolExecutionResult dict).
    """
    # Prefer file2brl for full-document translation; fall back to lou_translate
    binary_name = "file2brl" if use_file2brl else "lou_translate"
    binary = self.discover_binary(binary_name)
    if not binary and use_file2brl:
        binary_name = "lou_translate"
        binary = self.discover_binary("lou_translate")

    if not binary:
        return {
            "status": "unavailable",
            "reason": (
                "Neither 'file2brl' nor 'lou_translate' found on PATH. "
                "Install liblouis (https://liblouis.io) and ensure the "
                "binaries are accessible."
            ),
        }

    src  = Path(source_path)
    brf  = Path(tempfile.mkdtemp(prefix="liblouis-")) / (src.stem + ".brf")

    if binary_name == "file2brl":
        command = [binary, "-t", table, str(src), str(brf)]
    else:
        # lou_translate writes to stdout; redirect in the command template
        command = [binary, "-f", table, str(src)]

    result: ToolExecutionResult = self.toolchain.execute(
        tool_name="Liblouis",
        command=command,
        artifact_paths=[str(brf)],
    )

    # For lou_translate, stdout IS the BRF content
    if binary_name == "lou_translate" and result.status == "completed" and result.stdout:
        brf.write_text(result.stdout, encoding="utf-8")

    return {
        "status": result.status,
        "output_path": str(brf) if brf.exists() else None,
        "execution": asdict(result),
    }

Tools service

Purpose: tool-path resolution and command helper entrypoints.

Tools service — resolves external tool executables and augments PATH.

On import this module does nothing. Call bootstrap() (or init()) once at application startup to:

  1. Read tools.ini from the project root.
  2. Prepend any [paths] extra directories to os.environ["PATH"].
  3. Resolve each tool's executable via the config, then shutil.which().
  4. Cache resolved paths so the rest of the app can call resolve(name) without re-scanning.
Tool names recognised
  • "ace" → DAISY Ace
  • "epubcheck" → EPUBCheck
  • "pipeline" → DAISY Pipeline
  • "liblouis" → LibLouis CLI (lou_translate / file2brl / etc.)
  • "glow" → GLOW (ACB Large Print Toolkit, Community-Access)

bootstrap()

Read tools.ini, extend PATH, and cache resolved tool paths.

Safe to call multiple times — subsequent calls are no-ops.

Source code in accessibility_mgr/services/tools_service.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def bootstrap() -> None:
    """Read tools.ini, extend PATH, and cache resolved tool paths.

    Safe to call multiple times — subsequent calls are no-ops.
    """
    global _bootstrapped
    if _bootstrapped:
        return

    cfg = configparser.ConfigParser(default_section="DEFAULT")
    ini = _config_path()

    if ini.exists():
        cfg.read(ini)
        log.debug("tools_service: loaded %s", ini)
    else:
        log.warning(
            "tools_service: %s not found; using default tool names. "
            "Copy %s to tools.ini to customise.",
            ini,
            _example_config_path(),
        )

    # ── 1. Extend PATH with extra directories ─────────────────────────────
    raw_extra = cfg.get("paths", "extra", fallback="").strip()
    if raw_extra:
        extra_dirs = [d.strip() for d in raw_extra.splitlines() if d.strip()]
        if extra_dirs:
            current_path = os.environ.get("PATH", "")
            prepend = os.pathsep.join(extra_dirs)
            os.environ["PATH"] = prepend + os.pathsep + current_path
            log.info("tools_service: prepended to PATH: %s", prepend)

    # ── 2. Resolve each tool ───────────────────────────────────────────────
    for key, default_name in _DEFAULTS.items():
        configured = cfg.get("tools", key, fallback=default_name).strip()

        # If the user supplied an absolute path, use it directly.
        if os.path.isabs(configured):
            if os.path.isfile(configured) and os.access(configured, os.X_OK):
                _resolved[key] = configured
                log.info("tools_service: %s → %s (absolute path)", key, configured)
            else:
                _resolved[key] = None
                log.warning(
                    "tools_service: %s configured as '%s' but file not found or not executable.",
                    key,
                    configured,
                )
        else:
            # Bare name — search updated PATH.
            found = shutil.which(configured)
            if found:
                _resolved[key] = found
                log.info("tools_service: %s → %s", key, found)
            else:
                _resolved[key] = None
                log.warning(
                    "tools_service: '%s' (%s) not found on PATH. "
                    "Install the tool or set its path in tools.ini.",
                    configured,
                    key,
                )

    _bootstrapped = True

resolve(tool)

Return the resolved absolute path for tool, or None if not found.

Calls bootstrap() automatically on first use.

Example::

ace_bin = tools_service.resolve("ace")
if ace_bin is None:
    raise RuntimeError("DAISY Ace is not installed")
subprocess.run([ace_bin, "book.epub", "-o", "report"])
Source code in accessibility_mgr/services/tools_service.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def resolve(tool: str) -> str | None:
    """Return the resolved absolute path for *tool*, or None if not found.

    Calls ``bootstrap()`` automatically on first use.

    Example::

        ace_bin = tools_service.resolve("ace")
        if ace_bin is None:
            raise RuntimeError("DAISY Ace is not installed")
        subprocess.run([ace_bin, "book.epub", "-o", "report"])
    """
    if not _bootstrapped:
        bootstrap()
    return _resolved.get(tool)

status()

Return a copy of the resolved-tool map (useful for the Admin UI).

Source code in accessibility_mgr/services/tools_service.py 
146
147
148
149
150
def status() -> dict[str, str | None]:
    """Return a copy of the resolved-tool map (useful for the Admin UI)."""
    if not _bootstrapped:
        bootstrap()
    return dict(_resolved)

Worker runtime

Purpose: worker lifecycle and task execution runtime.

Background workflow worker runtime.

WorkerRuntime

Simple threaded worker runtime.

Foundation for: - distributed orchestration - SLA monitoring - retry execution - async workflow execution

Source code in accessibility_mgr/services/worker_runtime.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class WorkerRuntime:
    """Simple threaded worker runtime.

    Foundation for:
    - distributed orchestration
    - SLA monitoring
    - retry execution
    - async workflow execution
    """

    def __init__(
        self,
        queue_service: _QueueLike,
    ) -> None:
        self.queue_service = queue_service
        self._executions: list[WorkerExecution] = []
        self._running = False
        self._thread: threading.Thread | None = None

    def start(
        self,
        handler: Callable[[WorkflowJob], None],
        *,
        poll_interval: float = 1.0,
    ) -> None:
        if self._running:
            return

        self._running = True

        def _runner() -> None:
            while self._running:
                job = self.queue_service.next_job()

                if not job:
                    time.sleep(poll_interval)
                    continue

                execution = WorkerExecution(
                    worker_name="default-worker",
                    workflow_name=job.workflow_name,
                    asset_id=job.asset_id,
                    started_at=datetime.now(timezone.utc).isoformat(),
                    completed_at=None,
                    status="running",
                )

                self._executions.append(execution)

                try:
                    handler(job)
                    self.queue_service.complete_job(job)
                    execution.status = "completed"
                except Exception:
                    self.queue_service.fail_job(job)
                    execution.status = "failed"
                finally:
                    execution.completed_at = (
                        datetime.now(timezone.utc).isoformat()
                    )

        self._thread = threading.Thread(
            target=_runner,
            daemon=True,
        )
        self._thread.start()

    def stop(self) -> None:
        self._running = False

        if self._thread:
            self._thread.join(timeout=2)

    def list_executions(self) -> list[dict]:
        return [
            {
                "worker_name": execution.worker_name,
                "workflow_name": execution.workflow_name,
                "asset_id": execution.asset_id,
                "started_at": execution.started_at,
                "completed_at": execution.completed_at,
                "status": execution.status,
            }
            for execution in self._executions
        ]

Workflow DAG

Purpose: DAG representation for workflow steps and dependencies.

Dependency-aware workflow DAG orchestration — SQLite-backed.

WorkflowDAGService

SQLite-backed dependency-aware orchestration engine.

Source code in accessibility_mgr/services/workflow_dag.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class WorkflowDAGService:
    """SQLite-backed dependency-aware orchestration engine."""

    def __init__(self, database_path: Path | None = None) -> None:
        self.database_path = database_path or _default_db_path()
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(self.database_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _initialize(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS workflow_dag_node (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    workflow_name TEXT NOT NULL UNIQUE,
                    dependencies_json TEXT NOT NULL DEFAULT '[]',
                    retry_limit INTEGER NOT NULL DEFAULT 3,
                    status TEXT NOT NULL DEFAULT 'pending'
                )"""
            )

    def register_workflow(
        self,
        *,
        workflow_name: str,
        dependencies: list[str] | None = None,
        retry_limit: int = 3,
    ) -> dict[str, Any]:
        deps = dependencies or []
        with self._connect() as conn:
            conn.execute(
                "INSERT OR REPLACE INTO workflow_dag_node (workflow_name, dependencies_json, retry_limit, status) "
                "VALUES (?, ?, ?, 'pending')",
                (workflow_name, json.dumps(deps), retry_limit),
            )
        return {
            "workflow_name": workflow_name,
            "dependencies": deps,
            "retry_limit": retry_limit,
            "status": "pending",
        }

    def executable_workflows(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            rows = conn.execute(
                "SELECT * FROM workflow_dag_node WHERE status = 'pending'"
            ).fetchall()
            executable = []
            for row in rows:
                deps = json.loads(row["dependencies_json"])
                blocked = False
                for dep_name in deps:
                    dep = conn.execute(
                        "SELECT status FROM workflow_dag_node WHERE workflow_name = ?",
                        (dep_name,),
                    ).fetchone()
                    if dep and dep["status"] != "completed":
                        blocked = True
                        break
                if not blocked:
                    executable.append({
                        "workflow_name": row["workflow_name"],
                        "dependencies": deps,
                        "retry_limit": row["retry_limit"],
                        "status": row["status"],
                    })
            return executable

    def complete(self, workflow_name: str) -> None:
        with self._connect() as conn:
            conn.execute(
                "UPDATE workflow_dag_node SET status = 'completed' WHERE workflow_name = ?",
                (workflow_name,),
            )

    def fail(self, workflow_name: str) -> None:
        with self._connect() as conn:
            conn.execute(
                "UPDATE workflow_dag_node SET status = 'failed' WHERE workflow_name = ?",
                (workflow_name,),
            )

    def topology(self) -> list[dict[str, Any]]:
        with self._connect() as conn:
            return [
                {
                    "workflow_name": r["workflow_name"],
                    "dependencies": json.loads(r["dependencies_json"]),
                    "retry_limit": r["retry_limit"],
                    "status": r["status"],
                }
                for r in conn.execute(
                    "SELECT * FROM workflow_dag_node"
                ).fetchall()
            ]

Workflow queue

Purpose: workflow job data structures.

Workflow queue primitives and in-memory queue service.

Provides: - WorkflowJob dataclass shared across queue implementations. - WorkflowQueueService in-memory priority queue that implements the _QueueLike protocol used by WorkerRuntime.

The SQLite-backed PersistentWorkflowQueue in persistent_queue.py is the production-grade replacement; this in-memory version is useful for tests, lightweight deployments, and as a reference implementation.

WorkflowJob dataclass

Represents a single queued workflow execution request.

Source code in accessibility_mgr/services/workflow_queue.py 
22
23
24
25
26
27
28
29
30
@dataclass(slots=True)
class WorkflowJob:
    """Represents a single queued workflow execution request."""

    workflow_name: str
    asset_id: int
    priority: int
    status: str
    created_at: str

WorkflowQueueService

In-memory priority queue for workflow execution requests.

Thread-safe. Implements the _QueueLike protocol expected by WorkerRuntime: next_job() / complete_job() / fail_job() / list_jobs().

Source code in accessibility_mgr/services/workflow_queue.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class WorkflowQueueService:
    """In-memory priority queue for workflow execution requests.

    Thread-safe.  Implements the ``_QueueLike`` protocol expected by
    ``WorkerRuntime``: ``next_job()`` / ``complete_job()`` / ``fail_job()`` /
    ``list_jobs()``.
    """

    def __init__(self) -> None:
        self._queue: list[tuple[int, str, WorkflowJob]] = []
        self._counter = 0
        self._lock = threading.Lock()

    def enqueue(
        self,
        *,
        workflow_name: str,
        asset_id: int,
        priority: int = 5,
    ) -> WorkflowJob:
        """Add a new job to the queue and return it."""
        created_at = datetime.now(UTC).isoformat()
        job = WorkflowJob(
            workflow_name=workflow_name,
            asset_id=asset_id,
            priority=priority,
            status="queued",
            created_at=created_at,
        )
        with self._lock:
            self._counter += 1
            heapq.heappush(
                self._queue,
                (priority, self._counter, job),
            )
        return job

    def next_job(self) -> WorkflowJob | None:
        """Pop and return the highest-priority (lowest number) queued job.

        Returns ``None`` when the queue is empty.
        """
        with self._lock:
            while self._queue:
                _pri, _seq, job = heapq.heappop(self._queue)
                if job.status == "queued":
                    job.status = "running"
                    return job
        return None

    def complete_job(self, job: WorkflowJob) -> None:
        """Mark *job* as completed."""
        job.status = "completed"

    def fail_job(self, job: WorkflowJob) -> None:
        """Mark *job* as failed."""
        job.status = "failed"

    def list_jobs(self) -> list[dict[str, Any]]:
        """Return all jobs (queued, running, completed, failed) in insertion order."""
        with self._lock:
            return [
                {
                    "workflow_name": j.workflow_name,
                    "asset_id": j.asset_id,
                    "priority": j.priority,
                    "status": j.status,
                    "created_at": j.created_at,
                }
                for _pri, _seq, j in sorted(self._queue, key=lambda x: x[1])
            ]

complete_job(job)

Mark job as completed.

Source code in accessibility_mgr/services/workflow_queue.py 
83
84
85
def complete_job(self, job: WorkflowJob) -> None:
    """Mark *job* as completed."""
    job.status = "completed"

enqueue(*, workflow_name, asset_id, priority=5)

Add a new job to the queue and return it.

Source code in accessibility_mgr/services/workflow_queue.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def enqueue(
    self,
    *,
    workflow_name: str,
    asset_id: int,
    priority: int = 5,
) -> WorkflowJob:
    """Add a new job to the queue and return it."""
    created_at = datetime.now(UTC).isoformat()
    job = WorkflowJob(
        workflow_name=workflow_name,
        asset_id=asset_id,
        priority=priority,
        status="queued",
        created_at=created_at,
    )
    with self._lock:
        self._counter += 1
        heapq.heappush(
            self._queue,
            (priority, self._counter, job),
        )
    return job

fail_job(job)

Mark job as failed.

Source code in accessibility_mgr/services/workflow_queue.py 
87
88
89
def fail_job(self, job: WorkflowJob) -> None:
    """Mark *job* as failed."""
    job.status = "failed"

list_jobs()

Return all jobs (queued, running, completed, failed) in insertion order.

Source code in accessibility_mgr/services/workflow_queue.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def list_jobs(self) -> list[dict[str, Any]]:
    """Return all jobs (queued, running, completed, failed) in insertion order."""
    with self._lock:
        return [
            {
                "workflow_name": j.workflow_name,
                "asset_id": j.asset_id,
                "priority": j.priority,
                "status": j.status,
                "created_at": j.created_at,
            }
            for _pri, _seq, j in sorted(self._queue, key=lambda x: x[1])
        ]

next_job()

Pop and return the highest-priority (lowest number) queued job.

Returns None when the queue is empty.

Source code in accessibility_mgr/services/workflow_queue.py 
70
71
72
73
74
75
76
77
78
79
80
81
def next_job(self) -> WorkflowJob | None:
    """Pop and return the highest-priority (lowest number) queued job.

    Returns ``None`` when the queue is empty.
    """
    with self._lock:
        while self._queue:
            _pri, _seq, job = heapq.heappop(self._queue)
            if job.status == "queued":
                job.status = "running"
                return job
    return None