feat(#10): Wave C — multi-gate resume, TTY inline confirm, parked-run e2e

src/caw/cli.py

@@ -2,13 +2,16 @@
 
 Exit code contract:
 
-- 0: success (`caw run` / `caw resume`: the Run succeeded; `caw validate`:
-  the workflow is valid; `caw graph`: the plan was rendered)
-- 1: the Run finished with a failed Node (`caw run`, `caw resume`)
+- 0: success (`caw run` / `caw resume`: the Run succeeded, or parked at a
+  human gate awaiting approval; `caw validate`: the workflow is valid;
+  `caw graph`: the plan was rendered)
+- 1: the Run finished with a failed Node, or a human gate rejected the Run
+  (`caw run`, `caw resume`)
 - 2: config error (unreadable file or invalid workflow definition);
   config errors print exactly one `error:` line. `caw resume` also exits 2
   when the run id is unknown or the Run is not resume-eligible (it already
-  succeeded) — a refusal, with one `error:` line and no re-execution.
+  succeeded or was rejected) — a refusal, with one `error:` line and no
+  re-execution.
 - 3: infrastructure error (e.g. unwritable runs root, State database
   failure) — the Run could not be executed or completed (`caw run`,
   `caw resume`)
@@ -27,6 +30,7 @@
 import asyncio
 import json
 import sqlite3
+import sys
 from collections.abc import Coroutine
 from enum import StrEnum
 from pathlib import Path
@@ -60,7 +64,14 @@
     execute_run,
     resume_run,
 )
-from caw.model import Node, Predicate, Workflow, execution_order, normalize_workflow
+from caw.model import (
+    HumanGateNodeInputs,
+    Node,
+    Predicate,
+    Workflow,
+    execution_order,
+    normalize_workflow,
+)
 from caw.patterns import expander_names, get_expander
 from caw.report import GroupReportError, ReportFormat, render_group_report, render_report
 from caw.runlayout import run_dir, runs_root
@@ -455,12 +466,55 @@ def _report_and_exit(result: RunResult, workflow_label: str) -> None:
     typer.echo(f"run {result.run_id} succeeded")
 
 
+def _is_attended() -> bool:
+    """Whether this is an interactive (TTY) session that can prompt at a human gate (#10)."""
+    return sys.stdin.isatty() and sys.stdout.isatty()
+
+
+def _drive_tty_gates(result: RunResult, gate_prompts: dict[str, str | None]) -> RunResult:
+    """In an attended session, prompt at each awaiting gate and advance the run (#10).
+
+    A parked run in a TTY prompts inline for the awaiting gates — yes approves, no
+    rejects. Because ANY rejection ends the run (ADR 0010), the FIRST decline commits
+    immediately and stops prompting: the later gates' decisions can no longer matter,
+    so a subsequent prompt (or an abort/EOF at one) can never drop a recorded decline.
+    Approvals are committed once the pass approves every awaiting gate, looping until
+    the run reaches a terminal. In a non-TTY session the run stays parked for
+    `caw resume`, so this is a no-op.
+    """
+    while result.parked and _is_attended():
+        approvals: list[str] = []
+        declined: str | None = None
+        for node_id in result.awaiting_node_ids:
+            prompt = gate_prompts.get(node_id) or f"Approve gate {node_id!r}?"
+            if typer.confirm(prompt):
+                approvals.append(node_id)
+            else:
+                declined = node_id
+                break
+        if declined is not None:
+            return asyncio.run(resume_run(result.run_id, runs_root(), rejections=(declined,)))
+        result = asyncio.run(resume_run(result.run_id, runs_root(), approvals=tuple(approvals)))
+    return result
+
+
 @app.command()
 def run(workflow_file: Path) -> None:
-    """Run a workflow file and print a plain-text result."""
+    """Run a workflow file and print a plain-text result.
+
+    In an attended (TTY) session a human_gate prompts inline (#10): yes approves it and
+    the run continues, no rejects it and ends the run. In a non-TTY session the run
+    parks for `caw resume`.
+    """
     workflow = _load_normalized_workflow(workflow_file)
+    gate_prompts: dict[str, str | None] = {
+        node.id: node.inputs.prompt
+        for node in workflow.nodes
+        if isinstance(node.inputs, HumanGateNodeInputs)
+    }
     try:
         result = asyncio.run(execute_run(workflow, runs_root()))
+        result = _drive_tty_gates(result, gate_prompts)
     except (OSError, sqlite3.Error) as exc:
         typer.echo(f"error: {exc}", err=True)
         raise typer.Exit(code=3) from exc

src/caw/executor.py

@@ -584,6 +584,7 @@ def __init__(
         satisfied_seed: Mapping[str, str] | None = None,
         attempt_seed: Mapping[str, int] | None = None,
         started_seed: set[str] | None = None,
+        awaiting_seed: set[str] | None = None,
     ) -> None:
         self._state = state
         self._events = events
