From 0dba7aeb8a5602ee5ce32c75b1555fe4f2079676 Mon Sep 17 00:00:00 2001 From: PsiACE Date: Mon, 24 Aug 2026 02:49:36 +0800 Subject: [PATCH 1/4] refactor: move fork merge into sidecar --- src/bub/builtin/agent.py | 34 ++++-- src/bub/builtin/forkmerge.py | 152 +++++++++++++++++++++++ src/bub/builtin/hook_impl.py | 27 ++-- src/bub/builtin/spill.py | 1 + src/bub/builtin/tools.py | 37 ++++-- src/bub/sidecars.py | 12 +- src/bub/store.py | 102 +--------------- src/bub/tape.py | 32 ++--- tests/test_builtin_agent.py | 144 ++++------------------ tests/test_builtin_hook_impl.py | 25 +++- tests/test_file_tape_store_entry_ids.py | 28 +++-- tests/test_fork_store_merge_back.py | 101 --------------- tests/test_spill.py | 156 +++++++++++++----------- tests/test_subagent_tool.py | 42 +++++-- tests/test_tape.py | 76 +++++++++--- 15 files changed, 482 insertions(+), 487 deletions(-) create mode 100644 src/bub/builtin/forkmerge.py delete mode 100644 tests/test_fork_store_merge_back.py diff --git a/src/bub/builtin/agent.py b/src/bub/builtin/agent.py index 3924c4d9..c7d90803 100644 --- a/src/bub/builtin/agent.py +++ b/src/bub/builtin/agent.py @@ -8,7 +8,6 @@ import shlex import time from collections.abc import AsyncGenerator, AsyncIterator, Callable, Collection, Coroutine, Iterable -from contextlib import AsyncExitStack from dataclasses import dataclass, replace from datetime import UTC, datetime from functools import cached_property @@ -74,7 +73,7 @@ async def generator() -> AsyncIterator: return AsyncStreamEvents(generator()) @staticmethod - def _events_with_callback( + def finalize_stream( events: AsyncStreamEvents, callback: Callable[[], Coroutine[Any, Any, Any]] ) -> AsyncStreamEvents: async def generator() -> AsyncIterator[StreamEvent]: @@ -89,9 +88,8 @@ async def generator() -> AsyncIterator[StreamEvent]: async def run_stream( self, *, - session_id: str, + tape: Tape, prompt: str | list[dict], - state: TurnState, model: str | None = None, allowed_skills: Collection[str] | None = None, allowed_tools: Collection[str] | None = None, @@ -102,14 +100,6 @@ async def run_stream( StreamEvent("final", {"text": "error: empty prompt", "ok": False}), ]) - state.setdefault("session_id", session_id) - tape = self.tape.session_tape( - session_id, workspace_from_state(state), context=replace(self.tape.context, state=state) - ) - merge_back = not session_id.startswith("temp/") - stack = AsyncExitStack() - # The fork_tape context manager must not be exited until the last chunk of the stream is consumed. - tape = await stack.enter_async_context(tape.fork_tape(merge_back=merge_back)) await tape.ensure_bootstrap_anchor() if isinstance(prompt, str) and prompt.strip().startswith(","): result = await self._run_command(tape=tape, line=prompt.strip()) @@ -126,7 +116,25 @@ async def run_stream( allowed_tools=allowed_tools, ) - return self._events_with_callback(events, callback=stack.aclose) + return events + + def session_tape( + self, + session_id: str, + state: TurnState, + *, + source: Tape | None = None, + ) -> Tape: + """Build a plain session tape, optionally from the current agent tape.""" + + state.setdefault("session_id", session_id) + if source is None: + return self.tape.session_tape( + session_id, + workspace_from_state(state), + context=replace(self.tape.context, state=state), + ) + return source.with_context(replace(source.context, state=state)) async def _run_command(self, tape: Tape, *, line: str) -> str: line = line[1:].strip() diff --git a/src/bub/builtin/forkmerge.py b/src/bub/builtin/forkmerge.py new file mode 100644 index 00000000..ad97d42a --- /dev/null +++ b/src/bub/builtin/forkmerge.py @@ -0,0 +1,152 @@ +"""Fork and merge tape writes through a builtin sidecar.""" + +from __future__ import annotations + +import itertools +from collections.abc import Iterable +from dataclasses import dataclass, field, replace +from typing import Any, cast + +from loguru import logger + +from bub.sidecars import sidecar_owns_tape, sidecar_tape_name +from bub.store import AsyncTapeStore, InMemoryTapeStore, TapeQuery +from bub.tape import Tape, TapeEntry + +FORK_MERGE_SIDECAR_NAME = "forkmerge" + + +class _ForkTapeStore: + """In-memory write overlay owned by the forkmerge sidecar.""" + + def __init__(self, parent: AsyncTapeStore, tape: str, *, sidecars: Iterable[str] = ()) -> None: + self._parent = parent + self._store = InMemoryTapeStore() + self._tape = tape + 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 not in self._managed_tapes: + await self._parent.reset(tape) + return + self._store.reset(tape) + 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 query.tape not in self._reset_tapes: + try: + parent_entries = await self._parent.fetch_all(query) + except Exception: + parent_entries = [] + fork_entries: list[TapeEntry] = [] + for entry in self._store.read(query.tape) or []: + if entry.kind == "anchor": # noqa: SIM102 + if query._after_last or (query._after_anchor and entry.payload.get("name") == query._after_anchor): + fork_entries.clear() + parent_entries = [] + continue + if query._kinds and entry.kind not in query._kinds: + continue + fork_entries.append(entry) + entries = itertools.chain(parent_entries, fork_entries) + return itertools.islice(entries, query._limit) if query._limit is not None else entries + + @staticmethod + def _redact_prompt(prompt: list[dict]) -> Any: + if not isinstance(prompt, list): + return prompt + return [part for part in prompt if part.get("type") == "text"] + + @staticmethod + def _redact_payload(payload: dict) -> None: + if "content" in payload: + payload["content"] = _ForkTapeStore._redact_prompt(payload["content"]) + elif "prompt" in payload: + payload["prompt"] = _ForkTapeStore._redact_prompt(payload["prompt"]) + + 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: + 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) or [] + for entry in entries: + await self._parent.append(self._tape, entry) + total += len(entries) + if total: + logger.info('Merged {} entries into tape fork "{}"', total, self._tape) + + +@dataclass +class TapeFork: + """An isolated tape whose writes can be merged or discarded.""" + + tape: Tape + _store: _ForkTapeStore = field(repr=False) + _closed: bool = field(default=False, init=False, repr=False) + + async def merge(self) -> None: + if self._closed: + return + await self._store.merge_back() + self._closed = True + + async def discard(self) -> None: + self._closed = True + + +@dataclass(frozen=True) +class ForkMergeSidecar: + """Isolate tape writes and merge them only when requested.""" + + name: str = field(default=FORK_MERGE_SIDECAR_NAME, init=False) + owns_tape: bool = field(default=False, init=False) + + @classmethod + def mounted(cls, tape: Tape) -> ForkMergeSidecar: + sidecar = tape.get_sidecar(FORK_MERGE_SIDECAR_NAME) + if sidecar is None or not callable(getattr(sidecar, "fork", None)): + raise TypeError("forkmerge sidecar is not mounted") + return cast("ForkMergeSidecar", sidecar) + + async def fork(self, tape: Tape) -> TapeFork: + managed_sidecars = tuple( + sidecar_tape_name(tape.name, sidecar.name) for sidecar in tape.sidecars if sidecar_owns_tape(sidecar) + ) + store = _ForkTapeStore(tape.store, tape.name, sidecars=managed_sidecars) + return TapeFork(tape=replace(tape, store=store), _store=store) diff --git a/src/bub/builtin/hook_impl.py b/src/bub/builtin/hook_impl.py index 75c48843..5c93d4ad 100644 --- a/src/bub/builtin/hook_impl.py +++ b/src/bub/builtin/hook_impl.py @@ -227,12 +227,17 @@ async def build_prompt(self, message: ChannelMessage, session_id: str, state: Tu @hookimpl async def run_model_stream(self, prompt: str | list[dict], session_id: str, state: TurnState) -> AsyncStreamEvents: - return await self._get_agent().run_stream( - session_id=session_id, - prompt=prompt, - state=state, - model=state.get("model"), - ) + from bub.builtin.forkmerge import ForkMergeSidecar + + agent = self._get_agent() + tape = agent.session_tape(session_id, state) + tape_fork = await ForkMergeSidecar.mounted(tape).fork(tape) + try: + events = await agent.run_stream(tape=tape_fork.tape, prompt=prompt, model=state.get("model")) + except Exception: + await tape_fork.discard() + raise + return agent.finalize_stream(events, tape_fork.merge) @hookimpl def continue_prompt(self, prompt: str | list[dict], tape: Tape, state: StreamState) -> str: @@ -386,13 +391,19 @@ def provide_tape_store(self) -> TapeStore: return FileTapeStore(directory=bub.home / "tapes") - @hookimpl - def provide_tape_sidecar(self) -> TapeSidecar: + @hookimpl(specname="provide_tape_sidecar") + def provide_spill_sidecar(self) -> TapeSidecar: from bub.builtin.spill import SpillSettings, SpillStore from bub.configure import ensure_config return SpillStore(ensure_config(SpillSettings)) + @hookimpl(specname="provide_tape_sidecar") + def provide_forkmerge_sidecar(self) -> TapeSidecar: + from bub.builtin.forkmerge import ForkMergeSidecar + + return ForkMergeSidecar() + @hookimpl def build_tape_context(self) -> TapeContext: return default_tape_context() diff --git a/src/bub/builtin/spill.py b/src/bub/builtin/spill.py index e9e6fafc..612ed188 100644 --- a/src/bub/builtin/spill.py +++ b/src/bub/builtin/spill.py @@ -111,6 +111,7 @@ class SpillStore: settings: SpillSettings name: str = field(default=SPILL_SIDECAR_NAME, init=False) + owns_tape: bool = field(default=True, init=False) async def _record_write(self, tape: Tape, data: dict[str, object], *, run_id: str) -> None: try: diff --git a/src/bub/builtin/tools.py b/src/bub/builtin/tools.py index bfe11f8c..d193c598 100644 --- a/src/bub/builtin/tools.py +++ b/src/bub/builtin/tools.py @@ -328,6 +328,8 @@ async def web_fetch(url: str, headers: dict | None = None, timeout: int | None = @tool(name="subagent", context=True, model=SubAgentInput) async def run_subagent(param: SubAgentInput, *, context: ToolContext) -> str: """Run a task with sub-agent using specific model and session.""" + from bub.builtin.forkmerge import ForkMergeSidecar + agent = _get_agent(context) session_id = context.state.get("session_id", "temp/unknown") if param.session == "inherit": @@ -338,19 +340,30 @@ async def run_subagent(param: SubAgentInput, *, context: ToolContext) -> str: subagent_session = param.session state = {**context.state, "session_id": subagent_session} allowed_tools = resolve_tool_names(param.allowed_tools or None, exclude={"subagent"}) + source = context.tape if param.session in {"inherit", "temp"} else None + tape = agent.session_tape(subagent_session, state, source=source) + tape_fork = await ForkMergeSidecar.mounted(tape).fork(tape) output = "" - async for event in await agent.run_stream( - session_id=subagent_session, - prompt=param.prompt, - state=state, - model=param.model, - allowed_tools=allowed_tools, - allowed_skills=param.allowed_skills, - ): - if event.kind == "error": - output += f"[Error: {event.data.get('message', 'unknown error')}]" - elif event.kind == "text": - output += str(event.data.get("delta", "")) + try: + events = await agent.run_stream( + tape=tape_fork.tape, + prompt=param.prompt, + model=param.model, + allowed_tools=allowed_tools, + allowed_skills=param.allowed_skills, + ) + async for event in events: + if event.kind == "error": + output += f"[Error: {event.data.get('message', 'unknown error')}]" + elif event.kind == "text": + output += str(event.data.get("delta", "")) + except Exception: + await tape_fork.discard() + raise + if param.session == "temp": + await tape_fork.discard() + else: + await tape_fork.merge() return output diff --git a/src/bub/sidecars.py b/src/bub/sidecars.py index 2ca767ee..85954467 100644 --- a/src/bub/sidecars.py +++ b/src/bub/sidecars.py @@ -8,12 +8,22 @@ class TapeSidecar(Protocol): - """A named capability backed by a sibling tape.""" + """A named capability mounted beside a session tape.""" @property def name(self) -> str: ... +def sidecar_owns_tape(sidecar: TapeSidecar) -> bool: + """Return whether a sidecar owns a persistent sibling tape. + + Existing storage sidecars default to owning one. Capability-only sidecars, + such as the builtin forkmerge plugin, opt out explicitly. + """ + + return getattr(sidecar, "owns_tape", True) + + def sidecar_tape_name(owner: str, sidecar: str) -> str: """Return the physical tape name for a mounted sidecar.""" diff --git a/src/bub/store.py b/src/bub/store.py index 06184baf..6205f91b 100644 --- a/src/bub/store.py +++ b/src/bub/store.py @@ -2,7 +2,6 @@ import asyncio import inspect -import itertools import json import re import threading @@ -11,9 +10,8 @@ from datetime import UTC, datetime, time from datetime import date as date_type from pathlib import Path -from typing import Any, NoReturn, Protocol, Self, overload +from typing import NoReturn, Protocol, Self, overload -from loguru import logger from typing_extensions import TypeIs from bub.errors import BubError, ErrorKind @@ -280,104 +278,6 @@ def append(self, tape: str, entry: TapeEntry) -> None: self._raise() -class ForkTapeStore: - def __init__(self, parent: AsyncTapeStore, tape: str, *, sidecars: Iterable[str] = ()) -> None: - self._parent = parent - self._store = InMemoryTapeStore() - self._tape = tape - 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 not in self._managed_tapes: - await self._parent.reset(tape) - return - self._store.reset(tape) - 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 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 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) - 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: - if not isinstance(prompt, list): - return prompt - new_prompt = [] - for part in prompt: - if part.get("type") == "text": - new_prompt.append(part) - return new_prompt - - @staticmethod - def _redact_payload(payload: dict) -> None: - if "content" in payload: - payload["content"] = ForkTapeStore._redact_prompt(payload["content"]) - elif "prompt" in payload: - payload["prompt"] = ForkTapeStore._redact_prompt(payload["prompt"]) - - 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: - 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) or [] - for entry in entries: - await self._parent.append(self._tape, entry) - total += len(entries) - if total: - logger.info('Merged {} entries into tape fork "{}"', total, self._tape) - - class FileTapeStore(InMemoryQueryMixin): """TapeStore implementation that persists tapes as JSONL files under a directory.""" diff --git a/src/bub/tape.py b/src/bub/tape.py index 8cd50c47..03804932 100644 --- a/src/bub/tape.py +++ b/src/bub/tape.py @@ -2,11 +2,10 @@ from __future__ import annotations -import contextlib import hashlib import inspect import json -from collections.abc import AsyncGenerator, Callable, Coroutine, Iterable, Mapping +from collections.abc import Callable, Coroutine, Iterable, Mapping from dataclasses import asdict, dataclass, field, replace from datetime import UTC, datetime from pathlib import Path @@ -15,7 +14,7 @@ from pydantic import BaseModel from bub.errors import BubError -from bub.sidecars import TapeSidecar, sidecar_tape_name +from bub.sidecars import TapeSidecar, sidecar_owns_tape, sidecar_tape_name __all__ = [ "LAST_ANCHOR", @@ -229,10 +228,16 @@ def get_sidecar(self, name: str) -> TapeSidecar | 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: + sidecar = self.get_sidecar(name) + if sidecar is None: raise KeyError(f"tape sidecar {name!r} is not mounted") + if not sidecar_owns_tape(sidecar): + raise KeyError(f"tape sidecar {name!r} does not own a sibling tape") return sidecar_tape_name(self.name, name) + def _tape_sidecars(self) -> tuple[TapeSidecar, ...]: + return tuple(sidecar for sidecar in self.sidecars if sidecar_owns_tape(sidecar)) + 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"] @@ -442,6 +447,8 @@ 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") + if not sidecar_owns_tape(sidecar): + raise KeyError(f"tape sidecar {name!r} does not own a sibling tape") return sidecar async def archive_sidecar(self, name: str, *, reason: str = "manual") -> str: @@ -490,7 +497,7 @@ async def reset(self, *, archive: bool = False) -> str: if 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: + for sidecar in self._tape_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) @@ -498,7 +505,7 @@ async def reset(self, *, archive: bool = False) -> str: if archive_path is not None: state["archived"] = str(archive_path) await self.handoff(name="session/start", state=state) - for sidecar in self.sidecars: + for sidecar in self._tape_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( @@ -520,16 +527,3 @@ def session_tape(self, session_id: str, workspace: Path, context: TapeContext | workspace_hash + "__" + hashlib.md5(session_id.encode("utf-8"), usedforsecurity=False).hexdigest()[:16] ) return self.scoped(tape_name, context=context) - - @contextlib.asynccontextmanager - async def fork_tape(self, merge_back: bool = True) -> AsyncGenerator[Tape, None]: - from bub.store import ForkTapeStore - - 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 - finally: - if merge_back: - await fork_store.merge_back() diff --git a/tests/test_builtin_agent.py b/tests/test_builtin_agent.py index f28a88cc..9f0fb5c3 100644 --- a/tests/test_builtin_agent.py +++ b/tests/test_builtin_agent.py @@ -1,7 +1,6 @@ from __future__ import annotations -import contextlib -from collections.abc import AsyncGenerator, AsyncIterator +from collections.abc import AsyncIterator from typing import Any from unittest.mock import MagicMock, patch @@ -18,10 +17,6 @@ from bub.tape import TapeContext from bub.tools import REGISTRY, tool -# --------------------------------------------------------------------------- -# Agent.run() tests: merge_back logic and model passthrough -# --------------------------------------------------------------------------- - class _FakeModelRunner(ModelRunner): def __init__(self, settings: AgentSettings) -> None: @@ -80,27 +75,10 @@ async def _chat_stream(content: str) -> AsyncIterator[ChatCompletionChunk]: yield _chat_chunk(content) -class _ForkCapture: - """Captures fork_tape enter and exit behavior.""" - - def __init__(self) -> None: - self.merge_back_values: list[bool] = [] - self.exit_count = 0 - - @contextlib.asynccontextmanager - async def fork_tape(self, tape_name: str, merge_back: bool = True) -> AsyncGenerator[None, None]: - self.merge_back_values.append(merge_back) - try: - yield - finally: - self.exit_count += 1 - - class _FakeTape: - """Scoped tape stand-in for testing Agent.run().""" + """Scoped tape stand-in for testing Agent.run_stream().""" - def __init__(self, fork_capture: _ForkCapture) -> None: - self._fork = fork_capture + def __init__(self) -> None: self.name = "test-tape" self.context = TapeContext(state={}) self.messages: list[dict[str, Any]] = [] @@ -109,11 +87,6 @@ def __init__(self, fork_capture: _ForkCapture) -> None: async def ensure_bootstrap_anchor(self) -> None: pass - @contextlib.asynccontextmanager - async def fork_tape(self, merge_back: bool = True) -> AsyncGenerator[_FakeTape, None]: - async with self._fork.fork_tape(self.name, merge_back=merge_back): - yield self - async def read_messages(self) -> list[dict[str, Any]]: return list(self.messages) @@ -152,68 +125,21 @@ async def record_chat( self.events.append((self.name, "run", {"run_id": run_id, "model": model, "error": error is not None})) -class _FakeTapeFactory: - """Minimal tape factory stand-in for testing Agent.run().""" - - def __init__(self, fork_capture: _ForkCapture) -> None: - self.tape = _FakeTape(fork_capture) - self.context = self.tape.context - - def session_tape(self, session_id: str, workspace: Any, context: TapeContext | None = None) -> _FakeTape: - if context is not None: - self.tape.context = context - self.context = context - return self.tape - - -@pytest.mark.asyncio -async def test_agent_run_regular_session_merges_back() -> None: - """A regular (non-temp) session should merge tape entries back.""" - agent = _make_agent() - fork_capture = _ForkCapture() - agent.tape = _FakeTapeFactory(fork_capture) # type: ignore[assignment] - - result = await agent.run_stream(session_id="user/session1", prompt="hello", state={"_runtime_workspace": "/tmp"}) # noqa: S108 - - assert fork_capture.merge_back_values == [True] - assert fork_capture.exit_count == 0 - - [event async for event in result] - - assert fork_capture.merge_back_values == [True] - assert fork_capture.exit_count == 1 - - -@pytest.mark.asyncio -async def test_agent_run_temp_session_does_not_merge_back() -> None: - """A temp/ session should NOT merge tape entries back.""" - agent = _make_agent() - fork_capture = _ForkCapture() - agent.tape = _FakeTapeFactory(fork_capture) # type: ignore[assignment] - - result = await agent.run_stream(session_id="temp/abc123", prompt="hello", state={"_runtime_workspace": "/tmp"}) # noqa: S108 - - assert fork_capture.merge_back_values == [False] - assert fork_capture.exit_count == 0 - - [event async for event in result] - - assert fork_capture.merge_back_values == [False] - assert fork_capture.exit_count == 1 +def _fake_tape(tape: _FakeTape | None = None, session_id: str = "user/s1") -> _FakeTape: + session_tape = tape or _FakeTape() + session_tape.context.state.setdefault("session_id", session_id) + return session_tape @pytest.mark.asyncio async def test_agent_run_passes_model_to_llm() -> None: """The model parameter should be forwarded to any-llm.""" agent = _make_agent() - fork_capture = _ForkCapture() - fake_tapes = _FakeTapeFactory(fork_capture) - agent.tape = fake_tapes # type: ignore[assignment] + tape = _fake_tape() result = await agent.run_stream( - session_id="user/s1", + tape=tape, # type: ignore[arg-type] prompt="hello", - state={"_runtime_workspace": "/tmp"}, # noqa: S108 model="openai:gpt-4o", ) [event async for event in result] @@ -226,9 +152,8 @@ async def test_agent_run_passes_model_to_llm() -> None: @pytest.mark.asyncio async def test_agent_run_empty_prompt_returns_error() -> None: agent = _make_agent() - agent.tape = MagicMock() - result = await agent.run_stream(session_id="user/s1", prompt="", state={}) + result = await agent.run_stream(tape=_fake_tape(), prompt="") # type: ignore[arg-type] events = [event async for event in result] assert [(event.kind, event.data) for event in events] == [ @@ -241,11 +166,8 @@ async def test_agent_run_empty_prompt_returns_error() -> None: async def test_agent_run_model_defaults_to_none() -> None: """When model is not specified, settings.model is used for any-llm.""" agent = _make_agent() - fork_capture = _ForkCapture() - fake_tapes = _FakeTapeFactory(fork_capture) - agent.tape = fake_tapes # type: ignore[assignment] - result = await agent.run_stream(session_id="user/s1", prompt="hello", state={"_runtime_workspace": "/tmp"}) # noqa: S108 + result = await agent.run_stream(tape=_fake_tape(), prompt="hello") # type: ignore[arg-type] [event async for event in result] completion_kwargs = _model_runner(agent).completion_kwargs @@ -256,7 +178,7 @@ async def test_agent_run_model_defaults_to_none() -> None: @pytest.mark.asyncio async def test_agent_loop_awaits_continue_prompt_hook_with_stream_state() -> None: agent = _make_agent() - tape = _FakeTape(_ForkCapture()) + tape = _FakeTape() prompts: list[str | list[dict]] = [] continuation_prompts: list[str | list[dict]] = [] observed_usage: list[dict[str, Any] | None] = [] @@ -304,14 +226,11 @@ async def test_agent_run_model_override_does_not_mutate_default() -> None: sessions the way a process-global env var would. """ agent = _make_agent() - fork_capture = _ForkCapture() - agent.tape = _FakeTapeFactory(fork_capture) # type: ignore[assignment] default_model = agent.settings.model result = await agent.run_stream( - session_id="user/s1", + tape=_fake_tape(), # type: ignore[arg-type] prompt="hello", - state={"_runtime_workspace": "/tmp"}, # noqa: S108 model="openai:gpt-4o", ) [event async for event in result] @@ -325,9 +244,8 @@ async def test_agent_run_model_override_does_not_mutate_default() -> None: @pytest.mark.asyncio async def test_agent_run_injects_steering_messages_once_by_session() -> None: agent = _make_agent() - fork_capture = _ForkCapture() - fake_tapes = _FakeTapeFactory(fork_capture) - agent.tape = fake_tapes # type: ignore[assignment] + tape = _FakeTape() + tape = _fake_tape(tape) steering_inbox = InMemorySteeringInbox() agent.framework.get_steering_inbox.return_value = steering_inbox @@ -339,7 +257,7 @@ async def test_agent_run_injects_steering_messages_once_by_session() -> None: {"session_id": "user/s1", "content": "second steer"}, {"session_id": "user/s1"} ) - result = await agent.run_stream(session_id="user/s1", prompt="hello", state={"_runtime_workspace": "/tmp"}) # noqa: S108 + result = await agent.run_stream(tape=tape, prompt="hello") # type: ignore[arg-type] [event async for event in result] completion_kwargs = _model_runner(agent).completion_kwargs @@ -350,14 +268,14 @@ async def test_agent_run_injects_steering_messages_once_by_session() -> None: {"role": "user", "content": "second steer"}, {"role": "user", "content": "hello"}, ] - assert fake_tapes.tape.messages == [ + assert tape.messages == [ {"role": "user", "content": "first steer"}, {"role": "user", "content": "second steer"}, {"role": "user", "content": "hello"}, {"role": "assistant", "content": "done"}, ] - result = await agent.run_stream(session_id="user/s1", prompt="again", state={"_runtime_workspace": "/tmp"}) # noqa: S108 + result = await agent.run_stream(tape=tape, prompt="again") # type: ignore[arg-type] [event async for event in result] completion_kwargs = _model_runner(agent).completion_kwargs @@ -383,14 +301,10 @@ def denied_agent_tool() -> str: return "denied" agent = _make_agent() - fork_capture = _ForkCapture() - fake_tapes = _FakeTapeFactory(fork_capture) - agent.tape = fake_tapes # type: ignore[assignment] result = await agent.run_stream( - session_id="user/s1", + tape=_fake_tape(), # type: ignore[arg-type] prompt="hello", - state={"_runtime_workspace": "/tmp"}, # noqa: S108 allowed_tools=[" tests_allowed_agent_tool "], ) [event async for event in result] @@ -419,13 +333,10 @@ def internal_agent_tool() -> str: return "internal" agent = _make_agent() - fork_capture = _ForkCapture() - agent.tape = _FakeTapeFactory(fork_capture) # type: ignore[assignment] result = await agent.run_stream( - session_id="user/s1", + tape=_fake_tape(), # type: ignore[arg-type] prompt="hello", - state={"_runtime_workspace": "/tmp"}, # noqa: S108 allowed_tools=[visible_name, internal_name], ) [event async for event in result] @@ -441,14 +352,10 @@ def internal_agent_tool() -> str: @pytest.mark.asyncio async def test_agent_run_rejects_unknown_allowed_tools() -> None: agent = _make_agent() - fork_capture = _ForkCapture() - fake_tapes = _FakeTapeFactory(fork_capture) - agent.tape = fake_tapes # type: ignore[assignment] stream = await agent.run_stream( - session_id="user/s1", + tape=_fake_tape(), # type: ignore[arg-type] prompt="hello", - state={"_runtime_workspace": "/tmp"}, # noqa: S108 allowed_tools=["tests_missing_agent_tool"], ) @@ -466,12 +373,15 @@ async def test_run_command_model_switches_session_model_directly() -> None: session tape (merged back at end of turn), not a side effect of save_state. """ agent = _make_agent() - fork_capture = _ForkCapture() - agent.tape = _FakeTapeFactory(fork_capture) # type: ignore[assignment] state: dict[str, Any] = {"_runtime_workspace": "/tmp"} # noqa: S108 + tape = _FakeTape() + tape.context = TapeContext(state=state) assert "model" not in REGISTRY or REGISTRY["model"].context is True - stream = await agent.run_stream(session_id="user/s1", prompt=",model openai:gpt-4o", state=state) + stream = await agent.run_stream( + tape=_fake_tape(tape), # type: ignore[arg-type] + prompt=",model openai:gpt-4o", + ) events = [event async for event in stream] # state["model"] is mutated in place (tool context shares the framework state). diff --git a/tests/test_builtin_hook_impl.py b/tests/test_builtin_hook_impl.py index 12c639ad..4c272ffa 100644 --- a/tests/test_builtin_hook_impl.py +++ b/tests/test_builtin_hook_impl.py @@ -7,6 +7,7 @@ import pytest +from bub.builtin.forkmerge import ForkMergeSidecar from bub.builtin.hook_impl import AGENTS_FILE_NAME, DEFAULT_CONTINUE_PROMPT, DEFAULT_SYSTEM_PROMPT, BuiltinImpl from bub.channels.message import ChannelMessage from bub.framework import BubFramework @@ -32,6 +33,7 @@ def _fake_tape(home: Path) -> Tape: archive_path=home / "tapes", store=AsyncTapeStoreAdapter(InMemoryTapeStore()), context=TapeContext(), + sidecars=(ForkMergeSidecar(),), ) @@ -48,15 +50,32 @@ async def run(self, *, session_id: str, prompt: str, state: dict[str, object]) - self.run_calls.append((session_id, prompt, state)) return "agent-output" + def session_tape(self, session_id: str, state: dict[str, object], *, source=None) -> Tape: + state.setdefault("session_id", session_id) + context = TapeContext(state=state) + if source is not None: + return source.with_context(context) + return self.tape.session_tape(session_id, Path(state.get("_runtime_workspace", ".")), context=context) + + def finalize_stream(self, events: AsyncStreamEvents, callback) -> AsyncStreamEvents: + async def iterator(): + try: + async for event in events: + yield event + finally: + await callback() + + return AsyncStreamEvents(iterator()) + async def run_stream( self, *, - session_id: str, + tape: Tape, prompt: str, - state: dict[str, object], model: str | None = None, ) -> AsyncStreamEvents: - self.run_stream_calls.append((session_id, prompt, state, model)) + state = tape.context.state + self.run_stream_calls.append((str(state["session_id"]), prompt, state, model)) async def iterator(): yield StreamEvent("text", {"delta": "agent-output"}) diff --git a/tests/test_file_tape_store_entry_ids.py b/tests/test_file_tape_store_entry_ids.py index 0f73b91e..7637110f 100644 --- a/tests/test_file_tape_store_entry_ids.py +++ b/tests/test_file_tape_store_entry_ids.py @@ -2,22 +2,30 @@ import pytest +from bub.builtin.forkmerge import ForkMergeSidecar from bub.sidecars import sidecar_tape_name -from bub.store import AsyncTapeStoreAdapter, FileTapeStore, ForkTapeStore -from bub.tape import TapeEntry +from bub.store import AsyncTapeStoreAdapter, FileTapeStore +from bub.tape import Tape, TapeContext, TapeEntry @pytest.mark.asyncio async def test_file_tape_store_assigns_monotonic_ids_when_merging_forked_entries(tmp_path) -> None: parent = FileTapeStore(directory=tmp_path) - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "tape") - - await store.append("tape", TapeEntry.event(name="first", data={"n": 1})) - await store.merge_back() - - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "tape") - await store.append("tape", TapeEntry.event(name="second", data={"n": 2})) - await store.merge_back() + forkmerge = ForkMergeSidecar() + tape = Tape( + tmp_path, + AsyncTapeStoreAdapter(parent), + TapeContext(), + sidecars=(forkmerge,), + ).scoped("tape") + + first_fork = await forkmerge.fork(tape) + await first_fork.tape.append_event("first", {"n": 1}) + await first_fork.merge() + + second_fork = await forkmerge.fork(tape) + await second_fork.tape.append_event("second", {"n": 2}) + await second_fork.merge() entries = parent.read("tape") or [] assert [entry.id for entry in entries] == [1, 2] diff --git a/tests/test_fork_store_merge_back.py b/tests/test_fork_store_merge_back.py deleted file mode 100644 index e614900e..00000000 --- a/tests/test_fork_store_merge_back.py +++ /dev/null @@ -1,101 +0,0 @@ -from __future__ import annotations - -import pytest - -from bub.store import AsyncTapeStoreAdapter, ForkTapeStore, InMemoryTapeStore, TapeQuery -from bub.tape import TapeEntry - - -@pytest.mark.asyncio -async def test_fork_merge_back_true_merges_entries() -> None: - """With merge_back=True (default), forked entries are merged into the parent.""" - parent = InMemoryTapeStore() - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "test-tape") - - await store.append("test-tape", TapeEntry.event(name="step", data={"x": 1})) - await store.append("test-tape", TapeEntry.event(name="step", data={"x": 2})) - await store.merge_back() - - entries = parent.read("test-tape") - assert entries is not None - assert len(entries) == 2 - - -@pytest.mark.asyncio -async def test_fork_merge_back_false_discards_entries() -> None: - """With merge_back=False, forked entries are NOT merged into the parent.""" - parent = InMemoryTapeStore() - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "test-tape") - - await store.append("test-tape", TapeEntry.event(name="step", data={"x": 1})) - - entries = parent.read("test-tape") - # No entries should have been merged - assert entries is None or len(entries) == 0 - - -@pytest.mark.asyncio -async def test_merge_back_can_be_called_without_entries() -> None: - parent = InMemoryTapeStore() - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "test-tape") - - await store.merge_back() - - entries = parent.read("test-tape") - assert entries is None or len(entries) == 0 - - -@pytest.mark.asyncio -async def test_fork_reset_with_merge_back_false_preserves_parent_entries() -> None: - parent = InMemoryTapeStore() - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "test-tape") - parent.append("test-tape", TapeEntry.event(name="before", data={"x": 1})) - - await store.reset("test-tape") - await store.append("test-tape", TapeEntry.event(name="inside", data={"x": 2})) - - entries = parent.read("test-tape") - assert entries is not None - assert [entry.payload["name"] for entry in entries] == ["before"] - - -@pytest.mark.asyncio -async def test_fork_reset_with_merge_back_true_replaces_parent_entries() -> None: - parent = InMemoryTapeStore() - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "test-tape") - parent.append("test-tape", TapeEntry.event(name="before", data={"x": 1})) - - await store.reset("test-tape") - await store.append("test-tape", TapeEntry.event(name="inside", data={"x": 2})) - await store.merge_back() - - entries = parent.read("test-tape") - assert entries is not None - assert [entry.payload["name"] for entry in entries] == ["inside"] - - -@pytest.mark.asyncio -async def test_fork_reset_hides_parent_entries_during_fetch() -> None: - parent = InMemoryTapeStore() - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "test-tape") - parent.append("test-tape", TapeEntry.event(name="before", data={"x": 1})) - - await store.reset("test-tape") - await store.append("test-tape", TapeEntry.event(name="inside", data={"x": 2})) - - query = TapeQuery(tape="test-tape", store=store) - entries = list(await store.fetch_all(query)) - - assert [entry.payload["name"] for entry in entries] == ["inside"] - - -@pytest.mark.asyncio -async def test_reset_for_unbound_tape_resets_parent_immediately() -> None: - parent = InMemoryTapeStore() - store = ForkTapeStore(AsyncTapeStoreAdapter(parent), "other-tape") - parent.append("test-tape", TapeEntry.event(name="before", data={"x": 1})) - - await store.reset("test-tape") - - entries = parent.read("test-tape") - assert entries is None diff --git a/tests/test_spill.py b/tests/test_spill.py index b6ed58af..70b76f0d 100644 --- a/tests/test_spill.py +++ b/tests/test_spill.py @@ -7,6 +7,7 @@ import pytest from bub.builtin.context import default_tape_context +from bub.builtin.forkmerge import ForkMergeSidecar from bub.builtin.hook_impl import BuiltinImpl from bub.builtin.spill import ( SPILL_READ_MODEL_NAME, @@ -50,7 +51,12 @@ def _page_field(page: str, name: str) -> str: def _root_tape(tmp_path: Path, store: TapeStore, *, threshold: int = 1) -> Tape: spill = SpillStore(SpillSettings(threshold=threshold)) - return Tape(tmp_path, AsyncTapeStoreAdapter(store), default_tape_context(), sidecars=(spill,)).scoped("session") + return Tape( + tmp_path, + AsyncTapeStoreAdapter(store), + default_tape_context(), + sidecars=(spill, ForkMergeSidecar()), + ).scoped("session") async def _read_page( @@ -76,43 +82,45 @@ async def test_oversized_result_is_bounded_and_readable_across_merge(tmp_path: P root = _root_tape(tmp_path, parent) output = ("alpha🙂beta\n" * 5000) + "the-end" - 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 _read_page(context, handle, cursor=cursor, count=2) - 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 _read_page(context, handle, from_end=True) - 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 + tape_fork = await ForkMergeSidecar.mounted(root).fork(root) + tape = tape_fork.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 _read_page(context, handle, cursor=cursor, count=2) + 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 _read_page(context, handle, from_end=True) + 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 + await tape_fork.merge() persisted_context = ToolContext(tape=root, run_id="run-2") persisted = await _read_page(persisted_context, handle) @@ -124,20 +132,19 @@ async def test_spill_configuration_preserves_results_that_should_not_be_spilled( parent = InMemoryTapeStore() root = _root_tape(tmp_path, parent, threshold=100) - 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 - ) - - assert small.tool_results == ["tiny"] + tape_fork = await ForkMergeSidecar.mounted(root).fork(root) + context = ToolContext(tape=tape_fork.tape, run_id="run-1") + small = await _spill_executor().execute_async([(Tool(name="small", handler=lambda: "tiny"), {})], context=context) + assert small.tool_results == ["tiny"] + await tape_fork.merge() 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"), - ) + disabled_fork = await ForkMergeSidecar.mounted(disabled).fork(disabled) + execution = await _spill_executor().execute_async( + [(Tool(name="large", handler=lambda: "x" * 20_000), {})], + context=ToolContext(tape=disabled_fork.tape, run_id="run-2"), + ) + await disabled_fork.merge() assert execution.tool_results == ["x" * 20_000] @@ -146,13 +153,14 @@ async def test_temporary_fork_discards_spilled_content(tmp_path: Path) -> None: parent = InMemoryTapeStore() root = _root_tape(tmp_path, parent) - 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 _read_page(context, handle) + tape_fork = await ForkMergeSidecar.mounted(root).fork(root) + context = ToolContext(tape=tape_fork.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 _read_page(context, handle) + await tape_fork.discard() missing = await _read_page(ToolContext(tape=root), handle) assert "no spilled tool result" in missing @@ -181,22 +189,24 @@ async def test_tape_archive_preserves_spilled_results_and_clears_the_session(tmp sidecar = root.sidecar_tape_name(SPILL_SIDECAR_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, - ) + tape_fork = await ForkMergeSidecar.mounted(root).fork(root) + tape = tape_fork.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, + ) + await tape_fork.merge() result = await root.reset(archive=True) diff --git a/tests/test_subagent_tool.py b/tests/test_subagent_tool.py index c8d907c9..01a4045b 100644 --- a/tests/test_subagent_tool.py +++ b/tests/test_subagent_tool.py @@ -1,7 +1,8 @@ from __future__ import annotations +from types import SimpleNamespace from typing import Any -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock import pytest @@ -15,13 +16,28 @@ class FakeContext: def __init__(self, state: dict[str, Any]) -> None: self.state = state - self.tape = None + self.tape = state["_runtime_agent"].tape + + +class FakeTape: + def __init__(self, forkmerge: object) -> None: + self.forkmerge = forkmerge + + def get_sidecar(self, name: str) -> object | None: + return self.forkmerge if name == "forkmerge" else None class FakeAgent: def __init__(self) -> None: + self.tape_fork = SimpleNamespace(tape=object(), merge=AsyncMock(), discard=AsyncMock()) + self.forkmerge = SimpleNamespace(fork=AsyncMock(return_value=self.tape_fork)) + self.tape = FakeTape(self.forkmerge) + self.session_tape = MagicMock(side_effect=self._session_tape) self.run_stream = AsyncMock(side_effect=self._run_stream) + def _session_tape(self, session_id: str, state: dict[str, Any], *, source=None) -> FakeTape: + return source or self.tape + async def _run_stream(self, **kwargs: Any) -> AsyncStreamEvents: async def iterator(): yield StreamEvent("text", {"delta": "agent result"}) @@ -38,10 +54,12 @@ async def test_subagent_inherit_session() -> None: assert result == "agent result" agent.run_stream.assert_called_once() + tape_args = agent.session_tape.call_args.args + assert tape_args[0] == "user/abc" call_kwargs = agent.run_stream.call_args.kwargs - assert call_kwargs["session_id"] == "user/abc" assert call_kwargs["prompt"] == "do something" assert call_kwargs["model"] is None + agent.tape_fork.merge.assert_awaited_once() @pytest.mark.asyncio @@ -51,9 +69,10 @@ async def test_subagent_temp_session() -> None: await run_subagent.run(prompt="task", session="temp", context=ctx) - call_kwargs = agent.run_stream.call_args.kwargs - assert call_kwargs["session_id"].startswith("temp/") - assert call_kwargs["session_id"] != "user/abc" + subagent_session = agent.session_tape.call_args.args[0] + assert subagent_session.startswith("temp/") + assert subagent_session != "user/abc" + agent.tape_fork.discard.assert_awaited_once() @pytest.mark.asyncio @@ -63,8 +82,7 @@ async def test_subagent_custom_session() -> None: await run_subagent.run(prompt="task", session="custom/session-1", context=ctx) - call_kwargs = agent.run_stream.call_args.kwargs - assert call_kwargs["session_id"] == "custom/session-1" + assert agent.session_tape.call_args.args[0] == "custom/session-1" @pytest.mark.asyncio @@ -85,10 +103,9 @@ async def test_subagent_state_includes_session_id() -> None: await run_subagent.run(prompt="task", session="temp", context=ctx) - call_kwargs = agent.run_stream.call_args.kwargs - state = call_kwargs["state"] + session_id, state = agent.session_tape.call_args.args # Turn state should contain the subagent session_id, not the original - assert state["session_id"] == call_kwargs["session_id"] + assert state["session_id"] == session_id assert state["extra"] == "val" @@ -100,8 +117,7 @@ async def test_subagent_default_session_when_missing() -> None: await run_subagent.run(prompt="task", session="inherit", context=ctx) - call_kwargs = agent.run_stream.call_args.kwargs - assert call_kwargs["session_id"] == "temp/unknown" + assert agent.session_tape.call_args.args[0] == "temp/unknown" @pytest.mark.asyncio diff --git a/tests/test_tape.py b/tests/test_tape.py index fc7864a0..b539d8ea 100644 --- a/tests/test_tape.py +++ b/tests/test_tape.py @@ -4,8 +4,9 @@ import pytest -from bub.store import AsyncTapeStoreAdapter, ForkTapeStore, InMemoryTapeStore -from bub.tape import Tape, TapeContext +from bub.builtin.forkmerge import ForkMergeSidecar +from bub.store import AsyncTapeStoreAdapter, InMemoryTapeStore +from bub.tape import Tape, TapeContext, TapeEntry def test_tape_reexports_legacy_store_objects() -> None: @@ -28,28 +29,71 @@ def test_tape_reexports_legacy_store_objects() -> None: @pytest.mark.asyncio -async def test_tape_fork_binds_temporary_fork_store_to_scoped_tape(tmp_path: Path) -> None: +async def test_tape_fork_is_isolated_until_explicit_merge(tmp_path: Path) -> None: parent = InMemoryTapeStore() - root = Tape(tmp_path, AsyncTapeStoreAdapter(parent), TapeContext()).scoped("test-tape") + forkmerge = ForkMergeSidecar() + root = Tape(tmp_path, AsyncTapeStoreAdapter(parent), TapeContext(), sidecars=(forkmerge,)).scoped("test-tape") - async with root.fork_tape(merge_back=True) as forked: - first_store = forked.store + tape_fork = await forkmerge.fork(root) - assert isinstance(first_store, ForkTapeStore) - assert first_store is not root.store + await tape_fork.tape.append_event("step", {"value": 1}) + assert parent.read("test-tape") is None - await forked.append_event("step", {"value": 1}) - assert parent.read("test-tape") is None + await tape_fork.merge() assert [entry.payload["name"] for entry in parent.read("test-tape") or []] == ["step"] - async with root.fork_tape(merge_back=False) as forked: - second_store = forked.store - await forked.append_event("step", {"value": 2}) - assert isinstance(second_store, ForkTapeStore) - assert second_store is not first_store - assert [entry.payload["data"]["value"] for entry in parent.read("test-tape") or []] == [1] +@pytest.mark.asyncio +async def test_discarded_session_does_not_change_parent(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + forkmerge = ForkMergeSidecar() + root = Tape(tmp_path, AsyncTapeStoreAdapter(parent), TapeContext(), sidecars=(forkmerge,)).scoped("test-tape") + tape_fork = await forkmerge.fork(root) + + await tape_fork.tape.append_event("step", {"value": 2}) + await tape_fork.discard() + + assert parent.read("test-tape") is None + + +@pytest.mark.asyncio +async def test_discarding_a_reset_fork_preserves_parent_entries(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + parent.append("test-tape", TapeEntry.event(name="before")) + forkmerge = ForkMergeSidecar() + root = Tape(tmp_path, AsyncTapeStoreAdapter(parent), TapeContext(), sidecars=(forkmerge,)).scoped("test-tape") + tape_fork = await forkmerge.fork(root) + + await tape_fork.tape.reset() + await tape_fork.tape.append_event("inside", {}) + await tape_fork.discard() + + assert [entry.payload["name"] for entry in parent.read("test-tape") or []] == ["before"] + + +@pytest.mark.asyncio +async def test_merging_a_reset_fork_replaces_parent_entries(tmp_path: Path) -> None: + parent = InMemoryTapeStore() + parent.append("test-tape", TapeEntry.event(name="before")) + forkmerge = ForkMergeSidecar() + root = Tape(tmp_path, AsyncTapeStoreAdapter(parent), TapeContext(), sidecars=(forkmerge,)).scoped("test-tape") + tape_fork = await forkmerge.fork(root) + + await tape_fork.tape.reset() + await tape_fork.tape.append_event("inside", {}) + assert [entry.payload["name"] for entry in await tape_fork.tape.store.fetch_all(tape_fork.tape.query())] == [ + "session/start", + "handoff", + "inside", + ] + await tape_fork.merge() + + assert [entry.payload["name"] for entry in parent.read("test-tape") or []] == [ + "session/start", + "handoff", + "inside", + ] @pytest.mark.asyncio From 88ad6d6480b9513cb2360da12242bd535e20e113 Mon Sep 17 00:00:00 2001 From: PsiACE Date: Mon, 24 Aug 2026 02:49:43 +0800 Subject: [PATCH 2/4] docs: document forkmerge sidecar --- website/src/content/docs/docs/build/hooks.mdx | 8 ++++++-- website/src/content/docs/docs/concepts/turn-pipeline.mdx | 2 +- website/src/content/docs/docs/reference/hooks.mdx | 2 +- website/src/content/docs/docs/reference/types.mdx | 2 +- website/src/content/docs/zh-cn/docs/build/hooks.mdx | 8 ++++++-- .../content/docs/zh-cn/docs/concepts/turn-pipeline.mdx | 2 +- website/src/content/docs/zh-cn/docs/reference/hooks.mdx | 2 +- website/src/content/docs/zh-cn/docs/reference/types.mdx | 2 +- 8 files changed, 18 insertions(+), 10 deletions(-) diff --git a/website/src/content/docs/docs/build/hooks.mdx b/website/src/content/docs/docs/build/hooks.mdx index d986a300..a2e797be 100644 --- a/website/src/content/docs/docs/build/hooks.mdx +++ b/website/src/content/docs/docs/build/hooks.mdx @@ -137,7 +137,7 @@ The full plugin lives at [`bub-tapestore-sqlite`](https://github.com/bubbuild/bu ### Mount a tape sidecar -Use `provide_tape_sidecar` 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. +Use `provide_tape_sidecar` to mount a named capability beside each session tape. A storage sidecar owns a sibling tape by default; its plugin owns configuration and the data format. ```python from bub import hookimpl @@ -152,7 +152,11 @@ def provide_tape_sidecar(): return ArtifactSidecar() ``` -For a session tape named `session`, this mounts `session__sidecar__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")`. Each hook implementation contributes one sidecar. 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. +For a session tape named `session`, this mounts `session__sidecar__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 includes storage sidecars in merge, archive, and reset operations. + +Set `owns_tape = False` for a capability-only sidecar. It receives no physical sibling tape and owns its lifecycle. The builtin `forkmerge` sidecar follows this shape: it isolates writes over a plain session tape and returns a tape fork whose caller explicitly invokes `merge()` or `discard()`. + +Each hook implementation contributes one 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`. diff --git a/website/src/content/docs/docs/concepts/turn-pipeline.mdx b/website/src/content/docs/docs/concepts/turn-pipeline.mdx index 80f10f82..2067dd43 100644 --- a/website/src/content/docs/docs/concepts/turn-pipeline.mdx +++ b/website/src/content/docs/docs/concepts/turn-pipeline.mdx @@ -92,7 +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_sidecar` — the configured builtin spill sidecar. +- `provide_tape_sidecar` — the builtin `spill` and `forkmerge` sidecars over the configured plain tape store. - `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 baa2d036..21ade274 100644 --- a/website/src/content/docs/docs/reference/hooks.mdx +++ b/website/src/content/docs/docs/reference/hooks.mdx @@ -97,7 +97,7 @@ Each impl receives only the kwargs it declares. You can omit unused parameters f ### Tape sidecars -Each `provide_tape_sidecar` implementation contributes one named capability backed by a sibling tape 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 `__sidecar__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. +Each `provide_tape_sidecar` implementation contributes one named capability. Storage sidecars own a sibling tape in the active `TapeStore`; for example, a scoped session maps `spill` to `__sidecar__spill`, and the main tape includes it in merge, archive, and reset operations. A capability-only sidecar sets `owns_tape = False` and manages its own lifecycle. The builtin `forkmerge` sidecar uses that form to isolate writes over a plain session tape; its caller chooses `merge()` or `discard()`. Removing a provider stops mounting its sidecar but does not delete stored data. ### `on_error` observer safety diff --git a/website/src/content/docs/docs/reference/types.mdx b/website/src/content/docs/docs/reference/types.mdx index eb3b0e8c..e88b12a6 100644 --- a/website/src/content/docs/docs/reference/types.mdx +++ b/website/src/content/docs/docs/reference/types.mdx @@ -94,7 +94,7 @@ 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. `TapeSidecar` is the minimal named contract returned by `provide_tape_sidecar`; `sidecar_tape_name` resolves its sibling tape name. 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_sidecar`; `sidecar_tape_name` resolves the physical name for a sidecar that owns a sibling tape. See the [Hook reference](/docs/reference/hooks/) for storage and capability-only sidecar 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 55c933cf..afa4bafd 100644 --- a/website/src/content/docs/zh-cn/docs/build/hooks.mdx +++ b/website/src/content/docs/zh-cn/docs/build/hooks.mdx @@ -137,7 +137,7 @@ def provide_tape_store(): ### 挂载 tape sidecar -当插件需要一个与 session tape 共享生命周期的 sibling tape 时，使用 `provide_tape_sidecar`。sidecar 只声明稳定的 `name`；配置与数据格式由插件自己管理。 +使用 `provide_tape_sidecar` 在每条 session tape 旁挂载具名能力。存储型 sidecar 默认拥有一条 sibling tape；配置与数据格式由插件自己管理。 ```python from bub import hookimpl @@ -152,7 +152,11 @@ def provide_tape_sidecar(): return ArtifactSidecar() ``` -对名为 `session` 的 tape，这会在 active `TapeStore` 中挂载 `session__sidecar__artifacts`。工具代码可通过 `context.tape.get_sidecar("artifacts")` 取得 provider，并用 `context.tape.sidecar_tape_name("artifacts")` 得到物理名称。每个 hook implementation 提供一个 sidecar。主 tape 负责所有已挂载 sidecar 的 fork、merge、archive 和 reset。同名 provider 遵循普通 hook 优先级：最先出现的实现生效。 +对名为 `session` 的 tape，这会在 active `TapeStore` 中挂载 `session__sidecar__artifacts`。工具代码可通过 `context.tape.get_sidecar("artifacts")` 取得 provider，并用 `context.tape.sidecar_tape_name("artifacts")` 得到物理名称。主 tape 会在 merge、archive 和 reset 时一并处理存储型 sidecar。 + +能力型 sidecar 应设置 `owns_tape = False`。它没有物理 sibling tape，并自行管理生命周期。builtin `forkmerge` sidecar 采用这种形态：它在朴素 session tape 之上隔离写入，并返回由调用方显式 `merge()` 或 `discard()` 的 tape fork。 + +每个 hook implementation 提供一个 sidecar。同名 provider 遵循普通 hook 优先级：最先出现的实现生效。 builtin spill 插件拥有 `SpillStore`、`spill.read` 工具、数据格式和 `SpillSettings`。核心层不导入或识别 spill 实现。配置使用 `spill:` section 和 `BUB_SPILL_*` 环境变量，与 `AgentSettings` 相互独立。 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 12ed1f50..1609217a 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,7 +92,7 @@ hook 运行时会在两者间互相适配：若插件只实现 `run_model_stream - `dispatch_outbound` — 转发到绑定的 `ChannelRouter`。 - `system_prompt` — 将默认 prompt 与 workspace 的 `AGENTS.md` 拼接。 - `provide_tape_store` — 位于 `~/.bub/tapes` 下的文件型 tape store。 -- `provide_tape_sidecar` — 使用当前配置的 builtin spill sidecar。 +- `provide_tape_sidecar` — 挂载在当前朴素 tape store 之上的 builtin `spill` 与 `forkmerge` 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 d7a9a708..f20328ff 100644 --- a/website/src/content/docs/zh-cn/docs/reference/hooks.mdx +++ b/website/src/content/docs/zh-cn/docs/reference/hooks.mdx @@ -97,7 +97,7 @@ def _kwargs_for_impl(impl: Any, kwargs: dict[str, Any]) -> dict[str, Any]: ### Tape sidecar -每个 `provide_tape_sidecar` 实现提供一个由 active `TapeStore` 中 sibling tape 支撑的具名能力。Bub 在创建 agent root `Tape` 时挂载聚合结果；例如 scoped session 会把 sidecar `spill` 映射为 `__sidecar__spill`。fork、merge、archive 和 reset 会处理所有已挂载 sidecar，不要求存储插件实现新接口。移除 provider 只会停止挂载，不会删除已存数据。 +每个 `provide_tape_sidecar` 实现提供一个具名能力。存储型 sidecar 在 active `TapeStore` 中拥有一条 sibling tape；例如 scoped session 会把 `spill` 映射为 `__sidecar__spill`，主 tape 会在 merge、archive 和 reset 时一并处理它。能力型 sidecar 设置 `owns_tape = False`，自行管理生命周期。builtin `forkmerge` sidecar 采用这种形态，在朴素 session tape 之上隔离写入，由调用方选择 `merge()` 或 `discard()`。移除 provider 只会停止挂载，不会删除已存数据。 ### `on_error` 观察者安全性 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 b55aee87..80b56307 100644 --- a/website/src/content/docs/zh-cn/docs/reference/types.mdx +++ b/website/src/content/docs/zh-cn/docs/reference/types.mdx @@ -94,7 +94,7 @@ from bub.hooks.runtime import HookRuntime from bub.sidecars import TapeSidecar, sidecar_tape_name ``` -插件作者通常只需要 `hookimpl` 与对应 hook 使用的 payload 类型。`TapeSidecar` 是 `provide_tape_sidecar` 返回的最小具名契约；`sidecar_tape_name` 用于解析 sibling tape 名称。分发和故障隔离语义见 [Hook 参考](/zh-cn/docs/reference/hooks/)。 +插件作者通常只需要 `hookimpl` 与对应 hook 使用的 payload 类型。`TapeSidecar` 是 `provide_tape_sidecar` 返回的最小具名契约；`sidecar_tape_name` 用于解析拥有 sibling tape 的 sidecar 的物理名称。存储型与能力型 sidecar 的语义见 [Hook 参考](/zh-cn/docs/reference/hooks/)。 ## Channel 契约 From 50a19b72e4010c7c021537dd41121bcc5dee1038 Mon Sep 17 00:00:00 2001 From: PsiACE Date: Mon, 24 Aug 2026 12:14:33 +0800 Subject: [PATCH 3/4] refactor: attach callbacks to event streams --- src/bub/builtin/agent.py | 15 +-------------- src/bub/builtin/hook_impl.py | 2 +- src/bub/streaming.py | 14 +++++++++++++- tests/test_builtin_hook_impl.py | 10 ---------- website/src/content/docs/docs/reference/types.mdx | 2 +- .../content/docs/zh-cn/docs/reference/types.mdx | 2 +- 6 files changed, 17 insertions(+), 28 deletions(-) diff --git a/src/bub/builtin/agent.py b/src/bub/builtin/agent.py index c7d90803..48c242aa 100644 --- a/src/bub/builtin/agent.py +++ b/src/bub/builtin/agent.py @@ -7,7 +7,7 @@ import re import shlex import time -from collections.abc import AsyncGenerator, AsyncIterator, Callable, Collection, Coroutine, Iterable +from collections.abc import AsyncGenerator, AsyncIterator, Collection, Iterable from dataclasses import dataclass, replace from datetime import UTC, datetime from functools import cached_property @@ -72,19 +72,6 @@ async def generator() -> AsyncIterator: return AsyncStreamEvents(generator()) - @staticmethod - def finalize_stream( - events: AsyncStreamEvents, callback: Callable[[], Coroutine[Any, Any, Any]] - ) -> AsyncStreamEvents: - async def generator() -> AsyncIterator[StreamEvent]: - try: - async for event in events: - yield event - finally: - await callback() - - return AsyncStreamEvents(generator(), state=events._state) - async def run_stream( self, *, diff --git a/src/bub/builtin/hook_impl.py b/src/bub/builtin/hook_impl.py index 5c93d4ad..7a69236e 100644 --- a/src/bub/builtin/hook_impl.py +++ b/src/bub/builtin/hook_impl.py @@ -237,7 +237,7 @@ async def run_model_stream(self, prompt: str | list[dict], session_id: str, stat except Exception: await tape_fork.discard() raise - return agent.finalize_stream(events, tape_fork.merge) + return events.attach(tape_fork.merge) @hookimpl def continue_prompt(self, prompt: str | list[dict], tape: Tape, state: StreamState) -> str: diff --git a/src/bub/streaming.py b/src/bub/streaming.py index f2c57d42..cef74d36 100644 --- a/src/bub/streaming.py +++ b/src/bub/streaming.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from dataclasses import dataclass from typing import Any, Literal @@ -29,6 +29,18 @@ def __init__(self, iterator: AsyncIterator[StreamEvent], *, state: StreamState | def __aiter__(self) -> AsyncIterator[StreamEvent]: return self._iterator + def attach(self, callback: Callable[[], Awaitable[None]]) -> AsyncStreamEvents: + """Return a stream that awaits the callback when iteration ends.""" + + async def iterator() -> AsyncIterator[StreamEvent]: + try: + async for event in self: + yield event + finally: + await callback() + + return AsyncStreamEvents(iterator(), state=self._state) + @property def error(self) -> BubError | None: return self._state.error diff --git a/tests/test_builtin_hook_impl.py b/tests/test_builtin_hook_impl.py index 4c272ffa..b3d2ce83 100644 --- a/tests/test_builtin_hook_impl.py +++ b/tests/test_builtin_hook_impl.py @@ -57,16 +57,6 @@ def session_tape(self, session_id: str, state: dict[str, object], *, source=None return source.with_context(context) return self.tape.session_tape(session_id, Path(state.get("_runtime_workspace", ".")), context=context) - def finalize_stream(self, events: AsyncStreamEvents, callback) -> AsyncStreamEvents: - async def iterator(): - try: - async for event in events: - yield event - finally: - await callback() - - return AsyncStreamEvents(iterator()) - async def run_stream( self, *, diff --git a/website/src/content/docs/docs/reference/types.mdx b/website/src/content/docs/docs/reference/types.mdx index e88b12a6..7a9fee64 100644 --- a/website/src/content/docs/docs/reference/types.mdx +++ b/website/src/content/docs/docs/reference/types.mdx @@ -65,7 +65,7 @@ Channels and adapters may present these protocol-neutral choices to users. `BubF from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState ``` -`StreamEvent.kind` is one of `text`, `reasoning`, `tool_call`, `tool_result`, `usage`, `error`, or `final`. `AsyncStreamEvents` wraps an async iterator and exposes terminal `error` and `usage` state. +`StreamEvent.kind` is one of `text`, `reasoning`, `tool_call`, `tool_result`, `usage`, `error`, or `final`. `AsyncStreamEvents` wraps an async iterator and exposes terminal `error` and `usage` state. `attach(callback)` returns a stream that shares the same state and awaits the callback when iteration ends. ### Errors — `bub.errors` 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 80b56307..f47d5bd0 100644 --- a/website/src/content/docs/zh-cn/docs/reference/types.mdx +++ b/website/src/content/docs/zh-cn/docs/reference/types.mdx @@ -65,7 +65,7 @@ Channel 与 adapter 可向用户展示这些协议无关的选择项。`BubFrame from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState ``` -`StreamEvent.kind` 可以是 `text`、`reasoning`、`tool_call`、`tool_result`、`usage`、`error` 或 `final`。`AsyncStreamEvents` 包装异步迭代器，并暴露最终 `error` 与 `usage` 状态。 +`StreamEvent.kind` 可以是 `text`、`reasoning`、`tool_call`、`tool_result`、`usage`、`error` 或 `final`。`AsyncStreamEvents` 包装异步迭代器，并暴露最终 `error` 与 `usage` 状态。`attach(callback)` 返回共享相同状态的 stream，并在迭代结束时 await callback。 ### Error — `bub.errors` From 0038d962fb12e14284d74d604f95694a7e0d4905 Mon Sep 17 00:00:00 2001 From: PsiACE Date: Mon, 24 Aug 2026 22:43:42 +0800 Subject: [PATCH 4/4] fix: align forkmerge session behavior --- src/bub/builtin/agent.py | 16 ++++++-------- src/bub/builtin/hook_impl.py | 3 ++- src/bub/builtin/tools.py | 21 ++++++++----------- .../docs/docs/concepts/tape-and-context.mdx | 10 +++++---- .../zh-cn/docs/concepts/tape-and-context.mdx | 10 +++++---- 5 files changed, 29 insertions(+), 31 deletions(-) diff --git a/src/bub/builtin/agent.py b/src/bub/builtin/agent.py index 48c242aa..3ab792e9 100644 --- a/src/bub/builtin/agent.py +++ b/src/bub/builtin/agent.py @@ -109,19 +109,15 @@ def session_tape( self, session_id: str, state: TurnState, - *, - source: Tape | None = None, ) -> Tape: - """Build a plain session tape, optionally from the current agent tape.""" + """Return the ordinary tape addressed by a logical session id.""" state.setdefault("session_id", session_id) - if source is None: - return self.tape.session_tape( - session_id, - workspace_from_state(state), - context=replace(self.tape.context, state=state), - ) - return source.with_context(replace(source.context, state=state)) + return self.tape.session_tape( + session_id, + workspace_from_state(state), + context=replace(self.tape.context, state=state), + ) async def _run_command(self, tape: Tape, *, line: str) -> str: line = line[1:].strip() diff --git a/src/bub/builtin/hook_impl.py b/src/bub/builtin/hook_impl.py index 7a69236e..dc5b0dbe 100644 --- a/src/bub/builtin/hook_impl.py +++ b/src/bub/builtin/hook_impl.py @@ -237,7 +237,8 @@ async def run_model_stream(self, prompt: str | list[dict], session_id: str, stat except Exception: await tape_fork.discard() raise - return events.attach(tape_fork.merge) + finish = tape_fork.discard if session_id.startswith("temp/") else tape_fork.merge + return events.attach(finish) @hookimpl def continue_prompt(self, prompt: str | list[dict], tape: Tape, state: StreamState) -> str: diff --git a/src/bub/builtin/tools.py b/src/bub/builtin/tools.py index d193c598..c0c95709 100644 --- a/src/bub/builtin/tools.py +++ b/src/bub/builtin/tools.py @@ -340,10 +340,8 @@ async def run_subagent(param: SubAgentInput, *, context: ToolContext) -> str: subagent_session = param.session state = {**context.state, "session_id": subagent_session} allowed_tools = resolve_tool_names(param.allowed_tools or None, exclude={"subagent"}) - source = context.tape if param.session in {"inherit", "temp"} else None - tape = agent.session_tape(subagent_session, state, source=source) + tape = agent.session_tape(subagent_session, state) tape_fork = await ForkMergeSidecar.mounted(tape).fork(tape) - output = "" try: events = await agent.run_stream( tape=tape_fork.tape, @@ -352,18 +350,17 @@ async def run_subagent(param: SubAgentInput, *, context: ToolContext) -> str: allowed_tools=allowed_tools, allowed_skills=param.allowed_skills, ) - async for event in events: - if event.kind == "error": - output += f"[Error: {event.data.get('message', 'unknown error')}]" - elif event.kind == "text": - output += str(event.data.get("delta", "")) except Exception: await tape_fork.discard() raise - if param.session == "temp": - await tape_fork.discard() - else: - await tape_fork.merge() + + finish = tape_fork.discard if param.session == "temp" else tape_fork.merge + output = "" + async for event in events.attach(finish): + if event.kind == "error": + output += f"[Error: {event.data.get('message', 'unknown error')}]" + elif event.kind == "text": + output += str(event.data.get("delta", "")) return output 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 07310928..1d122172 100644 --- a/website/src/content/docs/docs/concepts/tape-and-context.mdx +++ b/website/src/content/docs/docs/concepts/tape-and-context.mdx @@ -36,7 +36,7 @@ A *handoff* is a constrained transition: write a new anchor, attach the minimum ### Per-session tape -Each session gets one tape. Bub computes the tape name from the workspace path and the session id (`TapeService.session_tape`): +Each session gets one tape. Bub computes the tape name from the workspace path and the session id (`Tape.session_tape`): ```python workspace_hash = md5(str(workspace.resolve()).encode()).hexdigest()[:16] @@ -62,7 +62,7 @@ Spill writes are recorded as `spill.write`. Framework-managed lifecycle outcomes ### 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. +Before the first turn on a tape, `Tape.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. ### default_tape_context: entries → OpenAI messages @@ -77,9 +77,11 @@ The context selector is a hook (`build_tape_context`), so plugins can replace it 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 +### Session and turn isolation -`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. +Each session id addresses an ordinary session tape. An inherited subagent resolves the parent's session id, a temporary subagent uses a throwaway session id, and a named subagent session addresses its own tape. + +The builtin `forkmerge` sidecar isolates one turn with an in-memory tape overlay. Its caller merges persistent runs or discards temporary runs. The sidecar does not create logical sessions or persistent fork tapes. ### auto_handoff on context overflow 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 59ed2ac0..a920178a 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 @@ -36,7 +36,7 @@ anchor **不是** 删除点。tape 完整保留 anchor 之前的所有内容；a ### per-session tape -每个 session 对应一条 tape。Bub 用 workspace 路径与 session id 计算 tape 名（`TapeService.session_tape`）： +每个 session 对应一条 tape。Bub 用 workspace 路径与 session id 计算 tape 名（`Tape.session_tape`）： ```python workspace_hash = md5(str(workspace.resolve()).encode()).hexdigest()[:16] @@ -62,7 +62,7 @@ spill 写入以 `spill.write` event 记录。框架管理的生命周期结果 ### ensure_bootstrap_anchor -在某条 tape 的第一次 turn 之前，`TapeService.ensure_bootstrap_anchor` 会检查是否存在 anchor entry。如果没有，则写入一条 `session/start` handoff，`state={"owner": "human"}`。这保证每条 tape 在 context 重建时都有起始 anchor。 +在某条 tape 的第一次 turn 之前，`Tape.ensure_bootstrap_anchor` 会检查是否存在 anchor entry。如果没有，则写入一条 `session/start` handoff，`state={"owner": "human"}`。这保证每条 tape 在 context 重建时都有起始 anchor。 ### default_tape_context：entry → OpenAI 消息 @@ -77,9 +77,11 @@ context selector 本身是个 hook（`build_tape_context`），插件可以用 带有 `context=False` 标记的 entry 会在 selector 运行前被移除。spill 运维 event 因此仍可在 tape 上查询，但不会进入默认 context 或插件自定义 context。 -### fork_tape +### session 与 turn 隔离 -`TapeService.fork_tape(tape_name, merge_back=True)` 是由 `ForkTapeStore.fork` 支撑的 async context manager。块内的写入发生在 fork 后的 tape 上；退出时合并回父 tape（若 `merge_back=False` 则丢弃）。可以用它跑子任务，避免污染父 session 的历史，直到决定保留结果。 +每个 session id 对应一条普通 session tape。inherited subagent 重新寻址 parent session id，临时 subagent 使用一次性 session id，命名 subagent session 则寻址自己的 tape。 + +builtin `forkmerge` sidecar 通过内存 tape overlay 隔离一次 turn。调用方 merge 持久运行或 discard 临时运行。sidecar 不创建逻辑 session，也不创建持久 fork tape。 ### context 溢出时的 auto_handoff