diff --git a/src/beaver_gateway/cli.py b/src/beaver_gateway/cli.py index c30af6b..cd9180d 100644 --- a/src/beaver_gateway/cli.py +++ b/src/beaver_gateway/cli.py @@ -190,8 +190,11 @@ async def _async_main() -> None: store=session_store, texts=gateway.texts, frontends=gateway.frontends, - envelope=Envelope(watch=gateway.watch, tz=gateway.tz), + envelope=Envelope( + watch=gateway.watch, tz=gateway.tz, recall=gateway.recall + ), distiller=gateway.distiller, + user_sink=gateway.user_sink, ) late.conversations = conversations scheduler = Scheduler( diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py index ac5dfea..4115d0f 100644 --- a/src/beaver_gateway/core/conversations.py +++ b/src/beaver_gateway/core/conversations.py @@ -99,6 +99,7 @@ __all__ = [ "DistillResult", "ForkResult", "SeedContext", + "UserSaid", ] _log = logging.getLogger("beaver_gateway.core.conversations") @@ -151,6 +152,17 @@ class NewDayContext: moved: int +@dataclass(frozen=True, slots=True) +class UserSaid: + """One message from the user as it enters a turn (``origin=user``).""" + + conversation_id: str + kind: str + title: str | None + text: str + at: datetime + + @dataclass(frozen=True, slots=True) class ConversationTexts: """Texts the gateway cannot invent for a setup. @@ -262,8 +274,10 @@ class Conversations: question_timeout: float = 600.0, envelope: Envelope | None = None, distiller: Distiller | None = None, + user_sink: Callable[[UserSaid], Awaitable[None] | None] | None = None, ) -> None: self._db = db + self._user_sink = user_sink self._distiller = distiller self._agents = agents self._backends = backends @@ -1696,7 +1710,8 @@ class Conversations: if head.priority == "user": origin = "user" prompt = head.text - envelope = self._envelope_for(conv) + await self._note_user(conv, head.text) + envelope = self._envelope_for(conv, head.text) if envelope: prompt += "\n\n" + envelope if len(batch) > 1: @@ -1734,10 +1749,37 @@ class Conversations: text=text, ) - def _envelope_for(self, conv: Conversation) -> str | None: - if conv.kind != "master" or self._envelope is None: + def _envelope_for(self, conv: Conversation, text: str = "") -> str | None: + """Master gets the whole envelope, a branch only the recall lines (§3.3).""" + if self._envelope is None: return None - return self._envelope.build() + if conv.kind == "master": + return self._envelope.build(text=text, kind="master") + if conv.kind == "branch": + return self._envelope.recall_only(text=text, kind="branch") + return None + + async def _note_user(self, conv: Conversation, text: str) -> None: + """Hand the user's text to the setup's sink. + + The setup keeps its own log of what Бобёр said, outside the + transcript; a failure there never blocks the turn. + """ + if self._user_sink is None: + return + message = UserSaid( + conversation_id=conv.external_id, + kind=conv.kind, + title=conv.title, + text=text, + at=datetime.now(UTC), + ) + try: + result = self._user_sink(message) + if inspect.isawaitable(result): + await result + except Exception: # noqa: BLE001 + _log.exception("user sink failed for %s", conv.external_id) def _observer( self, conv: Conversation, runner: _Runner, turn_id: str, origin: str diff --git a/src/beaver_gateway/core/envelope.py b/src/beaver_gateway/core/envelope.py index bcb57d7..4f3540d 100644 --- a/src/beaver_gateway/core/envelope.py +++ b/src/beaver_gateway/core/envelope.py @@ -9,17 +9,20 @@ it by the queue, with their own header. from __future__ import annotations +import logging from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING from zoneinfo import ZoneInfo if TYPE_CHECKING: - from collections.abc import Sequence + from collections.abc import Callable, Sequence from beaver_gateway.core.watch import Change, VaultWatch -__all__ = ["Envelope", "render"] +__all__ = ["Envelope", "RecallContext", "render"] + +_log = logging.getLogger(__name__) HEADER = ( "[конверт - фоновый сигнал, не обращение; " @@ -27,6 +30,15 @@ HEADER = ( ) +@dataclass(frozen=True, slots=True) +class RecallContext: + """What the setup's ``recall`` hook sees: the user's text and where it landed.""" + + text: str + kind: str + now: datetime + + @dataclass(slots=True) class Envelope: watch: VaultWatch | None = None @@ -35,15 +47,22 @@ class Envelope: per_file: int = 30 names_only_within: float = 600.0 last_at: datetime | None = None + recall: Callable[[RecallContext], str | None] | None = None + """Setup-side lookup run on the user's text at turn start: pointers into + the vault (a person's card, the agent's own notes, due dates) that the + gateway cannot know the paths of. Its lines go under the vault block; + a failure is logged and the envelope goes out without them.""" - def build(self, *, now: datetime | None = None) -> str: + def build( + self, *, now: datetime | None = None, text: str = "", kind: str = "master" + ) -> str: now = now or datetime.now(UTC) changes = self.watch.take() if self.watch is not None else [] names_only = ( self.last_at is not None and (now - self.last_at).total_seconds() < self.names_only_within ) - text = render( + out = render( now=now, tz=self.tz, changes=changes, @@ -53,7 +72,25 @@ class Envelope: per_file=self.per_file, ) self.last_at = now - return text + block = self.recall_block(text=text, kind=kind, now=now) + return f"{out}\n{block}" if block else out + + def recall_only( + self, *, text: str, kind: str, now: datetime | None = None + ) -> str | None: + """The recall lines under the header, without the vault diff (branches).""" + block = self.recall_block(text=text, kind=kind, now=now or datetime.now(UTC)) + return f"{HEADER}\n{block}" if block else None + + def recall_block(self, *, text: str, kind: str, now: datetime) -> str | None: + if self.recall is None or not text.strip(): + return None + try: + block = self.recall(RecallContext(text=text, kind=kind, now=now)) + except Exception: # noqa: BLE001 + _log.exception("recall hook failed") + return None + return block.strip() or None if block else None def render( diff --git a/src/beaver_gateway/core/registry.py b/src/beaver_gateway/core/registry.py index f9b3361..b8fd360 100644 --- a/src/beaver_gateway/core/registry.py +++ b/src/beaver_gateway/core/registry.py @@ -13,11 +13,12 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING if TYPE_CHECKING: - from collections.abc import Iterable, Iterator + from collections.abc import Awaitable, Callable, Iterable, Iterator from beaver_gateway.agents.base import BaseAgent - from beaver_gateway.core.conversations import ConversationTexts + from beaver_gateway.core.conversations import ConversationTexts, UserSaid from beaver_gateway.core.distill import Distiller + from beaver_gateway.core.envelope import RecallContext from beaver_gateway.core.rotation import RotationPolicy from beaver_gateway.core.scheduler import Budget, Job from beaver_gateway.core.watch import VaultWatch @@ -94,6 +95,12 @@ class Gateway: """When a master is rotated (§4.5); ``None`` keeps the defaults.""" watch: VaultWatch | None = None """Vault watcher feeding the envelope (§3.5, §4.6); ``None`` = no vault block.""" + recall: Callable[[RecallContext], str | None] | None = None + """Envelope lookup on the user's text: pointers into the vault (cards, + the agent's notes, due dates) the gateway knows no paths for (§3.3).""" + user_sink: Callable[[UserSaid], Awaitable[None] | None] | None = None + """Sees every user message as it enters a master or branch turn - the + setup's own grep-able log of what the user said, outside the transcript.""" budget: Budget | None = None """Subscription window past which non-critical jobs wait (§4.5).""" distiller: Distiller | None = None diff --git a/tests/test_envelope.py b/tests/test_envelope.py index 22c85a9..c84b097 100644 --- a/tests/test_envelope.py +++ b/tests/test_envelope.py @@ -5,7 +5,8 @@ from pathlib import Path from test_conversations import ScriptedClient, World, world -from beaver_gateway.core.envelope import HEADER, Envelope, render +from beaver_gateway.core.conversations import UserSaid +from beaver_gateway.core.envelope import HEADER, Envelope, RecallContext, render from beaver_gateway.core.watch import Change, VaultWatch, WatchRules __all__ = ["world"] @@ -148,3 +149,70 @@ async def test_master_turn_gets_envelope_after_text_and_before_injects( await world.settle(branch, 1) assert "[конверт" not in ScriptedClient.instances[-1].prompts[0] await asyncio.sleep(0) + + +async def test_recall_lines_follow_the_vault_block_and_reach_branches( + world: World, +) -> None: + root, watch = vault() + seen: list[tuple[str, str]] = [] + + def recall(ctx: RecallContext) -> str | None: + seen.append((ctx.kind, ctx.text)) + return "👤 Прохор - карточка `люди/Прохор.md`" if "Прохор" in ctx.text else None + + noted: list[UserSaid] = [] + world.conversations._envelope = Envelope(watch=watch, tz="UTC", recall=recall) # noqa: SLF001 + world.conversations._user_sink = noted.append # noqa: SLF001 + master = await world.conversations.create(kind="master", agent="a", origin="test") + append(root / "люди" / "Прохор.md", "новое\n") + watch.note(root / "люди" / "Прохор.md") + await world.conversations.post(master, "что там у Прохор") + await world.settle(master, 1) + prompt = ScriptedClient.instances[0].prompts[0] + head, _, rest = prompt.partition("\n\n") + assert head == "что там у Прохор" + lines = rest.splitlines() + assert lines[0] == HEADER + assert "люди/Прохор.md (+1)" in rest + assert lines[-1] == "👤 Прохор - карточка `люди/Прохор.md`" + assert rest.index("люди/Прохор.md (+1)") < rest.index("👤 Прохор") + assert seen == [("master", "что там у Прохор")] + assert [(m.kind, m.text) for m in noted] == [("master", "что там у Прохор")] + + branch = await world.conversations.spawn( + kind="branch", parent=master, seed="clean", text="про Прохор подробнее" + ) + await world.settle(branch, 1) + branch_prompt = ScriptedClient.instances[-1].prompts[0] + assert "про Прохор подробнее\n\n" + HEADER in branch_prompt + assert branch_prompt.endswith(f"{HEADER}\n👤 Прохор - карточка `люди/Прохор.md`") + assert "vault, изменено" not in branch_prompt + # the first branch turn carries the seed line above the text + assert seen[-1][0] == "branch" and seen[-1][1].endswith("про Прохор подробнее") + assert noted[-1].kind == "branch" + + await world.conversations.post(master, "ничего про людей") + await world.settle(master, 2) + second = ScriptedClient.instances[0].prompts[1] + assert "👤" not in second + assert second.partition("\n\n")[2].startswith(HEADER) + await asyncio.sleep(0) + + +async def test_failing_recall_or_sink_never_blocks_the_turn(world: World) -> None: + root, watch = vault() + + def boom(_: object) -> str: + msg = "nope" + raise RuntimeError(msg) + + world.conversations._envelope = Envelope(watch=watch, tz="UTC", recall=boom) # noqa: SLF001 + world.conversations._user_sink = boom # noqa: SLF001 + master = await world.conversations.create(kind="master", agent="a", origin="test") + await world.conversations.post(master, "hello") + await world.settle(master, 1) + prompt = ScriptedClient.instances[0].prompts[0] + assert prompt.startswith("hello\n\n" + HEADER) + assert root.exists() + await asyncio.sleep(0)