diff --git a/src/claude_code_api/backend.py b/src/claude_code_api/backend.py index 878e454..c66d902 100644 --- a/src/claude_code_api/backend.py +++ b/src/claude_code_api/backend.py @@ -136,6 +136,7 @@ class ClaudeCodeBackend: self._opts = options self._on_parse_error = on_parse_error self._sessions: dict[str, _LiveSession] = {} + self._active: dict[str, _LiveSession] = {} self._mcp_config_path: Path | None = None self._session_factory = _session_factory self._closed = False @@ -158,8 +159,15 @@ class ClaudeCodeBackend: ``_sessions`` (which is keyed by history fingerprint, not ``session_id``). Intended for debug surfaces (e.g. admin terminal viewer) that need to look up a session by id. + + Covers both idle pooled sessions (``_sessions``) and the turn(s) + currently running (``_active``) — so a live terminal is visible + *during* its turn, not only after it's repooled. ``_active`` wins + on key collision, but a session is never in both at once. """ - return {s.session_id: s.pty for s in self._sessions.values()} + out = {s.session_id: s.pty for s in self._sessions.values()} + out.update({sid: s.pty for sid, s in self._active.items()}) + return out async def complete(self, messages: list[Mapping[str, Any]]) -> AsyncIterator[Event]: """Run one turn against the matching session (or spawn one). @@ -223,6 +231,7 @@ class ClaudeCodeBackend: n_assistant = 0 n_user = 0 n_system = 0 + self._active[session.session_id] = session try: _log.info( "complete: sending user msg to session_id=%s (text_len=%d)", @@ -250,6 +259,8 @@ class ClaudeCodeBackend: with contextlib.suppress(Exception): await session.aclose() raise + finally: + self._active.pop(session.session_id, None) synthesized_cycle = synthesize_turn_messages(events) new_history = [*list(messages), *synthesized_cycle] diff --git a/tests/test_backend.py b/tests/test_backend.py index 89073c7..fed8ee9 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -234,6 +234,48 @@ async def test_complete_fresh_session_yields_events(tmp_path: Path) -> None: assert harness.seed_files == [] +@pytest.mark.asyncio +async def test_live_sessions_visible_during_turn_and_after(tmp_path: Path) -> None: + """A session must appear in ``live_sessions`` *while* its turn runs — so + the admin terminal viewer can attach mid-flight, not only once the turn + is done — and must stay visible as an idle pooled session afterwards so + snapshots / interaction with a quiet terminal keep working. + + Regression guard: the session is popped from the fingerprint pool (or + not yet pooled, on a fresh spawn) for the duration of the turn, so + without the in-flight ``_active`` registry an active session is + invisible to ``live_sessions``. + """ + scripts_per_session = [ + [ + [_user_rec("hi", "S0"), _assistant_rec("hello there", "S0")], + ], + ] + harness = FakeFactoryHarness(scripts_per_session) + backend = ClaudeCodeBackend( + BackendOptions(cwd=str(tmp_path)), + _session_factory=harness, + ) + + seen_live_mid = False + async for _event in backend.complete([{"role": "user", "content": "hi"}]): + live = backend.live_sessions + if live: + seen_live_mid = True + # The only live pty is the one the harness spawned for this turn. + assert list(live.values()) == [harness.spawned[0]] + + # The regression we're guarding against: visible *during* the turn. + assert seen_live_mid + # Still visible once the turn finishes — repooled idle session, so the + # viewer can snapshot / interact exactly as before. + assert list(backend.live_sessions.values()) == [harness.spawned[0]] + + await backend.aclose() + # Teardown clears the pool. + assert backend.live_sessions == {} + + # --- multi-turn fingerprint reuse ----------------------------------------