@@ -619,6 +620,11 @@ def __init__(
         # so the Run reaches a fixpoint and parks rather than running past the gate
         # (#10, ADR 0010).
         self._awaiting: set[str] = set()
+        # Gates already `awaiting` from a prior parked Run (seeded on resume): a gate
+        # in this set that re-parks is a CONTINUATION, not a new park, so it re-records
+        # `awaiting` silently and does NOT re-emit gate_awaiting — the trace marks the
+        # entry into awaiting once, not on every resume (#10).
+        self._awaiting_already: set[str] = set(awaiting_seed or ())
         # 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
@@ -753,7 +759,11 @@ def _park_gate(self, node: Node) -> None:
             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})
+        # Emit gate_awaiting only on the FIRST park — a fresh run, or a gate newly
+        # reached during this resume. A gate already awaiting from a prior parked Run
+        # re-parks silently: it never left awaiting, so the trace marks the entry once.
+        if node.id not in self._awaiting_already:
+            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.
@@ -1306,6 +1316,9 @@ async def resume_run(
                 f"not forward-compatible with this version"
             )
         node_statuses = state.node_statuses(run_id)
+        # Gates already awaiting before this resume: an unapproved one re-parks as a
+        # CONTINUATION, so the scheduler must not re-emit gate_awaiting for it (#10).
+        awaiting_before = {nid for nid, status in node_statuses.items() if status == AWAITING}
         # Duplicate decision ids collapse to one, order-preserving, so a repeated
         # --approve/--reject for a gate is idempotent rather than crashing on the
         # attempt PK or double-recording the rejection (#10 review).
@@ -1380,5 +1393,6 @@ async def resume_run(
             satisfied_seed=satisfied,
             attempt_seed=attempt_seed,
             started_seed=started_seed,
+            awaiting_seed=awaiting_before,
         )
         return await _drive_scheduler(scheduler, state, events, run_id)

tests/e2e/test_human_gate_parked_run.py

