Compare commits

..
15 Commits
Author SHA1 Message Date
hh 944b68e275 feat(backends,jobs,telegram,conversations): mcp calls carry the conversation origin, quiet branches close asleep, fresh topics get named 2026-10-23 22:24:40 +02:00
hh 7d812c7962 fix(agents,backends): claude sessions read sdk messages up to 16 mib 2026-10-19 15:14:13 +00:00
hh 8f26263d4a feat(conversations,jobs): a manual run of the rotation job rotates the master now 2026-10-19 13:51:15 +02:00
hh 7b36cb2156 feat(keystore,api,ui,docker): encrypted secrets store with a write-only admin page and a secret run helper for the model 2026-10-18 16:18:21 +02:00
hh 2a47c6771d feat(conversations,api,ui): rewind takes the last turns of a thread back, the cut session stays in the store 2026-10-13 02:10:03 +02:00
hh b0e5aaf599 feat(ui): finished turns keep their tool tree, reasoning and say bubbles, the thread stops dragging the reader down 2026-10-13 01:22:44 +02:00
hh 0fec15fb3b feat(telegram): reasoning rides under the draft status while the turn runs 2026-10-13 01:22:44 +02:00
hh 712c24de4e feat(backends,api,ui): process trees with cpu and stderr per session, interrupt, close and kill from the system page 2026-10-13 01:22:41 +02:00
hh 9875f08016 fix(events): a conversation's own stream carries its updates 2026-10-13 01:22:40 +02:00
hh 65564d8541 fix(vault,conversations): the envelope diff drops query blocks and frontmatter, history hands the envelope apart from the user's words 2026-10-13 01:22:36 +02:00
hh 1d1285de46 fix(agents): only models that send progress updates read thinking as progress 2026-09-22 20:53:05 +02:00
hh 120985deab feat(agents,backends,conversations,frontends,ui): progress thinking is told apart by its agent, raycast reasoning stays collapsed, read_conversation keeps progress 2026-09-22 20:51:23 +02:00
hh 13873f857b feat(frontends,ui): opus 5.5 progress updates show in the web panel and the vault transcript 2026-09-22 20:36:23 +02:00
hh bbce425eeb feat(backends,conversations,frontends): opus 5.5 progress updates reach telegram, current claude cli in the image 2026-09-22 18:02:01 +00:00
hh e58153628c fix(vault): the watcher ignores obsidian and lock files before it wakes 2026-09-09 12:28:19 +00:00
76 changed files with 4730 additions and 447 deletions

No files matched your search

