diff --git a/artemis/agents/flash/runner.py b/artemis/agents/flash/runner.py index 5a8f722..081ec28 100644 --- a/artemis/agents/flash/runner.py +++ b/artemis/agents/flash/runner.py @@ -18,9 +18,9 @@ Executes autonomous reactive mobile workflows through OpenAI Responses. Conversation shape (shared with the Pro operator, history redesign §3.2): the prompt is built every turn from a session :class:`TranscriptLedger` — -a byte-stable system prefix, the committed earlier turns (append-only; old UI -lists stripped at depth 1, screenshots resolved to visual summaries at depth -K, long spans chunk-compressed), and a fresh tail carrying the current +a byte-stable system prefix, the committed earlier turns (append-only within +a context phase; old observations and long spans compacted together only at +pressure thresholds), and a fresh tail carrying the current observation under a ``# CURRENT OBSERVATION [T+mm:ss]`` header. Actions run synchronously and their tool messages already carry the outcome; only a turn whose action failed ends with an ``--- Action Execution Result (T+mm:ss) ---`` @@ -30,7 +30,7 @@ session recording. Per-turn notices on the tail (the reasoning reminder after a silent turn, the no-tool-call nudge, the final-turn warning, the user-guidance wrapper) are -marked ephemeral: the ledger's scrub edge deletes them once the message leaves +marked ephemeral: the ledger's compaction pass deletes them when it rewrites the recent window, so they never reach the frozen history or a chunk capsule. The verbatim body of a user instruction is the exception: it stays in the active window until its turn is chunked, whose user lines then carry it on. @@ -83,7 +83,12 @@ from artemis.llm.router import ModelEndpoint from artemis.llm.structured import ParseFailure, parse_structured from artemis.mcp.action_executor import McpActionExecutor from artemis.mcp.observation import DeviceObservationError, require_observation -from artemis.memory.transcript import PRO_UI_LIST_MARKER, TranscriptLedger, mark_ephemeral +from artemis.memory.transcript import ( + OBSERVATION_SCOPE_RULE, + PRO_UI_LIST_MARKER, + TranscriptLedger, + mark_ephemeral, +) from artemis.services.llm import ( RobustChatModelWrapper, acomplete, @@ -272,6 +277,8 @@ class FlashRunner: image_scrub_depth_relaxed=getattr(cfg, "image_scrub_depth_relaxed", None), context_budget_tokens=getattr(cfg, "context_budget_tokens", None), start_ratio=getattr(cfg, "start_ratio", None), + soft_ratio=getattr(cfg, "soft_ratio", 0.7), + hard_ratio=getattr(cfg, "hard_ratio", 0.9), session_start=session_start if isinstance(session_start, (int, float)) else None, ) if engine is not None: @@ -1247,7 +1254,13 @@ class FlashRunner: """Installs the static prefix and captures the initial device state.""" ledger = self._build_ledger() ledger.set_static_prefix( - [SystemMessage(content=self._render_system_prompt(tools_declaration))] + [ + SystemMessage( + content=self._render_system_prompt(tools_declaration) + + "\n" + + OBSERVATION_SCOPE_RULE + ) + ] ) # Capture Initial State (Screenshot + UI Tree) @@ -1369,7 +1382,7 @@ class FlashRunner: ) pending_notices = [] - # S + F + A + tail: chunk compression and the scrub edge advance here. + # S + F + A + tail: history changes only at a compaction boundary. messages = ledger.render([tail]) turn_base = len(messages) - 1 current_tools = report_only_tools if is_final else self._get_tools() @@ -1427,7 +1440,7 @@ class FlashRunner: # ratio — only from a provider measurement, never from the # estimate (same as the Pro operator). ledger.record_prompt_tokens( - step_token_usage.get("prompt_tokens"), + step_token_usage.get("prompt_tokens") if measured else None, messages=messages[:-1] if measured else None, ) diff --git a/artemis/agents/operator/operator.py b/artemis/agents/operator/operator.py index 0913b0a..c81d787 100644 --- a/artemis/agents/operator/operator.py +++ b/artemis/agents/operator/operator.py @@ -330,6 +330,8 @@ class OperatorNode: image_scrub_depth_relaxed=getattr(cfg, "image_scrub_depth_relaxed", None), context_budget_tokens=getattr(cfg, "context_budget_tokens", None), start_ratio=getattr(cfg, "start_ratio", None), + soft_ratio=getattr(cfg, "soft_ratio", 0.7), + hard_ratio=getattr(cfg, "hard_ratio", 0.9), session_start=session_start if isinstance(session_start, (int, float)) else None, ) # L2/L3 chunk compression (M3) rides the same flag; without a @@ -392,7 +394,7 @@ class OperatorNode: """ from langchain_core.messages import HumanMessage - from artemis.memory.transcript import mark_ephemeral + from artemis.memory.transcript import OBSERVATION_SCOPE_RULE, mark_ephemeral ledger = self._ensure_transcript_ledger(state) @@ -407,6 +409,7 @@ class OperatorNode: if getattr(self.ctx, "plugins", None): prefix.append(self.ctx.plugins.prompt()) prefix.append(static_text) + prefix.append(OBSERVATION_SCOPE_RULE) ledger.set_static_prefix([SystemMessage(content="\n".join(prefix))]) # 2. F region cold start: an empty ledger over an existing step record @@ -496,7 +499,7 @@ class OperatorNode: tail_content.append(part) tail = HumanMessage(content=tail_content) # Per-turn notices (reminders, hints, guidance, incident blocks) are - # deleted by the scrub edge once the turn leaves the recent window. + # deleted only when phase compaction rewrites the recent window. if builder.ephemeral_indices: mark_ephemeral(tail, builder.ephemeral_indices) @@ -635,12 +638,8 @@ class OperatorNode: transcript_ledger.record_prompt_tokens( usage.get("prompt_tokens"), messages=current_messages ) - elif hasattr(transcript_ledger, "record_prompt_tokens"): - # No usage metadata from the provider: record an estimate so - # the start gate and thresholds still see a context base. - from artemis.memory.transcript import estimate_prompt_tokens - - transcript_ledger.record_prompt_tokens(estimate_prompt_tokens(current_messages)) + # Without provider usage the ledger estimates the complete next + # prompt at render time; do not label that estimate as a measurement. if response.content: if isinstance(response.content, str): diff --git a/artemis/config/agent.py b/artemis/config/agent.py index c8ce7ab..906bdbd 100644 --- a/artemis/config/agent.py +++ b/artemis/config/agent.py @@ -425,7 +425,7 @@ class MemoryRuntimeConfig(BaseModel): class MemoryTranscriptConfig(BaseModel): - """Scrub-edge options for the message transcript (agent.memory.transcript).""" + """Phased context compaction for the shared Flash/Pro transcript.""" enabled: bool = Field( default=True, @@ -443,7 +443,7 @@ class MemoryTranscriptConfig(BaseModel): default=3, ge=1, description=( - "Depth K of the screenshot scrub edge from the start gate up: the" + "At a compaction boundary, depth K of the screenshot scrub edge: the" " K-th most recent screenshot (the live one counts as depth 1) is" " replaced by its visual summary." ), @@ -452,11 +452,9 @@ class MemoryTranscriptConfig(BaseModel): default=6, ge=1, description=( - "Screenshot scrub depth while the operator's measured context is" - " below budget*start_ratio (nothing is being replaced yet, so" - " more raw screenshots stay in view). Must be >= image_scrub_depth;" - " set equal to it to disable the occupancy-driven depth. A change" - " of depth never rewrites an already-scrubbed message." + "Screenshot depth for an explicit low-occupancy compaction. Normal" + " turns retain previously sent images unchanged. Must be >=" + " image_scrub_depth." ), ) pending_grace_steps: int = Field( @@ -472,13 +470,10 @@ class MemoryTranscriptConfig(BaseModel): default=1, ge=1, description=( - "Text edge of the scrub: once a message is deeper than this many" - " image-bearing observations (1 = as soon as a newer observation" - " exists) its UI element list, plan recitation and per-turn" - " ephemeral blocks are removed. Kept shallow on purpose: only the" - " live observation carries an indexed element list, so a stale" - " [n] index can never be picked up as a target. Screenshots follow" - " image_scrub_depth / image_scrub_depth_relaxed." + "At phase compaction, remove old UI lists, plan recitations and" + " ephemeral notices beyond this observation depth. Between phases" + " old content stays immutable; only the latest observation may" + " supply current action targets." ), ) context_budget_tokens: int = Field( @@ -496,12 +491,9 @@ class MemoryTranscriptConfig(BaseModel): ge=0.0, lt=1.0, description=( - "Start gate: below budget*start_ratio of the operator's measured" - " prompt size, closed segments only prepare their capsule in the" - " background — the original turns stay in the transcript. Ready" - " chunks replace their turns once the measured context reaches" - " this ratio (the hard threshold overrides it). 0 restores" - " swap-as-soon-as-ready." + "Preparation gate: at budget*start_ratio begin background chunk" + " summaries without replacing the live prefix. Zero enables early" + " preparation, not immediate replacement. Swaps wait for soft_ratio." ), ) soft_ratio: float = Field( @@ -509,9 +501,9 @@ class MemoryTranscriptConfig(BaseModel): gt=0.0, lt=1.0, description=( - "Soft threshold: when the operator's last measured prompt size" - " reaches budget*soft_ratio, the oldest open segment is" - " chunk-compressed (L2)." + "At budget*soft_ratio, replace prepared summaries as one batch and" + " scrub old observations, starting a new stable context phase." + " Pending summaries retain original turns until ready or hard pressure." ), ) hard_ratio: float = Field( @@ -519,7 +511,7 @@ class MemoryTranscriptConfig(BaseModel): gt=0.0, le=1.0, description=( - "Hard threshold: when the operator's last measured prompt size" + "Hard threshold: when the projected current prompt size" " reaches budget*hard_ratio, ready chunks swap as L2 blocks first;" " only if that would not bring the estimated context back under the" " line does everything closed force-swap (pending chunks included)" diff --git a/artemis/context.py b/artemis/context.py index 06fd460..b0aa7a5 100644 --- a/artemis/context.py +++ b/artemis/context.py @@ -267,6 +267,9 @@ class ArtemisContext(BaseModel): operator's transcript prompt path (``agent.memory.transcript.enabled``). Lives beside ``step_memory``; the State never carries message payloads.""" + prompt_prefix_tracker: Any | None = None + """Task-local, hash-only prefix diagnostics shared by bound LLM wrappers.""" + async def __aenter__(self) -> ArtemisContext: return self diff --git a/artemis/llm/prefix_cache.py b/artemis/llm/prefix_cache.py new file mode 100644 index 0000000..4b293e4 --- /dev/null +++ b/artemis/llm/prefix_cache.py @@ -0,0 +1,65 @@ +"""Local prefix diagnostics; retain hashes only, never prompt/image payloads. + +These compare adapted messages, not the gateway's tokenized wire prefix. +Only provider-reported cached_tokens is a measured cache hit. +""" + +import hashlib +import json + +from langchain_core.messages import BaseMessage + + +def _digest(value) -> str: + return hashlib.sha256( + json.dumps(value, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest() + + +class PromptPrefixTracker: + def __init__(self): + self._previous: dict[str, tuple[str, tuple[str, ...]]] = {} + + def observe(self, identity: str, messages, options: dict) -> dict: + identity = _digest(identity) + if not isinstance(messages, (list, tuple)) or not all( + isinstance(message, BaseMessage) for message in messages + ): + return {"comparison": "unavailable"} + fingerprints = tuple( + _digest( + { + "role": message.type.removesuffix("MessageChunk").lower(), + "content": message.content, + "name": message.name, + "tool_calls": getattr(message, "tool_calls", None), + "tool_call_id": getattr(message, "tool_call_id", None), + } + ) + for message in messages + ) + options_hash = _digest(options) + previous = self._previous.get(identity) + # Task-scoped and bounded even if a caller tries many endpoint roles. + if identity not in self._previous and len(self._previous) >= 32: + self._previous.pop(next(iter(self._previous))) + self._previous[identity] = (options_hash, fingerprints) + common = 0 + if previous: + for old, new in zip(previous[1], fingerprints): + if old != new: + break + common += 1 + schema_changed = previous is not None and previous[0] != options_hash + return { + "comparison": "local_message_fingerprints", + "message_count": len(fingerprints), + "previous_message_count": len(previous[1]) if previous else None, + "shared_prefix_messages": common if previous else None, + "first_changed_message": ( + common + 1 if previous and common < len(previous[1]) else None + ), + "append_only": bool(previous and common == len(previous[1]) and not schema_changed), + "tools_or_format_changed": schema_changed, + "cold_start": previous is None, + } diff --git a/artemis/memory/chunking.py b/artemis/memory/chunking.py index a61c226..3d95f58 100644 --- a/artemis/memory/chunking.py +++ b/artemis/memory/chunking.py @@ -1074,7 +1074,12 @@ class HistoryChunkManager: soft = base_tokens is not None and base_tokens >= self._budget * self._soft_ratio hard = base_tokens is not None and base_tokens >= self._budget * self._hard_ratio - closed_any = self._close_new_segments(ledger, soft, hard) + preparing = base_tokens is not None and base_tokens >= self._budget * self._start_ratio + closed_any = ( + self._close_new_segments(ledger, soft, hard) + if hard or (preparing and (not soft or not self._awaiting)) + else False + ) if closed_any or soft or hard: # Failure rung of the degradation ladder: exhausted capsules are # re-dispatched (the lens retries a fallback model per attempt); @@ -1206,14 +1211,10 @@ class HistoryChunkManager: contiguous READY prefix may swap — an older pending segment keeps its (and every younger segment's) original text live. - Start gate: closing a segment (milestone/size triggers) only *prepares* - its capsule in the background; the original text is replaced only once - the operator's measured context has reached ``budget * start_ratio``. - Below that the ready chunks are held in the awaiting queue, so at low - occupancy the model keeps reading the full transcript while capsules - are already on hand for when pressure arrives. An unknown base (no - measured operator call yet, or a provider without usage metadata) - cannot gate and falls through to swap-on-ready. + The start gate enables background preparation. Only the soft gate + replaces live history, in one batch after all prepared headers are + ready. The hard gate may force a swap if summaries are still pending. + Missing provider usage uses the ledger's current prompt estimate. Hard threshold ladder (L2 first, L3 last): at ``budget * hard_ratio`` the ready prefix swaps as ordinary L2 chunk blocks when the estimated @@ -1262,7 +1263,13 @@ class HistoryChunkManager: else: swap = list(self._awaiting) # rung 2: force everything, fold fold = True - elif base_tokens is not None and base_tokens < self._budget * self._start_ratio: + elif base_tokens is None or base_tokens < self._budget * self._soft_ratio: + if base_tokens is not None: + self._announce_held(base_tokens, ledger) + return + elif len(ready_prefix) < len(self._awaiting): + # Batch the prepared summaries into one rewrite. Async completion + # must not cause a succession of partial prefix replacements. self._announce_held(base_tokens, ledger) return else: @@ -1290,7 +1297,7 @@ class HistoryChunkManager: # merged into the period paragraph); a pending chunk has no summary # size to report, so its trace carries no compression ratio. swapped = [e for e in swap if e["chunk"] is not None] - source_tokens_total = 0 + source_tokens_total = sum(ledger.turn_prompt_tokens(e["turns"]) for e in swap) replacement_tokens_total = 0 for entry in swapped: chunk = entry["chunk"] @@ -1298,7 +1305,6 @@ class HistoryChunkManager: chunk.summary_chars = len(render_chunk_block(chunk)) else: chunk.summary_chars = None - source_tokens_total += self._chars_to_tokens(chunk.source_chars, ledger) replacement_tokens_total += self._chars_to_tokens( self._replacement_chars(chunk, folded=fold), ledger ) @@ -1392,18 +1398,17 @@ class HistoryChunkManager: delta = 0 for entry in entries: chunk = entry["chunk"] - if chunk is None: - continue - delta += self._chars_to_tokens( - self._replacement_chars(chunk), ledger - ) - self._chars_to_tokens(chunk.source_chars, ledger) + replacement = self._replacement_chars(chunk) if chunk is not None else 0 + delta += self._chars_to_tokens(replacement, ledger) - ledger.turn_prompt_tokens( + entry["turns"] + ) return max(0, int(base_tokens) + delta) def _announce_held(self, base_tokens: int, ledger: Any = None) -> None: - """Once per chunk: its capsule is ready but the start gate holds the + """Once per chunk: its capsule is ready but the soft gate holds the original text live. The timeline line stays ``running`` with a ``held`` note so the UI can say the summary is ready and waiting.""" - threshold = int(self._budget * self._start_ratio) + threshold = int(self._budget * self._soft_ratio) for entry in self._awaiting: chunk = entry["chunk"] if chunk is None or chunk.status != "ready" or entry.get("held_announced"): @@ -1468,7 +1473,7 @@ class HistoryChunkManager: """The live context base for the start/soft/hard thresholds. An injected ``meter_getter`` wins (tests / custom wiring); otherwise - the base is the owning operator's last measured prompt size recorded + the base is the ledger's projected current prompt, calibrated from usage recorded on the ledger (:meth:`TranscriptLedger.record_prompt_tokens`). The session-wide token meter is deliberately NOT consulted: its ``last_prompt_tokens`` is overwritten by whichever agent called last @@ -1481,6 +1486,9 @@ class HistoryChunkManager: except Exception: return None try: + projected = getattr(ledger, "projected_prompt_tokens", None) + if isinstance(projected, int) and projected > 0: + return projected last = getattr(ledger, "last_prompt_tokens", None) return int(last) if last else None except (TypeError, ValueError): diff --git a/artemis/memory/transcript.py b/artemis/memory/transcript.py index e7fadc9..ac2a7b2 100644 --- a/artemis/memory/transcript.py +++ b/artemis/memory/transcript.py @@ -24,14 +24,11 @@ Messages are organized into four regions: (tool_calls and native thinking preserved by reference), the in-turn ToolMessages, and the turn's result message (every turn on the Pro path; only when an action failed on the Flash path). The - scrub edge (:class:`~artemis.agents.flash.context_compressor.ScrubEdgeCompressor`) - has two depths: a shallow text edge (UI list, plan recitation and ephemeral - per-turn blocks leave as soon as a newer observation exists) and a - screenshot edge that follows the occupancy tier (relaxed below the start - gate, tightened past it), where the screenshot is replaced in place by its - visual summary. Every rewrite happens once and is never undone; everything - newer is byte-identical to what the model was first shown. Messages are - never removed or reordered, so tool-call/response pairs are never split. + scrub edge runs only at a pressure-driven phase boundary, together with + chunk replacement. Normal renders preserve all previously sent content, + including screenshots and UI lists. Chunk summaries use a compact copy of + each turn; their background completion never rewrites the live prefix. + Tool-call/response pairs are never split. - **T (current tail)**: built fresh every turn by the operator; passed to :meth:`render` and only enters A when the turn is committed. @@ -74,8 +71,8 @@ RESTORED_HISTORY_HEADER = "[Restored history]" #: ``HumanMessage.additional_kwargs`` key listing the indices of content blocks that #: are only meaningful for the turn they were built for (reminders, hints, user -#: guidance, per-turn notices). The scrub edge deletes them at its text edge, so -#: they never reach the frozen region or a chunk capsule. The indices always +#: guidance, per-turn notices). At a compaction boundary the scrub edge deletes them; +#: capsule source copies also omit them. The indices always #: describe the message's *current* block layout: a rewrite that removes blocks #: before the text edge (the screenshot swap when the image edge is shallower) #: remaps the surviving indices in place, and the key is dropped once the blocks @@ -83,6 +80,13 @@ RESTORED_HISTORY_HEADER = "[Restored history]" #: marking costs the model nothing. EPHEMERAL_BLOCKS_KEY = "ephemeral_blocks" +OBSERVATION_SCOPE_RULE = ( + "历史截图、元素编号、坐标和逐轮提醒仅用于回顾。每次操作必须依据同一设备/来源的" + "最新观察;旧消息中的 CURRENT OBSERVATION 不表示仍然有效。最新任务计划和用户指导" + "优先于旧版本;旧的结束提示不得覆盖新指导。需要确认当前状态时重新观察,禁止从旧" + "元素列表选择目标或用历史截图证明当前操作成功。" +) + def mark_ephemeral(message: BaseMessage, indices) -> None: """Flag content blocks of ``message`` (by index) as ephemeral, see @@ -190,6 +194,10 @@ def render_turn_transcript(messages: list[BaseMessage]) -> str: #: (Gemini's flat per-image cost; the Flash runner uses the same figure). IMAGE_BLOCK_TOKENS = 258 +# Conservative growth allowance for a new screenshot before provider usage +# arrives. Removed images use IMAGE_BLOCK_TOKENS to avoid overstating savings. +PROJECTED_IMAGE_TOKENS = 2048 + #: Characters per token assumed before any calibrated measurement. DEFAULT_CHARS_PER_TOKEN = 4.0 @@ -268,8 +276,10 @@ class TranscriptLedger: pending_grace_steps: int = 3, xml_scrub_depth: int | None = None, image_scrub_depth_relaxed: int | None = None, - context_budget_tokens: int | None = None, + context_budget_tokens: int | None = 80000, start_ratio: float | None = None, + soft_ratio: float = 0.7, + hard_ratio: float = 0.9, clock: Callable[[], float] | None = None, session_start: float | None = None, ): @@ -308,6 +318,10 @@ class TranscriptLedger: # calls sharing the session can never masquerade as the operator's # context (see HistoryChunkManager._context_base_tokens). self._last_prompt_tokens: int | None = None + self._prompt_sample: tuple[int, int, int] | None = None + self._projected_prompt_tokens: int | None = None + self.context_phase = 0 + self.last_compaction: dict[str, Any] | None = None # Session-calibrated characters-per-token ratio (see # record_prompt_tokens): the chunker converts transcript characters # to tokens through it, so CJK-heavy sessions (~1 char/token on @@ -326,26 +340,21 @@ class TranscriptLedger: # it to decide whether to attach the shared reasoning reminder. self._last_turn_silent: bool = False - # Text edge (default 1): UI list, plan recitation and ephemeral blocks - # leave as soon as a newer observation exists; screenshots follow K. + # At phase compaction, strip old UI/plan/notices and resolve images. + # Between boundaries none of these fields is rewritten. self._xml_scrub_depth = xml_scrub_depth - # Occupancy-driven screenshot depth. Below the start gate - # (``last_prompt_tokens < budget * start_ratio``, or no measurement - # yet) the scrub edge sits at the relaxed depth so the model keeps - # more raw screenshots while nothing is being replaced; from the - # start gate up it tightens to ``image_scrub_depth``. Scrubbing is - # irreversible, so a change of depth never rewrites a message: - # tightening scrubs the extra depths once, relaxing lets new images - # live longer. See :attr:`effective_image_scrub_depth`. + # Depth settings apply only inside an explicit compaction pass. self._image_scrub_depth = max(1, int(image_scrub_depth)) self._image_scrub_depth_relaxed = ( max(self._image_scrub_depth, int(image_scrub_depth_relaxed)) if image_scrub_depth_relaxed is not None else self._image_scrub_depth ) - self._context_budget_tokens = int(context_budget_tokens) if context_budget_tokens else None + self._context_budget_tokens = int(context_budget_tokens or 80000) self._start_ratio = float(start_ratio) if start_ratio is not None else None + self._soft_ratio = float(soft_ratio) + self._hard_ratio = float(hard_ratio) # id(message) -> summary-job key (DataEngine step id). A side map keeps # the key out of the serialized message payload entirely. @@ -356,7 +365,8 @@ class TranscriptLedger: # whose failure depends on which side is imported first. from artemis.agents.flash.context_compressor import ScrubEdgeCompressor - self._compressor = ScrubEdgeCompressor( + self._compressor_type = ScrubEdgeCompressor + self._compressor_options = dict( summarizer=step_memory, prune_history_xml=prune_history_xml, image_scrub_depth=image_scrub_depth, @@ -366,6 +376,7 @@ class TranscriptLedger: strip_markers=(PRO_UI_LIST_MARKER, PLAN_RECITATION_MARKER), tail_offset=1, ) + self._compressor = ScrubEdgeCompressor(**self._compressor_options) # ------------------------------------------------------------------ # Session clock @@ -388,6 +399,27 @@ class TranscriptLedger: first measured call (or when the provider reports no usage).""" return self._last_prompt_tokens + @property + def projected_prompt_tokens(self) -> int | None: + """Current prompt including the fresh tail, estimated before compaction.""" + return self._projected_prompt_tokens + + def estimate_context_tokens(self, messages: list[BaseMessage]) -> int: + chars, images = measure_prompt_content(messages) + if self._prompt_sample is None: + # Without provider usage, count CJK conservatively rather than //4. + return max(1, self._last_prompt_tokens or 0, chars + images * PROJECTED_IMAGE_TOKENS) + prev_chars, prev_images, tokens = self._prompt_sample + return max( + 1, + tokens + + int((chars - prev_chars) / self._chars_per_token) + # New screenshots may tile to thousands of tokens. For removals, + # use the smaller lower bound rather than overclaiming savings. + + (images - prev_images) + * (PROJECTED_IMAGE_TOKENS if images >= prev_images else IMAGE_BLOCK_TOKENS), + ) + @property def last_turn_silent(self) -> bool: """True when the last committed turn had no visible reasoning text. @@ -455,7 +487,7 @@ class TranscriptLedger: @property def effective_image_scrub_depth(self) -> int: - """The screenshot depth the scrub edge uses at the next render: the + """The screenshot depth used at the next compaction boundary: the relaxed depth below the start gate, the tight depth from there up.""" if self.below_start_gate: return self._image_scrub_depth_relaxed @@ -514,6 +546,7 @@ class TranscriptLedger: if value <= 0: return self._last_prompt_tokens = value + self._projected_prompt_tokens = None if messages is None: return try: @@ -521,6 +554,7 @@ class TranscriptLedger: except Exception as exc: logger.debug(f"Prompt calibration skipped: {exc}", exc_info=True) return + self._prompt_sample = (text_chars, image_count, value) if text_chars <= 0: return @@ -763,11 +797,24 @@ class TranscriptLedger: return sum(len(self.turn_transcript(turn)) for turn in turns) def turn_transcript(self, turn: dict) -> str: - """Plain-text rendering of one committed turn exactly as it currently - stands in the active region (i.e. after the scrub edge), see - :func:`render_turn_transcript`. Chunk compression feeds this to the - segment capsule so the capsule digests precisely what it replaces.""" - return render_turn_transcript(self._active[turn["start"] : turn["end"]]) + """Compact capsule source without rewriting the live model prefix.""" + originals = self._active[turn["start"] : turn["end"]] + copies = [message.model_copy(deep=True) for message in originals] + keys = {id(new): self._step_keys.get(id(old)) for old, new in zip(originals, copies)} + options = { + **self._compressor_options, + "pending_grace_steps": 0, + "summary_key_getter": lambda message: keys.get(id(message)), + } + self._compressor_type(**options).compress(copies, image_scrub_depth=1) + return render_turn_transcript(copies) + + def turn_prompt_tokens(self, turns: list[dict]) -> int: + """Size actually removed by a swap, including retained image blocks.""" + chars, images = measure_prompt_content( + [message for turn in turns for message in self._active[turn["start"] : turn["end"]]] + ) + return self.chars_to_tokens(chars) + images * IMAGE_BLOCK_TOKENS def freeze_turns(self, turn_count: int, frozen_blocks: list[BaseMessage]) -> None: """Compression event: consume the oldest ``turn_count`` unchunked turns @@ -778,6 +825,7 @@ class TranscriptLedger: """ if turn_count <= 0: self._frozen_blocks = list(frozen_blocks) + self.context_phase += 1 return available = self._turns[self._chunked_turn_count :] if turn_count > len(available): @@ -788,36 +836,38 @@ class TranscriptLedger: self._chunked_turn_count += turn_count self._active_start = self._turns[self._chunked_turn_count - 1]["end"] self._frozen_blocks = list(frozen_blocks) + self.context_phase += 1 # ------------------------------------------------------------------ # Rendering # ------------------------------------------------------------------ - def render(self, tail: list[BaseMessage]) -> list[BaseMessage]: - """Advance compression and return ``S + F + A + tail``. + def compact_active(self) -> bool: + """Scrub old observations only at an explicit context phase boundary. - Order per turn: the occupancy band is read once (from the operator's - last measured prompt), the scrub edge advances over the active window - at that band's screenshot depth, then the chunker runs (a triggered - compression event deep-mutates the frozen region and advances the F/A - boundary). Scrub-before-chunk guarantees that the turns a chunk - closes over — all older than the sliding-window floor — have already - had their screenshots resolved when the relaxed depth is no deeper - than the floor plus one. The returned list is a fresh container: - appending to it (the operator's in-turn tool loop) never mutates the - ledger regions. + Copy messages first: a previously returned model input must remain + immutable even when this ledger starts a new phase. """ + active = self._active + copies = [message.model_copy(deep=True) for message in active] + protected = set() + for old, new in zip(active, copies, strict=True): + if id(old) in self._step_keys: + self._step_keys[id(new)] = self._step_keys.pop(id(old)) + if id(old) in self._fresh_tool_messages: + protected.add(id(new)) self._compressor.compress( - self._active, + copies, image_scrub_depth=self.effective_image_scrub_depth, - protected_message_ids=self._fresh_tool_messages, + protected_message_ids=protected, ) - self._fresh_tool_messages.clear() - if self._chunker is not None: - try: - self._chunker.on_render(self) - except Exception as e: - logger.error(f"History chunker render hook failed: {e}") + self._active = copies + return any( + old.content != new.content or old.additional_kwargs != new.additional_kwargs + for old, new in zip(active, copies, strict=True) + ) + + def _messages(self, tail: list[BaseMessage]) -> list[BaseMessage]: return [ *self._static, *self._restored, @@ -825,3 +875,35 @@ class TranscriptLedger: *self._active[self._active_start :], *tail, ] + + def render(self, tail: list[BaseMessage]) -> list[BaseMessage]: + """Append normally; batch history rewrites at pressure-driven boundaries.""" + before = self.estimate_context_tokens(self._messages(tail)) + self._projected_prompt_tokens = before + phase = self.context_phase + if self._chunker is not None: + try: + self._chunker.on_render(self) + except Exception as e: + logger.error(f"History chunker render hook failed: {e}") + if ( + self.context_phase != phase + or (self._chunker is None and before >= self._context_budget_tokens * self._soft_ratio) + or before >= self._context_budget_tokens * self._hard_ratio + ): + rewritten = self.compact_active() + after = self.estimate_context_tokens(self._messages(tail)) + if self.context_phase != phase or rewritten: + self.context_phase = phase + 1 + self.last_compaction = { + "phase": self.context_phase, + "reason": "context_pressure", + "before_tokens": before, + "after_tokens": after, + "estimated": True, + } + logger.info( + f"Context phase {self.context_phase}: {before} -> {after} tokens (estimated)" + ) + self._fresh_tool_messages.clear() + return self._messages(tail) diff --git a/artemis/resources/config/artemis.jsonc b/artemis/resources/config/artemis.jsonc index 0cc1fbe..ef47fb0 100644 --- a/artemis/resources/config/artemis.jsonc +++ b/artemis/resources/config/artemis.jsonc @@ -96,6 +96,7 @@ "enabled": true, // 历史压缩预算,允许范围 1024–300000;上下文参照上限固定为 300k。 "context_budget_tokens": 80000, + // 阶段内仅追加;35% 开始后台准备摘要,70% 成批压缩,90% 紧急折叠。 "start_ratio": 0.35, "soft_ratio": 0.7, "hard_ratio": 0.9, diff --git a/artemis/services/llm.py b/artemis/services/llm.py index c2194b5..40b8e7d 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -58,6 +58,7 @@ from artemis.llm.reliability import ( retry_policy_for, ) from artemis.llm.messages import adapt_responses_messages +from artemis.llm.prefix_cache import PromptPrefixTracker from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider from artemis.services.token_meter import record_llm_usage from artemis.llm.usage import usage_from_message @@ -671,6 +672,27 @@ class RobustChatModelWrapper: record_llm_usage(engine, response, source=self._endpoint_key()) def _diagnostics(self, args, kwargs): + prefix = None + if self.ctx is not None: + tracker = getattr(self.ctx, "prompt_prefix_tracker", None) + if not isinstance(tracker, PromptPrefixTracker): + tracker = PromptPrefixTracker() + self.ctx.prompt_prefix_tracker = tracker + bound_options = getattr(self.base_model, "kwargs", {}) + options = { + **(bound_options if isinstance(bound_options, dict) else {}), + **{ + key: kwargs[key] + for key in ("tools", "tool_choice", "response_format") + if key in kwargs + }, + } + identity = f"{kwargs.get('prompt_cache_key')}:{self.role}:{self._endpoint_key()}" + prefix = tracker.observe(identity, args[0] if args else kwargs.get("input"), options) + ledger = getattr(self.ctx, "transcript_ledger", None) + if ledger is not None: + prefix["context_phase"] = getattr(ledger, "context_phase", 0) + prefix["last_compaction"] = getattr(ledger, "last_compaction", None) return { "provider": self._provider_value(), "model": self.endpoint.model_name if self.endpoint else None, @@ -680,6 +702,7 @@ class RobustChatModelWrapper: "is_fallback": self.is_fallback, "images": input_images(args, kwargs), "prompt_cache_key": kwargs.get("prompt_cache_key"), + "prefix_cache": prefix, } async def _invoke_attempt(self, *args, **kwargs): diff --git a/config/artemis.jsonc b/config/artemis.jsonc index 0cc1fbe..ef47fb0 100644 --- a/config/artemis.jsonc +++ b/config/artemis.jsonc @@ -96,6 +96,7 @@ "enabled": true, // 历史压缩预算,允许范围 1024–300000;上下文参照上限固定为 300k。 "context_budget_tokens": 80000, + // 阶段内仅追加;35% 开始后台准备摘要,70% 成批压缩,90% 紧急折叠。 "start_ratio": 0.35, "soft_ratio": 0.7, "hard_ratio": 0.9, diff --git a/docs/ai-automation.md b/docs/ai-automation.md index f073a98..160bf7e 100644 --- a/docs/ai-automation.md +++ b/docs/ai-automation.md @@ -82,3 +82,19 @@ Flash 和 Pro 默认复用只读 Checker,以任务提示词和用户指导中 通过 `mobile_config(section="capabilities")` 可读取 `recording` 参数;用户授权后可保存全局默认。单任务用 `runtime_overrides={"recording":{"codec":"h264","computer_scope":"window","max_long_edge":1280,"fps":15}}`,批次在 `task_config.recording` 提供同样对象。JSON 配置的 `agent.recording` 覆盖环境默认;不要把录像参数当作模型截图参数。录屏开关继续独立且默认关闭。 运行信息的上下文参照上限固定为 300,000 tokens,前后端使用相同口径。历史压缩预算默认 80,000,JSON 的 `agent.memory.transcript.context_budget_tokens` 可调但不能超过 300,000;这是压缩触发预算,不是单次生成的输出 token 数,也不是整项任务累计 token 的限额。 + + +### 多轮上下文与缓存诊断 + +Flash、Pro(启用 transcript)和探索共用分阶段上下文。一个阶段内只追加新观察、工具结果和用户指导,保留此前发送的截图、元素列表和消息原文。行动只能使用同一来源的最新观察,旧画面不能证明当前状态。检查器和摘要模型保持独立职责,不混入执行模型的对话。 + +默认压缩预算为 80,000 tokens:估算占用达到 28,000 后开始后台准备摘要,56,000 后等待本批摘要齐备再一次性替换历史、清理旧观察,开始下一阶段;72,000 时可紧急折叠为带步骤索引的历史。摘要失败时原文保留至硬阈值,原始证据仍可经历史工具读取。触发估算包含本轮新消息和截图,不只看上一轮用量;没有上游 usage 时使用本地保守估算。该预算不是任务累计限额,也不改变 300k 上下文参照上限。 + +任务 trace 的 `llm_diagnostic` / `llm_request` 在原有 `usage`、耗时和 `prompt_cache_key` 外增加 `prefix_cache`: + +- `append_only`、`shared_prefix_messages`:相对同任务、角色和端点上次调用,输入是否仅追加、共享多少完整消息。 +- `first_changed_message`:首个变化消息的位置(从 1 开始);`tools_or_format_changed` 表示工具或格式参数发生变化。 +- `context_phase`、`last_compaction`:上下文阶段与上次压缩前后的估算 token 数。 +- `cold_start`:该角色/端点在当前进程内没有比较基准;重启后首次请求会重新建立基准。 + +以上为适配后消息的本地指纹比较,不是网关最终序列化后的 token 前缀,也不是命中率。诊断仅持有哈希,不额外保存提示词、图片正文或凭据。实际缓存率按上游返回的 `cached_tokens / input_tokens` 加权统计,usage 缺失不能当作零命中。首次请求、阶段压缩、模型/工具切换可能降低缓存率;网关的路由、缓存保留时间和实现也会影响结果。该实现使用完整多轮消息,不依赖网关支持 `previous_response_id`,也不承诺固定命中率。 diff --git a/tests/unit/agents/test_flash_runner_ledger.py b/tests/unit/agents/test_flash_runner_ledger.py index 29c6f8f..828ada9 100644 --- a/tests/unit/agents/test_flash_runner_ledger.py +++ b/tests/unit/agents/test_flash_runner_ledger.py @@ -283,13 +283,13 @@ def test_injected_instruction_body_outlives_its_wrapper(mock_context): assert guidance.wrapper.startswith(USER_GUIDANCE_MARKER) assert "don't send messages" not in guidance.wrapper - # Scrub edge parity: once the turn has left the live position (a newer - # committed observation exists) the wrapper is gone and the body is - # still in the committed turn. + # Normal turns preserve guidance verbatim; an explicit compaction removes + # the old wrapper but preserves the user's standing instruction. ledger.stage_turn([tail, AIMessage(content="ok")]) ledger.commit_staged(step_key="s1") ledger.stage_turn([runner._build_tail(ledger, 3, b"IMG", "[1] Send"), AIMessage(content="ok")]) ledger.commit_staged(step_key="s2") + ledger.compact_active() rendered = ledger.render([runner._build_tail(ledger, 4, b"IMG", "[1] Send")]) committed = next( m @@ -679,8 +679,8 @@ async def test_run_calibrates_the_ledger_only_from_measured_usage(mock_context): assert record.call_count == 2 first_call, second_call = record.call_args_list assert first_call.args[0] == 1200 and first_call.kwargs["messages"] is not None - # The report turn carries no usage metadata: estimated, recorded, not calibrated. - assert second_call.args[0] >= 1 and second_call.kwargs["messages"] is None + # Missing usage stays unknown; the next render computes its own estimate. + assert second_call.args[0] is None and second_call.kwargs["messages"] is None @pytest.mark.asyncio diff --git a/tests/unit/agents/test_operator_prompt_estimate.py b/tests/unit/agents/test_operator_prompt_estimate.py index 7367320..30de3fc 100644 --- a/tests/unit/agents/test_operator_prompt_estimate.py +++ b/tests/unit/agents/test_operator_prompt_estimate.py @@ -30,7 +30,7 @@ from artemis.agents.operator.operator import OperatorNode from artemis.config.agent import MemoryTranscriptConfig from artemis.context import ArtemisContext from artemis.memory.step_memory import StepMemoryService -from artemis.memory.transcript import TranscriptLedger, estimate_prompt_tokens +from artemis.memory.transcript import TranscriptLedger SCREENSHOT_B64 = base64.b64encode(b"fake-jpeg-bytes").decode("utf-8") @@ -92,7 +92,7 @@ def _llm(captured: list, usage_metadata=None): @pytest.mark.asyncio -async def test_operator_records_an_estimate_when_the_provider_reports_no_usage(): +async def test_operator_uses_projected_context_without_fabricating_provider_usage(): ctx = _ctx() captured: list = [] with patch("artemis.agents.operator.operator.get_llm", return_value=_llm(captured)): @@ -102,8 +102,8 @@ async def test_operator_records_an_estimate_when_the_provider_reports_no_usage() ledger = ctx.transcript_ledger assert isinstance(ledger, TranscriptLedger) assert captured, "the operator made no model call" - assert ledger.last_prompt_tokens == estimate_prompt_tokens(captured[-1]) - assert ledger.last_prompt_tokens > 0 + assert ledger.last_prompt_tokens is None + assert ledger.projected_prompt_tokens > 0 # No measurement, no calibration. assert ledger.chars_per_token == 4.0 diff --git a/tests/unit/agents/test_operator_transcript.py b/tests/unit/agents/test_operator_transcript.py index 91ebd7b..7b7b281 100644 --- a/tests/unit/agents/test_operator_transcript.py +++ b/tests/unit/agents/test_operator_transcript.py @@ -355,7 +355,7 @@ async def test_transcript_mode_two_turns_build_four_regions(tmp_path): # Text edge at depth 1: the committed observation has lost its plan # recitation as soon as the next tail exists (its screenshot stays # until the screenshot edge). - assert PLAN_RECITATION_MARKER not in str(committed_obs.content) + assert PLAN_RECITATION_MARKER in str(committed_obs.content) # retained until compaction # The live tail still recites the plan. assert PLAN_RECITATION_MARKER in str(turn2_messages[-1].content) # Message layout is S + A + tail. diff --git a/tests/unit/memory/test_history_chunking.py b/tests/unit/memory/test_history_chunking.py index 0ba7175..9c5c0ab 100644 --- a/tests/unit/memory/test_history_chunking.py +++ b/tests/unit/memory/test_history_chunking.py @@ -192,7 +192,17 @@ def _make( capsule_service=capsule, # No meter: the base comes from the ledger's operator-recorded prompt # size (None until recorded), exactly like production wiring. - meter_getter=meter, + meter_getter=meter + or ( + lambda: ( + 75_000 + if chunker.awaiting_chunks + and all( + capsule.get_summary(c.capsule_key) is not None for c in chunker.awaiting_chunks + ) + else 40_000 + ) + ), goal="TEST GOAL", ) ledger.attach_chunker(chunker) @@ -443,7 +453,8 @@ def test_segment_size_threshold_triggers_without_milestone(): assert len(chunker.chunks) >= 1 assert chunker.chunks[0].start_step_number == 1 total = sum(c.end_step_number - c.start_step_number + 1 for c in chunker.chunks) - assert total >= 4 + assert total >= 3 + assert chunker.awaiting_chunks # the next prepared batch has not swapped yet assert len(ledger.unchunked_turns()) >= 2 # floor respected @@ -458,7 +469,7 @@ def test_source_token_threshold_triggers_chunk(): def test_soft_threshold_compresses_oldest_open_segment(): steps = [_step(i) for i in range(1, 9)] - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, _, _ = _make( steps, min_active=2, max_steps=100, meter=lambda: meter["value"], auto_capsule=True ) @@ -843,7 +854,7 @@ def test_hard_threshold_folds_frozen_region_into_recall_only_era_with_step_index merged) plus the per-step minimal index of every folded chunk — step, session offset and action phrase, results only via search_history.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 13)] - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, _, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 8, _hashes(4)) capsule.resolve("chunk:1-4", _capsule(1, 4)) @@ -1083,7 +1094,7 @@ def test_failed_capsule_redispatches_and_retains_original(): """Failure rung of the ladder: an exhausted capsule job is re-dispatched at the next trigger/pressure render while the original turns stay live.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 9)] - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, _, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 8, _hashes(4)) key = chunker.awaiting_chunks[0].capsule_key @@ -1106,7 +1117,7 @@ def test_hard_threshold_force_swaps_pending_chunks_into_recall_only_era(): paragraph degrades to the milestone/step-range list, and ③'s minimal index survives for every folded step.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 9)] - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, _, _ = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 8, _hashes(4)) assert chunker.awaiting_chunks[0].status == "pending" @@ -1150,7 +1161,9 @@ def test_pressure_close_waits_for_min_steps_instead_of_one_turn_chunks(): _run_turns(ledger, chunker, 7, 8) # eligible beyond the claim: 1, then 2 assert [c.capsule_key for c in chunker.awaiting_chunks] == ["chunk:1-3"] _run_turns(ledger, chunker, 9, 9) - assert [c.capsule_key for c in chunker.awaiting_chunks] == ["chunk:1-3", "chunk:4-6"] + assert [c.capsule_key for c in chunker.awaiting_chunks] == [ + "chunk:1-3" + ] # finish this batch first def test_size_trigger_waits_for_min_steps_even_when_one_turn_exceeds_the_tokens(): @@ -1193,7 +1206,7 @@ def test_start_gate_holds_ready_chunk_below_start_ratio_then_swaps(): replaced only once the operator's measured context reaches budget*start_ratio; until then the ready chunk waits in the queue.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 11)] - meter = {"value": 20_000} # < 0.35 * 100k + meter = {"value": 40_000} # < 0.35 * 100k ledger, chunker, engine, _ = _make( steps, min_active=2, meter=lambda: meter["value"], auto_capsule=True ) @@ -1212,13 +1225,13 @@ def test_start_gate_holds_ready_chunk_below_start_ratio_then_swaps(): args = held[0]["payload"]["args"] assert args["note"] == "held" assert (args["context_tokens"], args["context_budget"], args["swap_at_tokens"]) == ( - 20_000, + 40_000, 100_000, - 35_000, + 70_000, ) # Same render cadence, context now past the gate → the ready prefix swaps. - meter["value"] = 40_000 + meter["value"] = 75_000 _run_turns(ledger, chunker, 9, 9, _hashes(4)) assert len(chunker.chunks) == 1 assert chunker.chunks[0].start_step_number == 1 @@ -1226,9 +1239,7 @@ def test_start_gate_holds_ready_chunk_below_start_ratio_then_swaps(): assert ledger.frozen_blocks != () done = engine.traces[-1] assert done["status"] == "success" - assert done["payload"]["args"]["context_tokens"] == 40_000 - ( - chunker.chunks[0].source_chars // 4 - ) + (chunker.chunks[0].summary_chars // 4) + assert done["payload"]["args"]["context_tokens"] < 75_000 def test_leading_stampless_turn_joins_the_next_segment_instead_of_pinning_the_queue(): @@ -1238,13 +1249,14 @@ def test_leading_stampless_turn_joins_the_next_segment_instead_of_pinning_the_qu the head of the awaiting queue, so no L2 swap ever happened and the hard line jumped straight to the fold. It joins the following segment.""" steps = [_step(i) for i in range(1, 13)] - meter = {"value": 50_000} # above the start gate + meter = {"value": 40_000} # above the start gate ledger, chunker, _, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) ledger.stage_turn([_observation(0), AIMessage(content="no tool call this turn")]) _run_turns(ledger, chunker, 1, 7, _hashes(4)) # commits turn 0 without a step key assert [c.capsule_key for c in chunker.awaiting_chunks] == ["chunk:1-4"] assert all(entry["chunk"] is not None for entry in chunker._awaiting) + meter["value"] = 75_000 capsule.resolve("chunk:1-4", _capsule(1, 4)) _run_turns(ledger, chunker, 8, 8, _hashes(4)) @@ -1278,24 +1290,79 @@ def test_partition_folds_leading_unstamped_turns_into_the_first_segment(): ] -def test_start_gate_zero_restores_swap_on_ready(): +def test_start_gate_zero_only_prepares_and_does_not_rewrite_prefix(): steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 9)] ledger, chunker, _, _ = _make( steps, min_active=2, meter=lambda: 1_000, auto_capsule=True, start_ratio=0.0 ) _run_turns(ledger, chunker, 1, 8, _hashes(4)) - assert len(chunker.chunks) == 1 - assert chunker.awaiting_chunks == () + assert chunker.chunks == () + assert chunker.awaiting_chunks[0].status == "ready" -def test_start_gate_unknown_base_falls_through_to_swap_on_ready(): - """No measured operator call (meter None, nothing recorded on the ledger): - the gate cannot decide and the legacy swap-on-ready behavior stands.""" +def test_unknown_usage_uses_current_prompt_estimate_and_preserves_small_prefix(): steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 9)] ledger, chunker, _, _ = _make(steps, min_active=2, auto_capsule=True) - assert ledger.last_prompt_tokens is None + chunker._meter_getter = None _run_turns(ledger, chunker, 1, 8, _hashes(4)) - assert len(chunker.chunks) == 1 + assert ledger.last_prompt_tokens is None + assert ledger.projected_prompt_tokens > 0 + assert chunker.chunks == () + assert ledger.context_phase == 0 + + +def test_phase_batches_async_summaries_then_preserves_the_new_prefix(): + steps = [_step(i) for i in range(1, 20)] + meter = {"value": 10_000} + ledger, chunker, _, capsule = _make(steps, meter=lambda: meter["value"]) + ledger.set_static_prefix([HumanMessage(content="Stable rules")]) + _run_turns(ledger, chunker, 1, 10, lambda i: "a" if i <= 4 else "b" if i <= 7 else "c") + assert capsule.submitted == [] # no background LLM work below preparation gate + tail = [_observation(11)] + original = ledger.render(tail) + fingerprint = [message.model_dump() for message in original] + meter["value"] = 40_000 + ledger.render(tail) + assert len(chunker.awaiting_chunks) == 2 + first, second = chunker.awaiting_chunks + capsule.resolve(first.capsule_key, _capsule(first.start_step_number, first.end_step_number)) + meter["value"] = 75_000 + assert [m.model_dump() for m in ledger.render(tail)] == fingerprint + assert ledger.context_phase == 0 # partial completion must not break the prefix + capsule.resolve(second.capsule_key, _capsule(second.start_step_number, second.end_step_number)) + compacted = ledger.render(tail) + assert ledger.context_phase == 1 + assert ledger.last_compaction["after_tokens"] < ledger.last_compaction["before_tokens"] + assert [m.model_dump() for m in original] == fingerprint # prior input objects never mutated + assert len(chunker.chunks) == 2 and not chunker.awaiting_chunks + + meter["value"] = 10_000 # provider confirms lower occupancy after compaction + # Commit the previous turn, then append a newly observed turn. + ledger.commit_staged(step_key="s10") + next_input = ledger.render(tail) + snapshot = [m.model_dump() for m in next_input] + turn = _turn(11) + turn[0] = tail[0] + ledger.stage_turn(turn) + ledger.commit_staged(step_key="s11") + following = ledger.render([_observation(12)]) + assert [m.model_dump() for m in following[: len(snapshot)]] == snapshot + assert ledger.context_phase == 1 + assert compacted[0].content == "Stable rules" + + +def test_new_large_tail_triggers_pressure_without_waiting_for_next_usage(): + steps = [_step(i) for i in range(1, 12)] + ledger, chunker, _, _ = _make(steps, auto_capsule=True) + chunker._meter_getter = None + previous = _run_turns(ledger, chunker, 1, 9, _hashes(4)) + ledger.record_prompt_tokens(10_000, previous) + assert ledger.context_phase == 0 + # New evidence already exceeds the hard line, despite a small last call. + ledger.render([HumanMessage(content="新证据" * 40_000)]) + assert ledger.projected_prompt_tokens > 90_000 + assert ledger.context_phase == 1 + assert ledger.frozen_blocks def _big_turn(i: int, chars: int = 4000) -> list: @@ -1323,7 +1390,7 @@ def test_hard_threshold_swaps_ready_prefix_as_l2_when_that_is_enough(): estimated context back under the hard line, the swap is an ordinary L2 event — no fold, and the younger pending segment stays awaiting.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 13)] - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, engine, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) # Steps 1–4 (hash-a) close at the switch; steps 5–8 (hash-b) close at the # next switch. Turns are ~1k tokens each so the L2 swap is worth ~4k. @@ -1352,7 +1419,7 @@ def test_hard_threshold_escalates_to_fold_when_l2_swap_is_not_enough(): and the whole frozen region folds into one recall-only era whose index lists every folded step — the pending chunk's steps too.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 13)] - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, engine, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 10, lambda i: "a" if i <= 4 else ("b" if i <= 8 else "c")) capsule.resolve("chunk:1-4", _capsule(1, 4)) @@ -1394,9 +1461,10 @@ def test_hard_fold_is_monotonic_later_l2_swaps_append_after_the_folded_era(): flips back to full chunk blocks.""" steps = [_step(i) for i in range(1, 17)] hash_for = lambda i: "a" if i <= 4 else ("b" if i <= 8 else ("c" if i <= 12 else "d")) # noqa: E731 - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, _, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 10, hash_for) + meter["value"] = 75_000 capsule.resolve("chunk:1-4", _capsule(1, 4)) meter["value"] = 99_000 _run_turns(ledger, chunker, 11, 11, hash_for) # hard: folds 1–4 and 5–8 @@ -1405,10 +1473,11 @@ def test_hard_fold_is_monotonic_later_l2_swaps_append_after_the_folded_era(): # Pressure gone (above the start gate, below soft): the next milestone # switch closes 9–12; once ready it swaps as an ordinary L2 chunk. - meter["value"] = 50_000 + meter["value"] = 40_000 _run_turns(ledger, chunker, 12, 15, hash_for) pending = [c for c in chunker.awaiting_chunks if c.status == "pending"] assert [c.capsule_key for c in pending] == ["chunk:9-12"] + meter["value"] = 75_000 capsule.resolve("chunk:9-12", _capsule(9, 12)) _run_turns(ledger, chunker, 16, 16, hash_for) @@ -1430,11 +1499,13 @@ def test_hard_threshold_with_nothing_awaiting_folds_frozen_region_once(): into a recall-only era without consuming turns, and repeated hard renders leave the frozen blocks byte-identical.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 13)] - meter = {"value": 50_000} + meter = {"value": 40_000} ledger, chunker, _, _ = _make( steps, min_active=2, meter=lambda: meter["value"], auto_capsule=True ) - _run_turns(ledger, chunker, 1, 7, _hashes(4)) + _run_turns(ledger, chunker, 1, 6, _hashes(4)) + meter["value"] = 75_000 + _run_turns(ledger, chunker, 7, 7, _hashes(4)) # Chunk 1–4 swapped as L2 at the step-7 render; s5..s6 remain = the floor. assert [c.capsule_key for c in chunker.chunks] == ["chunk:1-4"] assert chunker.awaiting_chunks == () @@ -1479,14 +1550,15 @@ def test_context_base_reads_the_ledger_operator_measurement_not_the_session_mete # Some other agent's small call lands on the session meter. get_meter("session-1").record({"prompt_tokens": 2_700, "completion_tokens": 10}) # The operator's own measured prompt is well below the gate. - ledger.record_prompt_tokens(20_000) - assert chunker._context_base_tokens(ledger) == 20_000 + chunker._meter_getter = None + ledger.record_prompt_tokens(40_000) + assert chunker._context_base_tokens(ledger) == 40_000 _run_turns(ledger, chunker, 1, 8, _hashes(4)) assert chunker.chunks == () # held: 20k < 35k assert [c.status for c in chunker.awaiting_chunks] == ["ready"] # Operator prompt grows past the gate → swap at the next render. - ledger.record_prompt_tokens(36_000) + ledger.record_prompt_tokens(75_000) _run_turns(ledger, chunker, 9, 9, _hashes(4)) assert len(chunker.chunks) == 1 @@ -1507,7 +1579,7 @@ def test_held_note_is_fresh_after_a_retry(): then readied while the start gate still holds it gets a fresh ``held`` note: the re-dispatch resets the once-per-chunk held marker.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 14)] - meter = {"value": 20_000} # < 0.35 * 100k: the gate holds ready chunks + meter = {"value": 40_000} # < 0.35 * 100k: the gate holds ready chunks ledger, chunker, engine, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 8, _hashes(4)) entry = chunker._awaiting[0] @@ -1797,7 +1869,7 @@ def test_calibrated_ratio_scales_the_size_trigger(): # Default ratio: the eligible text (two ~4k-char turns once turn 5 is # staged) is worth ~chars/4 tokens, under the 5k target → nothing closes. ledger, chunker, _, _ = _make( - steps, min_active=2, min_steps=2, max_steps=100, target_tokens=5000 + steps, min_active=2, min_steps=2, max_steps=100, target_tokens=5000, meter=lambda: 40_000 ) ledger.record_prompt_tokens(1000, messages=[HumanMessage(content="a" * 4000)]) _run_big_turns(ledger, chunker, 1, 5, lambda i: "hash-a") @@ -1807,7 +1879,7 @@ def test_calibrated_ratio_scales_the_size_trigger(): # Calibrated to 1 char/token: the same text is worth ~chars tokens → closes. ledger, chunker, _, _ = _make( - steps, min_active=2, min_steps=2, max_steps=100, target_tokens=5000 + steps, min_active=2, min_steps=2, max_steps=100, target_tokens=5000, meter=lambda: 40_000 ) ledger.record_prompt_tokens(4000, messages=[HumanMessage(content="中" * 4000)]) assert ledger.chars_per_token == pytest.approx(1.0) @@ -1840,6 +1912,7 @@ def test_compression_announces_one_tool_line_from_dispatch_to_swap(): assert args["source_tokens"] > 0 assert "context_tokens" not in args + meter["value"] = 75_000 capsule.resolve("chunk:1-4", _capsule(1, 4)) ledger.commit_staged(step_key="s8", validator_result={"status": "dispatched"}) ledger.render([_observation(9)]) @@ -1853,16 +1926,14 @@ def test_compression_announces_one_tool_line_from_dispatch_to_swap(): assert done_args["summary_tokens"] > 0 assert done_args["context_budget"] == 100_000 assert done_args["context_estimated"] is True - assert done_args["context_tokens"] == ( - 60_000 - done_args["source_tokens"] + done_args["summary_tokens"] - ) + assert 0 < done_args["context_tokens"] < 75_000 assert "forced" not in done_args def test_forced_swap_announces_fold_without_a_ratio(): """Forced folds report estimated context usage without a summary size.""" steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 9)] - meter = {"value": None} + meter = {"value": 40_000} ledger, chunker, engine, _ = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 8, _hashes(4)) assert any(t["status"] == "running" for t in engine.traces) @@ -1896,7 +1967,7 @@ def test_compression_trace_carries_a_plain_language_phase_on_every_path(): assert COMPRESSION_PHASES == ("summarizing", "ready", "applied", "failed") steps = [_step(i, "hash-a" if i <= 4 else "hash-b") for i in range(1, 15)] - meter = {"value": 20_000} # < 0.35 * 100k: the start gate holds ready chunks + meter = {"value": 40_000} # < 0.35 * 100k: the start gate holds ready chunks ledger, chunker, engine, capsule = _make(steps, min_active=2, meter=lambda: meter["value"]) _run_turns(ledger, chunker, 1, 8, _hashes(4)) chunk = chunker.awaiting_chunks[0] @@ -1933,9 +2004,14 @@ def test_compression_trace_carries_a_plain_language_phase_on_every_path(): assert held["payload"]["args"]["note"] == "held" # Context past the gate → the swap applies the summary. - meter["value"] = 40_000 + for pending in chunker.awaiting_chunks: + capsule.resolve( + pending.capsule_key, _capsule(pending.start_step_number, pending.end_step_number) + ) + meter["value"] = 75_000 _run_turns(ledger, chunker, 13, 13, lambda i: "hash-c") - assert [c.start_step_number for c in chunker.chunks] == [1] + assert chunker.chunks[0].start_step_number == 1 + assert not chunker.awaiting_chunks assert mine()[-1] == ("success", "applied") # Every announcement on every trace carries a known phase. diff --git a/tests/unit/memory/test_transcript_ledger.py b/tests/unit/memory/test_transcript_ledger.py index db48ded..dcec926 100644 --- a/tests/unit/memory/test_transcript_ledger.py +++ b/tests/unit/memory/test_transcript_ledger.py @@ -71,9 +71,19 @@ def _turn(i: int, tool_call_count: int = 1) -> list: return messages -def _play_turn(ledger: TranscriptLedger, i: int, *, prev_key=None, prev_result=None, **turn_kwargs): +def _play_turn( + ledger: TranscriptLedger, + i: int, + *, + prev_key=None, + prev_result=None, + compact=False, + **turn_kwargs, +): """Mimic the operator's real ordering: commit previous, render, stage.""" ledger.commit_staged(step_key=prev_key, validator_result=prev_result) + if compact: + ledger.compact_active() turn = _turn(i, **turn_kwargs) rendered = ledger.render([turn[0]]) ledger.stage_turn(turn) @@ -159,8 +169,8 @@ def test_recitation_and_ui_list_stripped_at_the_text_edge_and_screenshot_at_k(): until depth K and is then resolved in place; only the live tail ever carries an indexed element list.""" ledger = TranscriptLedger(step_memory=_service(), image_scrub_depth=3) - _play_turn(ledger, 1) - rendered = _play_turn(ledger, 2, prev_key="step-1") + _play_turn(ledger, 1, compact=True) + rendered = _play_turn(ledger, 2, compact=True, prev_key="step-1") obs1 = ledger.active_messages[0] # Turn 1 is now depth 2: text edge passed, screenshot kept. assert PLAN_RECITATION_MARKER not in str(obs1.content) @@ -168,10 +178,10 @@ def test_recitation_and_ui_list_stripped_at_the_text_edge_and_screenshot_at_k(): assert _has_image(obs1) original_obs1 = _fingerprint(obs1) - rendered = _play_turn(ledger, 3, prev_key="step-2") + rendered = _play_turn(ledger, 3, compact=True, prev_key="step-2") # Turn 1 reached depth 3: the screenshot is resolved. - assert not _has_image(obs1) - assert _fingerprint(obs1) != original_obs1 + assert not _has_image(ledger.active_messages[0]) + assert _fingerprint(ledger.active_messages[0]) != original_obs1 # Turn 2 (depth 2): text edge passed, screenshot still there; the live tail keeps both. obs2 = next(m for m in ledger.active_messages if "OBSERVATION [T+00:02]" in str(m.content)) assert PLAN_RECITATION_MARKER not in str(obs2.content) @@ -189,7 +199,7 @@ def test_depth_k_image_resolved_to_ready_visual_summary(): ledger = TranscriptLedger(step_memory=service, image_scrub_depth=3) for i in range(1, 5): - _play_turn(ledger, i, prev_key=f"step-{i - 1}" if i > 1 else None) + _play_turn(ledger, i, compact=True, prev_key=f"step-{i - 1}" if i > 1 else None) first_obs = ledger.active_messages[0] assert not _has_image(first_obs) @@ -212,7 +222,7 @@ def test_pending_grace_then_placeholder_never_backfilled(): ledger = TranscriptLedger(step_memory=service, image_scrub_depth=2, pending_grace_steps=1) # Enough turns to push turn 1 past K + grace. for i in range(1, 7): - _play_turn(ledger, i, prev_key=f"step-{i - 1}" if i > 1 else None) + _play_turn(ledger, i, compact=True, prev_key=f"step-{i - 1}" if i > 1 else None) first_obs = ledger.active_messages[0] assert not _has_image(first_obs) @@ -221,7 +231,7 @@ def test_pending_grace_then_placeholder_never_backfilled(): # A late summary must never mutate the frozen message. service._summaries["step-1"] = "Too late." - _play_turn(ledger, 7, prev_key="step-6") + _play_turn(ledger, 7, compact=True, prev_key="step-6") assert _fingerprint(ledger.active_messages[0]) == frozen @@ -232,7 +242,7 @@ def test_failed_summary_becomes_unavailable_placeholder(): ledger = TranscriptLedger(step_memory=service, image_scrub_depth=2, pending_grace_steps=5) for i in range(1, 4): - _play_turn(ledger, i, prev_key=f"step-{i - 1}" if i > 1 else None) + _play_turn(ledger, i, compact=True, prev_key=f"step-{i - 1}" if i > 1 else None) first_obs = ledger.active_messages[0] assert not _has_image(first_obs) @@ -473,7 +483,7 @@ def test_relaxed_depth_below_start_gate_keeps_more_screenshots(): assert ledger.occupancy is None assert ledger.effective_image_scrub_depth == 6 for i in range(1, 10): - _play_turn(ledger, i, prev_key=f"step-{i - 1}" if i > 1 else None) + _play_turn(ledger, i, compact=True, prev_key=f"step-{i - 1}" if i > 1 else None) # At the render of turn 9 its observation is the live tail (depth 1); # committed turns 8..5 are depths 2..5 and turn 4 (depth 6) is scrubbed. ledger.record_prompt_tokens(20_000) # 20% < 35% @@ -482,34 +492,34 @@ def test_relaxed_depth_below_start_gate_keeps_more_screenshots(): tight = TranscriptLedger(step_memory=_service(), image_scrub_depth=3) for i in range(1, 10): - _play_turn(tight, i, prev_key=f"step-{i - 1}" if i > 1 else None) + _play_turn(tight, i, compact=True, prev_key=f"step-{i - 1}" if i > 1 else None) assert _image_turns(tight) == [8] def test_crossing_the_start_gate_tightens_once_and_relaxing_never_backfills(): ledger = _banded_ledger() for i in range(1, 10): - _play_turn(ledger, i, prev_key=f"step-{i - 1}" if i > 1 else None) + _play_turn(ledger, i, compact=True, prev_key=f"step-{i - 1}" if i > 1 else None) assert _image_turns(ledger) == [5, 6, 7, 8] # Past the gate: the next render scrubs the extra depths in one pass # (turns 5–8 were depths 3–6 at that render; only turn 9 at depth 2 stays). ledger.record_prompt_tokens(40_000) # 40% >= 35% assert ledger.effective_image_scrub_depth == 3 - _play_turn(ledger, 10, prev_key="step-9") + _play_turn(ledger, 10, compact=True, prev_key="step-9") assert _image_turns(ledger) == [9] # Back under the gate: nothing is restored; new turns accumulate again # until four historical screenshots are in view. ledger.record_prompt_tokens(20_000) assert ledger.effective_image_scrub_depth == 6 - _play_turn(ledger, 11, prev_key="step-10") + _play_turn(ledger, 11, compact=True, prev_key="step-10") assert _image_turns(ledger) == [9, 10] - _play_turn(ledger, 12, prev_key="step-11") + _play_turn(ledger, 12, compact=True, prev_key="step-11") assert _image_turns(ledger) == [9, 10, 11] - _play_turn(ledger, 13, prev_key="step-12") + _play_turn(ledger, 13, compact=True, prev_key="step-12") assert _image_turns(ledger) == [9, 10, 11, 12] - _play_turn(ledger, 14, prev_key="step-13") # turn 9 reaches depth 6 + _play_turn(ledger, 14, compact=True, prev_key="step-13") # turn 9 reaches depth 6 assert _image_turns(ledger) == [10, 11, 12, 13] @@ -523,10 +533,8 @@ def test_no_band_configuration_keeps_the_fixed_depth(): assert relaxed_only.effective_image_scrub_depth == 3 # no gate → never relaxed -def test_scrub_edge_runs_before_the_chunker_at_render(): - """The chunker sees the active window after the scrub edge advanced, so a - chunk closing over turns older than the floor never captures a live - screenshot in its source transcript.""" +def test_chunker_sees_immutable_raw_history_before_a_phase_boundary(): + """Preparing capsules must not scrub any previously sent observation.""" ledger = _banded_ledger() seen: list[list[int]] = [] @@ -537,9 +545,9 @@ def test_scrub_edge_runs_before_the_chunker_at_render(): ledger.attach_chunker(Probe()) for i in range(1, 9): _play_turn(ledger, i, prev_key=f"step-{i - 1}" if i > 1 else None) - # At the render of turn 8 the scrub edge (depth 6) had already resolved - # turn 3 (depth 6); the probe saw turns 4–7 with images, never turn 3. - assert seen[-1] == [4, 5, 6, 7] + # A render below pressure preserves every historical image. + assert seen[-1] == list(range(1, 8)) + assert ledger.context_phase == 0 # --------------------------------------------------------------------------- @@ -580,6 +588,7 @@ def test_ephemeral_blocks_vanish_at_the_text_edge_and_not_before(): ledger.stage_turn(turn_with_notice(1)) for i in range(2, 4): ledger.commit_staged(step_key=f"step-{i - 1}") + ledger.compact_active() ledger.render([_observation(i)]) ledger.stage_turn(turn_with_notice(i)) @@ -600,15 +609,12 @@ def test_ephemeral_blocks_vanish_at_the_text_edge_and_not_before(): def test_rendering_past_the_edge_is_byte_identical_and_prefix_stable(): - """(iii) A render with nothing new is byte-identical to the previous one, - and across turns exactly one message — the observation reaching depth K — - differs from the previous render's prefix, so the request prefix before - it is what the provider's prompt cache already holds.""" + """Every full previous prompt, including its live tail, survives unchanged.""" ledger = TranscriptLedger(step_memory=_ready_service(12), image_scrub_depth=3) ledger.set_static_prefix([SystemMessage(content="STATIC")]) def snapshot(rendered: list) -> list[str]: - return [_fingerprint(m) for m in rendered[:-1]] # everything but the live tail + return [_fingerprint(m) for m in rendered] # includes the previously live observation previous: list[str] = [] for i in range(1, 11): @@ -620,24 +626,15 @@ def test_rendering_past_the_edge_is_byte_identical_and_prefix_stable(): ) current = snapshot(rendered) changed = [idx for idx, fp in enumerate(previous) if current[idx] != fp] - if i >= 3: - # The live tail is depth 1 and committed turn i-1 depth 2, so the - # turn reaching depth 3 is i-2; its observation heads its span. - expected = next( - idx - for idx, m in enumerate(rendered[:-1]) - if f"OBSERVATION [T+00:0{(i - 2) % 10}]" in str(m.content) - ) - assert changed == [expected], (i, changed) - else: - assert changed == [], (i, changed) + assert changed == [], (i, changed) previous = current # Same state, rendered again: byte-identical, including the tail. + ledger.commit_staged(step_key="step-10", validator_result={"status": "dispatched"}) tail = [_observation(11)] again = [_fingerprint(m) for m in ledger.render(tail)] assert again == [_fingerprint(m) for m in ledger.render(tail)] - assert again[:-1] == previous + assert again[: len(previous)] == previous # --------------------------------------------------------------------------- diff --git a/tests/unit/test_memory_config.py b/tests/unit/test_memory_config.py index a4b9fe2..3be6a0f 100644 --- a/tests/unit/test_memory_config.py +++ b/tests/unit/test_memory_config.py @@ -134,6 +134,8 @@ def test_xml_scrub_depth_positions_the_text_edge(): ledger.stage_turn([obs(1)]) ledger.commit_staged(step_key="s1") ledger.render([obs(2)]) + assert PRO_UI_LIST_MARKER in str(ledger.active_messages[0].content) + ledger.compact_active() # depth settings apply at a compaction boundary committed = ledger.active_messages[0] assert PRO_UI_LIST_MARKER not in str(committed.content) assert any(isinstance(b, dict) and b.get("type") == "image_url" for b in committed.content) diff --git a/tests/unit/test_plugin_result_delivery.py b/tests/unit/test_plugin_result_delivery.py index 312d079..0a057d8 100644 --- a/tests/unit/test_plugin_result_delivery.py +++ b/tests/unit/test_plugin_result_delivery.py @@ -65,7 +65,9 @@ def test_burst_results_reach_responses_once_with_images_before_compression(tmp_p actions[2].pop("post_image_name") actions[2]["attempts"] = ["Skipped"] report = {"status": "failed", "execution": actions} - ledger = TranscriptLedger(image_scrub_depth=1, pending_grace_steps=0) + ledger = TranscriptLedger( + image_scrub_depth=1, pending_grace_steps=0, context_budget_tokens=300_000 + ) _stage(ledger, actions) ledger.commit_staged( step_key="step1", @@ -84,13 +86,17 @@ def test_burst_results_reach_responses_once_with_images_before_compression(tmp_p assert len([item for item in payload if item.get("type") == "function_call_output"]) == 4 assert str(payload).count(base64.b64encode(image.read_bytes()).decode()) == 3 assert f"data:{Image.MIME[image_format]};base64," in str(payload) - # No duplicate outputs at the following turn; old images are eligible for scrubbing again. + # No duplicate outputs or rewritten images on the following ordinary turn. ledger.commit_staged( validator_result=report, tool_results=plugin_result_messages(report, engine) ) later = ledger.render([]) assert len([m for m in later if isinstance(m, ToolMessage)]) == 4 + assert str(later).count(base64.b64encode(image.read_bytes()).decode()) == 3 + ledger.compact_active() + later = ledger.render([]) assert str(later).count(base64.b64encode(image.read_bytes()).decode()) < 3 + assert str(messages).count(base64.b64encode(image.read_bytes()).decode()) == 3 def test_missing_plugin_screenshot_keeps_real_result_and_reports_missing_evidence(tmp_path): diff --git a/tests/unit/test_prefix_cache.py b/tests/unit/test_prefix_cache.py new file mode 100644 index 0000000..0640380 --- /dev/null +++ b/tests/unit/test_prefix_cache.py @@ -0,0 +1,63 @@ +import json +from types import SimpleNamespace + +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage + +from artemis.llm.prefix_cache import PromptPrefixTracker +from artemis.llm.router import ModelEndpoint +from artemis.services.llm import RobustChatModelWrapper + + +def test_prefix_diagnostics_distinguish_append_rewrite_and_tool_changes_without_payloads(): + tracker = PromptPrefixTracker() + messages = [SystemMessage(content="private rules"), HumanMessage(content="private screenshot")] + options = {"tools": [{"name": "click"}, {"name": "observe"}]} + assert tracker.observe("task:flash", messages, options)["cold_start"] + messages.extend( + [ + AIMessage(content="", tool_calls=[{"name": "click", "args": {"x": 5}, "id": "c1"}]), + ToolMessage(content="clicked", tool_call_id="c1"), + ] + ) + extended = tracker.observe("task:flash", messages, options) + assert extended["append_only"] and extended["shared_prefix_messages"] == 2 + messages[1] = HumanMessage(content="compressed summary") + rewritten = tracker.observe("task:flash", messages, options) + assert rewritten["first_changed_message"] == 2 and not rewritten["append_only"] + reordered = tracker.observe("task:flash", messages, {"tools": list(reversed(options["tools"]))}) + assert reordered["tools_or_format_changed"] and not reordered["append_only"] + assert tracker.observe("task:checker", messages, options)["cold_start"] + assert "private" not in json.dumps(tracker._previous) + + +def test_bound_wrappers_reuse_task_prefix_tracker_after_message_adaptation(): + class Model: + kwargs = {"tools": [{"type": "function", "name": "observe"}]} + + def bind_tools(self, *_args, **_kwargs): + return self + + ctx = SimpleNamespace(data_engine=SimpleNamespace(current_session_id="session")) + wrapper = RobustChatModelWrapper(Model(), ctx=ctx, endpoint=ModelEndpoint(), role="flash") + messages = [SystemMessage(content="rules"), HumanMessage(content="step 1")] + + def diagnostics(items): + bound = wrapper.bind_tools([]) # Flash binds again for each call + args, kwargs = bound._prepare_model_call((items,), {}) + return bound._diagnostics(args, kwargs) + + first = diagnostics(messages) + messages.append(HumanMessage(content="step 2")) + second = diagnostics(messages) + assert first["prefix_cache"]["cold_start"] + assert second["prefix_cache"]["append_only"] + assert first["prompt_cache_key"] == second["prompt_cache_key"] + assert not any("rules" in str(value) for value in ctx.prompt_prefix_tracker._previous.values()) + + ctx.transcript_ledger = SimpleNamespace( + context_phase=1, last_compaction={"reason": "context_pressure"} + ) + third = diagnostics(messages[:1] + [HumanMessage(content="history summary")]) + assert third["prefix_cache"]["context_phase"] == 1 + assert third["prefix_cache"]["last_compaction"]["reason"] == "context_pressure" + assert not third["prefix_cache"]["append_only"]