@@ -0,0 +1,104 @@
+"""Real-agent-CLI e2e: a real run parks at a human_gate and `caw report` renders it (#10).
+
+Deferred here from #90: the Reporter renders parked/awaiting status-agnostically (covered
+offline by the report-seam suite), so what this e2e adds is the REAL flow — a real agent
+Node runs through ``execute_run``, the run then parks at a downstream ``human_gate`` (ADR
+0010), and the report surfaces the parked run and the awaiting gate from persisted State.
+The agent is selected by ``CAW_E2E_AGENT`` (default ``claude``); the suite FAILS (never
+skips) when the selected CLI is absent.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from caw.adapter import AdapterRegistry
+from caw.executor import RunResult, execute_run
+from caw.model import Workflow, normalize_workflow
+from caw.report import ReportFormat, render_report
+from e2e import harness
+
+# A generous per-Node wall-clock budget so ordinary model latency never trips the
+# kernel's timeout; a genuine hang still fails rather than blocking forever.
+_NODE_TIMEOUT_S = 300.0
+_AGENT_ID = "agent"
+_GATE_ID = "gate"
+
+
+def _gated_agent_workflow(agent: str) -> Workflow:
+    """A real agent Node followed by a human_gate: agent -> gate (deploy is gated)."""
+    inputs: dict[str, Any] = {
+        "adapter": harness.adapter_for_agent(agent),
+        "prompt": "Reply with a one-word greeting.",
+        "env": list(harness.agent_env_names()),
+    }
+    run_args = harness.agent_run_args(agent)
+    if run_args:
+        inputs["args"] = list(run_args)
+    raw = {
+        "name": "e2e-gated",
+        "version": 1,
+        "nodes": [
+            {"id": _AGENT_ID, "kind": "agent", "timeout": _NODE_TIMEOUT_S, "inputs": inputs},
+            {
+                "id": _GATE_ID,
+                "kind": "human_gate",
+                "needs": [_AGENT_ID],
+                "inputs": {"prompt": "Approve the deploy?"},
+            },
+        ],
+    }
+    return normalize_workflow(raw, source="<e2e>")
+
+
+def _why(result: RunResult) -> str:
+    """A debuggable reason string surfacing failed Nodes' stderr in an assertion."""
+    return "; ".join(
+        f"{node.node_id}: {node.status}: {node.stderr.strip()}"
+        for node in result.node_results
+        if not node.succeeded
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_real_agent_run_parks_at_a_human_gate_and_reports_parked(
+    agent: str, tmp_path: Path
+) -> None:
+    # A real agent Node runs, then the run parks at the downstream human_gate: the run
+    # is `parked`, the agent node `succeeded`, the gate `awaiting`, and `caw report`
+    # surfaces all of that from persisted State in JSON and Markdown.
+    harness.require_agent_cli(agent)  # FAIL (not skip) when the selected CLI is absent
+    workflow = _gated_agent_workflow(agent)
+    runs_root = tmp_path / "runs"
+
+    async def do_run() -> RunResult:
+        return await execute_run(workflow, runs_root, registry=AdapterRegistry())
+
+    result = await harness.run_with_transient_retry(do_run)
+
+    assert result.status == "parked", f"expected a parked run: {_why(result)}"
+    assert result.awaiting_node_ids == (_GATE_ID,)
+
+    run_dir = runs_root / result.run_id
+    report: dict[str, Any] = json.loads(render_report(run_dir, ReportFormat.json))
+
+    assert report["status"] == "parked"
+    agent_node = next(item for item in report["nodes"] if item["id"] == _AGENT_ID)
+    assert agent_node["status"] == "succeeded", "the real agent node ran before the gate"
+    gate_node = next(item for item in report["nodes"] if item["id"] == _GATE_ID)
+    assert gate_node["status"] == "awaiting"
+    assert gate_node["error"] is None, "an awaiting gate is not a failure"
+    assert any(
+        event["type"] == "gate_awaiting" and event["data"]["node_id"] == _GATE_ID
+        for event in report["trace"]
+    )
+
+    # Markdown renders the same parked run without error: the awaiting gate is visible.
+    markdown = render_report(run_dir, ReportFormat.markdown)
+    assert f"# Run {result.run_id}" in markdown
+    assert _GATE_ID in markdown
+    assert "awaiting" in markdown

tests/test_cli_seam.py

@@ -1515,3 +1515,93 @@ def test_resume_reject_ends_the_run(
     assert "rejected" in rejected.output
     assert "deploy" not in rejected.output, "a rejected run never runs the gated downstream"
     assert "succeeded" not in rejected.output
+
+
+def test_run_in_a_tty_prompts_and_approves_inline(
+    write_workflow_data: Callable[[dict[str, Any]], Path],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # In an attended (TTY) session, `caw run` prompts at the gate inline; answering
+    # yes approves it and the run continues to success without a separate
+    # `caw resume` (#10, ADR 0010).
+    workflow_file = write_workflow_data(_gated_workflow_data())
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr("caw.cli._is_attended", lambda: True)
+
+    result = runner.invoke(app, ["run", str(workflow_file)], input="y\n")
+
+    assert result.exit_code == 0, result.output
+    assert "succeeded" in result.output
+    assert "node deploy attempt 1 exited 0" in result.output
+
+
+def test_run_in_a_tty_declines_and_rejects_inline(
+    write_workflow_data: Callable[[dict[str, Any]], Path],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Declining at the inline prompt rejects the gate and ends the run (exit 1).
+    workflow_file = write_workflow_data(_gated_workflow_data())
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr("caw.cli._is_attended", lambda: True)
+
+    result = runner.invoke(app, ["run", str(workflow_file)], input="n\n")
+
+    assert result.exit_code == 1, result.output
+    assert "rejected" in result.output
+    assert "deploy" not in result.output
+
+
+def _multi_gated_workflow_data() -> dict[str, Any]:
+    """Two parallel gated branches: build -> gate{A,B} -> deploy{A,B}."""
+    return {
+        "name": "multi-gated",
+        "version": 1,
+        "nodes": [
+            {"id": "build", "kind": "shell", "inputs": {"command": "echo built"}},
+            {"id": "gateA", "kind": "human_gate", "needs": ["build"], "inputs": {"prompt": "A?"}},
+            {"id": "gateB", "kind": "human_gate", "needs": ["build"], "inputs": {"prompt": "B?"}},
+            {
+                "id": "deployA",
+                "kind": "shell",
+                "needs": ["gateA"],
+                "inputs": {"command": "echo a"},
+            },
+            {
+                "id": "deployB",
+                "kind": "shell",
+                "needs": ["gateB"],
+                "inputs": {"command": "echo b"},
+            },
+        ],
+    }
+
+
+def test_run_in_a_tty_declines_the_first_of_two_gates_and_ends_the_run(
+    write_workflow_data: Callable[[dict[str, Any]], Path],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # With two parallel TTY gates, declining the FIRST ends the run immediately — the
+    # decline is committed before the second gate is ever prompted, so no later prompt
+    # or abort can drop it (#10, ADR 0010 review). Only ONE answer is supplied: if the
+    # CLI prompted both gates first, the second prompt would EOF/abort instead.
+    workflow_file = write_workflow_data(_multi_gated_workflow_data())
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr("caw.cli._is_attended", lambda: True)
+
+    result = runner.invoke(app, ["run", str(workflow_file)], input="n\n")
+
+    assert result.exit_code == 1, result.output
+    assert "rejected" in result.output
+    assert "parked" not in result.output, "the decline ended the run, not re-parked it"
+
+    run_dir = next((tmp_path / ".caw" / "runs").iterdir())
+    events = [
+        json.loads(line)
+        for line in (run_dir / "events.jsonl").read_text(encoding="utf-8").splitlines()
+    ]
+    assert any(event["type"] == "gate_rejected" for event in events), (
+        "the decline was persisted as a gate_rejected event"
+    )