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