diff --git a/CHANGELOG.md b/CHANGELOG.md index b8d61ee..df046f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,13 @@ Notable changes to Cloudfall. The format follows ### Added +- `agent investigate` streams: one bare JSON line with `_seq` per model turn + (its words and how many tools it called), one per tool call (the step and + what the tool handed back, `effect: created` when it wrote a decision), + and the saved investigation record last (`kind: investigation`), then the + `"_summary": true` line with `effects`, or the error envelope when it + failed. `--no-stream` returns every event in one envelope. A watcher, such + as the hosted demo, shows the agent working instead of waiting for the end - A decision record's approval says how it arrived, as `approval.via`: `terminal` when a person typed the decision id at a terminal. A record written before reads as `via: unknown`. `cloudfall why` tells it in the diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index 5ac8378..70d2b9b 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -141,10 +141,12 @@ InvestigationStore, ModelEndpoint, ModelUnavailableError, + Progress, StepOutcome, ToolResult, + Turn, host_text, - investigate, + investigation_events, ) from cloudfall.lifecycle import ( ERROR_EXECUTION_FAILED, @@ -2138,12 +2140,13 @@ def decisions_approve(args: ApproveDecisionArgs, ctx: Ctx) -> Decided: 93, description=( "The model ran out of turns or did not answer in the asked-for shape; " - "data holds the record" + "the event before holds the record" ), retryable=True, side_effects="none", suggestion=( - "read data.investigation.spec.steps, then run again or raise --max-turns" + "read investigation.spec.steps of the event of kind investigation, " + "then run again or raise --max-turns" ), ) @@ -2225,12 +2228,23 @@ def endpoint(self) -> ModelEndpoint: @dataclass(frozen=True, slots=True) -class Investigated: - """``agent investigate``: the investigation record.""" +class InvestigationEvent: + """``agent investigate``: one event of the stream. + + ``turn`` is the model's words and how many tools it called; ``call`` is one + tool call and what the tool handed back; the last event, ``investigation``, + is the record as it was saved. + """ effect: str - status: str - investigation: dict[str, object] = Out(ordered=True) + kind: str + turn: int | None = None + text: str | None = None + calls: int | None = None + step: dict[str, object] = Out(default_factory=dict, ordered=True) + result: dict[str, object] = Out(default_factory=dict, ordered=True) + status: str | None = None + investigation: dict[str, object] = Out(default_factory=dict, ordered=True) @agent.command( @@ -2243,6 +2257,7 @@ class Investigated: # An agent loop is not a tool for another agent. mcp=False, danger_level="mutating", + streaming=True, exit_codes=[ "CONFIG_INVALID", "PRECONDITION", @@ -2264,8 +2279,10 @@ class Investigated: ), ], ) -def agent_investigate(args: InvestigateArgs, ctx: Ctx) -> Investigated: - """Work the alert through the catalog and keep the record.""" +def agent_investigate( + args: InvestigateArgs, ctx: Ctx +) -> Iterator[InvestigationEvent]: + """Work the alert through the catalog, step by step, and keep the record.""" catalog = args.catalog() try: replay = ( @@ -2273,25 +2290,33 @@ def agent_investigate(args: InvestigateArgs, ctx: Ctx) -> Investigated: ) except RecordingError as error: raise Exit.PRECONDITION(error.detail, context={"code": error.code}) from error - run =_check_runner(ctx, args.root) if replay is None else replay.runner(args.root) + run = _check_runner(ctx, args.root) if replay is None else replay.runner(args.root) tools = [_agent_tool(args, operation, run) for operation in catalog.operations] store = InvestigationStore( directory=args.root / args.investigations, catalog=SchemaCatalog(args.schemas), ) + events = investigation_events( + args.alert, + tools, + args.endpoint(), + _chat_transport(ctx, args.api_key), + max_turns=args.max_turns, + replay=None if args.replay is None else args.replay.as_posix(), + ) try: - investigation = investigate( - args.alert, - tools, - args.endpoint(), - _chat_transport(ctx, args.api_key), - max_turns=args.max_turns, - replay=None if args.replay is None else args.replay.as_posix(), - ) + while True: + try: + progress = next(events) + except StopIteration as stop: + investigation = cast("Investigation", stop.value) + break + yield _progress_event(progress) except ModelUnavailableError as error: # Earlier turns may have recorded decisions: the investigation that # produced them is recorded too, then the model's failure is reported. failed = _save_investigation(store, error.investigation) + yield _investigation_event(failed) raise Exit.MODEL_UNAVAILABLE( error.detail, context={ @@ -2300,15 +2325,48 @@ def agent_investigate(args: InvestigateArgs, ctx: Ctx) -> Investigated: }, ) from error investigation = _save_investigation(store, investigation) - investigated = Investigated( + yield _investigation_event(investigation) + if investigation.status is not InvestigationStatus.ANSWERED: + message = f"the investigation ended {investigation.status.value}" + raise Exit.INVESTIGATION_INCOMPLETE(message) + + +_RECORDING_OUTCOMES = frozenset( + {StepOutcome.RAN, StepOutcome.PROPOSED, StepOutcome.FAILED} +) + + +def _progress_event(progress: Progress) -> InvestigationEvent: + """Return one turn or call as an event; a call that wrote a decision created it.""" + if isinstance(progress, Turn): + return InvestigationEvent( + effect="noop", + kind="turn", + turn=progress.turn, + text=progress.text, + calls=progress.calls, + ) + wrote = ( + progress.step.decision is not None + and progress.step.outcome in _RECORDING_OUTCOMES + ) + return InvestigationEvent( + effect="created" if wrote else "noop", + kind="call", + turn=progress.turn, + step=progress.step.as_dict(), + result=dict(progress.result.content), + ) + + +def _investigation_event(investigation: Investigation) -> InvestigationEvent: + """Return the saved record as the stream's last event.""" + return InvestigationEvent( effect="created", + kind="investigation", status=investigation.status.value, investigation=investigation.as_document(), ) - if investigation.status is not InvestigationStatus.ANSWERED: - message = f"the investigation ended {investigation.status.value}" - raise Exit.INVESTIGATION_INCOMPLETE(message, data=investigated) - return investigated def _save_investigation( diff --git a/sdk/src/cloudfall/investigation.py b/sdk/src/cloudfall/investigation.py index cbc9ad5..6c19f25 100644 --- a/sdk/src/cloudfall/investigation.py +++ b/sdk/src/cloudfall/investigation.py @@ -16,7 +16,7 @@ import json import re -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Generator, Mapping, Sequence from dataclasses import dataclass, replace from datetime import UTC, datetime from enum import StrEnum @@ -322,6 +322,28 @@ def save(self, investigation: Investigation) -> Investigation: raise InvestigationError(ERROR_INVESTIGATION_EXISTS, message) +@dataclass(frozen=True, slots=True) +class Turn: + """The model answered one turn: its words, and how many tools it called.""" + + turn: int + text: str + calls: int + + +@dataclass(frozen=True, slots=True) +class Called: + """One tool call of a turn, and what the tool handed back.""" + + turn: int + step: Step + result: ToolResult + + +type Progress = Turn | Called +"""What an investigation reports while it runs.""" + + def investigate( # noqa: PLR0913 - the loop's inputs, and what the record cites. alert: str, tools: Sequence[AgentTool], @@ -336,6 +358,30 @@ def investigate( # noqa: PLR0913 - the loop's inputs, and what the record cites When the endpoint fails or answers malformed once the run started, this raises ``ModelUnavailableError`` carrying the investigation so far. """ + events = investigation_events( + alert, tools, endpoint, transport, max_turns=max_turns, replay=replay + ) + while True: + try: + next(events) + except StopIteration as stop: + return cast("Investigation", stop.value) + + +def investigation_events( # noqa: PLR0913 - as investigate. + alert: str, + tools: Sequence[AgentTool], + endpoint: ModelEndpoint, + transport: ChatTransport, + *, + max_turns: int = DEFAULT_MAX_TURNS, + replay: str | None = None, +) -> Generator[Progress, None, Investigation]: + """Run the investigation, yielding each turn and call as it happens. + + The generator's return value is the investigation; a failing endpoint + raises ``ModelUnavailableError`` as ``investigate`` does. + """ started = _utc_now() by_name = {tool.name: tool for tool in tools} messages: list[Mapping[str, object]] = [ @@ -383,14 +429,19 @@ def record(ending: InvestigationStatus, finding: Finding | None) -> Investigatio usage = usage.plus(reply) message = _first_message(reply) messages.append(message) - calls = message.get("tool_calls") + calls = cast( + "Sequence[Mapping[str, object]]", message.get("tool_calls") or () + ) + text = str(message.get("content") or "").strip() + yield Turn(turn=turns, text=text, calls=len(calls)) if not calls: - answer = str(message.get("content") or "").strip() + answer = text status = InvestigationStatus.UNSTRUCTURED break - for call in cast("Sequence[Mapping[str, object]]", calls): + for call in calls: step, result = _run_call(call, by_name, seen) steps.append(step) + yield Called(turn=turns, step=step, result=result) messages.append( { "role": "tool", diff --git a/sdk/tests/test_investigation.py b/sdk/tests/test_investigation.py index b638e99..1f48fbc 100644 --- a/sdk/tests/test_investigation.py +++ b/sdk/tests/test_investigation.py @@ -325,6 +325,25 @@ def test_an_endpoint_must_be_plain_http(url: str) -> None: assert error.value.code == "agent_endpoint_invalid" +def _streamed(out: str) -> dict[str, Any]: + """Read a stream: its events, the saved record's event, and how it ended. + + Each event is a bare line with ``_seq``; the stream ends with the + ``_summary`` line, or with the error envelope when it failed. + """ + lines = [json.loads(line) for line in out.splitlines() if line.strip()] + events = [line for line in lines if "_seq" in line] + record = next( + (event for event in events if event["kind"] == "investigation"), None + ) + return { + "data": record, + "error": lines[-1].get("error"), + "events": events, + "end": lines[-1], + } + + class _Model(BaseHTTPRequestHandler): """An OpenAI-compatible endpoint answering from a class-level script. @@ -398,7 +417,7 @@ def test_the_cli_investigates_and_records_it( ] ) - payload = json.loads(capsys.readouterr().out) + payload = _streamed(capsys.readouterr().out) assert exit_code == 0, payload spec = payload["data"]["investigation"]["spec"] assert spec["status"] == "answered" @@ -463,7 +482,7 @@ def test_an_answer_that_is_not_json_exits_model_unavailable( server.shutdown() server.server_close() - payload = json.loads(capsys.readouterr().out) + payload = _streamed(capsys.readouterr().out) assert exit_code == 92, payload assert payload["error"]["context"]["code"] == "agent_model_answer_malformed" @@ -506,7 +525,7 @@ def test_a_model_failing_mid_run_records_the_investigation_and_exits_92( out = capsys.readouterr() assert "sk-test" not in out.out + out.err - payload = json.loads(out.out) + payload = _streamed(out.out) assert exit_code == 92, payload context = payload["error"]["context"] assert context["code"] == code @@ -543,7 +562,7 @@ def test_an_endpoint_with_a_query_or_fragment_is_refused_before_the_run( exit_code = _investigate_against(repository, key, url) out = capsys.readouterr() - payload = json.loads(out.out) + payload = _streamed(out.out) assert exit_code == 2, payload assert "sk-leak" not in out.out + out.err assert not (repository / "investigations").exists() @@ -563,7 +582,7 @@ def test_a_made_up_long_tool_name_is_still_recorded( exit_code = _investigate_against(repository, key, model_url) - payload = json.loads(capsys.readouterr().out) + payload = _streamed(capsys.readouterr().out) assert exit_code == 0, payload step = payload["data"]["investigation"]["spec"]["steps"][0] assert step["outcome"] == "unknown-tool" @@ -598,7 +617,7 @@ def test_what_the_record_would_refuse_is_refused_before_the_run( ] ) - payload = json.loads(capsys.readouterr().out) + payload = _streamed(capsys.readouterr().out) assert exit_code == 2, payload assert not (repository / "investigations").exists() @@ -616,7 +635,7 @@ def test_an_unreachable_endpoint_exits_model_unavailable( exit_code = _investigate_against(repository, key, f"http://127.0.0.1:{port}/v1") - payload = json.loads(capsys.readouterr().out) + payload = _streamed(capsys.readouterr().out) assert exit_code == 92, payload context = payload["error"]["context"] assert context["code"] == "agent_model_unreachable" @@ -707,7 +726,7 @@ def test_a_replay_answers_from_the_recording_and_runs_no_playbook( ] ) - payload = json.loads(capsys.readouterr().out) + payload = _streamed(capsys.readouterr().out) assert exit_code == 0, payload spec = payload["data"]["investigation"]["spec"] assert spec["replay"] == "recorded" @@ -746,7 +765,66 @@ def test_a_replay_of_no_recorded_run_is_refused_before_the_model( ] ) - payload = json.loads(capsys.readouterr().out) + payload = _streamed(capsys.readouterr().out) assert exit_code != 0, payload assert "recording_empty" in json.dumps(payload) assert not (repository / "investigations").exists() + + +def test_the_cli_streams_each_turn_and_call_then_the_record( + tmp_path: Path, capsys: pytest.CaptureFixture[str], model_url: str +) -> None: + repository = _repository(tmp_path) + (repository / "playbooks" / "facts.yml").write_text( + "---\n- hosts: localhost\n connection: local\n gather_facts: false\n" + " tasks:\n - name: Report\n ansible.builtin.debug:\n" + " msg: /var/log/shop holds 36G\n", + encoding="utf-8", + ) + key = tmp_path / "model.key" + key.write_text("sk-test\n", encoding="utf-8") + _Model.replies[:] = [ + _call("operation_facts", {}), + _say("ROOT CAUSE: /var/log/shop\nPROPOSED: none\nWHY: read only."), + ] + + assert _investigate_against(repository, key, model_url) == 0 + + stream = _streamed(capsys.readouterr().out) + events = stream["events"] + assert [event["_seq"] for event in events] == [1, 2, 3, 4] + assert [event["kind"] for event in events] == [ + "turn", + "call", + "turn", + "investigation", + ] + assert events[1]["effect"] == "created" + assert "/var/log/shop holds 36G" in events[1]["result"]["output"] + assert events[2]["text"].startswith("ROOT CAUSE") + # The decision the call wrote, and the saved record. + assert stream["end"]["_summary"] is True + assert stream["end"]["effects"] == {"created": 2, "noop": 2} + + +def test_an_incomplete_investigation_ends_with_the_record_then_93( + tmp_path: Path, capsys: pytest.CaptureFixture[str], model_url: str +) -> None: + repository = _repository(tmp_path) + key = tmp_path / "model.key" + key.write_text("sk-test\n", encoding="utf-8") + _Model.replies[:] = [_say("I am not sure what happened.")] + + exit_code = _investigate_against(repository, key, model_url) + + payload = _streamed(capsys.readouterr().out) + assert exit_code == 93, payload + assert payload["data"]["status"] == "unstructured" + assert payload["events"][-1]["kind"] == "investigation" + assert payload["end"]["error"] is not None + assert len(list((repository / "investigations").glob("*.json"))) == 1 + error = payload["error"] + assert error["code"] == "INVESTIGATION_INCOMPLETE" + # The error line holds no data: the record is the event before it. + assert "data.investigation" not in error["suggestion"] + assert "kind investigation" in error["suggestion"]