chat: persistent Agent SDK session per chat — decouple worker from turn (Phase 1)

Root fix for orphaned background tasks: the platform drove Claude Code as a
one-shot `claude -p` per turn (stdin ignored, process exits at turn end), so
run_in_background/Monitor work — and its task_notification — had no live harness
to return to. Now each chat session runs ONE long-lived Agent SDK query() with
streaming input; turns are user messages pushed onto it, and the session stays
warm between turns.

- claude-manager: persistent `query({ prompt: AsyncIterable, options })` per
  sessionKey (bypassPermissions, --resume, mcp via extraArgs, CLAUDECODE stripped).
  Single consumer loop maps every SDK message → ChatEvent, incl. post-turn
  task_started / task_notification. interrupt() = stop-turn; abort() = kill-session;
  30-min idle GC.
- stream-parser: processMessage() (object-level, reused by the SDK loop) + task
  message handling. ChatEvent/ServerMessage gain task:started / task:notification.
- API: the sidecar event subscription is now SESSION-scoped (no longer unsubscribes
  on 'result'), so background events after turn-end still reach the client. First
  turn opens the session; later turns push onto it. handleStop → interrupt (keeps
  session warm); disconnect/deleteSession → kill.
- protocol/sidecar-registry/user-instance: claude:interrupt command + interruptClaude.
- client: render task:started / task:notification in the transcript.

Verified end-to-end through the real chat WS: a run_in_background task's completion
arrives ~6s AFTER the turn's result; multi-turn on one warm session works.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/servers/sidecar/claude/claude-manager.ts

@@ -1,9 +1,9 @@
 import { join } from 'node:path';
-import type { Subprocess } from 'bun';
+import { query, type Query } from '@anthropic-ai/claude-agent-sdk';
 import type { ChatEvent } from '../../api/chat/types';
 import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult } from '../protocol';
 import { setClaudeSession, clearClaudeSession, getClaudeSession } from './state';
-import { parseStream } from './stream-parser';
+import { createParseState, processMessage } from './stream-parser';
 
 const SEND_TIMEOUT_MS = 30 * 60 * 1000;
 
@@ -15,8 +15,8 @@ const CLAUDE_BIN = '/usr/local/bin/claude';
 const HOST_HOME = process.env.HOME!;
 const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
 
-// Active streaming processes
-const activeProcs = new Map<string, Subprocess>();
+// Tear a persistent session down after this long with no new turn (see PersistentSession below).
+const IDLE_TIMEOUT_MS = 30 * 60 * 1000;
 
 // MCP config paths, set by user-instance at startup
 let mcpHostPath: string | undefined; // path on the host filesystem
@@ -25,7 +25,7 @@ export function setMcpConfigPath(hostPath: string): void {
   mcpHostPath = hostPath;
 }
 
