diff --git a/src/caw/cli.py b/src/caw/cli.py index fd66dc8..dc3fbb0 100644 --- a/src/caw/cli.py +++ b/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 diff --git a/src/caw/executor.py b/src/caw/executor.py index 3c0b610..97e790d 100644 --- a/src/caw/executor.py +++ b/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) diff --git a/tests/e2e/test_human_gate_parked_run.py b/tests/e2e/test_human_gate_parked_run.py new file mode 100644 index 0000000..31c69dd --- /dev/null +++ b/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="") + + +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 diff --git a/tests/test_cli_seam.py b/tests/test_cli_seam.py index fb1100c..7b86612 100644 --- a/tests/test_cli_seam.py +++ b/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" + ) diff --git a/tests/test_executor_seam.py b/tests/test_executor_seam.py index 6e430a7..18bc329 100644 --- a/tests/test_executor_seam.py +++ b/tests/test_executor_seam.py @@ -58,6 +58,81 @@ def gated_workflow() -> Workflow: return normalize_workflow(raw, source="") +def multi_gated_workflow() -> Workflow: + """Two parallel gated branches: build -> gate{A,B} -> deploy{A,B}.""" + raw: dict[str, Any] = { + "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"}, + }, + ], + } + return normalize_workflow(raw, source="") + + +@pytest.mark.asyncio +async def test_multi_gate_approve_one_reparks_then_approve_rest_completes(tmp_path: Path) -> None: + # Two parallel gates both park at the fixpoint; approving one advances its branch + # and re-parks on the rest, and approving the rest completes the run (#10, ADR 0010). + runs_root = tmp_path / "runs" + parked = await execute_run(multi_gated_workflow(), runs_root) + assert parked.status == "parked" + assert set(parked.awaiting_node_ids) == {"gateA", "gateB"} + + def node_statuses() -> dict[str, str]: + return { + row["node_id"]: row["status"] + for row in state_rows(single_run_dir(runs_root), "SELECT node_id, status FROM node") + } + + first = await resume_run(parked.run_id, runs_root, approvals=("gateA",)) + assert first.status == "parked", "the unapproved gate re-parks the run" + assert set(first.awaiting_node_ids) == {"gateB"} + statuses = node_statuses() + assert statuses["gateA"] == "succeeded" + assert statuses["deployA"] == "succeeded" + assert statuses["gateB"] == "awaiting" + assert "deployB" not in statuses, "the still-gated branch never ran" + + second = await resume_run(parked.run_id, runs_root, approvals=("gateB",)) + assert second.status == "succeeded" + statuses = node_statuses() + assert statuses["gateB"] == "succeeded" + assert statuses["deployB"] == "succeeded" + + +@pytest.mark.asyncio +async def test_a_re_parked_gate_does_not_re_emit_gate_awaiting(tmp_path: Path) -> None: + # A gate that stays awaiting across a resume (it was not approved) is not a NEW + # park: gate_awaiting fires once when it first parks, not again on every re-park, + # so the trace does not accumulate phantom parks for a continuously-awaiting gate. + runs_root = tmp_path / "runs" + parked = await execute_run(multi_gated_workflow(), runs_root) + await resume_run(parked.run_id, runs_root, approvals=("gateA",)) # gateB re-parks + + events = read_events(single_run_dir(runs_root)) + gate_b_awaiting = [ + event + for event in events + if event["type"] == "gate_awaiting" and event["data"]["node_id"] == "gateB" + ] + assert len(gate_b_awaiting) == 1, "gateB parked once; a re-park does not re-emit" + + @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`,