From a989d2909ddac9a84a4e77fa05aea13c4bf28b98 Mon Sep 17 00:00:00 2001 From: h Date: Sat, 29 Aug 2026 17:07:43 +0200 Subject: [PATCH] feat(injects,conversations): configurable inject header, urgent inject marks the turn it cut --- src/beaver_gateway/core/conversations.py | 11 +++++-- src/beaver_gateway/core/injects.py | 42 ++++++++++++++++++++++-- src/beaver_gateway/storage/models.py | 3 ++ tests/test_conversations.py | 15 +++++++++ 4 files changed, 66 insertions(+), 5 deletions(-) diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py index 31351eb..0611e43 100644 --- a/src/beaver_gateway/core/conversations.py +++ b/src/beaver_gateway/core/conversations.py @@ -39,6 +39,7 @@ from claude_agent_sdk import ( ) from sqlmodel import col, select +from beaver_gateway.core import injects from beaver_gateway.core.conversation_store import load_messages from beaver_gateway.core.distill import ( Digest, @@ -52,7 +53,7 @@ from beaver_gateway.core.distill import ( trim_summary, written_paths, ) -from beaver_gateway.core.injects import InjectQueue, inject_header +from beaver_gateway.core.injects import InjectQueue from beaver_gateway.core.kinds import KINDS, Kind, as_kind from beaver_gateway.core.transcript import ( messages_from_entries, @@ -157,6 +158,8 @@ class ConversationTexts: """ merge_prompt: str = _DEFAULT_MERGE_PROMPT + inject_header: Callable[[injects.InjectContext], str] = injects.inject_header + """Framing line(s) above each inject: who it is from, whether it cut a turn.""" interrupted: str = "прервано" answered: str = "Пользователь ответил: {answer}" unanswered: str = ( @@ -857,6 +860,7 @@ class Conversations: _log.info( "conversation %s: interrupted for urgent inject", conv.external_id ) + await self._queue.mark_interrupting(item) self._ensure_worker(cast("int", conv.id)) return item @@ -1650,7 +1654,10 @@ class Conversations: prompt += "\n\n" + _bundle(batch[1:]) else: origin = "inject" - prompt = "\n\n".join(f"{inject_header(i.origin)}\n{i.text}" for i in batch) + prompt = "\n\n".join( + f"{self._texts.inject_header(injects.context_of(i))}\n{i.text}" + for i in batch + ) seed = await self._pending_seed(conv) if seed: prompt = f"{seed}\n\n{prompt}" diff --git a/src/beaver_gateway/core/injects.py b/src/beaver_gateway/core/injects.py index ca3f0be..1a0c4e6 100644 --- a/src/beaver_gateway/core/injects.py +++ b/src/beaver_gateway/core/injects.py @@ -9,6 +9,7 @@ runs one worker per conversation over them. A row that is still from __future__ import annotations +from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING, Literal, cast @@ -23,14 +24,40 @@ if TYPE_CHECKING: from beaver_gateway.storage.db import Database from beaver_gateway.storage.models import Conversation -__all__ = ["PRIORITY_RANK", "InjectQueue", "Priority", "inject_header"] +__all__ = ["PRIORITY_RANK", "InjectContext", "InjectQueue", "Priority", "inject_header"] Priority = Literal["urgent", "user", "normal"] PRIORITY_RANK: dict[str, int] = {"urgent": 0, "user": 1, "normal": 2} +INTERRUPTED_TURN = ( + "[этот инжект прервал предыдущий тёрн: «Request interrupted» выше - " + "прерывание, не отказ от тулзы]" +) -def inject_header(origin: str) -> str: - return f"[инжект: {origin} - это не Бобёр, отвечать не нужно, голос не обязателен]" + +@dataclass(frozen=True, slots=True) +class InjectContext: + """What a header renderer knows about one queued inject.""" + + origin: str + priority: str + interrupted_turn: bool + + +def inject_header(ctx: InjectContext) -> str: + """Default framing; a setup overrides it via ``ConversationTexts.inject_header``.""" + head = ( + f"[инжект: {ctx.origin} - это не Бобёр, отвечать не нужно, голос не обязателен]" + ) + return f"{head}\n{INTERRUPTED_TURN}" if ctx.interrupted_turn else head + + +def context_of(item: InjectQueueItem) -> InjectContext: + return InjectContext( + origin=item.origin, + priority=item.priority, + interrupted_turn=bool(item.interrupted_turn), + ) class InjectQueue: @@ -83,6 +110,15 @@ class InjectQueue: ) -> None: await self._mark(items, status=status) + async def mark_interrupting(self, item: InjectQueueItem) -> None: + item.interrupted_turn = True + async with self._db.session() as session: + row = await session.get(InjectQueueItem, item.id) + if row is not None: + row.interrupted_turn = True + session.add(row) + await session.commit() + async def interrupted(self) -> Sequence[InjectQueueItem]: async with self._db.session() as session: result = await session.exec( diff --git a/src/beaver_gateway/storage/models.py b/src/beaver_gateway/storage/models.py index 1ec10a4..907c353 100644 --- a/src/beaver_gateway/storage/models.py +++ b/src/beaver_gateway/storage/models.py @@ -163,6 +163,9 @@ class InjectQueueItem(SQLModel, table=True): text: str status: str = Field(default="queued", index=True) turn_id: str | None = Field(default=None) + interrupted_turn: bool | None = Field(default=None) + """An urgent inject that cut a running turn; the header tells the model + the ``Request interrupted`` tool result above was this, not a refusal.""" created_at: datetime = Field(default_factory=_utcnow) delivered_at: datetime | None = Field(default=None) diff --git a/tests/test_conversations.py b/tests/test_conversations.py index efa5d91..47976f8 100644 --- a/tests/test_conversations.py +++ b/tests/test_conversations.py @@ -274,10 +274,25 @@ async def test_urgent_interrupts_and_goes_first(world: World) -> None: ] assert client.prompts[0] == "first" assert client.prompts[1].startswith("[инжект: крон") + assert "прервал предыдущий тёрн" in client.prompts[1] assert client.prompts[1].endswith("ALERT") assert client.prompts[2] == "second" +async def test_inject_header_is_configurable(world: World) -> None: + from beaver_gateway.core.conversations import ConversationTexts + + world.conversations._texts = ConversationTexts( # noqa: SLF001 + inject_header=lambda ctx: ( + f"[от {ctx.origin}, {ctx.priority}, cut={ctx.interrupted_turn}]" + ) + ) + conv = await world.conversations.create(kind="master", agent="a", origin="test") + await world.conversations.inject(conv, "hi", urgency="urgent", origin="panel") + await world.settle(conv, 1) + assert ScriptedClient.instances[0].prompts[0] == "[от panel, urgent, cut=False]\nhi" + + async def test_normal_injects_ride_with_the_next_user_message(world: World) -> None: conv = await world.conversations.create(kind="master", agent="a", origin="test") await world.conversations.inject(