-// ── Blocking send ──
+// ── Blocking send (one-shot; used by the non-streaming 'claude:spawn' command) ──
 
 type ClaudeCodeOutput = {
   result: string;
@@ -37,7 +37,7 @@ type ClaudeCodeOutput = {
 };
 
 export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCodeResult> {
-  const { prompt, sessionKey, email } = params;
+  const { prompt, sessionKey } = params;
 
   const existingSession = getClaudeSession(sessionKey);
 
@@ -54,8 +54,6 @@ export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCode
   }
 
   const spawnCmd = claudeArgs;
-  // Honor a caller-provided working directory (e.g. /chat runs from a dedicated general_chat_sessions
-  // dir); fall back to the owner's host home.
   const spawnCwd = params.cwd ?? HOST_HOME;
 
   const proc = Bun.spawn(spawnCmd, {
@@ -119,110 +117,200 @@ export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCode
   }
 }
 
-// ── Streaming send ──
+// ── Persistent streaming sessions (Agent SDK) ──
+//
+// Each sessionKey gets ONE long-lived `query()` driven by a streaming-input queue. The process stays
+// alive BETWEEN turns, so background work (Bash run_in_background, Monitor) and its `task_notification`
+// return to a live harness instead of being orphaned when the turn ends. A turn = one user message
+// pushed onto the input queue; the single consumer loop maps every SDK message (assistant text/tools,
+// tool results, turn `result`, and — crucially — post-turn `task_started`/`task_notification`) to a
+// ChatEvent and forwards it. Idle policy: torn down after IDLE_TIMEOUT_MS with no new turn, or on an
+// explicit kill (user "disconnect"). "Stop" is interrupt() — it ends the turn but keeps the session.
 
+type SdkUserMessage = {
+  type: 'user';
+  message: { role: 'user'; content: string };
+  parent_tool_use_id: null;
+  session_id: string;
+};
+
+type PersistentSession = {
+  sessionKey: string;
+  query: Query;
+  pushTurn: (prompt: string) => void;
+  closeInput: () => void;
+  abort: AbortController;
+  emit: (event: ChatEvent) => void;
+  isGenerating: boolean;
+  idleTimer?: ReturnType<typeof setTimeout>;
+};
+
+const sessions = new Map<string, PersistentSession>();
+
+/** A hand-rolled async iterable we can push turns onto and close on teardown. */
+function makeInputQueue() {
+  const buf: SdkUserMessage[] = [];
+  let wake: (() => void) | null = null;
+  let closed = false;
+  async function* gen(): AsyncGenerator<SdkUserMessage> {
+    while (true) {
+      if (buf.length) {
+        yield buf.shift()!;
+        continue;
+      }
+      if (closed) return;
+      await new Promise<void>((r) => {
+        wake = r;
+      });
+    }
+  }
+  return {
+    gen: gen(),
+    push(m: SdkUserMessage) {
+      buf.push(m);
+      wake?.();
+      wake = null;
+    },
+    close() {
+      closed = true;
+      wake?.();
+      wake = null;
+    },
+  };
+}
+
+function armIdle(session: PersistentSession): void {
+  if (session.idleTimer) clearTimeout(session.idleTimer);
+  session.idleTimer = setTimeout(() => {
+    killClaudeSession(session.sessionKey);
+  }, IDLE_TIMEOUT_MS);
+}
+
+function createSession(params: ClaudeSpawnStreamingParams, onEvent: (event: ChatEvent) => void): PersistentSession {
+  const { sessionKey } = params;
+  const input = makeInputQueue();
+  const abort = new AbortController();
+
+  const session: PersistentSession = {
+    sessionKey,
+    query: undefined as unknown as Query,
+    pushTurn: () => {},
+    closeInput: () => input.close(),
+    abort,
+    emit: onEvent,
+    isGenerating: false,
+  };
+
+  // Strip the nested-session guard vars so the SDK can spawn `claude` (mirrors the old spawn env clean).
+  const { CLAUDECODE: _c, CLAUDE_CODE_ENTRYPOINT: _e, CLAUDE_CODE_SSE_PORT: _s, ...cleanEnv } = process.env;
+
+  const resumeId = getClaudeSession(sessionKey) ?? params.resumeSessionId;
+  const subModel = params.model?.split('/')[1];
+
+  const q = query({
+    prompt: input.gen as AsyncIterable<SdkUserMessage>,
+    options: {
+      cwd: params.cwd ?? HOST_HOME,
+      permissionMode: 'bypassPermissions',
+      allowDangerouslySkipPermissions: true,
+      includePartialMessages: true,
+      abortController: abort,
+      pathToClaudeCodeExecutable: CLAUDE_BIN,
+      settingSources: ['user', 'project', 'local'],
+      env: cleanEnv as Record<string, string>,
+      stderr: (d: string) => {
+        if (d.trim()) console.error(`[claude:stream:${sessionKey}] ${d.slice(0, 300)}`);
+      },
+      ...(subModel ? { model: subModel } : {}),
+      ...(resumeId ? { resume: resumeId } : {}),
+      ...(mcpHostPath ? { extraArgs: { 'mcp-config': mcpHostPath } } : {}),
+    },
+  });
+
+  session.query = q;
+  session.pushTurn = (prompt: string) => {
+    if (session.idleTimer) clearTimeout(session.idleTimer);
+    session.isGenerating = true;
+    input.push({ type: 'user', message: { role: 'user', content: prompt }, parent_tool_use_id: null, session_id: sessionKey });
+  };
+
+  sessions.set(sessionKey, session);
+
+  // Single consumer loop for the session's whole life. Turn-end (`result`) and errors flip isGenerating
+  // and (re)arm the idle timer; the session process stays alive so later task_notifications still flow.
+  void (async () => {
+    const state = createParseState();
+    const emit = (event: ChatEvent) => {
+      if (event.type === 'result' || event.type === 'error' || event.type === 'stopped') {
+        session.isGenerating = false;
+        armIdle(session);
+      }
+      session.emit(event);
+    };
+    try {
+      for await (const msg of q as AsyncGenerator<Record<string, unknown>>) {
+        processMessage(msg, state, {
+          onEvent: emit,
+          onSessionId: (id: string) => setClaudeSession(sessionKey, id),
+        });
+      }
+    } catch (err) {
+      if (!abort.signal.aborted) {
+        session.emit({ type: 'error', message: err instanceof Error ? err.message : String(err) });
+      }
+    } finally {
+      if (session.idleTimer) clearTimeout(session.idleTimer);
+      sessions.delete(sessionKey);
+    }
+  })();
+
+  return session;
+}
+
+/** Start a turn: create the persistent session if needed, then push the prompt as a user message. */
 export async function spawnClaudeStreaming(
   params: ClaudeSpawnStreamingParams,
   onEvent: (event: ChatEvent) => void,
 ): Promise<void> {
-  const { prompt, sessionKey, email } = params;
-
-  const existingSession = getClaudeSession(sessionKey);
-
-  const claudeArgs = [
-    CLAUDE_BIN,
-    '-p',
-    prompt,
-    '--dangerously-skip-permissions',
-    '--output-format',
-    'stream-json',
-    '--verbose',
-    '--include-partial-messages',
-  ];
-
-  const { CLAUDECODE: _, ...cleanEnv } = process.env;
-  const mcpConfig = mcpHostPath;
-  if (mcpConfig) claudeArgs.push('--mcp-config', mcpConfig);
-
-  const subModel = params.model?.split('/')[1];
-  if (subModel) claudeArgs.push('--model', subModel);
-
-  // Resume: an in-memory mapping (subsequent turns of a live chat) takes precedence; otherwise a
-  // caller-supplied session uuid (reopening a session from the /chat list) resumes Claude's transcript.
-  const resumeId = existingSession ?? params.resumeSessionId;
-  if (resumeId) {
-    claudeArgs.push('--resume', resumeId);
-    if (!existingSession) setClaudeSession(sessionKey, resumeId);
+  let session = sessions.get(params.sessionKey);
+  if (session) {
+    session.emit = onEvent; // adopt the latest emitter (equivalent across turns; keeps events flowing)
+  } else {
+    session = createSession(params, onEvent);
   }
+  session.pushTurn(params.prompt);
+}
 
-  const spawnCmd = claudeArgs;
-  // Honor a caller-provided working directory (e.g. /chat runs from a dedicated general_chat_sessions
-  // dir); fall back to the owner's host home.
-  const spawnCwd = params.cwd ?? HOST_HOME;
-
-  const proc = Bun.spawn(spawnCmd, {
-    stdin: 'ignore',
-    stdout: 'pipe',
-    stderr: 'pipe',
-    cwd: spawnCwd,
-    env: cleanEnv as Record<string, string>,
-  });
-
-  activeProcs.set(sessionKey, proc);
-
-  const timeout = setTimeout(() => {
-    try {
-      proc.kill();
-    } catch {
-      /* already dead */
-    }
-    onEvent({ type: 'error', message: 'Claude Code timed out after 5 minutes' });
-  }, SEND_TIMEOUT_MS);
-
-  // Process NDJSON stream
+/** Interrupt the current turn but KEEP the session alive (the "stop" button). */
+export async function interruptClaudeSession(sessionKey: string): Promise<boolean> {
+  const session = sessions.get(sessionKey);
+  if (!session) return false;
   try {
-    const stdout = proc.stdout as ReadableStream<Uint8Array>;
-    const callbacks = {
-      onEvent,
-      onSessionId: (sessionId: string) => setClaudeSession(sessionKey, sessionId),
-    };
-
-    const state = await parseStream(stdout, callbacks);
-
-    clearTimeout(timeout);
-
-    if (!state.gotResult) {
-      const exitCode = await proc.exited;
-      const stderr = await new Response(proc.stderr as ReadableStream<Uint8Array>).text();
-      if (state.textBuffer) {
-        onEvent({ type: 'text', text: state.textBuffer });
-      }
-      if (exitCode !== 0) {
-        onEvent({ type: 'error', message: `Claude Code exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}` });
-      } else {
-        onEvent({ type: 'result', cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 } });
-      }
-    }
-  } catch (err) {
-    clearTimeout(timeout);
-    onEvent({ type: 'error', message: String(err) });
-  } finally {
-    activeProcs.delete(sessionKey);
+    await session.query.interrupt();
+    session.isGenerating = false;
+    return true;
+  } catch {
+    return false;
   }
 }
 
+/** Fully tear the session down (the "disconnect" action / idle GC): abort the query + close input. */
 export function killClaudeSession(sessionKey: string): boolean {
-  const proc = activeProcs.get(sessionKey);
-  if (proc) {
-    try {
-      proc.kill();
-    } catch {
-      /* already dead */
-    }
-    activeProcs.delete(sessionKey);
-    return true;
+  const session = sessions.get(sessionKey);
+  if (!session) return false;
+  if (session.idleTimer) clearTimeout(session.idleTimer);
+  try {
+    session.abort.abort();
+  } catch {
+    /* already aborted */
   }
-  return false;
+  try {
+    session.closeInput();
+  } catch {
+    /* noop */
+  }
+  sessions.delete(sessionKey);
+  return true;
 }
 
 export function clearSession(sessionKey: string): void {
@@ -230,5 +318,5 @@ export function clearSession(sessionKey: string): void {
 }
 
 export function getActiveSessionKeys(): string[] {
-  return Array.from(activeProcs.keys());
+  return Array.from(sessions.keys());
 }