Compare commits

...
5 Commits
32 changed files with 1606 additions and 290 deletions

No files matched your search

+24 -12
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 (
@@ -262,12 +269,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 +676,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 +696,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 +705,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
@@ -747,11 +756,14 @@ 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( stderr=lambda line: self._stderr(line.rstrip(), stderr),
"claude[%s]: %s", agent.name, line.rstrip()
),
) )
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[
+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:
+6 -4
View File
@@ -164,9 +164,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:
@@ -189,7 +189,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
@@ -205,7 +206,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
+25 -2
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,
@@ -482,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:
+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() :]
+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")
+64 -4
View File
@@ -55,6 +55,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
@@ -633,15 +634,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]:
@@ -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}"
@@ -994,10 +994,13 @@ class TelegramFrontend(Frontend):
draft.append(progress) draft.append(progress)
else: 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)
+43 -3
View File
@@ -13,6 +13,7 @@ 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
@@ -30,6 +31,8 @@ __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:
@@ -44,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."""
@@ -140,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)
) )
@@ -158,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,
@@ -186,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()
@@ -216,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] = []
+48
View File
@@ -396,6 +396,54 @@ 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_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")
+20
View File
@@ -1052,3 +1052,23 @@ def test_reply_text_keeps_progress_updates_and_drops_hidden_reasoning() -> None:
def test_reply_text_on_a_plain_answer_is_unchanged() -> None: def test_reply_text_on_a_plain_answer_is_unchanged() -> None:
blocks = [AnthropicTextBlock(type="text", text=" just an answer ")] blocks = [AnthropicTextBlock(type="text", text=" just an answer ")]
assert reply_text(blocks, thinking_is_progress=True) == "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:
+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([]) == ""
+38 -6
View File
@@ -1228,18 +1228,50 @@ def test_progress_updates_reach_the_draft_but_hidden_reasoning_does_not() -> Non
assert draft.status == texts.writing assert draft.status == texts.writing
def test_reasoning_from_an_agent_whose_thinking_is_not_progress_stays_hidden() -> None: def test_reasoning_rides_under_the_status_and_never_reaches_the_text() -> None:
draft = _progress_draft() draft = _progress_draft()
texts = TelegramTexts() texts = TelegramTexts()
stub = SimpleNamespace(_drafts={"k": draft}, texts=texts) stub = SimpleNamespace(_drafts={"k": draft}, texts=texts)
TelegramFrontend._feed_draft( feed = functools.partial(
cast("Any", stub), TelegramFrontend._feed_draft, cast("Any", stub), "k", thinking_is_progress=False
"k",
_thinking_delta("weighing it"),
thinking_is_progress=False,
) )
feed(_thinking_delta("weighing <it> "))
feed(_thinking_delta("against the note"))
assert draft.text == "" assert draft.text == ""
assert draft.status == texts.thinking 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: def test_a_tool_call_between_progress_updates_starts_a_new_paragraph() -> None:
+16
View File
@@ -313,6 +313,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");
} }
+18
View File
@@ -106,7 +106,9 @@ 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;
} }
@@ -144,17 +146,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;
@@ -162,6 +178,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[];
+6
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 "–";
+32 -9
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,27 +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;
const progress = this.conversation?.thinking_is_progress === true;
const text =
delta?.type === "thinking_delta" && progress
? delta.thinking
: delta?.text;
if (typeof text === "string" && text) { if (typeof text === "string" && text) {
turn.text = joinText(turn.text, text, turn.paused); turn.text = joinText(turn.text, 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")) {
@@ -365,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: [],
+239 -155
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,36 +190,13 @@
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[] {
@@ -222,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}
@@ -256,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 parts.length > 0 && !(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>
+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>
+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>
+310 -62
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>