diff --git a/docs/configuration.md b/docs/configuration.md index 435f54428..53f570f7e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -2268,16 +2268,12 @@ When a user is idle for longer than a configured threshold, nanobot **proactivel How it works: 1. **Idle detection**: On each idle tick (~1 s), checks whether an idle-session scan is due. By default, the full scan runs at most once per minute. -2. **Background compaction**: Idle sessions summarize the older live prefix via LLM and keep the most recent legal suffix (currently 8 messages). -3. **Summary injection**: When the user returns, the summary is injected as runtime context (one-shot, not persisted) alongside the retained recent suffix. -4. **Restart-safe resume**: The summary is also mirrored into session metadata so it can still be recovered after a process restart. +2. **Background compaction**: Older context is summarized while the most recent messages remain available. +3. **Session preservation**: The complete session history remains stored for later inspection and reuse. +4. **Restart-safe resume**: The compacted context remains available after a process restart. > [!NOTE] -> Mental model: "summarize older context, keep the freshest live turns, **and overwrite the session file with the compact form.**" It is not a full `session.clear()`, but it is a write — not a soft cursor move. -> -> Concretely, auto compact rewrites `sessions/.jsonl` in place: older messages (including their structured `tool_calls` / `tool_call_id` / `reasoning_content`) are replaced by just the retained recent suffix (currently 8 messages), while the archived prefix is preserved only as a plain-text summary appended to `memory/history.jsonl` (or a `[RAW] ...` flattened dump if LLM summarization fails). The original structured JSON of those turns is no longer recoverable from the session file. -> -> This differs from the **token-driven soft consolidation** that fires when a prompt exceeds the context budget: that path only advances an internal `last_consolidated` cursor and leaves the session file untouched, so the raw tool-call trail stays on disk and can still be replayed or audited. If you rely on that trail for debugging or auditing, set `idleCompactAfterMinutes` to `0` and let only the token-driven path run. +> Auto compact shortens the context sent to the model without deleting the session's structured message history. ## Timezone diff --git a/nanobot/agent/context.py b/nanobot/agent/context.py index 0c8f604ce..5d49f25d1 100644 --- a/nanobot/agent/context.py +++ b/nanobot/agent/context.py @@ -30,11 +30,7 @@ from nanobot.security.workspace_access import WorkspaceScopeResolver from nanobot.session.keys import last_channel_from_metadata from nanobot.session.manager import Session from nanobot.session.summary import SessionSummary -from nanobot.utils.helpers import ( - detect_image_mime, - load_bundled_template, - truncate_text_to_tokens, -) +from nanobot.utils.helpers import detect_image_mime, load_bundled_template from nanobot.utils.prompt_templates import render_template @@ -98,8 +94,6 @@ class ContextBuilder: BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md"] _SKIPPABLE_DEFAULTS = {"AGENTS.md", "USER.md"} _RUNTIME_CONTEXT_TAG = RUNTIME_CONTEXT_TAG - _MAX_RECENT_HISTORY = 50 - _MAX_HISTORY_TOKENS = 8_000 # hard cap on recent history section size (tokens) _RUNTIME_CONTEXT_END = RUNTIME_CONTEXT_END def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None): @@ -115,9 +109,6 @@ class ContextBuilder: session_summary: SessionSummary | None = None, workspace: Path | None = None, include_memory: bool = True, - include_memory_recent_history: bool = True, - session_key: str | None = None, - unified_session: bool = False, ) -> str: """Build the system prompt from identity, bootstrap files, memory, and skills.""" root = workspace or self.workspace @@ -155,29 +146,6 @@ class ContextBuilder: if skills_summary: parts.append(render_template("agent/skills_section.md", skills_summary=skills_summary)) - if include_memory_recent_history: - entries = self.memory.read_recent_history_for_prompt( - since_cursor=self.memory.get_last_dream_cursor(), - session_key=session_key, - unified_session=unified_session, - ) - if entries: - capped = entries[-self._MAX_RECENT_HISTORY:] - capped = self._without_duplicate_session_summary( - capped, - session_key=session_key, - session_summary=session_summary, - ) - if capped: - history_text = "\n".join( - f"- [{e['timestamp']}] {e['content']}" for e in capped - ) - history_text = truncate_text_to_tokens( - history_text, - self._MAX_HISTORY_TOKENS, - ) - parts.append("# Recent History\n\n" + history_text) - if session_summary: parts.append( "[Archived Context Summary]\n\n" @@ -187,25 +155,6 @@ class ContextBuilder: return "\n\n---\n\n".join(parts) - @staticmethod - def _without_duplicate_session_summary( - entries: list[dict[str, Any]], - *, - session_key: str | None, - session_summary: SessionSummary | None, - ) -> list[dict[str, Any]]: - """Drop the history entry already represented by the session summary.""" - if not session_summary: - return entries - for index in range(len(entries) - 1, -1, -1): - entry = entries[index] - if ( - entry.get("session_key") == session_key - and entry.get("content") == session_summary["text"] - ): - return [*entries[:index], *entries[index + 1:]] - return entries - def _get_identity(self, channel: str | None = None, workspace: Path | None = None) -> str: """Get the core identity section.""" root = workspace or self.workspace @@ -295,9 +244,6 @@ class ContextBuilder: runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None, workspace: Path | None = None, include_memory: bool = True, - include_memory_recent_history: bool = True, - session_key: str | None = None, - unified_session: bool = False, ) -> list[dict[str, Any]]: """Compatibility wrapper for callers that need merged adjacent roles.""" messages = self.build_transcript( @@ -312,9 +258,6 @@ class ContextBuilder: channel=channel, workspace=workspace, include_memory=include_memory, - include_memory_recent_history=include_memory_recent_history, - session_key=session_key, - unified_session=unified_session, ) current = messages[-1] if len(messages) < 2 or messages[-2].get("role") != current.get("role"): @@ -339,9 +282,6 @@ class ContextBuilder: channel: str | None = None, workspace: Path | None = None, include_memory: bool = True, - include_memory_recent_history: bool = True, - session_key: str | None = None, - unified_session: bool = False, ) -> list[dict[str, Any]]: """Build a model transcript while preserving the fresh-turn boundary.""" root = workspace or self.workspace @@ -353,9 +293,6 @@ class ContextBuilder: session_summary=transcript.session_summary, workspace=root, include_memory=include_memory, - include_memory_recent_history=include_memory_recent_history, - session_key=session_key, - unified_session=unified_session, ), }, *transcript.history, diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 9243504e0..d86b440a7 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -444,7 +444,6 @@ class AgentLoop: workspace_scopes=self.workspace_scopes, unified_session=unified_session, ), - unified_session=unified_session, ) self.auto_compact = AutoCompact( sessions=self.sessions, @@ -1109,9 +1108,6 @@ class AgentLoop: channel=request_ctx.channel, workspace=effective_scope.project_path, include_memory=session.policy.persist if session is not None else True, - include_memory_recent_history=not ephemeral, - session_key=session.key if session is not None else request_ctx.session_key, - unified_session=self._unified_session, ) if request_context is None: request_ctx = dataclasses.replace( @@ -1883,6 +1879,13 @@ class AgentLoop: session, runtime=runtime, ) + # Token consolidation may have committed a replacement checkpoint + # after the compact stage captured its summary for this request. + ctx.session, ctx.pending_summary = self.auto_compact.prepare_session( + session, + ctx.session_key, + ) + session = ctx.require_session() is_subagent = ctx.kind is TurnKind.SYSTEM and ctx.msg.sender_id == "subagent" _hist_kwargs: dict[str, Any] = { diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py index 97679c034..779a24a27 100644 --- a/nanobot/agent/memory.py +++ b/nanobot/agent/memory.py @@ -35,6 +35,7 @@ from nanobot.utils.helpers import ( estimate_prompt_tokens_chain, strip_think, truncate_text, + truncate_text_to_tokens, ) from nanobot.utils.prompt_templates import render_template from nanobot.utils.workspace_prompts import ( @@ -65,8 +66,6 @@ class MemoryStore: # durable files are tiny in practice (~5 KB total), but a runaway file must # not unbounded the prompt. _DREAM_FILE_EMBED_CAP = 8000 - _INTERNAL_HISTORY_SESSION_PREFIXES = ("cron:", "dream:") - _INTERNAL_HISTORY_SESSION_KEYS = {"heartbeat"} _LEGACY_ENTRY_START_RE = re.compile(r"^\[(\d{4}-\d{2}-\d{2}[^\]]*)\]\s*") _LEGACY_TIMESTAMP_RE = re.compile(r"^\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2})\]\s*") _LEGACY_RAW_MESSAGE_RE = re.compile( @@ -260,6 +259,29 @@ class MemoryStore: # -- history.jsonl — append-only, JSONL format --------------------------- + def _normalize_history_entry( + self, + entry: str, + *, + max_chars: int | None = None, + ) -> str: + """Return the exact bounded, model-safe text accepted by the journal.""" + limit = max_chars if max_chars is not None else _HISTORY_ENTRY_HARD_CAP + raw = entry.rstrip() + content = strip_think(raw) + if len(content) > limit: + if not self._oversize_logged: + self._oversize_logged = True + logger.warning( + "history entry exceeds {} chars ({}); truncating. " + "Usually means a caller forgot its own cap; " + "further occurrences suppressed.", + limit, + len(content), + ) + content = truncate_text(content, limit) + return content + def append_history( self, entry: str, @@ -274,27 +296,16 @@ class MemoryStore: persisted. If the cleaned content is empty but the raw entry wasn't, the record is persisted with an empty string rather than falling back to the raw leak — otherwise `strip_think`'s guarantees would be - undone by history replay / consolidation downstream. + undone when Dream consumes the journal entry. A defensive cap (*max_chars*, default ``_HISTORY_ENTRY_HARD_CAP``) is applied as a final safety net: individual callers should cap their own content more tightly; this default only exists to catch unintentional large writes (e.g. an LLM echoing its input back as a "summary"). """ - limit = max_chars if max_chars is not None else _HISTORY_ENTRY_HARD_CAP ts = datetime.now().strftime("%Y-%m-%d %H:%M") raw = entry.rstrip() - if len(raw) > limit: - if not self._oversize_logged: - self._oversize_logged = True - logger.warning( - "history entry exceeds {} chars ({}); truncating. " - "Usually means a caller forgot its own cap; " - "further occurrences suppressed.", - limit, len(raw), - ) - raw = truncate_text(raw, limit) - content = strip_think(raw) + content = self._normalize_history_entry(entry, max_chars=max_chars) # Cursor allocation and the append must be atomic: concurrent writers # could otherwise read the same current cursor and emit duplicates. with self._append_lock: @@ -302,7 +313,7 @@ class MemoryStore: if raw and not content: logger.debug( "history entry {} stripped to empty (likely template leak); " - "persisting empty content to avoid re-polluting context", + "persisting empty content to avoid re-polluting Dream input", cursor, ) record = {"cursor": cursor, "timestamp": ts, "content": content} @@ -392,36 +403,6 @@ class MemoryStore: """Return history entries with a valid cursor > *since_cursor*.""" return [e for e, c in self._iter_valid_entries() if c > since_cursor] - @classmethod - def _is_internal_history_session(cls, session_key: str | None) -> bool: - if not session_key: - return False - return ( - session_key in cls._INTERNAL_HISTORY_SESSION_KEYS - or session_key.startswith(cls._INTERNAL_HISTORY_SESSION_PREFIXES) - ) - - def read_recent_history_for_prompt( - self, - since_cursor: int, - *, - session_key: str | None, - unified_session: bool = False, - ) -> list[dict[str, Any]]: - """Return unprocessed history entries safe to inject into a turn prompt.""" - entries = self.read_unprocessed_history(since_cursor=since_cursor) - if session_key is None: - return entries - if not unified_session: - return [e for e in entries if e.get("session_key") == session_key] - - return [ - entry - for entry in entries - if (entry_session := entry.get("session_key")) == session_key - or not self._is_internal_history_session(entry_session) - ] - def compact_history(self) -> None: """Drop oldest processed entries without discarding pending Dream input.""" if self.max_history_entries <= 0: @@ -718,21 +699,28 @@ class MemoryStore: *, max_chars: int | None = None, session_key: str | None = None, - ) -> None: - """Fallback: dump raw messages to history.jsonl without LLM summarization.""" - limit = max_chars if max_chars is not None else _RAW_ARCHIVE_MAX_CHARS - formatted = truncate_text( - self._format_messages(public_history_messages(messages)), - limit, - ) - self.append_history( - f"[RAW] {len(messages)} messages\n" - f"{formatted}", - session_key=session_key, - ) + ) -> str: + """Persist and return a bounded raw checkpoint when summarization degrades.""" + checkpoint = self._build_raw_checkpoint(messages, max_chars=max_chars) + self.append_history(checkpoint, session_key=session_key) logger.warning( "Memory consolidation degraded: raw-archived {} messages", len(messages) ) + return checkpoint + + def _build_raw_checkpoint( + self, + messages: list[dict[str, Any]], + *, + max_chars: int | None = None, + ) -> str: + """Build the same bounded checkpoint as :meth:`raw_archive` without writing it.""" + limit = max_chars if max_chars is not None else _RAW_ARCHIVE_MAX_CHARS + checkpoint = ( + f"[RAW] {len(messages)} messages\n" + f"{self._format_messages(public_history_messages(messages))}" + ) + return self._normalize_history_entry(checkpoint, max_chars=limit) # ------------------------------------------------------------------ # Dream helpers @@ -787,12 +775,11 @@ class MemoryStore: # Memory ingestion and legacy context-pressure coordination # --------------------------------------------------------------------------- -# Individual history.jsonl writers cap their own payloads tightly; the -# _HISTORY_ENTRY_HARD_CAP at append_history() is a belt-and-suspenders default -# that catches any new caller that forgot to set its own cap. -_RAW_ARCHIVE_MAX_CHARS = 16_000 # fallback dump (LLM failed) -_ARCHIVE_SUMMARY_MAX_CHARS = 8_000 # LLM-produced consolidation summary -_HISTORY_ENTRY_HARD_CAP = 64_000 # emergency cap in append_history +# Raw fallbacks use a tighter cap. Completed model summaries may scale with the +# configured generation budget, while append_history() still enforces the +# emergency hard cap against pathological provider output. +_RAW_ARCHIVE_MAX_CHARS = 16_000 # fallback dump (LLM failed) +_HISTORY_ENTRY_HARD_CAP = 64_000 # emergency cap in append_history class MemoryArchiver: @@ -809,13 +796,45 @@ class MemoryArchiver: build_messages: Callable[..., list[dict[str, Any]]], get_tool_definitions: Callable[[], list[dict[str, Any]]], resolve_prompt_context: Callable[[Session], tuple[str | None, Path | None]] | None = None, - unified_session: bool = False, ) -> None: self.store = store self._build_messages = build_messages self._get_tool_definitions = get_tool_definitions self._resolve_prompt_context = resolve_prompt_context - self.unified_session = unified_session + + def _raw_checkpoint( + self, + messages: list[dict[str, Any]], + *, + session_key: str, + previous_summary: str | None, + max_tokens: int, + ) -> str: + """Persist the failed chunk and return a bounded replacement checkpoint.""" + raw = self.store.raw_archive(messages, session_key=session_key) + token_limit = max(1, max_tokens) + if not previous_summary: + return truncate_text_to_tokens(raw, token_limit) + + combined = ( + "[Previous archived context]\n" + f"{previous_summary}\n\n" + "[Newly archived raw context]\n" + f"{raw}" + ) + bounded = truncate_text_to_tokens(combined, token_limit) + if bounded == combined: + return combined + + # Keep evidence from both sides when their full concatenation cannot fit. + section_limit = max(1, (token_limit - 32) // 2) + return truncate_text_to_tokens( + "[Previous archived context]\n" + f"{truncate_text_to_tokens(previous_summary, section_limit)}\n\n" + "[Newly archived raw context]\n" + f"{truncate_text_to_tokens(raw, section_limit)}", + token_limit, + ) async def archive( self, @@ -825,48 +844,53 @@ class MemoryArchiver: session_key: str, request_messages: list[dict[str, Any]], request_tools: list[dict[str, Any]], + previous_summary: str | None = None, ) -> str | None: """Execute a prepared archive request and persist its result.""" if not messages: return None + + def raw_fallback() -> str: + return self._raw_checkpoint( + messages, + session_key=session_key, + previous_summary=previous_summary, + max_tokens=runtime.generation.max_tokens, + ) + try: with llm_usage_source("dream"): response = await runtime.provider.chat_with_retry( model=runtime.model, messages=request_messages, tools=request_tools, - tool_choice="none", temperature=runtime.generation.temperature, max_tokens=runtime.generation.max_tokens, reasoning_effort=runtime.generation.reasoning_effort, ) except Exception: logger.warning("Memory archive provider call failed, raw-dumping to history") - self.store.raw_archive(messages, session_key=session_key) - return None + return raw_fallback() if response.finish_reason in {"error", "length"}: logger.warning( "Memory archive provider did not complete ({}), raw-dumping to history", response.finish_reason, ) - self.store.raw_archive(messages, session_key=session_key) - return None + return raw_fallback() if response.has_tool_calls is True: logger.warning("Memory archive provider returned tool calls, raw-dumping to history") - self.store.raw_archive(messages, session_key=session_key) - return None + return raw_fallback() summary = response.content if not summary or not summary.strip(): logger.warning("Memory archive provider returned no summary, raw-dumping to history") - self.store.raw_archive(messages, session_key=session_key) - return None - if summary.strip() == "(nothing)": + return raw_fallback() + summary = self.store._normalize_history_entry(summary) + if not summary: + logger.warning("Memory archive provider summary was not safe to replay, raw-dumping") + return raw_fallback() + if summary == "(nothing)": return "(nothing)" - self.store.append_history( - summary, - max_chars=_ARCHIVE_SUMMARY_MAX_CHARS, - session_key=session_key, - ) + self.store.append_history(summary, session_key=session_key) return summary async def archive_session( @@ -881,13 +905,26 @@ class MemoryArchiver: messages = list(session.messages[session.last_archived:archive_end]) if not messages: return None + session_summary = session_summary_from_metadata( + session.metadata, + fallback_last_active=session.updated_at, + ) + previous_summary = session_summary["text"] if session_summary else None + + def raw_fallback() -> str: + return self._raw_checkpoint( + messages, + session_key=session.key, + previous_summary=previous_summary, + max_tokens=runtime.generation.max_tokens, + ) + if input_token_budget <= 0: logger.debug( "Memory archive has no safe input budget for {}; raw-dumping", session.key, ) - self.store.raw_archive(messages, session_key=session.key) - return None + return raw_fallback() prefix = Session( key=session.key, messages=list(session.messages[:archive_end]), @@ -903,13 +940,8 @@ class MemoryArchiver: "Memory archive cannot replay the full chunk for {}; raw-dumping", session.key, ) - self.store.raw_archive(messages, session_key=session.key) - return None - prompt = render_template( - "agent/consolidator_archive.md", - strip=True, - archive_count=len(archive_history), - ) + return raw_fallback() + prompt = render_template("agent/consolidator_archive.md", strip=True) channel = session.key.split(":", 1)[0] if ":" in session.key else None workspace: Path | None = None if self._resolve_prompt_context is not None: @@ -918,13 +950,8 @@ class MemoryArchiver: history=history, current_message=prompt, channel=channel, - session_summary=session_summary_from_metadata( - session.metadata, - fallback_last_active=session.updated_at, - ), + session_summary=session_summary, workspace=workspace, - session_key=session.key, - unified_session=self.unified_session, ) tools = self._get_tool_definitions() estimated, source = estimate_prompt_tokens_chain( @@ -941,14 +968,14 @@ class MemoryArchiver: input_token_budget, source, ) - self.store.raw_archive(messages, session_key=session.key) - return None + return raw_fallback() return await self.archive( messages, runtime=runtime, session_key=session.key, request_messages=request_messages, request_tools=tools, + previous_summary=previous_summary, ) @@ -964,20 +991,16 @@ class Consolidator: build_messages: Callable[..., list[dict[str, Any]]], get_tool_definitions: Callable[[], list[dict[str, Any]]], resolve_prompt_context: Callable[[Session], tuple[str | None, Path | None]] | None = None, - unified_session: bool = False, ): self.store = store self.sessions = sessions - self.unified_session = unified_session self._build_messages = build_messages self._get_tool_definitions = get_tool_definitions - self._resolve_prompt_context = resolve_prompt_context self.archiver = MemoryArchiver( store=store, build_messages=build_messages, get_tool_definitions=get_tool_definitions, resolve_prompt_context=resolve_prompt_context, - unified_session=unified_session, ) self._locks: weakref.WeakValueDictionary[str, asyncio.Lock] = ( weakref.WeakValueDictionary() @@ -1013,13 +1036,18 @@ class Consolidator: return [] return session.get_history() - def _persist_last_summary(self, session: Session, summary: str | None) -> None: - if summary and summary != "(nothing)": + @staticmethod + def _set_last_summary( + session: Session, + summary: str, + *, + last_active: datetime | None = None, + ) -> None: + if summary != "(nothing)": session.metadata["_last_summary"] = { "text": summary, - "last_active": session.updated_at.isoformat(), + "last_active": (last_active or session.updated_at).isoformat(), } - self.sessions.save(session) def estimate_session_prompt_tokens( self, @@ -1039,8 +1067,6 @@ class Consolidator: current_message="[token-probe]", channel=channel, session_summary=summary, - session_key=session.key, - unified_session=self.unified_session, ) return estimate_prompt_tokens_chain( runtime.provider, @@ -1057,24 +1083,6 @@ class Consolidator: - self._SAFETY_BUFFER ) - async def archive( - self, - messages: list[dict[str, Any]], - *, - runtime: LLMRuntime, - session_key: str, - request_messages: list[dict[str, Any]], - request_tools: list[dict[str, Any]], - ) -> str | None: - """Compatibility wrapper for the extracted MemoryArchiver.""" - return await self.archiver.archive( - messages, - runtime=runtime, - session_key=session_key, - request_messages=request_messages, - request_tools=request_tools, - ) - async def archive_session( self, session: Session, @@ -1101,26 +1109,23 @@ class Consolidator: The budget reserves space for completion tokens and a safety buffer so the LLM request never exceeds the context window. """ - if runtime.context_window_tokens <= 0: - return - lock = self.get_lock(session.key) async with lock: # Refresh session reference: AutoCompact may have replaced it. fresh = self.sessions.get_or_create(session.key) if fresh is not session: session = fresh + if runtime.context_window_tokens <= 0: + return if not session.messages: return budget = self._input_token_budget(runtime) - last_summary: str | None = None estimated, source = self.estimate_session_prompt_tokens( session, runtime=runtime, ) if estimated <= 0: - self._persist_last_summary(session, last_summary) return if estimated < budget: unarchived_count = len(session.messages) - session.last_archived @@ -1132,7 +1137,6 @@ class Consolidator: source, unarchived_count, ) - self._persist_last_summary(session, last_summary) return end_idx = self.pick_consolidation_boundary(session) @@ -1160,18 +1164,12 @@ class Consolidator: archive_end=end_idx, runtime=runtime, ) - # Advance either way: archive_session raw-archives on degradation, - # and replaying the same chunk would duplicate Memory material. - if summary: - last_summary = summary + if summary is None: + return + self._set_last_summary(session, summary) session.last_archived = end_idx self.sessions.save(session) - # Persist the last summary to session metadata so it can be injected - # into the runtime context on the next prepare_session() call, aligning - # the summary injection strategy with AutoCompact._archive(). - self._persist_last_summary(session, last_summary) - async def compact_idle_session( self, session_key: str, @@ -1209,12 +1207,10 @@ class Consolidator: archive_end=archive_end, runtime=runtime, ) + if summary is None: + return None - if summary and summary != "(nothing)": - session.metadata["_last_summary"] = { - "text": summary, - "last_active": last_active.isoformat(), - } + self._set_last_summary(session, summary, last_active=last_active) # A turn can append while the provider call is in flight. Advance only # through the captured batch so new messages remain eligible next time. diff --git a/nanobot/templates/agent/consolidator_archive.md b/nanobot/templates/agent/consolidator_archive.md index d3dffd157..3987762dc 100644 --- a/nanobot/templates/agent/consolidator_archive.md +++ b/nanobot/templates/agent/consolidator_archive.md @@ -1,27 +1,42 @@ -Create a memory overview for only the final {{ archive_count }} conversation messages immediately before this instruction. Earlier messages are context for resolving references; do not summarize them again. +Create a compact replacement checkpoint for this session. -Use [skip] unless a fact meets all SNIP criteria: -- Signal: would the user need to repeat this if forgotten? -- Novel: not just a restatement of another fact in this same conversation chunk -- Important: prevents rework or captures preferences / rules -- Persistent: still relevant after 2 weeks +When `[Archived Context Summary]` appears in the system prompt, update that previous checkpoint to reflect the current conversation state. -Also preserve a compact working-state handoff even when it is not Persistent: the active objective, current status, completed steps, unresolved blockers, next action, and exact identifiers needed to continue without rework. Mark these facts [ephemeral]. +## Merge rules -Format each fact as: -- [mark] fact content +- Use the latest correction or decision as the current version of a fact, and merge duplicates. +- Preserve exact names, identifiers, paths, commands, decisions, results, and unresolved blockers when they are needed to continue the session. +- Retain a fact already present in long-term memory when it is needed for session continuity. -Marks (choose the best match): -- [permanent] Core preferences, personal traits, habits — never becomes stale -- [durable] Technical discoveries, project knowledge, config details — valid for months -- [ephemeral] Active task state, temporary decisions — may change in weeks -- [correction] Correction to a previous memory — state what changed -- [skip] Conversational filler, code/source facts derivable from the repo, or audit-only breadcrumbs +## What to retain -Priority: user corrections and preferences > solutions > decisions > events > environment facts. +Always retain a compact working-state handoff: +- active objective +- current status +- completed results that constrain later work +- unresolved blockers +- next action +- exact identifiers needed for that action -Do not output facts already present in the system prompt's Recent History. +Mark working-state facts `[ephemeral]`. -Do not mark something [skip] merely because it might already exist in long-term memory. +For other facts, retain a candidate only when it meets all four SNIP criteria: +- Signal: remembering it saves the user from repeating it +- Novel: it adds a distinct fact to this checkpoint +- Important: losing it would cause rework or discard a preference or rule +- Persistent: it is expected to remain useful for at least two weeks -Return only formatted fact lines, or `(nothing)` if nothing noteworthy happened. +Assign each retained fact its best current mark: +- `[permanent]` for core preferences, personal traits, and habits that remain relevant indefinitely +- `[durable]` for technical discoveries, project knowledge, and configuration that remains valid for months +- `[ephemeral]` for active task state and temporary decisions that may change within weeks +- `[correction]` for the current fact that supersedes conflicting earlier long-term memory + +When space is limited, prioritize user corrections and preferences, then solutions, decisions, events, and environment facts. + +## Output + +Return one concise retained fact per line in this form: +- [mark] fact + +Use `(nothing)` when no fact qualifies and there is no active working state. diff --git a/tests/agent/test_auto_compact.py b/tests/agent/test_auto_compact.py index ef0a0f683..96e2e97ff 100644 --- a/tests/agent/test_auto_compact.py +++ b/tests/agent/test_auto_compact.py @@ -1302,9 +1302,9 @@ class TestSummaryPersistence: assert "_last_summary" in reloaded.metadata # Simulate /new command - session.clear() - loop.sessions.save(session) - loop.sessions.invalidate(session.key) + reloaded.clear() + loop.sessions.save(reloaded) + loop.sessions.invalidate(reloaded.key) # After /new, metadata should no longer contain _last_summary fresh = loop.sessions.get_or_create("cli:test") diff --git a/tests/agent/test_consolidator.py b/tests/agent/test_consolidator.py index 0f57973a9..712dc2568 100644 --- a/tests/agent/test_consolidator.py +++ b/tests/agent/test_consolidator.py @@ -1,4 +1,4 @@ -"""Tests for the lightweight Consolidator — append-only to HISTORY.md.""" +"""Tests for Memory checkpoint consolidation and history journaling.""" from dataclasses import replace from unittest.mock import AsyncMock, MagicMock @@ -6,7 +6,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest from nanobot.agent.memory import ( - _ARCHIVE_SUMMARY_MAX_CHARS, + _HISTORY_ENTRY_HARD_CAP, Consolidator, MemoryStore, ) @@ -26,6 +26,8 @@ from nanobot.session.manager import Session from nanobot.utils.llm_runtime import LLMRuntime from nanobot.utils.prompt_templates import render_template +_ARCHIVE_PROMPT = render_template("agent/consolidator_archive.md", strip=True) + @pytest.fixture def store(tmp_path): @@ -98,8 +100,15 @@ def _build_test_messages(**kwargs): ] -async def _archive(consolidator, messages, runtime, *, session_key="test:session"): - return await consolidator.archive( +async def _archive( + consolidator, + messages, + runtime, + *, + session_key="test:session", + previous_summary=None, +): + return await consolidator.archiver.archive( messages, runtime=runtime, session_key=session_key, @@ -108,6 +117,7 @@ async def _archive(consolidator, messages, runtime, *, session_key="test:session current_message="consolidate", ), request_tools=[], + previous_summary=previous_summary, ) @@ -201,7 +211,9 @@ class TestConsolidatorSummarize: mock_provider.chat_with_retry.side_effect = Exception("API error") messages = [{"role": "user", "content": "hello"}] result = await _archive(consolidator, messages, runtime) - assert result is None # no summary on raw dump fallback + assert result is not None + assert "[RAW]" in result + assert "hello" in result entries = store.read_unprocessed_history(since_cursor=0) assert len(entries) == 1 assert "[RAW]" in entries[0]["content"] @@ -226,23 +238,51 @@ class TestConsolidatorSummarize: entries = store.read_unprocessed_history(since_cursor=0) assert entries[0]["session_key"] == "slack:chat-2" + async def test_raw_fallback_represents_previous_checkpoint_and_new_chunk( + self, + consolidator, + mock_provider, + runtime, + ): + runtime = replace(runtime, generation=GenerationSettings(max_tokens=96)) + mock_provider.chat_with_retry.side_effect = RuntimeError("API error") + + result = await _archive( + consolidator, + [{"role": "user", "content": "NEW_MARKER " + "new " * 200}], + runtime, + previous_summary="OLD_MARKER " + "old " * 200, + ) + + assert result is not None + assert "[Previous archived context]" in result + assert "OLD_MARKER" in result + assert "[Newly archived raw context]" in result + assert "NEW_MARKER" in result + assert "... (truncated)" in result + async def test_summarize_skips_empty_messages(self, consolidator, runtime): result = await _archive(consolidator, [], runtime) assert result is None class TestConsolidatorPromptContract: - def test_archive_prompt_preserves_working_state_with_memory_facts(self): - prompt = render_template("agent/consolidator_archive.md", strip=True, archive_count=4) + def test_archive_prompt_requests_a_cumulative_replacement_checkpoint(self): + prompt = _ARCHIVE_PROMPT + for section in ("## Merge rules", "## What to retain", "## Output"): + assert section in prompt + assert "replacement checkpoint" in prompt + assert "[Archived Context Summary]" in prompt + assert "current conversation state" in prompt assert "SNIP" in prompt - assert "final 4 conversation messages" in prompt - for mark in ("[permanent]", "[durable]", "[ephemeral]", "[correction]", "[skip]"): + for mark in ("[permanent]", "[durable]", "[ephemeral]", "[correction]"): assert mark in prompt assert "working-state handoff" in prompt - assert "exact identifiers needed to continue without rework" in prompt - assert "Do not output facts already present in the system prompt's Recent History" in prompt - assert "Do not mark something [skip] merely because it might already exist" in prompt + assert "- [mark] fact" in prompt + assert "[skip]" not in prompt + assert "(nothing)" in prompt + assert "history.jsonl" not in prompt class TestConsolidatorArchiveErrorHandling: @@ -272,7 +312,8 @@ class TestConsolidatorArchiveErrorHandling: {"role": "assistant", "content": "Done, fixed the race condition."}, ] result = await _archive(consolidator, messages, runtime) - assert result is None + assert result is not None + assert "[RAW]" in result entries = store.read_unprocessed_history(since_cursor=0) assert len(entries) == 1 assert "[RAW]" in entries[0]["content"] @@ -436,9 +477,9 @@ class TestConsolidatorTokenBudget: assert [message["content"] for message in request["messages"][1:-1]] == [ f"m{i}" for i in range(50) ] - assert "final 50 conversation messages" in request["messages"][-1]["content"] + assert request["messages"][-1]["content"] == _ARCHIVE_PROMPT assert request["tools"] == [] - assert request["tool_choice"] == "none" + assert "tool_choice" not in request assert session.last_archived == 50 assert session.provider_state == _provider_state() @@ -460,8 +501,7 @@ class TestConsolidatorTokenBudget: consolidator.estimate_session_prompt_tokens = MagicMock( side_effect=[(1200, "tiktoken"), (400, "tiktoken")] ) - # LLM consolidation fails after raw_archive fires. - consolidator.archive_session = AsyncMock(return_value=None) + consolidator.archive_session = AsyncMock(return_value="[RAW] checkpoint") await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime) @@ -491,7 +531,7 @@ class TestConsolidatorTokenBudget: consolidator.estimate_session_prompt_tokens = MagicMock( return_value=(1200, "tiktoken") ) - consolidator.archive_session = AsyncMock(return_value=None) + consolidator.archive_session = AsyncMock(return_value="[RAW] checkpoint") await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime) @@ -613,27 +653,62 @@ class TestCompactIdleSession: assert reloaded.last_archived == 2 assert [message["content"] for message in reloaded.get_history()] == ["hello", "hi"] + @pytest.mark.asyncio + async def test_idle_compaction_with_no_new_messages_is_noop( + self, real_consolidator, mock_provider, store, runtime + ): + sessions = real_consolidator.sessions + session = sessions.get_or_create("cli:archived-idle") + session.add_message("user", "already archived") + session.add_message("assistant", "old answer") + session.last_archived = 2 + sessions.save(session) + sessions.invalidate("cli:archived-idle") + + result = await real_consolidator.compact_idle_session( + "cli:archived-idle", + runtime=runtime, + ) + + assert result == "" + mock_provider.chat_with_retry.assert_not_awaited() + reloaded = sessions.get_or_create("cli:archived-idle") + assert reloaded.last_archived == 2 + assert "_last_summary" not in reloaded.metadata + assert store.read_unprocessed_history(since_cursor=0) == [] + @pytest.mark.asyncio async def test_new_messages_advance_existing_archive_progress( self, real_consolidator, mock_provider, runtime ): - mock_provider.chat_with_retry.return_value = MagicMock( - content="Summary.", finish_reason="stop" - ) + mock_provider.chat_with_retry.side_effect = [ + MagicMock(content="First replacement checkpoint.", finish_reason="stop"), + MagicMock(content="Second replacement checkpoint.", finish_reason="stop"), + ] sessions = real_consolidator.sessions session = sessions.get_or_create("cli:incremental") session.add_message("user", "first user") session.add_message("assistant", "first assistant") sessions.save(session) - await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime) + first = await real_consolidator.compact_idle_session( + "cli:incremental", + runtime=runtime, + ) current = sessions.get_or_create("cli:incremental") current.add_message("user", "second user") current.add_message("assistant", "second assistant") sessions.save(current) - await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime) + second = await real_consolidator.compact_idle_session( + "cli:incremental", + runtime=runtime, + ) + assert first == "First replacement checkpoint." + assert second == "Second replacement checkpoint." assert mock_provider.chat_with_retry.await_count == 2 + latest_build = real_consolidator.archiver._build_messages.call_args_list[-1].kwargs + assert latest_build["session_summary"]["text"] == "First replacement checkpoint." latest_messages = mock_provider.chat_with_retry.await_args_list[-1].kwargs["messages"] assert [message["content"] for message in latest_messages[1:5]] == [ "first user", @@ -641,8 +716,91 @@ class TestCompactIdleSession: "second user", "second assistant", ] - assert "final 2 conversation messages" in latest_messages[-1]["content"] - assert sessions.get_or_create("cli:incremental").last_archived == 4 + assert latest_messages[-1]["content"] == _ARCHIVE_PROMPT + sessions.invalidate("cli:incremental") + reloaded = sessions.get_or_create("cli:incremental") + assert reloaded.last_archived == 4 + assert reloaded.metadata["_last_summary"]["text"] == second + + @pytest.mark.asyncio + async def test_raw_fallback_preserves_previous_checkpoint_and_new_chunk( + self, + real_consolidator, + mock_provider, + store, + runtime, + ): + mock_provider.chat_with_retry.side_effect = [ + LLMResponse(content="Earlier durable checkpoint.", finish_reason="stop"), + RuntimeError("LLM unavailable"), + ] + sessions = real_consolidator.sessions + session = sessions.get_or_create("cli:cumulative-fallback") + session.add_message("user", "first user") + session.add_message("assistant", "first answer") + sessions.save(session) + + await real_consolidator.compact_idle_session( + "cli:cumulative-fallback", + runtime=runtime, + ) + current = sessions.get_or_create("cli:cumulative-fallback") + current.add_message("user", "second user") + current.add_message("assistant", "newest working state") + sessions.save(current) + + fallback = await real_consolidator.compact_idle_session( + "cli:cumulative-fallback", + runtime=runtime, + ) + + assert fallback is not None + assert "[Previous archived context]" in fallback + assert "Earlier durable checkpoint." in fallback + assert "[Newly archived raw context]" in fallback + assert "newest working state" in fallback + entries = store.read_unprocessed_history(0) + assert entries[0]["content"] == "Earlier durable checkpoint." + assert entries[1]["content"].startswith("[RAW] 2 messages") + sessions.invalidate("cli:cumulative-fallback") + reloaded = sessions.get_or_create("cli:cumulative-fallback") + assert reloaded.metadata["_last_summary"]["text"] == fallback + + @pytest.mark.asyncio + async def test_nothing_keeps_previous_replacement_checkpoint( + self, + real_consolidator, + mock_provider, + runtime, + ): + mock_provider.chat_with_retry.side_effect = [ + LLMResponse(content="Existing checkpoint.", finish_reason="stop"), + LLMResponse(content="(nothing)", finish_reason="stop"), + ] + sessions = real_consolidator.sessions + session = sessions.get_or_create("cli:nothing-after-summary") + session.add_message("user", "important first turn") + session.add_message("assistant", "important result") + sessions.save(session) + await real_consolidator.compact_idle_session( + "cli:nothing-after-summary", + runtime=runtime, + ) + + current = sessions.get_or_create("cli:nothing-after-summary") + current.add_message("user", "thanks") + current.add_message("assistant", "you're welcome") + sessions.save(current) + result = await real_consolidator.compact_idle_session( + "cli:nothing-after-summary", + runtime=runtime, + ) + + assert result == "(nothing)" + sessions.invalidate("cli:nothing-after-summary") + reloaded = sessions.get_or_create("cli:nothing-after-summary") + assert reloaded.last_archived == 4 + assert reloaded.metadata["_last_summary"]["text"] == "Existing checkpoint." @pytest.mark.asyncio async def test_concurrent_append_remains_unarchived( @@ -792,11 +950,16 @@ class TestCompactIdleSession: result = await real_consolidator.compact_idle_session( "cli:nothing", runtime=runtime, max_suffix=4 ) + second = await real_consolidator.compact_idle_session( + "cli:nothing", runtime=runtime, max_suffix=4 + ) assert result == "(nothing)" + assert second == "" reloaded = sessions.get_or_create("cli:nothing") assert "_last_summary" not in reloaded.metadata assert real_consolidator.store.read_unprocessed_history(0) == [] + mock_provider.chat_with_retry.assert_awaited_once() @pytest.mark.asyncio async def test_llm_failure_preserves_history_but_advances_replay_boundary( @@ -813,7 +976,8 @@ class TestCompactIdleSession: result = await real_consolidator.compact_idle_session( "cli:fail", runtime=runtime, max_suffix=4 ) - assert result is None + assert result is not None + assert "[RAW]" in result # raw_archive should have been called (history.jsonl gets an entry) entries = store.read_unprocessed_history(since_cursor=0) @@ -823,6 +987,7 @@ class TestCompactIdleSession: assert len(reloaded.messages) == 20 assert reloaded.messages[0]["content"] == "u0" assert reloaded.last_archived == 20 + assert reloaded.metadata["_last_summary"]["text"] == result assert [m["content"] for m in reloaded.get_history(max_messages=20)] == [ "u6", "a6", @@ -863,11 +1028,10 @@ class TestCompactIdleSession: archived_call = mock_provider.chat_with_retry.call_args sent_messages = archived_call.kwargs["messages"] sent_content = [message.get("content") for message in sent_messages] - # The ordinary replay prefix contributes recent context, while the - # temporary instruction limits the new overview to the unarchived tail. + # The replacement overview covers all model-visible conversation context. assert "u0" not in sent_content assert "u26" in sent_content - assert "final 10 conversation messages" in sent_messages[-1]["content"] + assert sent_messages[-1]["content"] == _ARCHIVE_PROMPT @pytest.mark.asyncio async def test_full_archive_keeps_extended_legal_replay_suffix( @@ -956,9 +1120,9 @@ class TestCompactIdleSession: "user", ] assert sent_messages[2]["tool_calls"][0]["id"] == "call-1" - assert "final 4 conversation messages" in sent_messages[-1]["content"] + assert sent_messages[-1]["content"] == _ARCHIVE_PROMPT assert call["tools"] == tools - assert call["tool_choice"] == "none" + assert "tool_choice" not in call reloaded = sessions.get_or_create("cli:tool-history") assert len(reloaded.messages) == 4 @@ -996,7 +1160,8 @@ class TestCompactIdleSession: runtime=runtime, ) - assert result is None + assert result is not None + assert "[RAW]" in result entries = store.read_unprocessed_history(since_cursor=0) assert len(entries) == 1 assert entries[0]["content"].startswith("[RAW] ") @@ -1026,7 +1191,8 @@ class TestCompactIdleSession: runtime=runtime, ) - assert result is None + assert result is not None + assert "[RAW]" in result entries = store.read_unprocessed_history(since_cursor=0) assert len(entries) == 1 assert entries[0]["content"].startswith("[RAW] ") @@ -1052,7 +1218,8 @@ class TestCompactIdleSession: runtime=runtime, ) - assert result is None + assert result is not None + assert "[RAW]" in result mock_provider.chat_with_retry.assert_not_awaited() entries = store.read_unprocessed_history(since_cursor=0) assert len(entries) == 1 @@ -1060,7 +1227,7 @@ class TestCompactIdleSession: assert sessions.get_or_create("sdk:oversized").last_archived == 1 @pytest.mark.asyncio - async def test_incremental_scope_counts_only_model_visible_messages( + async def test_archive_context_contains_only_model_visible_messages( self, real_consolidator, mock_provider, @@ -1093,7 +1260,7 @@ class TestCompactIdleSession: "new user", "new answer", ] - assert "final 2 conversation messages" in sent[-1]["content"] + assert sent[-1]["content"] == _ARCHIVE_PROMPT @pytest.mark.asyncio async def test_reuses_real_prefix_for_unified_session_workspace( @@ -1126,8 +1293,6 @@ class TestCompactIdleSession: current_message="next project question", channel="websocket", workspace=project, - session_key=session.key, - unified_session=True, ) await loop.consolidator.compact_idle_session( @@ -1137,7 +1302,7 @@ class TestCompactIdleSession: sent_messages = runtime.provider.chat_with_retry.call_args.kwargs["messages"] assert sent_messages[:-1] == ordinary_messages[:-1] - assert "final 2 conversation messages" in sent_messages[-1]["content"] + assert sent_messages[-1]["content"] == _ARCHIVE_PROMPT system = sent_messages[0]["content"] assert "PROJECT_WORKSPACE_MARKER" in system assert "GLOBAL_WORKSPACE_MARKER" not in system @@ -1307,6 +1472,21 @@ class TestRawArchiveTruncation: assert len(entries) == 1 assert "hello" in entries[0]["content"] + def test_raw_archive_returns_the_sanitized_persisted_checkpoint(self, store): + messages = [ + { + "role": "user", + "content": "PRIVATE_REASONINGvisible result", + } + ] + + checkpoint = store.raw_archive(messages, session_key="cli:test") + + persisted = store.read_unprocessed_history(since_cursor=0)[0]["content"] + assert checkpoint == persisted + assert "PRIVATE_REASONING" not in checkpoint + assert "visible result" in checkpoint + def test_raw_archive_excludes_model_only_runtime_context(self, store): content, marker = append_runtime_context( "ship the feature", @@ -1338,21 +1518,40 @@ class TestRawArchiveTruncation: class TestArchivePersistence: - async def test_oversized_summary_is_capped_before_append( + async def test_archive_returns_the_sanitized_persisted_summary( + self, consolidator, mock_provider, store, runtime + ): + mock_provider.chat_with_retry.return_value = MagicMock( + content="PRIVATE_REASONINGsafe summary", + finish_reason="stop", + has_tool_calls=False, + ) + + summary = await _archive( + consolidator, + [{"role": "user", "content": "hi"}], + runtime, + ) + + persisted = store.read_unprocessed_history(since_cursor=0)[0]["content"] + assert summary == persisted == "safe summary" + + async def test_oversized_summary_uses_history_emergency_cap( self, consolidator, mock_provider, store, runtime ): """A pathologically large LLM summary must not land full-length in history.jsonl — that would re-open the #3412 bloat vector from the *success* path instead of the fallback path.""" mock_provider.chat_with_retry.return_value = MagicMock( - content="S" * (_ARCHIVE_SUMMARY_MAX_CHARS * 10), + content="S" * (_HISTORY_ENTRY_HARD_CAP * 2), finish_reason="stop", ) - await _archive( + summary = await _archive( consolidator, [{"role": "user", "content": "hi"}], runtime, ) entry = store.read_unprocessed_history(since_cursor=0)[0] - assert len(entry["content"]) <= _ARCHIVE_SUMMARY_MAX_CHARS + 50 + assert len(entry["content"]) <= _HISTORY_ENTRY_HARD_CAP + 50 + assert summary == entry["content"] diff --git a/tests/agent/test_context_builder.py b/tests/agent/test_context_builder.py index 55c014ba8..b9114f968 100644 --- a/tests/agent/test_context_builder.py +++ b/tests/agent/test_context_builder.py @@ -133,10 +133,7 @@ class TestLoadBootstrapFiles: (project / "SOUL.md").write_text("project soul collision", encoding="utf-8") (project / "USER.md").write_text("project user collision", encoding="utf-8") - result = ContextBuilder(agent_home).build_system_prompt( - workspace=project, - include_memory_recent_history=False, - ) + result = ContextBuilder(agent_home).build_system_prompt(workspace=project) assert "selected project rules" in result assert "global project rules" not in result @@ -152,10 +149,7 @@ class TestLoadBootstrapFiles: project.mkdir() (agent_home / "AGENTS.md").write_text("default workspace rules", encoding="utf-8") - result = ContextBuilder(agent_home).build_system_prompt( - workspace=project, - include_memory_recent_history=False, - ) + result = ContextBuilder(agent_home).build_system_prompt(workspace=project) assert "default workspace rules" not in result diff --git a/tests/agent/test_context_prompt_cache.py b/tests/agent/test_context_prompt_cache.py index b3ddcb32e..b852e0f26 100644 --- a/tests/agent/test_context_prompt_cache.py +++ b/tests/agent/test_context_prompt_cache.py @@ -3,7 +3,6 @@ from __future__ import annotations import datetime as datetime_module -import re from datetime import datetime as real_datetime from importlib.resources import files as pkg_files from pathlib import Path @@ -104,173 +103,6 @@ def test_provider_context_appended_after_user_content(tmp_path) -> None: assert user_pos < context_pos, "user content must precede provider context" -def test_unprocessed_history_injected_into_system_prompt(tmp_path) -> None: - """Entries in history.jsonl not yet consumed by Dream appear with timestamps.""" - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - builder.memory.append_history("User asked about weather in Tokyo") - builder.memory.append_history("Agent fetched forecast via web_search") - - prompt = builder.build_system_prompt() - assert "# Recent History" in prompt - assert "User asked about weather in Tokyo" in prompt - assert "Agent fetched forecast via web_search" in prompt - assert re.search(r"\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}\]", prompt) - - -def test_recent_history_injection_is_session_scoped(tmp_path) -> None: - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - builder.memory.append_history("legacy entry without session") - builder.memory.append_history("telegram history", session_key="telegram:chat-1") - builder.memory.append_history("slack history", session_key="slack:chat-2") - - prompt = builder.build_system_prompt(session_key="telegram:chat-1") - - assert "# Recent History" in prompt - assert "telegram history" in prompt - assert "slack history" not in prompt - assert "legacy entry without session" not in prompt - - -def test_session_summary_replaces_interleaved_recent_history_entry(tmp_path) -> None: - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - session_key = "unified:default" - overview = "CURRENT_SESSION_OVERVIEW_MARKER" - - builder.memory.append_history("another session event", session_key=session_key) - builder.memory.append_history(overview, session_key=session_key) - latest_cursor = builder.memory.append_history( - "later telegram event", - session_key="telegram:chat-1", - ) - summary = {"text": overview, "last_active": "2026-08-19T10:00:00"} - - prompt = builder.build_system_prompt( - session_key=session_key, - session_summary=summary, - unified_session=True, - ) - - assert "# Recent History" in prompt - assert "another session event" in prompt - assert "later telegram event" in prompt - assert "[Archived Context Summary]" in prompt - assert prompt.count(overview) == 1 - - builder.memory.set_last_dream_cursor(latest_cursor) - processed_prompt = builder.build_system_prompt( - session_key=session_key, - session_summary=summary, - unified_session=True, - ) - assert "# Recent History" not in processed_prompt - assert processed_prompt.count(overview) == 1 - - -def test_recent_history_injection_unified_excludes_cron_internals(tmp_path) -> None: - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - builder.memory.append_history("unified user history", session_key="unified:default") - builder.memory.append_history("channel user history", session_key="telegram:chat-1") - builder.memory.append_history("cron internal history", session_key="cron:job-1") - - prompt = builder.build_system_prompt( - session_key="unified:default", - unified_session=True, - ) - - assert "unified user history" in prompt - assert "channel user history" in prompt - assert "cron internal history" not in prompt - - -def test_cron_recent_history_can_see_own_history_and_unified_context(tmp_path) -> None: - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - builder.memory.append_history("unified user history", session_key="unified:default") - builder.memory.append_history("own cron history", session_key="cron:job-1") - builder.memory.append_history("other cron history", session_key="cron:job-2") - - prompt = builder.build_system_prompt( - session_key="cron:job-1", - unified_session=True, - ) - - assert "unified user history" in prompt - assert "own cron history" in prompt - assert "other cron history" not in prompt - - -def test_recent_history_capped_at_max(tmp_path) -> None: - """Only the most recent _MAX_RECENT_HISTORY entries are injected.""" - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - for i in range(builder._MAX_RECENT_HISTORY + 20): - builder.memory.append_history(f"entry-{i}") - - prompt = builder.build_system_prompt() - assert "entry-0" not in prompt - assert "entry-19" not in prompt - assert f"entry-{builder._MAX_RECENT_HISTORY + 19}" in prompt - - -def test_recent_history_truncated_at_max_tokens(tmp_path) -> None: - """Recent History section must be truncated to _MAX_HISTORY_TOKENS.""" - import tiktoken - - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - big_entry = "word " * (builder._MAX_HISTORY_TOKENS + 5_000) - builder.memory.append_history(big_entry) - - prompt = builder.build_system_prompt() - history_section = prompt.split("# Recent History\n\n", 1) - assert len(history_section) == 2 - - enc = tiktoken.get_encoding("cl100k_base") - assert len(enc.encode(history_section[1])) <= builder._MAX_HISTORY_TOKENS - - -def test_no_recent_history_when_dream_has_processed_all(tmp_path) -> None: - """If Dream has consumed everything, no Recent History section should appear.""" - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - cursor = builder.memory.append_history("already processed entry") - builder.memory.set_last_dream_cursor(cursor) - - prompt = builder.build_system_prompt() - assert "# Recent History" not in prompt - - -def test_partial_dream_processing_shows_only_remainder(tmp_path) -> None: - """When Dream has processed some entries, only the unprocessed ones appear.""" - workspace = _make_workspace(tmp_path) - builder = ContextBuilder(workspace) - - builder.memory.append_history("old conversation about Python") - c2 = builder.memory.append_history("old conversation about Rust") - builder.memory.append_history("recent question about Docker") - builder.memory.append_history("recent question about K8s") - - builder.memory.set_last_dream_cursor(c2) - - prompt = builder.build_system_prompt() - assert "# Recent History" in prompt - assert "old conversation about Python" not in prompt - assert "old conversation about Rust" not in prompt - assert "recent question about Docker" in prompt - assert "recent question about K8s" in prompt - - def test_execution_rules_in_system_prompt(tmp_path) -> None: """Execution rules should appear in the system prompt via the default templates.""" from nanobot.utils.helpers import sync_workspace_templates diff --git a/tests/agent/test_loop_consolidation_tokens.py b/tests/agent/test_loop_consolidation_tokens.py index 577b7f6af..eab8edc9d 100644 --- a/tests/agent/test_loop_consolidation_tokens.py +++ b/tests/agent/test_loop_consolidation_tokens.py @@ -7,11 +7,17 @@ from nanobot.bus.queue import MessageBus from nanobot.providers.base import LLMResponse -def _make_loop(tmp_path, *, estimated_tokens: int, context_window_tokens: int) -> AgentLoop: +def _make_loop( + tmp_path, + *, + estimated_tokens: int, + context_window_tokens: int, + max_tokens: int = 0, +) -> AgentLoop: from nanobot.providers.base import GenerationSettings provider = MagicMock() provider.get_default_model.return_value = "test-model" - provider.generation = GenerationSettings(max_tokens=0) + provider.generation = GenerationSettings(max_tokens=max_tokens) provider.estimate_prompt_tokens.return_value = (estimated_tokens, "test-counter") _response = LLMResponse(content="ok", tool_calls=[]) provider.chat_with_retry = AsyncMock(return_value=_response) @@ -56,6 +62,34 @@ async def test_prompt_above_threshold_triggers_consolidation(tmp_path) -> None: assert loop.consolidator.archive_session.await_count >= 1 +@pytest.mark.asyncio +async def test_token_consolidation_refreshes_summary_for_current_request(tmp_path) -> None: + loop = _make_loop(tmp_path, estimated_tokens=0, context_window_tokens=200) + loop.consolidator.archive_session = AsyncMock( # type: ignore[method-assign] + return_value="FRESH_CHECKPOINT" + ) + loop.consolidator.estimate_session_prompt_tokens = MagicMock( # type: ignore[method-assign] + return_value=(1000, "test") + ) + loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign] + + session = loop.sessions.get_or_create("cli:test") + session.messages = [ + {"role": role, "content": f"{role[0]}{turn}"} + for turn in range(10) + for role in ("user", "assistant") + ] + loop.sessions.save(session) + + await loop.process_direct("hello", session_key="cli:test") + + request_messages = loop.provider.chat_with_retry.await_args.kwargs["messages"] + system_prompt = request_messages[0]["content"] + assert "FRESH_CHECKPOINT" in system_prompt + assert all(message.get("content") != "u0" for message in request_messages) + assert loop.sessions.get_or_create("cli:test").last_archived == 12 + + @pytest.mark.asyncio async def test_prompt_above_threshold_uses_fixed_recent_tail(tmp_path) -> None: loop = _make_loop(tmp_path, estimated_tokens=1000, context_window_tokens=200) diff --git a/tests/agent/test_memory_store.py b/tests/agent/test_memory_store.py index 502a126f1..276eb3372 100644 --- a/tests/agent/test_memory_store.py +++ b/tests/agent/test_memory_store.py @@ -113,54 +113,6 @@ class TestHistoryWithCursor: entries = store.read_unprocessed_history(since_cursor=0) assert len(entries) == 2 - def test_prompt_history_filters_to_current_session(self, store): - store.append_history("legacy entry without session") - store.append_history("telegram entry", session_key="telegram:chat-1") - store.append_history("slack entry", session_key="slack:chat-2") - - entries = store.read_recent_history_for_prompt( - since_cursor=0, - session_key="telegram:chat-1", - ) - - assert [e["content"] for e in entries] == ["telegram entry"] - assert [e["content"] for e in store.read_unprocessed_history(0)] == [ - "legacy entry without session", - "telegram entry", - "slack entry", - ] - - def test_unified_prompt_history_excludes_internal_cron_sessions(self, store): - store.append_history("legacy entry without session") - store.append_history("unified entry", session_key="unified:default") - store.append_history("telegram entry", session_key="telegram:chat-1") - store.append_history("cron internal entry", session_key="cron:job-1") - - entries = store.read_recent_history_for_prompt( - since_cursor=0, - session_key="unified:default", - unified_session=True, - ) - - assert [e["content"] for e in entries] == [ - "legacy entry without session", - "unified entry", - "telegram entry", - ] - - def test_unified_cron_prompt_history_includes_own_cron_entry(self, store): - store.append_history("unified entry", session_key="unified:default") - store.append_history("other cron entry", session_key="cron:job-2") - store.append_history("own cron entry", session_key="cron:job-1") - - entries = store.read_recent_history_for_prompt( - since_cursor=0, - session_key="cron:job-1", - unified_session=True, - ) - - assert [e["content"] for e in entries] == ["unified entry", "own cron entry"] - def test_read_unprocessed_skips_entries_without_cursor(self, store): """Regression: entries missing the cursor key should be silently skipped.""" store.history_file.write_text( diff --git a/tests/agent/test_new_command_archival.py b/tests/agent/test_new_command_archival.py index 314a33e80..42493c812 100644 --- a/tests/agent/test_new_command_archival.py +++ b/tests/agent/test_new_command_archival.py @@ -8,6 +8,10 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from nanobot.utils.prompt_templates import render_template + +_ARCHIVE_PROMPT = render_template("agent/consolidator_archive.md", strip=True) + class TestNewCommandArchival: """Test /new archival behavior with the structured archive flow.""" @@ -117,7 +121,7 @@ class TestNewCommandArchival: await loop.aclose() sent = loop.provider.chat_with_retry.call_args.kwargs["messages"] assert sent[1:-1] == ordinary_history - assert "final 2 conversation messages" in sent[-1]["content"] + assert sent[-1]["content"] == _ARCHIVE_PROMPT @pytest.mark.asyncio async def test_new_clears_session_and_responds(self, tmp_path: Path) -> None: