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

* feat(executor): multi-gate resume and re-park event dedup (#10)

Two parallel human_gates both park at the fixpoint; approving one advances its
branch and the run re-parks on the rest, and approving the rest completes the
run -- which the slice-2/3 machinery already supported, now pinned by a test.

A gate that stays awaiting across a resume (unapproved) is a CONTINUATION, not a
new park: the scheduler is seeded with the gates already awaiting from the prior
parked run (awaiting_seed) and re-parks them silently, so gate_awaiting is
emitted once on first park rather than re-emitted on every resume. A gate newly
reached during a resume still emits on its first park.

Slice 5 of #10 (Wave C). TTY inline confirmation and the parked-run e2e follow.

* feat(cli): inline TTY confirmation at a human gate during caw run (#10)

In an attended (TTY) session `caw run` now prompts at each awaiting gate inline:
yes approves it and the run continues, no rejects it and ends the run. The CLI
loops run -> prompt -> resume until the run reaches a terminal, showing the gate's
prompt text. In a non-TTY session the run parks for `caw resume` (unchanged), so
the executor is untouched -- this is pure CLI orchestration over the existing
park/approve/reject primitives.

Slice 6 of #10 (Wave C). The real parked-run report e2e follows.

* test(e2e): real agent run parks at a human gate and reports parked (#10)

The parked-run e2e deferred from #90: a real agent Node runs through execute_run,
the run then parks at a downstream human_gate, and `caw report` surfaces the
parked run, the succeeded agent node, and the awaiting gate from persisted State
in JSON and Markdown. Confirms the Reporter renders parked/awaiting without change
(an awaiting gate is not a failure: its `error` is None), end to end with a real
agent CLI.

Slice 7 of #10 (Wave C).

* fix(cli): commit a TTY gate decline immediately, before prompting the rest (#10 review)

Addresses the #123 review (two reviewers, one correctness finding): a declined TTY
gate could be lost when several gates are awaiting. `_drive_tty_gates` prompted every
awaiting gate first, then resumed only after the loop -- so answering `n` to the first
of two parallel gates and then EOF/aborting at the second exited "Aborted" with the run
still `parked`, both gates `awaiting`, and no gate_rejected written.

Because ANY rejection ends the run (ADR 0010), the FIRST decline now commits
immediately and stops prompting: the later gates' decisions can no longer matter, so no
subsequent prompt or abort can drop a recorded decline. Approvals still batch until the
pass approves every gate.

Also completes the module exit-code contract for the gate semantics (parked = 0,
rejected = 1). Regression test: two parallel TTY gates with input `n\n` ends the run
rejected and persists a gate_rejected event (one answer suffices -- the second gate is
never prompted).

---------

Co-authored-by: haihong.qin <haihongqin@gmail.com>

Author: qinhaihong-red

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