' not in workspace
+ assert "1 · Entorno y dirección visual" in workspace
+ assert "Protagonistas y antagonistas" in workspace
+ assert "Arco y momentos de tráiler" in workspace
+
+
+def test_trailer_orientation_can_override_the_global_landscape_default_inline():
+ panel = PANEL.read_text(encoding="utf-8")
+ trailer = panel.split("{tab === 'trailer'", 1)[1].split("{tab === 'productions'", 1)[0]
+ handler = panel.split("const setStoryVideoFormat", 1)[1].split("useEffect", 1)[0]
+
+ assert "Portrait / Shorts" in panel
+ assert "disabled={!storyVideoOptionsReady}" in trailer
+ assert "provider: { ...project.provider, useGlobalProfile: false }" in handler
+ assert "if (project.provider.useGlobalProfile) return" not in handler
+ assert "Formato seleccionado" in panel
+ assert "aria-pressed={aspectRatio === option.value}" in panel
+ assert "Formato de vídeo actualizado:" in handler
+
+
+def test_trailer_supports_text_only_direct_video_without_visual_inputs():
+ panel = PANEL.read_text(encoding="utf-8")
+ store = STORE.read_text(encoding="utf-8")
+ director_chat = DIRECTOR_CHAT.read_text(encoding="utf-8")
+ pipeline = ROOT.joinpath("app", "services", "director_pipeline.py").read_text(encoding="utf-8")
+
+ trailer = panel.split("{tab === 'trailer'", 1)[1].split("{tab === 'productions'", 1)[0]
+ assert "Vídeo directo" in trailer
+ assert "T2V · sin imágenes" in trailer
+ assert "musicVideoGenerationMode: 'direct_video', protagonistConsistency: false" in trailer
+ assert "directVideoMasterReady" in trailer
+ assert "disabled={directVideo || directReferenceVideo}" in trailer
+ assert "const directVideo = state.directorMusicVideoTreatment.generation_mode === 'direct_video'" in store
+ assert "pipelineType === 'music_video' || directVideo" in store
+ assert "const isDirectVideo = musicVideoTreatment.generation_mode === 'direct_video'" in director_chat
+ assert '"music_video", "short_film_story"' in pipeline
+
+
+def test_direct_trailer_cast_approval_does_not_require_identity_images():
+ panel = PANEL.read_text(encoding="utf-8")
+ approval = panel.split("const approve =", 1)[1].split("const isApproved", 1)[0]
+
+ assert "const requiresVisualIdentities = !directVideo" in approval
+ assert "Character descriptions approved. Direct-video mode does not require identity images." in approval
+ assert "project.projectType === 'trailer'" in panel
+ assert "? trailerProductionIssues" in panel
+ assert "requiresVisualIdentities={!directVideo}" in panel
+
+
+def test_trailer_can_review_generate_reopen_and_reuse_ordered_assembly():
+ panel = PANEL.read_text(encoding="utf-8")
+ stage = panel.split("const stageTrailer", 1)[1].split("const writeStorySong", 1)[0]
+ reopen = panel.split("const reopenProduction", 1)[1].split(
+ "const restoreProductionSource", 1,
+ )[0]
+
+ assert "buildTrailerAdaptation" in panel
+ assert "kind: 'trailer'" in stage
+ assert "pipelineId: useStore.getState().pipelineId" in stage
+ assert "stageTrailer(true)" in panel
+ assert "stageTrailer(false)" in panel
+ assert "production.kind === 'trailer'" in reopen
+ assert "trailerOptions" in reopen
diff --git a/tests/test_story_montage_clip_history_ui.py b/tests/test_story_montage_clip_history_ui.py
new file mode 100644
index 00000000..3d48f06a
--- /dev/null
+++ b/tests/test_story_montage_clip_history_ui.py
@@ -0,0 +1,49 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+TIMELINE = ROOT / "ui" / "src" / "features" / "stories" / "StoryProductionTimeline.tsx"
+HANDOFF = ROOT / "ui" / "src" / "features" / "stories" / "directorClipHandoff.ts"
+MEDIA = ROOT / "ui" / "src" / "components" / "MainContent" / "MediaFeedItem.tsx"
+MAIN = ROOT / "ui" / "src" / "components" / "MainContent" / "MainContent.tsx"
+CLIENT = ROOT / "ui" / "src" / "api" / "client.ts"
+STORE = ROOT / "ui" / "src" / "stores" / "useStore.ts"
+
+
+def test_story_montage_exposes_slot_history_and_explicit_remake_action():
+ timeline = TIMELINE.read_text(encoding="utf-8")
+
+ assert "Historial de esta posición" in timeline
+ assert "En montaje:" in timeline
+ assert "Rehacer este clip" in timeline
+ assert "selectPipelineClipVideo" in timeline
+ assert "directorClipCreatorMetadata" in timeline
+ assert "writeDirectorClipReplacementTarget" in timeline
+ assert "fetchOutputMetadata(" in timeline
+ assert "switchWorkspace(targetWorkspace)" in timeline
+
+
+def test_creator_handoff_reduces_multiclip_metadata_to_one_exact_slot():
+ handoff = HANDOFF.read_text(encoding="utf-8")
+ store = STORE.read_text(encoding="utf-8")
+
+ assert "perClipFrames[clip.index]" in handoff
+ assert "source.per_clip_minimax_h3_references" in handoff
+ assert "params.prompt = attempt.prompt || clip.video_prompt" in handoff
+ assert "delete params[key]" in handoff
+ assert "params.repeat_generation = 1" in handoff
+ assert "newParams.minimax_h3_references" in store
+ assert "newParams.h3_model_profile" in store
+
+
+def test_generated_video_can_be_selected_and_returns_to_story_montage():
+ media = MEDIA.read_text(encoding="utf-8")
+ main = MAIN.read_text(encoding="utf-8")
+ client = CLIENT.read_text(encoding="utf-8")
+
+ assert "Usar en Montaje · clip" in media
+ assert "writeDirectorClipReplacementResult" in media
+ assert "setMediaFilter('stories')" in media
+ assert "Ajusta sus datos, genera una o varias versiones" in main
+ assert "Cancelar reemplazo" in main
+ assert "/video-selection" in client
diff --git a/tests/test_style_library.py b/tests/test_style_library.py
new file mode 100644
index 00000000..c51094b4
--- /dev/null
+++ b/tests/test_style_library.py
@@ -0,0 +1,99 @@
+import json
+
+import pytest
+from fastapi import HTTPException
+
+from routers.style_library import create_style_library_router
+from services.style_library import MINIMAX_H3_1K_SOURCE, StyleLibrary
+
+
+def _seed_library(tmp_path):
+ library = StyleLibrary(tmp_path / "styles")
+ library.raw_dir.mkdir(parents=True)
+ library.preview_dir.mkdir(parents=True)
+ source = {
+ **MINIMAX_H3_1K_SOURCE,
+ "revision": "revision-123",
+ "lastModified": "2026-08-10T02:58:09Z",
+ }
+ records = []
+ for number, prompt, group in (
+ (1, "Cinematic rain over a neon city", "Cinematic"),
+ (2, "Flat-color animated comedy", "Animation"),
+ (3, "Documentary wildlife close-up", "Documentary"),
+ ):
+ sample = f"{number:06d}"
+ style_id = f"minimax-h3-1k-{sample}"
+ (library.raw_dir / f"{sample}.txt").write_text(prompt, encoding="utf-8")
+ (library.raw_dir / f"{sample}.mp4").write_bytes(b"video")
+ (library.preview_dir / f"{style_id}.jpg").write_bytes(b"preview")
+ records.append({
+ "id": style_id,
+ "modelFamily": "minimax",
+ "title": f"Sample {sample}",
+ "prompt": prompt,
+ "collection": "MiniMax H3 1K",
+ "group": group,
+ "tags": [group.casefold()],
+ "sourceOrder": number,
+ "sourceFilename": f"{sample}.txt",
+ "videoFilename": f"{sample}.mp4",
+ "source": source,
+ "importedAt": 100 + number,
+ })
+ library.manifest_path.write_text(json.dumps({
+ "version": 1,
+ "source": source,
+ "styles": records,
+ "deletedIds": [],
+ "updatedAt": 200,
+ }), encoding="utf-8")
+ return library
+
+
+def test_styles_keep_source_attribution_and_support_filters_and_sorting(tmp_path):
+ library = _seed_library(tmp_path)
+
+ result = library.list_styles(
+ model_family="minimax",
+ collection="MiniMax H3 1K",
+ group="Animation",
+ query="comedy",
+ sort="prompt_asc",
+ )
+
+ assert result["total"] == 1
+ [style] = result["styles"]
+ assert style["source"]["id"] == "huggingface:ostris/minimax_h3_1k"
+ assert style["source"]["revision"] == "revision-123"
+ assert style["source"]["license"] is None
+ assert style["previewUrl"].endswith(f"/{style['id']}/preview")
+ assert result["facets"]["groups"] == ["Animation"]
+
+
+def test_style_deletion_is_tombstoned_and_removes_local_assets(tmp_path):
+ library = _seed_library(tmp_path)
+ style_id = "minimax-h3-1k-000002"
+
+ result = library.delete_style(style_id)
+
+ assert result["deleted"] is True
+ assert library.list_styles()["total"] == 2
+ manifest = json.loads(library.manifest_path.read_text(encoding="utf-8"))
+ assert style_id in manifest["deletedIds"]
+ assert not (library.raw_dir / "000002.mp4").exists()
+ assert not (library.raw_dir / "000002.txt").exists()
+ assert not (library.preview_dir / f"{style_id}.jpg").exists()
+
+
+def test_delete_endpoint_requires_explicit_confirmation(tmp_path):
+ library = _seed_library(tmp_path)
+ router = create_style_library_router(library)
+ endpoints = {route.path: route.endpoint for route in router.routes}
+ delete_endpoint = endpoints["/api/v1/style-library/styles/{style_id}"]
+
+ with pytest.raises(HTTPException, match="confirm=true") as captured:
+ delete_endpoint("minimax-h3-1k-000001", False)
+
+ assert captured.value.status_code == 400
+ assert delete_endpoint("minimax-h3-1k-000001", True)["deleted"] is True
diff --git a/tests/test_task_adapter_helpers.py b/tests/test_task_adapter_helpers.py
new file mode 100644
index 00000000..57e4c7c9
--- /dev/null
+++ b/tests/test_task_adapter_helpers.py
@@ -0,0 +1,210 @@
+"""Unit coverage for canonical-task adapter helpers without importing launch."""
+from __future__ import annotations
+
+import ast
+import math
+import os
+import re
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from services import resource_scheduler
+
+
+ROOT = Path(__file__).parents[1]
+LAUNCH_PATH = ROOT / "app" / "launch.py"
+SOURCE = LAUNCH_PATH.read_text(encoding="utf-8")
+TREE = ast.parse(SOURCE, filename=str(LAUNCH_PATH))
+
+
+class DummyHTTPException(Exception):
+ def __init__(self, *, status_code: int, detail: str):
+ super().__init__(detail)
+ self.status_code = status_code
+ self.detail = detail
+
+
+def _function(name: str) -> ast.FunctionDef:
+ for node in TREE.body:
+ if isinstance(node, ast.FunctionDef) and node.name == name:
+ return node
+ raise AssertionError(f"Function {name!r} not found")
+
+
+def _load_helpers(*names: str, save_path: Path | None = None) -> dict:
+ selected = [_function(name) for name in names]
+ module = ast.Module(body=selected, type_ignores=[])
+ ast.fix_missing_locations(module)
+ namespace = {
+ "HTTPException": DummyHTTPException,
+ "math": math,
+ "os": os,
+ "re": re,
+ "resource_scheduler": resource_scheduler,
+ "wgp": SimpleNamespace(server_config={
+ "save_path": str(save_path or ROOT / "outputs"),
+ "services": {"active_workspace": "active-one"},
+ }),
+ }
+ exec(compile(module, str(LAUNCH_PATH), "exec"), namespace)
+ return namespace
+
+
+def test_workspace_dir_accepts_only_exact_names_and_returns_real_paths(tmp_path):
+ helpers = _load_helpers("_get_active_workspace", "_workspace_dir", save_path=tmp_path)
+ workspace_dir = helpers["_workspace_dir"]
+
+ assert workspace_dir("default") == os.path.realpath(tmp_path)
+ assert workspace_dir(None) == os.path.realpath(tmp_path / "active-one")
+ assert workspace_dir("Project_2-test") == os.path.realpath(tmp_path / "Project_2-test")
+
+ for invalid in ("", ".", "..", "../escape", "nested/name", "nested\\name",
+ " leading", "trailing ", "has.dot", "café", 123):
+ with pytest.raises(DummyHTTPException) as error:
+ workspace_dir(invalid)
+ assert error.value.status_code == 400
+
+
+def test_workspace_dir_rejects_valid_named_symlink_that_escapes_base(tmp_path):
+ base = tmp_path / "outputs"
+ outside = tmp_path / "outside"
+ base.mkdir()
+ outside.mkdir()
+ link = base / "linked"
+ try:
+ link.symlink_to(outside, target_is_directory=True)
+ except (NotImplementedError, OSError):
+ pytest.skip("directory symlinks are unavailable on this platform")
+
+ workspace_dir = _load_helpers(
+ "_get_active_workspace", "_workspace_dir", save_path=base,
+ )["_workspace_dir"]
+ with pytest.raises(DummyHTTPException) as error:
+ workspace_dir("linked")
+ assert error.value.status_code == 400
+
+
+@pytest.mark.parametrize(
+ ("legacy_percent", "expected"),
+ [(0, 0.0), (1, 0.01), (8, 0.08), (50, 0.5), (100, 1.0)],
+)
+def test_canonical_legacy_progress_converts_percent_to_fraction(
+ legacy_percent, expected,
+):
+ progress = _load_helpers("_canonical_legacy_progress")[
+ "_canonical_legacy_progress"
+ ]
+ assert progress(0, 0, legacy_percent) == pytest.approx(expected)
+
+
+def test_canonical_legacy_progress_prioritizes_current_total_and_clamps():
+ progress = _load_helpers("_canonical_legacy_progress")[
+ "_canonical_legacy_progress"
+ ]
+ assert progress(1, 8, 100) == pytest.approx(0.125)
+ assert progress(10, 8, 0) == 1.0
+ assert progress(-1, 8, 100) == 0.0
+
+
+@pytest.mark.parametrize(
+ "adapter", ["_publish_generation_task", "_publish_generic_legacy_task"],
+)
+def test_legacy_adapters_use_the_canonical_progress_helper(adapter):
+ calls = {
+ node.func.id
+ for node in ast.walk(_function(adapter))
+ if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
+ }
+ assert "_canonical_legacy_progress" in calls
+
+
+def test_task_event_cursor_prefers_latest_valid_cursor():
+ cursor = _load_helpers("_task_event_cursor")["_task_event_cursor"]
+
+ assert cursor(3, "8") == 8
+ assert cursor("12", "broken") == 12
+ assert cursor(-4, -2) == 0
+
+
+def _load_publisher(name: str):
+ selected = [
+ _function(helper)
+ for helper in ("_task_legacy_id", "_task_status", "_task_timestamp", name)
+ ]
+ module = ast.Module(body=selected, type_ignores=[])
+ ast.fix_missing_locations(module)
+ captured = {}
+
+ def upsert(workspace, task_id, **fields):
+ captured.clear()
+ captured.update({"workspace": workspace, "id": task_id, **fields})
+ return dict(captured)
+
+ namespace = {
+ "datetime": __import__("datetime").datetime,
+ "resource_scheduler": resource_scheduler,
+ "time": __import__("time"),
+ "_GENERIC_TASK_CONFIG": {
+ "story-plan": ("Story Lab planning", "llm-planning", True),
+ "comic-plan": ("Comic planning", "llm-planning", True),
+ "video-editor": ("Video editor", "ffmpeg", False),
+ "model3d": ("3D generation", "model3d", False),
+ "rig": ("Character rigging", "rig", False),
+ },
+ "_upsert_canonical_task": upsert,
+ "_canonical_legacy_progress": lambda current, total, progress: 0.0,
+ }
+ exec(compile(module, str(LAUNCH_PATH), "exec"), namespace)
+ return namespace[name], captured
+
+
+def test_series_remote_lane_uses_normalized_origin_and_no_fake_acquisition():
+ publish, captured = _load_publisher("_publish_series_task")
+
+ publish({
+ "jobId": "series-plan-1",
+ "workspace": "default",
+ "status": "running",
+ "request": {
+ "writingProvider": "minimax",
+ "writingModel": "MiniMax-M3",
+ "writingBaseUrl": "https://api.minimax.io/v1",
+ },
+ }, "series-plan")
+
+ assert captured["resource_requirements"] == ["remote:https://api.minimax.io"]
+ assert captured["acquired_resources"] == []
+
+
+@pytest.mark.parametrize(
+ ("engine", "expected"),
+ [("procedural", "local_cpu:rig"), ("unirig", "local_gpu:0")],
+)
+def test_rig_adapter_declares_its_actual_engine_lane(engine, expected):
+ publish, captured = _load_publisher("_publish_generic_legacy_task")
+
+ publish({
+ "job_id": f"rig-{engine}",
+ "workspace": "default",
+ "status": "running",
+ "engine": engine,
+ }, "rig")
+
+ assert captured["resource_requirements"] == [expected]
+
+
+def test_generic_adapter_preserves_service_owned_task_identity():
+ publish, captured = _load_publisher("_publish_generic_legacy_task")
+
+ publish({
+ "job_id": "backend-id",
+ "task_id": "task-model3d-backend-id",
+ "root_task_id": "task-series-root",
+ "workspace": "default",
+ "status": "queued",
+ }, "model3d")
+
+ assert captured["id"] == "task-model3d-backend-id"
+ assert captured["root_id"] == "task-series-root"
diff --git a/tests/test_task_manager.py b/tests/test_task_manager.py
new file mode 100644
index 00000000..35791728
--- /dev/null
+++ b/tests/test_task_manager.py
@@ -0,0 +1,214 @@
+import sqlite3
+
+from services.task_manager import TaskRegistry, redact_sensitive_data, task_context_scope
+
+
+def test_task_registry_persists_ordered_events_and_transitions(tmp_path):
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ task = registry.create(
+ id="task-one", kind="image", title="Generate image", workflow="story",
+ status="queued", workspace="default", current=0, total=2,
+ )
+ registry.update(task["id"], status="running", phase="requesting", current=1)
+ completed = registry.update(
+ task["id"], status="completed", phase="completed", current=2,
+ result_refs=[{"kind": "image", "name": "frame.png"}],
+ )
+
+ assert completed["progress"] == 1
+ assert completed["completed_at"] >= completed["started_at"]
+ assert TaskRegistry(str(tmp_path), interrupt_stale=False).get(task["id"])["status"] == "completed"
+ events = registry.events(task["id"])
+ assert [event["sequence"] for event in events] == [1, 2, 3]
+ assert [event["type"] for event in events] == ["task.created", "task.updated", "task.updated"]
+
+
+def test_restart_marks_unfinished_task_interrupted_and_recoverable(tmp_path):
+ first = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ first.create(
+ id="task-running", kind="video", title="Render", status="running",
+ recoverable=True, workspace="default",
+ )
+
+ second = TaskRegistry(str(tmp_path), interrupt_stale=True)
+
+ task = second.get("task-running")
+ assert task["status"] == "interrupted"
+ assert task["recoverable"] is True
+ assert second.events("task-running")[-1]["type"] == "task.interrupted"
+
+
+def test_task_context_is_explicit_and_redacts_sensitive_metadata(tmp_path):
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ with task_context_scope(request_id="request-1", task_id="task-context"):
+ task = registry.create(
+ id="task-context", kind="llm", title="Plan", status="queued",
+ metadata={
+ "prompt": "secret",
+ "safe": "visible",
+ "token_usage": {"prompt": "7", "completion": 3, "total": 10, "calls": 1},
+ },
+ )
+
+ assert task["metadata"] == {
+ "safe": "visible",
+ "token_usage": {"prompt": 7, "completion": 3, "total": 10, "calls": 1},
+ }
+ assert registry.events("task-context")[0]["context"]["request_id"] == "request-1"
+
+
+def test_sensitive_redaction_covers_nested_provider_keys_headers_and_urls():
+ value = redact_sensitive_data({
+ "minimax_api_key": "secret-key",
+ "nested": {"Authorization": "Bearer abc.def", "safe": "visible"},
+ "url": "https://example.test/file?token=secret&ok=1",
+ })
+ assert value["minimax_api_key"] == "[REDACTED]"
+ assert value["nested"]["Authorization"] == "[REDACTED]"
+ assert value["nested"]["safe"] == "visible"
+ assert "secret" not in value["url"]
+
+
+def test_token_usage_is_normalized_on_create_update_and_reload(tmp_path):
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ created = registry.create(
+ id="task-tokens", kind="llm", title="Plan", status="running",
+ token_usage={"prompt": "11", "completion": -4, "total": 11.8, "calls": "1"},
+ )
+
+ assert created["token_usage"] == {
+ "prompt": 11, "completion": 0, "total": 11, "calls": 1,
+ }
+
+ updated = registry.update(
+ "task-tokens",
+ token_usage={"completion": "9", "total": "20", "calls": "2"},
+ event_type="task.tokens",
+ )
+ assert updated["token_usage"] == {
+ "prompt": 11, "completion": 9, "total": 20, "calls": 2,
+ }
+
+ reloaded = TaskRegistry(str(tmp_path), interrupt_stale=False).get("task-tokens")
+ assert reloaded["token_usage"] == updated["token_usage"]
+ assert registry.events("task-tokens")[-1]["changes"]["token_usage"] == updated["token_usage"]
+
+
+def test_active_tasks_cannot_be_dismissed(tmp_path):
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ registry.create(id="task-active", kind="llm", title="Plan", status="queued")
+ last_event_id = registry.latest_event_id()
+
+ try:
+ registry.delete("task-active")
+ except ValueError as exc:
+ assert "cancelled" in str(exc)
+ else: # pragma: no cover
+ raise AssertionError("active task was deleted")
+
+ assert registry.get("task-active")["status"] == "queued"
+ assert registry.events(after=last_event_id) == []
+
+
+def test_delete_emits_durable_tombstone_replayable_after_last_event_id(tmp_path):
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ registry.create(
+ id="task-dismissed", kind="image", title="Finished image",
+ status="completed", root_id="root-image",
+ )
+ last_event_id = registry.latest_event_id()
+
+ assert registry.delete("task-dismissed") is True
+ assert registry.get("task-dismissed") is None
+ assert registry.list() == []
+
+ replay = registry.events(after=last_event_id)
+ assert len(replay) == 1
+ tombstone = replay[0]
+ assert tombstone["event_id"] > last_event_id
+ assert tombstone["task_id"] == "task-dismissed"
+ assert tombstone["root_id"] == "root-image"
+ assert tombstone["sequence"] == 2
+ assert tombstone["type"] == "task.deleted"
+ changes = dict(tombstone["changes"])
+ deleted_at = changes.pop("deleted_at")
+ assert changes == {
+ "deleted": True,
+ "tombstone": True,
+ "task_id": "task-dismissed",
+ "root_id": "root-image",
+ "status": "completed",
+ }
+ assert deleted_at > 0
+ assert registry.wait_for_events(last_event_id, timeout=0.05) == replay
+
+ reloaded = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ assert reloaded.get("task-dismissed") is None
+ assert reloaded.latest_event_id() == tombstone["event_id"]
+ assert reloaded.events(after=last_event_id) == replay
+ assert reloaded.events("task-dismissed")[-1] == tombstone
+
+
+def test_existing_foreign_key_event_log_is_migrated_without_changing_cursors(tmp_path):
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ registry.create(
+ id="task-legacy", kind="video", title="Legacy task", status="completed",
+ )
+ original_events = registry.events("task-legacy")
+ original_cursor = registry.latest_event_id()
+
+ with sqlite3.connect(registry.path, isolation_level=None) as connection:
+ connection.executescript("""
+ BEGIN IMMEDIATE;
+ DROP INDEX idx_task_events_task;
+ ALTER TABLE task_events RENAME TO task_events_durable_source;
+ CREATE TABLE task_events (
+ event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+ task_id TEXT NOT NULL,
+ root_id TEXT NOT NULL,
+ sequence INTEGER NOT NULL,
+ timestamp REAL NOT NULL,
+ type TEXT NOT NULL,
+ changes TEXT NOT NULL,
+ context TEXT NOT NULL,
+ FOREIGN KEY(task_id) REFERENCES tasks(id) ON DELETE CASCADE,
+ UNIQUE(task_id, sequence)
+ );
+ INSERT INTO task_events
+ (event_id, task_id, root_id, sequence, timestamp, type, changes, context)
+ SELECT event_id, task_id, root_id, sequence, timestamp, type, changes, context
+ FROM task_events_durable_source;
+ DROP TABLE task_events_durable_source;
+ CREATE INDEX idx_task_events_task ON task_events(task_id, sequence);
+ COMMIT;
+ """)
+
+ reloaded = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ with sqlite3.connect(reloaded.path) as connection:
+ assert connection.execute("PRAGMA foreign_key_list(task_events)").fetchall() == []
+ assert reloaded.events("task-legacy") == original_events
+ assert reloaded.latest_event_id() == original_cursor
+
+ assert reloaded.delete("task-legacy") is True
+ tombstones = reloaded.events(after=original_cursor)
+ assert [event["type"] for event in tombstones] == ["task.deleted"]
+
+ restarted = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ assert restarted.events(after=original_cursor) == tombstones
+
+
+def test_compatibility_adapter_can_attach_an_existing_task_to_its_parent(tmp_path):
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ registry.create(id="task-parent", kind="series", title="Episode", status="running")
+ registry.create(id="task-child", kind="video", title="Shot", status="running")
+
+ child = registry.update(
+ "task-child", root_id="task-parent", parent_id="task-parent",
+ event_type="adapter.synced", force=True,
+ )
+
+ assert child["root_id"] == "task-parent"
+ assert child["parent_id"] == "task-parent"
+ assert [task["id"] for task in registry.list(root_id="task-parent")] == [
+ "task-child", "task-parent",
+ ]
diff --git a/tests/test_task_manager_active_listing.py b/tests/test_task_manager_active_listing.py
new file mode 100644
index 00000000..0a1b363a
--- /dev/null
+++ b/tests/test_task_manager_active_listing.py
@@ -0,0 +1,153 @@
+"""Regression tests for canonical task-list recovery semantics."""
+
+import itertools
+
+from app.services import task_manager
+from app.services.task_manager import ALL_STATUSES, TaskRegistry
+
+
+def _use_deterministic_clock(monkeypatch, start: int = 1_000):
+ ticks = itertools.count(start)
+ monkeypatch.setattr(task_manager, "_now", lambda: float(next(ticks)))
+
+
+def test_all_listing_keeps_old_active_task_beyond_terminal_history_limit(
+ tmp_path,
+ monkeypatch,
+):
+ _use_deterministic_clock(monkeypatch)
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ registry.create(
+ id="active-before-history",
+ kind="series",
+ title="Old active render",
+ status="running",
+ workspace="default",
+ )
+ for index in range(325):
+ registry.create(
+ id=f"terminal-{index:03d}",
+ kind="video",
+ title=f"Completed clip {index}",
+ status="completed",
+ workspace="default",
+ )
+
+ listed = registry.list(statuses=set(ALL_STATUSES), limit=300)
+
+ assert len(listed) == 301
+ assert sum(task["status"] == "completed" for task in listed) == 300
+ assert listed[-1]["id"] == "active-before-history"
+ assert len({task["id"] for task in listed}) == len(listed)
+ assert [task["updated_at"] for task in listed] == sorted(
+ (task["updated_at"] for task in listed),
+ reverse=True,
+ )
+
+ # The guarantee comes from SQLite state, not an in-memory side channel.
+ reopened = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ persisted = reopened.list(statuses=set(ALL_STATUSES), limit=300)
+ assert [task["id"] for task in persisted].count("active-before-history") == 1
+ assert len(persisted) == 301
+
+
+def test_mixed_status_and_root_filters_keep_all_matching_active_tasks(
+ tmp_path,
+ monkeypatch,
+):
+ _use_deterministic_clock(monkeypatch)
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ registry.create(
+ id="root-a-running",
+ root_id="root-a",
+ kind="series",
+ title="Running A",
+ status="running",
+ )
+ registry.create(
+ id="root-a-queued",
+ root_id="root-a",
+ kind="series",
+ title="Queued A",
+ status="queued",
+ )
+ registry.create(
+ id="root-b-running",
+ root_id="root-b",
+ kind="series",
+ title="Running B",
+ status="running",
+ )
+ registry.create(
+ id="root-a-failed-old",
+ root_id="root-a",
+ kind="series",
+ title="Failed A old",
+ status="failed",
+ )
+ registry.create(
+ id="root-a-failed-new",
+ root_id="root-a",
+ kind="series",
+ title="Failed A new",
+ status="failed",
+ )
+ registry.create(
+ id="root-a-completed",
+ root_id="root-a",
+ kind="series",
+ title="Completed A",
+ status="completed",
+ )
+ registry.create(
+ id="root-b-failed",
+ root_id="root-b",
+ kind="series",
+ title="Failed B",
+ status="failed",
+ )
+
+ listed = registry.list(
+ statuses={"running", "queued", "failed"},
+ root_id="root-a",
+ limit=1,
+ )
+
+ assert {task["id"] for task in listed} == {
+ "root-a-running",
+ "root-a-queued",
+ "root-a-failed-new",
+ }
+ assert len(listed) == 3
+ assert all(task["root_id"] == "root-a" for task in listed)
+ assert all(task["status"] in {"running", "queued", "failed"} for task in listed)
+ assert [task["updated_at"] for task in listed] == sorted(
+ (task["updated_at"] for task in listed),
+ reverse=True,
+ )
+
+
+def test_terminal_only_filter_still_uses_limit_and_invalid_filter_is_empty(
+ tmp_path,
+ monkeypatch,
+):
+ _use_deterministic_clock(monkeypatch)
+ registry = TaskRegistry(str(tmp_path), interrupt_stale=False)
+ for index in range(4):
+ registry.create(
+ id=f"failed-{index}",
+ kind="llm",
+ title=f"Failed plan {index}",
+ status="failed",
+ )
+ registry.create(
+ id="active",
+ kind="llm",
+ title="Active plan",
+ status="running",
+ )
+
+ listed = registry.list(statuses={"failed"}, limit=2)
+
+ assert [task["id"] for task in listed] == ["failed-3", "failed-2"]
+ assert registry.list(statuses={"not-a-status"}, limit=2) == []
diff --git a/tests/test_video_editor_replacement_ui.py b/tests/test_video_editor_replacement_ui.py
new file mode 100644
index 00000000..a74eb217
--- /dev/null
+++ b/tests/test_video_editor_replacement_ui.py
@@ -0,0 +1,39 @@
+"""Source contracts for replacing one Montage clip through Video Creation."""
+
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+EDITOR = ROOT / "ui" / "src" / "features" / "video-editor" / "VideoEditorPanel.tsx"
+HANDOFF = ROOT / "ui" / "src" / "features" / "video-editor" / "replacementHandoff.ts"
+MAIN = ROOT / "ui" / "src" / "components" / "MainContent" / "MainContent.tsx"
+FEED = ROOT / "ui" / "src" / "components" / "MainContent" / "MediaFeedItem.tsx"
+
+
+def test_selected_montage_clip_can_be_opened_in_video_creation():
+ editor = EDITOR.read_text(encoding="utf-8")
+
+ assert "Rehacer en Creación de vídeo" in editor
+ assert "fetchOutputMetadata(outputName)" in editor
+ assert "loadSettingsFromOutput()" in editor
+ assert "writeVideoEditorReplacementTarget" in editor
+ assert "setMediaFilter('videos')" in editor
+ assert "persistEditorDraft(clips, projectName, resolution, fps)" in editor
+
+
+def test_generated_video_can_replace_only_the_original_timeline_slot():
+ editor = EDITOR.read_text(encoding="utf-8")
+ handoff = HANDOFF.read_text(encoding="utf-8")
+ main = MAIN.read_text(encoding="utf-8")
+ feed = FEED.read_text(encoding="utf-8")
+
+ assert "maestro-video-editor-replacement-target-v1" in handoff
+ assert "maestro-video-editor-replacement-result-v1" in handoff
+ assert "Usar en posición" in feed
+ assert "writeVideoEditorReplacementResult" in feed
+ assert "clearVideoEditorReplacementTarget" in main
+ assert "readVideoEditorReplacementResult" in editor
+ assert "clip.id === replacement.clipId" in editor
+ assert "clearVideoEditorReplacementResult()" in editor
+ assert "clearVideoEditorReplacementTarget()" in editor
+ assert "persistEditorDraft(next, projectName, resolution, fps)" in editor
diff --git a/tests/test_video_editor_scheduler_jobs.py b/tests/test_video_editor_scheduler_jobs.py
new file mode 100644
index 00000000..8caa26bd
--- /dev/null
+++ b/tests/test_video_editor_scheduler_jobs.py
@@ -0,0 +1,463 @@
+"""Model-free scheduler contract tests for editor and comic animatic jobs.
+
+Importing ``app.launch`` initializes the full Maestro application, so this
+module extracts just the editor job functions from its AST and runs them with
+small in-memory fakes. No FFmpeg process, model, or GPU runtime is used.
+"""
+from __future__ import annotations
+
+import ast
+import copy
+from datetime import datetime
+import json
+import math
+import os
+import re
+import sys
+import threading
+import time
+import traceback
+import types
+import uuid
+from contextlib import contextmanager
+from functools import lru_cache
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+
+ROOT = Path(__file__).parents[1]
+LAUNCH = ROOT / "app" / "launch.py"
+LANE_KEY = "local_cpu:ffmpeg"
+
+
+class _HTTPException(Exception):
+ def __init__(self, *, status_code: int, detail: str):
+ super().__init__(detail)
+ self.status_code = status_code
+ self.detail = detail
+
+
+class _ResourceAcquireCancelled(Exception):
+ pass
+
+
+@lru_cache(maxsize=1)
+def _tree() -> ast.Module:
+ return ast.parse(LAUNCH.read_text(encoding="utf-8"), filename=str(LAUNCH))
+
+
+def _function(name: str) -> ast.FunctionDef | ast.AsyncFunctionDef:
+ for node in _tree().body:
+ if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name:
+ selected = copy.deepcopy(node)
+ selected.decorator_list = []
+ return selected
+ raise AssertionError(f"Function {name!r} not found in app/launch.py")
+
+
+def _load(*names: str, namespace: dict) -> dict:
+ module = ast.Module(body=[_function(name) for name in names], type_ignores=[])
+ ast.fix_missing_locations(module)
+ exec(compile(module, str(LAUNCH), "exec"), namespace)
+ return namespace
+
+
+def _module(name: str, **attrs) -> types.ModuleType:
+ module = types.ModuleType(name)
+ module.__dict__.update(attrs)
+ return module
+
+
+def _editor_body(workspace: str = "captured-editor") -> dict:
+ return {
+ "name": "Scheduler editor test",
+ "workspace": workspace,
+ "width": 1280,
+ "height": 720,
+ "fps": 30,
+ "clips": [{"source": "clip.mp4", "transition": "none"}],
+ }
+
+
+def _animatic_body(workspace: str = "captured-animatic") -> dict:
+ return {
+ "comic_id": "comic-contract",
+ "comic_title": "Scheduler animatic test",
+ "workspace": workspace,
+ "width": 1280,
+ "height": 720,
+ "fps": 30,
+ "transition": "crossfade",
+ "transition_duration": 0.35,
+ "panels": [{"source": "panel.webp", "duration": 1.5}],
+ }
+
+
+def _harness(monkeypatch, tmp_path: Path) -> dict:
+ events: list[tuple] = []
+ render_calls: list[str] = []
+ workspace_calls: list[str] = []
+ jobs: dict[str, dict] = {}
+ lane = SimpleNamespace(key=LANE_KEY)
+
+ class DeferredThread:
+ instances: list["DeferredThread"] = []
+
+ def __init__(self, *, target, args=(), kwargs=None, **_ignored):
+ self.target = target
+ self.args = tuple(args)
+ self.kwargs = dict(kwargs or {})
+ self.started = False
+ self.__class__.instances.append(self)
+
+ def start(self) -> None:
+ self.started = True
+ events.append(("thread_start", self.target.__name__))
+
+ def run_now(self) -> None:
+ self.target(*self.args, **self.kwargs)
+
+ class Coordinator:
+ @contextmanager
+ def acquire(self, requested_lane, *, task_id, description, cancelled):
+ events.append(("lane_wait", requested_lane.key, task_id, description))
+ if cancelled():
+ raise _ResourceAcquireCancelled(task_id)
+ events.append(("lane_acquire", requested_lane.key, task_id))
+ try:
+ yield
+ finally:
+ events.append(("lane_release", requested_lane.key, task_id))
+
+ def default_render_project(_clips, output_path, *, progress, **_settings):
+ render_calls.append("export")
+ progress(37, "Encoding editor timeline…")
+ Path(output_path).write_bytes(b"fake editor mp4")
+ return {"duration": 1.0}
+
+ def default_render_animatic(_panels, output_path, *, progress, **_settings):
+ render_calls.append("animatic")
+ progress(43, "Encoding comic animatic…")
+ Path(output_path).write_bytes(b"fake animatic mp4")
+ return {"duration": 1.5}
+
+ video_editor_module = _module(
+ "services.video_editor",
+ normalise_time_card_text=lambda value: str(value or "").strip(),
+ render_project=default_render_project,
+ render_comic_animatic=default_render_animatic,
+ )
+ services_module = _module("services", video_editor=video_editor_module)
+ services_module.__path__ = []
+ monkeypatch.setitem(sys.modules, "services", services_module)
+ monkeypatch.setitem(sys.modules, "services.video_editor", video_editor_module)
+
+ def workspace_dir(workspace=None) -> str:
+ workspace_calls.append(workspace)
+ path = tmp_path / str(workspace)
+ path.mkdir(parents=True, exist_ok=True)
+ return str(path)
+
+ def publish(job: dict, adapter: str) -> dict:
+ events.append(("publish", job["status"], job.get("phase"), copy.deepcopy(job)))
+ assert adapter == "video-editor"
+ return {"id": job["task_id"], "root_id": job["root_task_id"]}
+
+ namespace = {
+ "HTTPException": _HTTPException,
+ "copy": copy,
+ "json": json,
+ "os": os,
+ "re": re,
+ "threading": SimpleNamespace(Thread=DeferredThread),
+ "time": time,
+ "traceback": traceback,
+ "uuid": uuid,
+ "resource_scheduler": SimpleNamespace(
+ coordinator=Coordinator(),
+ ResourceAcquireCancelled=_ResourceAcquireCancelled,
+ ),
+ "_VIDEO_EDITOR_FFMPEG_LANE": lane,
+ "_VIDEO_EDITOR_TERMINAL": frozenset({"completed", "failed", "cancelled"}),
+ "_video_editor_jobs": jobs,
+ "_video_editor_jobs_lock": threading.RLock(),
+ "_publish_generic_legacy_task": publish,
+ "_workspace_dir": workspace_dir,
+ "_get_active_workspace": lambda: (_ for _ in ()).throw(
+ AssertionError("explicit workspace was not captured")
+ ),
+ }
+ _load(
+ "_public_video_editor_job",
+ "_publish_video_editor_job",
+ "_video_editor_job_snapshot",
+ "_video_editor_job_update",
+ "_register_video_editor_job",
+ "_video_editor_cancel_requested",
+ "_remove_video_editor_output_bundle",
+ "_finish_video_editor_cancelled",
+ "_video_editor_task_identity",
+ "_run_video_editor_export",
+ "start_video_editor_export",
+ "_run_comic_animatic",
+ "start_comic_animatic",
+ "cancel_video_editor_export",
+ namespace=namespace,
+ )
+ namespace["_resolve_video_editor_source"] = (
+ lambda source, workspace=None: str(tmp_path / str(workspace) / os.path.basename(source))
+ )
+ namespace["_resolve_comic_animatic_image"] = (
+ lambda source, workspace=None: str(tmp_path / str(workspace) / os.path.basename(source))
+ )
+ namespace.update({
+ "DeferredThread": DeferredThread,
+ "events": events,
+ "jobs": jobs,
+ "render_calls": render_calls,
+ "video_editor_module": video_editor_module,
+ "workspace_calls": workspace_calls,
+ })
+ return namespace
+
+
+@pytest.mark.parametrize(
+ ("start_name", "body_factory", "workspace"),
+ [
+ ("start_video_editor_export", _editor_body, "explicit-editor"),
+ ("start_comic_animatic", _animatic_body, "explicit-animatic"),
+ ],
+)
+def test_post_reserves_identity_and_publishes_queued_before_thread_under_250ms(
+ monkeypatch,
+ tmp_path,
+ start_name,
+ body_factory,
+ workspace,
+):
+ harness = _harness(monkeypatch, tmp_path)
+
+ started_at = time.monotonic()
+ response = harness[start_name](body_factory(workspace))
+ elapsed = time.monotonic() - started_at
+
+ assert elapsed < 0.250
+ assert response["job_id"]
+ assert response["task_id"].startswith("task-video-editor-")
+ assert response["root_task_id"] == response["task_id"]
+ assert response["workspace"] == workspace
+ assert response["status"] == "queued"
+ assert response["resource_requirements"] == [LANE_KEY]
+ assert harness["workspace_calls"] == [workspace]
+ assert [event[0] for event in harness["events"][:2]] == ["publish", "thread_start"]
+ assert harness["events"][0][1:3] == ("queued", "queued")
+ assert harness["render_calls"] == []
+
+
+@pytest.mark.parametrize(
+ ("start_name", "body_factory"),
+ [
+ ("start_video_editor_export", _editor_body),
+ ("start_comic_animatic", _animatic_body),
+ ],
+)
+def test_queued_cancel_never_acquires_ffmpeg_lane_or_renders(
+ monkeypatch,
+ tmp_path,
+ start_name,
+ body_factory,
+):
+ harness = _harness(monkeypatch, tmp_path)
+ response = harness[start_name](body_factory())
+
+ cancelled = harness["cancel_video_editor_export"](response["job_id"])
+ harness["DeferredThread"].instances[-1].run_now()
+
+ assert cancelled["status"] == "cancelled"
+ assert cancelled["phase"] == "cancelled"
+ assert cancelled["cancel_mode"] == "immediate"
+ assert harness["jobs"][response["job_id"]]["status"] == "cancelled"
+ assert harness["render_calls"] == []
+ assert not any(event[0].startswith("lane_") for event in harness["events"])
+ assert [event[1] for event in harness["events"] if event[0] == "publish"] == [
+ "queued",
+ "cancelled",
+ ]
+
+
+def test_running_cancel_waits_for_safe_boundary_then_removes_output_bundle(
+ monkeypatch,
+ tmp_path,
+):
+ harness = _harness(monkeypatch, tmp_path)
+ render_started = threading.Event()
+ release_render = threading.Event()
+ started_next_ffmpeg_step: list[bool] = []
+
+ def blocking_render(_clips, output_path, *, progress, **_settings):
+ harness["render_calls"].append("export")
+ Path(output_path).write_bytes(b"partial mp4")
+ Path(output_path).with_suffix(".meta.json").write_text("{}", encoding="utf-8")
+ progress(55, "Halfway through FFmpeg…")
+ render_started.set()
+ assert release_render.wait(timeout=2)
+ progress(75, "Current FFmpeg subprocess reached its safe boundary")
+ started_next_ffmpeg_step.append(True)
+ return {"duration": 2.0}
+
+ harness["video_editor_module"].render_project = blocking_render
+ response = harness["start_video_editor_export"](_editor_body())
+ deferred = harness["DeferredThread"].instances[-1]
+ worker = threading.Thread(target=deferred.run_now, daemon=True)
+ worker.start()
+ assert render_started.wait(timeout=2)
+
+ output_path = Path(deferred.args[3])
+ sidecar_path = output_path.with_suffix(".meta.json")
+ cancelling = harness["cancel_video_editor_export"](response["job_id"])
+
+ assert cancelling["status"] == "cancelling"
+ assert cancelling["phase"] == "cancelling"
+ assert cancelling["acquired_resources"] == [LANE_KEY]
+ assert output_path.exists() and sidecar_path.exists()
+ assert worker.is_alive()
+ assert not any(
+ event[0] == "publish" and event[1] == "cancelled"
+ for event in harness["events"]
+ )
+
+ release_render.set()
+ worker.join(timeout=2)
+ assert not worker.is_alive()
+ terminal = harness["jobs"][response["job_id"]]
+ assert terminal["status"] == "cancelled"
+ assert terminal["phase"] == "cancelled"
+ assert terminal["cancel_mode"] == "deferred"
+ assert terminal["acquired_resources"] == []
+ assert started_next_ffmpeg_step == []
+ assert not output_path.exists()
+ assert not sidecar_path.exists()
+
+ release_index = max(
+ index for index, event in enumerate(harness["events"])
+ if event[0] == "lane_release"
+ )
+ cancelled_index = max(
+ index for index, event in enumerate(harness["events"])
+ if event[0] == "publish" and event[1] == "cancelled"
+ )
+ assert cancelled_index > release_index
+
+
+@pytest.mark.parametrize(
+ ("start_name", "body_factory", "render_kind"),
+ [
+ ("start_video_editor_export", _editor_body, "export"),
+ ("start_comic_animatic", _animatic_body, "animatic"),
+ ],
+)
+def test_progress_and_terminal_publish_without_get_and_sidecar_keeps_task_hierarchy(
+ monkeypatch,
+ tmp_path,
+ start_name,
+ body_factory,
+ render_kind,
+):
+ harness = _harness(monkeypatch, tmp_path)
+ response = harness[start_name](body_factory())
+ deferred = harness["DeferredThread"].instances[-1]
+
+ deferred.run_now()
+
+ terminal = harness["jobs"][response["job_id"]]
+ assert terminal["status"] == "completed"
+ assert terminal["phase"] == "completed"
+ assert terminal["acquired_resources"] == []
+ assert harness["render_calls"] == [render_kind]
+
+ publications = [event[3] for event in harness["events"] if event[0] == "publish"]
+ statuses = [item["status"] for item in publications]
+ phases = [item["phase"] for item in publications]
+ assert statuses[0] == "queued"
+ assert "waiting_resource" in statuses
+ assert "running" in statuses
+ assert statuses[-1] == "completed"
+ assert "rendering" in phases
+ assert any(item["acquired_resources"] == [LANE_KEY] for item in publications)
+ assert any(0 < int(item["progress"]) < 100 for item in publications)
+
+ output_path = Path(deferred.args[-1])
+ sidecar_path = output_path.with_suffix(".meta.json")
+ assert output_path.exists()
+ assert sidecar_path.exists()
+ sidecar = json.loads(sidecar_path.read_text(encoding="utf-8"))
+ assert sidecar["job_id"] == response["job_id"]
+ assert sidecar["task_id"] == response["task_id"]
+ assert sidecar["root_task_id"] == response["root_task_id"]
+
+
+def test_canonical_adapter_exposes_real_lane_cancel_contract_and_control_route():
+ captured: list[dict] = []
+
+ def upsert(workspace, task_id, **fields):
+ value = {"workspace": workspace, "id": task_id, **fields}
+ captured.append(value)
+ return value
+
+ namespace = {
+ "HTTPException": _HTTPException,
+ "datetime": datetime,
+ "math": math,
+ "resource_scheduler": SimpleNamespace(
+ cpu_lane=lambda name: SimpleNamespace(key=f"local_cpu:{name}"),
+ ),
+ "time": time,
+ "_GENERIC_TASK_CONFIG": {
+ "video-editor": ("Video editor", "ffmpeg", False),
+ },
+ "_upsert_canonical_task": upsert,
+ }
+ _load(
+ "_task_legacy_id",
+ "_task_status",
+ "_task_timestamp",
+ "_canonical_legacy_progress",
+ "_publish_generic_legacy_task",
+ "_control_canonical_task",
+ namespace=namespace,
+ )
+ cancelled_ids: list[str] = []
+ namespace["cancel_video_editor_export"] = lambda job_id: cancelled_ids.append(job_id)
+
+ namespace["_publish_generic_legacy_task"]({
+ "job_id": "video-edit-contract",
+ "task_id": "task-video-editor-contract",
+ "root_task_id": "task-root-contract",
+ "workspace": "editing-room",
+ "status": "cancelling",
+ "phase": "cancelling",
+ "current": 55,
+ "total": 100,
+ "progress": 55,
+ "resource_lane": LANE_KEY,
+ "acquired_resources": [LANE_KEY],
+ "cancel_mode": "deferred",
+ "safe_boundary": "after_current_ffmpeg_render",
+ }, "video-editor")
+
+ task = captured[-1]
+ assert task["status"] == "running"
+ assert task["phase"] == "cancelling"
+ assert task["resource_requirements"] == [LANE_KEY]
+ assert task["acquired_resources"] == [LANE_KEY]
+ assert task["cancelable"] is False
+ assert task["metadata"]["cancel_mode"] == "deferred"
+ assert task["metadata"]["safe_boundary"] == "after_current_ffmpeg_render"
+
+ namespace["_control_canonical_task"]({
+ "backend_job_id": "video-edit-contract",
+ "metadata": {"adapter": "video-editor"},
+ }, "cancel")
+ assert cancelled_ids == ["video-edit-contract"]
diff --git a/ui/src/api/client.ts b/ui/src/api/client.ts
index 69098d5b..55167784 100644
--- a/ui/src/api/client.ts
+++ b/ui/src/api/client.ts
@@ -1,4 +1,6 @@
import { rememberPrompt } from '../lib/promptHistory'
+import { openCanonicalTaskEventStream } from '../lib/canonicalTaskEvents'
+import type { CanonicalTaskEvent, CanonicalTaskStreamState } from '../lib/canonicalTaskEvents'
import type { DirectorModelCompatibility, GenerationDetails, H3WindowPlan, ProductionPlan, ScailResolutionProfile } from '../types'
const BASE = '' // same origin in production; Vite proxy handles /api in dev
@@ -54,6 +56,10 @@ export interface ApiOutput {
favorite?: boolean
size: number
created_at: number
+ /** Time the generated asset was fully published. Older/imported assets
+ * fall back to the media file's modification time. */
+ completed_at?: number
+ completion_time_source?: 'metadata' | 'file'
url: string
/** Small static preview for image/video cards and saved 3D/scene assets. */
thumbnail_url?: string | null
@@ -67,7 +73,9 @@ export interface ApiOutput {
export interface ApiJobStatus {
job_id: string
- status: 'queued' | 'running' | 'completed' | 'failed' | 'cancelled'
+ task_id?: string | null
+ root_task_id?: string | null
+ status: 'queued' | 'waiting_resource' | 'running' | 'cancelling' | 'completed' | 'failed' | 'cancelled'
progress: number
step: number
total_steps: number
@@ -85,6 +93,7 @@ export interface ApiJobStatus {
* See `OomInfo` in types/index.ts. */
oom_info?: import('../types').OomInfo | null
generation_details?: GenerationDetails
+ h3_window_plan?: H3WindowPlan | null
}
export interface ApiTaskTiming {
@@ -96,6 +105,101 @@ export interface ApiTaskTiming {
phase_timings: Array<{ phase: string; seconds: number }>
}
+export type CanonicalTaskStatus =
+ | 'created' | 'queued' | 'waiting_resource' | 'running'
+ | 'completed' | 'failed' | 'cancelled' | 'interrupted'
+
+export interface CanonicalTask {
+ id: string
+ root_id: string
+ parent_id?: string | null
+ kind: string
+ title: string
+ workflow: string
+ status: CanonicalTaskStatus
+ phase: string
+ message: string
+ detail?: string
+ current: number
+ total: number
+ progress: number
+ detail_current: number
+ detail_total: number
+ created_at: number
+ queued_at?: number | null
+ started_at?: number | null
+ updated_at: number
+ completed_at?: number | null
+ provider?: string
+ model?: string
+ server_origin?: string
+ resource_requirements?: string[]
+ acquired_resources?: string[]
+ attempt: number
+ max_attempts: number
+ token_usage?: { prompt?: number; completion?: number; total?: number; calls?: number }
+ backend_job_id?: string
+ pipeline_id?: string
+ cancelable: boolean
+ resumable: boolean
+ recoverable: boolean
+ error?: { message?: string; retryable?: boolean } | null
+ result_refs?: string[]
+ metadata?: Record
+}
+
+export async function fetchCanonicalTasks(
+ workspace: string,
+ status: 'active' | 'all' = 'all',
+): Promise<{ workspace: string; tasks: CanonicalTask[] }> {
+ const query = new URLSearchParams({ workspace, status, limit: '300' })
+ const res = await fetch(`${BASE}/api/v1/tasks?${query}`)
+ if (!res.ok) throw new Error('Failed to fetch Maestro tasks')
+ return res.json()
+}
+
+export function subscribeCanonicalTaskEvents(
+ workspace: string,
+ onEvent: (event: CanonicalTaskEvent) => void,
+ onError?: () => void,
+ onStateChange?: (state: CanonicalTaskStreamState) => void,
+): () => void {
+ return openCanonicalTaskEventStream(BASE, workspace, onEvent, onError, onStateChange)
+}
+
+export async function upsertCanonicalClientTask(task: Record): Promise {
+ const res = await fetch(`${BASE}/api/v1/tasks/upsert`, {
+ method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ task }),
+ })
+ if (!res.ok) throw new Error('Failed to publish Maestro activity')
+ return res.json()
+}
+
+export async function cancelCanonicalTask(taskId: string, workspace: string): Promise {
+ const res = await fetch(`${BASE}/api/v1/tasks/${encodeURIComponent(taskId)}/cancel?workspace=${encodeURIComponent(workspace)}`, { method: 'POST' })
+ if (!res.ok) {
+ const error = await res.json().catch(() => ({ detail: 'Task cancellation failed' }))
+ throw new Error(error.detail || 'Task cancellation failed')
+ }
+ const payload = await res.json()
+ return payload.task
+}
+
+export async function resumeCanonicalTask(taskId: string, workspace: string): Promise {
+ const res = await fetch(`${BASE}/api/v1/tasks/${encodeURIComponent(taskId)}/resume?workspace=${encodeURIComponent(workspace)}`, { method: 'POST' })
+ if (!res.ok) {
+ const error = await res.json().catch(() => ({ detail: 'Task resume failed' }))
+ throw new Error(error.detail || 'Task resume failed')
+ }
+ const payload = await res.json()
+ return payload.task
+}
+
+export async function dismissCanonicalTask(taskId: string, workspace: string): Promise {
+ const res = await fetch(`${BASE}/api/v1/tasks/${encodeURIComponent(taskId)}?workspace=${encodeURIComponent(workspace)}`, { method: 'DELETE' })
+ if (!res.ok) throw new Error('Failed to dismiss Maestro task')
+}
+
// --- Models & Families ---
export async function fetchModels(): Promise<{ families: ApiFamily[]; models: ApiModel[] }> {
@@ -228,7 +332,13 @@ export async function fetchDefaults(modelType: string): Promise): Promise<{ job_id: string; h3_window_plan?: H3WindowPlan }> {
+export async function submitGeneration(params: Record): Promise<{
+ job_id: string
+ task_id?: string | null
+ root_task_id?: string | null
+ status: string
+ h3_window_plan?: H3WindowPlan
+}> {
const res = await fetch(`${BASE}/api/v1/generate`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
@@ -321,9 +431,31 @@ export interface MiniMaxMusicCandidate {
duration_seconds: number
provider: 'minimax'
model: string
+ task_id?: string
+ root_task_id?: string
+ taskId?: string
+ rootTaskId?: string
+}
+
+export interface MiniMaxMusicJob {
+ jobId: string
+ taskId: string
+ rootTaskId: string
+ workspace: string
+ status: 'queued' | 'waiting_resource' | 'running' | 'cancelling' | 'completed' | 'failed' | 'cancelled' | 'interrupted'
+ phase: string
+ message: string
+ current: number
+ total: number
+ progress: number
+ provider: 'minimax'
+ model: string
+ candidates: MiniMaxMusicCandidate[]
+ error?: string | null
+ statusCode?: number
}
-export async function generateStoryMusicCandidates(params: {
+export interface StoryMusicCandidateRequest {
prompt: string
lyrics: string
count: 1 | 2 | 3
@@ -331,8 +463,12 @@ export async function generateStoryMusicCandidates(params: {
reference_audio_filename?: string
instrumental?: boolean
workspace?: string
-}): Promise<{ candidates: MiniMaxMusicCandidate[] }> {
- const res = await fetch(`${BASE}/api/v1/stories/music-candidates`, {
+}
+
+export async function startStoryMusicCandidatesJob(
+ params: StoryMusicCandidateRequest,
+): Promise {
+ const res = await fetch(`${BASE}/api/v1/stories/music-candidates/jobs`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(params),
@@ -344,6 +480,75 @@ export async function generateStoryMusicCandidates(params: {
return res.json()
}
+export async function fetchStoryMusicCandidatesJob(jobId: string): Promise {
+ const res = await fetch(
+ `${BASE}/api/v1/stories/music-candidates/jobs/${encodeURIComponent(jobId)}`,
+ )
+ if (!res.ok) {
+ const error = await res.json().catch(() => ({ detail: 'MiniMax Music job not found' }))
+ throw new Error(error.detail || 'MiniMax Music job not found')
+ }
+ return res.json()
+}
+
+export async function cancelStoryMusicCandidatesJob(jobId: string): Promise {
+ const res = await fetch(
+ `${BASE}/api/v1/stories/music-candidates/jobs/${encodeURIComponent(jobId)}/cancel`,
+ { method: 'POST' },
+ )
+ if (!res.ok) {
+ const error = await res.json().catch(() => ({ detail: 'MiniMax Music cancellation failed' }))
+ throw new Error(error.detail || 'MiniMax Music cancellation failed')
+ }
+ return res.json()
+}
+
+export async function generateStoryMusicCandidates(
+ params: StoryMusicCandidateRequest,
+ options: {
+ onJobSubmitted?: (job: MiniMaxMusicJob) => void
+ onProgress?: (job: MiniMaxMusicJob) => void
+ } = {},
+): Promise<{
+ candidates: MiniMaxMusicCandidate[]
+ status: 'completed' | 'cancelled' | 'failed' | 'interrupted'
+ jobId: string
+ taskId: string
+ message: string
+}> {
+ let job = await startStoryMusicCandidatesJob(params)
+ options.onJobSubmitted?.(job)
+ let pollFailures = 0
+ while (!['completed', 'failed', 'cancelled', 'interrupted'].includes(job.status)) {
+ await new Promise(resolve => window.setTimeout(resolve, pollFailures ? Math.min(10_000, pollFailures * 1_500) : 1_000))
+ try {
+ job = await fetchStoryMusicCandidatesJob(job.jobId)
+ pollFailures = 0
+ options.onProgress?.(job)
+ } catch (error) {
+ pollFailures += 1
+ if (pollFailures >= 20) {
+ throw new Error(
+ `Could not reconnect to MiniMax Music job ${job.jobId}; its ID was preserved: ${(error as Error).message}`,
+ )
+ }
+ }
+ }
+ if (job.status === 'completed' || job.candidates.length > 0) {
+ return {
+ candidates: job.candidates,
+ status: job.status as 'completed' | 'cancelled' | 'failed' | 'interrupted',
+ jobId: job.jobId,
+ taskId: job.taskId,
+ message: job.message,
+ }
+ }
+ throw new Error(
+ `${job.statusCode ? `HTTP ${job.statusCode}: ` : ''}`
+ + (job.error || job.message || `MiniMax Music job ${job.status}`),
+ )
+}
+
export async function translateStoryLyrics(params: {
lyrics: string
targetLanguage: string
@@ -582,12 +787,14 @@ export async function saveScene(scene: import('../types').Scene, preview: string
return res.json()
}
-export function getFileUrl(filename: string): string {
- return `${BASE}/api/v1/file/${encodeURIComponent(filename)}`
+export function getFileUrl(filename: string, workspace?: string): string {
+ const query = workspace ? `?workspace=${encodeURIComponent(workspace)}` : ''
+ return `${BASE}/api/v1/file/${encodeURIComponent(filename)}${query}`
}
-export function getOutputThumbnailUrl(filename: string): string {
- return `${BASE}/api/v1/outputs/thumbnail/${encodeURIComponent(filename)}`
+export function getOutputThumbnailUrl(filename: string, workspace?: string): string {
+ const query = workspace ? `?workspace=${encodeURIComponent(workspace)}` : ''
+ return `${BASE}/api/v1/outputs/thumbnail/${encodeURIComponent(filename)}${query}`
}
export function getUploadUrl(filename: string): string {
@@ -636,12 +843,18 @@ export async function fetchStoredAsset(pathOrFilename: string): Promise {
+export async function fetchOutputMetadata(
+ name: string,
+ workspace?: string,
+): Promise {
// Retry with a per-attempt timeout. On a slow/high-latency link (e.g. the user
// is remote over VPN) the request can stall long enough that a single attempt
// hangs or is dropped by an intermediary; the old single-shot fetch then left
// the caller with no metadata and the "Load Settings" button a silent no-op.
- const url = `${BASE}/api/v1/outputs/${encodeURIComponent(name)}/metadata`
+ const workspaceQuery = workspace
+ ? `?workspace=${encodeURIComponent(workspace)}`
+ : ''
+ const url = `${BASE}/api/v1/outputs/${encodeURIComponent(name)}/metadata${workspaceQuery}`
const ATTEMPTS = 3
const PER_ATTEMPT_MS = 30000 // generous: the server may read embedded video metadata to recover a seed
let lastErr: unknown = null
@@ -669,6 +882,126 @@ export async function fetchOutputMetadata(name: string): Promise {
+ const res = await fetch(`${BASE}/api/v1/style-library/sources`, { cache: 'no-store' })
+ if (!res.ok) throw new Error('Could not load style sources')
+ const data = await res.json()
+ return data.sources || []
+}
+
+export async function fetchStyleLibrary(params: {
+ modelFamily?: string
+ sourceId?: string
+ collection?: string
+ group?: string
+ query?: string
+ sort?: string
+ offset?: number
+ limit?: number
+} = {}): Promise {
+ const query = new URLSearchParams()
+ if (params.modelFamily) query.set('model_family', params.modelFamily)
+ if (params.sourceId) query.set('source_id', params.sourceId)
+ if (params.collection) query.set('collection', params.collection)
+ if (params.group) query.set('group', params.group)
+ if (params.query) query.set('q', params.query)
+ if (params.sort) query.set('sort', params.sort)
+ if (params.offset) query.set('offset', String(params.offset))
+ if (params.limit) query.set('limit', String(params.limit))
+ const res = await fetch(`${BASE}/api/v1/style-library/styles?${query.toString()}`, { cache: 'no-store' })
+ if (!res.ok) throw new Error('Could not load styles')
+ return res.json()
+}
+
+export async function startMiniMaxStyleImport(): Promise {
+ const res = await fetch(`${BASE}/api/v1/style-library/imports/minimax-h3-1k`, { method: 'POST' })
+ if (!res.ok) throw new Error('Could not start the MiniMax style download')
+ return res.json()
+}
+
+export async function fetchStyleImport(jobId: string): Promise {
+ const res = await fetch(`${BASE}/api/v1/style-library/imports/${encodeURIComponent(jobId)}`, { cache: 'no-store' })
+ if (!res.ok) throw new Error('Could not load style import progress')
+ return res.json()
+}
+
+export async function deleteStyle(styleId: string): Promise<{ id: string; deleted: boolean }> {
+ const res = await fetch(`${BASE}/api/v1/style-library/styles/${encodeURIComponent(styleId)}?confirm=true`, {
+ method: 'DELETE',
+ })
+ if (!res.ok) {
+ const detail = await res.json().catch(() => ({ detail: 'Could not delete style' }))
+ throw new Error(detail.detail || 'Could not delete style')
+ }
+ return res.json()
+}
+
export async function fetchVideoExtraInfo(
name: string,
language: string,
@@ -728,12 +1061,19 @@ export interface VideoEditorProbe {
export interface VideoEditorExportJob {
job_id: string
- status: 'queued' | 'running' | 'completed' | 'failed'
+ task_id?: string | null
+ root_task_id?: string | null
+ workspace?: string
+ status: 'queued' | 'waiting_resource' | 'running' | 'cancelling' | 'completed' | 'failed' | 'cancelled'
+ phase?: string
progress: number
message: string
filename: string | null
url: string | null
error: string | null
+ acquired_resources?: string[]
+ cancel_mode?: 'immediate' | 'deferred' | string
+ safe_boundary?: string
result?: { duration: number; clip_count: number }
}
@@ -785,6 +1125,7 @@ export async function startVideoEditorExport(payload: {
width: number
height: number
fps: number
+ workspace?: string
clips: Array<{
name: string
source: string
@@ -812,7 +1153,7 @@ export async function startVideoEditorExport(payload: {
transition_text: string
transition_text_size: number
}>
-}): Promise<{ job_id: string }> {
+}): Promise {
const res = await fetch(`${BASE}/api/v1/video-editor/export`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
@@ -834,6 +1175,17 @@ export async function fetchVideoEditorExport(jobId: string): Promise {
+ const res = await fetch(`${BASE}/api/v1/video-editor/export/${encodeURIComponent(jobId)}/cancel`, {
+ method: 'POST',
+ })
+ if (!res.ok) {
+ const error = await res.json().catch(() => ({ detail: 'Could not cancel export' }))
+ throw new Error(error.detail || 'Could not cancel export')
+ }
+ return res.json()
+}
+
export async function startComicAnimatic(payload: {
comic_id: string
comic_title: string
@@ -842,6 +1194,7 @@ export async function startComicAnimatic(payload: {
fps: number
transition: string
transition_duration: number
+ workspace?: string
panels: Array<{
source: string
page_number: number
@@ -850,7 +1203,7 @@ export async function startComicAnimatic(payload: {
motion: string
script: string
}>
-}): Promise<{ job_id: string }> {
+}): Promise {
const res = await fetch(`${BASE}/api/v1/comics/animatic`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
@@ -1262,6 +1615,28 @@ export async function tagPipelineClip(pid: string, clipIndex: number, tag: strin
if (!res.ok) throw new Error('Failed to tag clip')
}
+export async function selectPipelineClipVideo(
+ pid: string,
+ clipIndex: number,
+ filename: string,
+): Promise<{
+ pipeline_id: string
+ clip_index: number
+ filename: string
+ attempt: import('../types').PipelineVideoAttempt
+}> {
+ const res = await fetch(`${BASE}/api/v1/director/pipelines/${encodeURIComponent(pid)}/clips/${clipIndex}/video-selection`, {
+ method: 'PUT',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ filename }),
+ })
+ if (!res.ok) {
+ const err = await res.json().catch(() => ({ error: 'Clip selection failed' }))
+ throw new Error(err.error || err.detail || 'Could not select this clip version')
+ }
+ return res.json()
+}
+
export async function startPipelineRepair(pid: string): Promise<{
pipeline_id: string
repair: import('../types').PipelineRepairState
@@ -2105,6 +2480,49 @@ export async function generateComicWithMiniMax(params: {
return res.json()
}
+export interface MiniMaxImageJob {
+ jobId: string
+ workspace: string
+ status: 'queued' | 'waiting_resource' | 'running' | 'cancelling' | 'completed' | 'failed' | 'cancelled'
+ phase: string
+ message: string
+ current: number
+ total: number
+ progress: number
+ taskId?: string | null
+ rootTaskId?: string | null
+ statusCode?: number
+ error?: string | null
+ result?: { asset: import('../features/comics/types').ComicAsset } | null
+}
+
+export async function startMiniMaxImageJob(params: {
+ prompt: string
+ aspect_ratio: string
+ subject_reference?: string
+ workspace: string
+}): Promise {
+ const res = await fetch(`${BASE}/api/v1/comics/generate/minimax/jobs`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify(params),
+ })
+ if (!res.ok) {
+ const err = await res.json().catch(() => ({ detail: 'MiniMax generation failed' }))
+ throw new Error(`HTTP ${res.status}: ${err.detail || 'MiniMax generation failed'}`)
+ }
+ return res.json()
+}
+
+export async function fetchMiniMaxImageJob(jobId: string): Promise {
+ const res = await fetch(`${BASE}/api/v1/comics/generate/minimax/jobs/${encodeURIComponent(jobId)}`)
+ if (!res.ok) {
+ const err = await res.json().catch(() => ({ detail: 'MiniMax image job not found' }))
+ throw new Error(`HTTP ${res.status}: ${err.detail || 'MiniMax image job not found'}`)
+ }
+ return res.json()
+}
+
export async function generateStorySection(params: {
scope: import('../features/stories/types').StoryGenerationScope
premise: string
@@ -2166,28 +2584,32 @@ export async function generateStorySection(params: {
}, 1000)
signal?.addEventListener('abort', onAbort, { once: true })
})
- const statusResponse = await fetch(
- `${BASE}/api/v1/stories/generate/status/${encodeURIComponent(accepted.jobId)}`,
- { signal },
+ const status = await getStoryGenerationStatusResilient(
+ accepted.jobId,
+ signal,
+ (attempt, delayMs) => onProgress?.({
+ ...accepted,
+ status: 'running',
+ stage: 'reconnecting',
+ message: `Mobile connection interrupted; retrying in ${Math.round(delayMs / 1000)}s (attempt ${attempt})…`,
+ current: 0,
+ total: 0,
+ }),
)
- if (!statusResponse.ok) {
- const err = await statusResponse.json().catch(() => ({ detail: 'Could not read Story Lab job' }))
- throw new Error(err.detail || 'Could not read Story Lab job')
- }
- const status = await statusResponse.json()
onProgress?.(status)
if (status.status === 'failed' || status.status === 'cancelled') {
throw new Error(`${status.error || status.message} Resume job: ${accepted.jobId}`)
}
if (status.status === 'completed') {
- if (!status.result?.result) throw new Error('Story Lab job completed without a draft')
+ const result = status.result?.result
+ if (!result) throw new Error('Story Lab job completed without a draft')
window.localStorage.setItem('maestro-last-story-plan-result', JSON.stringify({
jobId: accepted.jobId,
projectId: params.project.id,
scope: params.scope,
- result: status.result.result,
+ result,
}))
- return status.result
+ return { result }
}
}
} finally {
@@ -2437,9 +2859,15 @@ export async function resumeSeriesPlanJob(jobId: string): Promise {
+export async function applySeriesPlanJob(
+ jobId: string,
+ episodeResult?: import('../features/series/types').SeriesEpisode,
+): Promise {
return seriesResponse(fetch(
- `${BASE}/api/v1/series/plan/jobs/${encodeURIComponent(jobId)}/apply`, { method: 'POST' },
+ `${BASE}/api/v1/series/plan/jobs/${encodeURIComponent(jobId)}/apply`, {
+ method: 'POST', headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify(episodeResult ? { episodeResult } : {}),
+ },
), 'Could not apply Series planning proposal')
}
@@ -2549,6 +2977,41 @@ export async function approveSeriesAttempt(
), 'Could not approve Series shot attempt')
}
+export async function approveSeriesAttemptsBulk(
+ workspace: string,
+ seriesId: string,
+ episodeId: string,
+ selections: Array<{ shotId: string; attemptId: string }>,
+): Promise<{ seriesId: string; episodeId: string; revision: number; episode: import('../features/series/types').SeriesEpisode }> {
+ return seriesResponse(fetch(
+ `${BASE}/api/v1/series/${encodeURIComponent(seriesId)}/episodes/${encodeURIComponent(episodeId)}/attempts/approve-bulk`,
+ {
+ method: 'POST', headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ workspace, selections }),
+ },
+ ), 'Could not approve Series shot attempts')
+}
+
+export async function startSeriesEpisodeAssembly(
+ workspace: string, seriesId: string, episodeId: string,
+): Promise {
+ return seriesResponse(fetch(
+ `${BASE}/api/v1/series/${encodeURIComponent(seriesId)}/episodes/${encodeURIComponent(episodeId)}/assembly/start`,
+ {
+ method: 'POST', headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ workspace }),
+ },
+ ), 'Could not start Series episode assembly')
+}
+
+export async function fetchSeriesEpisodeAssembly(
+ jobId: string,
+): Promise {
+ return seriesResponse(fetch(
+ `${BASE}/api/v1/series/assembly/jobs/${encodeURIComponent(jobId)}`,
+ ), 'Could not read Series episode assembly')
+}
+
export async function rejectSeriesAttempt(
workspace: string, seriesId: string, episodeId: string, shotId: string, attemptId: string,
): Promise {
@@ -2589,6 +3052,8 @@ export async function cancelStoryGeneration(jobId: string): Promise {
export interface StoryGenerationStatus {
jobId: string
+ taskId?: string | null
+ rootTaskId?: string | null
status: string
message: string
stage: string
@@ -2598,9 +3063,37 @@ export interface StoryGenerationStatus {
result?: { result?: Record } | null
}
-export async function getStoryGenerationStatus(jobId: string): Promise {
+const STORY_STATUS_RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 15_000, 15_000, 15_000, 15_000]
+
+function isStoryStatusNetworkError(error: unknown): boolean {
+ return error instanceof TypeError || (error instanceof Error && error.name === 'TypeError')
+}
+
+function waitForStoryStatusRetry(delayMs: number, signal?: AbortSignal): Promise {
+ return new Promise((resolve, reject) => {
+ if (signal?.aborted) {
+ reject(new DOMException('Story generation cancelled', 'AbortError'))
+ return
+ }
+ const onAbort = () => {
+ window.clearTimeout(timer)
+ reject(new DOMException('Story generation cancelled', 'AbortError'))
+ }
+ const timer = window.setTimeout(() => {
+ signal?.removeEventListener('abort', onAbort)
+ resolve()
+ }, delayMs)
+ signal?.addEventListener('abort', onAbort, { once: true })
+ })
+}
+
+export async function getStoryGenerationStatus(
+ jobId: string,
+ signal?: AbortSignal,
+): Promise {
const response = await fetch(
`${BASE}/api/v1/stories/generate/status/${encodeURIComponent(jobId)}`,
+ { signal },
)
if (!response.ok) {
const error = await response.json().catch(() => ({ detail: 'Could not read Story Lab job' }))
@@ -2609,6 +3102,27 @@ export async function getStoryGenerationStatus(jobId: string): Promise void,
+): Promise {
+ for (let attempt = 0; ; attempt += 1) {
+ try {
+ return await getStoryGenerationStatus(jobId, signal)
+ } catch (error) {
+ if (signal?.aborted) throw new DOMException('Story generation cancelled', 'AbortError')
+ if (!isStoryStatusNetworkError(error)) throw error
+ if (attempt >= STORY_STATUS_RETRY_DELAYS_MS.length) {
+ throw new Error(`Connection to Maestro is still unavailable. The job remains saved. Resume job: ${jobId}`)
+ }
+ const delayMs = STORY_STATUS_RETRY_DELAYS_MS[attempt]
+ onRetry?.(attempt + 1, delayMs)
+ await waitForStoryStatusRetry(delayMs, signal)
+ }
+ }
+}
+
export async function resumeStoryGeneration(
jobId: string,
onProgress?: (progress: {
@@ -2639,7 +3153,18 @@ export async function resumeStoryGeneration(
}
for (;;) {
await new Promise(resolve => window.setTimeout(resolve, 1000))
- const status = await getStoryGenerationStatus(jobId)
+ const status = await getStoryGenerationStatusResilient(
+ jobId,
+ undefined,
+ (attempt, delayMs) => onProgress?.({
+ jobId,
+ status: 'running',
+ stage: 'reconnecting',
+ message: `Mobile connection interrupted; retrying in ${Math.round(delayMs / 1000)}s (attempt ${attempt})…`,
+ current: 0,
+ total: 0,
+ }),
+ )
onProgress?.(status)
if (status.status === 'failed' || status.status === 'cancelled') {
throw new Error(status.error || status.message)
@@ -2657,6 +3182,8 @@ export async function resumeStoryGeneration(
export type ComicPlanProgress = {
jobId?: string
+ taskId?: string | null
+ rootTaskId?: string | null
status: 'queued' | 'loading_llm' | 'planning' | 'planning_bible' | 'planning_page' | 'completed' | 'failed'
message: string
provider?: string
@@ -3298,7 +3825,9 @@ export async function analyzeAudio(params: {
export interface AudioAnalysisJobStatus {
job_id: string
- status: 'queued' | 'running' | 'completed' | 'failed' | 'cancelled'
+ task_id?: string
+ root_task_id?: string
+ status: 'queued' | 'waiting_resource' | 'running' | 'cancelling' | 'completed' | 'failed' | 'cancelled'
progress: number
step: number
total_steps: number
@@ -3313,7 +3842,8 @@ export async function startAudioAnalysisJob(params: {
transcribe?: boolean
extract_vocals?: boolean
lyrics_hint?: string
-}): Promise<{ job_id: string }> {
+ workspace?: string
+}): Promise<{ job_id: string; task_id: string; root_task_id: string }> {
const res = await fetch(`${BASE}/api/v1/audio/analyze/jobs`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
diff --git a/ui/src/components/ActivityFooter.tsx b/ui/src/components/ActivityFooter.tsx
index 677aaaaf..bcd303cb 100644
--- a/ui/src/components/ActivityFooter.tsx
+++ b/ui/src/components/ActivityFooter.tsx
@@ -1,610 +1,409 @@
-import { useEffect, useMemo, useState } from 'react'
-import { AlertCircle, CheckCircle2, ChevronDown, ChevronUp, ListVideo, Loader2 } from 'lucide-react'
+import { useEffect, useMemo, useRef, useState } from 'react'
+import { AlertCircle, CheckCircle2, ChevronDown, ChevronUp, CircleSlash2, ListVideo, Loader2 } from 'lucide-react'
+import * as api from '../api/client'
+import type { CanonicalTask } from '../api/client'
+import { applyCanonicalTaskEvent, canResumeCanonicalTask, canonicalTaskVisualState, reconcileCanonicalTaskSnapshot } from '../lib/canonicalTaskEvents'
import { useStore } from '../stores/useStore'
-import type { GenerationDetails } from '../types'
+const ACTIVE = new Set(['created', 'queued', 'waiting_resource', 'running'])
+const CONNECTED_RECONCILE_MS = 60_000
+const DISCONNECTED_POLL_MS = 5_000
const PHASE_LABELS: Record = {
planning: 'Planning',
- writing_scenes: 'Writing scenes',
- writing_prompts: 'Writing prompts',
- polishing_prompts: 'Polishing prompts',
+ known_series_research: 'Building series bible',
+ canon: 'Preparing series canon',
+ outline: 'Writing outline',
+ script: 'Writing script',
+ shots: 'Planning shots',
+ canon_validation: 'Validating canon',
+ canon_delta: 'Preparing canon changes',
+ rendering: 'Rendering',
generating_images: 'Generating images',
- regenerating_styled_references: 'Regenerating styled references',
- preview_ready: 'Ready for review',
generating_video: 'Generating video',
post_processing: 'Post-processing',
- preparing_comic_video: 'Preparing comic video',
- uploading_artwork: 'Uploading artwork',
- rendering_animatic: 'Rendering animatic',
- story_planning: 'Story Lab planning',
- story_music: 'Story Lab music',
- music_planning: 'Planning music',
- writing_song: 'Writing song',
- generating_music: 'Generating music',
- music_queue: 'Music queue',
- uploading_music_reference: 'Uploading music reference',
- uploading_audio: 'Uploading audio',
- trimming_audio: 'Trimming audio',
- analyzing_audio: 'Analyzing audio',
- loading_audio: 'Loading audio',
- detecting_beats: 'Detecting beats',
- identifying_sections: 'Identifying sections',
- loading_vocal_model: 'Loading vocal model',
- extracting_vocals: 'Extracting vocals',
- loading_transcription_model: 'Loading transcription model',
- transcribing: 'Transcribing',
- loading_diarization_model: 'Loading speaker model',
- identifying_speakers: 'Identifying speakers',
- finalizing: 'Finalizing analysis',
- classifying_sections: 'Classifying song sections',
- planning_clips: 'Planning clips',
- ready_for_visual_brief: 'Ready for visual brief',
- preparing_music_video: 'Preparing music video',
+ waiting_resource: 'Waiting for resource',
+ cancelling: 'Cancelling at a safe boundary',
+ completed: 'Completed',
+ failed: 'Failed',
+ cancelled: 'Cancelled',
+ interrupted: 'Interrupted',
}
-type ActivityStatus = 'queued' | 'running' | 'completed' | 'failed'
-
-interface ActivityView {
- id: string
- title: string
- status: ActivityStatus
- phase: string
- message: string
- current: number
- total: number
- percent: number
- detailMessage?: string
- detailCurrent?: number
- detailTotal?: number
- resourceMessage?: string
- generationDetails?: GenerationDetails
- tokenUsage?: {
- promptTokens?: number
- completionTokens?: number
- totalTokens?: number
- calls?: number
- }
- startedAt?: number
- phaseStartedAt?: number
- phaseCurrent?: number
- phaseTotal?: number
- updatedAt: number
- dismissible?: 'activity' | 'job'
+function epochMs(value?: number | null): number | undefined {
+ if (!value || !Number.isFinite(value)) return undefined
+ return value < 1_000_000_000_000 ? value * 1000 : value
}
-function exactModelLabel(name?: string, modelType?: string): string {
- const cleanName = name?.trim()
- const cleanType = modelType?.trim()
- if (cleanName && cleanType && cleanName !== cleanType) return `${cleanName} (${cleanType})`
- return cleanName || cleanType || ''
+function elapsed(task: CanonicalTask, now: number): string {
+ const start = epochMs(task.started_at || task.queued_at || task.created_at)
+ if (!start) return ''
+ const end = ACTIVE.has(task.status)
+ ? now
+ : epochMs(task.completed_at || task.updated_at) || now
+ const seconds = Math.max(0, Math.floor((end - start) / 1000))
+ const hours = Math.floor(seconds / 3600)
+ const minutes = Math.floor((seconds % 3600) / 60)
+ const remainder = seconds % 60
+ return hours
+ ? `${hours}:${minutes.toString().padStart(2, '0')}:${remainder.toString().padStart(2, '0')}`
+ : `${minutes}:${remainder.toString().padStart(2, '0')}`
}
-function currentModelLabel(details?: GenerationDetails, phase = ''): string {
- if (!details) return ''
- if (phase.includes('image')) {
- return exactModelLabel(details.image_model_name, details.image_model_type)
- || exactModelLabel(details.model_name, details.model_type)
- }
- if (phase.includes('video') || phase.includes('render') || phase.includes('post_process')) {
- return exactModelLabel(details.video_model_name, details.video_model_type)
- || exactModelLabel(details.model_name, details.model_type)
- }
- if (phase.includes('planning') || phase.includes('writing') || phase.includes('prompt')) {
- return [details.text_provider, details.text_model].filter(Boolean).join(' / ')
- || exactModelLabel(details.model_name, details.model_type)
- }
- return exactModelLabel(details.model_name, details.model_type)
- || exactModelLabel(details.video_model_name, details.video_model_type)
- || exactModelLabel(details.image_model_name, details.image_model_type)
- || [details.text_provider, details.text_model].filter(Boolean).join(' / ')
+function percent(task: CanonicalTask): number {
+ if (task.total > 0) return Math.max(0, Math.min(100, (task.current / task.total) * 100))
+ return Math.max(0, Math.min(100, Number(task.progress || 0) * 100))
}
-function humanReadableActivityMessage(row: ActivityView): string {
- if (row.status === 'running' && row.id.startsWith('pipeline:') && row.phase === 'planning') {
- const planner = [row.generationDetails?.text_provider, row.generationDetails?.text_model]
- .filter(Boolean)
- .join(' / ')
- const clipCount = row.generationDetails?.clip_count
- const target = clipCount ? `${clipCount} timed shot${clipCount === 1 ? '' : 's'}` : 'the timed shot plan'
- return `${planner || 'The planning LLM'} is writing ${target}: scene, action, camera and final generation prompt. Waiting for the remote response; image and video generation have not started.`
- }
- return row.detailMessage || row.message
+function phaseLabel(task: CanonicalTask): string {
+ return PHASE_LABELS[task.phase] || task.phase?.replaceAll('_', ' ') || task.status
}
-function generationRecipe(details?: GenerationDetails, phase = ''): string {
- if (!details) return ''
- const parts: string[] = []
- const currentModel = currentModelLabel(details, phase)
- if (currentModel) parts.push(`Using: ${currentModel}`)
-
- const imageModel = exactModelLabel(details.image_model_name, details.image_model_type)
- const videoModel = exactModelLabel(details.video_model_name, details.video_model_type)
- const textModel = [details.text_provider, details.text_model].filter(Boolean).join(' / ')
- if (textModel && textModel !== currentModel) parts.push(`text ${textModel}`)
- if (imageModel && imageModel !== currentModel) parts.push(`image ${imageModel}`)
- if (videoModel && videoModel !== currentModel) parts.push(`video ${videoModel}`)
+function resources(task: CanonicalTask): string {
+ const acquired = task.acquired_resources || []
+ const required = task.resource_requirements || []
+ if (acquired.length) return `Using ${acquired.join(' · ')}`
+ if (task.status === 'waiting_resource' && required.length) return `Waiting for ${required.join(' · ')}`
+ return required.length ? `Resources ${required.join(' · ')}` : ''
+}
- const resolution = phase.includes('image')
- ? details.image_resolution || details.resolution
- : phase.includes('video') || videoModel
- ? details.video_resolution || details.resolution
- : details.resolution || details.image_resolution
- const steps = phase.includes('image')
- ? details.image_steps || details.steps
- : phase.includes('video') || videoModel
- ? details.video_steps || details.steps
- : details.steps || details.image_steps
+function generationRecipe(task: CanonicalTask): string {
+ const metadata = task.metadata || {}
+ const details = (metadata.generation_details || metadata.settings || {}) as Record
+ const parts = [task.provider, task.model].filter(Boolean) as string[]
+ const addModel = (label: string, value: unknown) => {
+ if (!value) return
+ const model = String(value)
+ if (!parts.some(part => part === model || part.endsWith(` ${model}`))) {
+ parts.push(label ? `${label} ${model}` : model)
+ }
+ }
+ addModel('', details.model_name || details.model_type)
+ addModel('text', details.text_model)
+ addModel('image', details.image_model_name || details.image_model_type)
+ addModel('video', details.video_model_name || details.video_model_type)
+ const resolution = details.video_resolution || details.image_resolution || details.resolution
+ const seed = details.seed
+ const steps = details.video_steps || details.image_steps || details.steps || details.numInferenceSteps
if (resolution) parts.push(String(resolution))
- if (details.seed !== undefined) parts.push(`seed ${details.seed}`)
+ if (seed !== undefined) parts.push(`seed ${seed}`)
if (steps !== undefined) parts.push(`${steps} steps`)
if (details.guidance !== undefined) parts.push(`guidance ${details.guidance}`)
if (details.frames !== undefined) parts.push(`${details.frames} frames`)
if (details.duration_seconds !== undefined) parts.push(`${details.duration_seconds}s`)
- if (details.repeat !== undefined && details.repeat > 1) parts.push(`${details.repeat} outputs`)
- if (details.clip_count !== undefined) parts.push(`${details.clip_count} clips`)
+ if (details.dialogue_syllables !== undefined) {
+ parts.push(
+ `dialogue ${details.dialogue_syllables} syllables × ${details.dialogue_seconds_per_syllable}s → ${details.dialogue_duration_calculated}s calculated`
+ + (details.dialogue_duration_minimum_limited ? ' · H3 minimum applied' : ''),
+ )
+ } else if (details.dialogue_words !== undefined) {
+ parts.push(
+ `dialogue ${details.dialogue_words} words → ${details.dialogue_duration_calculated}s calculated`
+ + (details.dialogue_duration_minimum_limited ? ' · H3 minimum applied' : ''),
+ )
+ }
if (details.profile) parts.push(`profile ${details.profile}`)
- if (details.flow_shift !== undefined) parts.push(`flow shift ${details.flow_shift}`)
- if (details.audio_shift !== undefined) parts.push(`audio shift ${details.audio_shift}`)
+ if (details.flow_shift !== undefined || details.flowShift !== undefined) {
+ parts.push(`flow shift ${details.flow_shift ?? details.flowShift}`)
+ }
+ if (details.audio_shift !== undefined || details.audioShift !== undefined) {
+ parts.push(`audio shift ${details.audio_shift ?? details.audioShift}`)
+ }
if (details.turbo !== undefined) parts.push(`Turbo ${details.turbo ? 'on' : 'off'}`)
- return parts.join(' · ')
-}
-
-function generationTitle(details?: GenerationDetails): string {
- switch (details?.generation_mode) {
- case 'image': return 'Image generation'
- case 'video': return 'Video generation'
- case 'audio':
- case 'music': return 'Music generation'
- case 'model3d': return '3D generation'
- case 'avatar': return 'Video edit'
- default: return 'Generation job'
+ if (details.cache !== undefined) {
+ parts.push(details.cache
+ ? `Cache on${details.cache_type ? ` (${details.cache_type})` : ''}`
+ : 'Cache off')
}
-}
-
-function resourceMessage(schedule?: import('../api/client').PipelineResourceSchedule): string | undefined {
- if (!schedule?.lanes) return undefined
- const planning = schedule.lanes.planning?.label
- const images = schedule.lanes.images?.label
- const video = schedule.lanes.video?.label
- if (schedule.mode === 'remote-images+local-video') {
- const ready = schedule.images_total
- ? ` · images ${schedule.images_ready || 0}/${schedule.images_total}`
- : ''
- return `Parallel resources · ${images} → ${video}${ready}`
+ if (details.lora_count !== undefined) {
+ const loras = Array.isArray(details.loras) ? details.loras.map(String).filter(Boolean) : []
+ parts.push(details.lora_count
+ ? `${details.lora_count} LoRA${Number(details.lora_count) === 1 ? '' : 's'}${loras.length ? ` (${loras.join(', ')})` : ''}`
+ : 'LoRAs off')
}
- return `Resources · planning: ${planning || 'unknown'} · images: ${images || 'unknown'} · video: ${video || 'unknown'}`
-}
-
-function clampPercent(value: number): number {
- return Math.max(0, Math.min(100, Number.isFinite(value) ? value : 0))
-}
-
-function activityProgress(current: number, total: number, explicit?: number): number {
- if (total > 0) return clampPercent((current / total) * 100)
- return clampPercent((explicit || 0) * (explicit && explicit <= 1 ? 100 : 1))
-}
-
-function epochMilliseconds(value?: number): number | undefined {
- if (!value || !Number.isFinite(value)) return undefined
- return value < 1_000_000_000_000 ? value * 1000 : value
-}
-
-function formatElapsed(milliseconds: number): string {
- const seconds = Math.max(0, Math.floor(milliseconds / 1000))
- const hours = Math.floor(seconds / 3600)
- const minutes = Math.floor((seconds % 3600) / 60)
- const remainder = seconds % 60
- return hours > 0
- ? `${hours}:${minutes.toString().padStart(2, '0')}:${remainder.toString().padStart(2, '0')}`
- : `${minutes}:${remainder.toString().padStart(2, '0')}`
-}
-
-function formatEstimate(milliseconds: number): string {
- const minutes = Math.max(1, Math.ceil(milliseconds / 60_000))
- if (minutes < 60) return `${minutes}m`
- const hours = Math.floor(minutes / 60)
- const remainder = minutes % 60
- return remainder ? `${hours}h ${remainder}m` : `${hours}h`
-}
-
-function estimatedRemaining(row: ActivityView, now: number): string {
- if (row.status !== 'running') return ''
- const current = row.phaseCurrent ?? row.current
- const total = row.phaseTotal ?? row.total
- const startedAt = row.phaseStartedAt || row.startedAt
- if (!startedAt || current <= 0 || total <= current) return ''
- const elapsed = now - startedAt
- // A first very short sample is too noisy to be useful.
- if (elapsed < 10_000) return ''
- const remaining = (elapsed / current) * (total - current)
- if (!Number.isFinite(remaining) || remaining <= 0) return ''
- return `ETA ~${formatEstimate(remaining)}`
+ if (details.clip_count !== undefined) parts.push(`${details.clip_count} clips`)
+ return parts.join(' · ')
}
-/**
- * App-wide activity readout. Durable generation jobs and Director pipelines
- * are normalized together with user-visible foreground workflows, but each
- * row keeps its own message and progress so concurrent work is never mixed.
- */
export function ActivityFooter() {
- const jobs = useStore(s => s.jobs)
- const pipelineStatus = useStore(s => s.pipelineStatus)
- const activeDirectorPipelines = useStore(s => s.activeDirectorPipelines)
- const activities = useStore(s => s.activities)
- const stopGeneration = useStore(s => s.stopGeneration)
- const stopPipeline = useStore(s => s.stopPipeline)
- const removeActivity = useStore(s => s.removeActivity)
- const dismissJob = useStore(s => s.dismissJob)
- const setVideoWorkflowsOpen = useStore(s => s.setDashboardOpen)
+ const activeWorkspace = useStore(state => state.activeWorkspace)
+ const setVideoWorkflowsOpen = useStore(state => state.setDashboardOpen)
+ const [tasks, setTasks] = useState([])
+ const tasksRef = useRef([])
const [detailsOpen, setDetailsOpen] = useState(false)
- const [cancellingIds, setCancellingIds] = useState>(() => new Set())
- const [clock, setClock] = useState(() => Date.now())
+ const [clock, setClock] = useState(Date.now())
+ const [busyIds, setBusyIds] = useState>(() => new Set())
- const rows = useMemo(() => {
- const registered = Object.values(activities).map(activity => ({
- id: activity.id,
- title: activity.title || 'Maestro',
- status: activity.status,
- phase: activity.phase,
- message: activity.error || activity.message,
- current: activity.current || 0,
- total: activity.total || 0,
- percent: activityProgress(activity.current || 0, activity.total || 0, activity.progress),
- detailMessage: activity.detailMessage,
- detailCurrent: activity.detailCurrent,
- detailTotal: activity.detailTotal,
- generationDetails: activity.generationDetails,
- tokenUsage: activity.tokenUsage,
- startedAt: activity.startedAt,
- updatedAt: activity.updatedAt || activity.startedAt || 3,
- dismissible: activity.status === 'failed' ? 'activity' as const : undefined,
- }))
+ useEffect(() => {
+ let mounted = true
+ let refreshPending = false
+ let streamConnected = false
+ let pollTimer: number | null = null
+ let closeEvents: () => void = () => undefined
+ let unknownTaskBaseline = 0
- const recoveredPipelines: ActivityView[] = activeDirectorPipelines.map(pipeline => ({
- id: `pipeline:${pipeline.id}`,
- title: pipeline.pipeline_type === 'music_video' ? 'Music video' : 'Director pipeline',
- status: pipeline.status === 'paused' ? 'queued' : 'running',
- phase: pipeline.phase,
- message: pipeline.error || pipeline.progress?.message || 'Director is working…',
- current: pipeline.progress?.total_steps ? pipeline.progress.step : pipeline.progress?.current || 0,
- total: pipeline.progress?.total_steps || pipeline.progress?.total || 0,
- percent: activityProgress(
- pipeline.progress?.total_steps ? pipeline.progress.step : pipeline.progress?.current || 0,
- pipeline.progress?.total_steps || pipeline.progress?.total || 0,
- ),
- resourceMessage: resourceMessage(pipeline.resource_schedule),
- generationDetails: pipeline.generation_details,
- startedAt: epochMilliseconds(pipeline.created_at),
- phaseStartedAt: epochMilliseconds(pipeline.phase_started_at),
- phaseCurrent: pipeline.progress?.current || 0,
- phaseTotal: pipeline.progress?.total || 0,
- updatedAt: epochMilliseconds(pipeline.updated_at || pipeline.created_at) || 2,
- }))
+ const commitTasks = (next: CanonicalTask[]) => {
+ tasksRef.current = next
+ setTasks(next)
+ }
+ const refresh = async () => {
+ if (refreshPending) return
+ refreshPending = true
+ try {
+ const result = await api.fetchCanonicalTasks(activeWorkspace, 'all')
+ if (mounted) {
+ const snapshotBoundary = Math.max(
+ unknownTaskBaseline,
+ ...result.tasks.map(task => Number(task.updated_at || 0)),
+ )
+ unknownTaskBaseline = snapshotBoundary
+ commitTasks(reconcileCanonicalTaskSnapshot(
+ tasksRef.current,
+ result.tasks,
+ snapshotBoundary,
+ ))
+ }
+ } catch {
+ // Adaptive polling below remains the fallback during a restart.
+ } finally {
+ refreshPending = false
+ }
+ }
- const pipeline: ActivityView[] = pipelineStatus
- && ['running', 'failed', 'completed'].includes(pipelineStatus.status)
- && !activeDirectorPipelines.some(pipeline => pipeline.id === pipelineStatus.id)
- ? [{
- id: `pipeline:${pipelineStatus.id}`,
- title: 'Director pipeline',
- status: pipelineStatus.status === 'failed'
- ? 'failed'
- : pipelineStatus.status === 'completed' ? 'completed' : 'running',
- phase: pipelineStatus.phase,
- message: pipelineStatus.error || pipelineStatus.progress?.message || 'Director is working…',
- current: pipelineStatus.progress?.total_steps
- ? pipelineStatus.progress.step
- : pipelineStatus.progress?.current || 0,
- total: pipelineStatus.progress?.total_steps || pipelineStatus.progress?.total || 0,
- percent: activityProgress(
- pipelineStatus.progress?.total_steps ? pipelineStatus.progress.step : pipelineStatus.progress?.current || 0,
- pipelineStatus.progress?.total_steps || pipelineStatus.progress?.total || 0,
- ),
- resourceMessage: resourceMessage(pipelineStatus.resource_schedule),
- generationDetails: pipelineStatus.generation_details,
- startedAt: epochMilliseconds(pipelineStatus.created_at),
- phaseStartedAt: epochMilliseconds(pipelineStatus.phase_started_at),
- phaseCurrent: pipelineStatus.progress?.current || 0,
- phaseTotal: pipelineStatus.progress?.total || 0,
- updatedAt: epochMilliseconds(pipelineStatus.updated_at || pipelineStatus.created_at) || 2,
- }]
- : []
+ const schedulePoll = () => {
+ if (!mounted) return
+ if (pollTimer !== null) window.clearTimeout(pollTimer)
+ pollTimer = window.setTimeout(async () => {
+ pollTimer = null
+ await refresh()
+ schedulePoll()
+ }, streamConnected ? CONNECTED_RECONCILE_MS : DISCONNECTED_POLL_MS)
+ }
- const visibleJobs = jobs
- .filter((job, index) => !activities[job.id]
- && (job.status === 'running'
- || job.status === 'queued'
- || (index === 0 && (job.status === 'completed' || job.status === 'failed'))))
- .map((job): ActivityView => ({
- id: `job:${job.id}`,
- title: generationTitle(job.generationDetails),
- status: job.status === 'failed'
- ? 'failed'
- : job.status === 'completed' || job.status === 'cancelled'
- ? 'completed'
- : job.status === 'queued' ? 'queued' : 'running',
- phase: job.phase,
- message: job.error
- || (job.status === 'queued' && job.queuePosition
- ? `Queued · position ${job.queuePosition}`
- : job.message)
- || (job.status === 'queued' ? 'Queued' : 'Generation is running…'),
- current: job.totalSteps ? job.step : 0,
- total: job.totalSteps || 0,
- percent: activityProgress(job.step, job.totalSteps, job.progress),
- generationDetails: job.generationDetails,
- // Ordinary jobs deliberately omit queue wait from their timer. A
- // Director/music-video pipeline above keeps created_at as its total
- // workflow clock, including planning, generation and assembly.
- startedAt: job.startedAt,
- updatedAt: job.finishedAt || job.startedAt || job.createdAt || 1,
- dismissible: job.status === 'failed' ? 'job' as const : undefined,
- }))
+ tasksRef.current = []
+ setTasks([])
+ void refresh().finally(() => {
+ if (!mounted) return
+ closeEvents = api.subscribeCanonicalTaskEvents(
+ activeWorkspace,
+ event => {
+ const result = applyCanonicalTaskEvent(tasksRef.current, event, unknownTaskBaseline)
+ if (result.tasks !== tasksRef.current) commitTasks(result.tasks)
+ if (result.needsRefresh) void refresh()
+ },
+ () => undefined,
+ state => {
+ if (!mounted) return
+ streamConnected = state === 'open'
+ schedulePoll()
+ },
+ )
+ schedulePoll()
+ })
+ return () => {
+ mounted = false
+ closeEvents()
+ if (pollTimer !== null) window.clearTimeout(pollTimer)
+ }
+ }, [activeWorkspace])
- return [...registered, ...recoveredPipelines, ...pipeline, ...visibleJobs]
- .sort((left, right) => right.updatedAt - left.updatedAt)
- }, [activities, jobs, pipelineStatus, activeDirectorPipelines])
+ const roots = useMemo(() => {
+ const rootTasks = tasks.filter(task => !task.parent_id)
+ const active = rootTasks.filter(task => ACTIVE.has(task.status))
+ .sort((left, right) => right.updated_at - left.updated_at)
+ const recent = rootTasks.filter(task => !ACTIVE.has(task.status))
+ .sort((left, right) => right.updated_at - left.updated_at)
+ .slice(0, 12)
+ return [...active, ...recent]
+ }, [tasks])
+ const childrenByRoot = useMemo(() => {
+ const result = new Map()
+ for (const task of tasks) {
+ if (!task.parent_id) continue
+ const children = result.get(task.root_id) || []
+ children.push(task)
+ result.set(task.root_id, children)
+ }
+ for (const children of result.values()) children.sort((a, b) => a.created_at - b.created_at)
+ return result
+ }, [tasks])
+ const activeTasks = roots.filter(task => ACTIVE.has(task.status))
+ const failedTasks = roots.filter(task => task.status === 'failed' || task.status === 'interrupted')
+ const primary = activeTasks[0] || failedTasks[0] || roots[0] || null
- const activeRows = rows
- .filter(row => row.status === 'running' || row.status === 'queued')
- .sort((left, right) => {
- if (left.status !== right.status) return left.status === 'running' ? -1 : 1
- return right.updatedAt - left.updatedAt
- })
- const failedRows = rows.filter(row => row.status === 'failed')
- const completedRows = rows.filter(row => row.status === 'completed')
- // Prefer a cancellable backend job/pipeline over a foreground wrapper. A
- // wrapper may be newer because it mirrors the same child progress, but it
- // cannot stop the GPU worker itself and used to hide the useful Cancel.
- const primary = activeRows.find(row => (
- row.id.startsWith('job:')
- || row.id.startsWith('audio-analysis-')
- || row.id.startsWith('pipeline:')
- )) || activeRows[0] || failedRows[0] || completedRows[0] || null
- const isActive = activeRows.length > 0
- const hasError = !isActive && failedRows.length > 0
- const phase = primary
- ? PHASE_LABELS[primary.phase] || primary.phase?.replaceAll('_', ' ')
- : ''
- const message = primary ? humanReadableActivityMessage(primary) : 'Ready — no active jobs'
- const primaryModel = primary ? currentModelLabel(primary.generationDetails, primary.phase) : ''
useEffect(() => {
- if (!activeRows.length) return
- const interval = window.setInterval(() => setClock(Date.now()), 1000)
- return () => window.clearInterval(interval)
- }, [activeRows.length])
- const elapsed = (row: ActivityView) => row.startedAt
- ? formatElapsed((row.status === 'running' || row.status === 'queued' ? clock : row.updatedAt) - row.startedAt)
- : ''
- const phaseElapsed = (row: ActivityView) => row.phaseStartedAt
- ? formatElapsed((row.status === 'running' || row.status === 'queued' ? clock : row.updatedAt) - row.phaseStartedAt)
- : ''
- const canCancel = (row: ActivityView) => (
- (row.status === 'running' || row.status === 'queued')
- && (row.id.startsWith('job:') || row.id.startsWith('audio-analysis-') || row.id.startsWith('pipeline:'))
- )
- const cancelRow = (row: ActivityView) => {
- if (!canCancel(row) || cancellingIds.has(row.id)) return
- setCancellingIds(current => new Set(current).add(row.id))
- const operation = row.id.startsWith('pipeline:')
- ? stopPipeline(row.id.slice('pipeline:'.length))
- : Promise.resolve(stopGeneration(row.id.startsWith('job:') ? row.id.slice(4) : row.id))
- void operation.catch(error => {
- console.error('Failed to cancel activity:', error)
- }).finally(() => {
- setCancellingIds(current => {
+ if (!activeTasks.length) return
+ const timer = window.setInterval(() => setClock(Date.now()), 1000)
+ return () => window.clearInterval(timer)
+ }, [activeTasks.length])
+
+ const runControl = (task: CanonicalTask, action: 'cancel' | 'resume' | 'dismiss') => {
+ if (busyIds.has(task.id)) return
+ setBusyIds(current => new Set(current).add(task.id))
+ const operation = action === 'cancel'
+ ? api.cancelCanonicalTask(task.id, activeWorkspace)
+ : action === 'resume'
+ ? api.resumeCanonicalTask(task.id, activeWorkspace)
+ : api.dismissCanonicalTask(task.id, activeWorkspace)
+ void operation.then(result => {
+ const next = action === 'dismiss'
+ ? tasksRef.current.filter(item => item.id !== task.id)
+ : tasksRef.current.map(item => item.id === task.id ? result as CanonicalTask : item)
+ tasksRef.current = next
+ setTasks(next)
+ }).catch(error => console.error(`Failed to ${action} Maestro task`, error)).finally(() => {
+ setBusyIds(current => {
const next = new Set(current)
- next.delete(row.id)
+ next.delete(task.id)
return next
})
})
}
+ const copyId = (task: CanonicalTask) => {
+ void navigator.clipboard?.writeText(task.id)
+ }
+
+ const isActive = activeTasks.length > 0
+ const hasError = !isActive && failedTasks.length > 0
+ const primaryVisualState = primary ? canonicalTaskVisualState(primary.status) : 'neutral'
+ const primaryMessage = primary?.error?.message || primary?.detail || primary?.message || 'Ready — no active jobs'
+
return (