+26 -3
View File
@@ -45,17 +45,36 @@ RUN apt-get update \
&& rm -rf /var/lib/apt/lists/* && rm -rf /var/lib/apt/lists/*
# Bun native binary (glibc): `bunx` runs the stdio MCP servers declared in # Bun native binary (glibc): `bunx` runs the stdio MCP servers declared in
# the user's config. The claude CLI itself ships inside the # the user's config.
# claude-agent-sdk wheel (`_bundled/claude`), nothing to install here.
COPY --from=oven/bun:1-slim /usr/local/bin/bun /usr/local/bin/bun COPY --from=oven/bun:1-slim /usr/local/bin/bun /usr/local/bin/bun
RUN ln -s /usr/local/bin/bun /usr/local/bin/bunx RUN ln -s /usr/local/bin/bun /usr/local/bin/bunx
# The claude-agent-sdk wheel bundles a CLI, but it trails the npm release:
# 0.2.157 carries 2.1.277, which does not know claude-opus-5-5 and writes an
# [claude-code:unrecognized_model] warning per turn. Install the npm build and
# point BEAVER_CLAUDE_CLI at it; the adapter prefers it over the bundled one.
#
# Only the glibc platform package is pulled: the meta package also drags in the
# musl build, and this image is glibc for the reason stated at the top. The
# tree lands under /opt world-readable because the model runs as beaver-runner,
# not as root.
ARG CLAUDE_CODE_VERSION=2.1.280
RUN mkdir -p /opt/claude-code \
&& cd /opt/claude-code \
&& echo '{"name":"claude-cli-holder","private":true}' > package.json \
&& bun add @anthropic-ai/claude-code-linux-x64@${CLAUDE_CODE_VERSION} \
&& ln -sf /opt/claude-code/node_modules/@anthropic-ai/claude-code-linux-x64/claude \
/usr/local/bin/claude-cli \
&& chmod -R a+rX /opt/claude-code \
&& /usr/local/bin/claude-cli --version
ENV BEAVER_CLAUDE_CLI=/usr/local/bin/claude-cli
# The model process runs as `beaver-runner`, not as the gateway: the # The model process runs as `beaver-runner`, not as the gateway: the
# adapter spawns claude under this uid with a whitelisted environment. # adapter spawns claude under this uid with a whitelisted environment.
# `acl` lets the entrypoint grant it write access to the vault sub-mounts # `acl` lets the entrypoint grant it write access to the vault sub-mounts
# without chowning files that Obsidian Sync keeps rewriting as root. # without chowning files that Obsidian Sync keeps rewriting as root.
RUN apt-get update \ RUN apt-get update \
&& apt-get install -y --no-install-recommends acl \ && apt-get install -y --no-install-recommends acl sudo \
&& rm -rf /var/lib/apt/lists/* \ && rm -rf /var/lib/apt/lists/* \
&& useradd --system --uid 1001 --create-home --shell /usr/sbin/nologin beaver-runner && useradd --system --uid 1001 --create-home --shell /usr/sbin/nologin beaver-runner
@@ -64,6 +83,10 @@ ENV PATH=/app/.venv/bin:$PATH
COPY --from=builder /app/.venv /app/.venv COPY --from=builder /app/.venv /app/.venv
COPY --from=builder /app /app COPY --from=builder /app /app
COPY --from=ui /ui/build /app/ui/build COPY --from=ui /ui/build /app/ui/build
COPY --chmod=0755 docker/secret /usr/local/bin/secret
COPY --chmod=0755 docker/beaver-secret /usr/local/libexec/beaver-secret
COPY --chmod=0440 docker/sudoers /etc/sudoers.d/beaver-secret
RUN visudo -cf /etc/sudoers.d/beaver-secret
WORKDIR /app WORKDIR /app
ENTRYPOINT ["python", "-m"] ENTRYPOINT ["python", "-m"]
+6 -1
View File
@@ -33,6 +33,11 @@ is a full setup built on them.
aggregated in-process and handed to agents by name. aggregated in-process and handed to agents by name.
- **vault/** - watching a directory of notes for the envelope. - **vault/** - watching a directory of notes for the envelope.
- **security/** - bearer tokens, the audit log, credential redaction. - **security/** - bearer tokens, the audit log, credential redaction.
- **keystore/** - secrets the model runs commands with: values and files
sealed with `SECRETS_KEY` (AES-256-GCM) in the gateway's database, set
write-only from the admin, handed out only through `secret run NAME --
cmd` (one `sudo` rule for `beaver-runner`, a root-only socket, files in
`/dev/shm`, values masked in the command's output).
- **events/** - the in-process bus every frontend taps, and the stream - **events/** - the in-process bus every frontend taps, and the stream
event protocol backends emit. event protocol backends emit.
- **storage/** - SQLModel tables and the Postgres session store. - **storage/** - SQLModel tables and the Postgres session store.
@@ -40,7 +45,7 @@ is a full setup built on them.
`config.py` defines `Gateway` and loads a setup; `app.py` builds the runtime `config.py` defines `Gateway` and loads a setup; `app.py` builds the runtime
from it; `cli.py` is the entrypoint. The admin SPA in `ui/` is the browser from it; `cli.py` is the entrypoint. The admin SPA in `ui/` is the browser
window on all of it: conversations by day, the thread with the tree of tool window on all of it: conversations by day, the thread with the tree of tool
calls, the context of the last turn, memory, jobs, usage, tokens and the audit. calls, the context of the last turn, memory, jobs, usage, tokens, secrets and the audit.
![A conversation in the admin: the thread, the queue, the details](preview/shots/admin-thread.webp) ![A conversation in the admin: the thread, the queue, the details](preview/shots/admin-thread.webp)
+2
View File
@@ -0,0 +1,2 @@
#!/bin/sh
exec /usr/local/bin/python3 -I /app/src/beaver_gateway/keystore/cli.py --root "$@"
Executable
+2
View File
@@ -0,0 +1,2 @@
#!/bin/sh
exec /usr/local/bin/python3 -I /app/src/beaver_gateway/keystore/cli.py "$@"
+2
View File
@@ -0,0 +1,2 @@
Defaults:beaver-runner !syslog
beaver-runner ALL=(root) NOPASSWD: /usr/local/libexec/beaver-secret
+2 -1
View File
@@ -14,8 +14,9 @@ dependencies = [
"anthropic>=0.103.0", "anthropic>=0.103.0",
"anyio>=4.13.0", "anyio>=4.13.0",
"argon2-cffi>=25.1.0", "argon2-cffi>=25.1.0",
"claude-agent-sdk>=0.2.150", "claude-agent-sdk>=0.2.157",
"croniter>=6.2.4", "croniter>=6.2.4",
"cryptography>=48.0.0",
"fastapi>=0.136.1", "fastapi>=0.136.1",
"fastmcp>=3.3.1", "fastmcp>=3.3.1",
"greenlet>=3.5.0", "greenlet>=3.5.0",
+5
View File
@@ -36,3 +36,8 @@ class BaseAgent(BaseModel):
system_prompt: str system_prompt: str
expose_mcps: tuple[ExposedMcp, ...] = () expose_mcps: tuple[ExposedMcp, ...] = ()
accept_client_tools: bool = False accept_client_tools: bool = False
@property
def thinking_is_progress(self) -> bool:
"""Whether non-empty thinking is text for the human rather than reasoning."""
return False
+14 -1
View File
@@ -17,13 +17,19 @@ from beaver_gateway.agents.policy import PolicyRule # noqa: TC001
from beaver_gateway.agents.prompts import PromptSource # noqa: TC001 from beaver_gateway.agents.prompts import PromptSource # noqa: TC001
from beaver_gateway.conversations.kinds import KINDS, Kind from beaver_gateway.conversations.kinds import KINDS, Kind
__all__ = ["ClaudeAgent", "ClaudeOptions", "Prompts", "SkillSets"] __all__ = ["PROGRESS_MODELS", "ClaudeAgent", "ClaudeOptions", "Prompts", "SkillSets"]
PROGRESS_MODELS = frozenset({"claude-opus-5-5", "claude-fable-5-1"})
"""Models whose non-empty thinking is the text written between tool calls."""
class ClaudeOptions(BaseModel): class ClaudeOptions(BaseModel):
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
effort: str | None = None effort: str | None = None
thinking: Mapping[str, str] | None = None
"""Thinking config for the SDK; ``None`` keeps the CLI default."""
include_partial_messages: bool = True include_partial_messages: bool = True
"""Token-level deltas on the wire; off means one delta per finished block.""" """Token-level deltas on the wire; off means one delta per finished block."""
@@ -40,6 +46,7 @@ class ClaudeOptions(BaseModel):
"""Extra inherited variable names to let through the env whitelist.""" """Extra inherited variable names to let through the env whitelist."""
max_turns: int | None = None max_turns: int | None = None
max_buffer_size: int = 16 * 1024 * 1024
session_store_flush: str = "eager" session_store_flush: str = "eager"
"""``eager`` mirrors every transcript frame as it lands, so a gateway """``eager`` mirrors every transcript frame as it lands, so a gateway
killed mid-turn loses at most the frame in flight.""" killed mid-turn loses at most the frame in flight."""
@@ -118,6 +125,12 @@ class ClaudeAgent(BaseAgent):
raise ValueError(msg) raise ValueError(msg)
return self return self
@property
def thinking_is_progress(self) -> bool:
"""A progress model, unless ``display: summarized`` brings reasoning."""
summarized = (self.options.thinking or {}).get("display") == "summarized"
return self.model in PROGRESS_MODELS and not summarized
def prompt_for(self, kind: Kind) -> tuple[PromptSource, ...] | None: def prompt_for(self, kind: Kind) -> tuple[PromptSource, ...] | None:
return self.prompts.for_kind(kind) return self.prompts.for_kind(kind)
+38 -1
View File
@@ -33,6 +33,8 @@ from beaver_gateway.frontends.base import GatewayRuntime
from beaver_gateway.frontends.bearer import require_token from beaver_gateway.frontends.bearer import require_token
from beaver_gateway.frontends.root import build_root_app from beaver_gateway.frontends.root import build_root_app
from beaver_gateway.jobs.scheduler import Scheduler from beaver_gateway.jobs.scheduler import Scheduler
from beaver_gateway.keystore import SecretBox, SecretStore
from beaver_gateway.keystore import serve as serve_secrets
from beaver_gateway.mcp.internal_app import build_internal_app from beaver_gateway.mcp.internal_app import build_internal_app
from beaver_gateway.security.auth import TokenStore, scopes_with from beaver_gateway.security.auth import TokenStore, scopes_with
from beaver_gateway.storage import ( from beaver_gateway.storage import (
@@ -44,7 +46,7 @@ from beaver_gateway.storage import (
) )
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Iterable, Iterator from collections.abc import Iterable, Iterator, Sequence
from claude_agent_sdk import McpSdkServerConfig from claude_agent_sdk import McpSdkServerConfig
from fastmcp import FastMCP from fastmcp import FastMCP
@@ -120,6 +122,7 @@ async def run(gateway: Gateway, settings: Settings) -> None:
mcps = McpRegistry(gateway.mcps) mcps = McpRegistry(gateway.mcps)
db = Database(settings.database_url) db = Database(settings.database_url)
await db.create_all() await db.create_all()
secrets = SecretStore(db, _secret_box(settings))
token_store = TokenStore( token_store = TokenStore(
db, db,
bootstrap=TokenStore.parse_bootstrap(settings.bootstrap_tokens), bootstrap=TokenStore.parse_bootstrap(settings.bootstrap_tokens),
@@ -197,6 +200,7 @@ async def run(gateway: Gateway, settings: Settings) -> None:
public_url=gateway.public_url.rstrip("/") if gateway.public_url else None, public_url=gateway.public_url.rstrip("/") if gateway.public_url else None,
scopes=scopes_with(fe.scope for fe in gateway.frontends), scopes=scopes_with(fe.scope for fe in gateway.frontends),
send_file_roots=_roots(settings.send_file_roots), send_file_roots=_roots(settings.send_file_roots),
secrets=secrets,
) )
for fe in gateway.frontends: for fe in gateway.frontends:
fe.configure(runtime) fe.configure(runtime)
@@ -224,10 +228,43 @@ async def run(gateway: Gateway, settings: Settings) -> None:
tg.create_task(_serve_root(gateway, extra={"/hooks": hooks})) tg.create_task(_serve_root(gateway, extra={"/hooks": hooks}))
if gateway.watch is not None: if gateway.watch is not None:
tg.create_task(gateway.watch.run()) tg.create_task(gateway.watch.run())
if secrets.enabled:
tg.create_task(_serve_secrets(secrets, settings=settings, db=db))
for fe in gateway.frontends: for fe in gateway.frontends:
tg.create_task(fe.serve()) tg.create_task(fe.serve())
def _secret_box(settings: Settings) -> SecretBox | None:
if not settings.secrets_key:
return None
try:
return SecretBox.from_hex(settings.secrets_key)
except ValueError as exc:
_log.error("secrets are off: %s", exc)
return None
async def _serve_secrets(
store: SecretStore, *, settings: Settings, db: Database
) -> None:
async def used(names: Sequence[str]) -> None:
try:
async with db.session() as session:
await append_audit(
session,
actor="runner",
kind="secret_use",
detail={"names": list(names)},
)
except Exception: # noqa: BLE001
_log.exception("secret audit write failed")
try:
await serve_secrets(store, settings.secrets_socket, on_use=used)
except OSError:
_log.exception("secrets socket %s did not open", settings.secrets_socket)
async def _authorize_hook( async def _authorize_hook(
request: Request, *, runtime: GatewayRuntime, scope: str request: Request, *, runtime: GatewayRuntime, scope: str
) -> str: ) -> str:
+48 -13
View File
@@ -20,6 +20,7 @@ import tempfile
import time import time
import uuid import uuid
import warnings import warnings
from collections import deque
from collections.abc import Mapping from collections.abc import Mapping
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
@@ -49,7 +50,12 @@ from claude_agent_sdk import (
from beaver_gateway.agents import policy as policy_mod from beaver_gateway.agents import policy as policy_mod
from beaver_gateway.agents import prompts as prompt_assembly from beaver_gateway.agents import prompts as prompt_assembly
from beaver_gateway.backends.capture import TurnCapture, TurnUsage from beaver_gateway.backends.capture import TurnCapture, TurnUsage
from beaver_gateway.backends.sessions import Session, SessionClient, SessionPool from beaver_gateway.backends.sessions import (
STDERR_LINES,
Session,
SessionClient,
SessionPool,
)
from beaver_gateway.backends.transcript import ( from beaver_gateway.backends.transcript import (
build_entries, build_entries,
close_open_tool_uses, close_open_tool_uses,
@@ -71,6 +77,7 @@ from beaver_gateway.events.stream import (
build_thinking_delta, build_thinking_delta,
build_tool_use_block_start, build_tool_use_block_start,
) )
from beaver_gateway.security.redact import redact
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import ( from collections.abc import (
@@ -104,6 +111,7 @@ warnings.filterwarnings("ignore", category=CanUseToolShadowedWarning)
ASK_TOOL = "AskUserQuestion" ASK_TOOL = "AskUserQuestion"
__all__ = [ __all__ = [
"ORIGIN_HEADER",
"AuditSink", "AuditSink",
"ClaudeSdkBackend", "ClaudeSdkBackend",
"RunnerConfig", "RunnerConfig",
@@ -113,6 +121,9 @@ __all__ = [
"fingerprint", "fingerprint",
] ]
ORIGIN_HEADER = "X-Conversation-Origin"
"""Rides on every exposed MCP request; the gateway proxy passes it upstream."""
ENV_KEEP: tuple[str, ...] = ( ENV_KEEP: tuple[str, ...] = (
"PATH", "PATH",
"HOME", "HOME",
@@ -262,12 +273,7 @@ class ClaudeSdkBackend:
await self._pool.close(key) await self._pool.close(key)
async def interrupt(self, key: str) -> bool: async def interrupt(self, key: str) -> bool:
live = self._pool.get(key) return await self._pool.interrupt(key)
if live is None or not live.busy:
return False
live.interrupt_requested = True
await live.client.interrupt()
return True
def live(self, key: str) -> Session | None: def live(self, key: str) -> Session | None:
return self._pool.get(key) return self._pool.get(key)
@@ -674,7 +680,8 @@ class ClaudeSdkBackend:
async def _spawn( async def _spawn(
self, resume: str | None, *, key: str, spec: _SessionSpec self, resume: str | None, *, key: str, spec: _SessionSpec
) -> Session: ) -> Session:
options = self._build_options(resume, key=key, spec=spec) stderr: deque[str] = deque(maxlen=STDERR_LINES)
options = self._build_options(resume, key=key, spec=spec, stderr=stderr)
client = self._factory(options) client = self._factory(options)
await client.connect() await client.connect()
_log.info( _log.info(
@@ -693,6 +700,7 @@ class ClaudeSdkBackend:
session_id=resume, session_id=resume,
resumed=resume is not None, resumed=resume is not None,
pinned=spec.pinned, pinned=spec.pinned,
stderr=stderr,
) )
live.reader = asyncio.create_task(self._read(live)) live.reader = asyncio.create_task(self._read(live))
return live return live
@@ -701,7 +709,12 @@ class ClaudeSdkBackend:
return _RunnerClient(options, uid=self._uid) return _RunnerClient(options, uid=self._uid)
def _build_options( def _build_options(
self, resume: str | None, *, key: str, spec: _SessionSpec self,
resume: str | None,
*,
key: str,
spec: _SessionSpec,
stderr: deque[str] | None = None,
) -> ClaudeAgentOptions: ) -> ClaudeAgentOptions:
agent = self._agent agent = self._agent
opt = agent.options opt = agent.options
@@ -715,7 +728,14 @@ class ClaudeSdkBackend:
system_prompt = ( system_prompt = (
prompt_assembly.assemble(sources) if sources else agent.system_prompt prompt_assembly.assemble(sources) if sources else agent.system_prompt
) )
servers: dict[str, Any] = dict(self._servers) if spec.tools else {} servers: dict[str, Any] = (
{
name: {**server, "headers": {ORIGIN_HEADER: key}}
for name, server in self._servers.items()
}
if spec.tools
else {}
)
gateway = ( gateway = (
self._tool_server(key, spec.kind) self._tool_server(key, spec.kind)
if spec.tools and self._tool_server is not None and agent.gateway_tools if spec.tools and self._tool_server is not None and agent.gateway_tools
@@ -726,6 +746,7 @@ class ClaudeSdkBackend:
return ClaudeAgentOptions( return ClaudeAgentOptions(
model=agent.model or None, model=agent.model or None,
effort=cast("Any", opt.effort), effort=cast("Any", opt.effort),
thinking=cast("Any", dict(opt.thinking) if opt.thinking else None),
system_prompt=system_prompt, system_prompt=system_prompt,
setting_sources=[], setting_sources=[],
strict_mcp_config=True, strict_mcp_config=True,
@@ -746,11 +767,15 @@ class ClaudeSdkBackend:
plugins=cast("Any", plugins), plugins=cast("Any", plugins),
skills="all" if plugins else None, skills="all" if plugins else None,
max_turns=opt.max_turns, max_turns=opt.max_turns,
stderr=lambda line: _log.warning( max_buffer_size=opt.max_buffer_size,
"claude[%s]: %s", agent.name, line.rstrip() stderr=lambda line: self._stderr(line.rstrip(), stderr),
),
) )
def _stderr(self, line: str, tail: deque[str] | None) -> None:
_log.warning("claude[%s]: %s", self._agent.name, line)
if tail is not None:
tail.append(redact(line))
def _can_use_tool( def _can_use_tool(
self, key: str self, key: str
) -> Callable[ ) -> Callable[
@@ -933,7 +958,17 @@ async def _disconnect(live: Session) -> None:
_log.exception("disconnect failed for session %s", live.session_id) _log.exception("disconnect failed for session %s", live.session_id)
CLI_ENV = "BEAVER_CLAUDE_CLI"
"""Path to a ``claude`` CLI that wins over the one bundled in the wheel."""
def _claude_binary() -> str: def _claude_binary() -> str:
override = os.environ.get(CLI_ENV)
if override:
if not Path(override).is_file():
msg = f"{CLI_ENV} points at {override!r}, which is not a file"
raise FileNotFoundError(msg)
return override
bundled = Path(claude_agent_sdk.__file__).parent / "_bundled" / "claude" bundled = Path(claude_agent_sdk.__file__).parent / "_bundled" / "claude"
if bundled.is_file(): if bundled.is_file():
return str(bundled) return str(bundled)
+118
View File
@@ -0,0 +1,118 @@
"""What runs under the gateway: the process tree with memory and CPU, and the kill."""
from __future__ import annotations
import contextlib
import os
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import psutil
from beaver_gateway.security.redact import redact
__all__ = ["Processes", "subtree"]
_VALUE_HIDDEN = frozenset(
{
"--system-prompt",
"--append-system-prompt",
"--mcp-config",
"--settings",
"--agents",
}
)
_ARG_CLIP = 80
_COMMAND_CLIP = 240
@dataclass(slots=True)
class Processes:
root: int = field(default_factory=os.getpid)
_cpu_seen: dict[int, tuple[float, float]] = field(default_factory=dict)
def tree_parents_first(self) -> list[dict[str, Any]]:
try:
found = psutil.Process(self.root).children(recursive=True)
except psutil.Error:
return []
now = time.monotonic()
rows: dict[int, dict[str, Any]] = {}
for process in found:
with contextlib.suppress(psutil.Error), process.oneshot():
spent = process.cpu_times()
rows[process.pid] = {
"pid": process.pid,
"parent": process.ppid(),
"name": process.name(),
"command": _command(process.cmdline()),
"rss": process.memory_info().rss,
"cpu": self._rate(process.pid, spent.user + spent.system, now),
"age_seconds": round(time.time() - process.create_time(), 1),
}
self._cpu_seen = {p: seen for p, seen in self._cpu_seen.items() if p in rows}
children: dict[int, list[int]] = {}
for pid, row in rows.items():
parent = row["parent"] if row["parent"] in rows else self.root
children.setdefault(parent, []).append(pid)
ordered: list[dict[str, Any]] = []
def walk(parent: int, depth: int) -> None:
for pid in sorted(children.get(parent, ())):
ordered.append({**rows[pid], "depth": depth})
walk(pid, depth + 1)
walk(self.root, 0)
return ordered
def kill_own_descendant(self, pid: int) -> bool:
try:
owned = psutil.Process(self.root).children(recursive=True)
target = next((p for p in owned if p.pid == pid), None)
if target is None:
return False
doomed = [*target.children(recursive=True), target]
except psutil.Error:
return False
for process in doomed:
with contextlib.suppress(psutil.Error):
process.kill()
return True
def _rate(self, pid: int, spent: float, now: float) -> float | None:
before = self._cpu_seen.get(pid)
self._cpu_seen[pid] = (spent, now)
if before is None or now <= before[1]:
return None
return round(100 * max(spent - before[0], 0.0) / (now - before[1]), 1)
def subtree(rows: list[dict[str, Any]], pid: int | None) -> list[dict[str, Any]]:
start = next((i for i, row in enumerate(rows) if row["pid"] == pid), None)
if start is None:
return []
base = rows[start]["depth"]
out = [{**rows[start], "depth": 0}]
for row in rows[start + 1 :]:
if row["depth"] <= base:
break
out.append({**row, "depth": row["depth"] - base})
return out
def _command(argv: list[str]) -> str:
if not argv:
return ""
parts = [Path(argv[0]).name]
hide = False
for arg in argv[1:]:
if hide:
parts.append("…")
hide = False
continue
hide = arg in _VALUE_HIDDEN
parts.append(arg if len(arg) <= _ARG_CLIP else f"{arg[:_ARG_CLIP]}…")
command = redact(" ".join(parts))
return command if len(command) <= _COMMAND_CLIP else f"{command[:_COMMAND_CLIP]}…"
+50 -17
View File
@@ -13,12 +13,15 @@ import contextlib
import logging import logging
import os import os
import time import time
from collections import deque
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, Any, Protocol from typing import TYPE_CHECKING, Any, Protocol
import psutil import psutil
from beaver_gateway.backends.processes import Processes, subtree
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import ( from collections.abc import (
AsyncIterable, AsyncIterable,
@@ -43,6 +46,7 @@ DEFAULT_TTL: Mapping[str, float | None] = {
_EVICT_ORDER = {"fork": 0, "job": 0, "deep": 1, "branch": 2, "master": 3} _EVICT_ORDER = {"fork": 0, "job": 0, "deep": 1, "branch": 2, "master": 3}
_RSS_HEADROOM = 0.8 _RSS_HEADROOM = 0.8
STDERR_LINES = 200
class SessionClient(Protocol): class SessionClient(Protocol):
@@ -86,6 +90,7 @@ class Session:
"""Summary of the last task notification, the pending turn's prompt.""" """Summary of the last task notification, the pending turn's prompt."""
observer: Callable[[Any], None] | None = None observer: Callable[[Any], None] | None = None
"""The last turn's observer; sees subagent traffic between turns.""" """The last turn's observer; sees subagent traffic between turns."""
stderr: deque[str] = field(default_factory=lambda: deque(maxlen=STDERR_LINES))
@property @property
def busy(self) -> bool: def busy(self) -> bool:
@@ -121,6 +126,7 @@ class SessionPool:
self._max_live = max_live self._max_live = max_live
self._ttl = dict(ttl) self._ttl = dict(ttl)
self._reap_interval = reap_interval self._reap_interval = reap_interval
self._processes = Processes()
def __len__(self) -> int: def __len__(self) -> int:
return len(self._sessions) return len(self._sessions)
@@ -168,19 +174,6 @@ class SessionPool:
total += child.memory_info().rss total += child.memory_info().rss
return total return total
@staticmethod
def rss_of(session: Session) -> int | None:
pid = session.pid
if pid is None:
return None
try:
process = psutil.Process(pid)
return process.memory_info().rss + sum(
c.memory_info().rss for c in process.children(recursive=True)
)
except psutil.Error:
return None
def over_limit(self) -> bool: def over_limit(self) -> bool:
if self._rss_limit is not None: if self._rss_limit is not None:
return self.rss() > self._rss_limit * _RSS_HEADROOM return self.rss() > self._rss_limit * _RSS_HEADROOM
@@ -214,11 +207,21 @@ class SessionPool:
_log.info("closing session %s (%s, %s)", session.session_id, session.kind, key) _log.info("closing session %s (%s, %s)", session.session_id, session.kind, key)
if session.reader is not None and session.reader is not asyncio.current_task(): if session.reader is not None and session.reader is not asyncio.current_task():
session.reader.cancel() session.reader.cancel()
if session.attached:
session.inbox.put_nowait(RuntimeError(f"session {key} was closed"))
try: try:
await session.client.disconnect() await session.client.disconnect()
except Exception: # noqa: BLE001 except Exception: # noqa: BLE001
_log.exception("disconnect failed for session %s", session.session_id) _log.exception("disconnect failed for session %s", session.session_id)
async def interrupt(self, key: str) -> bool:
session = self._sessions.get(key)
if session is None or not session.busy:
return False
session.interrupt_requested = True
await session.client.interrupt()
return True
async def close_all(self, *, agent: str | None = None) -> None: async def close_all(self, *, agent: str | None = None) -> None:
for session in list(self._sessions.values()): for session in list(self._sessions.values()):
if agent is None or session.agent == agent: if agent is None or session.agent == agent:
@@ -246,15 +249,25 @@ class SessionPool:
_log.exception("session reaper failed") _log.exception("session reaper failed")
def snapshot(self) -> list[dict[str, Any]]: def snapshot(self) -> list[dict[str, Any]]:
return self.census()["sessions"]
def census(self) -> dict[str, Any]:
now = time.monotonic() now = time.monotonic()
return [ rows = self._processes.tree_parents_first()
owned: set[int] = set()
sessions: list[dict[str, Any]] = []
for s in self._sessions.values():
tree = subtree(rows, s.pid)
owned.update(row["pid"] for row in tree)
sessions.append(
{ {
"key": s.key, "key": s.key,
"agent": s.agent, "agent": s.agent,
"kind": s.kind, "kind": s.kind,
"session_id": s.session_id, "session_id": s.session_id,
"pid": s.pid, "pid": s.pid,
"rss": self.rss_of(s), "rss": sum(row["rss"] for row in tree) if tree else None,
"cpu": _cpu(tree),
"idle_seconds": round(now - s.last_used, 1), "idle_seconds": round(now - s.last_used, 1),
"age_seconds": round(now - s.created_at, 1), "age_seconds": round(now - s.created_at, 1),
"turns": s.turns, "turns": s.turns,
@@ -264,9 +277,29 @@ class SessionPool:
"pending_question": s.pending_question, "pending_question": s.pending_question,
"pinned": s.pinned, "pinned": s.pinned,
"dirty": s.dirty, "dirty": s.dirty,
"processes": tree,
} }
for s in self._sessions.values() )
] return {
"rss": sum(row["rss"] for row in rows),
"rss_limit": self._rss_limit,
"cpu": _cpu(rows),
"sessions": sessions,
"others": [row for row in rows if row["pid"] not in owned],
}
async def kill(self, pid: int) -> bool:
if not self._processes.kill_own_descendant(pid):
return False
for session in list(self._sessions.values()):
if session.pid == pid:
await self.close(session.key)
return True
def _cpu(rows: list[dict[str, Any]]) -> float | None:
rates = [row["cpu"] for row in rows if row["cpu"] is not None]
return round(sum(rates), 1) if rates else None
def cgroup_limit() -> int | None: def cgroup_limit() -> int | None:
+76 -10
View File
@@ -10,7 +10,7 @@ import hashlib
import uuid as _uuid import uuid as _uuid
from collections.abc import Mapping from collections.abc import Mapping
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any, cast
try: try:
from claude_agent_sdk._cli_version import __cli_version__ from claude_agent_sdk._cli_version import __cli_version__
@@ -20,19 +20,23 @@ except ImportError: # pragma: no cover
_cli_version = "2.1.248" _cli_version = "2.1.248"
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Iterable from collections.abc import Iterable, Sequence
__all__ = [ __all__ = [
"CLI_VERSION", "CLI_VERSION",
"build_entries", "build_entries",
"close_open_tool_uses", "close_open_tool_uses",
"fingerprint", "fingerprint",
"last_message_before",
"messages_from_entries", "messages_from_entries",
"open_tool_uses", "open_tool_uses",
"prompt_count", "prompt_count",
"render_messages", "render_messages",
"reply_text",
"spoken_text",
"strip_tool_entries", "strip_tool_entries",
"text_of", "text_of",
"turn_starts",
"window_entries", "window_entries",
] ]
@@ -162,9 +166,9 @@ def build_entries(
def messages_from_entries( def messages_from_entries(
entries: Iterable[Mapping[str, Any]], *, with_ts: bool = False entries: Iterable[Mapping[str, Any]], *, stamped: bool = False
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Messages as the model saw them; ``with_ts`` adds when each one was written.""" """Messages as the model saw them; ``stamped`` adds each one's entry id and time."""
out: list[dict[str, Any]] = [] out: list[dict[str, Any]] = []
last_message_id: str | None = None last_message_id: str | None = None
for entry in entries: for entry in entries:
@@ -187,7 +191,8 @@ def messages_from_entries(
"role": "user", "role": "user",
"content": _user_content(content), "content": _user_content(content),
} }
if with_ts: if stamped:
message_out["id"] = entry.get("uuid")
message_out["ts"] = entry.get("timestamp") message_out["ts"] = entry.get("timestamp")
out.append(message_out) out.append(message_out)
last_message_id = None last_message_id = None
@@ -203,7 +208,8 @@ def messages_from_entries(
out[-1]["content"].extend(blocks) out[-1]["content"].extend(blocks)
else: else:
message_out: dict[str, Any] = {"role": "assistant", "content": blocks} message_out: dict[str, Any] = {"role": "assistant", "content": blocks}
if with_ts: if stamped:
message_out["id"] = entry.get("uuid")
message_out["ts"] = entry.get("timestamp") message_out["ts"] = entry.get("timestamp")
out.append(message_out) out.append(message_out)
last_message_id = message_id last_message_id = message_id
@@ -434,13 +440,72 @@ def prompt_count(entries: Iterable[Mapping[str, Any]]) -> int:
return sum(1 for e in entries if _is_prompt_entry(e)) return sum(1 for e in entries if _is_prompt_entry(e))
def turn_starts(entries: Iterable[Mapping[str, Any]]) -> list[int]:
"""Where each turn opens: the index of every prompt the SDK was handed.
The CLI stamps only those with ``promptSource``; what it writes as a
user on its own (an interruption note, a skill body) stays inside the
turn it belongs to.
"""
return [
index
for index, entry in enumerate(entries)
if entry.get("promptSource") and _is_prompt_entry(entry)
]
def last_message_before(entries: Sequence[Mapping[str, Any]], index: int) -> str | None:
"""Uuid of the last ``user``/``assistant`` entry ahead of ``index``."""
for entry in reversed(entries[:index]):
uuid = entry.get("uuid")
if (
entry.get("type") in _PROMPT_TYPES
and isinstance(uuid, str)
and not entry.get("isSidechain")
):
return uuid
return None
def spoken_text(block: Any, *, thinking_is_progress: bool) -> str:
"""What a content block (dict or model) says to the human.
Text always; thinking only when the agent's thinking is progress
written for the human, never when it is reasoning.
"""
if isinstance(block, Mapping):
fields = cast("Mapping[str, Any]", block)
kind, text, thinking = (
fields.get("type"),
fields.get("text"),
fields.get("thinking"),
)
else:
kind = getattr(block, "type", None)
text, thinking = getattr(block, "text", None), getattr(block, "thinking", None)
if kind == "text":
return str(text or "").strip()
if kind == "thinking" and thinking_is_progress:
return str(thinking or "").strip()
return ""
def reply_text(content: Iterable[Any], *, thinking_is_progress: bool) -> str:
"""The reply as the human reads it: spoken blocks in order, a blank line apart."""
parts = (spoken_text(b, thinking_is_progress=thinking_is_progress) for b in content)
return "\n\n".join(p for p in parts if p)
def render_messages( def render_messages(
messages: Iterable[Mapping[str, Any]], *, window: int | None = None messages: Iterable[Mapping[str, Any]],
*,
thinking_is_progress: bool,
window: int | None = None,
) -> str: ) -> str:
"""Plain-text projection for ``read_conversation``: ``user:``/``assistant:`` turns. """Plain-text projection for ``read_conversation``: ``user:``/``assistant:`` turns.
Tool calls collapse to a one-line summary per assistant turn; tool Tool calls collapse to a one-line summary per assistant turn; tool
results and thinking are dropped. results and reasoning are dropped, progress thinking reads as text.
""" """
turns: list[str] = [] turns: list[str] = []
tools: list[str] = [] tools: list[str] = []
@@ -466,8 +531,9 @@ def render_messages(
turns.append("user:\n" + _text_of_content(content)) turns.append("user:\n" + _text_of_content(content))
continue continue
for block in _assistant_blocks(content): for block in _assistant_blocks(content):
if block.get("type") == "text" and block.get("text"): text = spoken_text(block, thinking_is_progress=thinking_is_progress)
current.append(str(block["text"])) if text:
current.append(text)
elif block.get("type") == "tool_use": elif block.get("type") == "tool_use":
tools.append(str(block.get("name", ""))) tools.append(str(block.get("name", "")))
flush() flush()
@@ -7,7 +7,7 @@ import contextlib
from typing import TYPE_CHECKING, Any, cast from typing import TYPE_CHECKING, Any, cast
from uuid import uuid4 from uuid import uuid4
from beaver_gateway.conversations.spawning import Spawning from beaver_gateway.conversations.rewinding import Rewinding
from beaver_gateway.conversations.state import Question from beaver_gateway.conversations.state import Question
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -16,7 +16,7 @@ if TYPE_CHECKING:
__all__ = ["Questions"] __all__ = ["Questions"]
class Questions(Spawning): class Questions(Rewinding):
async def ask(self, key: str, payload: dict[str, Any]) -> str | None: async def ask(self, key: str, payload: dict[str, Any]) -> str | None:
"""Show the question and wait; ``None`` when nobody answered in time.""" """Show the question and wait; ``None`` when nobody answered in time."""
conv = await self.get(key) conv = await self.get(key)
@@ -0,0 +1,114 @@
"""Taking turns back: the thread moves to a fork that ends before them."""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, cast
from claude_agent_sdk import fork_session_via_store
from beaver_gateway.backends.transcript import (
last_message_before,
messages_from_entries,
reply_text,
text_of,
turn_starts,
)
from beaver_gateway.conversations.spawning import Spawning
if TYPE_CHECKING:
from collections.abc import Sequence
from beaver_gateway.storage.models import Conversation
__all__ = ["Rewinding", "Rewound"]
_log = logging.getLogger(__name__)
PREVIEW = 280
@dataclass(frozen=True, slots=True)
class Rewound:
conversation: Conversation
turns: int
previous_session: str
class Rewinding(Spawning):
async def turns(
self, conv: Conversation, *, limit: int = 10
) -> list[dict[str, Any]]:
"""The latest turns that can be taken back, oldest first; the first stays."""
entries = await self.entries(conv)
starts = turn_starts(entries)
ends = [*starts[1:], len(entries)]
spans = list(zip(starts, ends, strict=True))[1:]
agent = self._agents.get(conv.agent_name)
progress = agent is not None and agent.thinking_is_progress
return [
self._turn(entries[start:end], thinking_is_progress=progress)
for start, end in spans[-limit:]
]
def _turn(
self, entries: Sequence[Any], *, thinking_is_progress: bool
) -> dict[str, Any]:
prompt, *rest = messages_from_entries(entries)
replies = (
reply_text(m["content"], thinking_is_progress=thinking_is_progress)
for m in reversed(rest)
if m["role"] == "assistant"
)
return {
"id": entries[0].get("uuid"),
"ts": entries[0].get("timestamp"),
"text": text_of(self._unwrapped(prompt)["content"])[:PREVIEW],
"reply": next((text for text in replies if text), "")[:PREVIEW],
"failed": any(e.get("isApiErrorMessage") for e in entries),
}
async def rewind(self, conv: Conversation, turn: str) -> Rewound:
"""Take ``turn`` and every turn after it out of what the model sees.
Nothing is deleted: the conversation moves to a fork of its session
that ends on the message before ``turn``, the cut session stays in
the store under its old id.
"""
row_id = cast("int", conv.id)
runner = self._runner(row_id)
if runner.lock.locked():
msg = f"conversation {conv.external_id} is mid-turn; interrupt it first"
raise RuntimeError(msg)
async with runner.lock:
row = await self.get_row(row_id) or conv
entries = await self.entries(row)
starts = turn_starts(entries)
cut = next((i for i in starts[1:] if entries[i].get("uuid") == turn), None)
keep = last_message_before(entries, cut) if cut is not None else None
if cut is None or keep is None:
msg = f"conversation {row.external_id} has no turn {turn} to take back"
raise LookupError(msg)
previous = cast("str", row.session_id)
forked = await fork_session_via_store(
self._store,
previous,
directory=str(self._claude_agent(row.agent_name).cwd),
up_to_message_id=keep,
)
await self._backend(row.agent_name).close(row.external_id)
async def apply(moved: Conversation) -> None:
moved.session_id = forked.session_id
row = await self._update(row, apply)
turns = sum(1 for start in starts if start >= cut)
_log.info(
"rewound %s by %d turns: session %s -> %s",
row.external_id,
turns,
previous,
forked.session_id,
)
return Rewound(conversation=row, turns=turns, previous_session=previous)
+12 -2
View File
@@ -77,9 +77,19 @@ class Rotation:
out.append((master, reason)) out.append((master, reason))
return out return out
async def tick(self, now: datetime | None = None) -> list[Conversation]: async def tick(
self, now: datetime | None = None, *, force: str | None = None
) -> list[Conversation]:
"""Rotate the masters that are due; ``force`` rotates every open master."""
rotated: list[Conversation] = [] rotated: list[Conversation] = []
for master, reason in await self.due(now): if force:
due = [
(m, force)
for m in await self._conversations.find(kind="master", status="open")
]
else:
due = await self.due(now)
for master, reason in due:
new = await self.rotate(master, reason) new = await self.rotate(master, reason)
if new is not None: if new is not None:
rotated.append(new) rotated.append(new)
+34 -3
View File
@@ -19,6 +19,7 @@ from beaver_gateway.backends.transcript import (
) )
from beaver_gateway.conversations.kinds import KINDS, Kind from beaver_gateway.conversations.kinds import KINDS, Kind
from beaver_gateway.conversations.state import State, iso from beaver_gateway.conversations.state import State, iso
from beaver_gateway.conversations.texts import split_service
from beaver_gateway.frontends.markdown.history import load_messages from beaver_gateway.frontends.markdown.history import load_messages
from beaver_gateway.storage.models import ( from beaver_gateway.storage.models import (
Conversation, Conversation,
@@ -352,6 +353,8 @@ class Rows(State):
for b in await self.bindings(conv) for b in await self.bindings(conv)
] ]
live = self._pool.get(conv.external_id) live = self._pool.get(conv.external_id)
agent = self._agents.get(conv.agent_name)
out["thinking_is_progress"] = agent is not None and agent.thinking_is_progress
out["live"] = live is not None out["live"] = live is not None
out["busy"] = live.busy if live is not None else False out["busy"] = live.busy if live is not None else False
runner = self._runners.get(cast("int", conv.id)) runner = self._runners.get(cast("int", conv.id))
@@ -466,7 +469,13 @@ class Rows(State):
return marked return marked
async def read(self, conv: Conversation, *, window: int | None = None) -> str: async def read(self, conv: Conversation, *, window: int | None = None) -> str:
return render_messages(await self.history(conv), window=window) return render_messages(
await self.history(conv),
thinking_is_progress=self._claude_agent(
conv.agent_name
).thinking_is_progress,
window=window,
)
async def history(self, conv: Conversation) -> list[dict[str, Any]]: async def history(self, conv: Conversation) -> list[dict[str, Any]]:
if conv.session_id is None: if conv.session_id is None:
@@ -474,9 +483,31 @@ class Rows(State):
return await load_messages( return await load_messages(
session, conversation_id=cast("int", conv.id) session, conversation_id=cast("int", conv.id)
) )
return messages_from_entries( messages = messages_from_entries(
cast("Any", await self.entries(conv)), with_ts=True cast("Any", await self.entries(conv)), stamped=True
) )
return [self._unwrapped(message) for message in messages]
def _unwrapped(self, message: dict[str, Any]) -> dict[str, Any]:
if message["role"] != "user":
return message
content = message["content"]
if isinstance(content, str):
text, service = split_service(content, self._texts)
return (
message
if service is None
else {**message, "content": text, "service": service}
)
for index, block in enumerate(content):
if block.get("type") != "text":
continue
text, service = split_service(str(block.get("text") or ""), self._texts)
if service is not None:
blocks = [*content]
blocks[index] = {**block, "text": text}
return {**message, "content": blocks, "service": service}
return message
async def entries(self, conv: Conversation, *, subpath: str = "") -> list[Any]: async def entries(self, conv: Conversation, *, subpath: str = "") -> list[Any]:
if conv.session_id is None: if conv.session_id is None:
+1 -1
View File
@@ -1,7 +1,7 @@
"""``Conversations`` - the service every frontend, job and gateway tool talks to. """``Conversations`` - the service every frontend, job and gateway tool talks to.
Built as layers, one file each: rows → seeds → turns → messaging → Built as layers, one file each: rows → seeds → turns → messaging →
spawning → questions → closing; this file adds start, stop and restart spawning → rewinding → questions → closing; this file adds start, stop and restart
recovery. A turn started by a user message streams back to whoever asked; recovery. A turn started by a user message streams back to whoever asked;
a turn started by an inject streams nowhere and can only speak via ``say``. a turn started by an inject streams nowhere and can only speak via ``say``.
""" """
+1 -2
View File
@@ -148,8 +148,7 @@ class Spawning(Messaging):
) )
if result.text.strip(): if result.text.strip():
await self.inject(parent, result.text, urgency="normal", origin="merge") await self.inject(parent, result.text, urgency="normal", origin="merge")
await self.set_status(conv, "merged") await self.mark_closed(await self.set_status(conv, "merged"))
await self.mark_closed(conv)
self._bus.publish( self._bus.publish(
"conversation.merged", "conversation.merged",
conversation_id=conv.external_id, conversation_id=conv.external_id,
+11
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
@@ -22,6 +23,7 @@ __all__ = [
"NewDayContext", "NewDayContext",
"SeedContext", "SeedContext",
"UserSaid", "UserSaid",
"split_service",
] ]
@@ -127,3 +129,12 @@ class ConversationTexts:
"; {fixed} open tool calls received tool_result «{interrupted}»" "; {fixed} open tool calls received tool_result «{interrupted}»"
) )
envelope: EnvelopeTexts = field(default_factory=EnvelopeTexts) envelope: EnvelopeTexts = field(default_factory=EnvelopeTexts)
def split_service(text: str, texts: ConversationTexts) -> tuple[str, str | None]:
bundle = re.escape(texts.bundle_header).replace(r"\{since\}", ".+")
heads = f"{re.escape(texts.envelope.header)}|{bundle}"
match = re.search(rf"(?:\A|\n\n)(?=(?:{heads})$)", text, re.MULTILINE)
if match is None:
return text, None
return text[: match.start()], text[match.end() :]
+8 -11
View File
@@ -23,7 +23,7 @@ from claude_agent_sdk import (
) )
from beaver_gateway.backends.capture import TurnCapture from beaver_gateway.backends.capture import TurnCapture
from beaver_gateway.backends.transcript import text_of from beaver_gateway.backends.transcript import reply_text, text_of
from beaver_gateway.conversations import injects from beaver_gateway.conversations import injects
from beaver_gateway.conversations.seeds import Seeds from beaver_gateway.conversations.seeds import Seeds
from beaver_gateway.conversations.state import Runner, aware, iso from beaver_gateway.conversations.state import Runner, aware, iso
@@ -68,14 +68,14 @@ class Turns(Seeds):
backend = self._backend(conv.agent_name) backend = self._backend(conv.agent_name)
turn_id = turn_id or f"turn_{uuid4().hex[:12]}" turn_id = turn_id or f"turn_{uuid4().hex[:12]}"
capture = capture or TurnCapture() capture = capture or TurnCapture()
resume = session_id if session_id is not None else conv.session_id
async with runner.lock: async with runner.lock:
runner.turn_id = turn_id runner.turn_id = turn_id
runner.origin = origin runner.origin = origin
runner.text = shown if shown is not None else _prompt_preview(messages) runner.text = shown if shown is not None else _prompt_preview(messages)
runner.started_at = datetime.now(UTC) runner.started_at = datetime.now(UTC)
runner.tools = {} runner.tools = {}
await self._mark_running(conv, turn_id) row = await self._mark_running(conv, turn_id)
resume = session_id if session_id is not None else row.session_id
before = await self.before_turn(conv) before = await self.before_turn(conv)
self._bus.publish( self._bus.publish(
"turn.start", "turn.start",
@@ -164,12 +164,9 @@ class Turns(Seeds):
): ):
acc.feed(event) acc.feed(event)
message = acc.finalize(model=agent.model) message = acc.finalize(model=agent.model)
reply = "\n\n".join( return reply_text(
getattr(b, "text", "") message.content, thinking_is_progress=agent.thinking_is_progress
for b in message.content ), capture
if getattr(b, "type", "") == "text"
).strip()
return reply, capture
async def before_turn(self, conv: Conversation) -> str | None: # noqa: ARG002 async def before_turn(self, conv: Conversation) -> str | None: # noqa: ARG002
return None return None
@@ -466,12 +463,12 @@ class Turns(Seeds):
except Exception: # noqa: BLE001 except Exception: # noqa: BLE001
_log.exception("rate limit write failed") _log.exception("rate limit write failed")
async def _mark_running(self, conv: Conversation, turn_id: str) -> None: async def _mark_running(self, conv: Conversation, turn_id: str) -> Conversation:
async def apply(row: Conversation) -> None: async def apply(row: Conversation) -> None:
row.running_turn = turn_id row.running_turn = turn_id
row.last_activity_at = datetime.now(UTC) row.last_activity_at = datetime.now(UTC)
await self._update(conv, apply) return await self._update(conv, apply)
async def _mark_done( async def _mark_done(
self, conv: Conversation, capture: TurnCapture, *, cut: bool = False self, conv: Conversation, capture: TurnCapture, *, cut: bool = False
+7 -1
View File
@@ -57,10 +57,16 @@ class EventBus:
event = await queue.get() event = await queue.get()
if ( if (
conversation_id is None conversation_id is None
or event.get("conversation_id") == conversation_id or _conversation_of(event) == conversation_id
): ):
yield event yield event
@property @property
def subscribers(self) -> int: def subscribers(self) -> int:
return len(self._subscribers) return len(self._subscribers)
def _conversation_of(event: Event) -> Any:
if event["type"].startswith("conversation."):
return event.get("id")
return event.get("conversation_id")
+9 -9
View File
@@ -17,7 +17,7 @@ from fastapi.responses import JSONResponse, StreamingResponse
from beaver_gateway.agents.claude import ClaudeAgent from beaver_gateway.agents.claude import ClaudeAgent
from beaver_gateway.backends.capture import TurnCapture from beaver_gateway.backends.capture import TurnCapture
from beaver_gateway.backends.transcript import fingerprint, text_of from beaver_gateway.backends.transcript import fingerprint, reply_text, text_of
from beaver_gateway.frontends.accumulate import StreamAccumulator from beaver_gateway.frontends.accumulate import StreamAccumulator
from beaver_gateway.frontends.base import Frontend from beaver_gateway.frontends.base import Frontend
from beaver_gateway.frontends.bearer import require_token from beaver_gateway.frontends.bearer import require_token
@@ -29,6 +29,7 @@ if TYPE_CHECKING:
from anthropic.types import Message, MessageParam from anthropic.types import Message, MessageParam
from beaver_gateway.agents.base import BaseAgent
from beaver_gateway.conversations.service import Conversations from beaver_gateway.conversations.service import Conversations
from beaver_gateway.events.stream import MessageStreamEvent from beaver_gateway.events.stream import MessageStreamEvent
from beaver_gateway.frontends.base import GatewayRuntime from beaver_gateway.frontends.base import GatewayRuntime
@@ -158,6 +159,7 @@ class AnthropicMessagesFrontend(Frontend):
capture=capture, capture=capture,
turn_id=turn_id, turn_id=turn_id,
history=history, history=history,
thinking_is_progress=agent.thinking_is_progress,
) )
else: else:
options: dict[str, Any] = {} options: dict[str, Any] = {}
@@ -170,7 +172,7 @@ class AnthropicMessagesFrontend(Frontend):
async def after(message: Message) -> None: async def after(message: Message) -> None:
await _broadcast_turn( await _broadcast_turn(
runtime, runtime,
agent_name=agent.name, agent=agent,
input_messages=messages, input_messages=messages,
output_message=message, output_message=message,
system=system_str, system=system_str,
@@ -235,12 +237,9 @@ async def _finish(
capture: TurnCapture, capture: TurnCapture,
turn_id: str, turn_id: str,
history: list[dict[str, str]] | None, history: list[dict[str, str]] | None,
thinking_is_progress: bool,
) -> None: ) -> None:
reply = "\n\n".join( reply = reply_text(message.content, thinking_is_progress=thinking_is_progress)
getattr(b, "text", "")
for b in message.content
if getattr(b, "type", "") == "text"
).strip()
tail = capture.synthesized_messages or [ tail = capture.synthesized_messages or [
{"role": "assistant", "content": [{"type": "text", "text": reply}]} {"role": "assistant", "content": [{"type": "text", "text": reply}]}
] ]
@@ -291,7 +290,7 @@ async def _sse(
async def _broadcast_turn( async def _broadcast_turn(
runtime: GatewayRuntime, runtime: GatewayRuntime,
*, *,
agent_name: str, agent: BaseAgent,
input_messages: list[MessageParam], input_messages: list[MessageParam],
output_message: Message, output_message: Message,
system: str | None, system: str | None,
@@ -299,9 +298,10 @@ async def _broadcast_turn(
if not runtime.turn_log_handlers: if not runtime.turn_log_handlers:
return return
record = TurnRecord( record = TurnRecord(
agent_name=agent_name, agent_name=agent.name,
input_messages=list(input_messages), input_messages=list(input_messages),
output_message=output_message, output_message=output_message,
thinking_is_progress=agent.thinking_is_progress,
system=system, system=system,
source=FRONTEND, source=FRONTEND,
) )
+199 -5
View File
@@ -1,11 +1,13 @@
"""``ApiFrontend`` - ``/api/*``: conversations, SSE, sessions, usage, limits. """``ApiFrontend`` - ``/api/*``: conversations, SSE, sessions, usage, limits.
Bearer scope ``api``; token and audit management need ``admin``. Every Bearer scope ``api``; tokens, secrets and audit need ``admin``. Every
write goes through conversations/service.py; the frontend only shapes JSON. write goes through conversations/service.py; the frontend only shapes JSON.
""" """
from __future__ import annotations from __future__ import annotations
import base64
import binascii
import json import json
import logging import logging
import secrets import secrets
@@ -34,6 +36,7 @@ from beaver_gateway.frontends.sse import (
sse_pack, sse_pack,
) )
from beaver_gateway.frontends.urls import frontend_url from beaver_gateway.frontends.urls import frontend_url
from beaver_gateway.keystore import SecretsDisabledError
from beaver_gateway.security import audit from beaver_gateway.security import audit
from beaver_gateway.security.auth import hash_token from beaver_gateway.security.auth import hash_token
from beaver_gateway.storage import ( from beaver_gateway.storage import (
@@ -46,6 +49,7 @@ from beaver_gateway.storage.models import (
Conversation, Conversation,
InjectQueueItem, InjectQueueItem,
RateLimit, RateLimit,
Secret,
Token, Token,
Usage, Usage,
) )
@@ -55,6 +59,7 @@ if TYPE_CHECKING:
from collections.abc import AsyncIterator, Iterable, Sequence from collections.abc import AsyncIterator, Iterable, Sequence
from pathlib import Path from pathlib import Path
from beaver_gateway.backends.sessions import Session, SessionPool
from beaver_gateway.conversations.service import Conversations from beaver_gateway.conversations.service import Conversations
from beaver_gateway.frontends.base import GatewayRuntime from beaver_gateway.frontends.base import GatewayRuntime
from beaver_gateway.jobs.scheduler import Job, Scheduler from beaver_gateway.jobs.scheduler import Job, Scheduler
@@ -78,6 +83,8 @@ MEMORY_MAX_ENTRIES = 5000
SEARCH_MIN = 2 SEARCH_MIN = 2
GRAPH_LIMIT = 120 GRAPH_LIMIT = 120
SEARCH_LIMIT = 20 SEARCH_LIMIT = 20
TURNS_LIMIT = 10
TURNS_MAX = 50
SNIPPET_MAX = 160 SNIPPET_MAX = 160
@@ -577,6 +584,42 @@ def build_app( # noqa: PLR0915
"text": result.text, "text": result.text,
} }
@app.get("/conversations/{public_id}/turns")
async def get_turns(public_id: str, request: Request) -> dict[str, Any]:
await require_token(request, runtime, scope=SCOPE)
conv = await conv_of(public_id)
limit = min(query_int(request, "limit", TURNS_LIMIT), TURNS_MAX)
return {
"id": conv.external_id,
"turns": await conversations.turns(conv, limit=limit),
}
@app.post("/conversations/{public_id}/rewind")
async def post_rewind(public_id: str, request: Request) -> dict[str, Any]:
token = await require_token(request, runtime, scope=ADMIN_SCOPE)
conv = await conv_of(public_id)
turn = text_of(await body_of(request), "turn")
try:
result = await conversations.rewind(conv, turn)
except LookupError as exc:
raise HTTPException(status.HTTP_404_NOT_FOUND, str(exc)) from exc
except RuntimeError as exc:
raise HTTPException(status.HTTP_409_CONFLICT, str(exc)) from exc
out = {
"turns": result.turns,
"session_id": result.conversation.session_id,
"previous_session": result.previous_session,
}
await audit.log(
runtime,
actor=f"token:{token}",
kind="api_rewind",
agent_name=conv.agent_name,
conversation=conv.external_id,
**out,
)
return {"id": conv.external_id, **out}
@app.post("/conversations/{public_id}/bind") @app.post("/conversations/{public_id}/bind")
async def post_bind(public_id: str, request: Request) -> dict[str, Any]: async def post_bind(public_id: str, request: Request) -> dict[str, Any]:
await require_token(request, runtime, scope=SCOPE) await require_token(request, runtime, scope=SCOPE)
@@ -633,15 +676,74 @@ def build_app( # noqa: PLR0915
await require_token(request, runtime, scope=SCOPE) await require_token(request, runtime, scope=SCOPE)
return _sse(runtime, conversation_id=None) return _sse(runtime, conversation_id=None)
def pool_of() -> SessionPool:
if runtime.pool is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "no session pool")
return runtime.pool
def session_of(key: str) -> Session:
session = pool_of().get(key)
if session is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, f"no live session {key}")
return session
@app.get("/sessions") @app.get("/sessions")
async def sessions(request: Request) -> dict[str, Any]: async def sessions(request: Request) -> dict[str, Any]:
await require_token(request, runtime, scope=SCOPE) await require_token(request, runtime, scope=SCOPE)
pool = runtime.pool if runtime.pool is None:
return { return {
"rss": pool.rss() if pool is not None else None, "rss": None,
"rss_limit": pool.rss_limit if pool is not None else None, "rss_limit": None,
"sessions": pool.snapshot() if pool is not None else [], "cpu": None,
"sessions": [],
"others": [],
} }
return runtime.pool.census()
@app.get("/sessions/{key}/stderr")
async def session_stderr(key: str, request: Request) -> dict[str, Any]:
await require_token(request, runtime, scope=ADMIN_SCOPE)
return {"key": key, "lines": list(session_of(key).stderr)}
@app.post("/sessions/{key}/interrupt")
async def session_interrupt(key: str, request: Request) -> dict[str, Any]:
token = await require_token(request, runtime, scope=SCOPE)
session = session_of(key)
interrupted = await pool_of().interrupt(key)
await audit.log(
runtime,
actor=f"token:{token}",
kind="session_interrupt",
agent_name=session.agent,
session=key,
interrupted=interrupted,
)
return {"key": key, "interrupted": interrupted}
@app.post("/sessions/{key}/close")
async def session_close(key: str, request: Request) -> dict[str, Any]:
token = await require_token(request, runtime, scope=ADMIN_SCOPE)
session = session_of(key)
await pool_of().close(key)
await audit.log(
runtime,
actor=f"token:{token}",
kind="session_close",
agent_name=session.agent,
session=key,
busy=session.busy,
)
return {"key": key, "closed": True}
@app.post("/processes/{pid}/kill")
async def process_kill(pid: int, request: Request) -> dict[str, Any]:
token = await require_token(request, runtime, scope=ADMIN_SCOPE)
if not await pool_of().kill(pid):
raise HTTPException(
status.HTTP_404_NOT_FOUND, f"no process {pid} under the gateway"
)
await audit.log(runtime, actor=f"token:{token}", kind="process_kill", pid=pid)
return {"pid": pid, "killed": True}
@app.get("/schedules") @app.get("/schedules")
async def schedules(request: Request) -> dict[str, Any]: async def schedules(request: Request) -> dict[str, Any]:
@@ -880,6 +982,58 @@ def build_app( # noqa: PLR0915
) )
return {"id": token_id, "revoked": True} return {"id": token_id, "revoked": True}
@app.get("/secrets")
async def secrets_list(request: Request) -> dict[str, Any]:
await require_token(request, runtime, scope=ADMIN_SCOPE)
store = runtime.secrets
rows = await store.rows() if store is not None else []
return {
"enabled": store is not None and store.enabled,
"secrets": [_secret_public(r) for r in rows],
}
@app.put("/secrets/{name}")
async def secret_put(name: str, request: Request) -> dict[str, Any]:
actor = await require_token(request, runtime, scope=ADMIN_SCOPE)
store = runtime.secrets
if store is None or not store.enabled:
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE, "SECRETS_KEY is not set"
)
data = await body_of(request)
description = data.get("description")
payload, filename = _secret_payload(data)
try:
row = await store.put(
name,
description=str(description) if description is not None else None,
data=payload,
filename=filename,
)
except SecretsDisabledError as exc:
raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, str(exc)) from exc
except ValueError as exc:
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
await audit.log(
runtime,
actor=f"token:{actor}",
kind="secret_set",
name=name,
replaced=payload is not None,
)
return {"secret": _secret_public(row)}
@app.delete("/secrets/{name}")
async def secret_delete(name: str, request: Request) -> dict[str, Any]:
actor = await require_token(request, runtime, scope=ADMIN_SCOPE)
store = runtime.secrets
if store is None or not await store.delete(name):
raise HTTPException(status.HTTP_404_NOT_FOUND, f"no secret {name}")
await audit.log(
runtime, actor=f"token:{actor}", kind="secret_delete", name=name
)
return {"name": name, "deleted": True}
@app.get("/audit") @app.get("/audit")
async def audit_list(request: Request) -> dict[str, Any]: async def audit_list(request: Request) -> dict[str, Any]:
await require_token(request, runtime, scope=ADMIN_SCOPE) await require_token(request, runtime, scope=ADMIN_SCOPE)
@@ -1156,6 +1310,46 @@ def _token_public(row: Token) -> dict[str, Any]:
} }
def _secret_public(row: Secret) -> dict[str, Any]:
return {
"name": row.name,
"description": row.description,
"kind": "file" if row.filename else "value",
"filename": row.filename,
"size": row.size,
"created_at": _iso(row.created_at),
"updated_at": _iso(row.updated_at),
"used_at": _iso(row.used_at),
}
def _secret_payload(data: dict[str, Any]) -> tuple[bytes | None, str | None]:
value, upload = data.get("value"), data.get("file")
if value is not None and upload is not None:
raise HTTPException(
status.HTTP_400_BAD_REQUEST, "send `value` or `file`, not both"
)
if isinstance(value, str):
if not value:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "`value` is empty")
return value.encode(), None
if isinstance(upload, dict):
filename, content = upload.get("name"), upload.get("content")
if not isinstance(filename, str) or not isinstance(content, str):
raise HTTPException(
status.HTTP_400_BAD_REQUEST, "`file` needs `name` and base64 `content`"
)
try:
return base64.b64decode(content, validate=True), filename
except binascii.Error as exc:
raise HTTPException(
status.HTTP_400_BAD_REQUEST, "`file.content` is not base64"
) from exc
if value is not None or upload is not None:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "`value` must be a string")
return None, None
def _detail(raw: str) -> Any: def _detail(raw: str) -> Any:
try: try:
return json.loads(raw) if raw else {} return json.loads(raw) if raw else {}
+3
View File
@@ -23,6 +23,7 @@ if TYPE_CHECKING:
from beaver_gateway.backends.base import Backend from beaver_gateway.backends.base import Backend
from beaver_gateway.conversations.kinds import Kind from beaver_gateway.conversations.kinds import Kind
from beaver_gateway.frontends.turn_record import TurnRecord from beaver_gateway.frontends.turn_record import TurnRecord
from beaver_gateway.keystore import SecretStore
from beaver_gateway.security.auth import TokenStore from beaver_gateway.security.auth import TokenStore
from beaver_gateway.storage import Database from beaver_gateway.storage import Database
from beaver_gateway.storage.models import Conversation, ConversationBinding from beaver_gateway.storage.models import Conversation, ConversationBinding
@@ -62,6 +63,8 @@ class GatewayRuntime:
"""Every token scope this gateway knows: the builtins plus each frontend's.""" """Every token scope this gateway knows: the builtins plus each frontend's."""
send_file_roots: tuple[Path, ...] = () send_file_roots: tuple[Path, ...] = ()
"""Directories ``send_file`` may read, from ``SEND_FILE_ROOTS``.""" """Directories ``send_file`` may read, from ``SEND_FILE_ROOTS``."""
secrets: SecretStore | None = None
"""Sealed secrets for ``secret run``; ``None`` in setups without them."""
class Frontend(ABC): class Frontend(ABC):
@@ -142,14 +142,16 @@ class CrossFrontendLogger:
new_user = record.input_messages[-1] new_user = record.input_messages[-1]
new_block = renderer.render_user_param(new_user) new_block = renderer.render_user_param(new_user)
new_block = renderer.append_to_body( new_block = renderer.append_to_body(
new_block, renderer.render_assistant_message(record.output_message) new_block,
renderer.render_assistant_message(
record.output_message,
thinking_is_progress=record.thinking_is_progress,
),
) )
new_body = renderer.append_to_body(body, new_block) new_body = renderer.append_to_body(body, new_block)
metadata = dict(parsed.metadata) metadata = dict(parsed.metadata)
else: else:
new_body = _render_full_history( new_body = _render_full_history(record)
record.input_messages, record.output_message
)
metadata = {} metadata = {}
new_body = renderer.append_to_body(new_body, renderer.USER_SCAFFOLD) new_body = renderer.append_to_body(new_body, renderer.USER_SCAFFOLD)
@@ -204,10 +206,10 @@ def _flatten_text(message: Any) -> str:
return "\n\n".join(c for c in chunks if c) return "\n\n".join(c for c in chunks if c)
def _render_full_history(messages: list[MessageParam], assistant: Any) -> str: def _render_full_history(record: TurnRecord) -> str:
"""Render an entire conversation (used when materializing a new logged file).""" """Render an entire conversation (used when materializing a new logged file)."""
blocks: list[str] = [] blocks: list[str] = []
for msg in messages: for msg in record.input_messages:
role = msg.get("role") role = msg.get("role")
if role == "user": if role == "user":
blocks.append(renderer.render_user_param(msg)) blocks.append(renderer.render_user_param(msg))
@@ -215,7 +217,11 @@ def _render_full_history(messages: list[MessageParam], assistant: Any) -> str:
content = msg.get("content", "") content = msg.get("content", "")
text = content if isinstance(content, str) else _content_to_text(content) text = content if isinstance(content, str) else _content_to_text(content)
blocks.append(f"### Assistant:\n\n{text.strip()}\n" if text.strip() else "") blocks.append(f"### Assistant:\n\n{text.strip()}\n" if text.strip() else "")
blocks.append(renderer.render_assistant_message(assistant)) blocks.append(
renderer.render_assistant_message(
record.output_message, thinking_is_progress=record.thinking_is_progress
)
)
body = "" body = ""
for block in blocks: for block in blocks:
if not block: if not block:
@@ -341,7 +341,11 @@ class MarkdownFrontend(Frontend):
file_path=file_path, file_path=file_path,
first_user_text=parsed.turns[0].text, first_user_text=parsed.turns[0].text,
) )
outcome = diff_and_fork(stored=stored_msgs, incoming=parsed.turns) outcome = diff_and_fork(
stored=stored_msgs,
incoming=parsed.turns,
thinking_is_progress=agent.thinking_is_progress,
)
capture = TurnCapture() capture = TurnCapture()
events = runtime.conversations.turn( events = runtime.conversations.turn(
conv, conv,
@@ -356,6 +360,7 @@ class MarkdownFrontend(Frontend):
file_path=file_path, file_path=file_path,
parsed=parsed, parsed=parsed,
model=agent.model or agent.name, model=agent.model or agent.name,
thinking_is_progress=agent.thinking_is_progress,
filename=filename, filename=filename,
write_disk=write_disk, write_disk=write_disk,
) )
@@ -371,6 +376,7 @@ class MarkdownFrontend(Frontend):
parsed=parsed, parsed=parsed,
message=message, message=message,
agent_name=agent.name, agent_name=agent.name,
thinking_is_progress=agent.thinking_is_progress,
conv_external_id=conv_external_id, conv_external_id=conv_external_id,
write_disk=write_disk, write_disk=write_disk,
) )
@@ -388,6 +394,7 @@ class MarkdownFrontend(Frontend):
agent_name=agent.name, agent_name=agent.name,
input_messages=list(parsed.messages), input_messages=list(parsed.messages),
output_message=message, output_message=message,
thinking_is_progress=agent.thinking_is_progress,
system=None, system=None,
source="markdown", source="markdown",
) )
@@ -534,7 +541,11 @@ class MarkdownFrontend(Frontend):
len(stored_msgs), len(stored_msgs),
len(parsed.turns), len(parsed.turns),
) )
outcome = diff_and_fork(stored=stored_msgs, incoming=parsed.turns) outcome = diff_and_fork(
stored=stored_msgs,
incoming=parsed.turns,
thinking_is_progress=agent.thinking_is_progress,
)
_log.info( _log.info(
"chat/stream: file=%s diff_and_fork divergence=%s " "chat/stream: file=%s diff_and_fork divergence=%s "
"backend_msgs=%d persist_msgs=%d", "backend_msgs=%d persist_msgs=%d",
@@ -567,7 +578,9 @@ class MarkdownFrontend(Frontend):
partial = acc.finalize(model=model) partial = acc.finalize(model=model)
if not partial.content: if not partial.content:
return None return None
rendered = renderer.render_assistant_message(partial) rendered = renderer.render_assistant_message(
partial, thinking_is_progress=agent.thinking_is_progress
)
new_body = renderer.append_to_body(parsed.body, rendered) new_body = renderer.append_to_body(parsed.body, rendered)
return reattach_frontmatter(parsed.metadata, new_body) return reattach_frontmatter(parsed.metadata, new_body)
@@ -593,7 +606,10 @@ class MarkdownFrontend(Frontend):
new_body = parsed.body new_body = parsed.body
if partial.content: if partial.content:
new_body = renderer.append_to_body( new_body = renderer.append_to_body(
new_body, renderer.render_assistant_message(partial) new_body,
renderer.render_assistant_message(
partial, thinking_is_progress=agent.thinking_is_progress
),
) )
new_body = renderer.append_to_body(new_body, _render_error_block(exc)) new_body = renderer.append_to_body(new_body, _render_error_block(exc))
if write_disk: if write_disk:
@@ -616,6 +632,7 @@ class MarkdownFrontend(Frontend):
parsed=parsed, parsed=parsed,
message=message, message=message,
agent_name=agent.name, agent_name=agent.name,
thinking_is_progress=agent.thinking_is_progress,
conv_external_id=conv_external_id, conv_external_id=conv_external_id,
write_disk=write_disk, write_disk=write_disk,
) )
@@ -633,6 +650,7 @@ class MarkdownFrontend(Frontend):
agent_name=agent.name, agent_name=agent.name,
input_messages=list(parsed.messages), input_messages=list(parsed.messages),
output_message=message, output_message=message,
thinking_is_progress=agent.thinking_is_progress,
system=None, system=None,
source="markdown", source="markdown",
) )
@@ -659,6 +677,7 @@ class MarkdownFrontend(Frontend):
file_path: Path, file_path: Path,
parsed: parser.ParsedFile, parsed: parser.ParsedFile,
model: str, model: str,
thinking_is_progress: bool,
filename: str, filename: str,
write_disk: bool = True, write_disk: bool = True,
) -> Any: ) -> Any:
@@ -675,7 +694,9 @@ class MarkdownFrontend(Frontend):
partial = acc.finalize(model=model) partial = acc.finalize(model=model)
if not partial.content: if not partial.content:
return return
rendered = renderer.render_assistant_message(partial) rendered = renderer.render_assistant_message(
partial, thinking_is_progress=thinking_is_progress
)
new_body = renderer.append_to_body(parsed.body, rendered) new_body = renderer.append_to_body(parsed.body, rendered)
await write_atomic( await write_atomic(
file_path, reattach_frontmatter(parsed.metadata, new_body) file_path, reattach_frontmatter(parsed.metadata, new_body)
@@ -698,7 +719,10 @@ class MarkdownFrontend(Frontend):
new_body = parsed.body new_body = parsed.body
if partial.content: if partial.content:
new_body = renderer.append_to_body( new_body = renderer.append_to_body(
new_body, renderer.render_assistant_message(partial) new_body,
renderer.render_assistant_message(
partial, thinking_is_progress=thinking_is_progress
),
) )
new_body = renderer.append_to_body(new_body, _render_error_block(exc)) new_body = renderer.append_to_body(new_body, _render_error_block(exc))
if write_disk: if write_disk:
@@ -715,11 +739,14 @@ class MarkdownFrontend(Frontend):
parsed: parser.ParsedFile, parsed: parser.ParsedFile,
message: Any, message: Any,
agent_name: str, agent_name: str,
thinking_is_progress: bool,
conv_external_id: str, conv_external_id: str,
write_disk: bool = True, write_disk: bool = True,
) -> str: ) -> str:
"""Render the assistant turn, refresh frontmatter, write if we own the file.""" """Render the assistant turn, refresh frontmatter, write if we own the file."""
rendered = renderer.render_assistant_message(message) rendered = renderer.render_assistant_message(
message, thinking_is_progress=thinking_is_progress
)
new_body = renderer.append_to_body(parsed.body, rendered) new_body = renderer.append_to_body(parsed.body, rendered)
new_body = renderer.append_to_body(new_body, renderer.USER_SCAFFOLD) new_body = renderer.append_to_body(new_body, renderer.USER_SCAFFOLD)
updated_metadata = dict(parsed.metadata) updated_metadata = dict(parsed.metadata)
@@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Any, cast
from sqlalchemy import delete from sqlalchemy import delete
from sqlmodel import select from sqlmodel import select
from beaver_gateway.backends.transcript import spoken_text
from beaver_gateway.storage.models import Conversation, ConversationMessage from beaver_gateway.storage.models import Conversation, ConversationMessage
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -180,7 +181,9 @@ class _StoredDisplayTurn:
text_segment_count: int text_segment_count: int
def _group_display_turns(stored: list[dict[str, Any]]) -> list[_StoredDisplayTurn]: def _group_display_turns(
stored: list[dict[str, Any]], *, thinking_is_progress: bool
) -> list[_StoredDisplayTurn]:
"""Walk raw stored messages, group them into Obsidian-visible turns. """Walk raw stored messages, group them into Obsidian-visible turns.
A user-prompt message (no ``tool_result`` blocks) opens a new turn; A user-prompt message (no ``tool_result`` blocks) opens a new turn;
@@ -210,7 +213,9 @@ def _group_display_turns(stored: list[dict[str, Any]]) -> list[_StoredDisplayTur
break break
group.append(m) group.append(m)
i += 1 i += 1
spoken, skeleton, text_count = _summarize_assistant_group(group) spoken, skeleton, text_count = _summarize_assistant_group(
group, thinking_is_progress=thinking_is_progress
)
out.append( out.append(
_StoredDisplayTurn( _StoredDisplayTurn(
role="assistant", role="assistant",
@@ -251,12 +256,13 @@ def _user_prompt_text(content: Any) -> str:
def _summarize_assistant_group( def _summarize_assistant_group(
group: list[dict[str, Any]], group: list[dict[str, Any]], *, thinking_is_progress: bool
) -> tuple[str, list[str], int]: ) -> tuple[str, list[str], int]:
"""Compute (spoken_text, tool_skeleton, text_segment_count) for a display group. """Compute (spoken_text, tool_skeleton, text_segment_count) for a display group.
Must mirror ``parser.parse_assistant_structure``: text blocks collapse into Must mirror what the renderer writes and the parser reads back: text and
one segment, tool_use becomes a skeleton entry, tool_result/thinking are invisible. progress thinking collapse into one segment, tool_use becomes a skeleton
entry, tool_result, reasoning and empty thinking are invisible.
""" """
from beaver_gateway.frontends.markdown.parser import TextSegment, ToolSegment from beaver_gateway.frontends.markdown.parser import TextSegment, ToolSegment
@@ -282,12 +288,10 @@ def _summarize_assistant_group(
for blk in content: for blk in content:
if not isinstance(blk, dict): if not isinstance(blk, dict):
continue continue
btype = blk.get("type") text = spoken_text(blk, thinking_is_progress=thinking_is_progress)
if btype == "text":
text = str(blk.get("text", "")).strip()
if text: if text:
pending.append(text) pending.append(text)
elif btype == "tool_use": elif blk.get("type") == "tool_use":
_flush() _flush()
segments.append(ToolSegment(name=str(blk.get("name", "")))) segments.append(ToolSegment(name=str(blk.get("name", ""))))
_flush() _flush()
@@ -299,7 +303,10 @@ def _summarize_assistant_group(
def diff_and_fork( def diff_and_fork(
*, stored: list[dict[str, Any]], incoming: list[ParsedTurn] *,
stored: list[dict[str, Any]],
incoming: list[ParsedTurn],
thinking_is_progress: bool,
) -> ForkOutcome: ) -> ForkOutcome:
"""Align the incoming parsed file against stored history. """Align the incoming parsed file against stored history.
@@ -314,11 +321,15 @@ def diff_and_fork(
) )
raise ValueError(msg) raise ValueError(msg)
stored_groups = _group_display_turns(stored) stored_groups = _group_display_turns(
stored, thinking_is_progress=thinking_is_progress
)
new_user_turn = incoming[-1] new_user_turn = incoming[-1]
prior_incoming = incoming[:-1] prior_incoming = incoming[:-1]
spliced_groups, divergence, edited = _walk_prefix(prior_incoming, stored_groups) spliced_groups, divergence, edited = _walk_prefix(
prior_incoming, stored_groups, thinking_is_progress=thinking_is_progress
)
if divergence is None and len(prior_incoming) < len(stored_groups): if divergence is None and len(prior_incoming) < len(stored_groups):
if _file_lags_store(stored_groups, len(prior_incoming), new_user_turn): if _file_lags_store(stored_groups, len(prior_incoming), new_user_turn):
@@ -358,7 +369,10 @@ def _file_lags_store(
def _walk_prefix( def _walk_prefix(
prior_incoming: list[ParsedTurn], stored_groups: list[_StoredDisplayTurn] prior_incoming: list[ParsedTurn],
stored_groups: list[_StoredDisplayTurn],
*,
thinking_is_progress: bool,
) -> tuple[list[list[dict[str, Any]]], int | None, bool]: ) -> tuple[list[list[dict[str, Any]]], int | None, bool]:
"""Walk incoming vs stored side-by-side until first divergence. """Walk incoming vs stored side-by-side until first divergence.
@@ -391,7 +405,9 @@ def _walk_prefix(
continue continue
if inc_skeleton and inc_text_count != st.text_segment_count: if inc_skeleton and inc_text_count != st.text_segment_count:
return spliced_groups, i, edited return spliced_groups, i, edited
spliced = _splice_assistant_group(stored_group=st, incoming=inc) spliced = _splice_assistant_group(
stored_group=st, incoming=inc, thinking_is_progress=thinking_is_progress
)
if spliced is None: if spliced is None:
return spliced_groups, i, edited return spliced_groups, i, edited
spliced_groups.append(spliced) spliced_groups.append(spliced)
@@ -426,7 +442,10 @@ def _assemble_tail(
def _splice_assistant_group( def _splice_assistant_group(
*, stored_group: _StoredDisplayTurn, incoming: ParsedTurn *,
stored_group: _StoredDisplayTurn,
incoming: ParsedTurn,
thinking_is_progress: bool,
) -> list[dict[str, Any]] | None: ) -> list[dict[str, Any]] | None:
"""Apply the file's prose to a stored assistant turn. """Apply the file's prose to a stored assistant turn.
@@ -434,27 +453,28 @@ def _splice_assistant_group(
markdown can't express. Falls back to rebuilding from the file's markdown can't express. Falls back to rebuilding from the file's
structure when the text blocks don't line up one-to-one. structure when the text blocks don't line up one-to-one.
""" """
preserved = _splice_in_place(stored_group=stored_group, incoming=incoming) preserved = _splice_in_place(
stored_group=stored_group,
incoming=incoming,
thinking_is_progress=thinking_is_progress,
)
if preserved is not None: if preserved is not None:
return preserved return preserved
return _splice_by_rebuild(stored_group=stored_group, incoming=incoming) return _splice_by_rebuild(stored_group=stored_group, incoming=incoming)
def _is_text_block(blk: object) -> bool:
if not isinstance(blk, dict):
return False
block = cast("dict[str, Any]", blk)
return block.get("type") == "text" and bool(str(block.get("text", "")).strip())
def _splice_in_place( def _splice_in_place(
*, stored_group: _StoredDisplayTurn, incoming: ParsedTurn *,
stored_group: _StoredDisplayTurn,
incoming: ParsedTurn,
thinking_is_progress: bool,
) -> list[dict[str, Any]] | None: ) -> list[dict[str, Any]] | None:
"""Copy the stored messages, swapping only their text block contents. """Copy the stored messages, swapping only their text block contents.
Preserves message/thinking-block boundaries that a rebuild would lose Preserves message/thinking-block boundaries that a rebuild would lose
(and that a cache-hit continuation depends on); returns ``None`` if (and that a cache-hit continuation depends on); returns ``None`` if
text blocks aren't 1:1 with incoming segments, so the caller rebuilds instead. text blocks aren't 1:1 with incoming segments or the prose runs through
signed thinking blocks, so the caller rebuilds instead.
""" """
from beaver_gateway.frontends.markdown.parser import TextSegment from beaver_gateway.frontends.markdown.parser import TextSegment
@@ -466,9 +486,12 @@ def _splice_in_place(
content = msg.get("content") content = msg.get("content")
if msg["role"] != "assistant" or not isinstance(content, list): if msg["role"] != "assistant" or not isinstance(content, list):
continue continue
positions.extend( for bi, blk in enumerate(cast("list[dict[str, Any]]", content)):
(mi, bi) for bi, blk in enumerate(content) if _is_text_block(blk) if not spoken_text(blk, thinking_is_progress=thinking_is_progress):
) continue
if blk["type"] == "thinking":
return None
positions.append((mi, bi))
if len(positions) != len(new_texts): if len(positions) != len(new_texts):
return None return None
@@ -12,8 +12,6 @@ from typing import TYPE_CHECKING, Any, cast
from anthropic.types import Message, TextBlock, ThinkingBlock from anthropic.types import Message, TextBlock, ThinkingBlock
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Iterable
from anthropic.types import MessageParam from anthropic.types import MessageParam
@@ -45,24 +43,25 @@ def render_assistant_text(text: str) -> str:
return f"### Assistant:\n\n{text.strip()}\n" return f"### Assistant:\n\n{text.strip()}\n"
def render_assistant_message(message: Message) -> str: def render_assistant_message(message: Message, *, thinking_is_progress: bool) -> str:
"""Render an assistant ``Message`` into a turn block. """Render an assistant ``Message`` into a turn block.
Tool-use blocks render to nothing; of the thinking blocks between two Tool-use blocks render to nothing. Progress thinking renders as text in
texts only the first one reaches the file. place and vanishes when empty; reasoning renders as a collapsed callout,
of which only the first between two texts reaches the file.
""" """
parts: list[str] = ["### Assistant:", ""] parts: list[str] = ["### Assistant:", ""]
after_thinking = False after_reasoning = False
for block in message.content: for block in message.content:
thinking = isinstance(block, ThinkingBlock) if isinstance(block, ThinkingBlock) and not thinking_is_progress:
if thinking and after_thinking: if not after_reasoning:
parts.extend((*_render_reasoning(block.thinking or ""), ""))
after_reasoning = True
continue continue
lines = list(_render_block(block)) text = _spoken(block)
if not lines: if text:
continue parts.extend((text, ""))
after_thinking = thinking after_reasoning = False
parts.extend(lines)
parts.append("")
return "\n".join(parts).rstrip() + "\n" return "\n".join(parts).rstrip() + "\n"
@@ -108,21 +107,17 @@ def summarize_tool_input(name: str, tool_input: object) -> str:
return name return name
def _render_block(block: object) -> Iterable[str]: def _render_reasoning(text: str) -> tuple[str, ...]:
lines = text.strip().splitlines() or [""]
return ("> [!thinking]-", *(f"> {line}" if line else ">" for line in lines))
def _spoken(block: object) -> str:
if isinstance(block, TextBlock): if isinstance(block, TextBlock):
text = (block.text or "").strip() return (block.text or "").strip()
if text:
yield text
return
if isinstance(block, ThinkingBlock): if isinstance(block, ThinkingBlock):
yield from _render_thinking(block.thinking or "") return (block.thinking or "").strip()
return return ""
def _render_thinking(text: str) -> Iterable[str]:
yield "> [!thinking]-"
for line in text.strip().splitlines() or [""]:
yield f"> {line}" if line else ">"
def adaptive_fence(content: str) -> str: def adaptive_fence(content: str) -> str:
@@ -9,6 +9,7 @@ from __future__ import annotations
import asyncio import asyncio
import contextlib import contextlib
import html
import logging import logging
import time import time
import zlib import zlib
@@ -26,6 +27,7 @@ __all__ = ["Draft"]
_log = logging.getLogger("beaver_gateway.frontends.telegram.drafts") _log = logging.getLogger("beaver_gateway.frontends.telegram.drafts")
_TAIL = 3500 _TAIL = 3500
_REASONING_TAIL = 400
_KEEPALIVE = 20.0 _KEEPALIVE = 20.0
@@ -47,6 +49,7 @@ class Draft:
self._interval = interval self._interval = interval
self.status = status self.status = status
self.text = "" self.text = ""
self.reasoning = ""
self._paused = False self._paused = False
self._dirty = True self._dirty = True
self._broken = False self._broken = False
@@ -80,9 +83,20 @@ class Draft:
self.text = self.text.rstrip() + "\n\n" + text.lstrip() self.text = self.text.rstrip() + "\n\n" + text.lstrip()
else: else:
self.text += text self.text += text
self.reasoning = ""
self._paused = False self._paused = False
self._dirty = True self._dirty = True
def think(self, text: str) -> None:
if text:
self.reasoning += text
self._dirty = True
def rethink(self) -> None:
if self.reasoning:
self.reasoning = ""
self._dirty = True
async def stop(self) -> None: async def stop(self) -> None:
"""Stop pushing and wait for the push in flight. """Stop pushing and wait for the push in flight.
@@ -146,8 +160,21 @@ class Draft:
) )
def _render(self) -> tuple[str, str]: def _render(self) -> tuple[str, str]:
rendered, plain = [to_html(self.status)], [self.status]
thought = _tail(self.reasoning.strip(), _REASONING_TAIL)
if thought:
rendered.append(f"<blockquote>{html.escape(thought)}</blockquote>")
plain.append(thought)
tail = self.text[-_TAIL:] tail = self.text[-_TAIL:]
if not tail.strip(): if tail.strip():
return to_html(self.status), self.status rendered.append(to_html_tail(self.text, _TAIL))
rendered = f"{to_html(self.status)}\n\n{to_html_tail(self.text, _TAIL)}" plain.append(tail)
return rendered, f"{self.status}\n\n{tail}" return "\n\n".join(rendered), "\n\n".join(plain)
def _tail(text: str, limit: int) -> str:
if len(text) <= limit:
return text
cut = text[-limit:]
_, _, rest = cut.partition(" ")
return f"… {rest or cut}"
@@ -268,7 +268,10 @@ class TelegramFrontend(Frontend):
) )
async def mark_closed(self, conv: Conversation) -> bool: async def mark_closed(self, conv: Conversation) -> bool:
return await self.mark_topic(conv) merged = conv.status == "merged"
return await self.mark_topic(
conv, self.texts.closed_prefix if merged else self.texts.idle_prefix
)
async def mark_topic(self, conv: Conversation, prefix: str | None = None) -> bool: async def mark_topic(self, conv: Conversation, prefix: str | None = None) -> bool:
"""Rename the topic; closeForumTopic does not exist in private chats.""" """Rename the topic; closeForumTopic does not exist in private chats."""
@@ -279,10 +282,7 @@ class TelegramFrontend(Frontend):
name = self._topic_names.get(target[1]) or conv.title or self.texts.branch name = self._topic_names.get(target[1]) or conv.title or self.texts.branch
if name.startswith(prefix): if name.startswith(prefix):
return True return True
await self.bot.edit_forum_topic( await self._rename(target[1], f"{prefix}{self._unmarked(name)}")
target[0], target[1], name=f"{prefix}{name}"[:128]
)
self._topic_names[target[1]] = f"{prefix}{name}"
return True return True
def file_note(self) -> str: def file_note(self) -> str:
@@ -462,6 +462,36 @@ class TelegramFrontend(Frontend):
attachments=attachments, attachments=attachments,
) )
async def _branch_in_topic(
self, topic: int, *, text: str, attachments: list[dict[str, Any]] | None
) -> None:
"""A fresh topic is named by the first line, a closed one loses its mark."""
current = self._topic_names.get(topic)
previous = None
if current is None:
previous = await self.conversations.find_bound(
frontend=FRONTEND, external_id=self._ext(topic)
)
known = current or (previous.title if previous is not None else None)
title = (self._unmarked(known) if known else "") or self._title_from(text)
await self._branch(topic, title=title, text=text, attachments=attachments)
if current == title:
return
try:
await self._rename(topic, title)
except TelegramAPIError as exc:
_log.warning("topic %s not renamed to %r: %s", topic, title, exc)
async def _rename(self, topic: int, name: str) -> None:
name = name[:128]
await self.bot.edit_forum_topic(self.chat_id, topic, name=name)
self._topic_names[topic] = name
def _unmarked(self, name: str) -> str:
for prefix in (self.texts.closed_prefix, self.texts.idle_prefix):
name = name.removeprefix(prefix)
return name
async def _handle(self, update: Update) -> None: async def _handle(self, update: Update) -> None:
if update.message is not None: if update.message is not None:
await self._on_message(update.message) await self._on_message(update.message)
@@ -592,9 +622,8 @@ class TelegramFrontend(Frontend):
topic = cast("int", thread_id) topic = cast("int", thread_id)
conv = await self._live(topic) conv = await self._live(topic)
if conv is None: if conv is None:
title = self._topic_names.get(topic) or self._title_from(text) await self._branch_in_topic(
await self._branch( topic, text=text, attachments=attachments or None
topic, title=title, text=text, attachments=attachments or None
) )
return return
pending = self.conversations.pending_question(conv.external_id) pending = self.conversations.pending_question(conv.external_id)
@@ -852,7 +881,10 @@ class TelegramFrontend(Frontend):
if event.get("origin") in ("user", "task"): if event.get("origin") in ("user", "task"):
await self._open_draft(key, event, target) await self._open_draft(key, event, target)
case "stream": case "stream":
self._feed_draft(key, event) agents = cast("GatewayRuntime", self._runtime).agents
agent = agents.get(conv.agent_name)
progress = agent is not None and agent.thinking_is_progress
self._feed_draft(key, event, thinking_is_progress=progress)
case "tool": case "tool":
draft = self._drafts.get(key) draft = self._drafts.get(key)
if draft is not None and event.get("parent_tool_use_id") is None: if draft is not None and event.get("parent_tool_use_id") is None:
@@ -971,7 +1003,9 @@ class TelegramFrontend(Frontend):
self._drafts[key] = draft self._drafts[key] = draft
draft.start() draft.start()
def _feed_draft(self, key: str, event: Event) -> None: def _feed_draft(
self, key: str, event: Event, *, thinking_is_progress: bool
) -> None:
draft = self._drafts.get(key) draft = self._drafts.get(key)
if draft is None or event.get("parent_tool_use_id") is not None: if draft is None or event.get("parent_tool_use_id") is not None:
return return
@@ -983,11 +1017,19 @@ class TelegramFrontend(Frontend):
draft.set_status(self.texts.writing) draft.set_status(self.texts.writing)
draft.append(str(delta.get("text") or "")) draft.append(str(delta.get("text") or ""))
elif delta.get("type") == "thinking_delta": elif delta.get("type") == "thinking_delta":
progress = str(delta.get("thinking") or "")
if progress and thinking_is_progress:
draft.set_status(self.texts.writing)
draft.append(progress)
else:
draft.set_status(self.texts.thinking) draft.set_status(self.texts.thinking)
draft.think(progress)
elif kind == "content_block_start": elif kind == "content_block_start":
block = raw.get("content_block") or {} block = raw.get("content_block") or {}
if block.get("type") == "tool_use": if block.get("type") == "tool_use":
draft.set_status(self._tool_status(block.get("name") or "", None)) draft.set_status(self._tool_status(block.get("name") or "", None))
elif block.get("type") == "thinking":
draft.rethink()
def _tool_status(self, name: Any, tool_input: Any) -> str: def _tool_status(self, name: Any, tool_input: Any) -> str:
label = status_label(str(name), tool_input, self.texts.tool_labels) label = status_label(str(name), tool_input, self.texts.tool_labels)
@@ -42,6 +42,7 @@ class TelegramTexts:
help_head: str = "General is the master, any other topic is a branch." help_head: str = "General is the master, any other topic is a branch."
branch: str = "branch" branch: str = "branch"
closed_prefix: str = "✅ " closed_prefix: str = "✅ "
idle_prefix: str = "💤 "
waiting: str = "⏳ thinking" waiting: str = "⏳ thinking"
thinking: str = "🤔 thinking" thinking: str = "🤔 thinking"
writing: str = "✍️ writing" writing: str = "✍️ writing"
@@ -46,6 +46,7 @@ class TurnRecord:
agent_name: str agent_name: str
input_messages: list[MessageParam] input_messages: list[MessageParam]
output_message: Message output_message: Message
thinking_is_progress: bool
system: str | None = None system: str | None = None
source: str = "anthropic" source: str = "anthropic"
+35 -3
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import logging import logging
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any, cast
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Awaitable, Callable, Coroutine from collections.abc import Awaitable, Callable, Coroutine
@@ -73,6 +73,23 @@ class JobRun:
) )
return True return True
async def inject_into(
self,
public_id: str | None,
text: str,
*,
urgency: Priority = "normal",
origin: str | None = None,
) -> bool:
"""Inject into that conversation, or the master once it is gone or over."""
conv = await self.conversations.get(public_id) if public_id else None
if conv is None or conv.status != "open":
return await self.inject_master(text, urgency=urgency, origin=origin)
await self.conversations.inject(
conv, text, urgency=urgency, origin=origin or self.job.name
)
return True
async def spawn_job( async def spawn_job(
self, self,
*, *,
@@ -113,14 +130,29 @@ class JobRun:
_log.exception("closing idle %s failed", conv.external_id) _log.exception("closing idle %s failed", conv.external_id)
return out return out
async def close_quiet(
self, *, kind: str = "branch", days: int = 3, limit: int | None = None
) -> list[Conversation]:
"""Close chats quiet for ``days`` without a digest, their windows marked."""
out: list[Conversation] = []
for conv in await self.conversations.idle(kind=kind, days=days, limit=limit):
if await self.conversations.busy(
conv
) or await self.conversations.queue.pending(cast("int", conv.id)):
continue
row = await self.conversations.close(conv)
await self.conversations.mark_closed(row)
out.append(row)
return out
async def retry_in(self, delay: timedelta) -> None: async def retry_in(self, delay: timedelta) -> None:
await self.scheduler.trigger( await self.scheduler.trigger(
self.job, self.payload, delay=delay, trigger=self.trigger self.job, self.payload, delay=delay, trigger=self.trigger
) )
async def rotate(self) -> list[Conversation]: async def rotate(self, *, force: str | None = None) -> list[Conversation]:
rotation = self.scheduler.rotation rotation = self.scheduler.rotation
return await rotation.tick() if rotation is not None else [] return await rotation.tick(force=force) if rotation is not None else []
def background(self, coro: Coroutine[Any, Any, Any]) -> None: def background(self, coro: Coroutine[Any, Any, Any]) -> None:
self.scheduler.background(coro) self.scheduler.background(coro)
+19
View File
@@ -0,0 +1,19 @@
"""Secrets the model runs commands with: sealed in the database, handed out to root."""
from beaver_gateway.keystore.server import serve
from beaver_gateway.keystore.store import (
NAME,
SecretBox,
SecretsDisabledError,
SecretStore,
UnknownSecretError,
)
__all__ = [
"NAME",
"SecretBox",
"SecretStore",
"SecretsDisabledError",
"UnknownSecretError",
"serve",
]
+289
View File
@@ -0,0 +1,289 @@
"""``secret ls`` and ``secret run NAME... -- command``: stdlib only, run by path.
The model's uid calls this file; it asks the root half (the same file
through ``sudo``) for the values, which reads them from the gateway's
root-only socket. The command then runs as the model's uid with each
value in its own variable, files in a private tmpfs directory, and the
values masked in what it prints.
"""
from __future__ import annotations
import base64
import contextlib
import json
import os
import re
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
from typing import IO, TYPE_CHECKING, Any
from urllib.parse import quote
if TYPE_CHECKING:
from io import BufferedReader
SOCKET = Path("/run/beaver-secrets/gateway.sock")
ROOT_HALF = ("sudo", "-n", "/usr/local/libexec/beaver-secret")
SHM = Path("/dev/shm") # noqa: S108
PREFIX = "beaver-secret-"
NAME = re.compile(r"[A-Z][A-Z0-9_]{0,63}")
MIN_MASKED = 6
HOLD = 1 << 16
CHUNK = 1 << 16
DRAIN = 2.0
USAGE = """usage:
secret ls names and descriptions
secret run NAME [NAME...] -- CMD run CMD with each secret in $NAME
A value secret arrives as its text; a file secret as the path to a
private copy that is removed when CMD exits. Values are masked in
CMD's output as <secret:NAME>.
"""
class SecretError(Exception):
pass
def request(payload: dict[str, Any], path: Path = SOCKET) -> dict[str, Any]:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
try:
conn.connect(str(path))
except OSError as exc:
msg = f"the gateway's secrets socket is not reachable ({exc.strerror})"
raise SecretError(msg) from exc
conn.sendall(json.dumps(payload).encode() + b"\n")
chunks = []
while chunk := conn.recv(CHUNK):
chunks.append(chunk)
reply = json.loads(b"".join(chunks) or b"{}")
if not reply.get("ok"):
raise SecretError(reply.get("error") or "the gateway gave no answer")
return reply
def root_half(argv: list[str]) -> int:
if argv[:1] == ["ls"] and len(argv) == 1:
payload: dict[str, Any] = {"op": "ls"}
elif (
argv[:1] == ["get"]
and len(argv) > 1
and all(NAME.fullmatch(a) for a in argv[1:])
):
payload = {"op": "get", "names": argv[1:]}
else:
print("beaver-secret: ls | get NAME...", file=sys.stderr)
return 2
try:
reply = request(payload)
except SecretError as exc:
reply = {"ok": False, "error": str(exc)}
sys.stdout.write(json.dumps(reply))
return 0 if reply.get("ok") else 1
def ask(argv: list[str]) -> dict[str, Any]:
if os.geteuid() == 0:
return request(
{"op": "ls"} if argv == ["ls"] else {"op": "get", "names": argv[1:]}
)
proc = subprocess.run( # noqa: S603
[*ROOT_HALF, *argv], capture_output=True, check=False
)
try:
reply = json.loads(proc.stdout)
except json.JSONDecodeError:
err = proc.stderr.decode(errors="replace").strip()
msg = f"the root helper failed: {err or f'exit {proc.returncode}'}"
raise SecretError(msg) from None
if not reply.get("ok"):
raise SecretError(reply.get("error") or "the root helper gave no answer")
return reply
def list_secrets() -> int:
rows = ask(["ls"])["secrets"]
if not rows:
print("no secrets yet; a human adds them in the admin, System -> Secrets")
return 0
width = max(len(r["name"]) for r in rows)
for r in rows:
kind = f"file {r['filename']}" if r.get("filename") else "value"
line = f"{r['name']:<{width}} {kind}"
if r.get("description"):
line += f" - {r['description']}"
print(line)
return 0
class Masker:
"""Streams bytes through, replacing every needle; never splits one."""
def __init__(self, needles: dict[bytes, bytes]) -> None:
self._needles = needles
self._longest = max((len(n) for n in needles), default=0)
ordered = sorted(needles, key=lambda n: -len(n))
self._pattern = (
re.compile(b"|".join(re.escape(n) for n in ordered)) if ordered else None
)
self._pending = b""
def feed(self, data: bytes) -> bytes:
if self._pattern is None:
return data
self._pending += data
cut = self._pending.rfind(b"\n") + 1
if len(self._pending) - cut > HOLD:
cut = max(cut, len(self._pending) - self._longest + 1)
for match in self._pattern.finditer(self._pending):
if match.start() < cut < match.end():
cut = match.end()
head, self._pending = self._pending[:cut], self._pending[cut:]
return self._mask(head)
def flush(self) -> bytes:
tail, self._pending = self._pending, b""
return self._mask(tail)
def _mask(self, data: bytes) -> bytes:
if self._pattern is None or not data:
return data
return self._pattern.sub(lambda m: self._needles[m[0]], data)
def needles_for(values: dict[str, bytes]) -> dict[bytes, bytes]:
out: dict[bytes, bytes] = {}
for name, data in values.items():
mask = f"<secret:{name}>".encode()
try:
text = data.decode()
except UnicodeDecodeError:
continue
for part in text.splitlines() or [text]:
for form in {part.strip(), quote(part.strip(), safe="")}:
if len(form) >= MIN_MASKED:
out[form.encode()] = mask
return out
def sweep(root: Path) -> None:
for stale in root.glob(f"{PREFIX}*"):
pid = stale.name.removeprefix(PREFIX).split("-", 1)[0]
if not pid.isdigit() or _alive(int(pid)):
continue
shutil.rmtree(stale, ignore_errors=True)
def _alive(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def _pump(src: BufferedReader, dst: IO[bytes], masker: Masker) -> None:
while chunk := src.read1(CHUNK):
dst.write(masker.feed(chunk))
dst.flush()
dst.write(masker.flush())
dst.flush()
def run(names: list[str], command: list[str], *, root: Path | None = None) -> int:
secrets = ask(["get", *names])["secrets"]
base = root or (SHM if SHM.is_dir() else Path(tempfile.gettempdir()))
sweep(base)
workdir = Path(tempfile.mkdtemp(prefix=f"{PREFIX}{os.getpid()}-", dir=base))
env = dict(os.environ)
values: dict[str, bytes] = {}
try:
for name in names:
item = secrets[name]
data = base64.b64decode(item["data"])
if item.get("filename"):
path = workdir / name / item["filename"]
path.parent.mkdir(mode=0o700)
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "wb") as fh:
fh.write(data)
env[name] = str(path)
else:
env[name] = data.decode(errors="surrogateescape")
values[name] = data
return _spawn(command, env, needles_for(values))
finally:
shutil.rmtree(workdir, ignore_errors=True)
def _spawn(command: list[str], env: dict[str, str], needles: dict[bytes, bytes]) -> int:
try:
child = subprocess.Popen( # noqa: S603
command, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
except (OSError, ValueError) as exc:
reason = exc.strerror if isinstance(exc, OSError) else str(exc)
print(f"secret: cannot run {command[0]}: {reason}", file=sys.stderr)
return 127
forwarded = (
(signal.SIGINT, signal.SIGTERM, signal.SIGHUP)
if threading.current_thread() is threading.main_thread()
else ()
)
previous = {
sig: signal.signal(sig, lambda s, _f: child.send_signal(s)) for sig in forwarded
}
pumps = [
threading.Thread(target=_pump, args=(src, dst, Masker(needles)), daemon=True)
for src, dst in (
(child.stdout, sys.stdout.buffer),
(child.stderr, sys.stderr.buffer),
)
]
for pump in pumps:
pump.start()
try:
code = child.wait()
for pump in pumps:
pump.join(DRAIN)
finally:
for sig, handler in previous.items():
signal.signal(sig, handler)
return 128 - code if code < 0 else code
def main(argv: list[str] | None = None) -> int:
args = sys.argv[1:] if argv is None else argv
if args[:1] == ["--root"]:
return root_half(args[1:])
try:
if args == ["ls"]:
return list_secrets()
if args[:1] == ["run"] and "--" in args:
split = args.index("--")
names, command = args[1:split], args[split + 1 :]
bad = [n for n in names if not NAME.fullmatch(n)]
if names and command and not bad:
return run(names, command)
if bad:
print(f"secret: invalid name {bad[0]!r}", file=sys.stderr)
return 2
except SecretError as exc:
print(f"secret: {exc}", file=sys.stderr)
return 1
print(USAGE, end="", file=sys.stderr)
return 2
if __name__ == "__main__":
with contextlib.suppress(KeyboardInterrupt):
sys.exit(main())
+108
View File
@@ -0,0 +1,108 @@
"""Root-only unix socket that hands sealed secrets to the ``secret`` helper."""
from __future__ import annotations
import asyncio
import base64
import contextlib
import json
import logging
from typing import TYPE_CHECKING, Any
from beaver_gateway.keystore.store import (
SecretsDisabledError,
UnknownSecretError,
check_name,
)
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable, Sequence
from pathlib import Path
from beaver_gateway.keystore.store import SecretStore
_log = logging.getLogger("beaver_gateway.keystore")
REQUEST_LIMIT = 1 << 16
async def serve(
store: SecretStore,
path: Path,
*,
on_use: Callable[[Sequence[str]], Awaitable[None]] | None = None,
) -> None:
"""Listen on ``path`` (mode 0600 in a 0700 directory) until cancelled."""
_clear(path)
async def handle(
reader: asyncio.StreamReader, writer: asyncio.StreamWriter
) -> None:
try:
line = await reader.readline()
reply = await answer(store, line, on_use=on_use)
writer.write(json.dumps(reply).encode() + b"\n")
await writer.drain()
except Exception: # noqa: BLE001
_log.exception("secret request failed")
finally:
writer.close()
with contextlib.suppress(Exception):
await writer.wait_closed()
server = await asyncio.start_unix_server(
handle, path=str(path), limit=REQUEST_LIMIT
)
_lock(path)
_log.info("secrets socket on %s", path)
async with server:
await server.serve_forever()
async def answer(
store: SecretStore,
line: bytes,
*,
on_use: Callable[[Sequence[str]], Awaitable[None]] | None = None,
) -> dict[str, Any]:
try:
request = json.loads(line)
op = request.get("op") if isinstance(request, dict) else None
if op == "ls":
return {"ok": True, "secrets": [_listed(r) for r in await store.rows()]}
if op == "get":
names = [check_name(str(n)) for n in request.get("names") or ()]
if not names:
return {"ok": False, "error": "no secret names given"}
revealed = await store.reveal(names)
if on_use is not None:
await on_use(names)
return {
"ok": True,
"secrets": {
r.name: {
"data": base64.b64encode(r.data).decode(),
"filename": r.filename,
}
for r in revealed
},
}
except UnknownSecretError as exc:
return {"ok": False, "error": f"unknown secret: {exc}"}
except (SecretsDisabledError, ValueError) as exc:
return {"ok": False, "error": str(exc)}
return {"ok": False, "error": "op must be `ls` or `get`"}
def _clear(path: Path) -> None:
path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
path.parent.chmod(0o700)
path.unlink(missing_ok=True)
def _lock(path: Path) -> None:
path.chmod(0o600)
def _listed(row: Any) -> dict[str, Any]:
return {"name": row.name, "description": row.description, "filename": row.filename}
+193
View File
@@ -0,0 +1,193 @@
"""Seal, list, replace and reveal secrets; the key never leaves this module."""
from __future__ import annotations
import os
import re
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from sqlmodel import col, select
from beaver_gateway.storage.models import Secret
if TYPE_CHECKING:
from collections.abc import Iterable, Sequence
from beaver_gateway.storage import Database
NAME = re.compile(r"[A-Z][A-Z0-9_]{0,63}")
"""A secret's name is the environment variable a command receives it under."""
KEY_BYTES = 32
NONCE_BYTES = 12
MAX_SIZE = 1 << 20
"""Largest value or file the store accepts, in bytes."""
class SecretsDisabledError(RuntimeError):
"""``SECRETS_KEY`` is unset or unusable, so nothing can be sealed or opened."""
class UnknownSecretError(LookupError):
"""A name that is not in the store."""
class SecretBox:
"""AES-256-GCM with the secret's name as associated data."""
__slots__ = ("_aead",)
def __init__(self, key: bytes) -> None:
if len(key) != KEY_BYTES:
msg = f"SECRETS_KEY must be {KEY_BYTES} bytes, got {len(key)}"
raise ValueError(msg)
self._aead = AESGCM(key)
@classmethod
def from_hex(cls, text: str) -> SecretBox:
try:
key = bytes.fromhex(text.strip())
except ValueError as exc:
msg = "SECRETS_KEY must be hex, e.g. `openssl rand -hex 32`"
raise ValueError(msg) from exc
return cls(key)
def seal(self, name: str, data: bytes) -> bytes:
nonce = os.urandom(NONCE_BYTES)
return nonce + self._aead.encrypt(nonce, data, name.encode())
def open(self, name: str, sealed: bytes) -> bytes:
nonce, body = sealed[:NONCE_BYTES], sealed[NONCE_BYTES:]
return self._aead.decrypt(nonce, body, name.encode())
@dataclass(frozen=True, slots=True)
class Revealed:
name: str
data: bytes
filename: str | None
def _now() -> datetime:
return datetime.now(UTC)
def check_name(name: str) -> str:
if not NAME.fullmatch(name):
msg = (
f"invalid secret name {name!r}: use A-Z, 0-9 and _, starting with a letter"
)
raise ValueError(msg)
return name
def clean_filename(filename: str) -> str:
base = os.path.basename(filename.replace("\\", "/")).strip() # noqa: PTH119
if base in ("", ".", ".."):
msg = f"invalid file name {filename!r}"
raise ValueError(msg)
return base
class SecretStore:
"""Rows of :class:`Secret`; ``box`` is ``None`` when no key is configured."""
__slots__ = ("_box", "_db")
def __init__(self, db: Database, box: SecretBox | None) -> None:
self._db = db
self._box = box
@property
def enabled(self) -> bool:
return self._box is not None
def _require_box(self) -> SecretBox:
if self._box is None:
msg = "SECRETS_KEY is not set"
raise SecretsDisabledError(msg)
return self._box
async def rows(self) -> Sequence[Secret]:
async with self._db.session() as session:
result = await session.exec(select(Secret).order_by(col(Secret.name)))
return result.all()
async def put(
self,
name: str,
*,
description: str | None = None,
data: bytes | None = None,
filename: str | None = None,
) -> Secret:
"""Create or replace; ``data=None`` keeps the value, sets the description."""
check_name(name)
if data is not None:
if len(data) > MAX_SIZE:
msg = f"secret {name} is {len(data)} bytes, the limit is {MAX_SIZE}"
raise ValueError(msg)
sealed = self._require_box().seal(name, data)
async with self._db.session() as session:
row = await session.get(Secret, name)
if row is None:
if data is None:
msg = f"secret {name} needs a value or a file"
raise ValueError(msg)
row = Secret(name=name, sealed=sealed)
if data is not None:
row.sealed = sealed
row.size = len(data)
row.filename = clean_filename(filename) if filename else None
if description is not None:
row.description = description.strip()
row.updated_at = _now()
session.add(row)
await session.commit()
await session.refresh(row)
return row
async def delete(self, name: str) -> bool:
async with self._db.session() as session:
row = await session.get(Secret, name)
if row is None:
return False
await session.delete(row)
await session.commit()
return True
async def reveal(self, names: Iterable[str]) -> list[Revealed]:
box = self._require_box()
wanted = list(dict.fromkeys(names))
async with self._db.session() as session:
rows = {
r.name: r
for r in (
await session.exec(
select(Secret).where(col(Secret.name).in_(wanted))
)
).all()
}
missing = [n for n in wanted if n not in rows]
if missing:
raise UnknownSecretError(", ".join(missing))
out = [_open(box, rows[n]) for n in wanted]
now = _now()
for row in rows.values():
row.used_at = now
session.add(row)
await session.commit()
return out
def _open(box: SecretBox, row: Secret) -> Revealed:
try:
data = box.open(row.name, row.sealed)
except InvalidTag as exc:
msg = f"secret {row.name} does not open: SECRETS_KEY changed since it was set"
raise SecretsDisabledError(msg) from exc
return Revealed(name=row.name, data=data, filename=row.filename)
+3
View File
@@ -28,6 +28,9 @@ KNOWN_KINDS: frozenset[str] = frozenset(
"logout", "logout",
"token_create", "token_create",
"token_revoke", "token_revoke",
"secret_set",
"secret_delete",
"secret_use",
} }
) )
"""Kinds the gateway currently emits; ``AuditLog.kind`` stays free-form.""" """Kinds the gateway currently emits; ``AuditLog.kind`` stays free-form."""
+7
View File
@@ -61,6 +61,13 @@ class Settings(BaseSettings):
"""Directories the ``send_file`` tool may read, separated by ``:``. A """Directories the ``send_file`` tool may read, separated by ``:``. A
frontend always adds the inbox its own attachments land in.""" frontend always adds the inbox its own attachments land in."""
secrets_key: str | None = None
"""Hex AES-256 key (``openssl rand -hex 32``) sealing the secrets table;
unset leaves secrets off. Never reaches the model's environment."""
secrets_socket: Path = Path("/run/beaver-secrets/gateway.sock")
"""Root-only socket the ``secret`` helper reads values from."""
bootstrap_tokens: str = "" bootstrap_tokens: str = ""
"""Out-of-band token seed: ``name1:value1,name2:value2``. Layers """Out-of-band token seed: ``name1:value1,name2:value2``. Layers
alongside DB-issued tokens; used for first-run setup and recovery.""" alongside DB-issued tokens; used for first-run setup and recovery."""
+21 -1
View File
@@ -10,7 +10,7 @@ from __future__ import annotations
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import Any from typing import Any
from sqlalchemy import JSON, Column, Index, UniqueConstraint, text from sqlalchemy import JSON, Column, Index, LargeBinary, UniqueConstraint, text
from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import Field, SQLModel from sqlmodel import Field, SQLModel
@@ -374,6 +374,25 @@ class JobRunRecord(SQLModel, table=True):
) )
class Secret(SQLModel, table=True):
"""A value or file the model may run a command with, sealed by ``SECRETS_KEY``.
``sealed`` is nonce + AES-GCM ciphertext bound to ``name``; nothing
here decrypts without the key, so a dump of the table stays closed.
"""
__tablename__ = "secrets"
name: str = Field(primary_key=True)
description: str = Field(default="")
filename: str | None = Field(default=None)
size: int = Field(default=0)
sealed: bytes = Field(sa_column=Column(LargeBinary, nullable=False))
created_at: datetime = Field(default_factory=_utcnow)
updated_at: datetime = Field(default_factory=_utcnow)
used_at: datetime | None = Field(default=None)
__all__ = [ __all__ = [
"AuditLog", "AuditLog",
"Conversation", "Conversation",
@@ -383,6 +402,7 @@ __all__ = [
"InjectQueueItem", "InjectQueueItem",
"JobRunRecord", "JobRunRecord",
"RateLimit", "RateLimit",
"Secret",
"TelegramUpdate", "TelegramUpdate",
"Token", "Token",
"TranscriptEntry", "TranscriptEntry",
+68 -5
View File
@@ -13,21 +13,26 @@ import difflib
import fnmatch import fnmatch
import logging import logging
import os import os
import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import UTC, datetime from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
from watchfiles import awatch from watchfiles import DefaultFilter, awatch
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Iterable from collections.abc import Iterable
__all__ = ["Change", "VaultWatch", "WatchRules"] from watchfiles import Change as FileChange
__all__ = ["Change", "VaultFilter", "VaultWatch", "WatchRules"]
_log = logging.getLogger("beaver_gateway.vault.watch") _log = logging.getLogger("beaver_gateway.vault.watch")
_FENCE = re.compile(r"^\s*(`{3,}|~{3,})\s*(.*?)\s*$")
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class WatchRules: class WatchRules:
@@ -42,6 +47,8 @@ class WatchRules:
names: tuple[str, ...] = () names: tuple[str, ...] = ()
ignore: tuple[str, ...] = () ignore: tuple[str, ...] = ()
suffixes: tuple[str, ...] = (".md",) suffixes: tuple[str, ...] = (".md",)
skip_fences: tuple[str, ...] = ("tasks", "dataview", "dataviewjs")
"""Fenced blocks of these languages are queries: ``full`` files drop them."""
def kind(self, rel: str, *, today: str) -> str | None: def kind(self, rel: str, *, today: str) -> str | None:
"""The most specific matching pattern wins, whichever list it is in.""" """The most specific matching pattern wins, whichever list it is in."""
@@ -74,6 +81,26 @@ class _Seen:
count: int count: int
IGNORE_DIRS = (".obsidian", ".trash", ".git")
"""Housekeeping directories whose churn never means an edit of the vault."""
IGNORE_NAMES = (".sync.lock", "*.lock", "*.tmp", "*~")
"""Lock and scratch file names, matched against the entity name."""
class VaultFilter(DefaultFilter):
"""Drops housekeeping paths inside ``awatch``, before the loop is woken."""
def __init__(self) -> None:
super().__init__(ignore_dirs=(*DefaultFilter.ignore_dirs, *IGNORE_DIRS))
def __call__(self, change: FileChange, path: str) -> bool:
name = Path(path).name
if any(fnmatch.fnmatchcase(name, pattern) for pattern in IGNORE_NAMES):
return False
return super().__call__(change, path)
class VaultWatch: class VaultWatch:
def __init__( def __init__(
self, root: Path, rules: WatchRules, *, tz: str = "UTC", debounce: float = 4.0 self, root: Path, rules: WatchRules, *, tz: str = "UTC", debounce: float = 4.0
@@ -98,6 +125,7 @@ class VaultWatch:
_log.info("watching %s (%d files in snapshot)", self.root, len(self._seen)) _log.info("watching %s (%d files in snapshot)", self.root, len(self._seen))
async for changes in awatch( async for changes in awatch(
self.root, self.root,
watch_filter=VaultFilter(),
debounce=int(self._debounce * 1000), debounce=int(self._debounce * 1000),
stop_event=self._stop, stop_event=self._stop,
ignore_permission_denied=True, ignore_permission_denied=True,
@@ -117,7 +145,7 @@ class VaultWatch:
if kind is None: if kind is None:
continue continue
with contextlib.suppress(OSError): with contextlib.suppress(OSError):
lines = _read_lines(path) lines = self._lines(path, full=kind == "full")
self._seen[rel] = _Seen( self._seen[rel] = _Seen(
lines=lines if kind == "full" else None, count=len(lines) lines=lines if kind == "full" else None, count=len(lines)
) )
@@ -135,7 +163,7 @@ class VaultWatch:
self._pending.pop(rel, None) self._pending.pop(rel, None)
return None return None
stat = path.stat() stat = path.stat()
lines = _read_lines(path) lines = self._lines(path, full=kind == "full")
seen = self._seen.get(rel) seen = self._seen.get(rel)
change = Change( change = Change(
path=rel, path=rel,
@@ -163,12 +191,16 @@ class VaultWatch:
for change in changes: for change in changes:
path = self.root / change.path path = self.root / change.path
with contextlib.suppress(OSError): with contextlib.suppress(OSError):
lines = _read_lines(path) lines = self._lines(path, full=change.full)
self._seen[change.path] = _Seen( self._seen[change.path] = _Seen(
lines=lines if change.full else None, count=len(lines) lines=lines if change.full else None, count=len(lines)
) )
return changes return changes
def _lines(self, path: Path, *, full: bool) -> list[str]:
lines = _read_lines(path)
return _prose(lines, self.rules.skip_fences) if full else lines
def _rel(self, path: Path) -> str: def _rel(self, path: Path) -> str:
return path.resolve().relative_to(self.root.resolve()).as_posix() return path.resolve().relative_to(self.root.resolve()).as_posix()
@@ -193,6 +225,37 @@ def _read_lines(path: Path) -> list[str]:
return path.read_text(encoding="utf-8", errors="replace").splitlines() return path.read_text(encoding="utf-8", errors="replace").splitlines()
def _prose(lines: list[str], skip_fences: tuple[str, ...]) -> list[str]:
start = 0
if lines and lines[0].strip() == "---":
closing = next(
(i for i, line in enumerate(lines[1:], 1) if line.strip() == "---"), None
)
start = closing + 1 if closing is not None else 0
out: list[str] = []
fence: str | None = None
skipping = False
for line in lines[start:]:
match = _FENCE.match(line)
if fence is None and match:
fence = match.group(1)
language = match.group(2).split(maxsplit=1)[0] if match.group(2) else ""
skipping = language.lower() in skip_fences
elif fence is not None and match and _closes(match, fence):
fence = None
if skipping:
skipping = False
continue
if not skipping and line.strip():
out.append(line)
return out
def _closes(match: re.Match[str], fence: str) -> bool:
mark = match.group(1)
return mark[0] == fence[0] and len(mark) >= len(fence) and not match.group(2)
def _added(old: list[str], new: list[str]) -> list[str]: def _added(old: list[str], new: list[str]) -> list[str]:
matcher = difflib.SequenceMatcher(a=old, b=new, autojunk=False) matcher = difflib.SequenceMatcher(a=old, b=new, autojunk=False)
added: list[str] = [] added: list[str] = []
+95 -1
View File
@@ -16,7 +16,7 @@ from claude_agent_sdk import (
UserMessage, UserMessage,
) )
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
from test_conversations import ScriptedClient, World from test_conversations import ScriptedClient, World, _three_turns
from beaver_gateway.frontends.markdown.history import rewrite_messages from beaver_gateway.frontends.markdown.history import rewrite_messages
from beaver_gateway.security.auth import TokenStore from beaver_gateway.security.auth import TokenStore
@@ -396,6 +396,100 @@ async def test_describe_snapshots_open_tools_and_records_rate_limit(
assert page["offset"] == 0 and len(page["entries"]) == 1 assert page["offset"] == 0 and len(page["entries"]) == 1
async def test_sessions_show_their_processes_and_take_orders(world: World) -> None:
world.backend._factory = ToolClient # noqa: SLF001
api = Api(world)
api.store.grant("api-only", "weak", scope="api")
weak = {"Authorization": "Bearer weak"}
conv = await world.conversations.create(kind="master", agent="a", origin="test")
key = conv.external_id
ScriptedClient.hold = asyncio.Event()
await world.conversations.post(conv, "go")
await asyncio.sleep(0.3)
census = await api.get("/sessions")
(row,) = census["sessions"]
assert row["key"] == key and row["busy"]
assert row["processes"] == [] and row["cpu"] is None
assert isinstance(census["others"], list)
live = world.pool.get(key)
assert live is not None
live.stderr.append("warning: something")
stderr = await api.get(f"/sessions/{key}/stderr")
assert stderr["lines"] == ["warning: something"]
for path in (f"/sessions/{key}/close", "/processes/1/kill"):
assert (await api.http.post(path, headers=weak)).status_code == 403
assert (
await api.http.get(f"/sessions/{key}/stderr", headers=weak)
).status_code == 403
closed = await api.http.post(f"/sessions/{key}/close", headers=HEADERS)
assert closed.json() == {"key": key, "closed": True}
assert world.pool.get(key) is None
deadline = asyncio.get_running_loop().time() + 5
while ("user", "failed") not in await world.statuses(conv):
assert asyncio.get_running_loop().time() < deadline, "the cut turn hangs"
await asyncio.sleep(0.02)
again = await api.http.post(f"/sessions/{key}/close", headers=HEADERS)
assert again.status_code == 404
gone = await api.http.post("/processes/999999999/kill", headers=HEADERS)
assert gone.status_code == 404
ScriptedClient.hold = asyncio.Event()
await world.conversations.post(conv, "once more")
await asyncio.sleep(0.3)
interrupted = await api.http.post(f"/sessions/{key}/interrupt", headers=weak)
assert interrupted.json() == {"key": key, "interrupted": True}
assert ScriptedClient.instances[-1].interrupted
kinds = [r["kind"] for r in (await api.get("/audit"))["records"]]
assert kinds == ["session_interrupt", "session_close"]
async def test_rewind_lists_turns_and_needs_admin_scope(world: World) -> None:
api = Api(world)
api.store.grant("api-only", "weak", scope="api")
weak = {"Authorization": "Bearer weak"}
sid, entries = await _three_turns(world)
conv = await world.conversations.create(
kind="master", agent="a", origin="test", session_id=sid
)
path = f"/conversations/{conv.external_id}"
listed = (await api.http.get(f"{path}/turns", headers=weak)).json()
assert [t["text"] for t in listed["turns"]] == ["two", "three"]
assert (await api.get(f"{path}/turns", {"limit": 1}))["turns"] == [
listed["turns"][1]
]
body = {"turn": listed["turns"][0]["id"]}
denied = await api.http.post(f"{path}/rewind", json=body, headers=weak)
assert denied.status_code == 403
missing = await api.http.post(
f"{path}/rewind", json={"turn": entries[0]["uuid"]}, headers=HEADERS
)
assert missing.status_code == 404
done = await api.http.post(f"{path}/rewind", json=body, headers=HEADERS)
assert done.status_code == 200, done.text
out = done.json()
assert out["turns"] == 2 and out["previous_session"] == sid
assert (await api.get(path))["session_id"] == out["session_id"] != sid
assert (await api.get(f"{path}/turns"))["turns"] == []
messages = (await api.get(f"{path}/history"))["messages"]
assert [m["content"] for m in messages] == [
"one",
[{"type": "text", "text": "done one"}],
]
(record,) = (await api.get("/audit"))["records"]
assert record["kind"] == "api_rewind"
assert record["detail"]["previous_session"] == sid
ScriptedClient.hold = asyncio.Event()
await world.conversations.post(conv, "busy")
await asyncio.sleep(0.2)
busy = await api.http.post(f"{path}/rewind", json=body, headers=HEADERS)
assert busy.status_code == 409
ScriptedClient.hold.set()
async def test_tokens_and_audit_need_admin_scope(world: World) -> None: async def test_tokens_and_audit_need_admin_scope(world: World) -> None:
api = Api(world) api = Api(world)
api.store.grant("api-only", "weak", scope="api") api.store.grant("api-only", "weak", scope="api")
+5 -1
View File
@@ -250,7 +250,11 @@ async def test_stream_envelope_and_index_rebase(cwd: Path) -> None:
assert opts.strict_mcp_config is True assert opts.strict_mcp_config is True
assert opts.permission_mode == "bypassPermissions" assert opts.permission_mode == "bypassPermissions"
assert opts.mcp_servers == { assert opts.mcp_servers == {
"firefly": {"type": "http", "url": "http://127.0.0.1:1/mcp/firefly/"} "firefly": {
"type": "http",
"url": "http://127.0.0.1:1/mcp/firefly/",
"headers": {"X-Conversation-Origin": "conv-1"},
}
} }
assert opts.session_store is not None assert opts.session_store is not None
assert opts.resume is None assert opts.resume is None
+214 -1
View File
@@ -20,7 +20,11 @@ from claude_agent_sdk import (
project_key_for_directory, project_key_for_directory,
) )
from anthropic.types import TextBlock as AnthropicTextBlock
from anthropic.types import ThinkingBlock
from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions
from beaver_gateway.agents.raycast import RaycastAgent
from beaver_gateway.backends.claude_sdk import ClaudeSdkBackend from beaver_gateway.backends.claude_sdk import ClaudeSdkBackend
from beaver_gateway.events.bus import EventBus from beaver_gateway.events.bus import EventBus
from beaver_gateway.conversations.service import ( from beaver_gateway.conversations.service import (
@@ -36,6 +40,7 @@ from beaver_gateway.backends.transcript import (
close_open_tool_uses, close_open_tool_uses,
open_tool_uses, open_tool_uses,
render_messages, render_messages,
reply_text,
strip_tool_entries, strip_tool_entries,
window_entries, window_entries,
) )
@@ -499,6 +504,105 @@ async def test_copy_seed_forks_parent_with_window(world: World) -> None:
assert (await world.conversations.get(child.external_id)).flags["seed"] is None assert (await world.conversations.get(child.external_id)).flags["seed"] is None
async def _three_turns(world: World) -> tuple[str, list[dict[str, Any]]]:
sid = str(uuid.uuid4())
history = [
{"role": "user", "content": "one"},
{"role": "assistant", "content": "done one"},
{"role": "user", "content": "two"},
{
"role": "assistant",
"content": [{"type": "tool_use", "id": "t1", "name": "Read", "input": {}}],
},
{
"role": "user",
"content": [{"type": "tool_result", "tool_use_id": "t1", "content": "x"}],
},
{"role": "assistant", "content": "done two"},
{"role": "user", "content": "three"},
{"role": "assistant", "content": "refused"},
]
entries = build_entries(history, session_id=sid, cwd=str(world.root), model="m")
entries[-1]["isApiErrorMessage"] = True
note = {
**entries[2],
"uuid": str(uuid.uuid4()),
"message": {
"role": "user",
"content": [{"type": "text", "text": "[Request interrupted by user]"}],
},
}
del note["promptSource"]
entries.insert(6, note)
await world.store.append(world.key(sid), entries)
return sid, entries
async def test_rewind_moves_the_thread_to_a_fork_that_ends_before_the_turn(
world: World,
) -> None:
sid, entries = await _three_turns(world)
conv = await world.conversations.create(
kind="master", agent="a", origin="test", session_id=sid
)
await world.conversations.post(conv, "live")
await world.settle(conv, 1)
assert world.pool.get(conv.external_id) is not None
turns = await world.conversations.turns(conv)
assert [(t["text"], t["reply"], t["failed"]) for t in turns] == [
("two", "done two", False),
("three", "refused", True),
]
assert [t["id"] for t in turns] == [entries[2]["uuid"], entries[7]["uuid"]]
result = await world.conversations.rewind(conv, turns[1]["id"])
assert result.turns == 1 and result.previous_session == sid
moved = result.conversation.session_id
assert moved != sid
assert (await world.conversations.get(conv.external_id)).session_id == moved
assert await world.store.load(world.key(sid)) == entries
kept = [
e["message"]["content"]
for e in await world.store.load(world.key(moved))
if e["type"] in ("user", "assistant")
]
assert len(kept) == 7 and kept[0] == "one"
assert kept[-1] == [{"type": "text", "text": "[Request interrupted by user]"}]
assert world.pool.get(conv.external_id) is None
await world.conversations.run_text_turn(conv, "after", origin="user")
assert conv.session_id == sid
assert ScriptedClient.instances[-1].options.resume == moved
(two,) = await world.conversations.turns(result.conversation)
assert two["text"] == "two"
again = await world.conversations.rewind(conv, two["id"])
assert again.turns == 1 and again.previous_session == moved
assert await world.conversations.turns(again.conversation) == []
first = (await world.store.load(world.key(again.conversation.session_id)))[0]
with pytest.raises(LookupError):
await world.conversations.rewind(conv, first["uuid"])
with pytest.raises(LookupError):
await world.conversations.rewind(conv, str(uuid.uuid4()))
async def test_rewind_waits_for_no_turn(world: World) -> None:
sid, entries = await _three_turns(world)
conv = await world.conversations.create(
kind="master", agent="a", origin="test", session_id=sid
)
ScriptedClient.hold = asyncio.Event()
await world.conversations.post(conv, "busy")
await asyncio.sleep(0.2)
with pytest.raises(RuntimeError, match="mid-turn"):
await world.conversations.rewind(conv, entries[7]["uuid"])
ScriptedClient.hold.set()
await world.settle(conv, 1)
assert (await world.conversations.get(conv.external_id)).session_id == sid
async def test_merge_injects_summary_into_parent(world: World) -> None: async def test_merge_injects_summary_into_parent(world: World) -> None:
sid = str(uuid.uuid4()) sid = str(uuid.uuid4())
await world.store.append( await world.store.append(
@@ -714,12 +818,86 @@ def test_transcript_helpers() -> None:
assert windowed[0]["message"]["content"] == "q2" and len(windowed) == 2 assert windowed[0]["message"]["content"] == "q2" and len(windowed) == 2
assert ( assert (
render_messages( render_messages(
[{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}] [{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}],
thinking_is_progress=True,
) )
== "user:\nq\n\nassistant:\na" == "user:\nq\n\nassistant:\na"
) )
def test_read_conversation_keeps_progress_and_drops_reasoning() -> None:
history = [
{"role": "user", "content": "status?"},
{
"role": "assistant",
"content": [
{"type": "text", "text": "Checking."},
{"type": "thinking", "thinking": "", "signature": "s"},
{"type": "tool_use", "id": "t1", "name": "komodo", "input": {}},
],
},
{
"role": "user",
"content": [{"type": "tool_result", "tool_use_id": "t1", "content": "ok"}],
},
{
"role": "assistant",
"content": [
{
"type": "thinking",
"thinking": "All up, reading logs.",
"signature": "s",
},
{"type": "tool_use", "id": "t2", "name": "komodo", "input": {}},
],
},
{
"role": "user",
"content": [{"type": "tool_result", "tool_use_id": "t2", "content": "ok"}],
},
{"role": "assistant", "content": [{"type": "text", "text": "Healthy."}]},
]
assert render_messages(history, thinking_is_progress=True) == (
"user:\nstatus?\n\nassistant:\nChecking.\n\nAll up, reading logs.\n\n"
"Healthy.\n(tools: komodo, komodo)"
)
assert render_messages(history, thinking_is_progress=False) == (
"user:\nstatus?\n\nassistant:\nChecking.\n\nHealthy.\n(tools: komodo, komodo)"
)
@pytest.mark.parametrize(
("model", "expected"),
[
("claude-opus-5-5", True),
("claude-fable-5-1", True),
("claude-opus-5", False),
("claude-sonnet-5", False),
("claude-haiku-4-5-20251001", False),
("claude-next-unknown", False),
],
)
def test_only_progress_models_read_their_thinking_as_progress(
tmp_path: Path, model: str, expected: bool
) -> None:
assert (
ClaudeAgent(name="c", model=model, cwd=tmp_path).thinking_is_progress
is expected
)
def test_summarized_display_and_raycast_are_never_progress(tmp_path: Path) -> None:
raycast = RaycastAgent(name="r", model="google-gemini-3.1-pro", system_prompt="")
summarized = ClaudeAgent(
name="s",
model="claude-opus-5-5",
cwd=tmp_path,
options=ClaudeOptions(thinking={"type": "adaptive", "display": "summarized"}),
)
assert not raycast.thinking_is_progress
assert not summarized.thinking_is_progress
async def test_agent_kinds_gate_create(world: World) -> None: async def test_agent_kinds_gate_create(world: World) -> None:
with pytest.raises(ValueError, match="does not serve kind 'deep'"): with pytest.raises(ValueError, match="does not serve kind 'deep'"):
await world.conversations.create(kind="deep", agent="a", origin="test") await world.conversations.create(kind="deep", agent="a", origin="test")
@@ -958,3 +1136,38 @@ async def test_subagent_result_inside_a_turn_does_not_end_it(world: World) -> No
assert client.prompts[-1].startswith("two") assert client.prompts[-1].startswith("two")
rows = await world.conversations.queue.recent(conv.id) rows = await world.conversations.queue.recent(conv.id)
assert [r.status for r in rows] == ["done", "done"] assert [r.status for r in rows] == ["done", "done"]
def test_reply_text_keeps_progress_updates_and_drops_hidden_reasoning() -> None:
blocks = [
ThinkingBlock(type="thinking", thinking="", signature="s"),
ThinkingBlock(type="thinking", thinking="looking it up", signature="s"),
AnthropicTextBlock(type="text", text="found it"),
]
assert reply_text(blocks, thinking_is_progress=True) == "looking it up\n\nfound it"
assert reply_text(blocks, thinking_is_progress=False) == "found it"
def test_reply_text_on_a_plain_answer_is_unchanged() -> None:
blocks = [AnthropicTextBlock(type="text", text=" just an answer ")]
assert reply_text(blocks, thinking_is_progress=True) == "just an answer"
async def test_a_conversation_stream_carries_updates_about_that_conversation() -> None:
bus = EventBus()
seen: list[tuple[str, str | None]] = []
async def collect() -> None:
async for event in bus.stream(conversation_id="c1"):
seen.append((event["type"], event.get("id")))
if len(seen) == 2:
return
task = asyncio.create_task(collect())
await asyncio.sleep(0)
bus.publish("conversation.updated", id="c2", running_turn=None)
bus.publish("conversation.updated", id="c1", running_turn=None)
bus.publish("stream", conversation_id="c2")
bus.publish("stream", conversation_id="c1")
await asyncio.wait_for(task, 1)
assert seen == [("conversation.updated", "c1"), ("stream", None)]
+54 -1
View File
@@ -7,7 +7,11 @@ from test_conversations import ScriptedClient, World, world
from beaver_gateway.conversations.service import UserSaid from beaver_gateway.conversations.service import UserSaid
from beaver_gateway.conversations.envelope import Envelope, RecallContext, render from beaver_gateway.conversations.envelope import Envelope, RecallContext, render
from beaver_gateway.conversations.texts import EnvelopeTexts from beaver_gateway.conversations.texts import (
ConversationTexts,
EnvelopeTexts,
split_service,
)
from beaver_gateway.vault.watch import Change, VaultWatch, WatchRules from beaver_gateway.vault.watch import Change, VaultWatch, WatchRules
__all__ = ["world"] __all__ = ["world"]
@@ -77,6 +81,37 @@ def test_watch_reports_only_added_lines_with_mtime() -> None:
assert only.added == ["- 14:00 снова"] assert only.added == ["- 14:00 снова"]
def test_a_templated_note_shows_its_text_without_queries_or_frontmatter() -> None:
root, watch = vault()
rel = f"дни/{watch.today()}.md"
diary = root / rel
watch._seen.pop(rel)
diary.write_text(
"---\ntags:\n - дни\n---\n## сводка\n- \n\n\n---\n\n# планы\nпросрочены:\n"
"```tasks\nnot done\ndue before 2026-10-13\n```\n"
"обязательно сегодня:\n~~~dataview\nlist\n~~~\n"
"```python\nprint(1)\n```\n06:04 - спать\n"
)
watch.note(diary)
(change,) = watch.take()
assert change.added == [
"## сводка",
"- ",
"---",
"# планы",
"просрочены:",
"обязательно сегодня:",
"```python",
"print(1)",
"```",
"06:04 - спать",
]
append(diary, "```tasks\ndone on 2026-10-13\n```\n07:10 - встал\n")
watch.note(diary)
(later,) = watch.take()
assert later.added == ["07:10 - встал"]
def test_envelope_respects_ceilings_and_names_only_window() -> None: def test_envelope_respects_ceilings_and_names_only_window() -> None:
root, watch = vault() root, watch = vault()
today = watch.today() today = watch.today()
@@ -129,6 +164,24 @@ def test_render_hits_total_ceiling() -> None:
assert lines[1] == "time: 2026-08-26 13:04 (Warsaw)" assert lines[1] == "time: 2026-08-26 13:04 (Warsaw)"
def test_the_users_words_part_from_what_the_gateway_put_under_them() -> None:
texts = ConversationTexts()
envelope = f"{HEADER}\ntime: 2026-10-13 10:00 (UTC)"
bundle = texts.bundle_header.format(since="2026-10-13T10:00:00+00:00")
injects = f"{bundle}\n- [cron] tick"
assert split_service(f"hello\n\n{envelope}", texts) == ("hello", envelope)
assert split_service(f"hello\n\n{envelope}\n\n{injects}", texts) == (
"hello",
f"{envelope}\n\n{injects}",
)
assert split_service(f"hello\n\n{injects}", texts) == ("hello", injects)
assert split_service(f"\n\n{envelope}", texts) == ("", envelope)
assert split_service("hello\n\n[a note of mine]\nmore", texts) == (
"hello\n\n[a note of mine]\nmore",
None,
)
async def test_master_turn_gets_envelope_after_text_and_before_injects( async def test_master_turn_gets_envelope_after_text_and_before_injects(
world: World, world: World,
) -> None: ) -> None:
+85
View File
@@ -0,0 +1,85 @@
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import Any, cast
from test_conversations import StubFrontend, World, world
from beaver_gateway.jobs.job import Job, JobRun
from beaver_gateway.storage.models import Conversation
__all__ = ["world"]
async def _noop(_: JobRun) -> None:
return None
def _run(world: World) -> JobRun:
return JobRun(
job=Job("test", _noop),
trigger="test",
payload={},
scheduler=cast("Any", SimpleNamespace(conversations=world.conversations)),
)
async def _quiet_since(world: World, conv: Conversation, ago: timedelta) -> None:
async def apply(row: Conversation) -> None:
row.last_activity_at = datetime.now(UTC) - ago
await world.conversations._update(conv, apply) # noqa: SLF001
async def test_close_quiet_closes_and_marks_only_idle_branches(world: World) -> None:
marked: list[tuple[str, str]] = []
class MarkingFrontend(StubFrontend):
async def mark_closed(self, conv: Conversation) -> bool:
marked.append((conv.external_id, conv.status))
return True
tg = MarkingFrontend("tg", ("master", "branch"))
tg.conversations = world.conversations
world.conversations._frontends = [tg] # noqa: SLF001
convs = world.conversations
master = await convs.create(kind="master", agent="a", session_id="m")
idle = await convs.create(kind="branch", agent="a", parent=master, session_id="1")
fresh = await convs.create(kind="branch", agent="a", parent=master, session_id="2")
waiting = await convs.create(
kind="branch", agent="a", parent=master, session_id="3"
)
await _quiet_since(world, idle, timedelta(days=4))
await _quiet_since(world, fresh, timedelta(days=1))
await _quiet_since(world, waiting, timedelta(days=4))
await convs.queue.push(
conversation_id=cast("int", waiting.id),
priority="normal",
origin="cron",
text="later",
)
closed = await _run(world).close_quiet(kind="branch", days=3)
assert [c.external_id for c in closed] == [idle.external_id]
assert marked == [(idle.external_id, "closed")]
assert (await convs.get(fresh.external_id)).status == "open"
assert (await convs.get(waiting.external_id)).status == "open"
async def test_inject_into_falls_back_to_the_master(world: World) -> None:
convs = world.conversations
master = await convs.create(kind="master", agent="a", session_id="m")
branch = await convs.create(kind="branch", agent="a", parent=master)
gone = await convs.create(kind="branch", agent="a", parent=master)
await convs.set_status(gone, "closed")
run = _run(world)
assert await run.inject_into(branch.external_id, "to branch", urgency="normal")
assert await run.inject_into(gone.external_id, "to closed", urgency="normal")
assert await run.inject_into(None, "to nobody", urgency="normal")
assert [i.text for i in await convs.queue.pending(branch.id)] == ["to branch"]
assert [i.text for i in await convs.queue.pending(master.id)] == [
"to closed",
"to nobody",
]
+299
View File
@@ -0,0 +1,299 @@
import asyncio
import base64
import os
import sys
import tempfile
from pathlib import Path
import pytest
from cryptography.exceptions import InvalidTag
from httpx import ASGITransport, AsyncClient
from sqlmodel import select
from beaver_gateway.app import AgentRegistry, McpRegistry
from beaver_gateway.frontends.api.frontend import build_app as build_api
from beaver_gateway.frontends.base import GatewayRuntime
from beaver_gateway.keystore import (
SecretBox,
SecretsDisabledError,
SecretStore,
UnknownSecretError,
cli,
serve,
)
from beaver_gateway.security.auth import TokenStore
from beaver_gateway.storage import Database
from beaver_gateway.storage.models import AuditLog, Secret
KEY = "11" * 32
HEADERS = {"Authorization": "Bearer tok"}
@pytest.fixture
async def db():
root = Path(tempfile.mkdtemp(prefix="beaver-keys-"))
database = Database(f"sqlite:///{root / 'k.db'}")
await database.create_all()
yield database
await database.dispose()
@pytest.fixture
def store(db: Database) -> SecretStore:
return SecretStore(db, SecretBox.from_hex(KEY))
def test_box_binds_the_value_to_its_name() -> None:
box = SecretBox.from_hex(KEY)
sealed = box.seal("A_KEY", b"hunter22")
assert b"hunter22" not in sealed
assert box.open("A_KEY", sealed) == b"hunter22"
with pytest.raises(InvalidTag):
box.open("OTHER", sealed)
with pytest.raises(ValueError, match="32 bytes"):
SecretBox.from_hex("abcd")
async def test_store_keeps_only_ciphertext_and_reveals_by_name(
store: SecretStore, db: Database
) -> None:
await store.put("NODEREAL_BSC", description="bsc rpc", data=b"nr-abcdef123")
await store.put("GCP_SA", data=b'{"k": 1}', filename="../sa.json")
async with db.session() as session:
rows = (await session.exec(select(Secret))).all()
assert all(b"nr-abcdef123" not in r.sealed for r in rows)
assert {r.name: r.filename for r in rows} == {
"GCP_SA": "sa.json",
"NODEREAL_BSC": None,
}
await store.put("NODEREAL_BSC", description=" rpc key ")
listed = {r.name: r for r in await store.rows()}
assert listed["NODEREAL_BSC"].description == "rpc key"
assert listed["NODEREAL_BSC"].used_at is None
revealed = await store.reveal(["NODEREAL_BSC", "GCP_SA"])
assert [(r.name, r.data) for r in revealed] == [
("NODEREAL_BSC", b"nr-abcdef123"),
("GCP_SA", b'{"k": 1}'),
]
assert all(r.used_at is not None for r in await store.rows())
with pytest.raises(UnknownSecretError):
await store.reveal(["NODEREAL_BSC", "MISSING"])
with pytest.raises(ValueError, match="invalid secret name"):
await store.put("lower", data=b"x")
with pytest.raises(ValueError, match="needs a value"):
await store.put("EMPTY", description="no value")
assert await store.delete("GCP_SA")
assert not await store.delete("GCP_SA")
async def test_another_key_or_no_key_opens_nothing(
store: SecretStore, db: Database
) -> None:
await store.put("A_KEY", data=b"value-123")
with pytest.raises(SecretsDisabledError, match="SECRETS_KEY changed"):
await SecretStore(db, SecretBox.from_hex("22" * 32)).reveal(["A_KEY"])
off = SecretStore(db, None)
assert not off.enabled
assert [r.name for r in await off.rows()] == ["A_KEY"]
with pytest.raises(SecretsDisabledError):
await off.reveal(["A_KEY"])
with pytest.raises(SecretsDisabledError):
await off.put("B_KEY", data=b"x")
@pytest.fixture
async def socket_path(store: SecretStore):
path = Path(tempfile.mkdtemp(prefix="bk-", dir="/tmp")) / "run" / "gw.sock" # noqa: S108
used: list[list[str]] = []
async def on_use(names):
used.append(list(names))
task = asyncio.create_task(serve(store, path, on_use=on_use))
for _ in range(100):
if path.exists():
break
await asyncio.sleep(0.01)
yield path, used
task.cancel()
async def test_socket_is_root_only_and_answers_ls_and_get(
store: SecretStore, socket_path
) -> None:
path, used = socket_path
assert path.stat().st_mode & 0o777 == 0o600
assert path.parent.stat().st_mode & 0o777 == 0o700
await store.put("API_TOKEN", description="the api", data=b"tok-123456")
ls = await asyncio.to_thread(cli.request, {"op": "ls"}, path)
assert ls["secrets"] == [
{"name": "API_TOKEN", "description": "the api", "filename": None}
]
got = await asyncio.to_thread(
cli.request, {"op": "get", "names": ["API_TOKEN"]}, path
)
assert base64.b64decode(got["secrets"]["API_TOKEN"]["data"]) == b"tok-123456"
assert used == [["API_TOKEN"]]
with pytest.raises(cli.SecretError, match="unknown secret: NOPE"):
await asyncio.to_thread(cli.request, {"op": "get", "names": ["NOPE"]}, path)
with pytest.raises(cli.SecretError, match="invalid secret name"):
await asyncio.to_thread(cli.request, {"op": "get", "names": ["../x"]}, path)
async def test_run_hands_values_and_files_and_masks_the_output(
store: SecretStore, socket_path, monkeypatch, capfd
) -> None:
path, _ = socket_path
await store.put("API_TOKEN", data=b"tok/123456")
await store.put("SA_JSON", data=b'{"private": "pem"}', filename="sa.json")
monkeypatch.setattr(
cli, "ask", lambda argv: cli.request({"op": "get", "names": argv[1:]}, path)
)
shm = Path(tempfile.mkdtemp(prefix="bk-shm-"))
stale = shm / f"{cli.PREFIX}999999-old"
stale.mkdir()
script = (
"import os, sys, pathlib;"
"p = pathlib.Path(os.environ['SA_JSON']);"
"print(p.read_text(), oct(p.stat().st_mode & 0o777), p.parent.parent.name.startswith('beaver-secret-'));"
"print('url', 'https://x/?key=' + os.environ['API_TOKEN'].replace('/', '%2F'));"
"sys.stderr.write('boom ' + os.environ['API_TOKEN'] + '\\n');"
"print(p, flush=True); sys.exit(3)"
)
code = await asyncio.to_thread(
cli.run, ["API_TOKEN", "SA_JSON"], [sys.executable, "-c", script], root=shm
)
out, err = capfd.readouterr()
assert code == 3
lines = out.splitlines()
assert lines[0] == '{"private": "pem"} 0o600 True'
assert lines[1] == "url https://x/?key=<secret:API_TOKEN>"
assert "tok/123456" not in out + err
assert err == "boom <secret:API_TOKEN>\n"
assert not Path(lines[2]).exists()
assert list(shm.iterdir()) == []
def test_masker_never_splits_a_value_across_chunks() -> None:
masker = cli.Masker(cli.needles_for({"K": b"abcdef123"}))
out = b"".join(
[
masker.feed(b"one abc"),
masker.feed(b"def1"),
masker.feed(b"23 two\nthr"),
masker.flush(),
]
)
assert out == b"one <secret:K> two\nthr"
assert cli.Masker({}).feed(b"as is") == b"as is"
blob = b"x" * (cli.HOLD + 10) + b"abcdef"
masker = cli.Masker(cli.needles_for({"K": b"abcdef123"}))
head = masker.feed(blob)
assert (
head + masker.feed(b"123") + masker.flush()
== b"x" * (cli.HOLD + 10) + b"<secret:K>"
)
assert cli.needles_for({"S": b"abc"}) == {}
def test_cli_rejects_bad_shapes(capsys) -> None:
assert cli.main(["run", "A_KEY"]) == 2
assert cli.main(["run", "bad-name", "--", "true"]) == 2
assert cli.main(["--root", "get", "../etc"]) == 2
assert cli.main(["--root", "rm"]) == 2
assert "usage" in capsys.readouterr().err
class _Runtime:
def __init__(self, db: Database, store: SecretStore | None) -> None:
self.tokens = TokenStore(db, bootstrap={"t": "tok"})
self.tokens.grant("weak", "weak", scope="api")
self.runtime = GatewayRuntime(
agents=AgentRegistry([]),
mcps=McpRegistry([]),
backends={},
token_store=self.tokens,
db=db,
conversations=object(),
bus=object(),
secrets=store,
)
self.http = AsyncClient(
transport=ASGITransport(app=build_api(self.runtime)), base_url="http://api"
)
async def test_api_writes_secrets_but_never_shows_a_value(
store: SecretStore, db: Database
) -> None:
api = _Runtime(db, store).http
assert (
await api.get("/secrets", headers={"Authorization": "Bearer weak"})
).status_code == 403
put = await api.put(
"/secrets/NODEREAL_BSC",
json={"description": "bsc", "value": "nr-secret-1"},
headers=HEADERS,
)
assert put.status_code == 200, put.text
upload = await api.put(
"/secrets/SA_JSON",
json={"file": {"name": "sa.json", "content": base64.b64encode(b"{}").decode()}},
headers=HEADERS,
)
assert upload.json()["secret"]["kind"] == "file"
rename = await api.put(
"/secrets/NODEREAL_BSC", json={"description": "rpc"}, headers=HEADERS
)
assert rename.json()["secret"]["size"] == len("nr-secret-1")
listed = await api.get("/secrets", headers=HEADERS)
assert listed.json()["enabled"] is True
assert "nr-secret-1" not in listed.text
assert [(s["name"], s["description"]) for s in listed.json()["secrets"]] == [
("NODEREAL_BSC", "rpc"),
("SA_JSON", ""),
]
assert (await store.reveal(["NODEREAL_BSC"]))[0].data == b"nr-secret-1"
bad = await api.put("/secrets/bad name", json={"value": "x"}, headers=HEADERS)
assert bad.status_code == 400
both = await api.put(
"/secrets/X_KEY",
json={"value": "x", "file": {"name": "a", "content": ""}},
headers=HEADERS,
)
assert both.status_code == 400
assert (await api.delete("/secrets/SA_JSON", headers=HEADERS)).status_code == 200
assert (await api.delete("/secrets/SA_JSON", headers=HEADERS)).status_code == 404
async with db.session() as session:
audit = (await session.exec(select(AuditLog))).all()
assert [a.kind for a in audit] == [
"secret_set",
"secret_set",
"secret_set",
"secret_delete",
]
assert all("nr-secret-1" not in a.detail_json for a in audit)
async def test_api_without_a_key_lists_but_refuses_writes(db: Database) -> None:
api = _Runtime(db, SecretStore(db, None)).http
assert (await api.get("/secrets", headers=HEADERS)).json() == {
"enabled": False,
"secrets": [],
}
put = await api.put("/secrets/A_KEY", json={"value": "x"}, headers=HEADERS)
assert put.status_code == 503
def test_cli_runs_with_isolated_stdlib_python() -> None:
source = Path(cli.__file__).read_text(encoding="utf-8")
imports = {
line.split()[1].split(".")[0]
for line in source.splitlines()
if line.startswith(("import ", "from "))
}
assert imports <= set(sys.stdlib_module_names) | {"__future__"}
assert os.access(cli.__file__, os.R_OK)
+151 -23
View File
@@ -1,6 +1,11 @@
from pathlib import Path
from anthropic.types import Message, TextBlock, ThinkingBlock, ToolUseBlock, Usage from anthropic.types import Message, TextBlock, ThinkingBlock, ToolUseBlock, Usage
from beaver_gateway.frontends.markdown import renderer from beaver_gateway.frontends.markdown import history, parser, renderer
from beaver_gateway.frontends.markdown.crossfront import CrossFrontendLogger
from beaver_gateway.frontends.markdown.parser import TextSegment, ToolSegment
from beaver_gateway.frontends.turn_record import TurnRecord
def _message(*blocks: object) -> Message: def _message(*blocks: object) -> Message:
@@ -24,38 +29,161 @@ def test_tool_calls_leave_no_blank_lines() -> None:
ToolUseBlock(type="tool_use", id="t3", name="Grep", input={"pattern": "y"}), ToolUseBlock(type="tool_use", id="t3", name="Grep", input={"pattern": "y"}),
TextBlock(type="text", text="second"), TextBlock(type="text", text="second"),
) )
assert renderer.render_assistant_message(message) == ( for progress in (True, False):
"### Assistant:\n\nfirst\n\nsecond\n" assert renderer.render_assistant_message(
message, thinking_is_progress=progress
) == ("### Assistant:\n\nfirst\n\nsecond\n")
def _progress_turn() -> Message:
return _message(
TextBlock(type="text", text="Checking the containers."),
ThinkingBlock(type="thinking", thinking="", signature="s"),
ToolUseBlock(type="tool_use", id="t1", name="komodo", input={}),
ThinkingBlock(type="thinking", thinking="", signature="s"),
ThinkingBlock(type="thinking", thinking="All up, reading logs.", signature="s"),
ToolUseBlock(type="tool_use", id="t2", name="komodo", input={}),
ThinkingBlock(type="thinking", thinking="Found it.\nOne more.", signature="s"),
ToolUseBlock(type="tool_use", id="t3", name="komodo", input={}),
TextBlock(type="text", text="The stack is healthy."),
) )
def test_thinking_renders_as_collapsed_callout() -> None: def _reasoning_turn() -> Message:
message = _message( return _message(
ThinkingBlock(type="thinking", thinking="a\nb", signature="s"), ThinkingBlock(type="thinking", thinking="one", signature=""),
ToolUseBlock(type="tool_use", id="t1", name="web_search", input={}),
ThinkingBlock(type="thinking", thinking="two", signature=""),
ToolUseBlock(type="tool_use", id="t2", name="read_page", input={}),
ThinkingBlock(type="thinking", thinking="three", signature=""),
TextBlock(type="text", text="answer"), TextBlock(type="text", text="answer"),
) ThinkingBlock(type="thinking", thinking="four\nfive", signature=""),
assert renderer.render_assistant_message(message) == (
"### Assistant:\n\n> [!thinking]-\n> a\n> b\n\nanswer\n"
)
def test_thinking_between_hidden_tool_calls_collapses_to_the_first() -> None:
message = _message(
ThinkingBlock(type="thinking", thinking="one", signature="s"),
ToolUseBlock(type="tool_use", id="t1", name="Read", input={}),
ThinkingBlock(type="thinking", thinking="two", signature="s"),
ToolUseBlock(type="tool_use", id="t2", name="Read", input={}),
ThinkingBlock(type="thinking", thinking="three", signature="s"),
TextBlock(type="text", text="answer"),
ThinkingBlock(type="thinking", thinking="four", signature="s"),
TextBlock(type="text", text="more"), TextBlock(type="text", text="more"),
) )
assert renderer.render_assistant_message(message) == (
REASONING_TURN_AS_BEFORE = (
"### Assistant:\n\n> [!thinking]-\n> one\n\nanswer\n\n" "### Assistant:\n\n> [!thinking]-\n> one\n\nanswer\n\n"
"> [!thinking]-\n> four\n\nmore\n" "> [!thinking]-\n> four\n> five\n\nmore\n"
) )
def _full_circle(
message: Message, *, thinking_is_progress: bool
) -> history.ForkOutcome:
body = renderer.render_user_text("status?") + renderer.render_assistant_message(
message, thinking_is_progress=thinking_is_progress
)
turns = parser.parse(body + renderer.render_user_text("thanks")).turns
stored = [
{"role": "user", "content": "status?"},
{"role": "assistant", "content": [b.model_dump() for b in message.content]},
]
outcome = history.diff_and_fork(
stored=stored, incoming=turns, thinking_is_progress=thinking_is_progress
)
assert outcome.divergence_index is None
assert not outcome.edited
assert outcome.messages[1] == stored[1]
return outcome
def test_progress_renders_as_text_in_place_and_empty_thinking_vanishes() -> None:
assert renderer.render_assistant_message(
_progress_turn(), thinking_is_progress=True
) == (
"### Assistant:\n\nChecking the containers.\n\nAll up, reading logs.\n\n"
"Found it.\nOne more.\n\nThe stack is healthy.\n"
)
def test_reasoning_renders_as_collapsed_callouts_as_before() -> None:
assert (
renderer.render_assistant_message(_reasoning_turn(), thinking_is_progress=False)
== REASONING_TURN_AS_BEFORE
)
def test_progress_round_trips_through_the_parser_and_the_stored_history() -> None:
message = _progress_turn()
body = renderer.render_assistant_message(message, thinking_is_progress=True)
assert parser.parse_assistant_structure(body.removeprefix("### Assistant:")) == [
TextSegment(
text="Checking the containers.\n\nAll up, reading logs.\n\n"
"Found it.\nOne more.\n\nThe stack is healthy."
)
]
_full_circle(message, thinking_is_progress=True)
def test_reasoning_round_trips_through_the_parser_and_the_stored_history() -> None:
message = _reasoning_turn()
body = renderer.render_assistant_message(message, thinking_is_progress=False)
assert parser.parse_assistant_structure(body.removeprefix("### Assistant:")) == [
TextSegment(text="answer"),
TextSegment(text="more"),
]
_full_circle(message, thinking_is_progress=False)
def test_an_edited_progress_turn_rebuilds_instead_of_rewriting_signed_blocks() -> None:
message = _progress_turn()
body = renderer.render_assistant_message(
message, thinking_is_progress=True
).replace("Found it", "Fixed it")
turns = parser.parse(
renderer.render_user_text("status?") + body + renderer.render_user_text("ok")
).turns
stored = [
{"role": "user", "content": "status?"},
{"role": "assistant", "content": [b.model_dump() for b in message.content]},
]
outcome = history.diff_and_fork(
stored=stored, incoming=turns, thinking_is_progress=True
)
assert outcome.edited
assert outcome.messages[1] == {
"role": "assistant",
"content": [{"type": "text", "text": turns[1].text}],
}
async def test_a_raycast_archive_with_reasoning_renders_as_before(
tmp_path: Path,
) -> None:
logger = CrossFrontendLogger(vault_path=tmp_path, logged_subdir="logged")
record = TurnRecord(
agent_name="beaver-gemini-pro-high",
input_messages=[{"role": "user", "content": "q"}],
output_message=_reasoning_turn(),
thinking_is_progress=False,
)
await logger.handle(record)
[written] = (tmp_path / "logged" / "beaver-gemini-pro-high").glob("*.md")
body = parser.parse(written.read_text()).body
assert f"---\n\n{REASONING_TURN_AS_BEFORE}\n---\n\n### User:" in body
def test_a_file_with_collapsed_thinking_callouts_parses_as_before() -> None:
body = (
"### User:\n\nq\n\n---\n\n### Assistant:\n\n"
"> [!thinking]-\n> weighing it\n> twice\n\nanswer\n\n"
"> [!tool]- Read · notes.md\n> input\n\n"
"> [!thinking]-\n>\n\nmore\n\n---\n\n### User:\n\nnext\n"
)
parsed = parser.parse(body)
assert parsed.turns[1].structure == (
TextSegment(text="answer"),
ToolSegment(name="Read"),
TextSegment(text="more"),
)
assert parsed.messages == [
{"role": "user", "content": "q"},
{"role": "assistant", "content": "answer\n\nmore"},
{"role": "user", "content": "next"},
]
def test_user_scaffold_matches_rendered_user_turn_shape() -> None: def test_user_scaffold_matches_rendered_user_turn_shape() -> None:
body = renderer.append_to_body("### Assistant:\n\nhi\n", renderer.USER_SCAFFOLD) body = renderer.append_to_body("### Assistant:\n\nhi\n", renderer.USER_SCAFFOLD)
assert body.endswith("\n\n---\n\n### User:\n\n") assert body.endswith("\n\n---\n\n### User:\n\n")
+69
View File
@@ -0,0 +1,69 @@
from __future__ import annotations
import asyncio
import contextlib
import socket
from typing import TYPE_CHECKING, Any
import pytest
import uvicorn
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.dependencies import get_http_headers
from beaver_gateway.backends.claude_sdk import ORIGIN_HEADER
from beaver_gateway.mcp.internal_app import build_internal_app
from beaver_gateway.mcp.types import McpServer
if TYPE_CHECKING:
from collections.abc import AsyncIterator
def _free_port() -> int:
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@contextlib.asynccontextmanager
async def _serving(app: Any, port: int) -> AsyncIterator[None]:
server = uvicorn.Server(
uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning")
)
task = asyncio.create_task(server.serve())
while not server.started:
await asyncio.sleep(0.02)
try:
yield
finally:
server.should_exit = True
await task
@pytest.mark.anyio
async def test_the_origin_header_reaches_the_upstream_mcp() -> None:
upstream = FastMCP("upstream")
@upstream.tool
def caller() -> str:
return get_http_headers().get(ORIGIN_HEADER.lower(), "")
up, gateway = _free_port(), _free_port()
app, urls, _ = build_internal_app(
[McpServer.http(name="t3", url=f"http://127.0.0.1:{up}/mcp")],
host="127.0.0.1",
port=gateway,
)
seen = []
async with (
_serving(upstream.http_app(transport="http", path="/mcp"), up),
_serving(app, gateway),
):
for conversation in ("conv-a", "conv-b"):
transport = StreamableHttpTransport(
urls["t3"], headers={ORIGIN_HEADER: conversation}
)
async with Client(transport) as client:
result = await client.call_tool("caller", {})
seen.append(result.data)
assert seen == ["conv-a", "conv-b"]
+92
View File
@@ -0,0 +1,92 @@
import os
import subprocess
import sys
import time
from beaver_gateway.backends.processes import Processes, _command, subtree
SLEEP = "import time; time.sleep(60)"
NEST = (
"import subprocess, sys, time; "
f"subprocess.Popen([sys.executable, '-c', {SLEEP!r}]); time.sleep(60)"
)
def spawn(code: str) -> subprocess.Popen[bytes]:
return subprocess.Popen([sys.executable, "-c", code])
def rows_under(processes: Processes, pid: int, count: int) -> list[dict]:
deadline = time.monotonic() + 5
while time.monotonic() < deadline:
rows = subtree(processes.tree_parents_first(), pid)
if len(rows) >= count:
return rows
time.sleep(0.05)
msg = f"{pid} never grew to {count} processes"
raise AssertionError(msg)
def test_the_tree_shows_what_the_gateway_started_and_what_it_costs() -> None:
child = spawn(SLEEP)
try:
processes = Processes()
(row,) = rows_under(processes, child.pid, 1)
assert row["parent"] == os.getpid()
assert row["depth"] == 0
assert row["rss"] > 0
assert row["cpu"] is None
assert "time.sleep" in row["command"]
(again,) = rows_under(processes, child.pid, 1)
assert again["cpu"] is not None
finally:
child.kill()
child.wait()
def test_kill_takes_the_branch_and_nothing_the_gateway_did_not_start() -> None:
parent = spawn(NEST)
try:
processes = Processes()
top, nested = rows_under(processes, parent.pid, 2)
assert (top["depth"], nested["depth"]) == (0, 1)
assert not processes.kill_own_descendant(os.getpid())
assert not processes.kill_own_descendant(os.getppid())
assert processes.kill_own_descendant(parent.pid)
parent.wait(timeout=5)
alive = {row["pid"] for row in processes.tree_parents_first()}
assert not alive & {top["pid"], nested["pid"]}
finally:
parent.kill()
parent.wait()
def test_a_subtree_ends_at_the_next_sibling() -> None:
rows = [
{"pid": 1, "depth": 0},
{"pid": 2, "depth": 1},
{"pid": 3, "depth": 2},
{"pid": 4, "depth": 1},
{"pid": 5, "depth": 0},
]
assert [(r["pid"], r["depth"]) for r in subtree(rows, 2)] == [(2, 0), (3, 1)]
assert [r["pid"] for r in subtree(rows, 1)] == [1, 2, 3, 4]
assert subtree(rows, 9) == []
assert subtree(rows, None) == []
def test_a_command_line_keeps_prompts_and_configs_to_itself() -> None:
argv = [
"/opt/claude/claude",
"--model",
"m",
"--system-prompt",
"you are a beaver",
"--mcp-config",
'{"gateway": {"headers": {"Authorization": "Bearer abc"}}}',
"x" * 100,
]
assert _command(argv) == (
"claude --model m --system-prompt … --mcp-config … " + "x" * 80 + "…"
)
assert _command([]) == ""
+11
View File
@@ -84,6 +84,17 @@ async def test_rotation_does_not_touch_a_master_mid_turn(world: World) -> None:
await world.settle(conv, 1) await world.settle(conv, 1)
async def test_force_rotates_a_master_that_is_not_due(world: World) -> None:
conv = await world.conversations.create(kind="master", agent="a", origin="test")
rotation = Rotation(world.conversations, POLICY)
assert await rotation.tick() == []
rotated = await rotation.tick(force="manual")
assert len(rotated) == 1
assert rotated[0].external_id != conv.external_id
assert (await world.conversations.get(conv.external_id)).status == "closed"
assert len(await world.conversations.find(kind="master", status="open")) == 1
async def test_rotation_order_handout_close_marks_moves_and_new_day( async def test_rotation_order_handout_close_marks_moves_and_new_day(
world: World, world: World,
) -> None: ) -> None:
+137 -2
View File
@@ -20,6 +20,7 @@ from beaver_gateway.frontends.base import GatewayRuntime
from beaver_gateway.frontends.files import SendFileError, SendFiles, method_for from beaver_gateway.frontends.files import SendFileError, SendFiles, method_for
from beaver_gateway.frontends.telegram import Attachments, TelegramFrontend from beaver_gateway.frontends.telegram import Attachments, TelegramFrontend
from beaver_gateway.frontends.telegram.drafts import Draft from beaver_gateway.frontends.telegram.drafts import Draft
from beaver_gateway.frontends.telegram.texts import TelegramTexts
from beaver_gateway.frontends.telegram.render import ( from beaver_gateway.frontends.telegram.render import (
LIMIT, LIMIT,
chunks, chunks,
@@ -518,10 +519,10 @@ async def test_commands_status_merge_and_new(stack: Stack) -> None:
frontend="telegram", external_id=f"{USER}/11" frontend="telegram", external_id=f"{USER}/11"
) )
assert branch.status == "merged" assert branch.status == "merged"
assert stack.bot.topics == ["🦫 General", "edit:11:✅ work"] assert stack.bot.topics == ["🦫 General", "edit:11:work", "edit:11:✅ work"]
stack.bot.message("/new отчёт") stack.bot.message("/new отчёт")
await stack.until(lambda: stack.sent_with("in a new topic"), what="new topic") await stack.until(lambda: stack.sent_with("in a new topic"), what="new topic")
assert stack.bot.topics == ["🦫 General", "edit:11:✅ work", "отчёт"] assert stack.bot.topics[-1] == "отчёт"
child = await stack.world.conversations.find_bound( child = await stack.world.conversations.find_bound(
frontend="telegram", external_id=f"{USER}/902" frontend="telegram", external_id=f"{USER}/902"
) )
@@ -1194,3 +1195,137 @@ async def test_a_refused_upload_comes_back_to_the_agent(stack: Stack) -> None:
outside = await _send(stack, master, path="/etc/passwd") outside = await _send(stack, master, path="/etc/passwd")
assert outside["is_error"] assert outside["is_error"]
assert "outside the allowed roots" in outside["content"][0]["text"] assert "outside the allowed roots" in outside["content"][0]["text"]
def _progress_draft() -> Draft:
return Draft(cast("Any", None), chat_id=USER, thread_id=None, turn_id="t")
def _thinking_delta(text: str) -> Any:
return {
"event": {
"type": "content_block_delta",
"index": 0,
"delta": {"type": "thinking_delta", "thinking": text},
}
}
def test_progress_updates_reach_the_draft_but_hidden_reasoning_does_not() -> None:
draft = _progress_draft()
texts = TelegramTexts()
stub = SimpleNamespace(_drafts={"k": draft}, texts=texts)
feed = functools.partial(
TelegramFrontend._feed_draft, cast("Any", stub), "k", thinking_is_progress=True
)
feed(_thinking_delta(""))
assert draft.text == ""
assert draft.status == texts.thinking
feed(_thinking_delta("checking the vault"))
assert draft.text == "checking the vault"
assert draft.status == texts.writing
def test_reasoning_rides_under_the_status_and_never_reaches_the_text() -> None:
draft = _progress_draft()
texts = TelegramTexts()
stub = SimpleNamespace(_drafts={"k": draft}, texts=texts)
feed = functools.partial(
TelegramFrontend._feed_draft, cast("Any", stub), "k", thinking_is_progress=False
)
feed(_thinking_delta("weighing <it> "))
feed(_thinking_delta("against the note"))
assert draft.text == ""
assert draft.status == texts.thinking
assert draft._render() == (
f"{texts.thinking}\n\n"
"<blockquote>weighing &lt;it&gt; against the note</blockquote>",
f"{texts.thinking}\n\nweighing <it> against the note",
)
feed(
{
"event": {
"type": "content_block_start",
"index": 2,
"content_block": {"type": "thinking"},
}
}
)
assert draft.reasoning == ""
feed(_thinking_delta("word " * 200))
shown = draft._render()[1].removeprefix(f"{texts.thinking}\n\n")
assert shown.startswith("… word")
assert len(shown) <= 402
feed(
{
"event": {
"type": "content_block_delta",
"index": 3,
"delta": {"type": "text_delta", "text": "Done."},
}
}
)
assert draft.reasoning == ""
assert draft._render()[1] == f"{texts.writing}\n\nDone."
def test_a_tool_call_between_progress_updates_starts_a_new_paragraph() -> None:
draft = _progress_draft()
texts = TelegramTexts()
stub = SimpleNamespace(
_drafts={"k": draft},
texts=texts,
_tool_status=lambda name, _input: texts.tool.format(label=name),
)
feed = functools.partial(
TelegramFrontend._feed_draft, cast("Any", stub), "k", thinking_is_progress=True
)
feed(_thinking_delta("reading the note"))
feed(
{
"event": {
"type": "content_block_start",
"index": 1,
"content_block": {"type": "tool_use", "name": "Read"},
}
}
)
feed(_thinking_delta("now writing it back"))
assert draft.text == "reading the note\n\nnow writing it back"
async def test_a_closed_topic_is_marked_asleep_and_wakes_with_a_new_branch(
stack: Stack,
) -> None:
stack.bot.message("кухня на неделю\nи ещё строка", thread=12)
await stack.until(lambda: stack.sent_with("кухня на неделю"), what="reply")
assert stack.bot.topics[-1] == "edit:12:кухня на неделю"
branch = await stack.world.conversations.find_bound(
frontend="telegram", external_id=f"{USER}/12"
)
closed = await stack.world.conversations.close(branch)
assert await stack.world.conversations.mark_closed(closed)
assert stack.bot.topics[-1] == "edit:12:💤 кухня на неделю"
stack.tg._topic_names.clear() # noqa: SLF001
stack.bot.message("а в четверг?", thread=12)
await stack.until(lambda: stack.sent_with("а в четверг?"), what="new branch")
assert stack.bot.topics[-1] == "edit:12:кухня на неделю"
fresh = await stack.world.conversations.find_bound(
frontend="telegram", external_id=f"{USER}/12"
)
assert fresh.id != branch.id
assert fresh.status == "open"
assert fresh.title == "кухня на неделю"
async def test_a_named_topic_keeps_its_name(stack: Stack) -> None:
stack.bot.topic_created("план", 13)
stack.bot.message("hello", thread=13)
await stack.until(lambda: stack.sent_with("hello"), what="reply")
assert not any(t.startswith("edit:13:") for t in stack.bot.topics)
+63
View File
@@ -0,0 +1,63 @@
import tempfile
from collections.abc import AsyncIterator
from pathlib import Path
import pytest
from watchfiles import Change as FileChange
from beaver_gateway.vault import watch as watch_module
from beaver_gateway.vault.watch import VaultFilter, VaultWatch, WatchRules
VAULT = Path("/vault")
FILTER = VaultFilter()
def wakes(rel: str) -> bool:
return FILTER(FileChange.modified, str(VAULT / rel))
def test_obsidian_never_wakes_the_watcher() -> None:
assert not wakes(".obsidian/.sync.lock")
assert not wakes(".obsidian/.sync.lock/data")
assert not wakes(".obsidian")
assert not wakes(".obsidian/workspace.json")
assert not wakes(".obsidian/plugins/obsidian-git/data.json")
def test_other_service_paths_never_wake_the_watcher() -> None:
assert not wakes(".trash/📅 дни/2026-09-01.md")
assert not wakes(".git/index")
def test_notes_still_wake_the_watcher() -> None:
assert wakes("📅 дни/2026-09-09.md")
assert wakes("мета/бобер/состояние.md")
assert wakes("заметка.md")
assert wakes("💻 проекты/бобер/план.md")
def test_lock_beside_a_note_is_quiet_but_the_note_is_not() -> None:
assert not wakes("📅 дни/.2026-09-09.md.lock")
assert not wakes("📅 дни/2026-09-09.md.tmp")
assert not wakes("📅 дни/2026-09-09.md~")
assert wakes("📅 дни/2026-09-09.md")
async def test_the_filter_is_applied_by_awatch_itself(
monkeypatch: pytest.MonkeyPatch,
) -> None:
seen: dict[str, object] = {}
async def fake_awatch(
*paths: object, **kwargs: object
) -> AsyncIterator[set[object]]:
seen.update(kwargs)
seen["paths"] = paths
for change in ():
yield change
monkeypatch.setattr(watch_module, "awatch", fake_awatch)
root = Path(tempfile.mkdtemp(prefix="beaver-vault-"))
watch = VaultWatch(root, WatchRules(names=("**",)), tz="UTC")
await watch.run()
assert isinstance(seen["watch_filter"], VaultFilter)
+48
View File
@@ -14,8 +14,12 @@ import type {
MemoryTree, MemoryTree,
ScopeRow, ScopeRow,
SearchResponse, SearchResponse,
SecretRow,
SecretsResponse,
SecretWrite,
SessionsResponse, SessionsResponse,
TokenRow, TokenRow,
TurnSummary,
UsageGroup, UsageGroup,
UsageResponse, UsageResponse,
VaultGraph, VaultGraph,
@@ -146,6 +150,14 @@ export class ApiClient {
return this.request<T>("PATCH", path, body ?? {}); return this.request<T>("PATCH", path, body ?? {});
} }
put<T>(path: string, body?: unknown): Promise<T> {
return this.request<T>("PUT", path, body ?? {});
}
delete<T>(path: string): Promise<T> {
return this.request<T>("DELETE", path);
}
async stream( async stream(
path: string, path: string,
onEvent: (event: BusEvent) => void, onEvent: (event: BusEvent) => void,
@@ -187,6 +199,14 @@ export class ApiClient {
return this.get(`/api/conversations/${id}/history`); return this.get(`/api/conversations/${id}/history`);
} }
turns(id: string, limit: number): Promise<{ turns: TurnSummary[] }> {
return this.get(`/api/conversations/${id}/turns`, { limit });
}
rewind(id: string, turn: string): Promise<{ turns: number }> {
return this.post(`/api/conversations/${id}/rewind`, { turn });
}
context(id: string, prompt = false): Promise<ContextResponse> { context(id: string, prompt = false): Promise<ContextResponse> {
return this.get(`/api/conversations/${id}/context`, { return this.get(`/api/conversations/${id}/context`, {
prompt: prompt ? 1 : undefined, prompt: prompt ? 1 : undefined,
@@ -313,6 +333,22 @@ export class ApiClient {
return this.get("/api/sessions"); return this.get("/api/sessions");
} }
sessionStderr(key: string): Promise<{ key: string; lines: string[] }> {
return this.get(`/api/sessions/${encodeURIComponent(key)}/stderr`);
}
interruptSession(key: string): Promise<{ interrupted: boolean }> {
return this.post(`/api/sessions/${encodeURIComponent(key)}/interrupt`);
}
closeSession(key: string): Promise<{ closed: boolean }> {
return this.post(`/api/sessions/${encodeURIComponent(key)}/close`);
}
killProcess(pid: number): Promise<{ killed: boolean }> {
return this.post(`/api/processes/${pid}/kill`);
}
jobs(): Promise<JobsResponse> { jobs(): Promise<JobsResponse> {
return this.get("/api/jobs"); return this.get("/api/jobs");
} }
@@ -367,6 +403,18 @@ export class ApiClient {
return this.post(`/api/tokens/${id}/revoke`); return this.post(`/api/tokens/${id}/revoke`);
} }
secrets(): Promise<SecretsResponse> {
return this.get("/api/secrets");
}
putSecret(name: string, body: SecretWrite): Promise<{ secret: SecretRow }> {
return this.put(`/api/secrets/${encodeURIComponent(name)}`, body);
}
deleteSecret(name: string): Promise<{ name: string; deleted: boolean }> {
return this.delete(`/api/secrets/${encodeURIComponent(name)}`);
}
audit(params?: { before?: number; limit?: number }): Promise<AuditPage> { audit(params?: { before?: number; limit?: number }): Promise<AuditPage> {
return this.get("/api/audit", params); return this.get("/api/audit", params);
} }
+49
View File
@@ -79,6 +79,7 @@ export interface ConversationInfo extends ConversationSummary {
parent: string | null; parent: string | null;
question: PendingQuestion | null; question: PendingQuestion | null;
queue: QueueItem[]; queue: QueueItem[];
thinking_is_progress?: boolean;
turn: TurnSnapshot | null; turn: TurnSnapshot | null;
} }
@@ -105,11 +106,21 @@ export interface ContentBlock {
export interface HistoryMessage { export interface HistoryMessage {
content: string | ContentBlock[]; content: string | ContentBlock[];
id?: string | null;
role: "user" | "assistant"; role: "user" | "assistant";
service?: string;
// When the first block of this message was written; absent for old rows. // When the first block of this message was written; absent for old rows.
ts?: string | null; ts?: string | null;
} }
export interface TurnSummary {
failed: boolean;
id: string;
reply: string;
text: string;
ts: string | null;
}
export interface EntriesPage { export interface EntriesPage {
entries: Record<string, unknown>[]; entries: Record<string, unknown>[];
id: string; id: string;
@@ -143,17 +154,31 @@ export interface AgentsResponse {
mcps: { name: string; kind: string }[]; mcps: { name: string; kind: string }[];
} }
export interface ProcessRow {
age_seconds: number;
command: string;
cpu: number | null;
depth: number;
name: string;
parent: number;
pid: number;
rss: number;
}
export interface SessionRow { export interface SessionRow {
age_seconds: number; age_seconds: number;
agent: string; agent: string;
busy: boolean; busy: boolean;
cpu: number | null;
dirty: boolean; dirty: boolean;
idle_seconds: number; idle_seconds: number;
key: string; key: string;
kind: string; kind: string;
pending: number;
pending_question: boolean; pending_question: boolean;
pid: number | null; pid: number | null;
pinned: boolean; pinned: boolean;
processes: ProcessRow[];
rss: number | null; rss: number | null;
running_turn: string | null; running_turn: string | null;
session_id: string | null; session_id: string | null;
@@ -161,6 +186,8 @@ export interface SessionRow {
} }
export interface SessionsResponse { export interface SessionsResponse {
cpu: number | null;
others: ProcessRow[];
rss: number | null; rss: number | null;
rss_limit: number | null; rss_limit: number | null;
sessions: SessionRow[]; sessions: SessionRow[];
@@ -260,6 +287,28 @@ export interface TokenRow {
scope: string; scope: string;
} }
export interface SecretRow {
created_at: string;
description: string;
filename: string | null;
kind: "file" | "value";
name: string;
size: number;
updated_at: string;
used_at: string | null;
}
export interface SecretsResponse {
enabled: boolean;
secrets: SecretRow[];
}
export interface SecretWrite {
description?: string;
file?: { content: string; name: string };
value?: string;
}
export interface AuditRecord { export interface AuditRecord {
actor: string; actor: string;
agent: string | null; agent: string | null;
+13
View File
@@ -68,6 +68,12 @@ export function fmtSeconds(seconds: number | null | undefined): string {
: fmtDuration(seconds * THOUSAND); : fmtDuration(seconds * THOUSAND);
} }
export function fmtCpu(percent: number | null | undefined): string {
return percent === null || percent === undefined
? "–"
: `${Math.round(percent)}%`;
}
export function fmtBytes(bytes: number | null | undefined): string { export function fmtBytes(bytes: number | null | undefined): string {
if (bytes === null || bytes === undefined) { if (bytes === null || bytes === undefined) {
return "–"; return "–";
@@ -103,6 +109,13 @@ export function fmtTime(iso: string | null | undefined): string {
: "–"; : "–";
} }
export function fmtMinute(iso: string | null | undefined): string {
const date = parseDate(iso);
return date
? date.toLocaleTimeString(undefined, { hour: "2-digit", minute: "2-digit" })
: "–";
}
export function fmtDateTime(iso: string | null | undefined): string { export function fmtDateTime(iso: string | null | undefined): string {
const date = parseDate(iso); const date = parseDate(iso);
return date return date
+1
View File
@@ -20,6 +20,7 @@ export const SYSTEM_PAGES: NavItem[] = [
{ href: "/system/agents", key: "", label: "Agents & endpoints" }, { href: "/system/agents", key: "", label: "Agents & endpoints" },
{ href: "/system/jobs", key: "", label: "Jobs" }, { href: "/system/jobs", key: "", label: "Jobs" },
{ href: "/system/tokens", key: "", label: "Tokens" }, { href: "/system/tokens", key: "", label: "Tokens" },
{ href: "/system/secrets", key: "", label: "Secrets" },
{ href: "/system/audit", key: "", label: "Audit" }, { href: "/system/audit", key: "", label: "Audit" },
]; ];
+8
View File
@@ -9,6 +9,7 @@ import FileTextIcon from "@lucide/svelte/icons/file-text";
import GitBranchIcon from "@lucide/svelte/icons/git-branch"; import GitBranchIcon from "@lucide/svelte/icons/git-branch";
import GitMergeIcon from "@lucide/svelte/icons/git-merge"; import GitMergeIcon from "@lucide/svelte/icons/git-merge";
import PencilIcon from "@lucide/svelte/icons/pencil"; import PencilIcon from "@lucide/svelte/icons/pencil";
import RewindIcon from "@lucide/svelte/icons/rewind";
import RotateCcwIcon from "@lucide/svelte/icons/rotate-ccw"; import RotateCcwIcon from "@lucide/svelte/icons/rotate-ccw";
import SendIcon from "@lucide/svelte/icons/send"; import SendIcon from "@lucide/svelte/icons/send";
import type { Component } from "svelte"; import type { Component } from "svelte";
@@ -38,6 +39,7 @@ export interface ActionScope {
onChanged: () => void; onChanged: () => void;
onOpen: (id: string) => void; onOpen: (id: string) => void;
onRename: () => void; onRename: () => void;
onRewind: () => void;
} }
const TELEGRAM = "telegram"; const TELEGRAM = "telegram";
@@ -103,6 +105,12 @@ export function conversationActions(
label: "Branch…", label: "Branch…",
run: scope.onBranch, run: scope.onBranch,
}); });
actions.push({
disabled: !info.session_id,
icon: RewindIcon,
label: "Rewind…",
run: scope.onRewind,
});
actions.push({ actions.push({
checked: remembers, checked: remembers,
gap: true, gap: true,
+34 -6
View File
@@ -30,6 +30,8 @@ export interface Turn {
// A tool call came between two pieces of text; the next piece starts a // A tool call came between two pieces of text; the next piece starts a
// paragraph, as the final message will show it. // paragraph, as the final message will show it.
paused: boolean; paused: boolean;
// What the model thought, when its thinking is reasoning and not progress.
reasoning: string;
resultSubtype: string | null; resultSubtype: string | null;
roots: string[]; roots: string[];
says: string[]; says: string[];
@@ -223,22 +225,47 @@ export class ActivityModel {
const turn = this.turnFor(turnId, ts, parent); const turn = this.turnFor(turnId, ts, parent);
const sdk = raw as Record<string, unknown>; const sdk = raw as Record<string, unknown>;
if (sdk.type === "content_block_start") { if (sdk.type === "content_block_start") {
const block = sdk.content_block as Record<string, unknown> | undefined; this.onBlockStart(turn, sdk.content_block, parent, ts);
} else if (sdk.type === "content_block_delta" && parent === null) {
this.onDelta(turn, sdk.delta as Record<string, unknown> | undefined);
}
}
private onBlockStart(
turn: Turn,
raw: unknown,
parent: string | null,
ts: string
): void {
const block = raw as Record<string, unknown> | undefined;
if (block?.type === "tool_use" && typeof block.id === "string") { if (block?.type === "tool_use" && typeof block.id === "string") {
this.ensureNode(turn, block.id, str(block.name) ?? "?", parent, ts); this.ensureNode(turn, block.id, str(block.name) ?? "?", parent, ts);
} else if (block?.type === "thinking" && parent === null) { } else if (block?.type === "thinking" && parent === null) {
turn.thinking += 1; turn.thinking += 1;
if (turn.reasoning) {
turn.reasoning += "\n\n";
} }
}
}
private onDelta(
turn: Turn,
delta: Record<string, unknown> | undefined
): void {
const thought = delta?.type === "thinking_delta" ? delta.thinking : null;
if (
typeof thought === "string" &&
this.conversation?.thinking_is_progress !== true
) {
turn.reasoning += thought;
return; return;
} }
if (sdk.type === "content_block_delta" && parent === null) { const text = thought ?? delta?.text;
const delta = sdk.delta as Record<string, unknown> | undefined; if (typeof text === "string" && text) {
if (delta?.type === "text_delta" && typeof delta.text === "string") { turn.text = joinText(turn.text, text, turn.paused);
turn.text = joinText(turn.text, delta.text, turn.paused);
turn.paused = false; turn.paused = false;
} }
} }
}
private onTool({ event, ts, turnId, parent }: Cursor): void { private onTool({ event, ts, turnId, parent }: Cursor): void {
if (!(turnId && typeof event.tool_use_id === "string")) { if (!(turnId && typeof event.tool_use_id === "string")) {
@@ -360,6 +387,7 @@ export class ActivityModel {
nodes: {}, nodes: {},
origin: origin ?? "?", origin: origin ?? "?",
paused: false, paused: false,
reasoning: "",
resultSubtype: null, resultSubtype: null,
roots: [], roots: [],
says: [], says: [],
+253 -158
View File
@@ -6,12 +6,16 @@
import ErrorNote from "$lib/components/error-note.svelte"; import ErrorNote from "$lib/components/error-note.svelte";
import { clip, fmtDateTime, fmtTime } from "$lib/format"; import { clip, fmtDateTime, fmtTime } from "$lib/format";
import { cn } from "$lib/utils"; import { cn } from "$lib/utils";
import type { ActivityModel } from "./activity.svelte"; import type { ActivityModel, ToolNode } from "./activity.svelte";
import { isLive, summarizeInput, toolLabel } from "./activity.svelte"; import { isLive } from "./activity.svelte";
import { cacheableHistory, historyKey } from "./history-cache"; import { cacheableHistory, historyKey } from "./history-cache";
import { usePanelHost } from "./host"; import { usePanelHost } from "./host";
import Markdown from "./markdown.svelte"; import Markdown from "./markdown.svelte";
import QuestionCard from "./question-card.svelte"; import QuestionCard from "./question-card.svelte";
import Reasoning from "./reasoning.svelte";
import SayBubble from "./say-bubble.svelte";
import ServiceLine from "./service-line.svelte";
import ToolNodeView from "./tool-node.svelte";
import TurnCard from "./turn-card.svelte"; import TurnCard from "./turn-card.svelte";
let { let {
@@ -26,25 +30,42 @@
refreshKey?: number; refreshKey?: number;
} = $props(); } = $props();
const RESULT_CLIP = 600; interface Step {
// Markdown for the last few screens renders at once; the rest waits block: ContentBlock;
// until scrolled near, so a long thread paints as fast as a short one. eager: boolean;
const EAGER_MARKDOWN = 40; key: string;
stampedAt: string | null;
}
type Row =
| { day: string; key: string; kind: "day" }
| { key: string; kind: "system"; text: string; ts: string | null }
| {
eager: boolean;
key: string;
kind: "user";
service: string | null;
texts: string[];
ts: string | null;
}
| { key: string; kind: "agent"; steps: Step[] };
const RESULT_CLIP = 20_000;
const RENDERED_AT_ONCE = 40;
const host = usePanelHost(); const host = usePanelHost();
const NEAR_BOTTOM_PX = 120; const AT_BOTTOM_PX = 8;
const TICK_MS = 1000; const TICK_MS = 1000;
const NO_NODES: Record<string, ToolNode> = {};
let messages = $state<HistoryMessage[] | null>(null); let messages = $state<HistoryMessage[] | null>(null);
let failure = $state<string | null>(null); let failure = $state<string | null>(null);
let showResults = $state(false);
let loadedAt = $state(new Date(0).toISOString()); let loadedAt = $state(new Date(0).toISOString());
let scroller = $state<HTMLDivElement | null>(null); let scroller = $state<HTMLDivElement | null>(null);
let body = $state<HTMLDivElement | null>(null); let body = $state<HTMLDivElement | null>(null);
const SETTLE_MS = 2500;
const SETTLE_STEP_MS = 120;
let settleUntil = 0;
let now = $state(Date.now()); let now = $state(Date.now());
let pinned = true; let pinned = true;
let readerRow: { node: Element; top: number } | null = null;
let ownScroll = false;
const tail = $derived( const tail = $derived(
model.turns model.turns
@@ -53,7 +74,11 @@
); );
const tailSize = $derived( const tailSize = $derived(
tail.reduce( tail.reduce(
(n, turn) => n + turn.text.length + Object.keys(turn.nodes).length, (n, turn) =>
n +
turn.text.length +
turn.reasoning.length +
Object.keys(turn.nodes).length,
0 0
) )
); );
@@ -71,46 +96,61 @@
try { try {
({ messages } = await client.history(conversationId)); ({ messages } = await client.history(conversationId));
loadedAt = new Date().toISOString(); loadedAt = new Date().toISOString();
settle();
host.cache?.set(historyKey(conversationId), cacheableHistory(messages)); host.cache?.set(historyKey(conversationId), cacheableHistory(messages));
} catch (cause) { } catch (cause) {
failure = cause instanceof Error ? cause.message : String(cause); failure = cause instanceof Error ? cause.message : String(cause);
} }
} }
// The scroller is column-reverse: its origin is the bottom, scrollTop is
// zero there and goes negative upward, and the browser keeps the view at
// the bottom while content grows - no matter how late markdown lands.
function nearBottom(): boolean {
return !scroller || Math.abs(scroller.scrollTop) < NEAR_BOTTOM_PX;
}
function toBottom() { function toBottom() {
if (scroller) { if (scroller) {
scroller.scrollTop = 0; scroller.scrollTop = 0;
} }
} }
async function scrollToBottom() { function topRow(): { node: Element; top: number } | null {
await tick(); if (!(scroller && body)) {
toBottom(); return null;
}
const edge = scroller.getBoundingClientRect().top;
const rows = body.children;
let low = 0;
let high = rows.length - 1;
while (low < high) {
const middle = Math.floor((low + high) / 2);
if (rows[middle].getBoundingClientRect().bottom > edge) {
high = middle;
} else {
low = middle + 1;
}
}
const node = rows[low];
return node ? { node, top: node.getBoundingClientRect().top } : null;
} }
// A fresh thread keeps growing after it is on screen: markdown renders function trackReader() {
// late, embeds load, fonts land. Hold the bottom for a while after a load. if (ownScroll) {
function settle() { ownScroll = false;
settleUntil = Date.now() + SETTLE_MS;
const step = () => {
if (!pinned || Date.now() > settleUntil) {
return; return;
} }
pinned = !scroller || Math.abs(scroller.scrollTop) < AT_BOTTOM_PX;
readerRow = pinned ? null : topRow();
}
function keepReaderRowInPlace() {
if (pinned) {
toBottom(); toBottom();
setTimeout(step, SETTLE_STEP_MS); return;
}; }
step(); if (!(scroller && readerRow?.node.isConnected)) {
return;
}
const drift = readerRow.node.getBoundingClientRect().top - readerRow.top;
const before = scroller.scrollTop;
scroller.scrollTop += drift;
ownScroll = scroller.scrollTop !== before;
} }
// The date changes rarely; say it once, then only the time.
function dayOf(iso: string | null | undefined): string { function dayOf(iso: string | null | undefined): string {
return iso ? iso.slice(0, 10) : ""; return iso ? iso.slice(0, 10) : "";
} }
@@ -121,7 +161,6 @@
$effect(() => { $effect(() => {
if (refreshKey >= 0) { if (refreshKey >= 0) {
pinned = true;
untrack(() => load()); untrack(() => load());
} }
}); });
@@ -129,21 +168,15 @@
$effect(() => { $effect(() => {
const count = (messages?.length ?? 0) + tailSize; const count = (messages?.length ?? 0) + tailSize;
if (pinned && count >= 0) { if (pinned && count >= 0) {
scrollToBottom(); tick().then(toBottom);
} }
}); });
// Markdown lands asynchronously (Obsidian renders after the DOM exists),
// so follow the content's height, not just the message count.
$effect(() => { $effect(() => {
if (!(scroller && body)) { if (!(scroller && body)) {
return; return;
} }
const observer = new ResizeObserver(() => { const observer = new ResizeObserver(keepReaderRowInPlace);
if (pinned && scroller) {
scroller.scrollTop = scroller.scrollHeight;
}
});
observer.observe(body); observer.observe(body);
return () => observer.disconnect(); return () => observer.disconnect();
}); });
@@ -157,42 +190,30 @@
return () => clearInterval(timer); return () => clearInterval(timer);
}); });
// Text the gateway or the CLI put in the user's seat: an inject header,
// an envelope, a subagent's report.
const SYSTEM_HEAD = /^(\[[^\]\n]+\]|<task-notification>)/; const SYSTEM_HEAD = /^(\[[^\]\n]+\]|<task-notification>)/;
const SECONDS = /:\d{2}$/; const SECONDS = /:\d{2}$/;
const DAY_TIME = /,?\s*\d{2}:\d{2}$/; const DAY_TIME = /,?\s*\d{2}:\d{2}$/;
let openSystem = $state<Set<number>>(new Set());
function systemText(message: HistoryMessage): string | null {
if (message.role !== "user" || typeof message.content !== "string") {
return null;
}
return SYSTEM_HEAD.test(message.content.trimStart())
? message.content
: null;
}
function toggleSystem(index: number) {
const next = new Set(openSystem);
if (next.has(index)) {
next.delete(index);
} else {
next.add(index);
}
openSystem = next;
}
// Keys count from the end: when the network copy grows the head past
// what the cache held, the tail keeps its nodes and its rendered markdown.
function keyOf(message: HistoryMessage, index: number, total: number) { function keyOf(message: HistoryMessage, index: number, total: number) {
return `${index - total}:${message.role}:${message.ts ?? ""}`; const fromEnd = `${index - total}:${message.role}:${message.ts ?? ""}`;
return message.id ?? fromEnd;
} }
function blocks(message: HistoryMessage): ContentBlock[] { function blocks(message: HistoryMessage): ContentBlock[] {
return typeof message.content === "string" if (typeof message.content === "string") {
? [{ text: message.content, type: "text" }] return [{ text: message.content, type: "text" }];
: message.content; }
if (!model.conversation?.thinking_is_progress) {
return message.content;
}
return message.content.flatMap((block) => {
if (block.type !== "thinking") {
return [block];
}
return block.thinking?.trim()
? [{ text: block.thinking, type: "text" }]
: [];
});
} }
function resultText(block: ContentBlock): string { function resultText(block: ContentBlock): string {
@@ -211,28 +232,127 @@
} }
return ""; return "";
} }
const results = $derived.by(() => {
const found = new Map<string, { failed: boolean; text: string }>();
for (const message of messages ?? []) {
if (typeof message.content === "string") {
continue;
}
for (const block of message.content) {
if (block.type === "tool_result" && block.tool_use_id) {
found.set(block.tool_use_id, {
failed: block.is_error === true,
text: clip(resultText(block), RESULT_CLIP),
});
}
}
}
return found;
});
function nodeOf(block: ContentBlock): ToolNode {
const result = block.id ? results.get(block.id) : undefined;
let status: ToolNode["status"] = "aborted";
if (result) {
status = result.failed ? "error" : "done";
}
return {
children: [],
endedAt: null,
id: block.id ?? "",
input: block.input ?? null,
name: block.name ?? "?",
parent: null,
result: result?.text ?? "",
startedAt: "",
status,
};
}
function said(block: ContentBlock): string | null {
const text = (block.input as { text?: unknown } | null | undefined)?.text;
return block.name?.endsWith("__say") && typeof text === "string"
? text
: null;
}
function userRow(
message: HistoryMessage,
parts: ContentBlock[],
key: string,
eager: boolean
): Row {
const ts = message.ts ?? null;
const texts = parts.flatMap((b) =>
b.type === "text" && b.text ? [b.text] : []
);
if (!message.service && SYSTEM_HEAD.test(texts[0]?.trimStart() ?? "")) {
return { key, kind: "system", text: texts.join("\n"), ts };
}
const service = message.service ?? null;
return { eager, key, kind: "user", service, texts, ts };
}
function stepsOf(
message: HistoryMessage,
parts: ContentBlock[],
key: string,
eager: boolean,
minute: string
): Step[] {
const ts = message.ts ?? null;
const stamped =
minuteOf(ts) !== minute ||
parts.some((b) => b.type === "text" && Boolean(b.text));
return parts.map((block, at) => ({
block,
eager,
key: `${key}:${at}`,
stampedAt: stamped && at === 0 ? ts : null,
}));
}
const rows = $derived.by(() => {
const out: Row[] = [];
const list = messages ?? [];
let run: Step[] | null = null;
let day = "";
let minute = "";
for (const [index, message] of list.entries()) {
const parts = blocks(message);
if (parts.every((b) => b.type === "tool_result")) {
continue;
}
const key = keyOf(message, index, list.length);
const eager = index >= list.length - RENDERED_AT_ONCE;
if (message.ts && dayOf(message.ts) !== day) {
day = dayOf(message.ts);
const label = fmtDateTime(message.ts).replace(DAY_TIME, "");
out.push({ day: label, key, kind: "day" });
run = null;
}
if (message.role === "user") {
out.push(userRow(message, parts, key, eager));
run = null;
} else if (run === null) {
run = stepsOf(message, parts, key, eager, minute);
out.push({ key, kind: "agent", steps: run });
} else {
run.push(...stepsOf(message, parts, key, eager, minute));
}
minute = message.ts ? minuteOf(message.ts) : minute;
}
return out;
});
</script> </script>
<div <div
class="flex min-h-0 flex-1 flex-col-reverse overflow-y-auto overflow-x-hidden px-3 py-3 [overflow-anchor:none] @md:px-6" class="flex min-h-0 flex-1 flex-col-reverse overflow-y-auto overflow-x-hidden px-3 py-3 [overflow-anchor:none] @md:px-6"
onscroll={() => { onscroll={trackReader}
if (Date.now() > settleUntil) {
pinned = nearBottom();
}
}}
bind:this={scroller} bind:this={scroller}
> >
<div class="flex min-w-0 max-w-full flex-col gap-3" bind:this={body}> <div class="flex min-w-0 max-w-full flex-col gap-3" bind:this={body}>
<button
class="rule-word self-end text-xs"
data-active={showResults}
onclick={() => {
showResults = !showResults;
}}
type="button"
>
{showResults ? "hide tool results" : "show tool results"}
</button>
{#if failure} {#if failure}
<ErrorNote message={failure} retry={load} /> <ErrorNote message={failure} retry={load} />
{:else if messages === null} {:else if messages === null}
@@ -245,97 +365,72 @@
title="Empty conversation" title="Empty conversation"
/> />
{:else} {:else}
{#each messages as message, index (keyOf(message, index, messages.length))} {#each rows as row (`${row.kind}:${row.key}`)}
{@const parts = blocks(message)} {#if row.kind === "day"}
{@const isResultOnly = parts.every((b) => b.type === "tool_result")} <p class="label-quiet self-center pt-2">{row.day}</p>
{@const system = systemText(message)} {:else if row.kind === "system"}
{@const newDay =
message.ts && dayOf(message.ts) !== dayOf(messages[index - 1]?.ts)}
{@const newMinute =
message.ts &&
minuteOf(message.ts) !== minuteOf(messages[index - 1]?.ts)}
{@const spoken =
message.role === "user" ||
parts.some((b) => b.type === "text" && Boolean(b.text))}
{#if newDay}
<p class="label-quiet self-center pt-2">
{fmtDateTime(message.ts).replace(DAY_TIME, "")}
</p>
{/if}
{#if system}
<div class="flex min-w-0 items-start justify-center gap-2"> <div class="flex min-w-0 items-start justify-center gap-2">
{#if message.ts} {#if row.ts}
<span <span
class="tabular shrink-0 py-1 text-[11px] text-muted-foreground opacity-60" class="tabular shrink-0 py-1 text-[11px] text-muted-foreground opacity-60"
title={fmtDateTime(message.ts)} title={fmtDateTime(row.ts)}
> >
{minuteOf(message.ts)} {minuteOf(row.ts)}
</span> </span>
{/if} {/if}
<button <ServiceLine text={row.text} />
aria-expanded={openSystem.has(index)}
class="@md:max-w-[75ch] min-w-0 max-w-full rounded-md px-2 py-1 text-left font-mono text-muted-foreground text-xs hover:bg-accent"
onclick={() => toggleSystem(index)}
type="button"
>
{#if openSystem.has(index)}
<span class="whitespace-pre-wrap break-words">{system}</span>
{:else}
<span class="block truncate"
>{clip(system.split("\n")[0], 120)}</span
>
{/if}
</button>
</div> </div>
{:else if !(isResultOnly && !showResults)} {:else if row.kind === "user"}
<div <div class="group flex min-w-0 max-w-full flex-col items-end gap-1.5">
class={cn( {#if row.ts}
"group flex min-w-0 max-w-full flex-col gap-1.5",
message.role === "user" && !isResultOnly && "items-end"
)}
>
{#if message.ts && (newMinute || spoken) && !isResultOnly}
<span <span
class="tabular px-1 text-[11px] text-muted-foreground opacity-60 transition-opacity group-hover:opacity-100" class="tabular px-1 text-[11px] text-muted-foreground opacity-60 transition-opacity group-hover:opacity-100"
title={fmtDateTime(message.ts)} title={fmtDateTime(row.ts)}
> >
{minuteOf(message.ts)} {minuteOf(row.ts)}
</span>
{/if}
{#each row.texts as text, index (index)}
<Markdown
class="min-w-0 @md:max-w-[75ch] max-w-full rounded-2xl rounded-br-md bg-primary/10 px-3.5 py-2"
eager={row.eager}
inert
{text}
/>
{/each}
{#if row.service}
<ServiceLine text={row.service} />
{/if}
</div>
{:else}
<div class="group flex min-w-0 max-w-full flex-col">
{#each row.steps as step (step.key)}
{@const block = step.block}
{#if step.stampedAt}
<span
class="tabular px-1 pt-1.5 text-[11px] text-muted-foreground opacity-60 transition-opacity group-hover:opacity-100"
title={fmtDateTime(step.stampedAt)}
>
{minuteOf(step.stampedAt)}
</span> </span>
{/if} {/if}
{#each parts as block, blockIndex (blockIndex)}
{#if block.type === "text" && block.text} {#if block.type === "text" && block.text}
<Markdown <Markdown
class={cn( class="min-w-0 @md:max-w-[75ch] max-w-full px-1 py-1"
"min-w-0 @md:max-w-[75ch] max-w-full", eager={step.eager}
message.role === "user"
? "rounded-2xl rounded-br-md bg-primary/10 px-3.5 py-2"
: "px-1 py-1"
)}
eager={index >= messages.length - EAGER_MARKDOWN}
text={block.text} text={block.text}
/> />
{:else if block.type === "tool_use"} {:else if block.type === "tool_use"}
<p {@const spoken = said(block)}
class="flex min-w-0 max-w-full items-baseline gap-2 px-1 text-xs" {#if spoken}
> <div class="py-1"><SayBubble text={spoken} /></div>
<span class="shrink-0 font-medium" {:else}
>{toolLabel(block.name ?? "?")}</span <ToolNodeView node={nodeOf(block)} nodes={NO_NODES} now={0} />
> {/if}
<span class="min-w-0 truncate text-muted-foreground"> {:else if block.type === "thinking" && block.thinking?.trim()}
{clip(summarizeInput(block.name ?? "", block.input), 160)} <Reasoning text={block.thinking} />
</span>
</p>
{:else if block.type === "tool_result" && showResults}
<pre
class={cn(
"max-h-48 max-w-full overflow-auto whitespace-pre-wrap break-all rounded-md p-2 text-xs",
block.is_error
? "bg-destructive/8 text-destructive"
: "bg-muted/40"
)}
>{clip(resultText(block), RESULT_CLIP)}</pre>
{:else if block.type === "thinking"} {:else if block.type === "thinking"}
<p class="px-1 text-kind-deep text-xs">thinking</p> <p class="px-1.5 py-1 text-kind-deep text-xs">thinking</p>
{/if} {/if}
{/each} {/each}
</div> </div>
@@ -13,6 +13,7 @@
import { usePanelHost } from "./host"; import { usePanelHost } from "./host";
import MenuItems, { type MenuKit } from "./menu-items.svelte"; import MenuItems, { type MenuKit } from "./menu-items.svelte";
import RenameDialog from "./rename-dialog.svelte"; import RenameDialog from "./rename-dialog.svelte";
import RewindDialog from "./rewind-dialog.svelte";
let { let {
client, client,
@@ -47,6 +48,7 @@
let failure = $state<string | null>(null); let failure = $state<string | null>(null);
let branchOpen = $state(false); let branchOpen = $state(false);
let renameOpen = $state(false); let renameOpen = $state(false);
let rewindOpen = $state(false);
const loaded = $derived(info ?? fetched); const loaded = $derived(info ?? fetched);
@@ -77,6 +79,9 @@
onRename: () => { onRename: () => {
renameOpen = true; renameOpen = true;
}, },
onRewind: () => {
rewindOpen = true;
},
}); });
const actions = $derived(loaded ? conversationActions(loaded, scope) : []); const actions = $derived(loaded ? conversationActions(loaded, scope) : []);
</script> </script>
@@ -141,4 +146,5 @@
bind:open={branchOpen} bind:open={branchOpen}
/> />
<RenameDialog {client} info={loaded} {onChanged} bind:open={renameOpen} /> <RenameDialog {client} info={loaded} {onChanged} bind:open={renameOpen} />
<RewindDialog {client} info={loaded} {onChanged} bind:open={rewindOpen} />
{/if} {/if}
+18
View File
@@ -0,0 +1,18 @@
const FENCE_HEAD =
/^((?:\s*(?:[-+*]|\d+[.)])\s+|\s*>)*\s*)(`{3,}|~{3,})[ \t]*([^\s`]+)(.*)$/gm;
export function inertFences(
text: string,
runs: (language: string) => boolean
): string {
return text.replace(
FENCE_HEAD,
(
line: string,
lead: string,
fence: string,
language: string,
rest: string
) => (runs(language.toLowerCase()) ? `${lead}${fence}text${rest}` : line)
);
}
+5 -1
View File
@@ -11,7 +11,11 @@ export interface PanelHost {
// Where the index and the last threads persist between opens. // Where the index and the last threads persist between opens.
cache?: PanelCache; cache?: PanelCache;
// Renders ``text`` into ``node``; returns the cleanup. Absent: marked + DOMPurify. // Renders ``text`` into ``node``; returns the cleanup. Absent: marked + DOMPurify.
markdown?: (node: HTMLElement, text: string) => (() => void) | undefined; markdown?: (
node: HTMLElement,
text: string,
options?: { inert?: boolean }
) => (() => void) | undefined;
name: "browser" | "obsidian"; name: "browser" | "obsidian";
openLink: (target: string, kind: LinkKind) => void; openLink: (target: string, kind: LinkKind) => void;
// Opens the note a conversation lives in (its markdown binding). // Opens the note a conversation lives in (its markdown binding).
+8 -2
View File
@@ -7,7 +7,13 @@
text, text,
class: className = "", class: className = "",
eager = true, eager = true,
}: { text: string; class?: string; eager?: boolean } = $props(); inert = false,
}: {
text: string;
class?: string;
eager?: boolean;
inert?: boolean;
} = $props();
const NEAR = "1200px"; const NEAR = "1200px";
@@ -50,7 +56,7 @@
const now = (next: string) => { const now = (next: string) => {
dispose?.(); dispose?.();
node.classList.remove("whitespace-pre-wrap"); node.classList.remove("whitespace-pre-wrap");
dispose = host.markdown?.(node, next); dispose = host.markdown?.(node, next, { inert });
}; };
const later = (next: string) => { const later = (next: string) => {
pending = next; pending = next;
+49
View File
@@ -0,0 +1,49 @@
<script lang="ts">
import ChevronRightIcon from "@lucide/svelte/icons/chevron-right";
import { cn } from "$lib/utils";
let { text, live = false }: { text: string; live?: boolean } = $props();
const PREVIEW = 400;
const SPACES = /\s+/g;
let expanded = $state(false);
const flat = $derived(text.replace(SPACES, " ").trim());
</script>
<div class="flex min-w-0 max-w-full flex-col">
<button
aria-expanded={expanded}
class="row-hover ledger-grid grid-cols-[auto_auto_minmax(0,1fr)] rounded-md px-1.5 py-1 text-left text-sm"
onclick={() => {
expanded = !expanded;
}}
type="button"
>
<ChevronRightIcon
class={cn(
"size-3.5 text-icon transition-transform duration-150",
expanded && "rotate-90"
)}
/>
<span class="size-2 rounded-full bg-kind-deep"></span>
<span class="flex min-w-0 items-baseline gap-2">
<span class="shrink-0 font-medium text-kind-deep">reasoning</span>
{#if live && !expanded}
<span class="truncate text-left text-muted-foreground" dir="rtl">
<bdi dir="ltr">{flat.slice(-PREVIEW)}</bdi>
</span>
{:else if !expanded}
<span class="truncate text-muted-foreground"
>{flat.slice(0, PREVIEW)}</span
>
{/if}
</span>
</button>
{#if expanded}
<p
class="mb-1 ml-7 @md:max-w-[75ch] whitespace-pre-wrap break-words text-muted-foreground text-sm"
>
{text.trim()}
</p>
{/if}
</div>
+210
View File
@@ -0,0 +1,210 @@
<script lang="ts">
import { tick, untrack } from "svelte";
import { toast } from "svelte-sonner";
import type { ApiClient } from "$lib/api/client";
import type { ConversationSummary, TurnSummary } from "$lib/api/types";
import EmptyState from "$lib/components/empty-state.svelte";
import ErrorNote from "$lib/components/error-note.svelte";
import { Button } from "$lib/components/ui/button";
import * as Dialog from "$lib/components/ui/dialog";
import { Skeleton } from "$lib/components/ui/skeleton";
import { clip, fmtDateTime, fmtMinute, shortId } from "$lib/format";
import { cn } from "$lib/utils";
let {
client,
info,
open = $bindable(false),
onChanged,
}: {
client: ApiClient;
info: Pick<ConversationSummary, "id" | "kind" | "running_turn" | "title">;
open?: boolean;
onChanged: () => void;
} = $props();
const LIMIT = 30;
const LINE = 160;
let turns = $state<TurnSummary[] | null>(null);
let failure = $state<string | null>(null);
let picked = $state<number | null>(null);
let busy = $state(false);
let list = $state<HTMLDivElement | null>(null);
const name = $derived(info.title || `${info.kind} ${shortId(info.id)}`);
const count = $derived(
turns === null || picked === null ? 0 : turns.length - picked
);
function plural(n: number): string {
return n === 1 ? "1 turn" : `${n} turns`;
}
function brokenTail(rows: TurnSummary[]): number | null {
let at = rows.length;
while (at > 0 && rows[at - 1].failed) {
at -= 1;
}
return at < rows.length ? at : null;
}
async function load() {
turns = null;
failure = null;
picked = null;
try {
({ turns } = await client.turns(info.id, LIMIT));
picked = brokenTail(turns);
await tick();
if (list) {
list.scrollTop = list.scrollHeight;
list
.querySelector(":checked")
?.closest("label")
?.scrollIntoView({ block: "nearest" });
}
} catch (error) {
failure = error instanceof Error ? error.message : String(error);
}
}
async function rewind() {
const turn = turns?.[picked ?? -1];
if (!turn) {
return;
}
busy = true;
try {
const result = await client.rewind(info.id, turn.id);
open = false;
toast.success(`Cut ${plural(result.turns)}`);
onChanged();
} catch (error) {
toast.error(error instanceof Error ? error.message : String(error));
} finally {
busy = false;
}
}
$effect(() => {
if (open) {
untrack(() => load());
}
});
</script>
<Dialog.Root bind:open>
<Dialog.Content>
<Dialog.Header>
<Dialog.Title>Rewind {name}</Dialog.Title>
<Dialog.Description>
Pick the turn to cut from. The model forgets it and every turn after it.
The old session stays in the database.
</Dialog.Description>
</Dialog.Header>
<div class="flex min-w-0 flex-col gap-3">
{#if failure}
<ErrorNote message={failure} retry={load} />
{:else if turns === null}
<div class="flex flex-col gap-3 py-2">
{#each [0, 1, 2] as row (row)}
<div class="flex flex-col gap-1.5">
<Skeleton class="h-3.5 w-3/5" />
<Skeleton class="h-3 w-4/5" />
</div>
{/each}
</div>
{:else if turns.length === 0}
<EmptyState
hint="The first turn of a thread stays, and this thread has no other."
title="Nothing to cut"
/>
{:else}
<div
aria-label="Turn to cut from"
class="-mx-2 grid max-h-[min(22rem,50dvh)] grid-cols-[auto_minmax(0,1fr)] gap-x-3 overflow-y-auto rounded-md pt-2"
role="radiogroup"
bind:this={list}
>
{#each turns as turn, index (turn.id)}
{@const gone = picked !== null && index >= picked}
<label
class={cn(
"relative col-span-2 grid cursor-pointer grid-cols-subgrid items-baseline border-t px-2 py-2 first:border-t-transparent has-focus-visible:ring-1 has-focus-visible:ring-ring has-focus-visible:ring-inset",
gone ? "bg-destructive/5" : "row-hover",
index === picked &&
"border-t-destructive/40 first:border-t-destructive/40"
)}
>
<input
checked={index === picked}
class="sr-only"
name="rewind-turn"
onchange={() => {
picked = index;
}}
type="radio"
>
<span
class="tabular text-muted-foreground text-xs"
title={fmtDateTime(turn.ts)}
>
{fmtMinute(turn.ts)}
</span>
<span class="flex min-w-0 flex-col">
<span
class={cn(
"truncate font-medium",
gone && "text-muted-foreground"
)}
>
{clip(turn.text, LINE) || "empty prompt"}
</span>
<span
class={cn(
"truncate text-xs",
turn.failed ? "text-destructive" : "text-muted-foreground"
)}
>
{clip(turn.reply, LINE) ||
(turn.failed ? "failed" : "no reply")}
</span>
</span>
{#if index === picked}
<span
class="-top-2 absolute right-2 rounded-md bg-[color-mix(in_oklab,var(--destructive)_12%,var(--popover))] px-1.5 font-medium text-destructive text-xs leading-4"
>
cut from here
</span>
{/if}
</label>
{/each}
</div>
{/if}
{#if info.running_turn}
<p class="text-warn text-xs">A turn is running. Interrupt it first.</p>
{/if}
</div>
<Dialog.Footer>
<Button
onclick={() => {
// biome-ignore lint/suspicious/noGlobalAssign: bindable prop, not window.open
open = false;
}}
variant="ghost"
>
Cancel
</Button>
{#if turns?.length}
<Button
disabled={busy || count === 0 || Boolean(info.running_turn)}
onclick={rewind}
variant="destructive"
>
{count === 0 ? "Cut" : `Cut ${plural(count)}`}
</Button>
{/if}
</Dialog.Footer>
</Dialog.Content>
</Dialog.Root>
+11
View File
@@ -0,0 +1,11 @@
<script lang="ts">
import MessageSquareIcon from "@lucide/svelte/icons/message-square";
import Markdown from "./markdown.svelte";
let { text }: { text: string } = $props();
</script>
<div class="mx-1 flex gap-2 rounded-md bg-primary/8 px-2.5 py-1.5">
<MessageSquareIcon class="mt-1 size-3.5 shrink-0 text-link" />
<Markdown class="min-w-0 flex-1" {text} />
</div>
+28
View File
@@ -0,0 +1,28 @@
<script lang="ts">
import { clip } from "$lib/format";
import { cn } from "$lib/utils";
let { text, class: className = "" }: { text: string; class?: string } =
$props();
const HEAD_CLIP = 120;
let expanded = $state(false);
</script>
<button
aria-expanded={expanded}
class={cn(
"min-w-0 @md:max-w-[75ch] max-w-full rounded-md px-2 py-1 text-left font-mono text-muted-foreground text-xs hover:bg-accent",
className
)}
onclick={() => {
expanded = !expanded;
}}
type="button"
>
{#if expanded}
<span class="whitespace-pre-wrap break-words">{text}</span>
{:else}
<span class="block truncate">{clip(text.split("\n")[0], HEAD_CLIP)}</span>
{/if}
</button>
+6 -1
View File
@@ -33,6 +33,11 @@
running: "bg-signal animate-pulse-dot", running: "bg-signal animate-pulse-dot",
}; };
const SHOW_DURATION_AFTER_MS = 2000; const SHOW_DURATION_AFTER_MS = 2000;
const timed = $derived(
node.status === "running"
? duration >= SHOW_DURATION_AFTER_MS
: node.endedAt !== null
);
</script> </script>
<div class={cn("flex flex-col", depth > 0 && "ml-3 border-l pl-2")}> <div class={cn("flex flex-col", depth > 0 && "ml-3 border-l pl-2")}>
@@ -59,7 +64,7 @@
<span class="truncate text-muted-foreground">{summary}</span> <span class="truncate text-muted-foreground">{summary}</span>
{/if} {/if}
</span> </span>
{#if node.status !== "running" || duration >= SHOW_DURATION_AFTER_MS} {#if timed}
<span class="tabular text-muted-foreground text-xs"> <span class="tabular text-muted-foreground text-xs">
{fmtDuration(duration)} {fmtDuration(duration)}
</span> </span>
+8 -6
View File
@@ -1,5 +1,4 @@
<script lang="ts"> <script lang="ts">
import MessageSquareIcon from "@lucide/svelte/icons/message-square";
import StatusPill from "$lib/components/status-pill.svelte"; import StatusPill from "$lib/components/status-pill.svelte";
import { import {
elapsedMs, elapsedMs,
@@ -11,6 +10,8 @@
import { cn } from "$lib/utils"; import { cn } from "$lib/utils";
import { isLive, ORIGIN_LABELS, type Turn } from "./activity.svelte"; import { isLive, ORIGIN_LABELS, type Turn } from "./activity.svelte";
import Markdown from "./markdown.svelte"; import Markdown from "./markdown.svelte";
import Reasoning from "./reasoning.svelte";
import SayBubble from "./say-bubble.svelte";
import ToolNodeView from "./tool-node.svelte"; import ToolNodeView from "./tool-node.svelte";
let { turn, now }: { turn: Turn; now: number } = $props(); let { turn, now }: { turn: Turn; now: number } = $props();
@@ -59,7 +60,7 @@
tool {toolCount === 1 ? "call" : "calls"} tool {toolCount === 1 ? "call" : "calls"}
</span> </span>
{/if} {/if}
{#if turn.thinking > 0 && turn.status === "running"} {#if turn.thinking > 0 && turn.status === "running" && !turn.reasoning}
<span class="text-kind-deep">thinking…</span> <span class="text-kind-deep">thinking…</span>
{/if} {/if}
{#if turn.usage} {#if turn.usage}
@@ -79,9 +80,13 @@
{:else if turn.userText} {:else if turn.userText}
<Markdown <Markdown
class="mx-1 rounded-md bg-muted/50 px-2.5 py-1.5" class="mx-1 rounded-md bg-muted/50 px-2.5 py-1.5"
inert
text={turn.userText} text={turn.userText}
/> />
{/if} {/if}
{#if turn.reasoning.trim()}
<Reasoning live={turn.status === "running"} text={turn.reasoning} />
{/if}
{#if turn.roots.length > 0} {#if turn.roots.length > 0}
<div class="flex flex-col"> <div class="flex flex-col">
{#each turn.roots as rootId (rootId)} {#each turn.roots as rootId (rootId)}
@@ -92,10 +97,7 @@
</div> </div>
{/if} {/if}
{#each turn.says as text, index (index)} {#each turn.says as text, index (index)}
<div class="mx-1 flex gap-2 rounded-md bg-primary/8 px-2.5 py-1.5"> <SayBubble {text} />
<MessageSquareIcon class="mt-1 size-3.5 shrink-0 text-link" />
<Markdown class="min-w-0 flex-1" {text} />
</div>
{/each} {/each}
{#if turn.text} {#if turn.text}
<Markdown class="mx-1 px-1" text={turn.text} /> <Markdown class="mx-1 px-1" text={turn.text} />
+68
View File
@@ -0,0 +1,68 @@
<script lang="ts">
import type { ProcessRow } from "$lib/api/types";
import { Button } from "$lib/components/ui/button";
import { fmtBytes, fmtCpu, fmtSeconds } from "$lib/format";
let {
rows,
onKill,
}: { rows: ProcessRow[]; onKill: (row: ProcessRow) => void } = $props();
const INDENT_REM = 0.875;
</script>
<div class="overflow-x-auto">
<table class="w-full text-sm">
<thead class="text-muted-foreground text-xs">
<tr class="border-b text-left">
<th class="py-1.5 pr-3 font-medium">pid</th>
<th class="w-full py-1.5 pr-3 font-medium">command</th>
<th class="py-1.5 pr-3 text-right font-medium">cpu</th>
<th class="py-1.5 pr-3 text-right font-medium">rss</th>
<th class="hidden py-1.5 pr-3 text-right font-medium sm:table-cell">
age
</th>
<th class="py-1.5"><span class="sr-only">actions</span></th>
</tr>
</thead>
<tbody>
{#each rows as row (row.pid)}
<tr class="border-b last:border-b-0">
<td class="tabular py-1 pr-3 text-muted-foreground text-xs">
{row.pid}
</td>
<td class="max-w-0 py-1 pr-3">
<span
class="block truncate font-mono text-xs"
title={row.command}
style:padding-left="{row.depth * INDENT_REM}rem"
>
{row.command || row.name}
</span>
</td>
<td class="tabular py-1 pr-3 text-right text-xs">
{fmtCpu(row.cpu)}
</td>
<td class="tabular whitespace-nowrap py-1 pr-3 text-right text-xs">
{fmtBytes(row.rss)}
</td>
<td
class="tabular hidden whitespace-nowrap py-1 pr-3 text-right text-muted-foreground text-xs sm:table-cell"
>
{fmtSeconds(row.age_seconds)}
</td>
<td class="py-0.5 text-right">
<Button
class="text-muted-foreground hover:text-destructive max-sm:h-10"
onclick={() => onKill(row)}
size="xs"
variant="ghost"
>
Kill
</Button>
</td>
</tr>
{/each}
</tbody>
</table>
</div>
+150
View File
@@ -0,0 +1,150 @@
<script lang="ts">
import { onMount, untrack } from "svelte";
import { base } from "$app/paths";
import type { ApiClient } from "$lib/api/client";
import type { ProcessRow, SessionRow } from "$lib/api/types";
import { Button } from "$lib/components/ui/button";
import { shortId } from "$lib/format";
import { ConversationFeed } from "$lib/panel/conversation.svelte";
import TurnCard from "$lib/panel/turn-card.svelte";
import ProcessTable from "./process-table.svelte";
let {
client,
row,
busy,
note,
onInterrupt,
onClose,
onKill,
}: {
client: ApiClient;
row: SessionRow;
busy: boolean;
note: string;
onInterrupt: (row: SessionRow) => void;
onClose: (row: SessionRow) => void;
onKill: (process: ProcessRow) => void;
} = $props();
const TICK_MS = 1000;
const feed = new ConversationFeed(
() => client,
untrack(() => row.key)
);
let stderr = $state<string[] | null>(null);
let now = $state(Date.now());
const turn = $derived(feed.model.running);
onMount(() => {
feed.start();
const timer = setInterval(() => {
now = Date.now();
}, TICK_MS);
return () => {
feed.stop();
clearInterval(timer);
};
});
$effect(() => {
const { key, idle_seconds: polled } = row;
if (polled >= 0) {
client
.sessionStderr(key)
.then((tail) => {
stderr = tail.lines;
})
.catch(() => undefined);
}
});
</script>
<div class="flex flex-col gap-4 px-3 pt-2 pb-4">
<div class="flex flex-wrap items-center gap-x-4 gap-y-2">
<p class="tabular min-w-0 flex-1 text-muted-foreground text-xs">
session
{shortId(row.session_id)}
{#if row.pid}
· pid {row.pid}
{/if}
{#if row.pending > 0}
· {row.pending}
buffered
{/if}
</p>
<div class="flex flex-wrap items-center gap-2">
{#if note}
<span class="text-muted-foreground text-xs" role="status">{note}</span>
{/if}
{#if !feed.error}
<Button
class="max-sm:h-10"
href="{base}/conversations/{row.key}"
size="sm"
variant="ghost"
>
Open thread
</Button>
{/if}
<Button
class="max-sm:h-10"
disabled={busy || !row.busy}
onclick={() => onInterrupt(row)}
size="sm"
variant="outline"
>
Interrupt turn
</Button>
<Button
class="max-sm:h-10"
disabled={busy}
onclick={() => onClose(row)}
size="sm"
variant="destructive"
>
Close session
</Button>
</div>
</div>
<section class="flex flex-col gap-1">
<h4 class="label-quiet">Now</h4>
{#if turn}
<TurnCard {now} {turn} />
{:else}
<p class="text-muted-foreground text-sm">
{row.busy
? "A turn is running; its events have not reached this page yet."
: "Nothing in flight."}
</p>
{/if}
</section>
<section class="flex flex-col gap-1">
<h4 class="label-quiet">Process tree</h4>
{#if row.processes.length === 0}
<p class="text-muted-foreground text-sm">
The process is not visible to the gateway.
</p>
{:else}
<ProcessTable {onKill} rows={row.processes} />
{/if}
</section>
<section class="flex flex-col gap-1">
<h4 class="label-quiet">stderr</h4>
{#if stderr?.length === 0}
<p class="text-muted-foreground text-sm">
Nothing on stderr since this process started.
</p>
{:else if stderr}
<pre
class="max-h-64 overflow-auto whitespace-pre-wrap break-all rounded-md bg-muted/40 p-2 text-xs"
>{stderr.join(
"\n"
)}</pre>
{/if}
</section>
</div>
+309 -61
View File
@@ -1,16 +1,39 @@
<script lang="ts"> <script lang="ts">
import ChevronRightIcon from "@lucide/svelte/icons/chevron-right";
import { onMount } from "svelte"; import { onMount } from "svelte";
import { base } from "$app/paths"; import type {
import type { SessionsResponse } from "$lib/api/types"; ProcessRow,
SessionRow,
SessionsResponse,
} from "$lib/api/types";
import ErrorNote from "$lib/components/error-note.svelte"; import ErrorNote from "$lib/components/error-note.svelte";
import { fmtBytes, fmtSeconds, shortId } from "$lib/format"; import { Button } from "$lib/components/ui/button";
import * as Dialog from "$lib/components/ui/dialog";
import { fmtBytes, fmtCpu, fmtSeconds, shortId } from "$lib/format";
import { gateway } from "$lib/gateway.svelte";
import { session } from "$lib/session.svelte"; import { session } from "$lib/session.svelte";
import KindMark from "$lib/shell/kind-mark.svelte"; import KindMark from "$lib/shell/kind-mark.svelte";
import ProcessTable from "$lib/system/process-table.svelte";
import SessionDetail from "$lib/system/session-detail.svelte";
import { cn } from "$lib/utils"; import { cn } from "$lib/utils";
const POLL_MS = 10_000; interface Order {
confirm: string;
lines: string[];
run: () => Promise<unknown>;
title: string;
}
const POLL_MS = 3000;
const SECOND_LOOK_FOR_CPU_MS = 1000;
const COLUMNS = 8;
let sessions = $state<SessionsResponse | null>(null); let sessions = $state<SessionsResponse | null>(null);
let failure = $state<string | null>(null); let failure = $state<string | null>(null);
let unfolded = $state<Set<string>>(new Set());
let order = $state<Order | null>(null);
let orderFailure = $state<string | null>(null);
let notes = $state<Record<string, string>>({});
let busy = $state(false);
async function load() { async function load() {
const { client } = session; const { client } = session;
@@ -25,66 +48,114 @@
} }
} }
function toggle(key: string) {
const next = new Set(unfolded);
if (!next.delete(key)) {
next.add(key);
}
unfolded = next;
}
function ask(next: Order) {
orderFailure = null;
order = next;
}
async function carryOut(pending: Order) {
busy = true;
try {
await pending.run();
order = null;
await load();
} catch (cause) {
orderFailure = cause instanceof Error ? cause.message : String(cause);
} finally {
busy = false;
}
}
async function interrupt(row: SessionRow) {
const { client } = session;
if (!client) {
return;
}
busy = true;
try {
const { interrupted } = await client.interruptSession(row.key);
notes = {
...notes,
[row.key]: interrupted ? "" : "Nothing was running.",
};
await load();
} catch (cause) {
const message = cause instanceof Error ? cause.message : String(cause);
notes = { ...notes, [row.key]: message };
} finally {
busy = false;
}
}
function askClose(row: SessionRow) {
const { client } = session;
if (!client) {
return;
}
const lines = [
"The claude process exits. The conversation stays and resumes from its transcript on the next message.",
];
if (row.busy) {
lines.push("A turn is running: it is cut and its reply is lost.");
}
if (row.pinned) {
lines.push("This session is pinned: the gateway never closes it itself.");
}
ask({
confirm: "Close session",
lines,
run: () => client.closeSession(row.key),
title: `Close the ${row.kind} session?`,
});
}
function askKill(process: ProcessRow) {
const { client } = session;
if (!client) {
return;
}
const owner = sessions?.sessions.find((row) => row.pid === process.pid);
const lines = [
`SIGKILL to ${process.name} and everything it started. Nothing gets to clean up.`,
];
if (owner) {
lines.push(
"It is a session's claude: the session is dropped, and the conversation resumes from its transcript on the next message."
);
}
ask({
confirm: "Kill",
lines,
run: () => client.killProcess(process.pid),
title: `Kill process ${process.pid}?`,
});
}
const titles = $derived(
new Map(gateway.conversations.map((row) => [row.id, row.title]))
);
onMount(() => { onMount(() => {
load(); load();
const second = setTimeout(load, SECOND_LOOK_FOR_CPU_MS);
const poll = setInterval(load, POLL_MS); const poll = setInterval(load, POLL_MS);
return () => clearInterval(poll); return () => {
clearTimeout(second);
clearInterval(poll);
};
}); });
</script> </script>
<section class="flex flex-col gap-3"> {#snippet state(s: SessionRow)}
<div class="flex items-baseline gap-3"> <span class={cn(s.busy ? "text-signal" : "text-muted-foreground")}>
<h2 class="font-medium text-sm">Live claude processes</h2>
{#if sessions}
<span class="tabular text-muted-foreground text-xs">
{sessions.sessions.length}
· {fmtBytes(sessions.rss)}
{#if sessions.rss_limit}
of {fmtBytes(sessions.rss_limit)}
{/if}
</span>
{/if}
</div>
{#if failure}
<ErrorNote message={failure} retry={load} />
{:else if !sessions}
<p class="text-muted-foreground text-sm">Loading…</p>
{:else if sessions.sessions.length === 0}
<p class="text-muted-foreground text-sm">
No claude processes alive. Sessions spawn on the first turn and are reaped
by idle time or memory pressure.
</p>
{:else}
<div class="overflow-x-auto rounded-lg border bg-strip">
<table class="w-full text-sm">
<thead class="text-muted-foreground text-xs">
<tr class="border-b text-left">
<th class="px-3 py-2 font-medium">conversation</th>
<th class="px-3 py-2 font-medium">agent</th>
<th class="px-3 py-2 font-medium">state</th>
<th class="px-3 py-2 text-right font-medium">rss</th>
<th class="px-3 py-2 text-right font-medium">idle</th>
<th class="px-3 py-2 text-right font-medium">age</th>
<th class="px-3 py-2 text-right font-medium">turns</th>
</tr>
</thead>
<tbody>
{#each sessions.sessions as s (s.key)}
<tr class="row-hover border-b last:border-b-0">
<td class="px-3 py-1.5">
<a
class="flex items-center gap-2 hover:underline"
href="{base}/conversations/{s.key}"
>
<KindMark kind={s.kind} />
<span class="tabular text-xs">{shortId(s.key)}</span>
</a>
</td>
<td class="px-3 py-1.5">{s.agent}</td>
<td class="px-3 py-1.5 text-xs">
<span
class={cn(s.busy ? "text-signal" : "text-muted-foreground")}
>
{s.busy ? "busy" : "idle"} {s.busy ? "busy" : "idle"}
</span> </span>
{#if s.pinned} {#if s.pinned}
@@ -96,19 +167,196 @@
{#if s.pending_question} {#if s.pending_question}
<span class="ml-1 text-attention-foreground">question</span> <span class="ml-1 text-attention-foreground">question</span>
{/if} {/if}
{/snippet}
<div class="flex flex-wrap items-baseline gap-3">
<h2 class="font-medium text-sm">Live claude processes</h2>
{#if sessions}
<span class="tabular text-muted-foreground text-xs">
{sessions.sessions.length}
{sessions.sessions.length === 1 ? "session" : "sessions"}
· {fmtBytes(sessions.rss)}
{#if sessions.rss_limit}
of {fmtBytes(sessions.rss_limit)}
{/if}
· cpu {fmtCpu(sessions.cpu)}
</span>
{/if}
</div>
<div class="flex flex-col gap-6">
{#if failure}
<ErrorNote message={failure} retry={load} />
{:else if !sessions}
<p class="text-muted-foreground text-sm">Loading…</p>
{:else}
<section class="flex flex-col gap-2">
<h3 class="label-quiet">Sessions</h3>
{#if sessions.sessions.length === 0}
<p class="text-muted-foreground text-sm">
No claude processes alive. Sessions spawn on the first turn and are
reaped by idle time or memory pressure.
</p>
{:else}
<div class="overflow-x-auto rounded-lg border bg-strip">
<table class="w-full text-sm">
<thead class="text-muted-foreground text-xs">
<tr class="border-b text-left">
<th class="px-3 py-2 font-medium">conversation</th>
<th class="hidden px-3 py-2 font-medium sm:table-cell">
agent
</th>
<th class="hidden px-3 py-2 font-medium sm:table-cell">
state
</th>
<th class="px-3 py-2 text-right font-medium">cpu</th>
<th class="px-3 py-2 text-right font-medium">rss</th>
<th
class="hidden px-3 py-2 text-right font-medium sm:table-cell"
>
idle
</th>
<th
class="hidden px-3 py-2 text-right font-medium sm:table-cell"
>
age
</th>
<th
class="hidden px-3 py-2 text-right font-medium sm:table-cell"
>
turns
</th>
</tr>
</thead>
<tbody>
{#each sessions.sessions as s (s.key)}
{@const expanded = unfolded.has(s.key)}
<tr class={cn("border-b", !expanded && "last:border-b-0")}>
<td class="max-w-0 px-1.5 py-1 sm:w-2/5">
<button
aria-expanded={expanded}
class="row-hover flex max-w-full items-center gap-2 rounded-md px-1.5 py-1 text-left max-sm:min-h-10"
onclick={() => toggle(s.key)}
type="button"
>
<ChevronRightIcon
class={cn(
"size-3.5 text-icon transition-transform duration-150",
expanded && "rotate-90"
)}
/>
<KindMark kind={s.kind} />
<span class="flex min-w-0 flex-col">
<span class="truncate font-medium">
{titles.get(s.key) ?? shortId(s.key)}
</span>
<span class="text-xs sm:hidden">
{@render state(s)}
</span>
</span>
</button>
</td>
<td class="hidden px-3 py-1.5 sm:table-cell">{s.agent}</td>
<td
class="hidden whitespace-nowrap px-3 py-1.5 text-xs sm:table-cell"
>
{@render state(s)}
</td> </td>
<td class="tabular px-3 py-1.5 text-right">{fmtBytes(s.rss)}</td>
<td class="tabular px-3 py-1.5 text-right"> <td class="tabular px-3 py-1.5 text-right">
{fmtCpu(s.cpu)}
</td>
<td class="tabular whitespace-nowrap px-3 py-1.5 text-right">
{fmtBytes(s.rss)}
</td>
<td
class="tabular hidden px-3 py-1.5 text-right sm:table-cell"
>
{fmtSeconds(s.idle_seconds)} {fmtSeconds(s.idle_seconds)}
</td> </td>
<td class="tabular px-3 py-1.5 text-right"> <td
class="tabular hidden px-3 py-1.5 text-right sm:table-cell"
>
{fmtSeconds(s.age_seconds)} {fmtSeconds(s.age_seconds)}
</td> </td>
<td class="tabular px-3 py-1.5 text-right">{s.turns}</td> <td
class="tabular hidden px-3 py-1.5 text-right sm:table-cell"
>
{s.turns}
</td>
</tr> </tr>
{#if expanded && session.client}
<tr class="border-b last:border-b-0">
<td class="bg-background/60" colspan={COLUMNS}>
<SessionDetail
{busy}
client={session.client}
note={notes[s.key] ?? ""}
onClose={askClose}
onInterrupt={interrupt}
onKill={askKill}
row={s}
/>
</td>
</tr>
{/if}
{/each} {/each}
</tbody> </tbody>
</table> </table>
</div> </div>
{/if} {/if}
</section> </section>
<section class="flex flex-col gap-2">
<h3 class="label-quiet">Outside any session</h3>
{#if sessions.others.length === 0}
<p class="text-muted-foreground text-sm">
Nothing else runs under the gateway.
</p>
{:else}
<ProcessTable onKill={askKill} rows={sessions.others} />
{/if}
</section>
{/if}
</div>
<Dialog.Root
onOpenChange={(next) => {
if (!next) {
order = null;
}
}}
open={order !== null}
>
<Dialog.Content>
{#if order}
{@const pending = order}
<Dialog.Header>
<Dialog.Title>{pending.title}</Dialog.Title>
{#each pending.lines as line (line)}
<Dialog.Description>{line}</Dialog.Description>
{/each}
{#if orderFailure}
<p class="text-destructive text-sm" role="alert">{orderFailure}</p>
{/if}
</Dialog.Header>
<Dialog.Footer>
<Button
disabled={busy}
onclick={() => {
order = null;
}}
variant="ghost"
>
Cancel
</Button>
<Button
disabled={busy}
onclick={() => carryOut(pending)}
variant="destructive"
>
{pending.confirm}
</Button>
</Dialog.Footer>
{/if}
</Dialog.Content>
</Dialog.Root>
+426
View File
@@ -0,0 +1,426 @@
<script lang="ts">
import FileIcon from "@lucide/svelte/icons/file";
import PlusIcon from "@lucide/svelte/icons/plus";
import { toast } from "svelte-sonner";
import type { SecretRow, SecretWrite } from "$lib/api/types";
import EmptyState from "$lib/components/empty-state.svelte";
import ErrorNote from "$lib/components/error-note.svelte";
import { Button } from "$lib/components/ui/button";
import * as Dialog from "$lib/components/ui/dialog";
import { Input } from "$lib/components/ui/input";
import { Label } from "$lib/components/ui/label";
import * as Tabs from "$lib/components/ui/tabs";
import { fmtBytes, fmtDateTime, fmtRelative } from "$lib/format";
import { session } from "$lib/session.svelte";
const NAME = /^[A-Z][A-Z0-9_]{0,63}$/;
const MAX_SIZE = 1024 * 1024;
let secrets = $state<SecretRow[] | null>(null);
let enabled = $state(true);
let failure = $state<string | null>(null);
let editOpen = $state(false);
let editing = $state<SecretRow | null>(null);
let secretName = $state("");
let description = $state("");
let mode = $state<"value" | "file">("value");
let value = $state("");
let files = $state<FileList | undefined>();
let busy = $state(false);
let doomed = $state<SecretRow | null>(null);
const file = $derived(files?.[0]);
const nameError = $derived(
!editing && secretName && !NAME.test(secretName)
? "Capital letters, digits and _, starting with a letter."
: null
);
const fileError = $derived(
file && file.size > MAX_SIZE ? "Files are capped at 1 MB." : null
);
const hasPayload = $derived(mode === "value" ? value !== "" : !!file);
const canSave = $derived(
!(busy || nameError || fileError) &&
(editing ? true : NAME.test(secretName) && hasPayload)
);
async function load() {
const { client } = session;
if (!client) {
return;
}
failure = null;
try {
({ secrets, enabled } = await client.secrets());
} catch (cause) {
failure = cause instanceof Error ? cause.message : String(cause);
}
}
$effect(() => {
load();
});
function openEditor(row: SecretRow | null) {
editing = row;
secretName = row?.name ?? "";
description = row?.description ?? "";
mode = row?.kind ?? "value";
value = "";
files = undefined;
editOpen = true;
}
function base64Of(blob: File): Promise<string> {
return new Promise((resolve, reject) => {
const reader = new FileReader();
reader.onload = () => {
const url = String(reader.result);
resolve(url.slice(url.indexOf(",") + 1));
};
reader.onerror = () => reject(reader.error);
reader.readAsDataURL(blob);
});
}
async function save() {
const { client } = session;
if (!client) {
return;
}
busy = true;
try {
const body: SecretWrite = { description };
if (mode === "value" && value) {
body.value = value;
}
if (mode === "file" && file) {
body.file = { content: await base64Of(file), name: file.name };
}
const target = editing?.name ?? secretName;
await client.putSecret(target, body);
toast.success(editing ? `Updated ${target}` : `Saved ${target}`);
editOpen = false;
value = "";
files = undefined;
await load();
} catch (cause) {
toast.error(cause instanceof Error ? cause.message : String(cause));
} finally {
busy = false;
}
}
async function remove(row: SecretRow) {
const { client } = session;
if (!client) {
return;
}
busy = true;
try {
await client.deleteSecret(row.name);
toast.success(`Deleted ${row.name}`);
doomed = null;
await load();
} catch (cause) {
toast.error(cause instanceof Error ? cause.message : String(cause));
} finally {
busy = false;
}
}
</script>
<div class="flex flex-wrap items-center gap-4">
<h2 class="font-medium text-sm">Secrets</h2>
<span class="text-muted-foreground text-xs">
write-only · the agent runs
<code class="rounded bg-muted px-1 py-0.5">secret run NAME -- cmd</code>
</span>
<Button
class="ml-auto"
disabled={!enabled}
onclick={() => openEditor(null)}
size="sm"
variant="outline"
>
<PlusIcon class="size-4" />
New secret
</Button>
</div>
<div class="flex flex-col gap-6">
{#if !enabled}
<div
class="flex flex-col gap-1 rounded-lg border border-warn/40 bg-warn/5 p-3 text-sm"
role="status"
>
<p class="font-medium">Secrets are off: SECRETS_KEY is not set.</p>
<p class="text-muted-foreground">
Put a key in the gateway's environment (<code
class="rounded bg-muted px-1 py-0.5"
>openssl rand -hex 32</code
>) and recreate the container. Existing names stay listed; values cannot
be set or read until then.
</p>
</div>
{/if}
{#if failure}
<ErrorNote message={failure} retry={() => load()} />
{:else if secrets === null}
<p class="text-muted-foreground text-sm">Loading…</p>
{:else if secrets.length === 0}
<EmptyState
hint="Keys and files the agent runs commands with. They are sealed with SECRETS_KEY in the gateway's database, reach a command as environment variables, and are never shown again after saving."
title="No secrets"
>
{#if enabled}
<Button onclick={() => openEditor(null)} size="sm" variant="outline">
Add the first secret
</Button>
{/if}
</EmptyState>
{:else}
<div class="overflow-x-auto rounded-lg border bg-strip">
<table class="w-full text-sm">
<thead class="text-muted-foreground text-xs">
<tr class="border-b text-left">
<th class="px-3 py-2 font-medium">name</th>
<th class="hidden px-3 py-2 font-medium sm:table-cell">
description
</th>
<th class="hidden px-3 py-2 font-medium sm:table-cell">kind</th>
<th class="hidden px-3 py-2 font-medium sm:table-cell">changed</th>
<th class="hidden px-3 py-2 font-medium sm:table-cell">
last used
</th>
<th class="px-3 py-2 text-right font-medium"></th>
</tr>
</thead>
<tbody>
{#each secrets as row (row.name)}
<tr class="row-hover border-b last:border-b-0">
<td class="px-3 py-1.5">
<code class="break-all font-medium">{row.name}</code>
<div
class="flex flex-col gap-0.5 pt-0.5 text-muted-foreground text-xs sm:hidden"
>
{#if row.description}
<span class="text-foreground/80">{row.description}</span>
{/if}
<span class="tabular">
{row.filename
? `${row.filename} · ${fmtBytes(row.size)}`
: "value"}
· changed {fmtRelative(row.updated_at)} ·
{row.used_at
? `used ${fmtRelative(row.used_at)}`
: "never used"}
</span>
</div>
</td>
<td class="hidden min-w-48 px-3 py-1.5 sm:table-cell">
{#if row.description}
{row.description}
{:else}
<span class="text-muted-foreground">–</span>
{/if}
</td>
<td
class="hidden whitespace-nowrap px-3 py-1.5 text-xs sm:table-cell"
>
{#if row.filename}
<span class="inline-flex items-center gap-1">
<FileIcon class="size-3.5 text-icon" />
{row.filename}
<span class="tabular text-muted-foreground"
>{fmtBytes(row.size)}</span
>
</span>
{:else}
<span class="text-muted-foreground">value</span>
{/if}
</td>
<td
class="tabular hidden whitespace-nowrap px-3 py-1.5 text-xs sm:table-cell"
title={fmtDateTime(row.updated_at)}
>
{fmtRelative(row.updated_at)}
</td>
<td
class="tabular hidden whitespace-nowrap px-3 py-1.5 text-xs sm:table-cell"
title={row.used_at ? fmtDateTime(row.used_at) : undefined}
>
{#if row.used_at}
{fmtRelative(row.used_at)}
{:else}
<span class="text-muted-foreground">never</span>
{/if}
</td>
<td
class="flex flex-col items-end px-1 py-1.5 sm:table-cell sm:whitespace-nowrap sm:px-3 sm:text-right"
>
<Button
disabled={!enabled}
onclick={() => openEditor(row)}
size="sm"
variant="ghost"
>
Replace
</Button>
<Button
class="text-destructive hover:text-destructive"
onclick={() => {
doomed = row;
}}
size="sm"
variant="ghost"
>
Delete
</Button>
</td>
</tr>
{/each}
</tbody>
</table>
</div>
{/if}
</div>
<Dialog.Root bind:open={editOpen}>
<Dialog.Content>
<Dialog.Header>
<Dialog.Title>
{editing ? `Replace ${editing.name}` : "New secret"}
</Dialog.Title>
<Dialog.Description>
{editing
? "Leave the value empty to change only the description."
: "Stored encrypted. After saving, the value is not shown anywhere."}
</Dialog.Description>
</Dialog.Header>
<form
autocomplete="off"
class="flex flex-col gap-3"
onsubmit={(event) => {
event.preventDefault();
save();
}}
>
{#if !editing}
<div class="flex flex-col gap-1.5">
<Label for="secret-name">Name</Label>
<Input
aria-invalid={nameError ? true : undefined}
autocapitalize="characters"
class="font-mono"
id="secret-name"
oninput={() => {
secretName = secretName.toUpperCase().replaceAll(/[\s-]/g, "_");
}}
placeholder="NODEREAL_BSC"
required
spellcheck={false}
bind:value={secretName}
/>
<p class="text-muted-foreground text-xs">
{nameError ?? `The command reads it as $${secretName || "NAME"}.`}
</p>
</div>
{/if}
<div class="flex flex-col gap-1.5">
<Label for="secret-description">Description</Label>
<Input
id="secret-description"
placeholder="What it opens and who issued it"
bind:value={description}
/>
</div>
<Tabs.Root bind:value={mode}>
<Tabs.List class="w-full">
<Tabs.Trigger value="value">Value</Tabs.Trigger>
<Tabs.Trigger value="file">File</Tabs.Trigger>
</Tabs.List>
<Tabs.Content class="flex flex-col gap-1.5 pt-2" value="value">
<Label class="sr-only" for="secret-value">Value</Label>
<Input
autocomplete="new-password"
class="font-mono"
id="secret-value"
placeholder={editing ? "New value" : "Paste the value"}
spellcheck={false}
type="password"
bind:value
/>
<p class="text-muted-foreground text-xs">
One line. Multi-line keys and JSON go in as a file.
</p>
</Tabs.Content>
<Tabs.Content class="flex flex-col gap-1.5 pt-2" value="file">
<Label class="sr-only" for="secret-file">File</Label>
<Input
aria-invalid={fileError ? true : undefined}
id="secret-file"
type="file"
bind:files
/>
<p class="text-muted-foreground text-xs">
{fileError ??
"Up to 1 MB. The command gets $NAME as the path to a private copy, removed when it exits."}
</p>
</Tabs.Content>
</Tabs.Root>
<Dialog.Footer>
<Button
onclick={() => {
editOpen = false;
}}
type="button"
variant="ghost"
>
Cancel
</Button>
<Button disabled={!canSave} type="submit">
{editing && !hasPayload ? "Save description" : "Save"}
</Button>
</Dialog.Footer>
</form>
</Dialog.Content>
</Dialog.Root>
<Dialog.Root
onOpenChange={(open) => {
if (!open) {
doomed = null;
}
}}
open={doomed !== null}
>
<Dialog.Content>
<Dialog.Header>
<Dialog.Title>Delete {doomed?.name}?</Dialog.Title>
<Dialog.Description>
Commands that ask for it will fail with “unknown secret”. The value
cannot be recovered.
</Dialog.Description>
</Dialog.Header>
<Dialog.Footer>
<Button
onclick={() => {
doomed = null;
}}
variant="ghost"
>
Cancel
</Button>
<Button
disabled={busy}
onclick={() => doomed && remove(doomed)}
variant="destructive"
>
Delete
</Button>
</Dialog.Footer>
</Dialog.Content>
</Dialog.Root>
Generated
+9 -8
View File
@@ -296,6 +296,7 @@ dependencies = [
{ name = "argon2-cffi" }, { name = "argon2-cffi" },
{ name = "claude-agent-sdk" }, { name = "claude-agent-sdk" },
{ name = "croniter" }, { name = "croniter" },
{ name = "cryptography" },
{ name = "fastapi" }, { name = "fastapi" },
{ name = "fastmcp" }, { name = "fastmcp" },
{ name = "greenlet" }, { name = "greenlet" },
@@ -338,8 +339,9 @@ requires-dist = [
{ name = "anthropic", specifier = ">=0.103.0" }, { name = "anthropic", specifier = ">=0.103.0" },
{ name = "anyio", specifier = ">=4.13.0" }, { name = "anyio", specifier = ">=4.13.0" },
{ name = "argon2-cffi", specifier = ">=25.1.0" }, { name = "argon2-cffi", specifier = ">=25.1.0" },
{ name = "claude-agent-sdk", specifier = ">=0.2.150" }, { name = "claude-agent-sdk", specifier = ">=0.2.157" },
{ name = "croniter", specifier = ">=6.2.4" }, { name = "croniter", specifier = ">=6.2.4" },
{ name = "cryptography", specifier = ">=48.0.0" },
{ name = "fastapi", specifier = ">=0.136.1" }, { name = "fastapi", specifier = ">=0.136.1" },
{ name = "fastmcp", specifier = ">=3.3.1" }, { name = "fastmcp", specifier = ">=3.3.1" },
{ name = "greenlet", specifier = ">=3.5.0" }, { name = "greenlet", specifier = ">=3.5.0" },
@@ -450,7 +452,7 @@ wheels = [
[[package]] [[package]]
name = "claude-agent-sdk" name = "claude-agent-sdk"
version = "0.2.150" version = "0.2.157"
source = { registry = "https://pypi.org/simple" } source = { registry = "https://pypi.org/simple" }
dependencies = [ dependencies = [
{ name = "anyio" }, { name = "anyio" },
@@ -458,13 +460,12 @@ dependencies = [
{ name = "mcp" }, { name = "mcp" },
{ name = "sniffio" }, { name = "sniffio" },
] ]
sdist = { url = "https://files.pythonhosted.org/packages/6d/1e/99bababecf24a4a55a0775665ff0be010aff26f653fc9be2f4f17a0b5ef0/claude_agent_sdk-0.2.150.tar.gz", hash = "sha256:3856e932f8d45514ddf5ddd540fb844678bb11638cabf9b939ce8b708e4eb352", size = 344692, upload-time = "2026-09-01T18:07:15.284Z" } sdist = { url = "https://files.pythonhosted.org/packages/b4/e3/444df1691c864a8349067686719b43e714fc43d3853fb1b2ebc2097e2a6e/claude_agent_sdk-0.2.157.tar.gz", hash = "sha256:4ce6f41bc965fcb4a9d11bd7a3cac266975c75b632f3d74a26f55f0fffb3160a", size = 346579, upload-time = "2026-09-18T18:21:10.306Z" }
wheels = [ wheels = [
{ url = "https://files.pythonhosted.org/packages/c3/13/b5b86d90c95e8a691143eeb45bbf2e1c250324817980464d6ea63ef0f789/claude_agent_sdk-0.2.150-py3-none-macosx_11_0_arm64.whl", hash = "sha256:8f58d746f3552faeac52883b5a502a8d9b5aef067ea0b68fae5c9bae0f39d73d", size = 85396845, upload-time = "2026-09-01T18:07:19.074Z" }, { url = "https://files.pythonhosted.org/packages/ec/e8/3e9cc23eb5a5a01a0e2ee931a17db2785f95407525d50366af06a4a5469c/claude_agent_sdk-0.2.157-py3-none-macosx_11_0_arm64.whl", hash = "sha256:dd6a2c7aa1901e43b827411d62f2327aed82f53ed8ca6b076eaa5be4a1fc681f", size = 92744001, upload-time = "2026-09-18T18:21:15.969Z" },
{ url = "https://files.pythonhosted.org/packages/d9/8f/c17da4fa9cd8bddb8ac1538456dd0fde3a43abb1dfc28ca448f85e2a10d3/claude_agent_sdk-0.2.150-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:fee3ae2dd457742d82070b5fa122070ee8e2e80b4f9b37a57ff0a8261ade656d", size = 89904483, upload-time = "2026-09-01T18:07:22.671Z" }, { url = "https://files.pythonhosted.org/packages/49/eb/e5095703a8efbe504ec468cab039543caabfc0ea3327c735a3d1b60e57ac/claude_agent_sdk-0.2.157-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:ebf3f514c61db9be01bbf6d1acb0d993247340583b3559d2df75e3822f618bbb", size = 97352326, upload-time = "2026-09-18T18:21:21.007Z" },
{ url = "https://files.pythonhosted.org/packages/8d/95/5b33717b807c6fbe29fca55a246eab9a09a80c9dd288ffeb31f05681b7fc/claude_agent_sdk-0.2.150-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:813535a0267271d91af29c456ec26137b34a8b7bc066647ea59efba6e3c26c93", size = 95270310, upload-time = "2026-09-01T18:07:26.61Z" }, { url = "https://files.pythonhosted.org/packages/d6/f4/64505beb3c757b863753d6556539554704ec9168cb96454dc6463b14df85/claude_agent_sdk-0.2.157-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:a58089e0c11364db2ec7a139e09b36a4fb9c95817a758bee7b28e0434f55c0b6", size = 102701030, upload-time = "2026-09-18T18:21:27.065Z" },
{ url = "https://files.pythonhosted.org/packages/9f/de/5a8b59e6a713ea3f4be88816cfd487b1a5b08c8ad9f042abe56d2decc184/claude_agent_sdk-0.2.150-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:cb65ff33060baa098b420232c630fa287d0f95e9469e7ceb706d0b7d8841888e", size = 95460277, upload-time = "2026-09-01T18:07:31.092Z" }, { url = "https://files.pythonhosted.org/packages/03/8f/2a7c121469fd16d1d6465bec180726c54f6e151b12e50835283b4285d1ce/claude_agent_sdk-0.2.157-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:bd126b2edee3a247b2c7c4cee1f5a3ea68ef748288ec0650db88dd3918e5471a", size = 102830653, upload-time = "2026-09-18T18:21:32.948Z" },
{ url = "https://files.pythonhosted.org/packages/dd/42/d147578a0df16c1b4b0deaebfb3d25e590ad5c16cb6b59a68f9dbb618be3/claude_agent_sdk-0.2.150-py3-none-win_amd64.whl", hash = "sha256:3b517a21c723854de778b38fa51660c5b68ee5fb6f28b2abf291a51471a7b378", size = 98078644, upload-time = "2026-09-01T18:07:35.734Z" },
] ]
[[package]] [[package]]