feat(envelope,conversations): recall hook under the user text, user_sink for the setup's reply log
This commit is contained in:
@@ -190,8 +190,11 @@ async def _async_main() -> None:
|
|||||||
store=session_store,
|
store=session_store,
|
||||||
texts=gateway.texts,
|
texts=gateway.texts,
|
||||||
frontends=gateway.frontends,
|
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,
|
distiller=gateway.distiller,
|
||||||
|
user_sink=gateway.user_sink,
|
||||||
)
|
)
|
||||||
late.conversations = conversations
|
late.conversations = conversations
|
||||||
scheduler = Scheduler(
|
scheduler = Scheduler(
|
||||||
|
|||||||
@@ -99,6 +99,7 @@ __all__ = [
|
|||||||
"DistillResult",
|
"DistillResult",
|
||||||
"ForkResult",
|
"ForkResult",
|
||||||
"SeedContext",
|
"SeedContext",
|
||||||
|
"UserSaid",
|
||||||
]
|
]
|
||||||
|
|
||||||
_log = logging.getLogger("beaver_gateway.core.conversations")
|
_log = logging.getLogger("beaver_gateway.core.conversations")
|
||||||
@@ -151,6 +152,17 @@ class NewDayContext:
|
|||||||
moved: int
|
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)
|
@dataclass(frozen=True, slots=True)
|
||||||
class ConversationTexts:
|
class ConversationTexts:
|
||||||
"""Texts the gateway cannot invent for a setup.
|
"""Texts the gateway cannot invent for a setup.
|
||||||
@@ -262,8 +274,10 @@ class Conversations:
|
|||||||
question_timeout: float = 600.0,
|
question_timeout: float = 600.0,
|
||||||
envelope: Envelope | None = None,
|
envelope: Envelope | None = None,
|
||||||
distiller: Distiller | None = None,
|
distiller: Distiller | None = None,
|
||||||
|
user_sink: Callable[[UserSaid], Awaitable[None] | None] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
self._db = db
|
self._db = db
|
||||||
|
self._user_sink = user_sink
|
||||||
self._distiller = distiller
|
self._distiller = distiller
|
||||||
self._agents = agents
|
self._agents = agents
|
||||||
self._backends = backends
|
self._backends = backends
|
||||||
@@ -1696,7 +1710,8 @@ class Conversations:
|
|||||||
if head.priority == "user":
|
if head.priority == "user":
|
||||||
origin = "user"
|
origin = "user"
|
||||||
prompt = head.text
|
prompt = head.text
|
||||||
envelope = self._envelope_for(conv)
|
await self._note_user(conv, head.text)
|
||||||
|
envelope = self._envelope_for(conv, head.text)
|
||||||
if envelope:
|
if envelope:
|
||||||
prompt += "\n\n" + envelope
|
prompt += "\n\n" + envelope
|
||||||
if len(batch) > 1:
|
if len(batch) > 1:
|
||||||
@@ -1734,10 +1749,37 @@ class Conversations:
|
|||||||
text=text,
|
text=text,
|
||||||
)
|
)
|
||||||
|
|
||||||
def _envelope_for(self, conv: Conversation) -> str | None:
|
def _envelope_for(self, conv: Conversation, text: str = "") -> str | None:
|
||||||
if conv.kind != "master" or self._envelope is None:
|
"""Master gets the whole envelope, a branch only the recall lines (§3.3)."""
|
||||||
|
if self._envelope is None:
|
||||||
return 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(
|
def _observer(
|
||||||
self, conv: Conversation, runner: _Runner, turn_id: str, origin: str
|
self, conv: Conversation, runner: _Runner, turn_id: str, origin: str
|
||||||
|
|||||||
@@ -9,17 +9,20 @@ it by the queue, with their own header.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
from zoneinfo import ZoneInfo
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from collections.abc import Sequence
|
from collections.abc import Callable, Sequence
|
||||||
|
|
||||||
from beaver_gateway.core.watch import Change, VaultWatch
|
from beaver_gateway.core.watch import Change, VaultWatch
|
||||||
|
|
||||||
__all__ = ["Envelope", "render"]
|
__all__ = ["Envelope", "RecallContext", "render"]
|
||||||
|
|
||||||
|
_log = logging.getLogger(__name__)
|
||||||
|
|
||||||
HEADER = (
|
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)
|
@dataclass(slots=True)
|
||||||
class Envelope:
|
class Envelope:
|
||||||
watch: VaultWatch | None = None
|
watch: VaultWatch | None = None
|
||||||
@@ -35,15 +47,22 @@ class Envelope:
|
|||||||
per_file: int = 30
|
per_file: int = 30
|
||||||
names_only_within: float = 600.0
|
names_only_within: float = 600.0
|
||||||
last_at: datetime | None = None
|
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)
|
now = now or datetime.now(UTC)
|
||||||
changes = self.watch.take() if self.watch is not None else []
|
changes = self.watch.take() if self.watch is not None else []
|
||||||
names_only = (
|
names_only = (
|
||||||
self.last_at is not None
|
self.last_at is not None
|
||||||
and (now - self.last_at).total_seconds() < self.names_only_within
|
and (now - self.last_at).total_seconds() < self.names_only_within
|
||||||
)
|
)
|
||||||
text = render(
|
out = render(
|
||||||
now=now,
|
now=now,
|
||||||
tz=self.tz,
|
tz=self.tz,
|
||||||
changes=changes,
|
changes=changes,
|
||||||
@@ -53,7 +72,25 @@ class Envelope:
|
|||||||
per_file=self.per_file,
|
per_file=self.per_file,
|
||||||
)
|
)
|
||||||
self.last_at = now
|
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(
|
def render(
|
||||||
|
|||||||
@@ -13,11 +13,12 @@ from dataclasses import dataclass, field
|
|||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
if 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.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.distill import Distiller
|
||||||
|
from beaver_gateway.core.envelope import RecallContext
|
||||||
from beaver_gateway.core.rotation import RotationPolicy
|
from beaver_gateway.core.rotation import RotationPolicy
|
||||||
from beaver_gateway.core.scheduler import Budget, Job
|
from beaver_gateway.core.scheduler import Budget, Job
|
||||||
from beaver_gateway.core.watch import VaultWatch
|
from beaver_gateway.core.watch import VaultWatch
|
||||||
@@ -94,6 +95,12 @@ class Gateway:
|
|||||||
"""When a master is rotated (§4.5); ``None`` keeps the defaults."""
|
"""When a master is rotated (§4.5); ``None`` keeps the defaults."""
|
||||||
watch: VaultWatch | None = None
|
watch: VaultWatch | None = None
|
||||||
"""Vault watcher feeding the envelope (§3.5, §4.6); ``None`` = no vault block."""
|
"""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
|
budget: Budget | None = None
|
||||||
"""Subscription window past which non-critical jobs wait (§4.5)."""
|
"""Subscription window past which non-critical jobs wait (§4.5)."""
|
||||||
distiller: Distiller | None = None
|
distiller: Distiller | None = None
|
||||||
|
|||||||
+69
-1
@@ -5,7 +5,8 @@ from pathlib import Path
|
|||||||
|
|
||||||
from test_conversations import ScriptedClient, World, world
|
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
|
from beaver_gateway.core.watch import Change, VaultWatch, WatchRules
|
||||||
|
|
||||||
__all__ = ["world"]
|
__all__ = ["world"]
|
||||||
@@ -148,3 +149,70 @@ async def test_master_turn_gets_envelope_after_text_and_before_injects(
|
|||||||
await world.settle(branch, 1)
|
await world.settle(branch, 1)
|
||||||
assert "[конверт" not in ScriptedClient.instances[-1].prompts[0]
|
assert "[конверт" not in ScriptedClient.instances[-1].prompts[0]
|
||||||
await asyncio.sleep(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)
|
||||||
|
|||||||
Reference in New Issue
Block a user