Stream agent investigate turn by turn

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

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,34 +2279,44 @@ 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 = (
             None if args.replay is None else Recording.load(args.root / args.replay)
         )
     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(

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",