From 3406e2e17892100aae628eacac2f310470b1319e Mon Sep 17 00:00:00 2001 From: h Date: Sun, 30 Aug 2026 20:34:12 +0200 Subject: [PATCH] feat(injects,conversations,scheduler,gateway_tools,api): wake priority, master and parent aliases, scheduled injects start a turn --- src/beaver_gateway/core/conversations.py | 58 ++++++++++++++++--- src/beaver_gateway/core/gateway_tools.py | 58 ++++++++++++++----- src/beaver_gateway/core/injects.py | 26 ++++++--- src/beaver_gateway/core/scheduler.py | 27 ++++++--- src/beaver_gateway/frontends/api/frontend.py | 24 +++++--- src/beaver_gateway/storage/models.py | 2 +- tests/test_conversations.py | 61 ++++++++++++++++++++ tests/test_scheduler.py | 1 + ui/src/lib/api/client.ts | 2 +- ui/src/lib/api/types.ts | 2 +- ui/src/lib/panel/queue-list.svelte | 1 + 11 files changed, 215 insertions(+), 47 deletions(-) diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py index c10594b..ac5dfea 100644 --- a/src/beaver_gateway/core/conversations.py +++ b/src/beaver_gateway/core/conversations.py @@ -103,6 +103,9 @@ __all__ = [ _log = logging.getLogger("beaver_gateway.core.conversations") +MASTER_ALIAS = "master" +PARENT_ALIAS = "parent" + SEEDS = ("clean", "morning", "copy", "brief") _STATUSES = ("open", "merged", "closed", "archived") _DEFAULT_MERGE_PROMPT = ( @@ -342,6 +345,23 @@ class Conversations: async with self._db.session() as session: return await session.get(Conversation, row_id) + async def resolve( + self, key: str, *, origin: Conversation | None = None + ) -> Conversation | None: + """A conversation by public id or alias. + + ``master`` is the open master, ``parent`` the parent of ``origin`` - + the names a job or a branch can use without knowing today's ids. + """ + key = key.strip() + if key == MASTER_ALIAS: + return await self._open_master() + if key == PARENT_ALIAS: + if origin is None or origin.parent_id is None: + return None + return await self.get_row(origin.parent_id) + return await self.get(key) + async def find( self, *, @@ -596,7 +616,7 @@ class Conversations: if live is not None and live.busy: return True pending = await self._queue.pending(cast("int", row.id)) - return any(i.priority in ("user", "urgent") for i in pending) + return any(i.priority in ("user", "urgent", "wake") for i in pending) # ---- routing ------------------------------------------------------- @@ -841,6 +861,22 @@ class Conversations: origin: str = "system", interrupt: bool = True, ) -> InjectQueueItem: + if conv.kind == "master" and conv.status != "open": + live = await self._open_master() + if live is None: + _log.error( + "inject (%s) for closed master %s: no open master, it stays there", + origin, + conv.external_id, + ) + else: + _log.info( + "inject (%s) for closed master %s goes to %s", + origin, + conv.external_id, + live.external_id, + ) + conv = live item = await self._queue.push( conversation_id=cast("int", conv.id), priority=urgency, @@ -923,12 +959,20 @@ class Conversations: return result async def schedule( - self, conv: Conversation, at: str, text: str, *, dedupe_key: str | None = None + self, + conv: Conversation, + at: str, + text: str, + *, + urgency: Priority = "wake", + dedupe_key: str | None = None, ) -> tuple[int | None, datetime]: if self.scheduler is None: msg = "no scheduler; `schedule` is unavailable" raise RuntimeError(msg) - return await self.scheduler.schedule(conv, at, text, dedupe_key=dedupe_key) + return await self.scheduler.schedule( + conv, at, text, urgency=urgency, dedupe_key=dedupe_key + ) async def schedules(self, conv: Conversation | None = None) -> list[dict[str, Any]]: return await self.scheduler.scheduled(conv) if self.scheduler else [] @@ -1632,15 +1676,15 @@ class Conversations: ) -> tuple[list[InjectQueueItem] | None, float | None]: if not items: return None, None - normals = [i for i in items if i.priority == "normal"] head = items[0] if head.priority == "urgent": return [head], None - if head.priority == "user": - return [head, *normals], None + tail = [i for i in items if i is not head and i.priority in ("wake", "normal")] + if head.priority in ("user", "wake"): + return [head, *tail], None age = (datetime.now(UTC) - _aware(head.created_at)).total_seconds() if age >= self._normal_window: - return normals, None + return [head, *tail], None return None, max(self._normal_window - age, 1.0) async def _run_batch( diff --git a/src/beaver_gateway/core/gateway_tools.py b/src/beaver_gateway/core/gateway_tools.py index dd8efd2..a661c46 100644 --- a/src/beaver_gateway/core/gateway_tools.py +++ b/src/beaver_gateway/core/gateway_tools.py @@ -12,8 +12,18 @@ from typing import TYPE_CHECKING, Any, cast from claude_agent_sdk import create_sdk_mcp_server, tool +from beaver_gateway.core.injects import URGENCY from beaver_gateway.core.kinds import as_kind +URGENCY_HELP = ( + "normal waits for the hourly batch or rides with the next turn, wake " + "starts a turn as soon as the target is idle, urgent cuts a running turn" +) +TARGET_HELP = ( + "conversation id, or `master` for the open master, `parent` for the " + "parent of this conversation" +) + if TYPE_CHECKING: from collections.abc import Iterable @@ -57,6 +67,12 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]: raise LookupError(msg) return conv + async def target(raw: Any) -> Any: + conv = await conversations.resolve(str(raw), origin=await current()) + if conv is None: + _log.warning("gateway tool in %s: conversation %r not found", key, raw) + return conv + @tool( "read_conversation", "Read another conversation (a branch, the master, a deep chat) as plain " @@ -64,16 +80,16 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]: { "type": "object", "properties": { - "id": {"type": "string", "description": "conversation id"}, + "id": {"type": "string", "description": TARGET_HELP}, "window": {"type": "integer", "minimum": 1}, }, "required": ["id"], }, ) async def read_conversation(args: dict[str, Any]) -> dict[str, Any]: - conv = await conversations.get(str(args["id"])) + conv = await target(args["id"]) if conv is None: - return _error(f"conversation {args['id']} not found") + return _error(f"conversation {args['id']} not found ({TARGET_HELP})") text = await conversations.read(conv, window=args.get("window")) return _text(text or "(empty)") @@ -137,14 +153,26 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]: @tool( "schedule", "Promise yourself an inject later: `at` is `+15m`, `+2h`, `+1d` or an " - "ISO datetime; `text` arrives in this conversation at that time.", - {"at": str, "text": str}, + "ISO datetime; `text` arrives in this conversation at that time and " + "starts a turn (urgency wake). " + URGENCY_HELP + ".", + { + "type": "object", + "properties": { + "at": {"type": "string"}, + "text": {"type": "string"}, + "urgency": {"type": "string", "enum": list(URGENCY), "default": "wake"}, + }, + "required": ["at", "text"], + }, ) async def schedule(args: dict[str, Any]) -> dict[str, Any]: conv = await current() try: job_id, when = await conversations.schedule( - conv, str(args["at"]), str(args["text"]) + conv, + str(args["at"]), + str(args["text"]), + urgency=cast("Any", args.get("urgency") or "wake"), ) except (RuntimeError, ValueError) as exc: return _error(str(exc)) @@ -152,15 +180,17 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]: @tool( "inject", - "Put a system-origin message into another conversation's queue.", + "Put a system-origin message into another conversation's queue. " + + URGENCY_HELP + + ". An error here means nothing was delivered.", { "type": "object", "properties": { - "conversation": {"type": "string"}, + "conversation": {"type": "string", "description": TARGET_HELP}, "text": {"type": "string"}, "urgency": { "type": "string", - "enum": ["normal", "urgent"], + "enum": list(URGENCY), "default": "normal", }, }, @@ -168,11 +198,13 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]: }, ) async def inject(args: dict[str, Any]) -> dict[str, Any]: - target = await conversations.get(str(args["conversation"])) - if target is None: - return _error(f"conversation {args['conversation']} not found") + conv = await target(args["conversation"]) + if conv is None: + return _error( + f"conversation {args['conversation']} not found ({TARGET_HELP})" + ) item = await conversations.inject( - target, + conv, str(args["text"]), urgency=cast("Any", args.get("urgency") or "normal"), origin="агент", diff --git a/src/beaver_gateway/core/injects.py b/src/beaver_gateway/core/injects.py index 1a0c4e6..9f9056e 100644 --- a/src/beaver_gateway/core/injects.py +++ b/src/beaver_gateway/core/injects.py @@ -1,10 +1,13 @@ -"""Persisted per-conversation queue with priorities ``urgent > user > normal`` (§3.4). +"""Persisted per-conversation queue, ``urgent > user > wake > normal`` (§3.4). One ``ClaudeSDKClient`` runs one turn at a time, so ordering has to happen before the client: the rows here are the queue, ``core/conversations`` -runs one worker per conversation over them. A row that is still -``running`` when the gateway starts was cut by a restart; it is flagged -``interrupted`` and never re-run. +runs one worker per conversation over them. ``urgent`` cuts a running +turn, ``user`` is the human, ``wake`` starts a turn as soon as the +conversation is idle and takes the queued normals with it, ``normal`` +waits for the batching window or rides with the next turn. A row that is +still ``running`` when the gateway starts was cut by a restart; it is +flagged ``interrupted`` and never re-run. """ from __future__ import annotations @@ -24,10 +27,19 @@ if TYPE_CHECKING: from beaver_gateway.storage.db import Database from beaver_gateway.storage.models import Conversation -__all__ = ["PRIORITY_RANK", "InjectContext", "InjectQueue", "Priority", "inject_header"] +__all__ = [ + "PRIORITY_RANK", + "URGENCY", + "InjectContext", + "InjectQueue", + "Priority", + "inject_header", +] -Priority = Literal["urgent", "user", "normal"] -PRIORITY_RANK: dict[str, int] = {"urgent": 0, "user": 1, "normal": 2} +Priority = Literal["urgent", "user", "wake", "normal"] +PRIORITY_RANK: dict[str, int] = {"urgent": 0, "user": 1, "wake": 2, "normal": 3} +URGENCY: tuple[Priority, ...] = ("normal", "wake", "urgent") +"""What the API and the tools accept for ``urgency``: every priority but ``user``.""" INTERRUPTED_TURN = ( "[этот инжект прервал предыдущий тёрн: «Request interrupted» выше - " diff --git a/src/beaver_gateway/core/scheduler.py b/src/beaver_gateway/core/scheduler.py index dcb1075..95f126c 100644 --- a/src/beaver_gateway/core/scheduler.py +++ b/src/beaver_gateway/core/scheduler.py @@ -93,6 +93,9 @@ class JobRun: ) -> bool: master = await self.master() if master is None: + _log.error( + "job %s: no open master, inject lost: %s", self.job.name, text[:200] + ) return False await self.conversations.inject( master, text, urgency=urgency, origin=origin or self.job.name @@ -311,9 +314,14 @@ class Scheduler: at: str, text: str, *, - urgency: Priority = "normal", + urgency: Priority = "wake", dedupe_key: str | None = None, ) -> tuple[int | None, datetime]: + """A deferred inject. + + ``wake`` by default: it was promised for a time, so it starts a turn + then instead of waiting for the normal window. + """ if self._queries is None: msg = "scheduler needs postgres; `schedule` is unavailable" raise RuntimeError(msg) @@ -355,19 +363,20 @@ class Scheduler: async def _deliver(self, job: PgJob) -> None: data = _decode(job.payload) - conv = await self.conversations.get(str(data.get("conversation", ""))) - if conv is not None and conv.status != "open" and conv.kind == "master": - masters = await self.conversations.find( - kind="master", status="open", limit=1 - ) - conv = masters[0] if masters else None + key = str(data.get("conversation", "")) + conv = await self.conversations.resolve(key) if conv is None: - _log.warning("scheduled inject #%s: conversation gone", job.id) + _log.error( + "scheduled inject #%s lost: conversation %r is gone: %s", + job.id, + key, + str(data.get("text", ""))[:200], + ) return await self.conversations.inject( conv, str(data.get("text", "")), - urgency=cast("Priority", data.get("urgency") or "normal"), + urgency=cast("Priority", data.get("urgency") or "wake"), origin="schedule", ) diff --git a/src/beaver_gateway/frontends/api/frontend.py b/src/beaver_gateway/frontends/api/frontend.py index c2a97f1..5906edc 100644 --- a/src/beaver_gateway/frontends/api/frontend.py +++ b/src/beaver_gateway/frontends/api/frontend.py @@ -31,6 +31,7 @@ from sqlmodel import col, select from beaver_gateway.core import audit from beaver_gateway.core.auth import VALID_SCOPES, hash_token from beaver_gateway.core.conversations import SEEDS, implied_title +from beaver_gateway.core.injects import URGENCY from beaver_gateway.core.kinds import Kind, as_kind from beaver_gateway.frontends._auth import require_token from beaver_gateway.frontends._sse import ( @@ -145,7 +146,7 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa return data async def conv_of(public_id: str) -> Conversation: - conv = await conversations.get(public_id) + conv = await conversations.resolve(public_id) if conv is None: raise HTTPException( status.HTTP_404_NOT_FOUND, f"unknown conversation {public_id}" @@ -158,6 +159,14 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa raise HTTPException(status.HTTP_400_BAD_REQUEST, f"missing `{key}`") return text + def urgency_of(data: dict[str, Any], default: str) -> Any: + urgency = str(data.get("urgency") or default) + if urgency not in URGENCY: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, f"urgency must be one of {URGENCY}" + ) + return urgency + def int_or_none(data: dict[str, Any], key: str) -> int | None: value = data.get(key) if value is None: @@ -343,15 +352,11 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa token = await require_token(request, runtime, scope=SCOPE) conv = await conv_of(public_id) data = await body_of(request) - urgency = str(data.get("urgency") or "normal") - if urgency not in ("normal", "urgent"): - raise HTTPException( - status.HTTP_400_BAD_REQUEST, "urgency must be normal|urgent" - ) + urgency = urgency_of(data, "normal") item = await conversations.inject( conv, text_of(data), - urgency=cast("Any", urgency), + urgency=urgency, origin=str(data.get("origin") or "api"), ) await audit.log( @@ -599,7 +604,10 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa data = await body_of(request) try: job_id, when = await conversations.schedule( - conv, text_of(data, "at"), text_of(data, "text") + conv, + text_of(data, "at"), + text_of(data, "text"), + urgency=urgency_of(data, "wake"), ) except (RuntimeError, ValueError) as exc: raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc diff --git a/src/beaver_gateway/storage/models.py b/src/beaver_gateway/storage/models.py index 907c353..0d8eb36 100644 --- a/src/beaver_gateway/storage/models.py +++ b/src/beaver_gateway/storage/models.py @@ -147,7 +147,7 @@ class ConversationBinding(SQLModel, table=True): class InjectQueueItem(SQLModel, table=True): - """Persisted per-conversation queue (§3.4), priorities ``urgent > user > normal``. + """Persisted per-conversation queue (§3.4), ``urgent > user > wake > normal``. ``status`` walks ``queued -> running -> done``; a row still ``running`` at startup was cut by a restart and becomes ``interrupted`` - it is diff --git a/tests/test_conversations.py b/tests/test_conversations.py index 00b241e..09d63ee 100644 --- a/tests/test_conversations.py +++ b/tests/test_conversations.py @@ -781,3 +781,64 @@ async def test_observer_publishes_tool_results_for_the_panel(world: World) -> No assert result["parent_tool_use_id"] == "toolu_0" assert result["turn_id"] == "t1" and result["is_error"] is False assert result["content"] == "a\nb" + + +async def test_wake_injects_start_a_turn_and_take_normals_along(world: World) -> None: + conv = await world.conversations.create(kind="master", agent="a", origin="test") + await world.conversations.inject(conv, "digest", urgency="normal", origin="watch") + await asyncio.sleep(0.2) + assert await world.statuses(conv) == [("normal", "queued")] + await world.conversations.inject( + conv, "reminder", urgency="wake", origin="schedule" + ) + await world.settle(conv, 2) + prompts = ScriptedClient.instances[0].prompts + assert len(prompts) == 1 + assert prompts[0].startswith("[инжект: schedule") + assert "reminder" in prompts[0] + assert "[инжект: watch" in prompts[0] + assert "digest" in prompts[0] + assert await world.statuses(conv) == [("normal", "done"), ("wake", "done")] + + +async def test_wake_does_not_cut_a_running_turn(world: World) -> None: + conv = await world.conversations.create(kind="master", agent="a", origin="test") + ScriptedClient.hold = asyncio.Event() + await world.conversations.post(conv, "first") + await asyncio.sleep(0.2) + await world.conversations.inject(conv, "later", urgency="wake", origin="schedule") + await asyncio.sleep(0.2) + assert await world.statuses(conv) == [("user", "running"), ("wake", "queued")] + assert not ScriptedClient.instances[0].interrupted + assert await world.conversations.busy(conv) + ScriptedClient.hold.set() + await world.settle(conv, 2) + assert await world.statuses(conv) == [("user", "done"), ("wake", "done")] + assert ScriptedClient.instances[0].prompts[1].endswith("later") + + +async def test_resolve_aliases_master_and_parent(world: World) -> None: + master = await world.conversations.create(kind="master", agent="a", origin="test") + branch = await world.conversations.create( + kind="branch", agent="a", parent=master, origin="test" + ) + resolve = world.conversations.resolve + assert (await resolve("master")).id == master.id + assert (await resolve(" master ")).id == master.id + assert (await resolve("parent", origin=branch)).id == master.id + assert await resolve("parent", origin=master) is None + assert await resolve("parent") is None + assert (await resolve(branch.external_id)).id == branch.id + assert await resolve("nope") is None + await world.conversations.set_status(master, "closed") + assert await resolve("master") is None + + +async def test_inject_into_a_closed_master_lands_in_the_open_one(world: World) -> None: + old = await world.conversations.create(kind="master", agent="a", origin="test") + old = await world.conversations.set_status(old, "closed") + new = await world.conversations.create(kind="master", agent="a", origin="test") + item = await world.conversations.inject(old, "late", urgency="normal", origin="x") + assert item.conversation_id == new.id + assert await world.statuses(old) == [] + assert await world.statuses(new) == [("normal", "queued")] diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index b9dd4e8..f7a0a55 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -88,6 +88,7 @@ async def test_schedule_survives_a_restart(world: World, pg: Pg) -> None: prompt = ScriptedClient_prompts(world)[0] assert prompt.startswith("[инжект: schedule") assert prompt.endswith("push X") + assert await world.statuses(conv) == [("wake", "done")] assert await world.conversations.schedules(conv) == [] await second.stop() diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index 5c0c41a..72a9a10 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -200,7 +200,7 @@ export class ApiClient { inject( id: string, text: string, - urgency: "normal" | "urgent", + urgency: "normal" | "wake" | "urgent", origin = "panel" ): Promise<{ id: string; item: number; priority: string }> { return this.post(`/api/conversations/${id}/inject`, { diff --git a/ui/src/lib/api/types.ts b/ui/src/lib/api/types.ts index 4de9cce..67d344e 100644 --- a/ui/src/lib/api/types.ts +++ b/ui/src/lib/api/types.ts @@ -47,7 +47,7 @@ export interface QueueItem { created_at: string; id: number; origin: string; - priority: "urgent" | "user" | "normal"; + priority: "urgent" | "user" | "wake" | "normal"; status: "queued" | "running" | "done" | "failed" | "interrupted"; text: string; } diff --git a/ui/src/lib/panel/queue-list.svelte b/ui/src/lib/panel/queue-list.svelte index 40c8e7b..6a037db 100644 --- a/ui/src/lib/panel/queue-list.svelte +++ b/ui/src/lib/panel/queue-list.svelte @@ -10,6 +10,7 @@ normal: "text-note", urgent: "text-destructive", user: "text-foreground", + wake: "text-foreground", };