diff --git a/agent_core/components/observers/trajectory.py b/agent_core/components/observers/trajectory.py index 8cd3981..129c1be 100644 --- a/agent_core/components/observers/trajectory.py +++ b/agent_core/components/observers/trajectory.py @@ -8,10 +8,10 @@ from __future__ import annotations import contextlib +import hashlib import json import os import time -from collections import deque from collections.abc import Callable, Iterable, Mapping from pathlib import Path from typing import IO, Any, Literal @@ -20,12 +20,14 @@ AgentLoopResult, BaseObserver, CompactionEvent, + ContextCompactionContext, Intervention, LoopConfig, ToolResult, TurnContext, ) from agent_core.messages import Message +from agent_core.runtime.loop.compact import COMPACTION_SEQ_KEY from agent_core.tool_content import redacted_for_trace, redacted_tool_result_content _FORMATS: tuple[str, ...] = ("json", "jsonl") @@ -34,8 +36,6 @@ # information and appear on almost every turn, so trajectories omit them. _ORDINARY_FINISH_REASONS = frozenset({"end_turn", "stop"}) _DIAGNOSTIC_KEYS = ("usage", "thinking_blocks", "finish_reason", "stop_details") -# Blocks a signature-recovery reset strips from assistant content. -_THINKING_BLOCK_TYPES = frozenset({"thinking", "redacted_thinking"}) def _json_default(value: Any) -> Any: @@ -65,6 +65,32 @@ def _json_default(value: Any) -> Any: _STREAM_ENCODER = json.JSONEncoder( ensure_ascii=False, separators=(",", ":"), default=_json_default, ) + + +def _line_digest(line: bytes) -> bytes: + """Digest of one rendered (redacted, clipped) spool line.""" + return hashlib.blake2b(line, digest_size=16).digest() + + +def _lineage(message: Any) -> list[Any]: + """Objects a shallow copy of *message* still shares with it. + + Compaction and thinking resets keep mutable content/tool-call containers + or their blocks. Scalar identities are not evidence: even separately + decoded nonempty strings can be shared through interpreter caches. + """ + if not isinstance(message, dict): + return [] + shared: list[Any] = [] + for key in ("content", "tool_calls"): + value = message.get(key) + if isinstance(value, (dict, list)) and value: + shared.append(value) + if isinstance(value, list): + shared.extend(item for item in value if isinstance(item, (dict, list)) and item) + return shared + + _DEFAULT_FORMAT_ENV_VARS = ( "AGENT_CORE_TRAJECTORY_FORMATS", "FRONTIER_AGENT_TRAJECTORY_FORMATS", @@ -147,8 +173,8 @@ class TrajectoryFileObserver(BaseObserver): # task never moved that work off the event loop; it only reordered it. # Awaiting them preserves event order even when a synthetic result needs # no async tool execution, where a background on_llm_response could run - # AFTER it. Coalescing and re-rendering only new or changed messages keep - # the awaited cost per flush small. + # AFTER it. Coalescing limits snapshot work; streamed value digests let + # unchanged spool prefixes survive while still detecting nested edits. critical: bool = True def __init__( @@ -221,16 +247,24 @@ def __init__( # True when no loop history was offered (a host driving the hooks # directly), so the observer assembles the conversation itself. self._owns_history: bool = False - # Diagnostic fields are separate from history, kept as [message, - # content fingerprint, details] in response order. Identity binds them - # first; the fingerprint re-binds them to the shallow copies that - # compaction and thinking resets substitute. Entries whose message left - # history are released at the next snapshot. - self._response_metadata: list[list[Any]] = [] - # One (message, signature, end offset) per history entry already in - # the spool. A snapshot keeps the longest unchanged prefix and - # re-renders only from the first new, replaced or edited message. - self._spool_entries: list[tuple[Any, tuple[Any, ...], int]] = [] + # Diagnostic fields are separate from history: id(message) -> + # [message, details], one entry per assistant in history. Identity + # binds them first; a replacement inherits an entry only when it is a + # shallow copy of exactly one vanished assistant (see ``_lineage``). + # Entries without diagnostics still compete, so repeated text cannot + # inherit another turn's details. Entries whose message left history + # are released at the next snapshot. + self._response_metadata: dict[int, list[Any]] = {} + # One (message, reference signature, line digest, end offset) per + # history entry already in the spool. A snapshot keeps the longest + # unchanged prefix and re-renders from the first difference. + self._spool_entries: list[tuple[Any, tuple[Any, ...], bytes, int]] = [] + # Reference signatures catch replaced messages and values. Edits nested + # inside a kept container need the rendered content compared, which + # costs a full re-render, so it runs only after a compaction or + # rollback notice and on forced (terminal) snapshots. + self._deep_check_due: bool = False + self._compaction_seq: Any = None # Set by on_loop_end. A cancellation arriving after it (while a later # observer still runs) must not overwrite the terminal snapshot. self._ended: bool = False @@ -345,8 +379,8 @@ def _signature(message: Any, details: Any) -> tuple[Any, ...]: History edits replace values (``message["content"] = rebuilt``) or whole messages, so identity of each value and content block detects - them without re-serialising. Holding the references keeps their ids - from being reused before the next comparison. + them without serialising anything. Holding the references keeps their + ids from being reused before the next comparison. """ if not isinstance(message, dict): return (message, details) @@ -362,59 +396,53 @@ def _same(a: Any, b: Any) -> bool: ) return a is b - @staticmethod - def _fingerprint(message: Message) -> int: - """Content identity that survives reasoning and thinking-block strips.""" - content = message.get("content") - if isinstance(content, list): - content = [ - block for block in content - if not isinstance(block, dict) - or block.get("type") not in _THINKING_BLOCK_TYPES - ] - ids = tuple( - tc.get("id") for tc in (message.get("tool_calls") or []) - if isinstance(tc, dict) - ) - return hash((ids, _STREAM_ENCODER.encode(content))) - def _resolve_annotations(self, history: list[Message]) -> dict[int, dict[str, Any]]: """Bind diagnostics to the history messages that carry them now.""" present = {id(message): message for message in history} bound: dict[int, list[Any]] = {} unmatched: list[list[Any]] = [] - for entry in self._response_metadata: - ref = entry[0] - if present.get(id(ref)) is ref and id(ref) not in bound: - bound[id(ref)] = entry + for key, entry in self._response_metadata.items(): + if present.get(key) is entry[0]: + bound[key] = entry else: unmatched.append(entry) if unmatched: - pending: dict[int, deque[list[Any]]] = {} + # Map each object a vanished assistant owned to that assistant. + # The entry still references the old message, so these ids + # cannot be reused while this lookup is alive. + owners: dict[int, dict[int, list[Any]]] = {} for entry in unmatched: - pending.setdefault(entry[1], deque()).append(entry) - for message in history: - if not pending: - break - if ( - not isinstance(message, dict) - or message.get("role") != "assistant" - or id(message) in bound - ): - continue - fingerprint = self._fingerprint(message) - queue = pending.get(fingerprint) - if queue: - entry = queue.popleft() - entry[0] = message - bound[id(message)] = entry - if not queue: - del pending[fingerprint] - self._response_metadata = [ - entry for entry in self._response_metadata - if bound.get(id(entry[0])) is entry - ] - return {key: entry[2] for key, entry in bound.items()} + for value in _lineage(entry[0]): + owners.setdefault(id(value), {})[id(entry)] = entry + if owners: + claims: dict[int, list[Message]] = {} + for message in history: + if ( + not isinstance(message, dict) + or message.get("role") != "assistant" + or id(message) in bound + ): + continue + sources: dict[int, list[Any]] = {} + for value in _lineage(message): + sources.update(owners.get(id(value), {})) + # A copy of several turns, or several copies of one turn, + # identifies nothing. Leave those diagnostics in JSONL. + if len(sources) == 1: + claims.setdefault(next(iter(sources)), []).append(message) + for entry in unmatched: + targets = claims.get(id(entry), []) + if len(targets) == 1: + entry[0] = targets[0] + bound[id(targets[0])] = entry + retained: dict[int, list[Any]] = {} + for message in history: + if not isinstance(message, dict) or message.get("role") != "assistant": + continue + entry = bound.get(id(message)) or [message, {}] + retained[id(message)] = entry + self._response_metadata = retained + return {key: entry[1] for key, entry in retained.items() if entry[1]} def _render(self, message: Any, details: dict[str, Any] | None) -> bytes: rendered = self._message_to_dict(message) @@ -430,9 +458,26 @@ def _render(self, message: Any, details: dict[str, Any] | None) -> bytes: # Already preserved verbatim as native content blocks. details = {k: v for k, v in details.items() if k != "thinking_blocks"} rendered.update(details) - return (_STREAM_ENCODER.encode(rendered) + "\n").encode("utf-8") - - def _sync_history(self) -> None: + try: + text = _STREAM_ENCODER.encode(rendered) + except (TypeError, ValueError, RecursionError) as exc: + # Non-string keys or a reference cycle in host metadata. Keep the + # message's place (and its tool pairing) instead of losing the + # whole snapshot. + fallback = { + key: rendered[key] for key in ("role", "tool_call_id", "tool_calls") + if key in rendered + } + fallback["content"] = str(rendered.get("content", "")) + fallback["trace_error"] = f"unserializable message: {exc}" + try: + text = _STREAM_ENCODER.encode(fallback) + except (TypeError, ValueError, RecursionError): + fallback.pop("tool_calls", None) + text = _STREAM_ENCODER.encode(fallback) + return (text + "\n").encode("utf-8") + + def _sync_history(self, *, deep: bool = False) -> None: """Bring the spool up to the real history, including in-place edits. A length/delta cursor is insufficient: compaction, rollback, thinking @@ -440,6 +485,10 @@ def _sync_history(self) -> None: The unchanged prefix stays on disk; from the first difference onwards the spool is truncated and re-rendered. Each message is rendered whole before it is written, so a failure leaves a consistent prefix. + + Prefix messages are compared by reference signature. *deep* also + re-renders them and compares the redacted line, which catches edits + nested inside a container the message kept. """ history = self._history if history is None: @@ -448,16 +497,24 @@ def _sync_history(self) -> None: entries = self._spool_entries path = self._message_spool_path() keep = 0 + first_line: bytes | None = None if path.exists(): limit = min(len(entries), len(history)) while keep < limit: message = history[keep] - ref, signature, _ = entries[keep] - current = self._signature(message, details_for.get(id(message))) - if ref is not message or not self._same(signature, current): + ref, signature, digest, _ = entries[keep] + details = details_for.get(id(message)) + if ref is not message or not self._same( + signature, self._signature(message, details), + ): break + if deep: + line = self._render(message, details) + if _line_digest(line) != digest: + first_line = line + break keep += 1 - offset = entries[keep - 1][2] if keep else 0 + offset = entries[keep - 1][3] if keep else 0 del entries[keep:] # Open even for an empty history so a rollback to [] clears the spool. with path.open("r+b" if keep else "wb") as fh: @@ -465,10 +522,13 @@ def _sync_history(self) -> None: fh.truncate() for message in history[keep:]: details = details_for.get(id(message)) - line = self._render(message, details) + line = first_line if first_line is not None else self._render(message, details) + first_line = None fh.write(line) offset += len(line) - entries.append((message, self._signature(message, details), offset)) + entries.append(( + message, self._signature(message, details), _line_digest(line), offset, + )) def _write_envelope(self, path: Path) -> None: """Stream one compatible JSON envelope without materialising it.""" @@ -553,7 +613,10 @@ def _flush_json(self, *, force: bool = False) -> None: return self._pending_flushes = 0 self._last_flush_at = time.monotonic() - self._sync_history() + deep = force or self._deep_check_due + self._sync_history(deep=deep) + if deep: + self._deep_check_due = False path = self._path("json") tmp = path.with_suffix(path.suffix + ".tmp") self._write_envelope(tmp) @@ -639,6 +702,8 @@ async def on_loop_start(self, config: LoopConfig) -> None: self._role_id = config.role_id self._max_turns = config.max_turns self._ended = False + self._deep_check_due = False + self._compaction_seq = None # After ``_task_id`` is set: it is the stem fallback when no explicit # ``filename`` was passed. self._publish_jsonl_path() @@ -665,22 +730,45 @@ async def on_before_llm(self, ctx: TurnContext) -> Intervention | None: # injections, compaction, early exits and cancellation. The loop # rewrites it in place, so binding once per run is enough. self._bind_history(ctx.messages) + self._note_compaction(ctx) return None async def on_llm_input(self, ctx: TurnContext) -> None: # Injections from on_before_llm have now been applied. Transient request # addenda belong to the request, not to durable conversation history. + self._note_compaction(ctx) self._flush_json() async def on_turn_end(self, ctx: TurnContext) -> Intervention | None: + self._note_compaction(ctx) self._flush_json() return None + async def on_context_compacted(self, ctx: ContextCompactionContext) -> None: + # Rollbacks and context-limit pops announce themselves here. + self._deep_check_due = True + + def _note_compaction(self, ctx: TurnContext) -> None: + """Schedule a deep check when a compactor ran since the last look. + + A compactor may edit blocks in place, leaving every reference intact, + and then the loop sends no ``on_context_compacted``. Its run counter in + loop metadata advances regardless. + """ + seq = (ctx.metadata or {}).get(COMPACTION_SEQ_KEY) + if seq != self._compaction_seq: + if self._compaction_seq is not None or seq: + self._deep_check_due = True + self._compaction_seq = seq + def _bind_history(self, messages: list[Message] | None) -> None: if "json" not in self._formats or self._owns_history or messages is None: return if self._history is None: self._history = messages + # Resumed histories can contain assistants without diagnostics; + # include them when deciding whether a future copy is ambiguous. + self._resolve_annotations(messages) @staticmethod def _assistant_from_ctx(ctx: TurnContext) -> Message: @@ -776,10 +864,11 @@ async def on_llm_response( self._history.append(assistant) if assistant is not None: details = {key: record[key] for key in _DIAGNOSTIC_KEYS if key in record} - if details: - self._response_metadata.append( - [assistant, self._fingerprint(assistant), details], - ) + entry = self._response_metadata.get(id(assistant)) + if entry is not None and entry[0] is assistant: + entry[1] = details + else: + self._response_metadata[id(assistant)] = [assistant, details] self._tool_calls_count += len(ctx.tool_calls or []) self._turns_used = ctx.turn self._flush_json() @@ -835,6 +924,7 @@ async def on_compaction(self, event: CompactionEvent) -> None: by the summariser's own output length, not by tool output, so it does not need the ``_BODY_MAX_CHARS`` guard that tool results do. """ + self._deep_check_due = True self._write_jsonl({ "t": "compaction", "turn": event.turn, @@ -881,7 +971,7 @@ async def on_loop_end(self, result: AgentLoopResult) -> None: self._close_message_spool(cleanup=snapshot_written) self._history = None self._owns_history = False - self._response_metadata = [] + self._response_metadata = {} async def on_loop_cancelled(self) -> None: """Flush current history and preserve crash-forensic sidecars.""" @@ -897,4 +987,4 @@ async def on_loop_cancelled(self) -> None: self._close_message_spool() self._history = None self._owns_history = False - self._response_metadata = [] + self._response_metadata = {} diff --git a/changes/trajectory-cache-consistency.fix.md b/changes/trajectory-cache-consistency.fix.md new file mode 100644 index 0000000..e4e6b81 --- /dev/null +++ b/changes/trajectory-cache-consistency.fix.md @@ -0,0 +1 @@ +Fix trajectory snapshots missing edits nested inside existing history containers. Routine snapshots still reuse the spool prefix by reference comparison; after a compactor run, a rollback or context-limit notice, and on completion or cancellation, the redacted rendered lines are compared as well, so inline image data is never hashed. Recover diagnostic annotations after history copies only when the copy shares mutable content/tool-call containers or blocks with exactly one vanished assistant, accounting for assistants without diagnostics and resumed history; cached strings cannot transfer another turn's usage or stop reason. Source deduplication uses object identity so cyclic metadata never triggers recursive message comparisons. Messages with unencodable host metadata (cycles, non-string keys) keep their place in the snapshot with a `trace_error` instead of failing it. Ambiguous or scalar-only copied annotations remain available in the JSONL event log when enabled. diff --git a/docs/agent-loop-boundary.md b/docs/agent-loop-boundary.md index 7ebe1b4..d24b673 100644 --- a/docs/agent-loop-boundary.md +++ b/docs/agent-loop-boundary.md @@ -48,8 +48,17 @@ replies, post-processed result bodies, injections, rollbacks and compaction. It preserves the normalizer's reasoning policy and native content blocks rather than reconstructing assistant messages from the executable subset of calls. Diagnostic fields (`usage`, `thinking_blocks`, `finish_reason`, `stop_details`) -remain annotations on the corresponding assistant message, including after -compaction or a thinking reset substitutes a copy of it; `thinking_blocks` is +remain annotations on the corresponding assistant message. After compaction +or a thinking reset substitutes a shallow copy, diagnostics move to the copy +only when it shares mutable content/tool-call containers or blocks with exactly +one vanished assistant and no other copy claims that assistant. Scalar object +identity is not evidence: separate JSON responses can share cached strings. +Ambiguous, deep-copied and plain-text-only copies without shared mutable +structure keep their diagnostics in JSONL when enabled. Source deduplication +uses entry identity and never compares message dictionaries by value, so cyclic +host metadata can reach the rendering fallback safely. +Messages without diagnostics, including resumed assistants, also count as +possible sources. `thinking_blocks` is omitted while the same blocks are still native assistant content. Values that are not JSON (datetimes, bytes, SDK objects) are written in a lossy string form rather than failing the snapshot. Images are redacted @@ -60,8 +69,11 @@ are outside this durable-history snapshot. JSON snapshots are coalesced during a run (LLM input and response, each tool result and turn end count toward `SWARM_TRAJECTORY_COALESCE_N`) and atomically -replaced; each snapshot re-renders only messages added or changed since the -previous one. Completion and cancellation force a snapshot of the latest +replaced. Routine snapshots compare the spool prefix by reference, which +catches replaced messages and values without serializing history. Edits nested +inside a kept container are found by comparing redacted rendered lines, a +full pass that runs after a compactor run, a rollback or context-limit notice, +and on completion or cancellation. Completion and cancellation force a snapshot of the latest history, and a cancellation that arrives after completion leaves the terminal snapshot untouched. Hosts that drive the hooks without loop history still get the assistant and tool entries built from the hook arguments. JSONL is the event diff --git a/tests/test_trajectory_history.py b/tests/test_trajectory_history.py index f0f90c7..5e5e20e 100644 --- a/tests/test_trajectory_history.py +++ b/tests/test_trajectory_history.py @@ -369,8 +369,11 @@ async def test_diagnostics_survive_a_signed_thinking_reset(tmp_path): history = [{"role": "user", "content": "start"}, assistant] await trajectory.on_llm_response(_turn(history, ai_text="a", thinking_blocks=blocks, usage={"input_tokens": 3})) - # _reset_signed_thinking_history substitutes a copy without the blocks. - history[1] = {**assistant, "content": [{"type": "text", "text": "a"}]} + # The real reset substitutes a copy without the blocks. + from agent_core.runtime.loop.agent_loop import _reset_signed_thinking_history + + _reset_signed_thinking_history(history) + assert history[1] is not assistant await trajectory.on_loop_end(AgentLoopResult(messages=history)) last = _messages(tmp_path / "trace.json")[-1] assert last["usage"] == {"input_tokens": 3} @@ -453,17 +456,282 @@ def counting(message): monkeypatch.setattr(trajectory, "_message_to_dict", counting) history = [{"role": "user", "content": f"m{i}"} for i in range(3)] trajectory._history = history - trajectory._flush_json(force=True) + trajectory._flush_json() history.append({"role": "user", "content": "m3"}) - trajectory._flush_json(force=True) + trajectory._flush_json() assert rendered == ["m0", "m1", "m2", "m3"] # Value replacement in place (as image eviction does) is detected. history[1]["content"] = "edited" - trajectory._flush_json(force=True) + trajectory._flush_json() assert rendered[4:] == ["edited", "m2", "m3"] # So is a rewrite that removes messages. history[:] = [history[0], history[3]] - trajectory._flush_json(force=True) + trajectory._flush_json() assert [m["content"] for m in _messages(tmp_path / "trace.json")] == ["m0", "m3"] assert rendered[7:] == ["m3"] + + +@pytest.mark.parametrize("field", ["text", "arguments", "image_meta", "usage"]) +async def test_snapshots_detect_nested_history_edits(tmp_path, field): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + assistant = { + "role": "assistant", "content": [{"type": "text", "text": "old"}], + "tool_calls": [_call("c")], "image_meta": [{"label": "old", "tokens": 1}], + } + usage = {"prompt_tokens": 10, "details": {"cached": 1}} + history = [assistant] + await trajectory.on_llm_response(_turn(history, usage=usage)) + trajectory._flush_json(force=True) + if field == "text": + assistant["content"][0]["text"] = "new" + elif field == "arguments": + assistant["tool_calls"][0]["function"]["arguments"] = '{"value":"new"}' + elif field == "image_meta": + assistant["image_meta"][0]["label"] = "new" + else: + usage["details"]["cached"] = 2 + # Completion must detect edits even when every container keeps its identity. + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + saved = _messages(tmp_path / "trace.json")[0] + assert _without_diagnostics([saved]) == history + assert saved["usage"] == usage + + +@pytest.mark.parametrize("copies", [1, 2]) +@pytest.mark.parametrize("first_has_details", [False, True]) +async def test_duplicate_assistant_copies_do_not_guess_diagnostics(tmp_path, copies, first_has_details): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace") + history = [{"role": "assistant", "content": "same"}] + await trajectory.on_llm_response(_turn( + history, usage={"prompt_tokens": 10} if first_has_details else None, + finish_reason="length" if first_has_details else "stop", + )) + history.append({"role": "assistant", "content": "same"}) + await trajectory.on_llm_response(_turn(history, turn=2, usage={"prompt_tokens": 20}, + finish_reason="refusal")) + # A compactor copies the retained tail, optionally dropping the first turn. + history[:] = [dict(message) for message in history[-copies:]] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert _messages(tmp_path / "trace.json") == history + events = [json.loads(line) for line in (tmp_path / "trace.jsonl").read_text().splitlines()] + assert [event.get("usage") for event in events if event["t"] == "llm"] == [ + {"prompt_tokens": 10} if first_has_details else None, {"prompt_tokens": 20}, + ] + + +async def test_identity_keeps_diagnostics_when_other_assistant_has_same_text(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + history = [{"role": "assistant", "content": "same"}] + await trajectory.on_llm_response(_turn(history, usage={"prompt_tokens": 10})) + history.append({"role": "assistant", "content": "same"}) + await trajectory.on_llm_response(_turn(history, turn=2, usage={"prompt_tokens": 20})) + history[:] = [history[-1]] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert _messages(tmp_path / "trace.json")[0]["usage"] == {"prompt_tokens": 20} + + +async def test_copied_diagnostic_does_not_attach_to_multiple_candidates(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + history = [{"role": "assistant", "content": "same"}] + await trajectory.on_llm_response(_turn(history, usage={"prompt_tokens": 10})) + history[:] = [dict(history[0]), dict(history[0])] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert _messages(tmp_path / "trace.json") == history + + +async def test_resumed_assistant_without_details_counts_as_an_ambiguous_source(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + history = [{"role": "assistant", "content": "same"}] + await trajectory.on_before_llm(_turn(history)) + history.append({"role": "assistant", "content": "same"}) + await trajectory.on_llm_response(_turn(history, usage={"prompt_tokens": 20})) + history[:] = [dict(history[0])] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert _messages(tmp_path / "trace.json") == history + + +async def test_identity_disambiguates_a_copy_with_identical_text(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + history = [{"role": "assistant", "content": "same", "tool_calls": [_call("first")]}] + await trajectory.on_llm_response(_turn(history, usage={"prompt_tokens": 10})) + history.append({"role": "assistant", "content": "same", "tool_calls": [_call("second")]}) + await trajectory.on_llm_response(_turn(history, turn=2, usage={"prompt_tokens": 20})) + history[-1] = dict(history[-1]) + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert [m["usage"]["prompt_tokens"] for m in _messages(tmp_path / "trace.json")] == [10, 20] + + +async def test_loop_compactor_can_edit_a_native_block_in_place(tmp_path): + class EditBlock: + def compact(self, messages, keep_recent): + for message in messages: + if message.get("role") == "assistant" and isinstance(message.get("content"), list): + for block in message["content"]: + if block.get("type") == "text": + block["text"] = "compacted text" + return messages + + messages, _, _ = await _run(tmp_path, [ + LLMResponse(content=[ + {"type": "thinking", "thinking": "r", "signature": "s"}, + {"type": "text", "text": "original"}, + ], tool_calls=[_call("c")]), LLMResponse(content="done"), + ], model_profile=ModelProfile(model_id="test", provider="test", thinking_format="content_block"), + config=LoopConfig(max_turns=3, compact_after_turns=0, compactor=EditBlock(), + loop_policy=LoopPolicy(no_tool_behavior="stop"))) + first = next(m for m in messages if m["role"] == "assistant") + assert first["content"][-1]["text"] == "compacted text" + + +async def test_routine_snapshots_do_not_reserialize_the_prefix(tmp_path, monkeypatch): + from agent_core.runtime.loop.compact import COMPACTION_SEQ_KEY + + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + rendered = [] + original = trajectory._message_to_dict + + def counting(message): + rendered.append(message.get("content")) + return original(message) + + monkeypatch.setattr(trajectory, "_message_to_dict", counting) + image = {"type": "image_url", "image_url": {"url": "data:image/png;base64," + "A" * 4096}} + block = {"type": "text", "text": "old"} + history = [{"role": "user", "content": [block, image]}, {"role": "assistant", "content": "a"}] + metadata = {} + await trajectory.on_before_llm(_turn(history, metadata=metadata)) + await trajectory.on_llm_input(_turn(list(history), metadata=dict(metadata))) + await trajectory.on_turn_end(_turn(history, metadata=metadata)) + assert len(rendered) == 2 + + # A nested edit with no compaction is not looked for on routine snapshots. + block["text"] = "edited" + await trajectory.on_llm_input(_turn(list(history), metadata=dict(metadata))) + assert len(rendered) == 2 + # A compactor run (its counter advances even without a notice) schedules + # one deep comparison of the redacted lines, which finds the edit. + metadata[COMPACTION_SEQ_KEY] = 1 + await trajectory.on_llm_input(_turn(list(history), metadata=dict(metadata))) + assert _messages(tmp_path / "trace.json")[0]["content"][0]["text"] == "edited" + assert "AAAA" not in (tmp_path / "trace.json").read_text() + count = len(rendered) + await trajectory.on_turn_end(_turn(history, metadata=metadata)) + assert len(rendered) == count + + +async def test_rollback_notice_schedules_a_deep_check(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + block = {"type": "text", "text": "old"} + history = [{"role": "user", "content": [block]}] + await trajectory.on_before_llm(_turn(history)) + await trajectory.on_turn_end(_turn(history)) + block["text"] = "edited" + await trajectory.on_context_compacted(None) + await trajectory.on_turn_end(_turn(history)) + assert _messages(tmp_path / "trace.json")[0]["content"][0]["text"] == "edited" + + +async def test_diagnostics_follow_an_in_place_edit_and_a_later_copy(tmp_path): + from agent_core.runtime.loop.agent_loop import _reset_signed_thinking_history + + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + blocks = [{"type": "thinking", "thinking": "r", "signature": "s"}] + text = {"type": "text", "text": "original"} + history = [{"role": "assistant", "content": [*blocks, text], "tool_calls": [_call("c")]}, + {"role": "tool", "tool_call_id": "c", "content": "ok"}] + await trajectory.on_llm_response(_turn(history, thinking_blocks=blocks, + usage={"prompt_tokens": 10})) + trajectory._flush_json(force=True) + text["text"] = "compacted text" + _reset_signed_thinking_history(history) + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + first = _messages(tmp_path / "trace.json")[0] + assert first["content"] == [{"type": "text", "text": "compacted text"}] + assert first["usage"] == {"prompt_tokens": 10} + + +async def test_a_new_assistant_with_equal_text_never_inherits_a_vanished_copy(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + first = {"role": "assistant", "content": [{"type": "text", "text": "ok"}]} + history = [first] + await trajectory.on_llm_response(_turn(history, usage={"prompt_tokens": 10}, + finish_reason="length")) + history[0] = dict(first) + trajectory._flush_json(force=True) + assert _messages(tmp_path / "trace.json")[0]["usage"] == {"prompt_tokens": 10} + # Within one coalesced window the copy is dropped and an unrelated + # assistant with the same text arrives without its own diagnostics. + history[:] = [{"role": "assistant", "content": [{"type": "text", "text": "ok"}]}] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert _messages(tmp_path / "trace.json") == history + + +async def test_unencodable_history_values_keep_the_snapshot_and_pairing(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + await trajectory.on_loop_start(LoopConfig()) + cycle = {} + cycle["self"] = cycle + history = [ + {"role": "assistant", "content": "", "tool_calls": [_call("c")]}, + {"role": "tool", "tool_call_id": "c", "content": "ok", "meta": cycle}, + {"role": "tool", "tool_call_id": "d", "content": "ok", "meta": {(1, 2): "tuple key"}}, + {"role": "user", "content": "after"}, + ] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + messages = _messages(tmp_path / "trace.json") + assert [m.get("tool_call_id") for m in messages] == [None, "c", "d", None] + assert "unserializable" in messages[1]["trace_error"] + assert "unserializable" in messages[2]["trace_error"] + assert messages[-1]["content"] == "after" + + +@pytest.mark.parametrize("field", ["content", "reasoning_content"]) +async def test_shared_scalar_strings_do_not_transfer_diagnostics(tmp_path, field): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + first = json.loads('{"role":"assistant","content":"0"}') + replacement = json.loads('{"role":"assistant","content":"0"}') + # CPython's JSON decoder caches single-character strings across responses. + assert first["content"] is replacement["content"] + if field == "reasoning_content": + first["reasoning_content"] = first["content"] + replacement["reasoning_content"] = replacement["content"] + first["content"] = "old reply" + replacement["content"] = "new reply" + history = [first] + await trajectory.on_llm_response(_turn(history, usage={"prompt_tokens": 10}, + finish_reason="length")) + # The host replaces a discarded turn with an unobserved synthetic reply. + history[:] = [replacement] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert _messages(tmp_path / "trace.json") == history + + +async def test_copy_lineage_deduplication_does_not_compare_cyclic_messages(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace", formats=["json"]) + block = {"type": "text", "text": "same"} + history = [] + for turn in (1, 2): + cycle = {} + cycle["self"] = cycle + history.append({"role": "assistant", "content": [block], "meta": cycle}) + await trajectory.on_llm_response(_turn(history, turn=turn, usage={"prompt_tokens": 10 * turn})) + # Both vanished sources share a block but have separate cyclic metadata. + # Comparing the message dicts by value recurses before rendering can fall back. + history[:] = [dict(history[-1]), {"role": "user", "content": "after"}] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + messages = _messages(tmp_path / "trace.json") + assert messages[0]["role"] == "assistant" + assert "unserializable" in messages[0]["trace_error"] + assert messages[1] == {"role": "user", "content": "after"} + + +async def test_plain_text_copy_does_not_guess_diagnostics_from_scalar_aliases(tmp_path): + trajectory = TrajectoryFileObserver(tmp_path, filename="trace") + history = [{"role": "assistant", "content": "plain text", "reasoning_content": "reason"}] + await trajectory.on_llm_response(_turn(history, usage={"prompt_tokens": 10})) + history[:] = [dict(history[0])] + await trajectory.on_loop_end(AgentLoopResult(messages=history)) + assert _messages(tmp_path / "trace.json") == history + events = [json.loads(line) for line in (tmp_path / "trace.jsonl").read_text().splitlines()] + assert next(event for event in events if event["t"] == "llm")["usage"] == {"prompt_tokens": 10}