feat(#10): Wave A — human_gate node kind + await parking

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:

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))

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,8 @@
 from caw.status import (
     ERRORED,
     FAILED,
+    PARKED,
+    REJECTED,
     SKIPPED,
     SUCCEEDED,
     TIMED_OUT,
@@ -165,16 +168,33 @@ 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
     node_results: tuple[NodeResult, ...]
     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:
+        # 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
@@ -184,6 +204,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 +480,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 +600,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 +635,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 +689,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 +728,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 +1045,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 +1123,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__
@@ -1070,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
 

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
@@ -636,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."""

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,