fix(conversation_store,admin,cli): keep session when file lags db, sort pty by recency

This commit is contained in:
hh
2026-07-26 00:25:07 +02:00
parent d2981116d4
commit 90fe92d014
7 changed files with 113 additions and 7 deletions
+10
View File
@@ -171,6 +171,16 @@ class ClaudeCodeOptions(BaseModel):
terminal assistant has been seen. Bound on extra latency when terminal assistant has been seen. Bound on extra latency when
``wait_for_turn_duration=True``.""" ``wait_for_turn_duration=True``."""
idle_session_ttl: float = 1800.0
"""Seconds a pooled claude session may sit unused before it is
terminated. The backend pools live PTYs by conversation-history
fingerprint; when a conversation forks, the session behind the old
fingerprint becomes unreachable but stays resident — one live
``claude`` (hundreds of MB) each. 30 minutes keeps an active chat
warm across a coffee break while bounding what a fork storm can
strand. There is no count cap: concurrent conversations should all
stay warm. ``0`` disables reaping."""
class ClaudeAgent(BaseAgent): class ClaudeAgent(BaseAgent):
"""Agent backed by ``claude-code-api``. """Agent backed by ``claude-code-api``.
@@ -199,6 +199,7 @@ def _build_backend_options(
startup_delay=opt.startup_delay, startup_delay=opt.startup_delay,
file_wait_timeout=opt.file_wait_timeout, file_wait_timeout=opt.file_wait_timeout,
turn_duration_timeout=opt.turn_duration_timeout, turn_duration_timeout=opt.turn_duration_timeout,
idle_session_ttl=opt.idle_session_ttl,
) )
+26
View File
@@ -21,7 +21,9 @@ it just keeps the aggregator running for anyone who needs it.
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import contextlib
import logging import logging
import signal
from contextlib import AsyncExitStack from contextlib import AsyncExitStack
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
@@ -60,9 +62,33 @@ def main() -> None:
logging.basicConfig( logging.basicConfig(
level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s" level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s"
) )
_install_sigterm_handler()
asyncio.run(_async_main(), loop_factory=uvloop.new_event_loop) asyncio.run(_async_main(), loop_factory=uvloop.new_event_loop)
def _install_sigterm_handler() -> None:
"""Turn SIGTERM into a normal interpreter exit.
Python's default SIGTERM disposition kills the process outright, so
neither ``AsyncExitStack`` unwinding nor ``atexit`` hooks run. That
matters because every ``claude`` we spawn lives in its own session
(ptyprocess calls ``setsid``), which makes it immune to the signal
that took us down — a hard SIGTERM leaves one orphaned CLI per live
session, each holding hundreds of MB. Raising ``KeyboardInterrupt``
instead routes ``docker stop`` / ``systemctl stop`` through the same
shutdown path as Ctrl-C, which does reap them.
Only installed when we own the main thread's signal handlers; under
an embedding host that isn't ours to take.
"""
def _raise_interrupt(_signum: int, _frame: object) -> None:
raise KeyboardInterrupt
with contextlib.suppress(ValueError, OSError):
signal.signal(signal.SIGTERM, _raise_interrupt)
async def _async_main() -> None: async def _async_main() -> None:
# Populate ``os.environ`` from ``.env`` before anything else so the # Populate ``os.environ`` from ``.env`` before anything else so the
# user's ``config.py`` can read its own secrets via ``os.environ[...]`` # user's ``config.py`` can read its own secrets via ``os.environ[...]``
@@ -398,6 +398,15 @@ def diff_and_fork(
spliced_groups, divergence = _walk_prefix(prior_incoming, stored_groups) spliced_groups, divergence = _walk_prefix(prior_incoming, stored_groups)
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):
# Not a deletion — the file simply never received turns we
# already ran. Adopt the stored tail verbatim so history stays
# structured and its fingerprint still matches the live
# session's. See ``_file_lags_store`` for why this matters.
spliced_groups.extend(
list(g.messages) for g in stored_groups[len(prior_incoming) :]
)
else:
# Incoming truncated stored (user deleted some past turns). # Incoming truncated stored (user deleted some past turns).
# Truncate stored to match. # Truncate stored to match.
divergence = len(prior_incoming) divergence = len(prior_incoming)
@@ -415,6 +424,38 @@ def diff_and_fork(
) )
def _file_lags_store(
stored_groups: list[_StoredDisplayTurn], prior_len: int, new_user_turn: ParsedTurn
) -> bool:
"""Is the shorter incoming file a stale view rather than a deletion?
A file with fewer display turns than the DB has two possible causes,
and they need opposite handling:
* the user deleted trailing turns — we should truncate to match;
* the turn ran, was persisted, but its reply never made it back into
the ``.md`` (the render lost a race with the user's next prompt, or
the reply rendered to nothing visible). The file is simply behind.
The tell is the prompt the user is submitting right now: if the DB
already holds it at exactly the position the file stops at, this is a
re-submission of a turn we've already run, not a deletion. Nobody
deletes a turn and immediately retypes it verbatim.
Getting this wrong is expensive and self-sustaining. Forking here
flattens every post-divergence turn into plain text (losing tool_use /
tool_result structure), persists that flattened history, and changes
the conversation fingerprint — so the backend's session pool misses,
spawns a fresh ``claude``, reseeds it from a multi-MB JSONL, and
strands the previous process. The file still lags afterwards, so the
next turn does it again.
"""
if prior_len >= len(stored_groups):
return False
candidate = stored_groups[prior_len]
return candidate.role == "user" and candidate.spoken_text == new_user_turn.text
def _walk_prefix( def _walk_prefix(
prior_incoming: list[ParsedTurn], stored_groups: list[_StoredDisplayTurn] prior_incoming: list[ParsedTurn], stored_groups: list[_StoredDisplayTurn]
) -> tuple[list[list[dict[str, Any]]], int | None]: ) -> tuple[list[list[dict[str, Any]]], int | None]:
@@ -567,13 +567,19 @@ def _collect_pty_sessions(runtime: GatewayRuntime) -> list[dict[str, Any]]:
(currently only ``ClaudeCodeBackendAdapter``). Other backend types (currently only ``ClaudeCodeBackendAdapter``). Other backend types
are quietly skipped — the admin terminal viewer only makes sense for are quietly skipped — the admin terminal viewer only makes sense for
PTY-backed agents. PTY-backed agents.
Sorted most-recently-used first. The underlying mapping is keyed by
conversation fingerprint and reshuffled on every pool hit, so its
iteration order carries no meaning a human could act on.
""" """
now = time.time()
out: list[dict[str, Any]] = [] out: list[dict[str, Any]] = []
for agent_name, backend in runtime.backends.items(): for agent_name, backend in runtime.backends.items():
live = getattr(backend, "live_sessions", None) live = getattr(backend, "live_sessions", None)
if not isinstance(live, dict): if not isinstance(live, dict):
continue continue
for session_id, pty in live.items(): for session_id, pty in live.items():
last_used = getattr(pty, "last_activity_at", None) or now
out.append( out.append(
{ {
"agent": agent_name, "agent": agent_name,
@@ -582,11 +588,28 @@ def _collect_pty_sessions(runtime: GatewayRuntime) -> list[dict[str, Any]]:
"buffer_size": len(pty.captured_output()) "buffer_size": len(pty.captured_output())
if hasattr(pty, "captured_output") if hasattr(pty, "captured_output")
else 0, else 0,
"last_used": last_used,
"idle": _humanize_duration(now - last_used),
"age": _humanize_duration(
now - (getattr(pty, "created_at", None) or now)
),
} }
) )
out.sort(key=lambda s: s["last_used"], reverse=True)
return out return out
def _humanize_duration(seconds: float) -> str:
"""Render a duration as the coarsest unit that still reads precisely."""
secs = max(0, int(seconds))
if secs < 60:
return f"{secs}s"
if secs < 3600:
return f"{secs // 60}m {secs % 60}s"
hours, rem = divmod(secs, 3600)
return f"{hours}h {rem // 60}m"
def _find_pty(runtime: GatewayRuntime, session_id: str) -> Any: def _find_pty(runtime: GatewayRuntime, session_id: str) -> Any:
"""Locate a live PTY by its claude session_id, across all backends.""" """Locate a live PTY by its claude session_id, across all backends."""
for backend in runtime.backends.values(): for backend in runtime.backends.values():
@@ -4,8 +4,9 @@
{% block content %} {% block content %}
<h2>Live PTY sessions</h2> <h2>Live PTY sessions</h2>
<p class="muted"> <p class="muted">
Each row is a running <code>claude</code> subprocess. Open one to see what's Each row is a running <code>claude</code> subprocess, most recently used first.
currently rendered on its TUI and (if needed) type into it directly. Open one to see what's currently rendered on its TUI and (if needed) type into
it directly. Sessions idle past the agent's TTL are terminated automatically.
</p> </p>
{% if sessions %} {% if sessions %}
@@ -15,6 +16,8 @@
<tr> <tr>
<th>Session ID</th> <th>Session ID</th>
<th>Agent</th> <th>Agent</th>
<th>Idle</th>
<th>Age</th>
<th>PID</th> <th>PID</th>
<th>Buffer</th> <th>Buffer</th>
<th></th> <th></th>
@@ -25,6 +28,8 @@
<tr> <tr>
<td><code>{{ s.session_id }}</code></td> <td><code>{{ s.session_id }}</code></td>
<td>{{ s.agent }}</td> <td>{{ s.agent }}</td>
<td>{{ s.idle }}</td>
<td class="muted">{{ s.age }}</td>
<td><span class="pill">{{ s.pid or "?" }}</span></td> <td><span class="pill">{{ s.pid or "?" }}</span></td>
<td>{{ s.buffer_size }} bytes</td> <td>{{ s.buffer_size }} bytes</td>
<td> <td>
Generated
+2 -2
View File
@@ -287,7 +287,7 @@ local = [
{ name = "raycast-api", version = "0.1.0", source = { editable = "../raycast-api" } }, { name = "raycast-api", version = "0.1.0", source = { editable = "../raycast-api" } },
] ]
prod = [ prod = [
{ name = "claude-code-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#4f6585b1ac16746de3724c36b5211a24181fd948" } }, { name = "claude-code-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#89065c2f4ed8db5607ae49e236a8e398719c8072" } },
{ name = "raycast-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/raycast-api.git#e73894c8e435da5c0709f92f69f11bcd0dab9afe" } }, { name = "raycast-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/raycast-api.git#e73894c8e435da5c0709f92f69f11bcd0dab9afe" } },
] ]
@@ -419,7 +419,7 @@ wheels = [
[[package]] [[package]]
name = "claude-code-api" name = "claude-code-api"
version = "0.1.0" version = "0.1.0"
source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#4f6585b1ac16746de3724c36b5211a24181fd948" } source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#89065c2f4ed8db5607ae49e236a8e398719c8072" }
resolution-markers = [ resolution-markers = [
"python_full_version >= '3.14'", "python_full_version >= '3.14'",
"python_full_version < '3.14'", "python_full_version < '3.14'",