From d47d22fee20cc742b150a33d064a9092e9d07e25 Mon Sep 17 00:00:00 2001 From: "haihong.qin" Date: Wed, 17 Jun 2026 14:59:42 +0800 Subject: [PATCH 1/3] feat(model): add the human_gate node kind (#10) The third node kind (ADR 0010): a human_gate parks the Run for approval. Its only input is an optional prompt for the TTY confirmation; it carries none of the shell/agent subprocess fields and emits no `when`-producible field, so any `when` ref to a gate is a config error. Registered in the kind->inputs and kind->producible maps and the node-kind Literal. Slice 1 of #10 (Wave A). Parking behavior follows. --- src/caw/model.py | 38 ++++++++++++++++++++++----- tests/test_model.py | 64 ++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 95 insertions(+), 7 deletions(-) diff --git a/src/caw/model.py b/src/caw/model.py index 75a9af9..0e5af09 100644 --- a/src/caw/model.py +++ b/src/caw/model.py @@ -148,14 +148,36 @@ def _adapter_must_be_known(cls, adapter: str, info: ValidationInfo) -> str: return adapter +class HumanGateNodeInputs(BaseModel): + """Inputs of a human_gate Node: it parks the Run for approval (#10, ADR 0010). + + A human_gate spawns no process — it records itself ``awaiting`` and the Run + parks at the scheduler fixpoint until approved or rejected. Its only input is an + optional ``prompt`` shown at the interactive TTY confirmation; it carries none + of the shell/agent subprocess fields (``timeout``/``retries``/``env``/``cwd``/ + ``output_schema``), which do not apply to a node that runs nothing. On approval + the gate emits ``{"approved": true}`` as its normalized output, but that field + is intentionally NOT ``when``-producible in v0.1 (declining ends the Run, so a + downstream branch-on-decision has no meaning yet). + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + kind: Literal["human_gate"] = "human_gate" + prompt: str | None = None + + # The node-level `kind` is the single source of truth (#62): it selects which # inputs model is built, so the top-level kind, the `caw graph` plan, and the # executor dispatch can never disagree. Because the concrete model is chosen by # kind (not by a discriminated union), validation errors name the authored field # directly (`inputs.command`) with no injected discriminator tag to strip. -_INPUTS_MODEL_FOR_KIND: dict[str, type[ShellNodeInputs | AgentNodeInputs]] = { +_INPUTS_MODEL_FOR_KIND: dict[ + str, type[ShellNodeInputs | AgentNodeInputs | HumanGateNodeInputs] +] = { "shell": ShellNodeInputs, "agent": AgentNodeInputs, + "human_gate": HumanGateNodeInputs, } @@ -187,10 +209,10 @@ def _resolve_inputs_paths(inputs: dict[str, Any], context: Any) -> dict[str, Any def _build_inputs( - model: type[ShellNodeInputs | AgentNodeInputs], + model: type[ShellNodeInputs | AgentNodeInputs | HumanGateNodeInputs], inputs: dict[str, Any], context: Any, -) -> ShellNodeInputs | AgentNodeInputs: +) -> ShellNodeInputs | AgentNodeInputs | HumanGateNodeInputs: """Build the kind's inputs model, re-raising any failure under the `inputs` field. Validating the inputs model here (rather than via a discriminated union on the @@ -254,6 +276,10 @@ def _reprefix_inputs(error: Any) -> Any: _PRODUCIBLE_FIELDS_FOR_KIND: dict[str, frozenset[str]] = { "shell": frozenset({"stdout", "exit_status"}), "agent": frozenset({"stdout", "exit_status", "structured_output"}), + # A human_gate runs no process, so it emits no stdout/exit_status; its + # approval output (`approved`) is deliberately not `when`-referenceable in + # v0.1 (ADR 0010). An empty set makes any `when` ref to a gate a config error. + "human_gate": frozenset(), } @@ -541,13 +567,13 @@ def _concat(children: list[tuple[PredicateRef, ...]]) -> tuple[PredicateRef, ... class Node(BaseModel): - """A unit of work in a Workflow; v0.1 supports shell and agent Nodes.""" + """A unit of work in a Workflow; v0.1 supports shell, agent, and human_gate Nodes.""" model_config = ConfigDict(frozen=True, extra="forbid") id: str - kind: Literal["shell", "agent"] - inputs: ShellNodeInputs | AgentNodeInputs + kind: Literal["shell", "agent", "human_gate"] + inputs: ShellNodeInputs | AgentNodeInputs | HumanGateNodeInputs needs: tuple[str, ...] = () # A node-level `when` predicate gates whether the Node runs (#7): a false # predicate marks the Node `skipped` without executing it. `when` is the ONLY diff --git a/tests/test_model.py b/tests/test_model.py index 509251d..7808d43 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -6,7 +6,14 @@ import pytest from caw.config import WorkflowConfigError -from caw.model import Node, ShellNodeInputs, Workflow, execution_order, normalize_workflow +from caw.model import ( + HumanGateNodeInputs, + Node, + ShellNodeInputs, + Workflow, + execution_order, + normalize_workflow, +) def shell_node(node_id: str, *needs: str) -> Node: @@ -15,6 +22,61 @@ def shell_node(node_id: str, *needs: str) -> Node: ) +def test_human_gate_node_normalizes_with_an_optional_prompt() -> None: + # A human_gate Node is a recognized kind (#10, ADR 0010): it parks the run for + # approval. Its only input is an optional prompt shown at the TTY confirmation; + # it spawns no process, so it carries none of the shell/agent subprocess fields. + raw: dict[str, Any] = { + "name": "sample", + "version": 1, + "nodes": [ + {"id": "build", "kind": "shell", "inputs": {"command": "echo hi"}}, + { + "id": "gate", + "kind": "human_gate", + "needs": ["build"], + "inputs": {"prompt": "Approve deploy?"}, + }, + ], + } + + workflow = normalize_workflow(raw, source="workflow.yaml") + + gate = next(node for node in workflow.nodes if node.id == "gate") + assert gate.kind == "human_gate" + assert isinstance(gate.inputs, HumanGateNodeInputs) + assert gate.inputs.prompt == "Approve deploy?" + + +def test_when_cannot_reference_a_human_gate_output() -> None: + # A human_gate emits no `when`-producible field (ADR 0010): its approval is not + # a branch source in v0.1, so any `when` ref to a gate is a config error — like + # a ref to a field a node's kind never emits (#75). + raw: dict[str, Any] = { + "name": "sample", + "version": 1, + "nodes": [ + {"id": "gate", "kind": "human_gate", "inputs": {"prompt": "ok?"}}, + { + "id": "after", + "kind": "shell", + "needs": ["gate"], + "inputs": {"command": "echo hi"}, + "when": { + "ref": {"node": "gate", "field": "exit_status"}, + "op": "equals", + "value": 0, + }, + }, + ], + } + + with pytest.raises(WorkflowConfigError) as excinfo: + normalize_workflow(raw, source="workflow.yaml") + + assert "gate" in str(excinfo.value) + + def test_cycle_error_names_only_the_cycle_members_not_downstream_nodes() -> None: raw: dict[str, Any] = { "name": "sample", From 092e2e5ffa669201f8807c5b0ccef4c50f2c563e Mon Sep 17 00:00:00 2001 From: "haihong.qin" Date: Wed, 17 Jun 2026 15:11:26 +0800 Subject: [PATCH 2/3] feat(executor): park the run at a human_gate (#10) A human_gate node now parks the run at the scheduler fixpoint instead of executing: the gate is recorded `awaiting`, the run `parked` (no finished_at and no run_finished -- it is not finished), downstream stays unreached, and a gate_awaiting event marks the park. `caw run` reports the parked run and its awaiting gate and exits 0 (a park is not a failure) in a non-TTY session. Extends the #30 vocabulary owners with the gate statuses (parked/awaiting/ rejected) and events (gate_*). RunResult carries awaiting_node_ids and reports `parked`; a human_gate reaching the node dispatch is a guarded invariant breach. Slice 2 of #10 (Wave A). Approval/rejection follow in Wave B. --- src/caw/cli.py | 8 ++++ src/caw/events.py | 5 +++ src/caw/executor.py | 76 +++++++++++++++++++++++++++++++++++-- src/caw/state.py | 29 +++++++++++++- src/caw/status.py | 18 +++++++-- tests/test_cli_seam.py | 39 +++++++++++++++++++ tests/test_executor_seam.py | 52 +++++++++++++++++++++++++ 7 files changed, 219 insertions(+), 8 deletions(-) diff --git a/src/caw/cli.py b/src/caw/cli.py index dbd6232..cb84828 100644 --- a/src/caw/cli.py +++ b/src/caw/cli.py @@ -432,6 +432,14 @@ def _report_and_exit(result: RunResult, workflow_label: str) -> None: typer.echo(_failure_line(workflow_label, node_result)) if node_result.stderr: _echo_stderr_excerpt(node_result) + if result.parked: + # A parked Run is neither succeeded nor failed: it awaits approval at one or + # more human gates (#10, ADR 0010). Name the awaiting gates and exit 0 — a + # park is not a failure; the run is advanced later via `caw resume`. + for node_id in result.awaiting_node_ids: + typer.echo(f"node {node_id} awaiting approval") + typer.echo(f"run {result.run_id} parked at a human gate") + return for node_id in result.skipped_node_ids: typer.echo(f"node {node_id} skipped {_skip_reason(result, node_id)}") if not result.succeeded: diff --git a/src/caw/events.py b/src/caw/events.py index 4d71e5c..fcc82a3 100644 --- a/src/caw/events.py +++ b/src/caw/events.py @@ -22,6 +22,11 @@ "node_finished", "node_skipped", "node_retrying", + # Human Gate events (#10, ADR 0010): a gate parks the Run (awaiting), and an + # approval/rejection advances or ends it. + "gate_awaiting", + "gate_approved", + "gate_rejected", ] EVENT_TYPES: frozenset[str] = frozenset(get_args(EventType)) diff --git a/src/caw/executor.py b/src/caw/executor.py index ed7091f..5df9b4e 100644 --- a/src/caw/executor.py +++ b/src/caw/executor.py @@ -20,6 +20,7 @@ from caw.events import EventLog from caw.model import ( AgentNodeInputs, + HumanGateNodeInputs, Node, ShellNodeInputs, Workflow, @@ -32,6 +33,7 @@ from caw.status import ( ERRORED, FAILED, + PARKED, SKIPPED, SUCCEEDED, TIMED_OUT, @@ -165,6 +167,11 @@ class RunResult: ``join: any`` Node whose every dependency skipped). A ``blocked`` skip carries a ``skipped_blockers`` entry; the others carry none, so a Reporter renders a closed gate distinctly from withheld-by-failure work. + + ``awaiting_node_ids`` names the human_gate Nodes the Run is parked on (#10): a + non-empty set means the Run did not finish — it parked at the scheduler + fixpoint awaiting approval — and its ``status`` is ``parked`` regardless of how + the already-run Nodes fared. """ run_id: str @@ -172,6 +179,12 @@ class RunResult: skipped_node_ids: tuple[str, ...] = () skipped_blockers: Mapping[str, str] = field(default_factory=dict) skipped_causes: Mapping[str, str] = field(default_factory=dict) + awaiting_node_ids: tuple[str, ...] = () + + @property + def parked(self) -> bool: + """Whether the Run parked at a Human Gate rather than reaching a terminal (#10).""" + return bool(self.awaiting_node_ids) @property def succeeded(self) -> bool: @@ -184,6 +197,10 @@ def succeeded(self) -> bool: @property def status(self) -> RunStatus: + # A parked Run is neither succeeded nor failed — it is awaiting approval, so + # `parked` takes precedence over the run-down node outcomes (#10). + if self.parked: + return PARKED return SUCCEEDED if self.succeeded else FAILED @@ -456,6 +473,13 @@ async def _execute_node(node: Node, registry: AdapterRegistry) -> NodeResult: """ if isinstance(node.inputs, ShellNodeInputs): return await _execute_shell_node(node) + if isinstance(node.inputs, HumanGateNodeInputs): + # A human_gate launches no Attempt: the scheduler parks it in place + # (_park_gate). Reaching this dispatch means the scheduler failed to + # intercept it — an internal invariant breach, not a node failure (#10). + raise RuntimeError( + f"human_gate node {node.id!r} must be parked by the scheduler, not executed" + ) try: return await _execute_agent_node(node, registry) except Exception as exc: @@ -569,6 +593,12 @@ def __init__( # ``_skipped`` on every visit (which made a wide skip cone quadratic). self._skipped_set: set[str] = set() self._result_ids: set[str] = set() + # human_gate Nodes parked in place: recorded `awaiting`, launched as no + # task. Like a done Node they are excluded from readiness so they are never + # re-launched, but unlike a succeeded Node they do NOT satisfy dependents — + # so the Run reaches a fixpoint and parks rather than running past the gate + # (#10, ADR 0010). + self._awaiting: set[str] = set() # Per-Node Attempt bookkeeping for the in-run retry loop (#6). ``_attempt`` # is the Attempt NUMBER the next launch of a Node uses, so re-launched # Nodes write distinct ``attempt`` rows ((run_id, node_id, attempt) is the @@ -598,7 +628,7 @@ def in_flight_node_ids(self) -> tuple[str, ...]: def _ready_nodes(self) -> list[Node]: """Nodes whose needs are all satisfied and that are neither running nor done.""" running = {node.id for node in self._in_flight.values()} - done = self._result_ids | self._skipped_set | self._satisfied.keys() + done = self._result_ids | self._skipped_set | self._satisfied.keys() | self._awaiting return [ node for node in self._ordered @@ -652,6 +682,20 @@ def _launch_ready(self) -> None: while True: progressed = False for node in self._ready_nodes(): + if node.kind == "human_gate": + # A gate consumes no concurrency slot and launches no task. A + # gate whose own `when` gate closed is skipped like any Node, + # propagating to its dependents (#7); otherwise it parks in place + # (#10, ADR 0010). + if node.when is not None and not evaluate_predicate( + node.when, self._output_of + ): + self._skip_with_cause(node.id, cause=SKIP_WHEN_FALSE, blocker=None) + progressed = True + break + self._park_gate(node) + progressed = True + continue if len(self._in_flight) >= self._concurrency: break if node.when is not None and not evaluate_predicate(node.when, self._output_of): @@ -677,6 +721,20 @@ def _launch_ready(self) -> None: if not progressed: return + def _park_gate(self, node: Node) -> None: + """Record a human_gate Node as awaiting and park it in place (#10, ADR 0010). + + The gate launches no Attempt and holds no concurrency slot: it is recorded + ``awaiting`` in State and the Event trace and tracked in ``_awaiting`` so + readiness excludes it (never re-launched) WITHOUT satisfying its dependents, + so the Run reaches a fixpoint and parks once nothing else can progress. + """ + if node.id in self._awaiting: + return + self._state.record_node_awaiting(run_id=self._run_id, node_id=node.id) + self._awaiting.add(node.id) + self._events.append("gate_awaiting", {"node_id": node.id}) + def _record_attempt(self, node: Node, result: NodeResult) -> None: """Record one Attempt's outcome in State and the Event trace. @@ -980,6 +1038,11 @@ async def run(self) -> RunResult: skipped_node_ids=tuple(self._skipped), skipped_blockers=dict(self._skipped_blockers), skipped_causes=dict(self._skipped_causes), + # Deterministic declaration order (execution_order), so a parked Run's + # awaiting gates render stably (#10). + awaiting_node_ids=tuple( + node.id for node in self._ordered if node.id in self._awaiting + ), ) async def _drain_in_flight(self) -> None: @@ -1053,11 +1116,18 @@ async def _drive_scheduler( is recorded ``errored`` over every in-flight Node without masking the original exception. Shared by ``execute_run`` and ``resume_run`` so a resumed Run finalizes identically to a fresh one (#6). + + A Run that parked at a Human Gate is NOT finished: it is recorded ``parked`` + (no ``finished_at``) and emits no ``run_finished`` — the ``gate_awaiting`` + Events already mark the park, and a resume advances it (#10, ADR 0010). """ try: run_result = await scheduler.run() - state.record_run_finished(run_id=run_id, status=run_result.status, finished_at=_now()) - events.append("run_finished", {"status": run_result.status}) + if run_result.parked: + state.record_run_parked(run_id=run_id) + else: + state.record_run_finished(run_id=run_id, status=run_result.status, finished_at=_now()) + events.append("run_finished", {"status": run_result.status}) except BaseException as exc: message = str(exc) error = f"{type(exc).__name__}: {message}" if message else type(exc).__name__ diff --git a/src/caw/state.py b/src/caw/state.py index 82bddd9..f8130f6 100644 --- a/src/caw/state.py +++ b/src/caw/state.py @@ -19,7 +19,7 @@ from types import TracebackType from typing import Any -from caw.status import ERRORED, RUNNING, SKIPPED, NodeStatus, RunStatus +from caw.status import AWAITING, ERRORED, PARKED, RUNNING, SKIPPED, NodeStatus, RunStatus _SCHEMA = """ CREATE TABLE IF NOT EXISTS run ( @@ -125,6 +125,18 @@ def record_run_errored(self, run_id: str, error: str, finished_at: str) -> None: (ERRORED, error, finished_at, run_id), ) + def record_run_parked(self, run_id: str) -> None: + """Mark a Run ``parked`` at a Human Gate (#10, ADR 0010). + + A parked Run is not finished — it awaits approval and will resume — so, + unlike ``record_run_finished``, it sets no ``finished_at``. A resume flips + it back to ``running`` (``record_run_running``) before advancing. + """ + self._execute( + "UPDATE run SET status = ? WHERE run_id = ?", + (PARKED, run_id), + ) + def record_node_started(self, run_id: str, node_id: str) -> None: self._execute( "INSERT INTO node (run_id, node_id, status) VALUES (?, ?, ?)", @@ -173,6 +185,21 @@ def record_node_skipped(self, run_id: str, node_id: str, cause: str | None = Non (run_id, node_id, SKIPPED, cause), ) + def record_node_awaiting(self, run_id: str, node_id: str) -> None: + """Record a human_gate Node as ``awaiting`` approval (#10, ADR 0010). + + A gate launches no Attempt, so like a skipped Node it has no prior + ``running`` row on first park: it is inserted straight into ``awaiting``. + A plain resume of an already-parked Run re-parks the same gate, whose row + now exists, so the write UPSERTs on the ``(run_id, node_id)`` PK to stay + idempotent rather than breaching it. + """ + self._execute( + "INSERT INTO node (run_id, node_id, status) VALUES (?, ?, ?) " + "ON CONFLICT(run_id, node_id) DO UPDATE SET status = excluded.status", + (run_id, node_id, AWAITING), + ) + def record_attempt( self, run_id: str, diff --git a/src/caw/status.py b/src/caw/status.py index b0ae051..a5207ff 100644 --- a/src/caw/status.py +++ b/src/caw/status.py @@ -34,15 +34,25 @@ ERRORED: Final = "errored" CANCELLED: Final = "cancelled" SKIPPED: Final = "skipped" +# Human Gate statuses (#10, ADR 0010): a ``parked`` Run waits at a gate; an ``awaiting`` +# Node is the gate holding the Run; ``rejected`` is a human "no" that ended the Run, applied +# to both the declined gate Node and the Run. +PARKED: Final = "parked" +AWAITING: Final = "awaiting" +REJECTED: Final = "rejected" # A Run is in flight (``running``), finished cleanly (``succeeded``), finished with a failed # Node (``failed``), was prevented from producing a result by an Adapter/internal fault -# (``errored``), or was cancelled (``cancelled``). -RunStatus = Literal["running", "succeeded", "failed", "errored", "cancelled"] +# (``errored``), was cancelled (``cancelled``), is ``parked`` at a Human Gate awaiting +# approval, or was ``rejected`` by a human (#10). +RunStatus = Literal["running", "succeeded", "failed", "errored", "cancelled", "parked", "rejected"] # A Node is in flight (``running``) or reached a terminal outcome: ``succeeded``, one of the -# failure kinds (``failed`` / ``timed_out`` / ``errored``), or ``skipped`` (never attempted). -NodeStatus = Literal["running", "succeeded", "failed", "timed_out", "errored", "skipped"] +# failure kinds (``failed`` / ``timed_out`` / ``errored``), ``skipped`` (never attempted), +# ``awaiting`` (a Human Gate holding the Run), or ``rejected`` (a declined gate; #10). +NodeStatus = Literal[ + "running", "succeeded", "failed", "timed_out", "errored", "skipped", "awaiting", "rejected" +] # The Error Classification failure kinds a failed Node Attempt carries (a subset of # NodeStatus); ``None`` on a NodeResult means the Attempt succeeded. diff --git a/tests/test_cli_seam.py b/tests/test_cli_seam.py index ad9ced8..a30b313 100644 --- a/tests/test_cli_seam.py +++ b/tests/test_cli_seam.py @@ -1411,3 +1411,42 @@ def test_run_malformed_agent_node_is_a_config_error_before_executing_anything( assert lines[0].startswith("error:") assert "summarize" in lines[0], "the error names the node id" assert not (tmp_path / ".caw").exists(), "no run directory is created for invalid input" + + +def test_run_parking_at_a_human_gate_exits_zero_and_reports_parked( + write_workflow_data: Callable[[dict[str, Any]], Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # `caw run` reaching a human_gate parks cleanly in a non-TTY session (ADR 0010): + # the process exits 0 (a park is not a failure), names the awaiting gate, and does + # not report the Run succeeded. + workflow_file = write_workflow_data( + { + "name": "gated", + "version": 1, + "nodes": [ + {"id": "build", "kind": "shell", "inputs": {"command": "echo built"}}, + { + "id": "gate", + "kind": "human_gate", + "needs": ["build"], + "inputs": {"prompt": "Approve?"}, + }, + { + "id": "deploy", + "kind": "shell", + "needs": ["gate"], + "inputs": {"command": "echo deployed"}, + }, + ], + } + ) + monkeypatch.chdir(tmp_path) + + result = runner.invoke(app, ["run", str(workflow_file)]) + + assert result.exit_code == 0, result.output + assert "parked" in result.output + assert "gate" in result.output + assert "succeeded" not in result.output, "a parked run is not reported succeeded" diff --git a/tests/test_executor_seam.py b/tests/test_executor_seam.py index 378ad25..cbceeed 100644 --- a/tests/test_executor_seam.py +++ b/tests/test_executor_seam.py @@ -39,6 +39,58 @@ def shell_workflow(*nodes: ShellNodeSpec) -> Workflow: return normalize_workflow(raw, source="") +def gated_workflow() -> Workflow: + """A build -> human_gate -> deploy Workflow: deploy is gated behind approval.""" + raw: dict[str, Any] = { + "name": "gated", + "version": 1, + "nodes": [ + {"id": "build", "kind": "shell", "inputs": {"command": "echo built"}}, + {"id": "gate", "kind": "human_gate", "needs": ["build"], "inputs": {"prompt": "ok?"}}, + { + "id": "deploy", + "kind": "shell", + "needs": ["gate"], + "inputs": {"command": "echo deployed"}, + }, + ], + } + return normalize_workflow(raw, source="") + + +@pytest.mark.asyncio +async def test_a_run_reaching_a_human_gate_parks(tmp_path: Path) -> None: + # A human_gate parks the Run at the fixpoint (ADR 0010): the gate goes `awaiting`, + # the Run goes `parked`, State is persisted, downstream never runs, and the park + # is recorded as a gate_awaiting Event — all offline, with no TTY. + runs_root = tmp_path / "runs" + + result = await execute_run(gated_workflow(), runs_root) + + assert result.status == "parked" + assert result.awaiting_node_ids == ("gate",) + ran = {node.node_id for node in result.node_results} + assert "build" in ran and "deploy" not in ran, "downstream of the gate never runs" + + run_dir = single_run_dir(runs_root) + statuses = { + row["node_id"]: row["status"] + for row in state_rows(run_dir, "SELECT node_id, status FROM node") + } + assert statuses["build"] == "succeeded" + assert statuses["gate"] == "awaiting" + assert "deploy" not in statuses, "an unreached node has no row" + assert state_rows(run_dir, "SELECT status FROM run")[0]["status"] == "parked" + + events = read_events(run_dir) + assert any( + event["type"] == "gate_awaiting" and event["data"]["node_id"] == "gate" for event in events + ) + assert not any(event["type"] == "run_finished" for event in events), ( + "a parked Run is not finished" + ) + + def conditional_workflow(*nodes: dict[str, Any]) -> Workflow: """Build a shell Workflow from raw node dicts carrying `when` / `join` (#7). From 1721612f614f790e330b52d1681ab454e0f4174f Mon Sep 17 00:00:00 2001 From: "haihong.qin" Date: Wed, 17 Jun 2026 15:36:24 +0800 Subject: [PATCH 3/3] fix(human-gate): align gate validation, parked success, and rejected eligibility with ADR 0010 (#10 review) Addresses the #120 review (two reviewers, three converging findings): - Reject `retries`/`timeout` on a human_gate node. ADR 0010 says the subprocess-shaped fields do not apply (a gate runs no process); the inputs-level ones were already forbidden, but the node-level `retries`/`timeout` still validated and were silently ignored. They are now a config error. - A parked RunResult is no longer `succeeded`. `succeeded` returned True whenever the run-down nodes succeeded, even when parked -- risking a Pattern Controller that branches on `result.succeeded` misclassifying a parked iteration as a finished success. A parked run is now `succeeded == False` (status stays `parked`). - `rejected` is non-resumable. The status was added to the vocabulary but `_NON_RESUMABLE_RUN_STATUSES` still held only `succeeded`; ADR 0010 and CONTEXT.md refuse a rejected run alongside succeeded. `parked` stays resumable. Tests: a gate with retries/timeout is a config error; a parked run reports succeeded=False; is_resumable refuses rejected/succeeded and admits parked/failed. --- src/caw/executor.py | 27 ++++++++++++++++++--------- src/caw/model.py | 21 +++++++++++++++++++++ tests/test_executor_seam.py | 27 +++++++++++++++++++++++++++ tests/test_model.py | 19 +++++++++++++++++++ 4 files changed, 85 insertions(+), 9 deletions(-) diff --git a/src/caw/executor.py b/src/caw/executor.py index 5df9b4e..e728890 100644 --- a/src/caw/executor.py +++ b/src/caw/executor.py @@ -34,6 +34,7 @@ ERRORED, FAILED, PARKED, + REJECTED, SKIPPED, SUCCEEDED, TIMED_OUT, @@ -188,6 +189,12 @@ def parked(self) -> bool: @property def succeeded(self) -> bool: + # A parked Run is NOT a successful terminal — it awaits approval (#10) — so + # it is never `succeeded` even though its already-run Nodes all succeeded. + # This keeps consumers that branch on `succeeded` (e.g. Pattern Controllers) + # from treating a parked iteration as a finished success. + if self.parked: + return False # A Run fails iff an ATTEMPTED Node failed. A failure-driven (`blocked`) # skip always coincides with a failed Node already in `node_results`, so # it is captured here; a benign skip — a closed `when` gate or a fully @@ -1140,19 +1147,21 @@ class ResumeError(Exception): """Raised when a Run cannot be resumed: it is absent or not resume-eligible (#6).""" -# A Run that already SUCCEEDED has nothing left to do, so resuming it is refused; -# every other recorded status — a failed run, an errored/cancelled (interrupted) -# run, even a run still marked ``running`` because it was killed mid-flight — has -# incomplete work and IS resumable. Eligibility lives here so the entry point and -# the CLI share one rule. -_NON_RESUMABLE_RUN_STATUSES = frozenset({SUCCEEDED}) +# A Run that already SUCCEEDED has nothing left to do, and a REJECTED Run is a +# decided human "no" (#10, ADR 0010) — both are refused. Every other recorded +# status — a failed run, an errored/cancelled (interrupted) run, a `parked` run +# awaiting approval, even a run still marked ``running`` because it was killed +# mid-flight — has work left and IS resumable. Eligibility lives here so the entry +# point and the CLI share one rule. +_NON_RESUMABLE_RUN_STATUSES = frozenset({SUCCEEDED, REJECTED}) def is_resumable(run_status: str | None) -> bool: - """Whether a Run with this recorded status can be resumed (#6). + """Whether a Run with this recorded status can be resumed (#6, #10). - ``None`` (an unknown Run) is not resumable; a ``succeeded`` Run is not (nothing - to do); any other terminal/interrupted status is. + ``None`` (an unknown Run) is not resumable; a ``succeeded`` or ``rejected`` Run + is not (a decided terminal); any other terminal/interrupted status — including + ``parked`` — is. """ return run_status is not None and run_status not in _NON_RESUMABLE_RUN_STATUSES diff --git a/src/caw/model.py b/src/caw/model.py index 0e5af09..8a423b1 100644 --- a/src/caw/model.py +++ b/src/caw/model.py @@ -662,6 +662,27 @@ def _when_refs_must_be_dependencies(self) -> "Node": ) return self + @model_validator(mode="after") + def _human_gate_rejects_subprocess_fields(self) -> "Node": + # A human_gate spawns no process, so the subprocess-shaped Node fields do + # not apply (ADR 0010). The inputs-level ones (env/cwd/artifacts/ + # output_schema) are already forbidden by HumanGateNodeInputs; retries and + # timeout are node-level, so reject a non-default value here rather than + # silently ignoring an authored control that cannot affect a parked gate. + if self.kind != "human_gate": + return self + if self.retries != 0: + raise ValueError( + f"node {self.id!r} is a human_gate, which runs no process, so `retries` " + f"does not apply" + ) + if self.timeout is not None: + raise ValueError( + f"node {self.id!r} is a human_gate, which runs no process, so `timeout` " + f"does not apply" + ) + return self + class Workflow(BaseModel): """A normalized Workflow IR for one Run.""" diff --git a/tests/test_executor_seam.py b/tests/test_executor_seam.py index cbceeed..dda213c 100644 --- a/tests/test_executor_seam.py +++ b/tests/test_executor_seam.py @@ -91,6 +91,33 @@ async def test_a_run_reaching_a_human_gate_parks(tmp_path: Path) -> None: ) +@pytest.mark.asyncio +async def test_a_parked_run_is_not_reported_succeeded(tmp_path: Path) -> None: + # A parked Run is not a successful terminal — it awaits approval (#10) — so + # RunResult.succeeded is False even though its already-run Nodes all succeeded, + # keeping controllers that branch on `succeeded` from misclassifying it. + result = await execute_run(gated_workflow(), tmp_path / "runs") + + assert result.status == "parked" + assert result.succeeded is False + assert all(node.succeeded for node in result.node_results), ( + "the run-down nodes still succeeded" + ) + + +def test_rejected_and_succeeded_runs_are_not_resumable() -> None: + # ADR 0010 Resume Eligibility: a `rejected` Run is refused like `succeeded`; a + # `parked` Run, by contrast, is resumable (advanced by approve/reject), and the + # other interrupted terminals stay resumable (#6). + from caw.executor import is_resumable + + assert is_resumable("succeeded") is False + assert is_resumable("rejected") is False + assert is_resumable("parked") is True + assert is_resumable("failed") is True + assert is_resumable(None) is False + + def conditional_workflow(*nodes: dict[str, Any]) -> Workflow: """Build a shell Workflow from raw node dicts carrying `when` / `join` (#7). diff --git a/tests/test_model.py b/tests/test_model.py index 7808d43..cacb1e9 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -77,6 +77,25 @@ def test_when_cannot_reference_a_human_gate_output() -> None: assert "gate" in str(excinfo.value) +@pytest.mark.parametrize(("field", "value"), [("retries", 3), ("timeout", 1.0)]) +def test_human_gate_rejects_subprocess_node_fields(field: str, value: object) -> None: + # A human_gate spawns no process, so the subprocess-shaped Node fields do not + # apply (ADR 0010): authoring `retries`/`timeout` on a gate is a config error, + # not a silently-ignored value. + raw: dict[str, Any] = { + "name": "sample", + "version": 1, + "nodes": [ + {"id": "gate", "kind": "human_gate", "inputs": {"prompt": "ok?"}, field: value}, + ], + } + + with pytest.raises(WorkflowConfigError) as excinfo: + normalize_workflow(raw, source="workflow.yaml") + + assert field in str(excinfo.value) + + def test_cycle_error_names_only_the_cycle_members_not_downstream_nodes() -> None: raw: dict[str, Any] = { "name": "sample",