From 1052da58ccc37a6c1f367eda75120aa676b2a18d Mon Sep 17 00:00:00 2001 From: Gateway <519563+gateway@users.noreply.github.com> Date: Fri, 23 Oct 2026 22:18:31 +0700 Subject: [PATCH 1/3] Keep Graph recovery and execution under one run owner --- apps/api/app/graph/runtime.py | 66 +++- .../tests/test_graph_recovery_ownership.py | 315 ++++++++++++++++++ docs/runtime-and-supervision.md | 9 + scripts/static-analysis/full-baseline.json | 60 ++-- scripts/static-analysis/python-baseline.json | 80 ++--- 5 files changed, 440 insertions(+), 90 deletions(-) create mode 100644 apps/api/tests/test_graph_recovery_ownership.py diff --git a/apps/api/app/graph/runtime.py b/apps/api/app/graph/runtime.py index 24d35d52..7e5a1116 100644 --- a/apps/api/app/graph/runtime.py +++ b/apps/api/app/graph/runtime.py @@ -316,6 +316,10 @@ class GraphRuntime: def __init__(self) -> None: # Cancellation handlers and workers share finalization in this process. self._cancellation_lock = threading.RLock() + # The supported installation has one API process. A restart drops these + # reservations so persisted orphaned runs can be recovered by its successor. + self._execution_lock = threading.Lock() + self._active_runs: set[str] = set() executors: List[GraphExecutor] = [ PromptTextExecutor(), PromptConcatExecutor(), @@ -396,14 +400,42 @@ def create_run(self, workflow_id: str, workflow: GraphWorkflow, *, start: bool = return self._shape_run(run) def start_run(self, run_id: str) -> None: + if not self._claim_run(run_id): + return + try: + self._start_run_worker(run_id) + except Exception: + self._release_run(run_id) + raise + + def _claim_run(self, run_id: str) -> bool: + with self._execution_lock: + if run_id in self._active_runs: + return False + self._active_runs.add(run_id) + return True + + def _release_run(self, run_id: str) -> None: + with self._execution_lock: + self._active_runs.remove(run_id) + + def _start_run_worker(self, run_id: str, *, resume: bool = False, start: bool = True) -> None: + if not start: + return thread = threading.Thread( - target=self.execute_run, - args=(run_id,), - name=f"graph-run-{run_id}", + target=self._execute_owned_run, + kwargs={"run_id": run_id, "resume": resume}, + name=f"graph-{'recover' if resume else 'run'}-{run_id}", daemon=True, ) thread.start() + def _execute_owned_run(self, run_id: str, *, resume: bool = False) -> None: + try: + self._execute_run(run_id, resume=resume) + finally: + self._release_run(run_id) + def recover_interrupted_runs(self, *, start: bool = True, limit: int = 100) -> int: recovered = 0 for run in store.list_graph_runs(limit=limit): @@ -423,8 +455,22 @@ def recover_interrupted_runs(self, *, start: bool = True, limit: int = 100) -> i return recovered def recover_run(self, run_id: str, *, start: bool = True) -> Dict[str, object]: + if not self._claim_run(run_id): + return {"recovered": False, "started": False, "run_id": run_id} + worker_started = False + try: + # Recovery preparation must not overwrite an acknowledged cancellation. + with self._cancellation_lock: + result = self._recover_run(run_id, start=start) + worker_started = bool(result["started"]) + return result + finally: + if not worker_started: + self._release_run(run_id) + + def _recover_run(self, run_id: str, *, start: bool) -> Dict[str, object]: run = store.get_graph_run(run_id) - if not run or not _is_interrupted_run(run): + if not run or not _is_interrupted_run(run) or self._finish_cancelled_run_if_requested(run_id): return {"recovered": False, "started": False, "run_id": run_id} workflow = materialize_workflow_defaults(GraphWorkflow(**run["workflow_json"])) compiled = compile_workflow(workflow) @@ -527,9 +573,7 @@ def recover_run(self, run_id: str, *, start: bool = True) -> Dict[str, object]: existing_recovery_metrics = run.get("metrics_json") if isinstance(run.get("metrics_json"), dict) else {} if not recovered_node_ids and not resumable_node_ids and existing_recovery_metrics.get("recovered_from_interruption") is True: - if start: - thread = threading.Thread(target=self.execute_run, kwargs={"run_id": run_id, "resume": True}, name=f"graph-recover-{run_id}", daemon=True) - thread.start() + self._start_run_worker(run_id, resume=True, start=start) return { "recovered": True, "started": start, @@ -579,9 +623,7 @@ def recover_run(self, run_id: str, *, start: bool = True) -> Dict[str, object]: "terminal_provider_failures": terminal_provider_failures, }, ) - if start: - thread = threading.Thread(target=self.execute_run, kwargs={"run_id": run_id, "resume": True}, name=f"graph-recover-{run_id}", daemon=True) - thread.start() + self._start_run_worker(run_id, resume=True, start=start) return { "recovered": True, "started": start, @@ -592,6 +634,10 @@ def recover_run(self, run_id: str, *, start: bool = True) -> Dict[str, object]: } def execute_run(self, run_id: str, *, resume: bool = False) -> None: + if self._claim_run(run_id): + self._execute_owned_run(run_id, resume=resume) + + def _execute_run(self, run_id: str, *, resume: bool) -> None: run = store.get_graph_run(run_id) if not run: return diff --git a/apps/api/tests/test_graph_recovery_ownership.py b/apps/api/tests/test_graph_recovery_ownership.py new file mode 100644 index 00000000..9c2eea0b --- /dev/null +++ b/apps/api/tests/test_graph_recovery_ownership.py @@ -0,0 +1,315 @@ +"""Recovery ownership through the real router/runtime/store; normal CI fixtures only.""" +from concurrent.futures import ThreadPoolExecutor +import copy +import importlib +import threading +from types import SimpleNamespace + +import pytest + + +def _prepare(client, app_modules, monkeypatch, *, interrupted=True): + module = importlib.import_module("app.graph.runtime") + from app.graph.schemas import GraphWorkflow + + workflow = GraphWorkflow(name="Recovery ownership", nodes=[ + {"id": "upstream", "type": "model.kie.nano_banana_pro", "fields": {"prompt": "Existing job", "resolution": "1K"}}, + {"id": "downstream", "type": "model.kie.nano_banana_pro", "fields": {"prompt": "One execution", "resolution": "1K"}}, + ]) + saved = client.post("/media/graph/workflows", json=workflow.model_dump(mode="json")) + assert saved.status_code == 200, saved.text + runtime, store = module.runtime, app_modules["store"] + run = runtime.create_run(saved.json()["workflow_id"], workflow, start=False) + batch, jobs = store.create_batch_and_jobs( + {"status": "processing", "model_key": "nano-banana-pro", "task_mode": "text_to_image"}, + [{"model_key": "nano-banana-pro", "task_mode": "text_to_image", "status": "running", "provider_task_id": "ownership-existing-job"}], + ) + job_id, batch_id = jobs[0]["job_id"], batch["batch_id"] + store.append_graph_run_event(run.run_id, "kie.submitted", {"job_id": job_id, "batch_id": batch_id}, node_id="upstream") + if interrupted: + store.update_graph_run(run.run_id, {"status": "failed", "error": "Graph run was interrupted before completion."}) + store.update_graph_run_node(run.run_id, "upstream", {"status": "failed", "metrics_json": {"job_id": job_id, "batch_id": batch_id}}) + calls, adoptions, workers = [], [], [] + entered, release = threading.Event(), threading.Event() + + def adopt(**kwargs): + adoptions.append(kwargs["job_id"]) + return {} + + def execute(node, context): + calls.append(node.id) + return {} + + class Worker(threading.Thread): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + workers.append(self) + + monkeypatch.setattr(module, "threading", SimpleNamespace(Thread=Worker, Lock=threading.Lock, RLock=threading.RLock)) + monkeypatch.setattr(module, "wait_for_existing_kie_job", adopt) + monkeypatch.setattr(runtime.executors["model.kie"], "execute", execute) + return SimpleNamespace(module=module, runtime=runtime, store=store, run_id=run.run_id, job_id=job_id, + calls=calls, adoptions=adoptions, workers=workers, entered=entered, release=release) + + +def _snapshot(case): + return copy.deepcopy({"run": case.store.get_graph_run(case.run_id), "nodes": case.store.list_graph_run_nodes(case.run_id), + "events": case.store.list_graph_run_events(case.run_id)}) + + +def _recover(client, case): + response = client.post(f"/media/graph/runs/{case.run_id}/recover") + assert response.status_code == 200, response.text + return response.json() + + +def _finish(case): + case.release.set() + for worker in case.workers: + worker.join(10) + assert not any(worker.is_alive() for worker in case.workers) + + +def _assert_one_result(case): + result = case.store.get_graph_run(case.run_id) + events = case.store.list_graph_run_events(case.run_id) + assert result["status"] == "completed" + assert case.calls.count("downstream") == 1 + assert sum(event["event_type"] == "run.completed" for event in events) == 1 + assert sum(event["event_type"] == "node.completed" and event["node_id"] == "downstream" for event in events) == 1 + + +@pytest.mark.parametrize("phase", ["polling", "downstream"]) +def test_repeated_and_concurrent_recovery_keeps_one_worker(client, app_modules, monkeypatch, phase): + case = _prepare(client, app_modules, monkeypatch) + + def adopt(**kwargs): + case.adoptions.append(kwargs["job_id"]) + if phase == "polling": + case.entered.set() + assert case.release.wait(10) + return {} + + def execute(node, context): + case.calls.append(node.id) + if phase == "downstream": + case.entered.set() + assert case.release.wait(10) + return {} + + monkeypatch.setattr(case.module, "wait_for_existing_kie_job", adopt) + monkeypatch.setattr(case.runtime.executors["model.kie"], "execute", execute) + try: + _recover(client, case) + assert case.entered.wait(10) + before = _snapshot(case) + _recover(client, case) + barrier = threading.Barrier(4) + + def recover_together(): + barrier.wait(10) + return _recover(client, case) + + with ThreadPoolExecutor(max_workers=3) as pool: + requests = [pool.submit(recover_together) for _ in range(3)] + barrier.wait(10) + for request in requests: + request.result(10) + assert _snapshot(case) == before + assert len(case.workers) == 1 + finally: + _finish(case) + assert case.adoptions == [case.job_id] + assert case.calls == ["downstream"] + _assert_one_result(case) + + +def test_recovery_does_not_touch_an_active_original_worker(client, app_modules, monkeypatch): + case = _prepare(client, app_modules, monkeypatch, interrupted=False) + + def execute(node, context): + case.calls.append(node.id) + if node.id == "upstream": + case.entered.set() + assert case.release.wait(10) + return {} + + monkeypatch.setattr(case.runtime.executors["model.kie"], "execute", execute) + try: + case.runtime.start_run(case.run_id) + assert case.entered.wait(10) + before = _snapshot(case) + _recover(client, case) + _recover(client, case) + assert _snapshot(case) == before + assert len(case.workers) == 1 + finally: + _finish(case) + assert case.adoptions == [] + _assert_one_result(case) + + +def test_start_reserves_ownership_before_its_thread_is_scheduled(client, app_modules, monkeypatch): + case = _prepare(client, app_modules, monkeypatch, interrupted=False) + pending = [] + + class DeferredWorker: + def __init__(self, target, args=(), kwargs=None, **unused): + self.target, self.args, self.kwargs = target, args, kwargs or {} + + def start(self): + pending.append(self) + + monkeypatch.setattr(case.module.threading, "Thread", DeferredWorker) + case.runtime.start_run(case.run_id) + before = _snapshot(case) + _recover(client, case) + _recover(client, case) + assert len(pending) == 1 and _snapshot(case) == before + worker = pending[0] + worker.target(*worker.args, **worker.kwargs) + _assert_one_result(case) + + +def test_direct_resume_cannot_compete_with_an_existing_worker(client, app_modules, monkeypatch): + case = _prepare(client, app_modules, monkeypatch) + case.runtime.recover_run(case.run_id, start=False) + + def adopt(**kwargs): + case.adoptions.append(kwargs["job_id"]) + case.entered.set() + assert case.release.wait(10) + return {} + + monkeypatch.setattr(case.module, "wait_for_existing_kie_job", adopt) + first = case.module.threading.Thread(target=lambda: case.runtime.execute_run(case.run_id, resume=True)) + returned = threading.Event() + + def second_resume(): + case.runtime.execute_run(case.run_id, resume=True) + returned.set() + + second = case.module.threading.Thread(target=second_resume) + try: + first.start() + assert case.entered.wait(10) + before = _snapshot(case) + second.start() + assert returned.wait(5), "competing resume did not return while the owner was paused" + assert _snapshot(case) == before + finally: + _finish(case) + assert case.adoptions == [case.job_id] + _assert_one_result(case) + + +def test_new_runtime_recovers_an_orphan_and_completed_retry_is_a_noop(client, app_modules, monkeypatch): + case = _prepare(client, app_modules, monkeypatch) + restarted = case.module.GraphRuntime() + monkeypatch.setattr(restarted.executors["model.kie"], "execute", lambda node, context: case.calls.append(node.id) or {}) + result = restarted.recover_run(case.run_id, start=False) + assert result["recovered"] and not result["started"] + restarted.execute_run(case.run_id, resume=True) + _assert_one_result(case) + assert case.adoptions == [case.job_id] + before = _snapshot(case) + assert not restarted.recover_run(case.run_id)["started"] + assert _snapshot(case) == before + + +@pytest.mark.parametrize("entry", ["start", "recover"]) +def test_thread_start_failure_releases_the_reservation(client, app_modules, monkeypatch, entry): + case = _prepare(client, app_modules, monkeypatch, interrupted=entry == "recover") + worker_type = case.module.threading.Thread + + class BrokenWorker: + def __init__(self, **kwargs): + pass + + def start(self): + raise RuntimeError("Controlled thread start failure") + + monkeypatch.setattr(case.module.threading, "Thread", BrokenWorker) + with pytest.raises(RuntimeError, match="Controlled thread start failure"): + if entry == "recover": + case.runtime.recover_run(case.run_id) + else: + case.runtime.start_run(case.run_id) + monkeypatch.setattr(case.module.threading, "Thread", worker_type) + if entry == "recover": + _recover(client, case) + else: + case.runtime.start_run(case.run_id) + _finish(case) + _assert_one_result(case) + + +def test_exception_before_execution_releases_ownership(client, app_modules, monkeypatch): + case = _prepare(client, app_modules, monkeypatch, interrupted=False) + read = case.store.get_graph_run + + def unavailable(run_id): + raise RuntimeError("Controlled store read failure") + + monkeypatch.setattr(case.store, "get_graph_run", unavailable) + with pytest.raises(RuntimeError, match="Controlled store read failure"): + case.runtime.execute_run(case.run_id) + monkeypatch.setattr(case.store, "get_graph_run", read) + case.runtime.execute_run(case.run_id) + _assert_one_result(case) + + +@pytest.mark.parametrize("status", ["cancelling", "cancelled"]) +def test_recovery_never_restarts_cancellation(client, app_modules, monkeypatch, status): + case = _prepare(client, app_modules, monkeypatch) + case.store.update_graph_run(case.run_id, {"status": status}) + _recover(client, case) + assert case.workers == [] and case.calls == [] and case.adoptions == [] + assert case.store.get_graph_run(case.run_id)["status"] == "cancelled" + before = _snapshot(case) + _recover(client, case) + assert _snapshot(case) == before + + +def test_cancellation_during_recovery_preparation_survives_worker_start(client, app_modules, monkeypatch): + case = _prepare(client, app_modules, monkeypatch) + case.store.update_graph_run(case.run_id, {"status": "running", "error": None}) + get_job, cancel = case.store.get_job, case.runtime.cancel_run + pending = [] + cancelling = threading.Event() + + def paused_job(job_id): + case.entered.set() + assert case.release.wait(10) + return get_job(job_id) + + def cancel_request(run_id): + cancelling.set() + return cancel(run_id) + + class DeferredWorker: + def __init__(self, target, args=(), kwargs=None, **unused): + self.target, self.args, self.kwargs = target, args, kwargs or {} + + def start(self): + pending.append(self) + + monkeypatch.setattr(case.store, "get_job", paused_job) + monkeypatch.setattr(case.runtime, "cancel_run", cancel_request) + monkeypatch.setattr(case.module.threading, "Thread", DeferredWorker) + with ThreadPoolExecutor(max_workers=2) as pool: + recovery = pool.submit(_recover, client, case) + try: + assert case.entered.wait(10) + cancellation = pool.submit(client.post, f"/media/graph/runs/{case.run_id}/cancel") + assert cancelling.wait(10) + finally: + case.release.set() + recovery.result(10) + response = cancellation.result(10) + assert response.status_code == 200 and response.json()["status"] == "cancelled" + before = _snapshot(case) + worker, = pending + worker.target(*worker.args, **worker.kwargs) + assert _snapshot(case) == before + assert case.calls == [] and case.adoptions == [] diff --git a/docs/runtime-and-supervision.md b/docs/runtime-and-supervision.md index 81fc443b..2c52ea19 100644 --- a/docs/runtime-and-supervision.md +++ b/docs/runtime-and-supervision.md @@ -17,6 +17,15 @@ Important: This means the correct production-style setup is to supervise the API process, not to expect the runner to resurrect its own parent process. +Run one API process per installation/database, as the supplied launchers and +supervisor configurations do. Graph execution and recovery share per-run +ownership in that process; recovery requests for an owned run leave it untouched. +The reservation covers worker scheduling, recovery preparation and execution, +and is released when the worker finishes or fails. After the previous API process +has stopped, its successor can recover persisted orphaned runs. Multiple API +workers or overlapping API processes sharing a database are not supported by this +ownership mechanism. + ## What Happens On Restart On API startup, the runner: diff --git a/scripts/static-analysis/full-baseline.json b/scripts/static-analysis/full-baseline.json index 46f14be0..44c249f0 100644 --- a/scripts/static-analysis/full-baseline.json +++ b/scripts/static-analysis/full-baseline.json @@ -1159,26 +1159,6 @@ "lines": 6, "format": "python" }, - { - "id": "2e31d545fdeab42341b8a86153e206b76c1e87a5692d0abb6b0c82e65a632936", - "endpoints": [ - { - "file": "apps/api/app/graph/runtime.py", - "start": 582, - "end": 589, - "context": "1c6b60b766ec56a283a3e98444f988c71c54ac20bc2f65b3e5e3aa69783a4950" - }, - { - "file": "apps/api/app/graph/runtime.py", - "start": 530, - "end": 537, - "context": "813900d6d0e7503f82db2ef6f6f0936fb66b91dffda3646cfc99ea0123d8a4e0" - } - ], - "tokens": 57, - "lines": 8, - "format": "python" - }, { "id": "2fbfc15ad482814c441cd5bb64efc377bd0cf6e2527a8dfbe6fdb19162f19642", "endpoints": [ @@ -2039,26 +2019,6 @@ "lines": 11, "format": "typescript" }, - { - "id": "5c61be402a3b34b4ac5a6831a897ef0312a9463a6d63a62f97cc319a4838b9f2", - "endpoints": [ - { - "file": "apps/api/app/graph/runtime.py", - "start": 126, - "end": 132, - "context": "0f34f05319e985057e28c2db8a2fd7ed3fb751e7e3b7419c0ee4998c07d527b1" - }, - { - "file": "apps/api/app/graph/validator.py", - "start": 213, - "end": 219, - "context": "4d1137aa222b91461804010a22f7125cf123dc79f5e1fb914435127413e95c64" - } - ], - "tokens": 58, - "lines": 7, - "format": "python" - }, { "id": "5cb250a1878f741a6b8699d7e3ed869bf98f65d046645588c1b6dd8354d2419e", "endpoints": [ @@ -2599,6 +2559,26 @@ "lines": 16, "format": "python" }, + { + "id": "6fad70e133fc4f283891f1c9b479f2d83b79171e894296c83d6309cdd8d6a764", + "endpoints": [ + { + "file": "apps/api/app/graph/runtime.py", + "start": 126, + "end": 132, + "context": "5ca91c090462d2ee4cba1f6080676ff0056247e9eb6f74e23e65da70afb9f6ae" + }, + { + "file": "apps/api/app/graph/validator.py", + "start": 213, + "end": 219, + "context": "4d1137aa222b91461804010a22f7125cf123dc79f5e1fb914435127413e95c64" + } + ], + "tokens": 58, + "lines": 7, + "format": "python" + }, { "id": "6ff2ad6c65fe2fba60c3355ebb1efc874563514b349ea49d4d11b6e82ba3f574", "endpoints": [ diff --git a/scripts/static-analysis/python-baseline.json b/scripts/static-analysis/python-baseline.json index e4d37210..a5c92af0 100644 --- a/scripts/static-analysis/python-baseline.json +++ b/scripts/static-analysis/python-baseline.json @@ -1410,6 +1410,17 @@ "rule": "ruff/F841", "severity": 2 }, + { + "column": 9, + "file": "apps/api/app/graph/runtime.py", + "id": "2390caca7ba0a2f38caecd7de5086494f9d44f34f0f6c645c7657227b980c6c4", + "line": 640, + "message": "`_execute_run` is too complex (21 > 8)", + "metric": 21, + "owner": "GraphRuntime/_execute_run", + "rule": "ruff/C901", + "severity": 2 + }, { "column": 1, "file": "apps/api/app/graph/routes.py", @@ -1763,17 +1774,6 @@ "rule": "radon/CC", "severity": 2 }, - { - "column": 9, - "file": "apps/api/app/graph/runtime.py", - "id": "2a69d6f769d0e1ea8f7794d7697b52ae997a057397aeba3a9151cb0335ad01c6", - "line": 429, - "message": "`recover_run` is too complex (16 > 8)", - "metric": 16, - "owner": "GraphRuntime/recover_run", - "rule": "ruff/C901", - "severity": 2 - }, { "column": 1, "file": "apps/api/tests/test_assistant_kernel.py", @@ -4147,6 +4147,17 @@ "rule": "ruff/C901", "severity": 2 }, + { + "column": 5, + "file": "apps/api/app/graph/runtime.py", + "id": "6943b3afe68243dd58003955cd58209cda8969914b68953b5e12f4c0c360f355", + "line": 640, + "message": "Radon CC 63 exceeds 8", + "metric": 63, + "owner": "GraphRuntime/_execute_run", + "rule": "radon/CC", + "severity": 2 + }, { "column": 1, "file": "apps/api/tests/test_assistant_image_models.py", @@ -5093,17 +5104,6 @@ "rule": "radon/CC", "severity": 2 }, - { - "column": 9, - "file": "apps/api/app/graph/runtime.py", - "id": "7e638cd0228ec36a7b9ae2a49eae0fb6705e63a97a85c693f150fd909c1f70df", - "line": 598, - "message": "`execute_run` is too complex (22 > 8)", - "metric": 22, - "owner": "GraphRuntime/execute_run", - "rule": "ruff/C901", - "severity": 2 - }, { "column": 5, "file": "scripts/test_gallery_cursor.py", @@ -6258,17 +6258,6 @@ "rule": "ruff/E701", "severity": 2 }, - { - "column": 5, - "file": "apps/api/app/graph/runtime.py", - "id": "9de158574bbde1ef906ca18a7c898ff021e473cfd05413d0171e2763eab028ed", - "line": 429, - "message": "Radon CC 35 exceeds 8", - "metric": 35, - "owner": "GraphRuntime/recover_run", - "rule": "radon/CC", - "severity": 2 - }, { "column": 1, "file": "apps/api/app/assistant/run_confirmation.py", @@ -6950,6 +6939,17 @@ "rule": "radon/CC", "severity": 2 }, + { + "column": 5, + "file": "apps/api/app/graph/runtime.py", + "id": "ae1940c8f5d0cf571b1ad0013ae418bac067f49983fa826b14633b86ff5d5040", + "line": 471, + "message": "Radon CC 34 exceeds 8", + "metric": 34, + "owner": "GraphRuntime/_recover_run", + "rule": "radon/CC", + "severity": 2 + }, { "column": 5, "file": "apps/api/app/graph/storyboard_sheet_renderer.py", @@ -9509,14 +9509,14 @@ "severity": 2 }, { - "column": 5, + "column": 9, "file": "apps/api/app/graph/runtime.py", - "id": "ec7fc36b6a0f1b7b20ffbf77f82730930a9639256109ee72a2ee5b404b76a7fb", - "line": 598, - "message": "Radon CC 72 exceeds 8", - "metric": 72, - "owner": "GraphRuntime/execute_run", - "rule": "radon/CC", + "id": "ec4fc426c38c24a8d71a7594ec1550bed1829eb869ed5fd8a035f0da7dc93888", + "line": 471, + "message": "`_recover_run` is too complex (14 > 8)", + "metric": 14, + "owner": "GraphRuntime/_recover_run", + "rule": "ruff/C901", "severity": 2 }, { From 1c3fc92d56fcf9c1473226939840b9e14e9e91b4 Mon Sep 17 00:00:00 2001 From: Gateway <519563+gateway@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:17:42 +0700 Subject: [PATCH 2/3] Use the official Ubuntu archive for CI dependencies --- .github/workflows/tests.yml | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index cf4b5caa..5e5af1ef 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -115,6 +115,11 @@ jobs: python -m venv .venv/static-analysis .venv/static-analysis/bin/python -m pip install -r scripts/static-analysis/requirements.txt + - name: Use the Ubuntu archive for CI packages + run: | + # Avoid the stalled Azure mirror; retain APT's existing signing keys and suites. + printf '%s\n' 'https://archive.ubuntu.com/ubuntu/' | sudo tee /etc/apt/apt-mirrors.txt + - name: Install Playwright Chromium run: npx playwright install --with-deps chromium @@ -122,7 +127,11 @@ jobs: run: node --test scripts/graph_text_wrapping.test.mjs - name: Install media tooling - run: sudo apt-get update && sudo apt-get install -y ffmpeg + run: | + sudo apt-get update + sudo apt-get install -y ffmpeg + ffmpeg -version + ffprobe -version - name: Run quality gates env: From f847daf6f5bbfcda08f53facd9b76202ba5476c3 Mon Sep 17 00:00:00 2001 From: Gateway <519563+gateway@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:37:00 +0700 Subject: [PATCH 3/3] Connect recovery fixtures and verify downstream result propagation --- .../tests/test_graph_recovery_ownership.py | 31 ++++++++++++++----- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/apps/api/tests/test_graph_recovery_ownership.py b/apps/api/tests/test_graph_recovery_ownership.py index 9c2eea0b..008c2432 100644 --- a/apps/api/tests/test_graph_recovery_ownership.py +++ b/apps/api/tests/test_graph_recovery_ownership.py @@ -1,13 +1,19 @@ """Recovery ownership through the real router/runtime/store; normal CI fixtures only.""" -from concurrent.futures import ThreadPoolExecutor import copy import importlib import threading +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace import pytest +def _image_output(node_id): + from app.graph.schemas import GraphOutputRef + + return {"image": [GraphOutputRef(kind="value", media_type="image", value={"node_id": node_id})]} + + def _prepare(client, app_modules, monkeypatch, *, interrupted=True): module = importlib.import_module("app.graph.runtime") from app.graph.schemas import GraphWorkflow @@ -15,6 +21,10 @@ def _prepare(client, app_modules, monkeypatch, *, interrupted=True): workflow = GraphWorkflow(name="Recovery ownership", nodes=[ {"id": "upstream", "type": "model.kie.nano_banana_pro", "fields": {"prompt": "Existing job", "resolution": "1K"}}, {"id": "downstream", "type": "model.kie.nano_banana_pro", "fields": {"prompt": "One execution", "resolution": "1K"}}, + {"id": "preview", "type": "preview.image"}, + ], edges=[ + {"id": "upstream-downstream", "source": "upstream", "source_port": "image", "target": "downstream", "target_port": "image_refs"}, + {"id": "downstream-preview", "source": "downstream", "source_port": "image", "target": "preview", "target_port": "image"}, ]) saved = client.post("/media/graph/workflows", json=workflow.model_dump(mode="json")) assert saved.status_code == 200, saved.text @@ -34,11 +44,11 @@ def _prepare(client, app_modules, monkeypatch, *, interrupted=True): def adopt(**kwargs): adoptions.append(kwargs["job_id"]) - return {} + return _image_output("upstream") def execute(node, context): calls.append(node.id) - return {} + return _image_output(node.id) class Worker(threading.Thread): def __init__(self, *args, **kwargs): @@ -77,6 +87,11 @@ def _assert_one_result(case): assert case.calls.count("downstream") == 1 assert sum(event["event_type"] == "run.completed" for event in events) == 1 assert sum(event["event_type"] == "node.completed" and event["node_id"] == "downstream" for event in events) == 1 + downstream = case.store.get_graph_run_node(case.run_id, "downstream") + assert downstream["input_snapshot_json"]["image_refs"] == {"node_id": "upstream"} + preview = case.store.get_graph_run_node(case.run_id, "preview") + assert preview["status"] == "completed" + assert [ref["value"] for ref in preview["output_snapshot_json"]["image"]] == [{"node_id": "downstream"}] @pytest.mark.parametrize("phase", ["polling", "downstream"]) @@ -88,14 +103,14 @@ def adopt(**kwargs): if phase == "polling": case.entered.set() assert case.release.wait(10) - return {} + return _image_output("upstream") def execute(node, context): case.calls.append(node.id) if phase == "downstream": case.entered.set() assert case.release.wait(10) - return {} + return _image_output(node.id) monkeypatch.setattr(case.module, "wait_for_existing_kie_job", adopt) monkeypatch.setattr(case.runtime.executors["model.kie"], "execute", execute) @@ -132,7 +147,7 @@ def execute(node, context): if node.id == "upstream": case.entered.set() assert case.release.wait(10) - return {} + return _image_output(node.id) monkeypatch.setattr(case.runtime.executors["model.kie"], "execute", execute) try: @@ -179,7 +194,7 @@ def adopt(**kwargs): case.adoptions.append(kwargs["job_id"]) case.entered.set() assert case.release.wait(10) - return {} + return _image_output("upstream") monkeypatch.setattr(case.module, "wait_for_existing_kie_job", adopt) first = case.module.threading.Thread(target=lambda: case.runtime.execute_run(case.run_id, resume=True)) @@ -206,7 +221,7 @@ def second_resume(): def test_new_runtime_recovers_an_orphan_and_completed_retry_is_a_noop(client, app_modules, monkeypatch): case = _prepare(client, app_modules, monkeypatch) restarted = case.module.GraphRuntime() - monkeypatch.setattr(restarted.executors["model.kie"], "execute", lambda node, context: case.calls.append(node.id) or {}) + monkeypatch.setattr(restarted.executors["model.kie"], "execute", lambda node, context: case.calls.append(node.id) or _image_output(node.id)) result = restarted.recover_run(case.run_id, start=False) assert result["recovered"] and not result["started"] restarted.execute_run(case.run_id, resume=True)