From 210e0b6ded549f699657c792e825e689646ba2de Mon Sep 17 00:00:00 2001 From: Romamo Date: Sun, 11 Oct 2026 21:18:58 +0300 Subject: [PATCH 1/2] Tell which agent turn asked for a decision in cloudfall why An investigation step now keeps the turn that made its call, and why reads the investigation records beside the decisions: a decision an agent asked for is told with the model, the endpoint, the alert, the turn, and that turn's reasoning with the endpoint's response and request ids. The record could already say what the model thought, but only by reading two files by hand. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01APTdZZw8Jw9PQF4TD55wMv --- CHANGELOG.md | 14 ++ .../v1/agent-investigation.schema.json | 5 + docs/operator-guide.md | 4 +- sdk/src/cloudfall/agent_server.py | 3 + sdk/src/cloudfall/app.py | 24 ++- sdk/src/cloudfall/investigation.py | 5 + sdk/src/cloudfall/why.py | 184 +++++++++++++++++- sdk/tests/test_investigation.py | 1 + sdk/tests/test_why.py | 148 +++++++++++++- 9 files changed, 374 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 48c519a..fd076d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ Notable changes to Cloudfall. The format follows ## [Unreleased] +### Added + +- `cloudfall why` tells which agent turn asked for a decision. Each step of + an investigation record now keeps the `turn` that made its call (optional + in `agent-investigation.schema.json`; older records have none). `why` + reads the investigation records (`--investigations`, default + `investigations`, served to MCP clients from the same place) and adds to a + decision's story the model, the endpoint, the alert, the turn, and that + turn's reasoning with the endpoint's response and request ids; the answer + carries the same as `agent`. A decision a later call only repeated keeps + the turn that asked first. An investigation record that is not valid JSON + or does not match its schema stops `why` with `RECORD_INVALID` + (`why_investigation_invalid`) + ## [0.6.0] - 2026-10-10 ### Added diff --git a/config/schemas/v1/agent-investigation.schema.json b/config/schemas/v1/agent-investigation.schema.json index 0cc2003..567015f 100644 --- a/config/schemas/v1/agent-investigation.schema.json +++ b/config/schemas/v1/agent-investigation.schema.json @@ -106,6 +106,11 @@ }, "decision": { "$ref": "common.schema.json#/$defs/resourceId" + }, + "turn": { + "type": "integer", + "minimum": 1, + "description": "The turn of the trace that made this call. Optional: records written before it was kept have none." } }, "required": [ diff --git a/docs/operator-guide.md b/docs/operator-guide.md index 0223288..4eefa92 100644 --- a/docs/operator-guide.md +++ b/docs/operator-guide.md @@ -80,7 +80,9 @@ it. Autonomy in the next section is computed from it. The [brownfield design](brownfield-design.md) generalises this entry to every declared operation as a decision record with the fleet snapshot and the check-mode diff, and `cloudfall why` queries those records by host, -operation or time window. The operator's receipts here are the older +operation or time window; for a decision an agent asked for through +`agent investigate`, it also tells the turn and the model's reasoning +there. The operator's receipts here are the older form of the same entry and are not yet read by `cloudfall why`. To run persistently, wrap the command in a systemd service on the diff --git a/sdk/src/cloudfall/agent_server.py b/sdk/src/cloudfall/agent_server.py index 70d3abf..712f53f 100644 --- a/sdk/src/cloudfall/agent_server.py +++ b/sdk/src/cloudfall/agent_server.py @@ -33,6 +33,7 @@ ) from cloudfall.catalog import CATALOG_DIRECTORY, RiskLevel from cloudfall.decision import DECISION_DIRECTORY +from cloudfall.investigation import INVESTIGATION_DIRECTORY from cloudfall.project import ( PROJECT_DIRECTORY_VARIABLE, ProjectError, @@ -355,6 +356,8 @@ def bound_arguments(args: McpServeArgs) -> dict[str, object]: "env_receipts": str(root.path(args.env_receipts)), "operations": str(args.operations), "decisions": str(args.decisions), + # why reads the agent's turns where agent investigate writes them. + "investigations": INVESTIGATION_DIRECTORY, } return { **shared, diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index a6df8d8..7d213bf 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -256,7 +256,13 @@ ValidatedConfig, validate_config, ) -from cloudfall.why import WhyError, WhyQuery, answer, render_why_document +from cloudfall.why import ( + Investigations, + WhyError, + WhyQuery, + answer, + render_why_document, +) if TYPE_CHECKING: from collections.abc import Callable, Sequence @@ -3166,10 +3172,18 @@ class WhyArgs(DecisionsArgs): default=(), description="Only decisions with this status (repeatable)", ) + investigations: Path = Flag( + default=Path(INVESTIGATION_DIRECTORY), + description=( + "Directory holding the investigation records, to tell which agent " + f"turn asked for each decision (default: {INVESTIGATION_DIRECTORY})" + ), + ) def __post_init__(self) -> None: """Refuse a time the question cannot be asked with.""" DecisionsArgs.__post_init__(self) + _inside_project(self.investigations, "investigations") try: self.query() except WhyError as error: @@ -3212,11 +3226,11 @@ class WhyPayload(Payload): ) def why(args: WhyArgs, _ctx: Ctx) -> WhyPayload: """Answer from the record alone: no catalog, no fleet.""" - store = DecisionStore( - directory=args.root / args.decisions, catalog=SchemaCatalog(args.schemas) - ) + catalog = SchemaCatalog(args.schemas) + store = DecisionStore(directory=args.root / args.decisions, catalog=catalog) try: - result = answer(store, args.query()) + investigations = Investigations.load(args.root / args.investigations, catalog) + result = answer(store, args.query(), investigations) except (DecisionError, WhyError) as error: raise Exit.RECORD_INVALID(error.detail, context={"code": error.code}) from error return WhyPayload(result.as_dict()) diff --git a/sdk/src/cloudfall/investigation.py b/sdk/src/cloudfall/investigation.py index f521dfc..f33af18 100644 --- a/sdk/src/cloudfall/investigation.py +++ b/sdk/src/cloudfall/investigation.py @@ -205,6 +205,8 @@ class Step: arguments: Mapping[str, object] outcome: StepOutcome decision: str | None + turn: int | None = None + """The turn of the trace that made the call; the record links the two by it.""" def as_dict(self) -> dict[str, object]: """Serialize the step for the record.""" @@ -215,6 +217,8 @@ def as_dict(self) -> dict[str, object]: } if self.decision is not None: document["decision"] = self.decision + if self.turn is not None: + document["turn"] = self.turn return document @@ -542,6 +546,7 @@ def record(ending: InvestigationStatus, finding: Finding | None) -> Investigatio break for call in calls: step, result = _run_call(call, by_name, seen) + step = replace(step, turn=turns) steps.append(step) yield Called(turn=turns, step=step, result=result) messages.append( diff --git a/sdk/src/cloudfall/why.py b/sdk/src/cloudfall/why.py index fbd5af1..e345ce2 100644 --- a/sdk/src/cloudfall/why.py +++ b/sdk/src/cloudfall/why.py @@ -11,26 +11,37 @@ names that host, as its target or in the per-host evidence of its check, run or verify stage. A group pattern is not expanded, because expanding it would need an inventory the record deliberately does not depend on. + +A decision an agent asked for is told with the turn that asked for it: +the investigation records each tool call's decision and turn, and the +turn's trace keeps the model's reasoning and the endpoint's ids. """ from __future__ import annotations +import json from dataclasses import dataclass from datetime import UTC, datetime from html import escape from typing import TYPE_CHECKING, cast +from jsonschema import ValidationError + from cloudfall.decision import ApprovalChannel, DecisionStatus, Requirement from cloudfall.domain import Hostname, ResourceId +from cloudfall.investigation import INVESTIGATION_SCHEMA, StepOutcome if TYPE_CHECKING: from collections.abc import Iterable, Mapping, Sequence + from pathlib import Path from cloudfall.decision import Approval, Decision, DecisionStore, RunRecord + from cloudfall.validation import SchemaCatalog _ERROR_INSTANT_INVALID = "why_instant_invalid" _ERROR_WINDOW_INVERTED = "why_window_inverted" _ERROR_STATUS_UNKNOWN = "why_status_unknown" +_ERROR_INVESTIGATION_INVALID = "why_investigation_invalid" class WhyError(RuntimeError): @@ -165,15 +176,109 @@ class Explanation: decision: Decision hosts: tuple[str, ...] story: tuple[str, ...] + proposer: Proposer | None = None def as_dict(self) -> dict[str, object]: """Serialize the explanation with the record it is drawn from.""" - return { + document: dict[str, object] = { "id": self.decision.decision_id.value, "hosts": list(self.hosts), "story": list(self.story), "decision": self.decision.as_document(), } + if self.proposer is not None: + document["agent"] = self.proposer.as_dict() + return document + + +@dataclass(frozen=True, slots=True) +class Proposer: + """The agent turn that asked for a decision, as its investigation recorded it.""" + + investigation: str + model: str + endpoint: str + alert: str + turn: int | None + """None in a record written before a step kept its turn.""" + reasoning: str | None + response_id: str | None + request_id: str | None + + def as_dict(self) -> dict[str, object]: + """Serialize the turn for the answer, as the investigation has it.""" + document: dict[str, object] = { + "investigation": self.investigation, + "model": self.model, + "endpoint": self.endpoint, + "alert": self.alert, + } + optional = { + "turn": self.turn, + "reasoning": self.reasoning, + "responseId": self.response_id, + "requestId": self.request_id, + } + document.update({key: value for key, value in optional.items() if value}) + return document + + def sentences(self) -> tuple[str, ...]: + """Tell who asked for the decision, and what it was thinking then.""" + where = f"in turn {self.turn} of" if self.turn is not None else "in" + # An alert that ends its own sentence is not given a second stop. + stop = "" if self.alert.rstrip().endswith((".", "!", "?")) else "." + asked = ( + f"The agent {self.model} at {self.endpoint} asked for it {where} " + f"investigation {self.investigation}, working the alert " + f'"{self.alert.strip()}"{stop}' + ) + if self.turn is None: + return ( + asked, + "That investigation predates recording which turn made each " + "call, so its reasoning is in its trace but not tied to this one.", + ) + if self.reasoning is None: + return (asked, "The endpoint returned no reasoning for that turn.") + served = ", ".join( + f"{label} {value}" + for label, value in ( + ("response", self.response_id), + ("request", self.request_id), + ) + if value is not None + ) + return ( + asked, + "Its reasoning in that turn, as the endpoint returned it: " + f'"{self.reasoning}"' + (f" ({served})." if served else "."), + ) + + +@dataclass(frozen=True, slots=True) +class Investigations: + """Which agent turn asked for each decision, from the investigation records.""" + + proposers: Mapping[str, Proposer] + + @classmethod + def load(cls, directory: Path, catalog: SchemaCatalog) -> Investigations: + """Read every investigation record in the directory; none is no link. + + A record that is not valid JSON or does not match its schema is + refused: an answer drawn from it could cite a turn that never was. + A decision a later call only repeated keeps the turn that asked first. + """ + proposers: dict[str, Proposer] = {} + if not directory.is_dir(): + return cls(proposers) + for path in sorted(directory.glob("*.json")): + document = _investigation(path, catalog) + metadata = cast("Mapping[str, str]", document["metadata"]) + spec = cast("Mapping[str, object]", document["spec"]) + for decision, proposer in _proposers(metadata["id"], spec): + proposers.setdefault(decision, proposer) + return cls(proposers) @dataclass(frozen=True, slots=True) @@ -195,23 +300,32 @@ def as_dict(self) -> dict[str, object]: } -def answer(store: DecisionStore, query: WhyQuery) -> WhyAnswer: +def answer( + store: DecisionStore, + query: WhyQuery, + investigations: Investigations | None = None, +) -> WhyAnswer: """Answer the question from the record, newest decision first.""" matched = [decision for decision in store.list() if query.matches(decision)] matched.sort(key=lambda decision: decision.proposed_at, reverse=True) + proposers = investigations.proposers if investigations is not None else {} return WhyAnswer( query=query, directory=str(store.directory), - explanations=tuple(explain(decision) for decision in matched), + explanations=tuple( + explain(decision, proposers.get(decision.decision_id.value)) + for decision in matched + ), ) -def explain(decision: Decision) -> Explanation: - """Tell one decision from its record alone.""" +def explain(decision: Decision, proposer: Proposer | None = None) -> Explanation: + """Tell one decision from its record, and the agent turn that asked for it.""" return Explanation( decision=decision, hosts=tuple(sorted(hosts_of(decision))), - story=tuple(_story(decision)), + story=tuple(_story(decision, proposer)), + proposer=proposer, ) @@ -367,10 +481,12 @@ def _status(value: object) -> DecisionStatus: raise WhyError(_ERROR_STATUS_UNKNOWN, message) from error -def _story(decision: Decision) -> Iterable[str]: +def _story(decision: Decision, proposer: Proposer | None) -> Iterable[str]: """Yield the record as sentences: saw, proposed, checked, approved, ended.""" yield _saw(decision) yield _proposed(decision) + if proposer is not None: + yield from proposer.sentences() yield _checked(decision) if decision.replay is not None: yield ( @@ -487,3 +603,57 @@ def _ended(decision: Decision) -> str: def _present(value: object) -> bool: return value is not None and not (isinstance(value, str) and not value.strip()) + + +def _investigation(path: Path, catalog: SchemaCatalog) -> Mapping[str, object]: + try: + document = cast( + "Mapping[str, object]", json.loads(path.read_text(encoding="utf-8")) + ) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + message = f"investigation record is not valid JSON: {path}: {error}" + raise WhyError(_ERROR_INVESTIGATION_INVALID, message) from error + try: + catalog.validate_named(INVESTIGATION_SCHEMA, document) + except ValidationError as error: + field = ".".join(str(part) for part in error.absolute_path) or "(root)" + message = ( + f"investigation record does not match its schema: {path}: " + f"{field}: {error.message}" + ) + raise WhyError(_ERROR_INVESTIGATION_INVALID, message) from error + return document + + +def _proposers( + identifier: str, spec: Mapping[str, object] +) -> Iterable[tuple[str, Proposer]]: + """Yield each decision the investigation's own calls recorded, with its turn.""" + model = cast("Mapping[str, str]", spec["model"]) + trace = { + int(cast("int", turn["turn"])): turn + for turn in cast("Sequence[Mapping[str, object]]", spec.get("trace", ())) + } + for step in cast("Sequence[Mapping[str, object]]", spec["steps"]): + decision = step.get("decision") + if decision is None or step["outcome"] == StepOutcome.REPEATED.value: + continue + number = cast("int | None", step.get("turn")) + turn = trace.get(number, {}) if number is not None else {} + yield ( + str(decision), + Proposer( + investigation=identifier, + model=model["name"], + endpoint=model["endpoint"], + alert=str(spec["alert"]), + turn=number, + reasoning=_text(turn.get("reasoning")), + response_id=_text(turn.get("responseId")), + request_id=_text(turn.get("requestId")), + ), + ) + + +def _text(value: object) -> str | None: + return value if isinstance(value, str) and value.strip() else None diff --git a/sdk/tests/test_investigation.py b/sdk/tests/test_investigation.py index d62061a..0a11cd5 100644 --- a/sdk/tests/test_investigation.py +++ b/sdk/tests/test_investigation.py @@ -150,6 +150,7 @@ def test_the_model_reads_proposes_and_answers() -> None: StepOutcome.RAN, StepOutcome.PROPOSED, ] + assert [step.as_dict()["turn"] for step in done.steps] == [1, 2] assert done.turns == 3 assert (done.usage.prompt, done.usage.completion) == (250, 25) tool_reply = script.requests[1]["messages"] diff --git a/sdk/tests/test_why.py b/sdk/tests/test_why.py index da43bc7..1eb2381 100644 --- a/sdk/tests/test_why.py +++ b/sdk/tests/test_why.py @@ -17,7 +17,29 @@ approve, propose, ) -from cloudfall.why import Instant, WhyError, WhyQuery, answer, render_why_html +from cloudfall.domain import ResourceId +from cloudfall.investigation import ( + INVESTIGATION_DIRECTORY, + Investigation, + InvestigationStatus, + InvestigationStore, + ModelEndpoint, + Step, + StepOutcome, + ToolCallText, + Turn, + Usage, +) +from cloudfall.resources import default_schema_directory +from cloudfall.validation import SchemaCatalog +from cloudfall.why import ( + Instant, + Investigations, + WhyError, + WhyQuery, + answer, + render_why_html, +) from test_catalog import _repository from test_decision import MOMENT, _proposal, _Runner, _store @@ -347,3 +369,127 @@ def test_an_empty_record_renders_an_empty_page(tmp_path: Path) -> None: assert "The record holds no decision this question is about." in page assert "0 decisions" in page + + +def _investigated(repository: Path, decision: str, *, turn: int | None = 2) -> None: + """Record an investigation whose second turn asked for the decision.""" + asked = Turn( + turn=2, + text="", + reasoning="Disk is full of shop logs; rotate them first.", + tool_calls=(ToolCallText(name="operation_deploy", arguments="{}"),), + response_id="chatcmpl-2", + request_id="req-2", + elapsed_ms=900, + prompt_tokens=100, + completion_tokens=10, + reasoning_tokens=8, + ) + InvestigationStore( + directory=repository / INVESTIGATION_DIRECTORY, + catalog=SchemaCatalog(default_schema_directory()), + ).save( + Investigation( + investigation_id=ResourceId.from_boundary("investigation-20260921143000"), + alert="ALERT web-1 down", + endpoint=ModelEndpoint( + base_url="https://models.example/v1/", model="nemotron" + ), + started_at="2026-09-21T14:30:00Z", + finished_at="2026-09-21T14:30:10Z", + status=InvestigationStatus.ANSWERED, + turns=2, + usage=Usage(prompt=100, completion=10), + steps=( + Step("operation_deploy", {}, StepOutcome.PROPOSED, decision, turn), + Step("operation_deploy", {}, StepOutcome.REPEATED, decision, turn), + ), + answer="", + finding=None, + trace=(asked,), + ) + ) + + +def _investigations(repository: Path) -> Investigations: + return Investigations.load( + repository / INVESTIGATION_DIRECTORY, + SchemaCatalog(default_schema_directory()), + ) + + +def test_a_decision_an_agent_asked_for_is_told_with_its_turn(tmp_path: Path) -> None: + repository = _repository(tmp_path) + deploy, facts = _record(repository) + _investigated(repository, deploy.decision_id.value) + + told = answer(_store(repository), WhyQuery(), _investigations(repository)) + + by_id = {entry.decision.decision_id: entry for entry in told.explanations} + story = by_id[deploy.decision_id].story + assert story[2] == ( + "The agent nemotron at https://models.example/v1/ asked for it in turn 2 " + 'of investigation investigation-20260921143000, working the alert "ALERT ' + 'web-1 down".' + ) + assert story[3] == ( + "Its reasoning in that turn, as the endpoint returned it: " + '"Disk is full of shop logs; rotate them first." (response chatcmpl-2, ' + "request req-2)." + ) + assert by_id[deploy.decision_id].as_dict()["agent"] == { + "investigation": "investigation-20260921143000", + "model": "nemotron", + "endpoint": "https://models.example/v1/", + "alert": "ALERT web-1 down", + "turn": 2, + "reasoning": "Disk is full of shop logs; rotate them first.", + "responseId": "chatcmpl-2", + "requestId": "req-2", + } + assert "agent" not in by_id[facts.decision_id].as_dict() + assert all("agent" not in line for line in by_id[facts.decision_id].story) + + +def test_an_investigation_from_before_turns_were_kept_says_so(tmp_path: Path) -> None: + repository = _repository(tmp_path) + deploy, _ = _record(repository) + _investigated(repository, deploy.decision_id.value, turn=None) + + told = answer( + _store(repository), + WhyQuery.from_boundary(operation="deploy"), + _investigations(repository), + ) + + story = told.explanations[0].story + assert story[2].endswith('working the alert "ALERT web-1 down".') + assert " in investigation investigation-20260921143000" in story[2] + assert story[3] == ( + "That investigation predates recording which turn made each call, so " + "its reasoning is in its trace but not tied to this one." + ) + + +def test_the_cli_tells_the_agent_turn_and_refuses_a_broken_investigation( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + repository = _repository(tmp_path) + deploy, _ = _record(repository) + _investigated(repository, deploy.decision_id.value) + + exit_code = main(["why", "--repository", str(repository), "--host", "web-1"]) + + payload = json.loads(capsys.readouterr().out) + assert exit_code == 0 + assert payload["data"]["answers"][0]["agent"]["turn"] == 2 + + (repository / INVESTIGATION_DIRECTORY / "broken.json").write_text( + "{", encoding="utf-8" + ) + exit_code = main(["why", "--repository", str(repository)]) + + payload = json.loads(capsys.readouterr().out) + assert payload["error"]["code"] == "RECORD_INVALID" + assert payload["error"]["context"]["code"] == "why_investigation_invalid" + assert exit_code != 0 From bf9b89b362bed7fb6e6b764c507c21b182ad8c69 Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Sun, 11 Oct 2026 21:35:36 +0300 Subject: [PATCH 2/2] Move the why agent-turn entry into a changelog fragment Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 14 -------------- changelog.d/78-why-agent-turn.md | 13 +++++++++++++ 2 files changed, 13 insertions(+), 14 deletions(-) create mode 100644 changelog.d/78-why-agent-turn.md diff --git a/CHANGELOG.md b/CHANGELOG.md index fd076d2..48c519a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,20 +6,6 @@ Notable changes to Cloudfall. The format follows ## [Unreleased] -### Added - -- `cloudfall why` tells which agent turn asked for a decision. Each step of - an investigation record now keeps the `turn` that made its call (optional - in `agent-investigation.schema.json`; older records have none). `why` - reads the investigation records (`--investigations`, default - `investigations`, served to MCP clients from the same place) and adds to a - decision's story the model, the endpoint, the alert, the turn, and that - turn's reasoning with the endpoint's response and request ids; the answer - carries the same as `agent`. A decision a later call only repeated keeps - the turn that asked first. An investigation record that is not valid JSON - or does not match its schema stops `why` with `RECORD_INVALID` - (`why_investigation_invalid`) - ## [0.6.0] - 2026-10-10 ### Added diff --git a/changelog.d/78-why-agent-turn.md b/changelog.d/78-why-agent-turn.md new file mode 100644 index 0000000..f470a4f --- /dev/null +++ b/changelog.d/78-why-agent-turn.md @@ -0,0 +1,13 @@ +### Added + +- `cloudfall why` tells which agent turn asked for a decision. Each step of + an investigation record now keeps the `turn` that made its call (optional + in `agent-investigation.schema.json`; older records have none). `why` + reads the investigation records (`--investigations`, default + `investigations`, served to MCP clients from the same place) and adds to a + decision's story the model, the endpoint, the alert, the turn, and that + turn's reasoning with the endpoint's response and request ids; the answer + carries the same as `agent`. A decision a later call only repeated keeps + the turn that asked first. An investigation record that is not valid JSON + or does not match its schema stops `why` with `RECORD_INVALID` + (`why_investigation_invalid`) (#78)