From 9b3b47ca651f99b6024d1737156c2d169226a430 Mon Sep 17 00:00:00 2001 From: Adam Gohain Date: Tue, 4 Aug 2026 10:22:43 -0400 Subject: [PATCH] feat(workflows): add durable inference commands --- client/src/contexts/WorkflowContext.js | 19 +- client/src/contexts/WorkflowContext.test.js | 41 +++++ client/src/views/ModelInference.js | 10 ++ client/src/views/ModelInference.test.js | 43 +++-- server_api/main.py | 190 ++++++++++++++++---- server_api/workflows/router.py | 61 ++++++- tests/test_pytc_runtime_routes.py | 179 +++++++++++++++++- tests/test_workflow_routes.py | 8 +- 8 files changed, 486 insertions(+), 65 deletions(-) diff --git a/client/src/contexts/WorkflowContext.js b/client/src/contexts/WorkflowContext.js index 1a258b29..fab6ab22 100644 --- a/client/src/contexts/WorkflowContext.js +++ b/client/src/contexts/WorkflowContext.js @@ -46,7 +46,10 @@ export const WorkflowContext = createContext(null); const PENDING_RUNTIME_ACTION_KEY = "pytc.workflow.pendingRuntimeAction.v1"; const PENDING_RUNTIME_ACTION_TTL_MS = 6 * 60 * 60 * 1000; -const PERSISTABLE_RUNTIME_KINDS = new Set(["monitor_training"]); +const PERSISTABLE_RUNTIME_KINDS = new Set([ + "monitor_training", + "monitor_inference", +]); const isPersistableRuntimeAction = (kind) => PERSISTABLE_RUNTIME_KINDS.has(kind); @@ -955,11 +958,19 @@ export function WorkflowProvider({ children }) { workflow.id, durableCommand.id, ); - if ((approvedEffects?.runtime_action || {}).kind === "start_training") { + const runtimeKind = (approvedEffects?.runtime_action || {}).kind; + if ( + runtimeKind === "start_training" || + runtimeKind === "start_inference" + ) { + const monitorKind = + runtimeKind === "start_training" + ? "monitor_training" + : "monitor_inference"; registerPendingRuntimeAction( { - id: `monitor_training:${Date.now()}`, - kind: "monitor_training", + id: `${monitorKind}:${Date.now()}`, + kind: monitorKind, commandId: durableCommand.id, commandResult, clientEffects: approvedEffects, diff --git a/client/src/contexts/WorkflowContext.test.js b/client/src/contexts/WorkflowContext.test.js index 80a5341a..fb9aa407 100644 --- a/client/src/contexts/WorkflowContext.test.js +++ b/client/src/contexts/WorkflowContext.test.js @@ -589,6 +589,47 @@ describe("WorkflowProvider", () => { }); }); + it("monitors an approved durable inference command without queuing a browser launch", async () => { + approveAgentAction.mockResolvedValue({ + workflow: { ...baseWorkflow, stage: "inference" }, + client_effects: { + navigate_to: "inference", + set_inference_image_path: "/tmp/image.h5", + set_inference_checkpoint_path: "/tmp/checkpoint.pth.tar", + set_inference_output_path: "/tmp/inference-output", + runtime_action: { kind: "start_inference" }, + }, + commands: [ + { + id: 23, + title: "Start inference", + command: "pytc inference", + }, + ], + }); + + renderProvider({ + inferenceState: { + setInputImage: jest.fn(), + setCheckpointPath: jest.fn(), + setOutputPath: jest.fn(), + }, + }); + await screen.findByText("setup"); + + fireEvent.click(screen.getByText("Approve proposal")); + + await waitFor(() => { + expect(runWorkflowCommand).toHaveBeenCalledWith(1, 23); + expect(screen.getByText("monitor_inference")).toBeTruthy(); + }); + const persisted = JSON.parse( + window.sessionStorage.getItem("pytc.workflow.pendingRuntimeAction.v1"), + ); + expect(persisted?.action?.kind).toBe("monitor_inference"); + expect(persisted?.action?.commandId).toBe(23); + }); + it("exposes direct client-effect execution for chat action cards", async () => { const setOutputPath = jest.fn(); diff --git a/client/src/views/ModelInference.js b/client/src/views/ModelInference.js index 5bffc17f..7a4bb9dd 100644 --- a/client/src/views/ModelInference.js +++ b/client/src/views/ModelInference.js @@ -259,6 +259,16 @@ function ModelInference({ isInferring, setIsInferring }) { startInferenceRun(action); }, [consumeRuntimeAction, pendingRuntimeAction, startInferenceRun]); + useEffect(() => { + if (pendingRuntimeAction?.kind !== "monitor_inference") return; + const action = pendingRuntimeAction; + consumeRuntimeAction?.(action.id); + terminalLoggedRef.current = false; + setIsInferring(true); + setInferenceStatus("Model run accepted. Monitoring process..."); + refreshInferenceLogs(); + }, [consumeRuntimeAction, pendingRuntimeAction, setIsInferring]); + const handleStartButton = async () => { await startInferenceRun(); }; diff --git a/client/src/views/ModelInference.test.js b/client/src/views/ModelInference.test.js index 8585140d..ea17638d 100644 --- a/client/src/views/ModelInference.test.js +++ b/client/src/views/ModelInference.test.js @@ -8,12 +8,24 @@ import { getInferenceStatus, syncWorkflowInferenceRuntime, } from "../api"; +import { launchInferenceFromContext } from "../runtime/modelLaunch"; const mockAppendWorkflowEvent = jest.fn(); const mockRefreshWorkflow = jest.fn(); const mockRefreshEvents = jest.fn(); const mockRefreshInsights = jest.fn(); const mockRefreshEvidence = jest.fn(); +const mockConsumeRuntimeAction = jest.fn(); +const mockWorkflowContext = { + workflow: { id: 42, stage: "inference" }, + appendEvent: mockAppendWorkflowEvent, + refreshWorkflow: mockRefreshWorkflow, + refreshEvents: mockRefreshEvents, + refreshInsights: mockRefreshInsights, + refreshEvidence: mockRefreshEvidence, + pendingRuntimeAction: null, + consumeRuntimeAction: mockConsumeRuntimeAction, +}; jest.mock("../api", () => ({ getInferenceLogs: jest.fn(), @@ -28,16 +40,7 @@ jest.mock("../runtime/modelLaunch", () => ({ })); jest.mock("../contexts/WorkflowContext", () => ({ - useWorkflow: () => ({ - workflow: { id: 42, stage: "inference" }, - appendEvent: mockAppendWorkflowEvent, - refreshWorkflow: mockRefreshWorkflow, - refreshEvents: mockRefreshEvents, - refreshInsights: mockRefreshInsights, - refreshEvidence: mockRefreshEvidence, - pendingRuntimeAction: null, - consumeRuntimeAction: jest.fn(), - }), + useWorkflow: () => mockWorkflowContext, })); jest.mock("../components/Configurator", () => () =>
Configurator
); @@ -70,6 +73,7 @@ describe("ModelInference", () => { beforeEach(() => { jest.useFakeTimers(); jest.clearAllMocks(); + mockWorkflowContext.pendingRuntimeAction = null; getInferenceLogs.mockResolvedValue({ phase: "finished", metadata: {} }); getInferenceStatus.mockResolvedValue({ isRunning: false, @@ -135,4 +139,23 @@ describe("ModelInference", () => { ); }); }); + + it("monitors an accepted durable inference command without launching it again", async () => { + mockWorkflowContext.pendingRuntimeAction = { + id: "monitor_inference:23", + kind: "monitor_inference", + commandId: 23, + }; + const { setIsInferring } = renderInference({ isInferring: false }); + + await act(async () => { + await Promise.resolve(); + }); + expect(mockConsumeRuntimeAction).toHaveBeenCalledWith( + "monitor_inference:23", + ); + expect(setIsInferring).toHaveBeenCalledWith(true); + expect(getInferenceLogs).toHaveBeenCalled(); + expect(launchInferenceFromContext).not.toHaveBeenCalled(); + }); }); diff --git a/server_api/main.py b/server_api/main.py index 564d86a6..b0f5997e 100644 --- a/server_api/main.py +++ b/server_api/main.py @@ -818,6 +818,85 @@ def _build_training_body_from_command( } +def _build_inference_body_from_command( + command: WorkflowCommand, + workflow, +) -> dict[str, Any]: + """Build a worker payload from durable inference-command input. + + Command inputs intentionally retain client effects rather than browser-built + YAML. The server resolves the selected preset at submission time so the + worker launch is replayable without requiring an open browser tab. + """ + command_input = decode_json(command.input_json) + client_effects = command_input.get("client_effects") + if not isinstance(client_effects, dict): + client_effects = {} + command_arguments = command_input.get("arguments") + if not isinstance(command_arguments, dict): + command_arguments = {} + + config_origin_path = _first_string( + command_input.get("configOriginPath"), + command_input.get("config_origin_path"), + command_input.get("inference_config_preset"), + client_effects.get("set_inference_config_preset"), + workflow.config_path, + ) + inference_config = _first_string( + command_input.get("inferenceConfig"), + command_input.get("inference_config"), + ) + if not inference_config: + if not config_origin_path: + raise HTTPException( + status_code=400, + detail="Inference command is missing a config preset or config text.", + ) + config_origin_path, inference_config = _read_pytc_config_content( + config_origin_path + ) + + output_path = _first_string( + command_input.get("outputPath"), + command_input.get("output_path"), + client_effects.get("set_inference_output_path"), + workflow.inference_output_path, + ) + image_path = _first_string( + command_input.get("inputImagePath"), + command_input.get("image_path"), + client_effects.get("set_inference_image_path"), + workflow.image_path, + workflow.dataset_path, + ) + checkpoint_path = _first_string( + command_input.get("checkpointPath"), + command_input.get("checkpoint_path"), + command_arguments.get("checkpoint"), + client_effects.get("set_inference_checkpoint_path"), + workflow.checkpoint_path, + ) + run_id = _first_string( + command_input.get("run_id"), + command_input.get("runId"), + f"workflow-command-{command.id}", + ) + + return { + "inferenceConfig": inference_config, + "configOriginPath": config_origin_path, + "outputPath": output_path or "", + "inputImagePath": image_path, + "checkpointPath": checkpoint_path, + "arguments": {"checkpoint": checkpoint_path}, + "workflowId": workflow.id, + "workflow_id": workflow.id, + "command_id": command.id, + "run_id": run_id, + } + + def _workflow_command_run_response( workflow, command: WorkflowCommand, @@ -857,7 +936,7 @@ def _fail_command_operation( expected_status=operation.status, error_payload=error_payload, lease_owner=( - "server_api.training_runner" if operation.status == "running" else None + operation.lease_owner if operation.status == "running" else None ), commit=False, ) @@ -2592,7 +2671,7 @@ async def run_workflow_command( ) if not command: raise HTTPException(status_code=404, detail="Workflow command not found.") - if command.command_type != "start_training": + if command.command_type not in {"start_training", "start_inference"}: raise HTTPException( status_code=400, detail=f"Unsupported workflow command type: {command.command_type}", @@ -2619,6 +2698,17 @@ async def run_workflow_command( status_code=409, detail="Workflow command was already submitted." ) + is_training = command.command_type == "start_training" + operation_type = "start_training" if is_training else "start_inference" + runtime_mode = "training" if is_training else "inference" + runner_name = ( + "server_api.training_runner" if is_training else "server_api.inference_runner" + ) + worker_endpoint = ( + "/start_model_training" if is_training else "/start_model_inference" + ) + event_prefix = "training" if is_training else "inference" + operation_query = db.query(WorkflowOperation).filter( WorkflowOperation.workflow_id == workflow.id, WorkflowOperation.command_id == command.id, @@ -2634,7 +2724,7 @@ async def run_workflow_command( operation = create_workflow_operation( db, workflow_id=workflow.id, - operation_type="start_training", + operation_type=operation_type, idempotency_key=( f"workflow-command:{command.id}:attempt:{operation_query.count() + 1}" ), @@ -2663,12 +2753,16 @@ async def run_workflow_command( ) try: - body = _build_training_body_from_command(command, workflow) - body = _runtime_body_with_workflow_fallbacks(body, workflow, mode="training") + body = ( + _build_training_body_from_command(command, workflow) + if is_training + else _build_inference_body_from_command(command, workflow) + ) + body = _runtime_body_with_workflow_fallbacks(body, workflow, mode=runtime_mode) command = mark_workflow_command_running( db, command, - lease_owner="server_api.training_runner", + lease_owner=runner_name, commit=False, ) operation = transition_workflow_operation( @@ -2676,45 +2770,57 @@ async def run_workflow_command( operation, status="running", expected_status="queued", - lease_owner="server_api.training_runner", + lease_owner=runner_name, metadata={"run_id": body.get("run_id")}, commit=False, ) db.commit() db.refresh(command) db.refresh(operation) - update_workflow_fields( - db, - workflow, - { - "stage": "retraining_staged", - "training_output_path": body.get("outputPath"), - "config_path": body.get("configOriginPath"), - }, - commit=True, - ) + workflow_updates = { + "stage": "retraining_staged" if is_training else "inference", + "config_path": body.get("configOriginPath"), + } + if is_training: + workflow_updates["training_output_path"] = body.get("outputPath") + else: + workflow_updates["inference_output_path"] = body.get("outputPath") + workflow_updates["checkpoint_path"] = body.get("checkpointPath") + update_workflow_fields(db, workflow, workflow_updates, commit=True) + event_payload = { + "run_id": body.get("run_id"), + "command_id": command.id, + "outputPath": body.get("outputPath"), + "configOriginPath": body.get("configOriginPath"), + "inputImagePath": body.get("inputImagePath"), + "source": "workflow_command_runner", + } + if is_training: + event_payload.update( + { + "logPath": body.get("logPath"), + "inputLabelPath": body.get("inputLabelPath"), + } + ) + else: + event_payload["checkpointPath"] = body.get("checkpointPath") started_event = append_event_for_workflow_if_present( db, workflow_id=workflow.id, actor="system", - event_type="training.started", + event_type=f"{event_prefix}.started", stage=workflow.stage, - summary="Started model training from a durable workflow command.", - payload={ - "run_id": body.get("run_id"), - "command_id": command.id, - "outputPath": body.get("outputPath"), - "logPath": body.get("logPath"), - "configOriginPath": body.get("configOriginPath"), - "inputImagePath": body.get("inputImagePath"), - "inputLabelPath": body.get("inputLabelPath"), - "source": "workflow_command_runner", - }, - idempotency_key=f"workflow-command:{command.id}:training.started", + summary=( + "Started model training from a durable workflow command." + if is_training + else "Started model inference from a durable workflow command." + ), + payload=event_payload, + idempotency_key=f"workflow-command:{command.id}:{event_prefix}.started", ) worker_data = _proxy_to_worker( "post", - "/start_model_training", + worker_endpoint, json_body=body, timeout=30, ) @@ -2736,7 +2842,7 @@ async def run_workflow_command( status="succeeded", expected_status="running", result_payload=operation_result, - lease_owner="server_api.training_runner", + lease_owner=runner_name, commit=False, ) db.commit() @@ -2760,15 +2866,19 @@ async def run_workflow_command( db, workflow_id=workflow.id, actor="system", - event_type="training.failed", + event_type=f"{event_prefix}.failed", stage=workflow.stage, - summary="Failed to start model training from a durable workflow command.", + summary=( + "Failed to start model training from a durable workflow command." + if is_training + else "Failed to start model inference from a durable workflow command." + ), payload={ "command_id": command.id, "source": "workflow_command_runner", **error_payload, }, - idempotency_key=f"workflow-command:{command.id}:training.failed", + idempotency_key=f"workflow-command:{command.id}:{event_prefix}.failed", ) raise except Exception as exc: @@ -2787,15 +2897,19 @@ async def run_workflow_command( db, workflow_id=workflow.id, actor="system", - event_type="training.failed", + event_type=f"{event_prefix}.failed", stage=workflow.stage, - summary="Failed to start model training from a durable workflow command.", + summary=( + "Failed to start model training from a durable workflow command." + if is_training + else "Failed to start model inference from a durable workflow command." + ), payload={ "command_id": command.id, "source": "workflow_command_runner", **error_payload, }, - idempotency_key=f"workflow-command:{command.id}:training.failed", + idempotency_key=f"workflow-command:{command.id}:{event_prefix}.failed", ) raise HTTPException(status_code=500, detail=error_payload) from exc diff --git a/server_api/workflows/router.py b/server_api/workflows/router.py index e8a6c228..f5b0feba 100644 --- a/server_api/workflows/router.py +++ b/server_api/workflows/router.py @@ -9124,8 +9124,29 @@ async def approve_agent_action( except ValidationError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc proposal.approval_status = "approved" + runtime_action = client_effects.get("runtime_action") + server_executes_inference = ( + isinstance(runtime_action, dict) + and runtime_action.get("kind") == "start_inference" + ) + if server_executes_inference: + update_payload: Dict[str, Any] = {"stage": "inference"} + if client_effects.get("set_inference_output_path"): + update_payload["inference_output_path"] = client_effects[ + "set_inference_output_path" + ] + if client_effects.get("set_inference_checkpoint_path"): + update_payload["checkpoint_path"] = client_effects[ + "set_inference_checkpoint_path" + ] + if client_effects.get("set_inference_config_preset"): + update_payload["config_path"] = client_effects[ + "set_inference_config_preset" + ] + update_workflow_fields(db, workflow, update_payload, commit=False) db.commit() db.refresh(proposal) + db.refresh(workflow) approved = append_workflow_event( db, @@ -9151,15 +9172,23 @@ async def approve_agent_action( workflow_id=workflow.id, actor="system", event_type=( - "evaluation.agent_action_approved" - if server_executes_evaluation - else "agent.client_effects_approved" + "inference.run_approved" + if server_executes_inference + else ( + "evaluation.agent_action_approved" + if server_executes_evaluation + else "agent.client_effects_approved" + ) ), stage=workflow.stage, summary=( - "Approved agent evaluation action for server execution." - if server_executes_evaluation - else "Approved in-app assistant action for client execution." + "Inference run approved for server execution." + if server_executes_inference + else ( + "Approved agent evaluation action for server execution." + if server_executes_evaluation + else "Approved in-app assistant action for client execution." + ) ), payload={ "proposal_event_id": proposal.id, @@ -9175,6 +9204,24 @@ async def approve_agent_action( operation_payload = None receipt = None approved_client_effects = dict(client_effects) + commands = [] + if server_executes_inference: + command = create_workflow_command( + db, + workflow_id=workflow.id, + command_type="start_inference", + idempotency_key=f"agent-proposal:{proposal.id}:start_inference", + actor="agent", + source_event_id=proposal.id, + approval_event_id=approved.id, + input_payload={ + "client_effects": client_effects, + "workflow_stage": workflow.stage, + "proposal_event_id": proposal.id, + }, + commit=True, + ) + commands = [_command_response(command)] if server_executes_evaluation: requested_correlation_id = params.get("correlation_id") operation, receipt = stage_and_execute_compute_evaluation_proposal( @@ -9200,7 +9247,7 @@ async def approve_agent_action( **approved_client_effects, "workflow_stage": workflow.stage, }, - commands=[], + commands=commands, operation=operation_payload, receipt=receipt, ) diff --git a/tests/test_pytc_runtime_routes.py b/tests/test_pytc_runtime_routes.py index ce53075a..1a30ee17 100644 --- a/tests/test_pytc_runtime_routes.py +++ b/tests/test_pytc_runtime_routes.py @@ -597,6 +597,179 @@ def fake_worker(method, endpoint, json_body=None, **_kwargs): f"workflow-command-{command['id']}", ) + def test_durable_inference_command_runner_submits_once_and_replays_result(self): + """Inference commands use the server-owned durable submission path.""" + workflow_id = self._workflow_id() + project_root = pathlib.Path(self.temp_dir.name) / "inference-command-project" + image_path = project_root / "data" / "image" / "infer_im.h5" + checkpoint_path = project_root / "outputs" / "checkpoint_00001.pth.tar" + output_path = project_root / "outputs" / "inference" + image_path.parent.mkdir(parents=True) + checkpoint_path.parent.mkdir(parents=True) + output_path.mkdir(parents=True) + image_path.write_text("image", encoding="utf-8") + checkpoint_path.write_text("checkpoint", encoding="utf-8") + + db = self.SessionLocal() + try: + command = WorkflowCommand( + workflow_id=workflow_id, + command_type="start_inference", + status="queued", + idempotency_key="test:durable-inference-command", + actor="agent", + input_json=encode_json( + { + "inferenceConfig": "INFERENCE: {}\n", + "configOriginPath": "configs/MitoEM/Mito25-Local-BC.yaml", + "outputPath": str(output_path), + "inputImagePath": str(image_path), + "arguments": {"checkpoint": str(checkpoint_path)}, + } + ), + ) + db.add(command) + db.commit() + db.refresh(command) + command_id = command.id + finally: + db.close() + + captured = {} + + def fake_worker(method, endpoint, json_body=None, **_kwargs): + captured["calls"] = captured.get("calls", 0) + 1 + captured["method"] = method + captured["endpoint"] = endpoint + captured["json_body"] = json_body + return {"status": "started", "pid": 4343} + + with patch("server_api.main._proxy_to_worker", side_effect=fake_worker): + response = self.client.post( + f"/api/workflows/{workflow_id}/commands/{command_id}/run" + ) + replay = self.client.post( + f"/api/workflows/{workflow_id}/commands/{command_id}/run" + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(replay.status_code, 200) + payload = response.json() + self.assertEqual(payload["command"]["status"], "submitted") + self.assertEqual(payload["command"]["attempt_count"], 1) + self.assertEqual(payload["operation"]["operation_type"], "start_inference") + self.assertEqual(payload["operation"]["status"], "succeeded") + self.assertEqual(payload["operation"]["command_id"], command_id) + self.assertEqual( + payload["operation"]["idempotency_key"], + f"workflow-command:{command_id}:attempt:1", + ) + self.assertEqual(replay.json()["operation"]["id"], payload["operation"]["id"]) + self.assertEqual(captured["calls"], 1) + self.assertEqual(captured["method"], "post") + self.assertEqual(captured["endpoint"], "/start_model_inference") + self.assertEqual(captured["json_body"]["workflowId"], workflow_id) + self.assertEqual(captured["json_body"]["command_id"], command_id) + self.assertEqual( + captured["json_body"]["run_id"], f"workflow-command-{command_id}" + ) + self.assertEqual( + captured["json_body"]["inputImagePath"], str(image_path.resolve()) + ) + self.assertEqual( + captured["json_body"]["arguments"]["checkpoint"], + str(checkpoint_path.resolve()), + ) + + events_response = self.client.get(f"/api/workflows/{workflow_id}/events") + self.assertEqual(events_response.status_code, 200) + started_events = [ + event + for event in events_response.json() + if event["event_type"] == "inference.started" + ] + self.assertEqual(len(started_events), 1) + self.assertEqual(started_events[0]["payload"]["command_id"], command_id) + self.assertEqual( + started_events[0]["payload"]["run_id"], + f"workflow-command-{command_id}", + ) + + def test_durable_inference_command_retry_uses_a_new_operation_attempt(self): + workflow_id = self._workflow_id() + project_root = pathlib.Path(self.temp_dir.name) / "retry-inference-command" + image_path = project_root / "image.h5" + checkpoint_path = project_root / "checkpoint.pth.tar" + output_path = project_root / "output" + project_root.mkdir(parents=True) + output_path.mkdir() + image_path.write_text("image", encoding="utf-8") + checkpoint_path.write_text("checkpoint", encoding="utf-8") + + db = self.SessionLocal() + try: + command = WorkflowCommand( + workflow_id=workflow_id, + command_type="start_inference", + status="queued", + idempotency_key="test:retryable-inference-command", + actor="user", + input_json=encode_json( + { + "inferenceConfig": "INFERENCE: {}\n", + "outputPath": str(output_path), + "inputImagePath": str(image_path), + "arguments": {"checkpoint": str(checkpoint_path)}, + } + ), + ) + db.add(command) + db.commit() + db.refresh(command) + command_id = command.id + finally: + db.close() + + with patch( + "server_api.main._proxy_to_worker", + side_effect=HTTPException(status_code=503, detail="worker unavailable"), + ): + failed_response = self.client.post( + f"/api/workflows/{workflow_id}/commands/{command_id}/run" + ) + + self.assertEqual(failed_response.status_code, 503) + commands_response = self.client.get(f"/api/workflows/{workflow_id}/commands") + command_payload = next( + item for item in commands_response.json() if item["id"] == command_id + ) + self.assertEqual(command_payload["status"], "retry_pending") + self.assertEqual(command_payload["attempt_count"], 1) + + operations_response = self.client.get( + f"/api/workflows/{workflow_id}/operations" + ) + self.assertEqual(operations_response.status_code, 200) + failed_operation = operations_response.json()[0] + self.assertEqual(failed_operation["operation_type"], "start_inference") + self.assertEqual(failed_operation["status"], "failed") + self.assertEqual(failed_operation["error"]["status_code"], 503) + + with patch( + "server_api.main._proxy_to_worker", + return_value={"status": "started", "pid": 4344}, + ): + retry_response = self.client.post( + f"/api/workflows/{workflow_id}/commands/{command_id}/run" + ) + + self.assertEqual(retry_response.status_code, 200) + self.assertEqual(retry_response.json()["operation"]["status"], "succeeded") + self.assertEqual( + retry_response.json()["operation"]["idempotency_key"], + f"workflow-command:{command_id}:attempt:2", + ) + def test_durable_training_command_failure_records_retryable_operation(self): workflow_id = self._workflow_id() project_root = pathlib.Path(self.temp_dir.name) / "failed-command-project" @@ -756,14 +929,12 @@ def test_runtime_log_lines_are_exported_to_app_event_log(self): self.assertEqual(records[1]["stream"], "stdout") def test_detect_chunk_tile_mismatch_for_direct_h5_volume(self): - diagnostic = model_service._detect_chunk_tile_mismatch( - """ + diagnostic = model_service._detect_chunk_tile_mismatch(""" DATASET: DO_CHUNK_TITLE: 1 IMAGE_NAME: /tmp/train-volume.h5 LABEL_NAME: /tmp/train-label.h5 -""" - ) +""") self.assertIsNotNone(diagnostic) self.assertEqual(diagnostic["code"], "tile_dataset_direct_volume_mismatch") diff --git a/tests/test_workflow_routes.py b/tests/test_workflow_routes.py index e01a08ee..1c47eb68 100644 --- a/tests/test_workflow_routes.py +++ b/tests/test_workflow_routes.py @@ -512,7 +512,11 @@ def test_agent_action_approve_and_reject_flow(self): ) self.assertEqual(effects_approval.status_code, 200) effects_payload = effects_approval.json() - self.assertEqual(effects_payload["commands"], []) + self.assertEqual(len(effects_payload["commands"]), 1) + self.assertEqual( + effects_payload["commands"][0]["command_type"], "start_inference" + ) + self.assertEqual(effects_payload["commands"][0]["status"], "queued") self.assertEqual( effects_payload["client_effects"]["set_inference_output_path"], "/tmp/inference-out", @@ -526,7 +530,7 @@ def test_agent_action_approve_and_reject_flow(self): event_types = [event["event_type"] for event in events_response.json()] self.assertIn("agent.proposal_approved", event_types) self.assertIn("agent.proposal_rejected", event_types) - self.assertIn("agent.client_effects_approved", event_types) + self.assertIn("inference.run_approved", event_types) self.assertIn("retraining.staged", event_types) def test_agent_plan_preview_control_and_bundle_export(self):