From 3cd61be96d5cf0921f5b0980e43095ded569d238 Mon Sep 17 00:00:00 2001 From: lda Date: Tue, 8 Sep 2026 10:22:17 +0700 Subject: [PATCH] sched: route manual runs through durable admission before dispatch (T06) --- src/wf_api/runs.py | 47 ++++++++--- tests/wf_api/test_run_admission_path.py | 104 ++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 13 deletions(-) create mode 100644 tests/wf_api/test_run_admission_path.py diff --git a/src/wf_api/runs.py b/src/wf_api/runs.py index 1a690d55..f4cc7d3c 100644 --- a/src/wf_api/runs.py +++ b/src/wf_api/runs.py @@ -31,6 +31,8 @@ from .run_lifecycle import ( has_blocking_diagnostics, load_stored_run, mark_resume_blocked, + materialize_admitted_view, + persist_admission, persist_stopped_run, restore_interrupted_run, validate_pinned_resume_environment, @@ -98,27 +100,46 @@ class WorkflowRunApi: max_steps=limits.max_steps, ) - plan = raw_plan_from_artifact(artifact) - run = await self.context.runtime.run_workflow_from_plan( - plan, - workflow_input, + # Durable admission ordering: recheck -> allocate/freeze -> persist + # admission -> materialize view -> dispatch captured -> persist + # stopped -> reconcile. A failed durable admission never dispatches, + # and dispatch never re-resolves the deployment. + store = self._run_store() + run_id = store.allocate_run_id() + environment = create_pinned_environment( deployment=deployment, artifact=artifact, - saved_subgraph_tree=tree, + tree=tree, + ) + admission = persist_admission( + store=store, + run_id=run_id, + environment=environment, + resolved_input=workflow_input, + max_steps=limits.max_steps, + ) + materialize_admitted_view(store=store, admission=admission) + plan = raw_plan_from_artifact(admission.environment.root_artifact) + captured_tree = saved_subgraph_tree_from_snapshots( + admission.environment.child_artifacts + ) + run = await self.context.runtime.run_workflow_from_plan( + plan, + dict(admission.resolved_input), + deployment=admission.environment.deployment, + artifact=admission.environment.root_artifact, + saved_subgraph_tree=captured_tree, limits=limits, ) record = persist_stopped_run( - store=self._run_store(), - environment=create_pinned_environment( - deployment=deployment, - artifact=artifact, - tree=tree, - ), + store=store, + environment=admission.environment, run=run, + run_id=run_id, ) return _run_payload( - deployment=deployment, - artifact=artifact, + deployment=admission.environment.deployment, + artifact=admission.environment.root_artifact, status=run.status.value, run_id=record.id, resume_readiness=record.resume_readiness.value, diff --git a/tests/wf_api/test_run_admission_path.py b/tests/wf_api/test_run_admission_path.py new file mode 100644 index 00000000..39c07fcc --- /dev/null +++ b/tests/wf_api/test_run_admission_path.py @@ -0,0 +1,104 @@ +"""Admission path through the run API: persist before dispatch (T06).""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +from tests.wf_mcp.test_support import echo_tool +from tests.wf_mcp.workflow_surface.conftest import echo_artifact +from wf_api.runs import WorkflowRunApi +from wf_artifacts import ( + FileRunStore, + FileWorkflowArtifactStore, + StoredRunStatus, + WorkflowDeployment, +) +from wf_mcp.broker import WfMcpService +from wf_mcp.broker.service.workflow_operation_context import context_from_service +from wf_mcp.models import ConnectionConfig +from wf_mcp.storage import FileStore + + +def _api_with_echo(root: Path) -> tuple[WorkflowRunApi, FileRunStore]: + + artifact_store = FileWorkflowArtifactStore(root) + artifact_store.save_artifact(echo_artifact()) + artifact_store.save_deployment( + WorkflowDeployment( + id="echo.personal", + artifact_id="echo", + artifact_version=1, + bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}], + ) + ) + service = WfMcpService( + store=FileStore(root / "mcp"), + artifact_store=artifact_store, + run_store=FileRunStore(root / "mcp"), + ) + service.register_connection( + ConnectionConfig(id="demo.personal", server="demo", account="personal") + ) + service.register_specs("demo.personal", echo_tool) + context = context_from_service(service) + assert context.run_store is not None + assert isinstance(context.run_store, FileRunStore) + return WorkflowRunApi(context), context.run_store + + +def test_manual_run_persists_admission_before_dispatch(tmp_path: Path) -> None: + api, store = _api_with_echo(tmp_path / "admit") + result = asyncio.run( + api.run_deployment( + deployment_id="echo.personal", + workflow_input={"text": "hi"}, + ) + ) + assert result["run_id"] is not None + admission = store.get_admission(result["run_id"]) + assert admission.resolved_input == {"text": "hi"} + assert admission.environment.deployment.id == "echo.personal" + record = store.get_run(result["run_id"]) + assert record.status is not StoredRunStatus.ADMITTED + assert record.latest_checkpoint_id is not None + + +def test_fault_before_admission_never_dispatches(tmp_path: Path, monkeypatch) -> None: + api, store = _api_with_echo(tmp_path / "fault-before") + + def _fail_save(admission) -> None: # type: ignore[no-untyped-def] + raise OSError("injected admission failure") + + # Admission persist precedes dispatch in run_deployment ordering: a failure + # here must propagate before any run view exists, so no dispatch could + # have produced a stopped checkpoint. + monkeypatch.setattr(store, "save_admission", _fail_save) + try: + asyncio.run( + api.run_deployment( + deployment_id="echo.personal", + workflow_input={"text": "hi"}, + ) + ) + raise AssertionError("fault must propagate") + except OSError: + pass + assert store.list_admissions() == [] + assert store.list_runs() == [] + + +def test_captured_invocation_freezes_input(tmp_path: Path) -> None: + api, store = _api_with_echo(tmp_path / "freeze") + payload = {"text": "hi", "nested": {"n": 1}} + result = asyncio.run( + api.run_deployment( + deployment_id="echo.personal", + workflow_input=payload, + ) + ) + assert result["run_id"] is not None + payload["text"] = "mutated" + payload["nested"]["n"] = 999 + admission = store.get_admission(result["run_id"]) + assert admission.resolved_input == {"text": "hi", "nested": {"n": 1}}