feat: persist isolated utility export jobs

config.yaml

@@ -48,6 +48,11 @@ downloads:
   max_workers: 2
   process_timeout_seconds: 7200
   max_height: 1080
+  # Utility exports live below media_base and never participate in slots.
+  export_directory: exports
+  export_max_height: 1080
+  # Bounds queued + running utility work; Player slots retain their own bound.
+  max_queued_exports: 20
 
 router:
   required_mode: video

controller/config.py

@@ -63,6 +63,9 @@ class DownloadConfig:
     max_workers: int = 2
     process_timeout_seconds: float = 2 * 3600.0
     max_height: int = 1080
+    export_directory: Path = Path("exports")
+    export_max_height: int = 1080
+    max_queued_exports: int = 20
 
 
 @dataclass(frozen=True)
@@ -172,6 +175,20 @@ def _optional_path(value: Any, name: str) -> Path | None:
     return Path(text)
 
 
+def _relative_directory(value: Any, name: str) -> Path:
+    text = _text(value, name)
+    assert text is not None
+    path = Path(text)
+    if (
+        path.is_absolute()
+        or not path.parts
+        or any(part in {".", ".."} for part in path.parts)
+        or "\\" in text
+    ):
+        raise ConfigError(f"{name} must be a safe relative directory")
+    return path
+
+
 def _env_value(env: Mapping[str, str], key: str, fallback: Any) -> Any:
     return env[key] if key in env else fallback
 
@@ -277,6 +294,19 @@ def load_config(
         max_workers=_integer(downloads_raw.get("max_workers", 2), "downloads.max_workers", maximum=10),
         process_timeout_seconds=_number(downloads_raw.get("process_timeout_seconds", 2 * 3600), "downloads.process_timeout_seconds", minimum=1.0),
         max_height=_integer(downloads_raw.get("max_height", 1080), "downloads.max_height"),
+        export_directory=_relative_directory(
+            downloads_raw.get("export_directory", "exports"),
+            "downloads.export_directory",
+        ),
+        export_max_height=_integer(
+            downloads_raw.get("export_max_height", 1080),
+            "downloads.export_max_height",
+        ),
+        max_queued_exports=_integer(
+            downloads_raw.get("max_queued_exports", 20),
+            "downloads.max_queued_exports",
+            maximum=1000,
+        ),
     )
     if downloads.cookie_required and downloads.cookie_path is None:
         raise ConfigError("downloads.cookie_path is required when cookie_required is true")

controller/models.py

@@ -10,6 +10,7 @@ MEDIA_STATES = frozenset(
 )
 LIVE_JOB_STATES = frozenset({"reserved", "running"})
 LIVE_GUARD_STATES = frozenset({"active", "review_required"})
+UTILITY_EXPORT_MODES = frozenset({"mp4", "m4a", "mp3"})
 
 
 @dataclass(frozen=True)
@@ -54,6 +55,27 @@ class DownloadJob:
     error: str | None
 
 
+@dataclass(frozen=True)
+class ExportReservation:
+    status: str
+    job_id: str | None = None
+
+
+@dataclass(frozen=True)
+class ExportJob:
+    job_id: str
+    mode: str
+    url: str
+    status: str
+    submitter: str | None
+    note: str | None
+    requested_at: float
+    started_at: float | None
+    finished_at: float | None
+    output_paths: tuple[str, ...]
+    error: str | None
+
+
 @dataclass(frozen=True)
 class DeleteToken:
     slot: int

controller/store.py

