diff --git a/README.md b/README.md index a4410385..8b39a25b 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,7 @@ Lines starting with `,` enter internal command mode (`,help`, `,skill name=my-sk | `BUB_MAX_STEPS` | unlimited | Tool-use loop limit; must be a positive integer | | `BUB_MAX_TOKENS` | `16384` | Max tokens per model call | | `BUB_MODEL_TIMEOUT_SECONDS` | — | Model call timeout (seconds) | +| `BUB_SPILL_THRESHOLD` | `4096` | Estimated tokens before tool output spills; `0` disables | ## Background diff --git a/env.example b/env.example index 191a8ab9..287942ba 100644 --- a/env.example +++ b/env.example @@ -10,6 +10,8 @@ # BUB_MAX_STEPS=50 # BUB_MAX_TOKENS=16384 # BUB_MODEL_TIMEOUT_SECONDS=300 +# Estimated tokens (4 chars each) above which string tool results move to a chunked spill tape. 0 disables. +# BUB_SPILL_THRESHOLD=4096 # BUB_HOME=~/.bub # --------------------------------------------------------------------------- diff --git a/src/bub/builtin/agent.py b/src/bub/builtin/agent.py index da92c822..2de31406 100644 --- a/src/bub/builtin/agent.py +++ b/src/bub/builtin/agent.py @@ -59,7 +59,12 @@ def tape(self) -> Tape: tape_store = InMemoryTapeStore() if not is_async_tape_store(tape_store): tape_store = AsyncTapeStoreAdapter(tape_store) - return Tape(bub.home / "tapes", tape_store, self.framework.build_tape_context()) + return Tape( + bub.home / "tapes", + tape_store, + self.framework.build_tape_context(), + sidecars=self.framework.get_tape_sidecars(), + ) @staticmethod def _events_from_iterable(iterable: Iterable) -> AsyncStreamEvents: diff --git a/src/bub/builtin/hook_impl.py b/src/bub/builtin/hook_impl.py index 4df649dd..76f8ab60 100644 --- a/src/bub/builtin/hook_impl.py +++ b/src/bub/builtin/hook_impl.py @@ -19,8 +19,9 @@ from bub.envelope import Envelope, content_of, field_of from bub.framework import BubFramework from bub.hooks import hookimpl -from bub.hooks.interception import ToolCall, ToolCallDecision +from bub.hooks.interception import ToolCall, ToolCallDecision, ToolCallResult from bub.model_selection import ModelChoice, ModelOptions +from bub.sidecars import TapeSidecar from bub.store import TapeStore from bub.streaming import AsyncStreamEvents from bub.tape import TapeContext @@ -66,7 +67,7 @@ class BuiltinImpl: """Default hook implementations for basic runtime operations.""" def __init__(self, framework: BubFramework) -> None: - from bub.builtin import tools # noqa: F401 + from bub.builtin import spill, tools # noqa: F401 self.framework = framework self._agent: Agent | None = None @@ -362,6 +363,13 @@ def provide_tape_store(self) -> TapeStore: return FileTapeStore(directory=bub.home / "tapes") + @hookimpl + def provide_tape_sidecars(self) -> list[TapeSidecar]: + from bub.builtin.spill import SpillSettings, SpillStore + from bub.configure import ensure_config + + return [SpillStore(ensure_config(SpillSettings))] + @hookimpl def build_tape_context(self) -> TapeContext: return default_tape_context() @@ -409,3 +417,27 @@ async def before_tool_call( else: guidance = f"Tool `{call.tool}` does not exist. No similar tool is available." return ToolCallDecision.replace(guidance) + + @hookimpl + async def after_tool_call( + self, + call: ToolCall, + result: ToolCallResult, + state: TurnState, + ) -> None: + from bub.builtin.spill import SpillStore + + if result.error is not None or not isinstance(result.result, str): + return + tape = state.get("_runtime_tape") + if tape is None: + return + spill = SpillStore.mounted(tape) + if spill is None: + return + result.result = await spill.spill_tool_result( + tape, + result.result, + tool=call.tool, + run_id=call.run_id, + ) diff --git a/src/bub/builtin/spill.py b/src/bub/builtin/spill.py new file mode 100644 index 00000000..fbc4d7dc --- /dev/null +++ b/src/bub/builtin/spill.py @@ -0,0 +1,314 @@ +"""Chunked storage and bounded reads for oversized tool results.""" + +from __future__ import annotations + +import uuid +from collections.abc import Iterator +from dataclasses import dataclass, field +from typing import Any + +from loguru import logger +from pydantic import Field +from pydantic_settings import SettingsConfigDict + +from bub import config +from bub.configure import Settings +from bub.errors import BubError, ErrorKind +from bub.sidecars import sidecar_tape_name +from bub.tape import TapeEntry, TapeQuery +from bub.tools import ToolContext, tool + +SPILL_READ_TOOL_NAME = "spill.read" +SPILL_READ_MODEL_NAME = SPILL_READ_TOOL_NAME.replace(".", "_") +SPILL_CHUNK_BYTES = 16_384 +MAX_READ_CHUNKS = 4 +PREVIEW_CHARS = 600 +SPILL_SIDECAR_NAME = "spill" + + +@config(name="spill") +class SpillSettings(Settings): + """Configuration owned by the builtin spill sidecar.""" + + model_config = SettingsConfigDict(env_prefix="BUB_SPILL_", extra="ignore", env_file=".env") + + threshold: int = Field( + default=4096, + ge=0, + description="Estimated tokens (4 chars each) above which string tool results move to the spill sidecar.", + ) + + +def spill_tape_name(session_tape: str) -> str: + return sidecar_tape_name(session_tape, SPILL_SIDECAR_NAME) + + +def _chunk_anchor(handle: str, index: int) -> str: + return f"spill/{handle}/chunk/{index}" + + +def _manifest_anchor(handle: str) -> str: + return f"spill/{handle}/manifest" + + +def _utf8_chunks(encoded: bytes, chunk_bytes: int = SPILL_CHUNK_BYTES) -> Iterator[str]: + start = 0 + while start < len(encoded): + end = min(start + chunk_bytes, len(encoded)) + while end < len(encoded) and encoded[end] & 0xC0 == 0x80: + end -= 1 + yield encoded[start:end].decode("utf-8") + start = end + + +def _preview(text: str) -> str: + if len(text) <= PREVIEW_CHARS: + return text + head = PREVIEW_CHARS // 2 + tail = PREVIEW_CHARS - head + omitted = len(text) - PREVIEW_CHARS + return f"{text[:head]}\n...[{omitted:,} chars omitted]...\n{text[-tail:]}" + + +@dataclass(frozen=True) +class SpillManifest: + handle: str + chunks: int + bytes: int + chars: int + lines: int + + @classmethod + def from_entry(cls, entry: TapeEntry, handle: str) -> SpillManifest | None: + if entry.kind != "event" or entry.payload.get("name") != "spill.manifest": + return None + data = entry.payload.get("data") + if not isinstance(data, dict) or data.get("handle") != handle: + return None + values = (data.get("chunks"), data.get("bytes"), data.get("chars"), data.get("lines")) + if not all(isinstance(value, int) and not isinstance(value, bool) and value >= 0 for value in values): + return None + return cls( + handle=handle, + chunks=data["chunks"], + bytes=data["bytes"], + chars=data["chars"], + lines=data["lines"], + ) + + +@dataclass(frozen=True) +class SpillPage: + manifest: SpillManifest + content: str + start: int + stop: int + next_cursor: int + complete: bool + + +class IncompleteSpillError(RuntimeError): + """Raised when a manifest points to missing or invalid chunks.""" + + +@dataclass(frozen=True) +class SpillStore: + """Store spill chunks as ordinary entries in a sibling tape.""" + + settings: SpillSettings + name: str = field(default=SPILL_SIDECAR_NAME, init=False) + + @classmethod + def mounted(cls, tape: Any) -> SpillStore | None: + sidecar = tape.get_sidecar(cls.name) + return sidecar if isinstance(sidecar, cls) else None + + async def _record_write(self, tape: Any, data: dict[str, object], *, run_id: str) -> None: + try: + await tape.append_event("spill.write", data, run_id=run_id, context=False) + except Exception as exc: + logger.warning("spill write event failed run_id={} error={}", run_id, exc) + + async def spill_tool_result(self, tape: Any, result: str, *, tool: str, run_id: str) -> str: + threshold = self.settings.threshold + if threshold <= 0 or tool in {SPILL_READ_TOOL_NAME, SPILL_READ_MODEL_NAME} or len(result) < threshold * 4: + return result + + handle = uuid.uuid4().hex + encoded = result.encode("utf-8") + encoded_bytes = len(encoded) + chunk_count = 0 + spill_tape = tape.sidecar_tape_name(self.name) + try: + for index, chunk in enumerate(_utf8_chunks(encoded)): + await tape.store.append(spill_tape, TapeEntry.anchor(_chunk_anchor(handle, index))) + await tape.store.append( + spill_tape, + TapeEntry.tool_result([chunk], spill_handle=handle, spill_chunk=index), + ) + chunk_count = index + 1 + await tape.store.append(spill_tape, TapeEntry.anchor(_manifest_anchor(handle))) + await tape.store.append( + spill_tape, + TapeEntry.event( + "spill.manifest", + { + "handle": handle, + "chunks": chunk_count, + "bytes": encoded_bytes, + "chars": len(result), + "lines": result.count("\n") + 1, + "tool": tool, + }, + spill_handle=handle, + run_id=run_id, + ), + ) + except Exception as exc: + logger.warning("tool result spill failed tool={} error={}", tool, exc) + await self._record_write( + tape, + { + "status": "error", + "handle": handle, + "bytes": encoded_bytes, + "tool": tool, + "error": str(exc), + }, + run_id=run_id, + ) + return f"[tool output truncated: {encoded_bytes:,} bytes; spill storage failed]\n{_preview(result)}" + + await self._record_write( + tape, + { + "status": "ok", + "handle": handle, + "bytes": encoded_bytes, + "chunks": chunk_count, + "tool": tool, + }, + run_id=run_id, + ) + + return ( + f"[tool output spilled: {encoded_bytes:,} bytes in {chunk_count:,} chunks; handle: {handle}]\n" + f"[read with: {SPILL_READ_MODEL_NAME}(handle={handle!r}, cursor=0, count=1, from_end=False)]\n" + f"{_preview(result)}" + ) + + async def manifest(self, tape: Any, handle: str) -> SpillManifest | None: + query = ( + TapeQuery(tape=tape.sidecar_tape_name(self.name), store=tape.store) + .after_anchor(_manifest_anchor(handle)) + .kinds("event") + .limit(1) + ) + try: + entries = list(await tape.store.fetch_all(query)) + except BubError as exc: + if exc.kind is ErrorKind.NOT_FOUND: + return None + raise + if not entries: + return None + return SpillManifest.from_entry(entries[0], handle) + + async def read( + self, + tape: Any, + handle: str, + *, + cursor: int, + count: int, + from_end: bool, + ) -> SpillPage | None: + manifest = await self.manifest(tape, handle) + if manifest is None: + return None + + count = min(count, MAX_READ_CHUNKS) + if from_end: + stop = max(0, manifest.chunks - cursor) + start = max(0, stop - count) + next_cursor = cursor + (stop - start) + complete = start == 0 + else: + start = min(cursor, manifest.chunks) + stop = min(start + count, manifest.chunks) + next_cursor = stop + complete = stop == manifest.chunks + + if start == stop: + return SpillPage(manifest, "", start, stop, next_cursor, True) + + query = ( + TapeQuery(tape=tape.sidecar_tape_name(self.name), store=tape.store) + .after_anchor(_chunk_anchor(handle, start)) + .kinds("tool_result") + .limit(stop - start) + ) + try: + entries = list(await tape.store.fetch_all(query)) + except BubError as exc: + if exc.kind is ErrorKind.NOT_FOUND: + raise IncompleteSpillError(f"missing chunk {start} for handle {handle!r}") from exc + raise + if len(entries) != stop - start: + raise IncompleteSpillError(f"missing chunks for handle {handle!r}") + + chunks: list[str] = [] + for index, entry in enumerate(entries, start=start): + results = entry.payload.get("results") + if ( + entry.meta.get("spill_handle") != handle + or entry.meta.get("spill_chunk") != index + or not isinstance(results, list) + or len(results) != 1 + or not isinstance(results[0], str) + ): + raise IncompleteSpillError(f"invalid chunk {index} for handle {handle!r}") + chunks.append(results[0]) + + return SpillPage(manifest, "".join(chunks), start, stop, next_cursor, complete) + + +@tool(context=True, name=SPILL_READ_TOOL_NAME) +async def spill_read( + handle: str, + cursor: int = 0, + count: int = 1, + from_end: bool = False, + *, + context: ToolContext, +) -> str: + """Read bounded chunks from an oversized tool result stored in the current session's spill tape.""" + if cursor < 0: + return "`cursor` must be >= 0." + if count < 1: + return "`count` must be >= 1." + + spill = SpillStore.mounted(context.tape) + if spill is None: + return "spill sidecar unavailable in this context." + try: + page = await spill.read( + context.tape, + handle, + cursor=cursor, + count=min(count, MAX_READ_CHUNKS), + from_end=from_end, + ) + except IncompleteSpillError as exc: + return f"[incomplete spilled tool result: {exc}]" + if page is None: + return f"[no spilled tool result for handle {handle!r}]" + + shown = f"{page.start}-{page.stop - 1}" if page.stop > page.start else "none" + return ( + f"[spilled tool result: {page.manifest.bytes:,} bytes, {page.manifest.chunks:,} chunks]\n" + f"chunks: {shown}\n" + f"next_cursor: {page.next_cursor}\n" + f"complete: {str(page.complete).lower()}\n" + f"content:\n{page.content}" + ) diff --git a/src/bub/builtin/tools.py b/src/bub/builtin/tools.py index 3e4cb916..bfe11f8c 100644 --- a/src/bub/builtin/tools.py +++ b/src/bub/builtin/tools.py @@ -290,8 +290,7 @@ async def tape_search(param: SearchInput, *, context: ToolContext) -> str: @tool(context=True, name="tape.reset") async def tape_reset(archive: bool = False, *, context: ToolContext) -> str: """Reset the current tape, optionally archiving it.""" - result = await context.tape.reset(archive=archive) - return result + return cast(str, await context.tape.reset(archive=archive)) @tool(context=True, name="tape.handoff") diff --git a/src/bub/framework.py b/src/bub/framework.py index b11f1cb9..a84f253c 100644 --- a/src/bub/framework.py +++ b/src/bub/framework.py @@ -22,6 +22,7 @@ from bub.hooks.runtime import _SKIP_VALUE, HookRuntime from bub.hooks.specs import BUB_HOOK_NAMESPACE, BubHookSpecs from bub.model_selection import ModelOptions +from bub.sidecars import TapeSidecar from bub.store import AsyncTapeStore, TapeStore from bub.tape import TapeContext from bub.turn import TurnResult, TurnState @@ -361,6 +362,13 @@ async def running(self) -> AsyncGenerator[contextlib.AsyncExitStack, None]: def get_tape_store(self) -> TapeStore | AsyncTapeStore | None: return self._tape_store + def get_tape_sidecars(self) -> tuple[TapeSidecar, ...]: + sidecars: dict[str, TapeSidecar] = {} + for provided in self._hook_runtime.call_many_sync("provide_tape_sidecars"): + for sidecar in provided: + sidecars.setdefault(sidecar.name, sidecar) + return tuple(sidecars.values()) + def get_steering_inbox(self) -> SteeringInbox | None: return self._steering_inbox diff --git a/src/bub/hooks/interception.py b/src/bub/hooks/interception.py index acca58ab..cb1d26e1 100644 --- a/src/bub/hooks/interception.py +++ b/src/bub/hooks/interception.py @@ -95,9 +95,9 @@ def deny(cls, message: str) -> ToolCallDecision: return cls(action="deny", message=message) -@dataclass(frozen=True) +@dataclass class ToolCallResult: - """Terminal outcome of one tool invocation exposed to hooks.""" + """Terminal outcome exposed to hooks; successful results may be replaced in place.""" run_id: str tool: str @@ -165,7 +165,7 @@ async def before_tool_call(self, call: ToolCall, state: TurnState) -> tuple[Tool return call, ToolCallDecision.proceed() async def after_tool_call(self, call: ToolCall, result: ToolCallResult, state: TurnState) -> None: - """Notify every observer; return values are ignored.""" + """Notify every implementation with one shared, mutable outcome.""" await self._safe_calls("after_tool_call", lambda: {"call": call, "state": state, "result": result}) diff --git a/src/bub/hooks/specs.py b/src/bub/hooks/specs.py index 6ddf93f3..cc695d45 100644 --- a/src/bub/hooks/specs.py +++ b/src/bub/hooks/specs.py @@ -19,6 +19,7 @@ ToolCallResult, ) from bub.model_selection import ModelOptions +from bub.sidecars import TapeSidecar from bub.store import AsyncTapeStore, TapeStore from bub.streaming import AsyncStreamEvents from bub.tape import TapeContext @@ -155,10 +156,12 @@ def before_tool_call(self, call: ToolCall, state: TurnState) -> ToolCallDecision @hookspec def after_tool_call(self, call: ToolCall, result: ToolCallResult, state: TurnState) -> None: - """Observe the terminal outcome of one tool invocation. + """Handle the terminal outcome of one tool invocation. Fires for success, failure (``result.error`` set), denial and - replacement. Return values are ignored; exceptions are logged. + replacement. An implementation may replace a successful value by + assigning ``result.result``. Return values are ignored; exceptions + are logged. """ @hookspec @@ -171,6 +174,11 @@ def provide_tape_store(self) -> TapeStore | AsyncTapeStore | None: """Provide a tape store instance for Bub's conversation recording feature.""" raise NotImplementedError + @hookspec + def provide_tape_sidecars(self) -> list[TapeSidecar]: + """Provide capabilities backed by sibling tapes mounted on every session tape.""" + raise NotImplementedError + @hookspec def provide_channels(self, message_handler: MessageHandler) -> list[Channel]: """Provide a list of channels for receiving messages.""" diff --git a/src/bub/sidecars.py b/src/bub/sidecars.py new file mode 100644 index 00000000..b1d24f5b --- /dev/null +++ b/src/bub/sidecars.py @@ -0,0 +1,18 @@ +"""Contracts for tapes mounted beside a session tape.""" + +from __future__ import annotations + +from typing import Protocol + + +class TapeSidecar(Protocol): + """A named capability backed by a sibling tape.""" + + @property + def name(self) -> str: ... + + +def sidecar_tape_name(owner: str, sidecar: str) -> str: + """Return the physical tape name for a mounted sidecar.""" + + return f"{owner}__{sidecar}" diff --git a/src/bub/store.py b/src/bub/store.py index addb39e5..d319af94 100644 --- a/src/bub/store.py +++ b/src/bub/store.py @@ -281,40 +281,46 @@ def append(self, tape: str, entry: TapeEntry) -> None: class ForkTapeStore: - def __init__(self, parent: AsyncTapeStore, tape: str) -> None: + def __init__(self, parent: AsyncTapeStore, tape: str, *, sidecars: Iterable[str] = ()) -> None: self._parent = parent self._store = InMemoryTapeStore() self._tape = tape - self._tape_was_reset = False + self._sidecars = tuple(dict.fromkeys(sidecars)) + self._managed_tapes = {tape, *self._sidecars} + self._reset_tapes: set[str] = set() async def list_tapes(self) -> list[str]: return await self._parent.list_tapes() async def reset(self, tape: str) -> None: - if tape != self._tape: + if tape not in self._managed_tapes: await self._parent.reset(tape) return self._store.reset(tape) - self._tape_was_reset = True + self._reset_tapes.add(tape) async def fetch_all(self, query: TapeQuery[AsyncTapeStore]) -> Iterable[TapeEntry]: + if query.tape not in self._managed_tapes: + return await self._parent.fetch_all(query) + parent_entries: Iterable[TapeEntry] = [] - if not (query.tape == self._tape and self._tape_was_reset): + if query.tape not in self._reset_tapes: try: parent_entries = await self._parent.fetch_all(query) except Exception: parent_entries = [] this_entries: list[TapeEntry] = [] for entry in self._store.read(query.tape) or []: - if query._kinds and entry.kind not in query._kinds: - continue if entry.kind == "anchor": # noqa: SIM102 if query._after_last or (query._after_anchor and entry.payload.get("name") == query._after_anchor): this_entries.clear() parent_entries = [] continue + if query._kinds and entry.kind not in query._kinds: + continue this_entries.append(entry) - return itertools.chain(parent_entries, this_entries) + entries = itertools.chain(parent_entries, this_entries) + return itertools.islice(entries, query._limit) if query._limit is not None else entries @staticmethod def _redact_prompt(prompt: list[dict]) -> Any: @@ -335,18 +341,41 @@ def _redact_payload(payload: dict) -> None: async def append(self, tape: str, entry: TapeEntry) -> None: self._redact_payload(entry.payload) + if tape not in self._managed_tapes: + await self._parent.append(tape, entry) + return self._store.append(tape, entry) async def merge_back(self) -> None: - if self._tape_was_reset: + total = 0 + for sidecar in self._sidecars: + entries = self._store.read(sidecar) or [] + try: + if sidecar in self._reset_tapes: + await self._parent.reset(sidecar) + for entry in entries: + await self._parent.append(sidecar, entry) + except Exception as exc: + logger.warning('Failed to merge sidecar "{}" into tape "{}": {}', sidecar, self._tape, exc) + self._store.append( + self._tape, + TapeEntry.event( + "sidecar.merge", + {"tape": sidecar, "status": "error", "error": str(exc)}, + context=False, + ), + ) + else: + total += len(entries) + + if self._tape in self._reset_tapes: await self._parent.reset(self._tape) - entries = self._store.read(self._tape) - if not entries: - return - count = len(entries) + entries = self._store.read(self._tape) or [] for entry in entries: await self._parent.append(self._tape, entry) - logger.info(f'Merged {count} entries into tape "{self._tape}"') + total += len(entries) + if total: + logger.info('Merged {} entries into tape fork "{}"', total, self._tape) class FileTapeStore(InMemoryQueryMixin): @@ -430,13 +459,7 @@ def _tape_file(self, tape: str) -> TapeFile: return self._tape_files[tape] def list_tapes(self) -> list[str]: - result: list[str] = [] - for file in self._directory.glob("*.jsonl"): - filename = file.stem - if filename.count("__") != 1: - continue - result.append(filename) - return result + return sorted(file.stem for file in self._directory.glob("*.jsonl")) def reset(self, tape: str) -> None: self._tape_file(tape).reset() diff --git a/src/bub/tape.py b/src/bub/tape.py index b4b3fc54..8cd50c47 100644 --- a/src/bub/tape.py +++ b/src/bub/tape.py @@ -15,6 +15,7 @@ from pydantic import BaseModel from bub.errors import BubError +from bub.sidecars import TapeSidecar, sidecar_tape_name __all__ = [ "LAST_ANCHOR", @@ -200,6 +201,7 @@ class Tape: archive_path: Path store: AsyncTapeStore context: TapeContext + sidecars: tuple[TapeSidecar, ...] = field(default=(), repr=False) _name: str | None = field(default=None, repr=False) @property @@ -219,6 +221,18 @@ def query(self) -> TapeQuery[AsyncTapeStore]: return TapeQuery(tape=self.name, store=self.store) + def get_sidecar(self, name: str) -> TapeSidecar | None: + """Return a mounted sidecar by its public name.""" + + return next((sidecar for sidecar in self.sidecars if sidecar.name == name), None) + + def sidecar_tape_name(self, name: str) -> str: + """Return the sibling tape name for a mounted sidecar.""" + + if self.get_sidecar(name) is None: + raise KeyError(f"tape sidecar {name!r} is not mounted") + return sidecar_tape_name(self.name, name) + async def info(self) -> TapeInfo: entries = list(await self.store.fetch_all(self.query())) anchors = [(i, entry) for i, entry in enumerate(entries) if entry.kind == "anchor"] @@ -286,7 +300,8 @@ async def append_event(self, name: str, payload: dict[str, Any], **meta: Any) -> async def read_messages(self) -> list[dict[str, Any]]: query = self.context.build_query(self.query()) entries = await self.store.fetch_all(query) - messages = build_messages(entries, self.context) + context_entries = (entry for entry in entries if entry.meta.get("context") is not False) + messages = build_messages(context_entries, self.context) if inspect.isawaitable(messages): messages = await messages return messages @@ -362,25 +377,141 @@ def _extract_usage(response: object) -> dict[str, Any] | None: return payload if isinstance(payload, dict) else None return None - async def _archive(self) -> Path: - tape_name = self.name - stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + async def _archive_tape(self, tape_name: str, stamp: str) -> Path: + from bub.store import TapeQuery + self.archive_path.mkdir(parents=True, exist_ok=True) archive_path = self.archive_path / f"{tape_name}.jsonl.{stamp}.bak" with archive_path.open("w", encoding="utf-8") as f: - for entry in await self.store.fetch_all(self.query()): + query = TapeQuery(tape=tape_name, store=self.store) + for entry in await self.store.fetch_all(query): f.write(json.dumps(asdict(entry), ensure_ascii=False) + "\n") return archive_path + @staticmethod + def _sidecar_lifecycle_data( + *, + sidecar: str, + status: str, + reason: str, + archive_path: Path | None = None, + error: Exception | None = None, + cause: str | None = None, + ) -> dict[str, Any]: + data: dict[str, Any] = {"sidecar": sidecar, "status": status, "reason": reason} + if archive_path is not None: + data["archive"] = str(archive_path) + if error is not None: + data["error"] = str(error) + if cause is not None: + data["cause"] = cause + return data + + async def _try_archive_sidecar( + self, + sidecar: TapeSidecar, + *, + reason: str, + stamp: str | None = None, + ) -> tuple[Path | None, dict[str, Any]]: + archive_stamp = stamp or datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + try: + archive_path = await self._archive_tape(sidecar_tape_name(self.name, sidecar.name), archive_stamp) + except Exception as exc: + return None, self._sidecar_lifecycle_data( + sidecar=sidecar.name, + status="error", + reason=reason, + error=exc, + ) + return archive_path, self._sidecar_lifecycle_data( + sidecar=sidecar.name, + status="ok", + reason=reason, + archive_path=archive_path, + ) + + async def _try_reset_sidecar(self, sidecar: TapeSidecar, *, reason: str) -> dict[str, Any]: + try: + await self.store.reset(sidecar_tape_name(self.name, sidecar.name)) + except Exception as exc: + return self._sidecar_lifecycle_data(sidecar=sidecar.name, status="error", reason=reason, error=exc) + return self._sidecar_lifecycle_data(sidecar=sidecar.name, status="ok", reason=reason) + + def _require_sidecar(self, name: str) -> TapeSidecar: + sidecar = self.get_sidecar(name) + if sidecar is None: + raise KeyError(f"tape sidecar {name!r} is not mounted") + return sidecar + + async def archive_sidecar(self, name: str, *, reason: str = "manual") -> str: + """Archive one mounted sidecar without changing the main tape.""" + + sidecar = self._require_sidecar(name) + archive_path, event_data = await self._try_archive_sidecar(sidecar, reason=reason) + await self.append_event("sidecar.archive", event_data, context=False) + return ( + f"Archived {name}: {archive_path}" + if archive_path is not None + else f"{name} archive failed: {event_data['error']}" + ) + + async def reset_sidecar(self, name: str, *, archive: bool = False, reason: str = "gc") -> str: + """Reset one mounted sidecar and record the outcome on the main tape.""" + + sidecar = self._require_sidecar(name) + archive_path: Path | None = None + archive_data: dict[str, Any] | None = None + if archive: + archive_path, archive_data = await self._try_archive_sidecar(sidecar, reason=reason) + + if archive_data is not None and archive_data["status"] == "error": + reset_data = self._sidecar_lifecycle_data( + sidecar=name, + status="skipped", + reason=reason, + cause="archive_failed", + ) + else: + reset_data = await self._try_reset_sidecar(sidecar, reason=reason) + if archive_data is not None: + await self.append_event("sidecar.archive", archive_data, context=False) + await self.append_event("sidecar.reset", reset_data, context=False) + + if reset_data["status"] == "error": + return f"{name} reset failed: {reset_data['error']}" + if reset_data["status"] == "skipped" and archive_data is not None: + return f"{name} archive failed: {archive_data['error']}; {name} reset skipped" + return f"Archived {name}: {archive_path}" if archive_path is not None else "ok" + async def reset(self, *, archive: bool = False) -> str: archive_path: Path | None = None + sidecar_archives: dict[str, dict[str, Any]] = {} if archive: - archive_path = await self._archive() + stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + archive_path = await self._archive_tape(self.name, stamp) + for sidecar in self.sidecars: + _, sidecar_archive = await self._try_archive_sidecar(sidecar, reason="tape.reset", stamp=stamp) + sidecar_archives[sidecar.name] = sidecar_archive await self.store.reset(self.name) state = {"owner": "human"} if archive_path is not None: state["archived"] = str(archive_path) await self.handoff(name="session/start", state=state) + for sidecar in self.sidecars: + archive_data = sidecar_archives.get(sidecar.name) + if archive_data is not None and archive_data["status"] == "error": + reset_data = self._sidecar_lifecycle_data( + sidecar=sidecar.name, + status="skipped", + reason="tape.reset", + cause="archive_failed", + ) + else: + reset_data = await self._try_reset_sidecar(sidecar, reason="tape.reset") + if archive_data is not None: + await self.append_event("sidecar.archive", archive_data, context=False) + await self.append_event("sidecar.reset", reset_data, context=False) return f"Archived: {archive_path}" if archive_path else "ok" def session_tape(self, session_id: str, workspace: Path, context: TapeContext | None = None) -> Tape: @@ -394,7 +525,8 @@ def session_tape(self, session_id: str, workspace: Path, context: TapeContext | async def fork_tape(self, merge_back: bool = True) -> AsyncGenerator[Tape, None]: from bub.store import ForkTapeStore - fork_store = ForkTapeStore(self.store, self.name) + managed_sidecars = tuple(sidecar_tape_name(self.name, sidecar.name) for sidecar in self.sidecars) + fork_store = ForkTapeStore(self.store, self.name, sidecars=managed_sidecars) forked = replace(self, store=fork_store) try: yield forked diff --git a/src/bub/tools.py b/src/bub/tools.py index 057046b4..6031491e 100644 --- a/src/bub/tools.py +++ b/src/bub/tools.py @@ -15,7 +15,6 @@ from bub.errors import BubError, ErrorKind from bub.hooks.interception import ToolCall, ToolCallResult -from bub.tape import Tape if TYPE_CHECKING: from bub.hooks.interception import AgentHooks @@ -25,7 +24,7 @@ class ToolContext: """Runtime context passed to tools that opt into context.""" - tape: Tape + tape: Any run_id: str | None = None state: dict[str, Any] = field(default_factory=dict) @@ -243,6 +242,8 @@ async def _handle_tool_response_async( arguments=dict(tool_args), ) hook_state = context.state if context is not None else {} + if self._hooks is not None and context is not None: + hook_state["_runtime_tape"] = context.tape started = time.monotonic() if self._hooks is not None: call, short_circuit = await self._apply_before_tool_call(call, hook_state, started) @@ -255,8 +256,8 @@ async def _handle_tool_response_async( await self._fire_after_tool_call(call, hook_state, started, error=exc) raise else: - await self._fire_after_tool_call(call, hook_state, started, result=result) - return result + outcome = await self._fire_after_tool_call(call, hook_state, started, result=result) + return outcome.result async def _invoke_normalized(self, tool_obj: Tool, call: ToolCall, context: ToolContext | None) -> Any: """Run the tool with errors normalized to BubError.""" @@ -310,8 +311,8 @@ def raise_denied() -> Any: return call, raise_denied if decision.action == "replace": - await self._fire_after_tool_call(call, hook_state, started, result=decision.result) - return call, lambda: decision.result + outcome = await self._fire_after_tool_call(call, hook_state, started, result=decision.result) + return call, lambda: outcome.result return call, None async def _fire_after_tool_call( @@ -322,9 +323,7 @@ async def _fire_after_tool_call( *, result: Any = None, error: Exception | None = None, - ) -> None: - if self._hooks is None: - return + ) -> ToolCallResult: duration_ms = int((time.monotonic() - started) * 1000) outcome = ToolCallResult( run_id=call.run_id, @@ -334,7 +333,9 @@ async def _fire_after_tool_call( error=error, duration_ms=duration_ms, ) - await self._hooks.after_tool_call(call, outcome, state=state) + if self._hooks is not None: + await self._hooks.after_tool_call(call, outcome, state=state) + return outcome # Central registry for tools. Tools defined with the @tool decorator are automatically added here. diff --git a/tests/test_agent_hooks.py b/tests/test_agent_hooks.py index 0a181f90..5ce6fbd6 100644 --- a/tests/test_agent_hooks.py +++ b/tests/test_agent_hooks.py @@ -186,6 +186,19 @@ def failing(cmd: str) -> str: assert observed[1].error.kind is not None assert "bad" in observed[1].tool + @pytest.mark.asyncio + async def test_after_tool_call_can_replace_the_result_seen_by_the_model(self) -> None: + class BoundResult: + @hookimpl + def after_tool_call(self, call: ToolCall, result: ToolCallResult, state: dict) -> None: + if isinstance(result.result, str): + result.result = f"bounded:{result.result}" + + executor = ToolExecutor(hooks=make_hooks(BoundResult())) + execution = await executor.execute_async([(self.tool(), {"cmd": "ls"})]) + + assert execution.tool_results == ["bounded:ran:ls"] + @pytest.mark.asyncio async def test_modified_arguments_reach_handler(self) -> None: class Rewrite: diff --git a/tests/test_builtin_agent.py b/tests/test_builtin_agent.py index 7d8a01e0..bc5a01a5 100644 --- a/tests/test_builtin_agent.py +++ b/tests/test_builtin_agent.py @@ -38,6 +38,7 @@ def _make_agent() -> Agent: framework.get_tape_store.return_value = None framework.get_steering_inbox.return_value = None framework.get_system_prompt.return_value = "" + framework.get_tape_sidecars.return_value = () async def build_prompt(message: dict[str, Any], session_id: str, state: dict[str, Any]) -> str: return str(message["content"]) diff --git a/tests/test_builtin_hook_impl.py b/tests/test_builtin_hook_impl.py index 2d8d95f4..ac546dd8 100644 --- a/tests/test_builtin_hook_impl.py +++ b/tests/test_builtin_hook_impl.py @@ -420,6 +420,14 @@ def test_provide_tape_store_uses_bub_home_directory(tmp_path: Path, monkeypatch: assert store._directory == tmp_path / "tapes" +def test_builtin_mounts_the_spill_sidecar(tmp_path: Path) -> None: + from bub.builtin.spill import SpillStore + + _, impl, _ = _build_impl(tmp_path) + + assert isinstance(impl.provide_tape_sidecars()[0], SpillStore) + + def test_before_tool_call_ignores_known_tool(tmp_path: Path) -> None: _, impl, _ = _build_impl(tmp_path) import asyncio diff --git a/tests/test_file_tape_store_entry_ids.py b/tests/test_file_tape_store_entry_ids.py index 40986f68..d5378dc0 100644 --- a/tests/test_file_tape_store_entry_ids.py +++ b/tests/test_file_tape_store_entry_ids.py @@ -21,3 +21,11 @@ async def test_file_tape_store_assigns_monotonic_ids_when_merging_forked_entries entries = parent.read("tape") or [] assert [entry.id for entry in entries] == [1, 2] assert [entry.payload.get("name") for entry in entries] == ["first", "second"] + + +def test_file_tape_store_lists_main_and_sidecar_tapes(tmp_path) -> None: + store = FileTapeStore(directory=tmp_path) + store.append("session__id", TapeEntry.event(name="main")) + store.append("session__id__spill", TapeEntry.event(name="spill")) + + assert store.list_tapes() == ["session__id", "session__id__spill"] diff --git a/tests/test_fork_store_merge_back.py b/tests/test_fork_store_merge_back.py index e614900e..c8ecf4d1 100644 --- a/tests/test_fork_store_merge_back.py +++ b/tests/test_fork_store_merge_back.py @@ -99,3 +99,28 @@ async def test_reset_for_unbound_tape_resets_parent_immediately() -> None: entries = parent.read("test-tape") assert entries is None + + +@pytest.mark.asyncio +async def test_sidecar_merge_failure_records_event_and_still_merges_main_tape() -> None: + class BrokenSidecarStore(InMemoryTapeStore): + def append(self, tape: str, entry: TapeEntry) -> None: + if tape == "session__spill": + raise OSError("sidecar unavailable") + super().append(tape, entry) + + parent = BrokenSidecarStore() + store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "session", sidecars=("session__spill",)) + await store.append("session__spill", TapeEntry.tool_result(["full output"])) + await store.append("session", TapeEntry.tool_result(["ref"])) + + await store.merge_back() + + entries = parent.read("session") or [] + assert any(entry.kind == "tool_result" and entry.payload["results"] == ["ref"] for entry in entries) + merge_event = next(entry for entry in entries if entry.payload.get("name") == "sidecar.merge") + assert merge_event.payload["data"] == { + "tape": "session__spill", + "status": "error", + "error": "sidecar unavailable", + } diff --git a/tests/test_framework.py b/tests/test_framework.py index 44ca77c3..5bc1c7f1 100644 --- a/tests/test_framework.py +++ b/tests/test_framework.py @@ -118,6 +118,33 @@ def system_prompt(self, prompt: str, state: dict[str, str]) -> str | None: assert prompt == "low\n\nhigh" +def test_get_tape_sidecars_combines_plugins_and_prefers_the_highest_priority_name() -> None: + framework = BubFramework() + + class Sidecar: + def __init__(self, name: str, source: str) -> None: + self.name = name + self.source = source + + class LowPriorityPlugin: + @hookimpl + def provide_tape_sidecars(self): + return [Sidecar("shared", "low"), Sidecar("low-only", "low")] + + class HighPriorityPlugin: + @hookimpl + def provide_tape_sidecars(self): + return [Sidecar("shared", "high"), Sidecar("high-only", "high")] + + framework._plugin_manager.register(LowPriorityPlugin(), name="low") + framework._plugin_manager.register(HighPriorityPlugin(), name="high") + + sidecars = {sidecar.name: sidecar for sidecar in framework.get_tape_sidecars()} + + assert set(sidecars) == {"shared", "low-only", "high-only"} + assert cast(Any, sidecars["shared"]).source == "high" + + @pytest.mark.asyncio async def test_running_enters_tape_store_once_and_reuses_it() -> None: framework = BubFramework() diff --git a/tests/test_settings.py b/tests/test_settings.py index 02127a43..b4455d3f 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -4,6 +4,8 @@ from unittest.mock import patch from bub.builtin.settings import DEFAULT_MODEL, AgentSettings, load_settings +from bub.builtin.spill import SpillSettings +from bub.configure import ensure_config def _settings_with_env(env: dict[str, str]) -> AgentSettings: @@ -134,6 +136,19 @@ def test_settings_client_args_can_be_disabled() -> None: assert settings.completion_args == {} +def test_spill_sidecar_settings_can_be_configured_or_disabled() -> None: + with patch.dict("os.environ", {"BUB_SPILL_THRESHOLD": "64"}, clear=True): + assert SpillSettings().threshold == 64 + with patch.dict("os.environ", {"BUB_SPILL_THRESHOLD": "0"}, clear=True): + assert SpillSettings().threshold == 0 + + +def test_spill_sidecar_settings_load_from_the_plugin_section(load_config) -> None: + load_config("spill:\n threshold: 64") + + assert ensure_config(SpillSettings).threshold == 64 + + def test_load_settings_returns_defaults_without_loaded_config() -> None: with patch.dict(os.environ, {}, clear=True): settings = load_settings() diff --git a/tests/test_spill.py b/tests/test_spill.py new file mode 100644 index 00000000..3241fcb9 --- /dev/null +++ b/tests/test_spill.py @@ -0,0 +1,435 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from bub.builtin.context import default_tape_context +from bub.builtin.hook_impl import BuiltinImpl +from bub.builtin.spill import ( + SPILL_READ_MODEL_NAME, + SPILL_READ_TOOL_NAME, + SpillSettings, + SpillStore, + spill_read, + spill_tape_name, +) +from bub.builtin.tools import render_tools_prompt +from bub.hooks.interception import ToolCall, ToolCallDecision, ToolCallResult +from bub.store import AsyncTapeStoreAdapter, FileTapeStore, InMemoryTapeStore +from bub.tape import Tape, TapeContext, TapeEntry +from bub.tools import Tool, ToolContext, ToolExecutor, model_tools + + +class _SpillHooks: + async def before_tool_call(self, call: ToolCall, state: dict[str, Any]) -> tuple[ToolCall, ToolCallDecision]: + return call, ToolCallDecision.proceed() + + async def after_tool_call(self, call: ToolCall, result: ToolCallResult, state: dict[str, Any]) -> None: + await BuiltinImpl.after_tool_call(self, call, result, state) # type: ignore[arg-type] + + +def _spill_executor() -> ToolExecutor: + return ToolExecutor(hooks=_SpillHooks()) # type: ignore[arg-type] + + +def _handle_from_ref(ref: str) -> str: + return ref.split("handle: ", 1)[1].split("]", 1)[0] + + +def _page_content(page: str) -> str: + return page.split("content:\n", 1)[1] + + +def _page_field(page: str, name: str) -> str: + prefix = f"{name}: " + return next(line.removeprefix(prefix) for line in page.splitlines() if line.startswith(prefix)) + + +def _root_tape(tmp_path: Path, store: InMemoryTapeStore, *, threshold: int = 1) -> Tape: + spill = SpillStore(SpillSettings(threshold=threshold)) + return Tape(tmp_path, AsyncTapeStoreAdapter(store), default_tape_context(), sidecars=(spill,)).scoped("session") + + +@pytest.mark.asyncio +async def test_oversized_result_is_bounded_and_readable_across_merge(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + root = _root_tape(tmp_path, parent) + output = ("alpha🙂beta\n" * 5000) + "the-end" + sidecar = spill_tape_name(root.name) + + async with root.fork_tape() as tape: + context = ToolContext(tape=tape, run_id="run-1") + tool = Tool(name="large", handler=lambda: output) + execution = await _spill_executor().execute_async([(tool, {})], context=context) + + ref = execution.tool_results[0] + assert isinstance(ref, str) + assert "tool output spilled" in ref + assert len(ref) < 2000 + handle = _handle_from_ref(ref) + + cursor = 0 + restored: list[str] = [] + while True: + page = await spill_read.run(handle=handle, cursor=cursor, count=2, context=context) + restored.append(_page_content(page)) + if _page_field(page, "complete") == "true": + break + cursor = int(_page_field(page, "next_cursor")) + + assert "".join(restored) == output + + tail = await spill_read.run(handle=handle, cursor=0, count=1, from_end=True, context=context) + assert _page_content(tail).endswith("the-end") + + await tape.record_chat( + run_id="run-1", + system_prompt=None, + new_messages=[], + response_text=None, + tool_calls=[{"id": "call-1", "type": "function", "function": {"name": "large", "arguments": "{}"}}], + tool_results=execution.tool_results, + ) + request_messages = await tape.read_messages() + request_body = json.dumps(request_messages, ensure_ascii=False) + assert handle in request_body + assert output not in request_body + + assert parent.read(sidecar) is None + + persisted_context = ToolContext(tape=root, run_id="run-2") + persisted = await spill_read.run(handle=handle, cursor=0, count=1, context=persisted_context) + assert _page_content(persisted) == restored[0][: len(_page_content(persisted))] + assert parent.read(sidecar) + write_events = [ + entry + for entry in parent.read(root.name) or [] + if entry.kind == "event" and entry.payload.get("name") == "spill.write" + ] + assert len(write_events) == 1 + assert write_events[0].payload["data"]["status"] == "ok" + assert write_events[0].payload["data"]["handle"] == handle + + +@pytest.mark.asyncio +async def test_small_results_and_errors_are_not_spilled(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + root = _root_tape(tmp_path, parent, threshold=100) + sidecar = spill_tape_name(root.name) + + def fail() -> str: + raise ValueError("boom") + + async with root.fork_tape() as tape: + context = ToolContext(tape=tape, run_id="run-1") + small = await _spill_executor().execute_async( + [(Tool(name="small", handler=lambda: "tiny"), {})], context=context + ) + spill_page = await _spill_executor().execute_async( + [(Tool(name=SPILL_READ_MODEL_NAME, handler=lambda: "x" * 20_000), {})], context=context + ) + failed = await _spill_executor().execute_async([(Tool(name="failed", handler=fail), {})], context=context) + + assert small.tool_results == ["tiny"] + assert spill_page.tool_results == ["x" * 20_000] + assert failed.error is not None + + assert parent.read(sidecar) is None + + disabled = _root_tape(tmp_path, parent, threshold=0).scoped("disabled") + async with disabled.fork_tape() as tape: + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: "x" * 20_000), {})], + context=ToolContext(tape=tape, run_id="run-2"), + ) + assert execution.tool_results == ["x" * 20_000] + + +@pytest.mark.asyncio +async def test_temporary_fork_discards_spilled_content(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + root = _root_tape(tmp_path, parent) + sidecar = spill_tape_name(root.name) + + async with root.fork_tape(merge_back=False) as tape: + context = ToolContext(tape=tape, run_id="run-1") + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: "x" * 20_000), {})], context=context + ) + handle = _handle_from_ref(execution.tool_results[0]) + assert "content:" in await spill_read.run(handle=handle, context=context) + + assert parent.read(sidecar) is None + missing = await spill_read.run(handle=handle, context=ToolContext(tape=root)) + assert "no spilled tool result" in missing + + +@pytest.mark.asyncio +async def test_spill_failure_degrades_to_a_bounded_result(tmp_path: Path) -> None: + class BrokenStore: + async def list_tapes(self) -> list[str]: + return [] + + async def reset(self, tape: str) -> None: + pass + + async def fetch_all(self, query: Any) -> list[Any]: + return [] + + async def append(self, tape: str, entry: Any) -> None: + raise OSError("disk full") + + spill = SpillStore(SpillSettings(threshold=1)) + tape = Tape(tmp_path, BrokenStore(), TapeContext(), sidecars=(spill,)).scoped("session") + context = ToolContext(tape=tape, run_id="run-1") + output = "x" * 100_000 + + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: output), {})], context=context + ) + + result = execution.tool_results[0] + assert execution.error is None + assert isinstance(result, str) + assert "spill storage failed" in result + assert len(result) < 2000 + + +@pytest.mark.asyncio +async def test_unknown_handle_and_invalid_read_bounds_are_friendly(tmp_path: Path) -> None: + root = _root_tape(tmp_path, InMemoryTapeStore()) + context = ToolContext(tape=root) + + assert "no spilled tool result" in await spill_read.run(handle="missing", context=context) + assert await spill_read.run(handle="missing", cursor=-1, context=context) == "`cursor` must be >= 0." + assert await spill_read.run(handle="missing", count=0, context=context) == "`count` must be >= 1." + + +@pytest.mark.asyncio +async def test_spill_uses_the_regular_tape_store_contract(tmp_path: Path) -> None: + store = FileTapeStore(tmp_path / "tapes") + spill = SpillStore(SpillSettings(threshold=1)) + root = Tape(tmp_path, AsyncTapeStoreAdapter(store), default_tape_context(), sidecars=(spill,)).scoped("session") + output = "stored through the native tape store\n" * 1000 + + async with root.fork_tape() as tape: + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: output), {})], + context=ToolContext(tape=tape, run_id="run-1"), + ) + handle = _handle_from_ref(execution.tool_results[0]) + + context = ToolContext(tape=root, run_id="run-2") + first_page = await spill_read.run(handle=handle, count=1, context=context) + + assert output.startswith(_page_content(first_page)) + + +def test_spill_read_uses_the_builtin_tool_naming_convention() -> None: + assert spill_read.name == SPILL_READ_TOOL_NAME == "spill.read" + assert model_tools([spill_read])[0].name == SPILL_READ_MODEL_NAME == "spill_read" + assert "spill_read(handle, cursor?, count?, from_end?)" in render_tools_prompt([spill_read]) + + +@pytest.mark.asyncio +async def test_spilled_result_keeps_the_recorded_model_prefix_stable(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + root = _root_tape(tmp_path, parent) + output = "cache-prefix\n" * 5000 + await root.ensure_bootstrap_anchor() + + async with root.fork_tape() as tape: + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: output), {})], + context=ToolContext(tape=tape, run_id="run-1"), + ) + ref = execution.tool_results[0] + assert isinstance(ref, str) + assert f"[read with: {SPILL_READ_MODEL_NAME}(" in ref + await tape.record_chat( + run_id="run-1", + system_prompt=None, + new_messages=[{"role": "user", "content": "produce a large result"}], + response_text=None, + tool_calls=[{"id": "call-1", "type": "function", "function": {"name": "large", "arguments": "{}"}}], + tool_results=execution.tool_results, + ) + + cached_prefix = await root.read_messages() + serialized_prefix = json.dumps(cached_prefix, ensure_ascii=False, separators=(",", ":")) + assert serialized_prefix == json.dumps(await root.read_messages(), ensure_ascii=False, separators=(",", ":")) + assert output not in serialized_prefix + + await root.record_chat( + run_id="run-2", + system_prompt=None, + new_messages=[{"role": "user", "content": "continue"}], + response_text="done", + ) + + extended_messages = await root.read_messages() + assert extended_messages[: len(cached_prefix)] == cached_prefix + + +@pytest.mark.asyncio +async def test_tape_reset_clears_the_spill_sidecar_with_the_main_tape(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + root = _root_tape(tmp_path, parent) + sidecar = spill_tape_name(root.name) + await root.ensure_bootstrap_anchor() + + async with root.fork_tape() as tape: + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: "old output\n" * 5000), {})], + context=ToolContext(tape=tape, run_id="run-1"), + ) + ref = execution.tool_results[0] + assert isinstance(ref, str) + handle = _handle_from_ref(ref) + await tape.record_chat( + run_id="run-1", + system_prompt=None, + new_messages=[{"role": "user", "content": "produce output"}], + response_text=None, + tool_calls=[{"id": "call-1", "type": "function", "function": {"name": "large", "arguments": "{}"}}], + tool_results=execution.tool_results, + ) + + assert parent.read(sidecar) + + async with root.fork_tape() as tape: + await tape.reset() + missing = await spill_read.run(handle=handle, context=ToolContext(tape=tape)) + assert "no spilled tool result" in missing + assert parent.read(sidecar) + + assert parent.read(sidecar) is None + assert "no spilled tool result" in await spill_read.run(handle=handle, context=ToolContext(tape=root)) + assert [entry.payload.get("name") for entry in parent.read(root.name) or [] if entry.kind == "anchor"] == [ + "session/start" + ] + + +@pytest.mark.asyncio +async def test_tape_archive_preserves_main_and_spill_as_sibling_tapes(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + root = _root_tape(tmp_path, parent) + sidecar = spill_tape_name(root.name) + await root.ensure_bootstrap_anchor() + + async with root.fork_tape() as tape: + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: "archived output\n" * 5000), {})], + context=ToolContext(tape=tape, run_id="run-1"), + ) + ref = execution.tool_results[0] + assert isinstance(ref, str) + handle = _handle_from_ref(ref) + await tape.record_chat( + run_id="run-1", + system_prompt=None, + new_messages=[{"role": "user", "content": "archive this"}], + response_text=None, + tool_calls=[{"id": "call-1", "type": "function", "function": {"name": "large", "arguments": "{}"}}], + tool_results=execution.tool_results, + ) + + result = await root.reset(archive=True) + + main_archive = Path(result.removeprefix("Archived: ")) + spill_archives = list(tmp_path.glob(f"{sidecar}.jsonl.*.bak")) + assert main_archive.exists() + assert len(spill_archives) == 1 + assert handle in main_archive.read_text(encoding="utf-8") + assert handle in spill_archives[0].read_text(encoding="utf-8") + assert parent.read(sidecar) is None + assert "no spilled tool result" in await spill_read.run(handle=handle, context=ToolContext(tape=root)) + + +@pytest.mark.asyncio +async def test_spill_sidecar_can_be_archived_and_reset_without_changing_main_context(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + root = _root_tape(tmp_path, parent) + sidecar = spill_tape_name(root.name) + await root.ensure_bootstrap_anchor() + + async with root.fork_tape() as tape: + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: "gc output\n" * 5000), {})], + context=ToolContext(tape=tape, run_id="run-1"), + ) + ref = execution.tool_results[0] + assert isinstance(ref, str) + handle = _handle_from_ref(ref) + await tape.record_chat( + run_id="run-1", + system_prompt=None, + new_messages=[{"role": "user", "content": "retain the main tape"}], + response_text=None, + tool_calls=[{"id": "call-1", "type": "function", "function": {"name": "large", "arguments": "{}"}}], + tool_results=execution.tool_results, + ) + + messages_before = await root.read_messages() + archive_result = await root.archive_sidecar("spill", reason="gc") + + assert archive_result.startswith("Archived spill: ") + assert parent.read(sidecar) + assert await root.read_messages() == messages_before + + reset_result = await root.reset_sidecar("spill", reason="gc") + + assert reset_result == "ok" + assert parent.read(sidecar) is None + assert await root.read_messages() == messages_before + assert "no spilled tool result" in await spill_read.run(handle=handle, context=ToolContext(tape=root)) + lifecycle_events = [ + entry + for entry in parent.read(root.name) or [] + if entry.kind == "event" and entry.payload.get("name") in {"sidecar.archive", "sidecar.reset"} + ] + assert [ + (entry.payload["name"], entry.payload["data"]["sidecar"], entry.payload["data"]["status"]) + for entry in lifecycle_events + ] == [ + ("sidecar.archive", "spill", "ok"), + ("sidecar.reset", "spill", "ok"), + ] + + +@pytest.mark.asyncio +async def test_failed_spill_archive_preserves_sidecar_without_blocking_main_reset(tmp_path: Path) -> None: + class BrokenSidecarArchiveStore(InMemoryTapeStore): + def fetch_all(self, query: Any) -> Any: + if query.tape.endswith("__spill"): + raise OSError("spill archive unavailable") + return super().fetch_all(query) + + parent = BrokenSidecarArchiveStore() + root = _root_tape(tmp_path, parent) + await root.ensure_bootstrap_anchor() + sidecar = spill_tape_name(root.name) + parent.append(sidecar, TapeEntry.event("spill.manifest")) + + result = await root.reset(archive=True) + + assert result.startswith("Archived: ") + assert parent.read(sidecar) + assert [entry.payload.get("name") for entry in parent.read(root.name) or [] if entry.kind == "anchor"] == [ + "session/start" + ] + lifecycle_events = [ + entry + for entry in parent.read(root.name) or [] + if entry.kind == "event" and entry.payload.get("name") in {"sidecar.archive", "sidecar.reset"} + ] + assert [ + (entry.payload["name"], entry.payload["data"]["sidecar"], entry.payload["data"]["status"]) + for entry in lifecycle_events + ] == [ + ("sidecar.archive", "spill", "error"), + ("sidecar.reset", "spill", "skipped"), + ] diff --git a/tests/test_tape.py b/tests/test_tape.py index 04e25335..fc7864a0 100644 --- a/tests/test_tape.py +++ b/tests/test_tape.py @@ -88,3 +88,23 @@ async def test_tape_info_omits_cache_hit_rate_when_usage_has_no_cache_details(tm info = await tape.info() assert info.last_token_cache_hit_rate is None + + +@pytest.mark.asyncio +async def test_context_excluded_entries_do_not_reach_custom_context_selectors(tmp_path: Path) -> None: + def select_events(entries, _context): + return [ + {"role": "assistant", "content": str(entry.payload.get("name"))} + for entry in entries + if entry.kind == "event" + ] + + tape = Tape( + tmp_path, + AsyncTapeStoreAdapter(InMemoryTapeStore()), + TapeContext(anchor=None, select=select_events), + ).scoped("test-tape") + await tape.append_event("visible", {}) + await tape.append_event("hidden", {}, context=False) + + assert await tape.read_messages() == [{"role": "assistant", "content": "visible"}] diff --git a/website/src/content/docs/docs/build/hooks.mdx b/website/src/content/docs/docs/build/hooks.mdx index 80db0092..7721c3cb 100644 --- a/website/src/content/docs/docs/build/hooks.mdx +++ b/website/src/content/docs/docs/build/hooks.mdx @@ -135,6 +135,27 @@ def provide_tape_store(): The full plugin lives at [`bub-tapestore-sqlite`](https://github.com/bubbuild/bub-contrib/tree/main/packages/bub-tapestore-sqlite). For stores that need cleanup, return a generator instead — Bub treats it as a context manager. +### Mount a tape sidecar + +Use `provide_tape_sidecars` when a plugin needs a sibling tape with the same lifecycle as the session tape. A sidecar only declares a stable `name`; its plugin owns configuration and the data format. + +```python +from bub import hookimpl + + +class ArtifactSidecar: + name = "artifacts" + + +@hookimpl +def provide_tape_sidecars(): + return [ArtifactSidecar()] +``` + +For a session tape named `session`, this mounts `session__artifacts` in the active `TapeStore`. Tool code can retrieve its provider with `context.tape.get_sidecar("artifacts")` and resolve the physical name with `context.tape.sidecar_tape_name("artifacts")`. The main tape handles fork, merge, archive, and reset for every mounted sidecar. Providers with the same name follow normal hook priority: the first one wins. + +The builtin spill plugin owns `SpillStore`, the `spill.read` tool, its data format, and `SpillSettings`. The core does not import or recognize the spill implementation. Configuration uses the `spill:` section and `BUB_SPILL_*` environment variables independently of `AgentSettings`. + ## 6. Add a channel A **channel** is an inbound/outbound surface — CLI, Telegram, WeChat, a scheduled trigger. `provide_channels` lets your plugin contribute one or more `Channel` subclasses. @@ -266,7 +287,7 @@ class ShellPolicy: return None # proceed unchanged ``` -Observe terminal outcomes — `after_llm_call` / `after_tool_call` receive the original exception object on failure (`result.error`); cancelled calls do not produce an observation: +Handle terminal outcomes — `after_llm_call` / `after_tool_call` receive the original exception object on failure (`result.error`); cancelled calls do not produce an observation. An `after_tool_call` implementation may replace a successful value by assigning `result.result`: ```python from bub import hookimpl @@ -281,6 +302,8 @@ class Metrics: @hookimpl def after_tool_call(self, call, result, state): print(f"tool {call.tool} {result.duration_ms}ms error={result.error!r}") + if result.error is None and call.tool == "web_search": + result.result = str(result.result)[:4_000] ``` All payloads carry `run_id`, matching the tape entry meta, so metrics can be joined against the recorded conversation. diff --git a/website/src/content/docs/docs/concepts/tape-and-context.mdx b/website/src/content/docs/docs/concepts/tape-and-context.mdx index 0eec86e9..340ad5e2 100644 --- a/website/src/content/docs/docs/concepts/tape-and-context.mdx +++ b/website/src/content/docs/docs/concepts/tape-and-context.mdx @@ -1,6 +1,6 @@ --- title: Tape and context -description: The tape primitives, append-only invariant, anchors and handoffs, and how Bub turns a per-session tape into a model context window. +description: The tape primitives, append-only invariant, spill sidecars, anchors and handoffs, and how Bub turns a per-session tape into a model context window. sidebar: order: 3 --- @@ -48,6 +48,18 @@ This means the same `session_id` produces a different tape in a different worksp The default `provide_tape_store` returns a `FileTapeStore` rooted at `~/.bub/tapes/`. Plugins replace it by providing their own `provide_tape_store` (for example, a SQLite or HTTP-backed store). +### Spill sidecar + +The builtin spill plugin mounts a `SpillStore` through `provide_tape_sidecars`, registers `spill.read`, and bounds large results through the existing `after_tool_call` hook. The sidecar itself has no tool-result interception contract. Large results are stored in a sibling tape named `__spill`. The sidecar uses the same `TapeStore` as the session tape, so existing storage plugins do not need a spill-specific interface. Bub writes UTF-8-safe chunks followed by a manifest; the manifest is the completion marker for the stored result. + +The main tape keeps a bounded preview and an opaque handle instead of the complete result. The `spill.read` tool reads a bounded page by handle and cursor, including pages counted from the end. Content explicitly returned by `spill.read` is recorded as a normal bounded tool result. + +The sidecar follows the session tape through forks, merges, archive, and reset, but remains a separate tape and is never scanned while constructing the main context. It can also be archived or reset independently with `Tape.archive_sidecar("spill")` and `Tape.reset_sidecar("spill")`. When reset requests an archive, Bub preserves the sidecar if that archive fails. + +Spill configuration belongs to the sidecar plugin: use `spill.threshold` in `config.yml` or `BUB_SPILL_THRESHOLD` in the environment. Setting it to `0` stops new writes without unmounting the sidecar, so existing handles remain readable and tape lifecycle operations still include it. + +Spill writes are recorded as `spill.write`. Framework-managed lifecycle outcomes use `sidecar.archive`, `sidecar.reset`, and `sidecar.merge`; lifecycle data identifies the affected sidecar or physical tape. These entries are marked as context-excluded before any context selector runs, so they remain available for operations and audit without changing model messages or prompt-cache prefixes. A sidecar persistence failure is recorded but does not prevent the main tape from merging or resetting. + ### ensure_bootstrap_anchor Before the first turn on a tape, `TapeService.ensure_bootstrap_anchor` checks for an anchor entry. If none exists, it writes a `session/start` handoff with `state={"owner": "human"}`. This guarantees that context reconstruction has a starting anchor on every tape. @@ -63,6 +75,8 @@ Before the first turn on a tape, `TapeService.ensure_bootstrap_anchor` checks fo The context selector is a hook (`build_tape_context`), so plugins can replace it with a different strategy — compaction, summarization, retrieval — without touching the rest of the pipeline. +Entries marked `context=False` are removed before the selector runs. This is how operational spill events remain queryable on the tape without appearing in either the default context or a plugin-defined context. + ### fork_tape `TapeService.fork_tape(tape_name, merge_back=True)` is an async context manager backed by `ForkTapeStore.fork`. Inside the block, writes happen on a forked tape; on exit, they are merged back into the parent tape (or discarded if `merge_back=False`). Use this to run a sub-task without polluting the parent session's history until you decide to keep the result. diff --git a/website/src/content/docs/docs/concepts/turn-pipeline.mdx b/website/src/content/docs/docs/concepts/turn-pipeline.mdx index 55437776..d291543d 100644 --- a/website/src/content/docs/docs/concepts/turn-pipeline.mdx +++ b/website/src/content/docs/docs/concepts/turn-pipeline.mdx @@ -92,6 +92,7 @@ The default observer (`BuiltinImpl.on_error`) sends an error envelope through `d - `dispatch_outbound` — forwards through the bound `ChannelRouter`. - `system_prompt` — combines a default prompt with the workspace `AGENTS.md`. - `provide_tape_store` — file-backed tape store under `~/.bub/tapes`. +- `provide_tape_sidecars` — the configured builtin spill sidecar. - `provide_channels` — registers the built-in `cli` and `telegram` adapters. Plugins override any of these by registering a higher-priority implementation; later-registered plugins run first. diff --git a/website/src/content/docs/docs/reference/hooks.mdx b/website/src/content/docs/docs/reference/hooks.mdx index b083cdd1..93e529f8 100644 --- a/website/src/content/docs/docs/reference/hooks.mdx +++ b/website/src/content/docs/docs/reference/hooks.mdx @@ -29,13 +29,14 @@ For the *why* and *how* of each stage see [Turn pipeline](/docs/concepts/turn-pi | `on_error` | observer | `(stage: str, error: Exception, message: Envelope \| None) -> None` | none | `HookRuntime.notify_error` / `notify_error_sync` | Failures inside an `on_error` impl are caught and logged so other observers still run. | | `system_prompt` | broadcast (joined) | `(prompt, state) -> str` | prompt fragment | `BubFramework.get_system_prompt` (`call_many_sync`) | Results are reversed and joined with `\n\n`; truthy fragments only. | | `provide_tape_store` | firstresult | `() -> TapeStore \| AsyncTapeStore` | tape store | `BubFramework.running()` | Resolved once when the runtime scope opens; sync/async iterators are entered as context managers. | +| `provide_tape_sidecars` | sync-only consumer (deduped) | `() -> list[TapeSidecar]` | mounted sidecars | `BubFramework.get_tape_sidecars` | Sidecars are deduplicated by `name`; the first value in hook priority order wins. | | `provide_channels` | sync-only consumer (deduped) | `(message_handler: MessageHandler) -> list[Channel]` | channels | `BubFramework.get_channels` (`call_many_sync`) | Channels are deduplicated by `Channel.name`; the first channel seen in hook priority order wins. | | `build_tape_context` | firstresult | `() -> TapeContext` | tape context | `BubFramework.build_tape_context` (`call_first_sync`) | Sync-only; awaitable returns are skipped. | | `admit_message` | firstresult | `(session_id, message, turn) -> AdmitDecision \| None` | turn admission decision | `ChannelManager` | Runs before channel scheduling. `None` keeps default concurrent scheduling; decision types are listed in [Core contracts](/docs/reference/types/). | | `before_llm_call` | chained | `(request: LlmCallRequest, state: TurnState) -> LlmCallRequest \| LlmCallDecision \| None` | modified request or finish decision | `ModelRunner.run` via `AgentHooks` | Impls chain in LIFO order; each sees the previous impl's request. `LlmCallDecision.finish(text)` skips the provider call. Raising impls are logged and skipped. | | `after_llm_call` | observer | `(request: LlmCallRequest, result: LlmCallResult, state: TurnState) -> None` | none | `ModelRunner.run` via `AgentHooks` | Fires exactly once per completed call: success or `Exception` failure. Cancellation / consumer `aclose()` is not observed. `result.error` is the original exception. | | `before_tool_call` | chained | `(call: ToolCall, state: TurnState) -> ToolCallDecision \| None` | decision | `ToolExecutor` via `AgentHooks` | Per tool invocation. `proceed(arguments=…)` folds argument changes; `replace(result)` / `deny(message)` short-circuit. Veto only via the decision object — exceptions are logged and skipped. | -| `after_tool_call` | observer | `(call: ToolCall, result: ToolCallResult, state: TurnState) -> None` | none | `ToolExecutor` via `AgentHooks` | Fires for success, failure and deny/replace. Cancellation is not observed. `result.error` is the original `BubError`. | +| `after_tool_call` | terminal handler | `(call: ToolCall, result: ToolCallResult, state: TurnState) -> None` | none | `ToolExecutor` via `AgentHooks` | Fires for success, failure and deny/replace. Cancellation is not observed. Assigning `result.result` replaces the successful value returned to the model; `result.error` is the original `BubError`. | ## How hooks are invoked @@ -67,7 +68,7 @@ def _iter_hookimpls(self, hook_name: str) -> list[Any]: Async calls (`call_first`, `call_many`) `await` any awaitable result. Sync calls (`*_sync`) check `inspect.isawaitable(value)` and emit `hook.async_not_supported hook= adapter=`, then skip the value. -Bootstrap hooks **must** be synchronous: `register_cli_commands`, `onboard_config`, `provide_channels`, `provide_tape_store`, `build_tape_context`, plus `system_prompt`. +Bootstrap hooks **must** be synchronous: `register_cli_commands`, `onboard_config`, `provide_channels`, `provide_tape_store`, `provide_tape_sidecars`, `build_tape_context`, plus `system_prompt`. :::caution Defining an async coroutine implementation for a sync-only hook registers the hook, but runtime sync dispatch logs `hook.async_not_supported` and skips its return value. `bub hooks` confirms discovery, not that the implementation is usable in sync dispatch. @@ -93,6 +94,10 @@ Each impl receives only the kwargs it declares. You can omit unused parameters f `BubFramework.get_tape_store()` returns `None` outside the scope. +### Tape sidecars + +`provide_tape_sidecars` contributes named capabilities backed by sibling tapes in the active `TapeStore`. Bub mounts the combined set when it constructs the agent's root `Tape`; a scoped session then maps sidecar `spill` to `__spill`. Fork, merge, archive, and reset operate on every mounted sidecar without requiring a new storage interface. Removing a provider stops mounting its sidecar but does not delete stored data. + ### `on_error` observer safety `notify_error` and `notify_error_sync` wrap each impl in a `try`/`except`; observer failures are logged (`hook.on_error_failed stage=… adapter=…`) but never propagate. This guarantees one broken observer does not block the others, and prevents an `on_error` from masking the original exception. @@ -103,7 +108,7 @@ The four `*_llm_call` / `*_tool_call` hooks are dispatched through the `AgentHoo - **Chaining** — `before_llm_call` and `before_tool_call` run every implementation in LIFO order and *fold* modifications: each impl receives the request/call as modified by the impls before it. The first short-circuiting decision (`LlmCallDecision.finish`, `ToolCallDecision.replace`/`deny`) stops the chain. - **Fault isolation** — a raising implementation is logged (`hook.agent_hook_failed`) and skipped, never fatal to the turn. Blocking is only expressible through decision objects, so a broken policy plugin cannot veto by crashing. -- **Exactly-once terminal observation** — `after_llm_call` and `after_tool_call` fire exactly once per call for real completions: success or `Exception` failure. Cancellation and consumer close (`BaseException`) intentionally bypass after hooks. `result.error` carries the **original exception object** (a `BubError` for tool failures with `kind`/`details` intact). +- **Exactly-once terminal handling** — `after_llm_call` and `after_tool_call` fire exactly once per call for real completions: success or `Exception` failure. Cancellation and consumer close (`BaseException`) intentionally bypass after hooks. `result.error` carries the **original exception object** (a `BubError` for tool failures with `kind`/`details` intact). `after_tool_call` implementations share one outcome, so assigning `result.result` changes the successful value seen by later implementations and the model. Payload dataclasses (`LlmCallRequest`, `LlmCallResult`, `ToolCall`, `ToolCallDecision`, `ToolCallResult`, `LlmCallDecision`) live beside those semantics in `src/bub/hooks/interception.py`. Every payload carries `run_id`, matching the `run_id` meta on tape entries, so observers can correlate hook events with the tape. Rewritten `model`/`max_tokens` from `before_llm_call` are honored end-to-end: the provider receives them and the tape records the effective model. diff --git a/website/src/content/docs/docs/reference/settings.mdx b/website/src/content/docs/docs/reference/settings.mdx index df2cd8c3..ee06c0a1 100644 --- a/website/src/content/docs/docs/reference/settings.mdx +++ b/website/src/content/docs/docs/reference/settings.mdx @@ -73,6 +73,23 @@ Provider-specific defaults are gathered at startup by scanning `os.environ` for OpenAI Codex OAuth does not need a separate request-format setting. After `bub login openai`, Bub detects the stored Codex OAuth token when the selected model uses the `openai:` provider and no custom API base is set. +## Spill sidecar — `SpillSettings` + +Defined in `src/bub/builtin/spill.py` and registered by the builtin sidecar plugin: + +```python +@config(name="spill") +class SpillSettings(Settings): + model_config = SettingsConfigDict(env_prefix="BUB_SPILL_", extra="ignore", env_file=".env") + threshold: int = Field(default=4096, ge=0) +``` + +Loaded under the YAML `spill:` section. + +| Env var | Default | YAML key (`spill.*`) | Description | +| --- | --- | --- | --- | +| `BUB_SPILL_THRESHOLD` | `4096` | `threshold` | Estimated tokens (4 chars each) above which string tool results are stored in the spill sidecar. Set to `0` to stop creating new spills while keeping the sidecar mounted for existing handles and lifecycle operations. | + ## Channels — `ChannelSettings` Defined in `src/bub/channels/manager.py`: diff --git a/website/src/content/docs/docs/reference/types.mdx b/website/src/content/docs/docs/reference/types.mdx index 5add963d..912083aa 100644 --- a/website/src/content/docs/docs/reference/types.mdx +++ b/website/src/content/docs/docs/reference/types.mdx @@ -91,9 +91,10 @@ from bub.hooks.interception import ( ToolCallResult, ) from bub.hooks.runtime import HookRuntime +from bub.sidecars import TapeSidecar, sidecar_tape_name ``` -Plugin authors normally need only `hookimpl` plus the payload types used by their hooks. See the [Hook reference](/docs/reference/hooks/) for dispatch and fault-isolation semantics. +Plugin authors normally need only `hookimpl` plus the payload types used by their hooks. `TapeSidecar` is the minimal named contract returned by `provide_tape_sidecars`; `sidecar_tape_name` resolves its sibling tape name. See the [Hook reference](/docs/reference/hooks/) for dispatch and fault-isolation semantics. ## Channel contracts diff --git a/website/src/content/docs/zh-cn/docs/build/hooks.mdx b/website/src/content/docs/zh-cn/docs/build/hooks.mdx index 198b387d..fb6c6e4c 100644 --- a/website/src/content/docs/zh-cn/docs/build/hooks.mdx +++ b/website/src/content/docs/zh-cn/docs/build/hooks.mdx @@ -135,6 +135,27 @@ def provide_tape_store(): 完整插件位于 [`bub-tapestore-sqlite`](https://github.com/bubbuild/bub-contrib/tree/main/packages/bub-tapestore-sqlite)。需要清理的库可改为返回生成器 —— Bub 会以上下文管理器对待它。 +### 挂载 tape sidecar + +当插件需要一个与 session tape 共享生命周期的 sibling tape 时，使用 `provide_tape_sidecars`。sidecar 只声明稳定的 `name`；配置与数据格式由插件自己管理。 + +```python +from bub import hookimpl + + +class ArtifactSidecar: + name = "artifacts" + + +@hookimpl +def provide_tape_sidecars(): + return [ArtifactSidecar()] +``` + +对名为 `session` 的 tape，这会在 active `TapeStore` 中挂载 `session__artifacts`。工具代码可通过 `context.tape.get_sidecar("artifacts")` 取得 provider，并用 `context.tape.sidecar_tape_name("artifacts")` 得到物理名称。主 tape 负责所有已挂载 sidecar 的 fork、merge、archive 和 reset。同名 provider 遵循普通 hook 优先级：最先出现的实现生效。 + +builtin spill 插件拥有 `SpillStore`、`spill.read` 工具、数据格式和 `SpillSettings`。核心层不导入或识别 spill 实现。配置使用 `spill:` section 和 `BUB_SPILL_*` 环境变量，与 `AgentSettings` 相互独立。 + ## 6. 新增通道 **通道**是一个收发端 —— CLI、Telegram、微信、定时触发器。`provide_channels` 让插件贡献一个或多个 `Channel` 子类。 @@ -266,7 +287,7 @@ class ShellPolicy: return None # proceed unchanged ``` -观察终态 —— `after_llm_call` / `after_tool_call` 在失败时拿到**原始异常对象**(`result.error`);被取消的调用不产生观察: +处理终态 —— `after_llm_call` / `after_tool_call` 在失败时拿到**原始异常对象**（`result.error`）；被取消的调用不产生观察。`after_tool_call` 实现可以通过赋值 `result.result` 替换成功结果： ```python from bub import hookimpl @@ -281,6 +302,8 @@ class Metrics: @hookimpl def after_tool_call(self, call, result, state): print(f"tool {call.tool} {result.duration_ms}ms error={result.error!r}") + if result.error is None and call.tool == "web_search": + result.result = str(result.result)[:4_000] ``` 所有载荷携带 `run_id`,与 tape 条目 meta 对应,指标可与记录的会话对齐。 diff --git a/website/src/content/docs/zh-cn/docs/concepts/tape-and-context.mdx b/website/src/content/docs/zh-cn/docs/concepts/tape-and-context.mdx index 7b855436..c1768159 100644 --- a/website/src/content/docs/zh-cn/docs/concepts/tape-and-context.mdx +++ b/website/src/content/docs/zh-cn/docs/concepts/tape-and-context.mdx @@ -1,6 +1,6 @@ --- title: Tape 与 context -description: tape 原语、append-only 不变式、anchor 与 handoff，以及 Bub 如何把 per-session tape 转成模型 context window。 +description: tape 原语、append-only 不变式、spill sidecar、anchor 与 handoff，以及 Bub 如何把 per-session tape 转成模型 context window。 sidebar: order: 3 --- @@ -48,6 +48,18 @@ tape_name = f"{workspace_hash}__{session_hash}" 默认的 `provide_tape_store` 返回根目录在 `~/.bub/tapes/` 的 `FileTapeStore`。插件通过提供自己的 `provide_tape_store`（例如 SQLite 或 HTTP 后端）来替换它。 +### spill sidecar + +builtin spill 插件通过 `provide_tape_sidecars` 挂载 `SpillStore`、注册 `spill.read`，并通过现有 `after_tool_call` hook 限制大结果。sidecar 本身没有工具结果拦截约定。较大的结果存放在名为 `__spill` 的 sibling tape 中。sidecar 与 session tape 使用同一个 `TapeStore`，因此现有存储插件不需要实现 spill 专用接口。Bub 依次写入 UTF-8 安全的 chunk，最后写入 manifest；manifest 是该结果已完整存储的提交标记。 + +主 tape 只保留有界预览和 opaque handle，不保存完整结果。`spill.read` 工具按 handle 与 cursor 有界读取，也支持从末尾开始读取。由 `spill.read` 明确返回的内容会作为普通的有界 tool result 记录。 + +sidecar 与 session tape 一起参与 fork、merge、archive 和 reset，但始终是独立 tape，构造主 context 时不会扫描它。也可以通过 `Tape.archive_sidecar("spill")` 和 `Tape.reset_sidecar("spill")` 单独 archive 或 reset sidecar。当 reset 要求先 archive 时，如果 archive 失败，Bub 会保留 sidecar。 + +spill 配置归 sidecar 插件自己所有：在 `config.yml` 使用 `spill.threshold`，或设置环境变量 `BUB_SPILL_THRESHOLD`。设为 `0` 只停止新写入，不会卸载 sidecar，因此已有 handle 仍可读取，tape 生命周期操作也仍会包含它。 + +spill 写入以 `spill.write` event 记录。框架管理的生命周期结果使用 `sidecar.archive`、`sidecar.reset` 和 `sidecar.merge`，其数据会标明受影响的 sidecar 或物理 tape。它们会在任何 context selector 运行前被排除，因此可用于运维和审计，同时不会改变模型消息或 prompt cache 前缀。sidecar 持久化失败会被记录，但不会阻止主 tape merge 或 reset。 + ### ensure_bootstrap_anchor 在某条 tape 的第一次 turn 之前，`TapeService.ensure_bootstrap_anchor` 会检查是否存在 anchor entry。如果没有，则写入一条 `session/start` handoff，`state={"owner": "human"}`。这保证每条 tape 在 context 重建时都有起始 anchor。 @@ -63,6 +75,8 @@ tape_name = f"{workspace_hash}__{session_hash}" context selector 本身是个 hook（`build_tape_context`），插件可以用其他策略（压缩、摘要、检索）替换它，而无需触动 pipeline 其余部分。 +带有 `context=False` 标记的 entry 会在 selector 运行前被移除。spill 运维 event 因此仍可在 tape 上查询，但不会进入默认 context 或插件自定义 context。 + ### fork_tape `TapeService.fork_tape(tape_name, merge_back=True)` 是由 `ForkTapeStore.fork` 支撑的 async context manager。块内的写入发生在 fork 后的 tape 上；退出时合并回父 tape（若 `merge_back=False` 则丢弃）。可以用它跑子任务，避免污染父 session 的历史，直到决定保留结果。 diff --git a/website/src/content/docs/zh-cn/docs/concepts/turn-pipeline.mdx b/website/src/content/docs/zh-cn/docs/concepts/turn-pipeline.mdx index e90fe38f..52992ee9 100644 --- a/website/src/content/docs/zh-cn/docs/concepts/turn-pipeline.mdx +++ b/website/src/content/docs/zh-cn/docs/concepts/turn-pipeline.mdx @@ -92,6 +92,7 @@ hook 运行时会在两者间互相适配：若插件只实现 `run_model_stream - `dispatch_outbound` — 转发到绑定的 `ChannelRouter`。 - `system_prompt` — 将默认 prompt 与 workspace 的 `AGENTS.md` 拼接。 - `provide_tape_store` — 位于 `~/.bub/tapes` 下的文件型 tape store。 +- `provide_tape_sidecars` — 使用当前配置的 builtin spill sidecar。 - `provide_channels` — 注册内置的 `cli` 与 `telegram` adapter。 插件通过注册更高优先级的实现来覆写其中任一项；晚注册的插件先执行。 diff --git a/website/src/content/docs/zh-cn/docs/reference/hooks.mdx b/website/src/content/docs/zh-cn/docs/reference/hooks.mdx index 1cd6a54f..5fcf08a3 100644 --- a/website/src/content/docs/zh-cn/docs/reference/hooks.mdx +++ b/website/src/content/docs/zh-cn/docs/reference/hooks.mdx @@ -29,13 +29,14 @@ description: BubHookSpecs 中每个钩子的类型、签名、返回值与调用 | `on_error` | observer | `(stage: str, error: Exception, message: Envelope \| None) -> None` | none | `HookRuntime.notify_error` / `notify_error_sync` | `on_error` 实现内部抛出的异常会被吞掉并写日志,确保其他观察者继续运行。 | | `system_prompt` | broadcast (joined) | `(prompt, state) -> str` | prompt fragment | `BubFramework.get_system_prompt` (`call_many_sync`) | 结果先反转再用 `\n\n` 拼接,只保留真值片段。 | | `provide_tape_store` | firstresult | `() -> TapeStore \| AsyncTapeStore` | tape store | `BubFramework.running()` | 仅在 runtime 作用域开启时解析一次;返回同步或异步迭代器时会被作为 context manager 进入。 | +| `provide_tape_sidecars` | sync-only consumer（去重） | `() -> list[TapeSidecar]` | 挂载的 sidecar | `BubFramework.get_tape_sidecars` | 按 `name` 去重；hook 优先级中最先出现的值生效。 | | `provide_channels` | sync-only consumer (deduped) | `(message_handler: MessageHandler) -> list[Channel]` | channels | `BubFramework.get_channels` (`call_many_sync`) | 按 `Channel.name` 去重；在钩子优先级顺序中最先出现的 channel 胜出。 | | `build_tape_context` | firstresult | `() -> TapeContext` | tape context | `BubFramework.build_tape_context` (`call_first_sync`) | 仅同步;awaitable 返回会被跳过。 | | `admit_message` | firstresult | `(session_id, message, turn) -> AdmitDecision \| None` | turn admission decision | `ChannelManager` | 调度 channel message 前调用。返回 `None` 保持默认并发调度；decision 类型见 [核心契约](/zh-cn/docs/reference/types/)。 | | `before_llm_call` | chained | `(request: LlmCallRequest, state: TurnState) -> LlmCallRequest \| LlmCallDecision \| None` | 修改后的 request 或 finish 决定 | `ModelRunner.run` 经 `AgentHooks` | 实现按 LIFO 链式执行,每个实现看到的是前一个实现修改后的 request。返回 `LlmCallDecision.finish(text)` 跳过 provider 调用。实现抛异常仅记日志并跳过。 | | `after_llm_call` | observer | `(request: LlmCallRequest, result: LlmCallResult, state: TurnState) -> None` | 无 | `ModelRunner.run` 经 `AgentHooks` | 每次完成的调用恰好触发一次:成功或 `Exception` 失败。取消 / 消费方 `aclose()` 不观察。`result.error` 是原始异常。 | | `before_tool_call` | chained | `(call: ToolCall, state: TurnState) -> ToolCallDecision \| None` | decision | `ToolExecutor` 经 `AgentHooks` | 逐次工具调用。`proceed(arguments=…)` 折叠参数修改;`replace(result)` / `deny(message)` 短路。否决只能通过 decision 对象——异常仅记日志并跳过。 | -| `after_tool_call` | observer | `(call: ToolCall, result: ToolCallResult, state: TurnState) -> None` | 无 | `ToolExecutor` 经 `AgentHooks` | 成功、失败、deny/replace 会触发。取消不观察。`result.error` 是原始 `BubError`。 | +| `after_tool_call` | 终态处理 | `(call: ToolCall, result: ToolCallResult, state: TurnState) -> None` | 无 | `ToolExecutor` 经 `AgentHooks` | 成功、失败、deny/replace 会触发。取消不观察。赋值 `result.result` 会替换返回给模型的成功结果；`result.error` 是原始 `BubError`。 | ## 钩子如何被调用 @@ -67,7 +68,7 @@ def _iter_hookimpls(self, hook_name: str) -> list[Any]: 异步调用 (`call_first`、`call_many`) 会 `await` 任何 awaitable 返回值。同步调用 (`*_sync`) 通过 `inspect.isawaitable(value)` 判断,若为真则发出 `hook.async_not_supported hook= adapter=` 告警并跳过该值。 -启动期钩子 **必须** 同步:`register_cli_commands`、`onboard_config`、`provide_channels`、`provide_tape_store`、`build_tape_context`,以及 `system_prompt`。 +启动期钩子 **必须** 同步:`register_cli_commands`、`onboard_config`、`provide_channels`、`provide_tape_store`、`provide_tape_sidecars`、`build_tape_context`,以及 `system_prompt`。 :::caution 为 sync-only hook 定义 async coroutine 实现时，hook 仍会被注册；但同步分发会记录 `hook.async_not_supported` 并跳过其返回值。`bub hooks` 只能确认发现成功，不能确认它会在同步分发中生效。 @@ -93,6 +94,10 @@ def _kwargs_for_impl(impl: Any, kwargs: dict[str, Any]) -> dict[str, Any]: `BubFramework.get_tape_store()` 在作用域之外返回 `None`。 +### Tape sidecar + +`provide_tape_sidecars` 提供由 active `TapeStore` 中 sibling tape 支撑的具名能力。Bub 在创建 agent root `Tape` 时挂载聚合结果；例如 scoped session 会把 sidecar `spill` 映射为 `__spill`。fork、merge、archive 和 reset 会处理所有已挂载 sidecar，不要求存储插件实现新接口。移除 provider 只会停止挂载，不会删除已存数据。 + ### `on_error` 观察者安全性 `notify_error` 与 `notify_error_sync` 把每个实现包在 `try`/`except` 中;观察者失败会写入日志 (`hook.on_error_failed stage=… adapter=…`) 但不会向上传递。这样可以保证某个观察者出错不会阻塞其他观察者,也不会让 `on_error` 掩盖原始异常。 @@ -103,7 +108,7 @@ def _kwargs_for_impl(impl: Any, kwargs: dict[str, Any]) -> dict[str, Any]: - **链式** — `before_llm_call` 与 `before_tool_call` 按 LIFO 顺序执行全部实现并**折叠**修改:每个实现收到的是前序实现修改后的 request/call。第一个短路决定(`LlmCallDecision.finish`、`ToolCallDecision.replace`/`deny`)终止链。 - **错误隔离** — 实现抛异常只记日志(`hook.agent_hook_failed`)并跳过,绝不炸掉回合。阻断只能通过 decision 对象表达,坏掉的策略插件无法靠 crash 否决。 -- **终态恰好一次** — `after_llm_call` 与 `after_tool_call` 对真实完成的调用恰好触发一次:成功或 `Exception` 失败。取消与消费方关闭(`BaseException`)刻意不进入 after 钩子。`result.error` 携带**原始异常对象**(工具失败为完整 `BubError`,`kind`/`details` 保留)。 +- **终态处理恰好一次** — `after_llm_call` 与 `after_tool_call` 对真实完成的调用恰好触发一次：成功或 `Exception` 失败。取消与消费方关闭（`BaseException`）刻意不进入 after hook。`result.error` 携带**原始异常对象**（工具失败为完整 `BubError`，保留 `kind`/`details`）。所有 `after_tool_call` 实现共享同一个 outcome，因此赋值 `result.result` 会改变后续实现以及模型看到的成功结果。 载荷数据类(`LlmCallRequest`、`LlmCallResult`、`ToolCall`、`ToolCallDecision`、`ToolCallResult`、`LlmCallDecision`)与执行语义一同位于 `src/bub/hooks/interception.py`。所有载荷携带 `run_id`,与 tape 条目 meta 中的 `run_id` 对应,观察者可将钩子事件与 tape 记录对齐。`before_llm_call` 改写的 `model`/`max_tokens` 全链路生效:provider 收到改写值,tape 记录 effective model。 diff --git a/website/src/content/docs/zh-cn/docs/reference/settings.mdx b/website/src/content/docs/zh-cn/docs/reference/settings.mdx index 5a0fcc4d..fbb07b77 100644 --- a/website/src/content/docs/zh-cn/docs/reference/settings.mdx +++ b/website/src/content/docs/zh-cn/docs/reference/settings.mdx @@ -73,6 +73,23 @@ class AgentSettings(Settings): OpenAI Codex OAuth 不需要单独的请求格式设置。运行 `bub login openai` 后,当模型使用 `openai:` provider 且没有自定义 API base 时,Bub 会检测并使用本地保存的 Codex OAuth token。 +## Spill sidecar —— `SpillSettings` + +定义于 `src/bub/builtin/spill.py`，由 builtin sidecar 插件注册： + +```python +@config(name="spill") +class SpillSettings(Settings): + model_config = SettingsConfigDict(env_prefix="BUB_SPILL_", extra="ignore", env_file=".env") + threshold: int = Field(default=4096, ge=0) +``` + +从 YAML 的 `spill:` section 加载。 + +| 环境变量 | 默认值 | YAML 字段（`spill.*`） | 描述 | +| --- | --- | --- | --- | +| `BUB_SPILL_THRESHOLD` | `4096` | `threshold` | 字符串工具结果超过该估算 token 数（每 token 按 4 字符估算）时写入 spill sidecar。设为 `0` 会停止产生新 spill，但 sidecar 仍保持挂载，已有 handle 和生命周期操作不受影响。 | + ## Channels —— `ChannelSettings` 定义于 `src/bub/channels/manager.py`: diff --git a/website/src/content/docs/zh-cn/docs/reference/types.mdx b/website/src/content/docs/zh-cn/docs/reference/types.mdx index 52e7066c..f437451a 100644 --- a/website/src/content/docs/zh-cn/docs/reference/types.mdx +++ b/website/src/content/docs/zh-cn/docs/reference/types.mdx @@ -91,9 +91,10 @@ from bub.hooks.interception import ( ToolCallResult, ) from bub.hooks.runtime import HookRuntime +from bub.sidecars import TapeSidecar, sidecar_tape_name ``` -插件作者通常只需要 `hookimpl` 与对应 hook 使用的 payload 类型。分发和故障隔离语义见 [Hook 参考](/zh-cn/docs/reference/hooks/)。 +插件作者通常只需要 `hookimpl` 与对应 hook 使用的 payload 类型。`TapeSidecar` 是 `provide_tape_sidecars` 返回的最小具名契约；`sidecar_tape_name` 用于解析 sibling tape 名称。分发和故障隔离语义见 [Hook 参考](/zh-cn/docs/reference/hooks/)。 ## Channel 契约