@@ -25,6 +25,8 @@ from .models import (
     DeleteToken,
     DownloadJob,
     DownloadReservation,
+    ExportJob,
+    ExportReservation,
     MigrationError,
     MigrationReport,
     PlayerObservation,
@@ -33,10 +35,11 @@ from .models import (
     SlotProtectedError,
     SlotRecord,
     StaleOperationError,
+    UTILITY_EXPORT_MODES,
 )
 
 
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 LEGACY_MIGRATION_NAME = "legacy_playlists_yaml_v1"
 MAX_LEGACY_YAML_BYTES = 8 * 1024 * 1024
 
@@ -104,6 +107,24 @@ CREATE UNIQUE INDEX IF NOT EXISTS one_live_download_per_slot
 ON download_jobs(slot_number)
 WHERE status IN ('reserved','running');
 
+CREATE TABLE IF NOT EXISTS export_jobs (
+    job_id TEXT PRIMARY KEY,
+    mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')),
+    url TEXT NOT NULL,
+    status TEXT NOT NULL CHECK (status IN
+        ('queued','running','completed','failed','interrupted')),
+    submitter TEXT,
+    note TEXT,
+    requested_at REAL NOT NULL,
+    started_at REAL,
+    finished_at REAL,
+    output_paths_json TEXT NOT NULL DEFAULT '[]',
+    error TEXT
+);
+
+CREATE INDEX IF NOT EXISTS export_jobs_recent
+ON export_jobs(requested_at DESC, job_id DESC);
+
 CREATE TABLE IF NOT EXISTS slot_guards (
     guard_id TEXT PRIMARY KEY,
     slot_number INTEGER NOT NULL REFERENCES slots(slot_number),
@@ -401,6 +422,13 @@ class ControllerStore:
             """,
             (now,),
         )
+        connection.execute(
+            """
+            INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
+            VALUES (3, 'utility_export_jobs', ?)
+            """,
+            (now,),
+        )
         connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
 
     def _upgrade_database(self, connection: sqlite3.Connection) -> None:
@@ -409,12 +437,13 @@ class ControllerStore:
         version = connection.execute("PRAGMA user_version").fetchone()[0]
         if version == SCHEMA_VERSION:
             return
-        if version != 1:
+        if version not in {1, 2}:
             raise MigrationError(
-                f"unsupported Controller database schema {version}; expected 1 or {SCHEMA_VERSION}"
+                f"unsupported Controller database schema {version}; expected 1, 2, or {SCHEMA_VERSION}"
             )
         connection.execute("BEGIN IMMEDIATE")
         try:
+            if version < 2:
                 columns = {
                     row["name"]
                     for row in connection.execute(
@@ -432,6 +461,37 @@ class ControllerStore:
                     """,
                     (self._now(),),
                 )
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS export_jobs (
+                    job_id TEXT PRIMARY KEY,
+                    mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')),
+                    url TEXT NOT NULL,
+                    status TEXT NOT NULL CHECK (status IN
+                        ('queued','running','completed','failed','interrupted')),
+                    submitter TEXT,
+                    note TEXT,
+                    requested_at REAL NOT NULL,
+                    started_at REAL,
+                    finished_at REAL,
+                    output_paths_json TEXT NOT NULL DEFAULT '[]',
+                    error TEXT
+                )
+                """
+            )
+            connection.execute(
+                """
+                CREATE INDEX IF NOT EXISTS export_jobs_recent
+                ON export_jobs(requested_at DESC, job_id DESC)
+                """
+            )
+            connection.execute(
+                """
+                INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
+                VALUES (3, 'utility_export_jobs', ?)
+                """,
+                (self._now(),),
+            )
             connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
         except BaseException:
             connection.rollback()
@@ -837,6 +897,15 @@ class ControllerStore:
                         row["generation"],
                     ),
                 )
+            connection.execute(
+                """
+                UPDATE export_jobs
+                SET status = 'interrupted', finished_at = ?,
+                    error = 'export interrupted by Controller restart'
+                WHERE status IN ('queued','running')
+                """,
+                (now,),
+            )
 
     def list_slots(self) -> list[SlotRecord]:
         with self._lock:
@@ -1136,6 +1205,133 @@ class ControllerStore:
                 error=row["error"],
             )
 
+    @staticmethod
+    def _row_to_export_job(row: sqlite3.Row) -> ExportJob:
+        try:
+            raw_paths = json.loads(row["output_paths_json"])
+        except (TypeError, ValueError, json.JSONDecodeError):
+            raw_paths = []
+        output_paths = tuple(
+            value for value in raw_paths if isinstance(value, str)
+        ) if isinstance(raw_paths, list) else ()
+        return ExportJob(
+            job_id=row["job_id"],
+            mode=row["mode"],
+            url=row["url"],
+            status=row["status"],
+            submitter=row["submitter"],
+            note=row["note"],
+            requested_at=row["requested_at"],
+            started_at=row["started_at"],
+            finished_at=row["finished_at"],
+            output_paths=output_paths,
+            error=row["error"],
+        )
+
+    def reserve_export_job(
+        self,
+        url: str,
+        mode: str,
+        submitter: str | None,
+        note: str | None,
+        *,
+        max_active: int,
+        job_id: str | None = None,
+    ) -> ExportReservation:
+        if not isinstance(url, str) or not url.strip():
+            raise ValueError("export URL must be non-empty")
+        if mode not in UTILITY_EXPORT_MODES:
+            raise ValueError("unsupported utility export mode")
+        if not isinstance(max_active, int) or isinstance(max_active, bool) or max_active < 1:
+            raise ValueError("max_active must be a positive integer")
+        job_id = job_id or uuid.uuid4().hex
+        with self._transaction() as connection:
+            active = connection.execute(
+                """
+                SELECT COUNT(*) FROM export_jobs
+                WHERE status IN ('queued','running')
+                """
+            ).fetchone()[0]
+            if active >= max_active:
+                return ExportReservation(status="queue_full")
+            connection.execute(
+                """
+                INSERT INTO export_jobs
+                    (job_id, mode, url, status, submitter, note, requested_at)
+                VALUES (?, ?, ?, 'queued', ?, ?, ?)
+                """,
+                (job_id, mode, url.strip(), submitter, note, self._now()),
+            )
+            return ExportReservation(status="queued", job_id=job_id)
+
+    def mark_export_running(self, job_id: str) -> bool:
+        with self._transaction() as connection:
+            return bool(
+                connection.execute(
+                    """
+                    UPDATE export_jobs SET status = 'running', started_at = ?
+                    WHERE job_id = ? AND status = 'queued'
+                    """,
+                    (self._now(), job_id),
+                ).rowcount
+            )
+
+    def complete_export_job(
+        self,
+        job_id: str,
+        output_paths: Sequence[str],
+    ) -> bool:
+        normalized = tuple(output_paths)
+        if not normalized or any(
+            not isinstance(path, str) or not path for path in normalized
+        ):
+            raise ValueError("completed export must contain output paths")
+        with self._transaction() as connection:
+            return bool(
+                connection.execute(
+                    """
+                    UPDATE export_jobs
+                    SET status = 'completed', finished_at = ?,
+                        output_paths_json = ?, error = NULL
+                    WHERE job_id = ? AND status = 'running'
+                    """,
+                    (self._now(), json.dumps(normalized), job_id),
+                ).rowcount
+            )
+
+    def fail_export_job(self, job_id: str, error: str) -> bool:
+        with self._transaction() as connection:
+            return bool(
+                connection.execute(
+                    """
+                    UPDATE export_jobs
+                    SET status = 'failed', finished_at = ?, error = ?
+                    WHERE job_id = ? AND status IN ('queued','running')
+                    """,
+                    (self._now(), error, job_id),
+                ).rowcount
+            )
+
+    def get_export_job(self, job_id: str) -> ExportJob | None:
+        with self._lock:
+            row = self.connection.execute(
+                "SELECT * FROM export_jobs WHERE job_id = ?", (job_id,)
+            ).fetchone()
+            return None if row is None else self._row_to_export_job(row)
+
+    def list_export_jobs(self, *, limit: int = 20) -> list[ExportJob]:
+        if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 100:
+            raise ValueError("export job limit must be between 1 and 100")
+        with self._lock:
+            rows = self.connection.execute(
+                """
+                SELECT * FROM export_jobs
+                ORDER BY requested_at DESC, job_id DESC LIMIT ?
+                """,
+                (limit,),
+            ).fetchall()
+            return [self._row_to_export_job(row) for row in rows]
+
     def protect_slot(
         self,
         slot: int,

tests/test_config_contracts.py

@@ -82,6 +82,9 @@ class ConfigurationTests(unittest.TestCase):
         self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0)
         self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0)
         self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0)
+        self.assertEqual(config.downloads.export_directory, Path("exports"))
+        self.assertEqual(config.downloads.export_max_height, 1080)
+        self.assertEqual(config.downloads.max_queued_exports, 20)
 
     def test_documented_environment_overrides_apply(self):
         env = {
@@ -175,6 +178,34 @@ class ConfigurationTests(unittest.TestCase):
                 with self.assertRaisesRegex(ConfigError, f"{field} must be an absolute path"):
                     load_config(path, environ={})
 
+    def test_export_directory_must_stay_relative_to_media_root(self):
+        invalid = ("/tmp/exports", "../exports", "nested/../exports", r"nested\\exports")
+        for value in invalid:
+            with self.subTest(value=value), tempfile.TemporaryDirectory() as directory:
+                path = Path(directory) / "config.yaml"
+                path.write_text(
+                    f"downloads:\n  export_directory: {value!r}\n",
+                    encoding="utf-8",
+                )
+                with self.assertRaisesRegex(ConfigError, "safe relative directory"):
+                    load_config(path, environ={})
+
+    def test_export_limits_are_bounded_positive_integers(self):
+        cases = (
+            ("export_max_height", 0),
+            ("max_queued_exports", 0),
+            ("max_queued_exports", 1001),
+        )
+        for field, value in cases:
+            with self.subTest(field=field), tempfile.TemporaryDirectory() as directory:
+                path = Path(directory) / "config.yaml"
+                path.write_text(
+                    f"downloads:\n  {field}: {value}\n",
+                    encoding="utf-8",
+                )
+                with self.assertRaisesRegex(ConfigError, field):
+                    load_config(path, environ={})
+
 
 class ImportLifecycleTests(unittest.TestCase):
     def test_controller_core_import_does_not_connect_or_create_runtime_paths(self):

tests/test_store.py

@@ -50,7 +50,7 @@ class FreshStoreTests(StoreTestCase):
             self.assertTrue(all(slot.media_state == "empty" for slot in slots))
             self.assertEqual(store.player_observation().phase, "unknown")
 
-    def test_schema_one_database_upgrades_external_reconciliation_atomically(self):
+    def test_schema_one_database_upgrades_reconciliation_and_exports_atomically(self):
         with self.open_store() as store:
             database_path = store.path
         connection = sqlite3.connect(database_path)
@@ -67,7 +67,7 @@ class FreshStoreTests(StoreTestCase):
         with self.open_store() as upgraded:
             self.assertEqual(
                 upgraded.connection.execute("PRAGMA user_version").fetchone()[0],
-                2,
+                3,
             )
             columns = {
                 row["name"]
@@ -80,6 +80,131 @@ class FreshStoreTests(StoreTestCase):
                 "SELECT name FROM schema_migrations WHERE version = 2"
             ).fetchone()
             self.assertEqual(migration["name"], "external_playback_reconciliation")
+            export_migration = upgraded.connection.execute(
+                "SELECT name FROM schema_migrations WHERE version = 3"
+            ).fetchone()
+            self.assertEqual(export_migration["name"], "utility_export_jobs")
+            tables = {
+                row["name"]
+                for row in upgraded.connection.execute(
+                    "SELECT name FROM sqlite_master WHERE type = 'table'"
+                ).fetchall()
+            }
+            self.assertIn("export_jobs", tables)
+
+    def test_schema_two_database_adds_only_utility_export_jobs(self):
+        with self.open_store() as store:
+            database_path = store.path
+        connection = sqlite3.connect(database_path)
+        try:
+            connection.execute("DROP TABLE export_jobs")
+            connection.execute("DELETE FROM schema_migrations WHERE version = 3")
+            connection.execute("PRAGMA user_version = 2")
+            connection.commit()
+        finally:
+            connection.close()
+
+        with self.open_store() as upgraded:
+            self.assertEqual(
+                upgraded.connection.execute("PRAGMA user_version").fetchone()[0],
+                3,
+            )
+            self.assertEqual(upgraded.list_export_jobs(), [])
+
+    def test_export_jobs_are_slot_free_bounded_and_durable(self):
+        with self.open_store() as store:
+            first = store.reserve_export_job(
+                "https://www.youtube.com/watch?v=one",
+                "mp4",
+                "Alice",
+                "Sunday",
+                max_active=1,
+                job_id="export-one",
+            )
+            self.assertEqual(first.status, "queued")
+            self.assertEqual(first.job_id, "export-one")
+            self.assertEqual(
+                store.reserve_export_job(
+                    "https://www.youtube.com/watch?v=two",
+                    "mp3",
+                    None,
+                    None,
+                    max_active=1,
+                ).status,
+                "queue_full",
+            )
+            self.assertTrue(store.mark_export_running("export-one"))
+            self.assertTrue(
+                store.complete_export_job(
+                    "export-one",
+                    ("exports/video/export-one/Service.mp4",),
+                )
+            )
+            job = store.get_export_job("export-one")
+            self.assertEqual(job.mode, "mp4")
+            self.assertEqual(job.status, "completed")
+            self.assertEqual(job.submitter, "Alice")
+            self.assertEqual(
+                job.output_paths,
+                ("exports/video/export-one/Service.mp4",),
+            )
+            self.assertTrue(all(slot.media_state == "empty" for slot in store.list_slots()))
+
+    def test_restart_marks_live_exports_interrupted_without_touching_slots(self):
+        store = self.open_store()
+        store.reserve_export_job(
+            "https://www.youtube.com/playlist?list=one",
+            "m4a",
+            None,
+            None,
+            max_active=2,
+            job_id="queued-export",
+        )
+        store.reserve_export_job(
+            "https://www.youtube.com/watch?v=two",
+            "mp3",
+            None,
+            None,
+            max_active=2,
+            job_id="running-export",
+        )
+        store.mark_export_running("running-export")
+        store.close()
+
+        with self.open_store() as reopened:
+            jobs = {job.job_id: job for job in reopened.list_export_jobs()}
+            self.assertEqual(jobs["queued-export"].status, "interrupted")
+            self.assertEqual(jobs["running-export"].status, "interrupted")
+            self.assertTrue(
+                all("Controller restart" in job.error for job in jobs.values())
+            )
+            self.assertTrue(
+                all(slot.media_state == "empty" for slot in reopened.list_slots())
+            )
+
+    def test_export_failures_do_not_corrupt_player_inventory(self):
+        with self.open_store() as store:
+            slot = store.reserve_download(
+                "https://www.youtube.com/watch?v=managed",
+                None,
+                None,
+                job_id="managed",
+            )
+            store.fail_download(
+                "managed", slot.slot, slot.generation, "managed failure"
+            )
+            before = store.get_slot(slot.slot)
+            export = store.reserve_export_job(
+                "https://www.youtube.com/watch?v=export",
+                "mp3",
+                None,
+                None,
+                max_active=5,
+                job_id="failed-export",
+            )
+            self.assertEqual(export.status, "queued")
+            self.assertTrue(store.fail_export_job("failed-export", "ffmpeg failed"))
+            self.assertEqual(store.get_slot(slot.slot), before)
 
     def test_slot_reservation_is_committed_before_worker_transition(self):
         with self.open_store() as store: