pastilhasandClaude Opus 5
fe0012635a
pm2 inherits the environment of whoever ran pm2 start, so restarting this sidecar from inside a claude code terminal — which is how it is restarted most of the time — bakes that terminal's session into the daemon. right now this process is carrying CLAUDE_CODE_MESSAGING_SOCKET for an unrelated pid that has been alive for an hour and a half. three of these were already stripped; the rest arrived with 2.x and were never added. this is hygiene, not the fix for today's hang — a spawn was verified to succeed with the whole set present — but a child attaching to a stranger's ipc socket is not a failure anyone would recognise from the symptom. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
716 lines
31 KiB
TypeScript
716 lines
31 KiB
TypeScript
import { existsSync } from 'node:fs';
|
|
import { spawn } from 'node:child_process';
|
|
import { homedir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import { query, type Query, type SpawnOptions, type SpawnedProcess } from '@anthropic-ai/claude-agent-sdk';
|
|
import type { ChatEvent, PromptImage } from '../../api/chat/types';
|
|
import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult, LiveClaudeSession } from '../protocol';
|
|
import { setClaudeSession, clearClaudeSession, getClaudeSession } from './state';
|
|
import { createParseState, processMessage } from './stream-parser';
|
|
import { spawnClaudeAsMember } from './spawn-as-member';
|
|
import { claudeBinIn } from '@@/os-user-claude';
|
|
|
|
const SEND_TIMEOUT_MS = 30 * 60 * 1000;
|
|
|
|
// Where the Claude Code CLI lives. This was hardcoded to /usr/local/bin/claude, which dated from the
|
|
// bwrap-sandboxed architecture: the jail ro-bound /usr and saw nothing else, so the installer's real
|
|
// target (~/.local/bin/claude) had to be symlinked into a path the sandbox could reach. That sandbox
|
|
// is gone, and the hardcoded path made the sidecar unrunnable anywhere it does not exist — a stock
|
|
// macOS host has no /usr/local/bin at all.
|
|
//
|
|
// Resolution order mirrors OPENCODE_BIN in the opencode sidecar: an explicit pin, then PATH, then the
|
|
// locations Anthropic's installer actually writes to.
|
|
function resolveClaudeBin(): string {
|
|
const pinned = process.env.CLAUDE_BIN;
|
|
if (pinned) return pinned;
|
|
|
|
const onPath = Bun.which('claude');
|
|
if (onPath) return onPath;
|
|
|
|
const candidates = [join(homedir(), '.local', 'bin', 'claude'), '/usr/local/bin/claude', '/opt/homebrew/bin/claude'];
|
|
return candidates.find((candidate) => existsSync(candidate)) ?? 'claude';
|
|
}
|
|
|
|
const CLAUDE_BIN = resolveClaudeBin();
|
|
console.log(`[claude] CLI resolved to ${CLAUDE_BIN}`);
|
|
|
|
/**
|
|
* Variables the `claude` CLI injects to describe ITS OWN session, which must never reach a `claude` we
|
|
* spawn ourselves.
|
|
*
|
|
* They arrive here by an ordinary accident: PM2 inherits the environment of whoever ran `pm2 start`, so
|
|
* restarting this sidecar from inside a Claude Code terminal — which is how it is restarted most of the
|
|
* time — bakes that terminal's session into the daemon. On 2026-08-14 this process was carrying
|
|
* `CLAUDE_CODE_MESSAGING_SOCKET` for an unrelated PID that had been alive for an hour and a half.
|
|
*
|
|
* Only three of these were stripped before (`CLAUDECODE`, `CLAUDE_CODE_ENTRYPOINT`, `CLAUDE_CODE_SSE_PORT`);
|
|
* the rest are newer and arrived with 2.x. Stripping them is hygiene rather than a fix — a spawn was
|
|
* verified to succeed with the whole set present — but the failure it prevents is a child attaching to a
|
|
* stranger's IPC socket, which would be extremely hard to recognise from the symptom.
|
|
*/
|
|
const NESTED_SESSION_ENV = [
|
|
'CLAUDECODE',
|
|
'CLAUDE_CODE_ENTRYPOINT',
|
|
'CLAUDE_CODE_SSE_PORT',
|
|
'CLAUDE_CODE_CHILD_SESSION',
|
|
'CLAUDE_CODE_MESSAGING_SOCKET',
|
|
'CLAUDE_CODE_MESSAGING_TOKEN',
|
|
'CLAUDE_CODE_SESSION_ID',
|
|
'CLAUDE_CODE_EXECPATH',
|
|
'CLAUDE_PID',
|
|
] as const;
|
|
|
|
/** `process.env` minus the parent session's fingerprint. */
|
|
function envWithoutParentSession(): Record<string, string> {
|
|
const out: Record<string, string> = { ...process.env } as Record<string, string>;
|
|
for (const name of NESTED_SESSION_ENV) delete out[name];
|
|
return out;
|
|
}
|
|
|
|
// Capture original HOME before user-instance overrides it
|
|
const HOST_HOME = process.env.HOME!;
|
|
import { DATA_PATH } from '../../data-path';
|
|
|
|
// 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
|
|
|
|
export function setMcpConfigPath(hostPath: string): void {
|
|
mcpHostPath = hostPath;
|
|
}
|
|
|
|
// ── Blocking send (one-shot; used by the non-streaming 'claude:spawn' command) ──
|
|
|
|
type ClaudeCodeOutput = {
|
|
result: string;
|
|
session_id: string;
|
|
cost_usd: number;
|
|
input_tokens: number;
|
|
output_tokens: number;
|
|
is_error: boolean;
|
|
};
|
|
|
|
export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCodeResult> {
|
|
const { prompt, sessionKey } = params;
|
|
|
|
const existingSession = getClaudeSession(sessionKey, params.userId);
|
|
|
|
const claudeArgs = [CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'json'];
|
|
|
|
const mcpConfig = mcpHostPath;
|
|
if (mcpConfig) claudeArgs.push('--mcp-config', mcpConfig);
|
|
|
|
const subModel = params.model?.split('/')[1];
|
|
if (subModel) claudeArgs.push('--model', subModel);
|
|
|
|
if (existingSession) {
|
|
claudeArgs.push('--resume', existingSession);
|
|
}
|
|
|
|
const spawnCmd = claudeArgs;
|
|
const spawnCwd = params.cwd ?? HOST_HOME;
|
|
|
|
const proc = Bun.spawn(spawnCmd, {
|
|
stdin: 'pipe',
|
|
stdout: 'pipe',
|
|
stderr: 'pipe',
|
|
cwd: spawnCwd,
|
|
env: process.env as Record<string, string>,
|
|
});
|
|
|
|
const timeout = setTimeout(() => {
|
|
try {
|
|
proc.kill();
|
|
} catch {
|
|
/* already dead */
|
|
}
|
|
}, SEND_TIMEOUT_MS);
|
|
|
|
try {
|
|
const [stdout, stderr] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text()]);
|
|
const exitCode = await proc.exited;
|
|
clearTimeout(timeout);
|
|
|
|
if (exitCode !== 0 && !stdout.trim()) {
|
|
throw new Error(`claude exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}`);
|
|
}
|
|
|
|
let output: ClaudeCodeOutput;
|
|
try {
|
|
output = JSON.parse(stdout) as ClaudeCodeOutput;
|
|
} catch {
|
|
return {
|
|
text: stdout.trim() || '(no response)',
|
|
sessionId: sessionKey,
|
|
model: 'claude-code',
|
|
cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 },
|
|
};
|
|
}
|
|
|
|
if (output.is_error) {
|
|
throw new Error(output.result || 'Claude Code returned an error');
|
|
}
|
|
|
|
if (output.session_id) {
|
|
setClaudeSession(sessionKey, output.session_id, params.userId);
|
|
}
|
|
|
|
return {
|
|
text: output.result || '(no response)',
|
|
sessionId: sessionKey,
|
|
model: 'claude-code',
|
|
cost: {
|
|
inputTokens: output.input_tokens ?? 0,
|
|
outputTokens: output.output_tokens ?? 0,
|
|
totalUSD: output.cost_usd ?? 0,
|
|
},
|
|
};
|
|
} catch (err) {
|
|
clearTimeout(timeout);
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ── 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.
|
|
|
|
/**
|
|
* `content` is an Anthropic `MessageParam` content: a bare string, or the block array we build when the
|
|
* turn carries images. Blocks are what make an attached screenshot actually reach the model — the string
|
|
* form can only carry the client's `[Attached image: …]` placeholder, which is a label, not the picture.
|
|
*/
|
|
type ContentBlock =
|
|
| { type: 'text'; text: string }
|
|
| { type: 'image'; source: { type: 'base64'; media_type: string; data: string } };
|
|
|
|
type SdkUserMessage = {
|
|
type: 'user';
|
|
message: { role: 'user'; content: string | ContentBlock[] };
|
|
parent_tool_use_id: null;
|
|
session_id: string;
|
|
};
|
|
|
|
type PersistentSession = {
|
|
sessionKey: string;
|
|
/** Whose session this is. The map is global and `sessionKey` arrives in a client message. */
|
|
userId: number;
|
|
query: Query;
|
|
pushTurn: (prompt: string, images?: PromptImage[]) => void;
|
|
closeInput: () => void;
|
|
abort: AbortController;
|
|
emit: (event: ChatEvent) => void;
|
|
isGenerating: boolean;
|
|
pendingTasks: Set<string>; // background tasks started but not yet notified; suppress idle-GC while non-empty
|
|
/**
|
|
* The user pressed stop and we are waiting for the turn to fall over. The SDK reports an interrupt as
|
|
* an ordinary failed `result` — `is_error` with no text — which is indistinguishable downstream from
|
|
* the harness actually breaking, and reached the user as "Claude Code returned an error". Only the
|
|
* side that called `interrupt()` knows better, so it says so here.
|
|
*/
|
|
interrupted: boolean;
|
|
/**
|
|
* When the `PreCompact` hook fired, so the `compact_boundary` that closes it can carry how long the
|
|
* silence lasted. The harness reports the boundary but not the duration, and the duration is the part
|
|
* that explains the wait.
|
|
*/
|
|
compactStartedAt?: number;
|
|
idleTimer?: ReturnType<typeof setTimeout>;
|
|
/**
|
|
* Fires when a turn that claims to be generating has emitted nothing for too long.
|
|
*
|
|
* The idle timer above answers the opposite question — how long a session with NO turn in flight may
|
|
* sit before it is collected — so neither one covers a turn that is wedged. Nothing did: a turn could
|
|
* stop producing events and stay `isGenerating` forever, and every client showed a spinner with no
|
|
* timeout of its own. On 2026-08-08 that ran for seventeen minutes inside a compaction and was
|
|
* indistinguishable, from the phone, from a dead chat.
|
|
*/
|
|
stallTimer?: ReturnType<typeof setTimeout>;
|
|
};
|
|
|
|
/*
|
|
How long a generating turn may say nothing before we call it stalled.
|
|
|
|
Generous on purpose, because the legitimate silences here are long: compaction narrates nothing for
|
|
as long as it takes (minutes on a large conversation), and a single deep tool call can be quiet for a
|
|
while too. This is a backstop against turns that will never speak again, not a latency budget — too
|
|
tight and it would kill work that was about to succeed, which is worse than the hang it prevents.
|
|
*/
|
|
const STALL_TIMEOUT_MS = 10 * 60 * 1000;
|
|
/** Compaction gets longer still: it is the known-slowest silent phase, and the one that stalled. */
|
|
const COMPACT_STALL_TIMEOUT_MS = 20 * 60 * 1000;
|
|
|
|
const sessions = new Map<string, PersistentSession>();
|
|
|
|
/**
|
|
* The owner's spawn — what the SDK would do by default, written out so it can be wrapped by `watchChild`.
|
|
*
|
|
* A member's turn already supplies its own (`spawnClaudeAsMember`) because it has to go through `setpriv`.
|
|
* The owner had no such function and therefore no place to observe the child, which is precisely why its
|
|
* death was invisible. The existence check mirrors the SDK's default: without it a bad `CLAUDE_BIN` fails
|
|
* as a write to a closed pipe several seconds later, naming nothing useful.
|
|
*/
|
|
function spawnClaudeAsOwner({ command, args, cwd, env, signal }: SpawnOptions): SpawnedProcess {
|
|
if (!existsSync(command)) throw new Error(`claude CLI not found at ${command}`);
|
|
const child = spawn(command, args, { cwd, env, signal, stdio: ['pipe', 'pipe', 'pipe'] });
|
|
// Non-null by construction: 'pipe' on all three. Mirrors the cast in `spawn-as-member.ts`.
|
|
return child as unknown as SpawnedProcess;
|
|
}
|
|
|
|
/**
|
|
* Drop a session whose `claude` process is gone, and tell whoever was waiting.
|
|
*
|
|
* This closes the hole that made a dead agent look like a slow one **forever**. The SDK runs two
|
|
* independent tasks per session: the consumer loop (`for await (const msg of q)`) and an input pump that
|
|
* writes the queue to the child's stdin. The consumer loop's `finally` is what removes a session from the
|
|
* map — but when the CHILD dies, it is the input pump that fails, with `ProcessTransport is not ready for
|
|
* writing`, and that rejection neither ends the consumer loop nor is caught anywhere. So the loop stayed
|
|
* parked on a stream with no writer, `finally` never ran, the session stayed in the map, and
|
|
* `spawnClaudeStreaming` handed every later turn to the same corpse. Each one pushed a message onto a
|
|
* queue nobody drained: no error, no result, no timeout. The client spun forever, and the only trace was
|
|
* an `unhandledRejection` line in the sidecar log.
|
|
*
|
|
* Observed on 2026-08-14; the session had to be cleared with `pm2 restart officer-claude-code`.
|
|
*
|
|
* Emitting only while a turn is in flight is deliberate. A child that exits between turns is invisible to
|
|
* the user, and an error bubble arriving in a chat nobody is looking at would be noise — dropping the map
|
|
* entry is the whole repair there, because the next turn then builds a fresh session and resumes the
|
|
* transcript by id.
|
|
*/
|
|
function dropDeadSession(session: PersistentSession, detail: string): void {
|
|
// Identity, not key: a later turn may have already replaced this entry, and killing its session because
|
|
// its predecessor's process exited would break the live conversation instead of a dead one.
|
|
if (sessions.get(session.sessionKey) !== session) return;
|
|
sessions.delete(session.sessionKey);
|
|
if (session.idleTimer) clearTimeout(session.idleTimer);
|
|
if (session.stallTimer) clearTimeout(session.stallTimer);
|
|
session.idleTimer = undefined;
|
|
session.stallTimer = undefined;
|
|
|
|
const wasGenerating = session.isGenerating;
|
|
session.isGenerating = false;
|
|
// A deliberate teardown (kill / idle GC) aborts first, and its exit is not news.
|
|
if (session.abort.signal.aborted) return;
|
|
|
|
console.error(`[claude:exit:${session.sessionKey}] ${detail}`);
|
|
if (!wasGenerating) return;
|
|
session.emit({
|
|
type: 'error',
|
|
message: 'The agent process exited unexpectedly. Your conversation is safe — send again to continue.',
|
|
});
|
|
}
|
|
|
|
/** Wrap a spawn so the session self-heals when its child goes away. */
|
|
function watchChild(
|
|
inner: (options: SpawnOptions) => SpawnedProcess,
|
|
session: PersistentSession,
|
|
): (options: SpawnOptions) => SpawnedProcess {
|
|
return (options: SpawnOptions): SpawnedProcess => {
|
|
const child = inner(options);
|
|
child.on('exit', (code, signal) =>
|
|
dropDeadSession(session, `claude exited (code=${code ?? 'null'} signal=${signal ?? 'null'})`),
|
|
);
|
|
child.on('error', (err: Error) => dropDeadSession(session, `claude failed to start: ${err.message}`));
|
|
return child;
|
|
};
|
|
}
|
|
|
|
/** 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;
|
|
},
|
|
};
|
|
}
|
|
|
|
/**
|
|
* (Re)arm the stall watchdog. Called on every emitted event, so any sign of life pushes it back.
|
|
*
|
|
* On expiry it ends the turn the same way a real failure would — `isGenerating` off, idle re-armed,
|
|
* and an `error` the client can render — rather than tearing the session down. The agent process is
|
|
* left alive deliberately: it may still be working, and the next turn resumes it. What this guarantees
|
|
* is that the CLIENT is told, which is the part that was missing.
|
|
*/
|
|
function armStall(session: PersistentSession): void {
|
|
if (session.stallTimer) clearTimeout(session.stallTimer);
|
|
if (!session.isGenerating) {
|
|
session.stallTimer = undefined;
|
|
return;
|
|
}
|
|
const compacting = session.compactStartedAt !== undefined;
|
|
session.stallTimer = setTimeout(
|
|
() => {
|
|
if (!session.isGenerating) return;
|
|
const waited = Math.round((compacting ? COMPACT_STALL_TIMEOUT_MS : STALL_TIMEOUT_MS) / 60000);
|
|
console.error(
|
|
`[claude:stall:${session.sessionKey}] no events for ${waited}m${compacting ? ' (compacting)' : ''} — ending the turn`,
|
|
);
|
|
session.isGenerating = false;
|
|
session.compactStartedAt = undefined;
|
|
session.interrupted = false;
|
|
session.emit({
|
|
type: 'error',
|
|
message: compacting
|
|
? `Compaction has produced nothing for ${waited} minutes and has been given up on. Your conversation is safe — send again to continue.`
|
|
: `The agent has produced nothing for ${waited} minutes and has been given up on. Your conversation is safe — send again to continue.`,
|
|
});
|
|
// A background job can be silent for far longer than this and still land its `task_notification`, so
|
|
// a stall with tasks outstanding keeps the old behaviour and leaves the session alone.
|
|
if (session.pendingTasks.size > 0) return;
|
|
|
|
// Otherwise tear it down rather than leaving it armed for the next turn.
|
|
//
|
|
// This used to keep the session — "it may still be working, and the next turn resumes it" — which is
|
|
// the right instinct for a SLOW agent and exactly wrong for a wedged one: a session that has said
|
|
// nothing for ten minutes because its transport is broken stays broken, so every later turn hangs
|
|
// the same way and the message above ("send again to continue") is a lie. Killing costs a resume,
|
|
// which is what the message already promises; the transcript id survives in `claudeSessions`, so the
|
|
// next turn continues the same conversation.
|
|
killClaudeSession(session.sessionKey, session.userId);
|
|
},
|
|
compacting ? COMPACT_STALL_TIMEOUT_MS : STALL_TIMEOUT_MS,
|
|
);
|
|
}
|
|
|
|
function armIdle(session: PersistentSession): void {
|
|
if (session.idleTimer) clearTimeout(session.idleTimer);
|
|
session.idleTimer = setTimeout(() => {
|
|
// Heartbeat: never GC a session that's mid-turn or still has background tasks running — a long
|
|
// silent run_in_background job would otherwise be killed along with its pending task_notification.
|
|
if (session.isGenerating || session.pendingTasks.size > 0) {
|
|
armIdle(session);
|
|
return;
|
|
}
|
|
// The idle GC is this process acting on its own session, so it passes the session's own owner.
|
|
killClaudeSession(session.sessionKey, session.userId);
|
|
}, 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,
|
|
userId: params.userId,
|
|
query: undefined as unknown as Query,
|
|
pushTurn: () => {},
|
|
closeInput: () => input.close(),
|
|
abort,
|
|
emit: onEvent,
|
|
isGenerating: false,
|
|
pendingTasks: new Set<string>(),
|
|
interrupted: false,
|
|
};
|
|
|
|
// Strip the nested-session guard vars so the SDK can spawn `claude` (mirrors the old spawn env clean).
|
|
const cleanEnv = envWithoutParentSession();
|
|
|
|
const resumeId = getClaudeSession(sessionKey, params.userId) ?? params.resumeSessionId;
|
|
const subModel = params.model?.split('/')[1];
|
|
|
|
const q = query({
|
|
prompt: input.gen as AsyncIterable<SdkUserMessage>,
|
|
options: {
|
|
// HOST_HOME is this process's home — the owner's. Defaulting a member's turn to it would start them in
|
|
// a directory they cannot read, and the failure would look like a broken agent rather than a wrong cwd.
|
|
cwd: params.cwd ?? params.member?.home ?? HOST_HOME,
|
|
permissionMode: 'bypassPermissions',
|
|
allowDangerouslySkipPermissions: true,
|
|
includePartialMessages: true,
|
|
// The only warning that compaction is about to happen. Everything else the harness does narrates
|
|
// itself through the message stream; compaction goes silent for as long as it takes — 2.5 minutes
|
|
// in the worst case on disk here — and the stream resumes with no explanation of the gap. The hook
|
|
// returns immediately and never throws: it is a notification, and it must not be able to stall or
|
|
// fail the compaction it is announcing.
|
|
hooks: {
|
|
PreCompact: [
|
|
{
|
|
hooks: [
|
|
async (input) => {
|
|
session.compactStartedAt = Date.now();
|
|
const trigger = 'trigger' in input && input.trigger === 'manual' ? 'manual' : 'auto';
|
|
session.emit({ type: 'compact:start', trigger });
|
|
// Re-arm on the compaction budget: this hook fires as the long silence BEGINS, so the
|
|
// deadline the turn is holding was sized for ordinary work and is about to be wrong.
|
|
armStall(session);
|
|
return { continue: true };
|
|
},
|
|
],
|
|
},
|
|
],
|
|
},
|
|
abortController: abort,
|
|
// A member's turn runs their own install as their own Linux account; the owner's runs as it always has.
|
|
//
|
|
// `settingSources` is why the binary and the spawn have to move together: it makes `~/.claude`
|
|
// authoritative for settings, and `~` is decided by the HOME the process gets. Pointing the SDK at a
|
|
// member's binary while spawning as the service user would read the OWNER'S settings and credential
|
|
// while executing the member's code — the worst of both, and it would look like it worked.
|
|
//
|
|
// Both branches go through `watchChild`: the owner's spawn exists only so there is something to
|
|
// wrap (see `spawnClaudeAsOwner`), because a child nobody watches is a session that can die silently.
|
|
...(params.member
|
|
? {
|
|
pathToClaudeCodeExecutable: claudeBinIn(params.member.home),
|
|
spawnClaudeCodeProcess: watchChild(spawnClaudeAsMember(params.member), session),
|
|
}
|
|
: {
|
|
pathToClaudeCodeExecutable: CLAUDE_BIN,
|
|
spawnClaudeCodeProcess: watchChild(spawnClaudeAsOwner, session),
|
|
}),
|
|
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 } : {}),
|
|
// The owner's MCP config, and only ever the owner's. `mcpHostPath` is module-level, written once at
|
|
// this process's bootstrap, and its `env` carries OFFICER_AUTH_TOKEN — a JWT that signs as the owner.
|
|
// Handing it to a member's turn would either spawn their MCP server holding the owner's token, or (once
|
|
// that file is 0600, which it now is) point their `claude` at a file it cannot read and fail obscurely.
|
|
//
|
|
// So a member gets no MCP config at all. What they SHOULD get — their own generated config with a token
|
|
// scoped to them, or nothing until per-user tools exist — is an open design question; `undefined` is
|
|
// the correct answer until it is settled, and is strictly better than the owner's.
|
|
...(mcpHostPath && !params.member ? { extraArgs: { 'mcp-config': mcpHostPath } } : {}),
|
|
},
|
|
});
|
|
|
|
session.query = q;
|
|
session.pushTurn = (prompt: string, images?: PromptImage[]) => {
|
|
if (session.idleTimer) clearTimeout(session.idleTimer);
|
|
session.isGenerating = true;
|
|
// A turn that dies before its FIRST event is the case the emit-path arming cannot reach.
|
|
armStall(session);
|
|
// Images first, then the text: the model reads what it is looking at before what to do about it.
|
|
const content: string | ContentBlock[] = images?.length
|
|
? [
|
|
...images.map(
|
|
(img): ContentBlock => ({
|
|
type: 'image',
|
|
source: { type: 'base64', media_type: img.mediaType, data: img.data },
|
|
}),
|
|
),
|
|
{ type: 'text', text: prompt },
|
|
]
|
|
: prompt;
|
|
input.push({
|
|
type: 'user',
|
|
message: { role: 'user', content },
|
|
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 = (raw: ChatEvent) => {
|
|
// A turn we interrupted ends in a failed `result`. That is the stop landing, not a fault.
|
|
const event: ChatEvent =
|
|
raw.type === 'error' && session.interrupted
|
|
? { type: 'stopped' }
|
|
: // The boundary knows what it dropped; only this side knows how long it took, because the start
|
|
// came from a hook rather than from the stream.
|
|
raw.type === 'compact:done' && session.compactStartedAt
|
|
? { ...raw, durationMs: Date.now() - session.compactStartedAt }
|
|
: raw;
|
|
if (event.type === 'compact:done') session.compactStartedAt = undefined;
|
|
if (event.type === 'task:started') {
|
|
// Work is running — hold off idle-GC until it finishes.
|
|
session.pendingTasks.add(event.taskId);
|
|
if (session.idleTimer) {
|
|
clearTimeout(session.idleTimer);
|
|
session.idleTimer = undefined;
|
|
}
|
|
} else if (event.type === 'task:notification') {
|
|
session.pendingTasks.delete(event.taskId);
|
|
if (!session.isGenerating && session.pendingTasks.size === 0) armIdle(session);
|
|
} else if (event.type === 'result' || event.type === 'error' || event.type === 'stopped') {
|
|
session.isGenerating = false;
|
|
// Whatever ended the turn, the interrupt is spent — a later genuine error must not wear it.
|
|
session.interrupted = false;
|
|
if (session.pendingTasks.size === 0) armIdle(session);
|
|
}
|
|
// Any event at all is a sign of life, so push the stall deadline back. Placed after the branch
|
|
// above so a terminal event disarms rather than re-arms it.
|
|
armStall(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, params.userId),
|
|
});
|
|
}
|
|
} 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);
|
|
if (session.stallTimer) clearTimeout(session.stallTimer);
|
|
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> {
|
|
let session = sessions.get(params.sessionKey);
|
|
if (session && session.userId !== params.userId) {
|
|
// A live session belongs to whoever started it. Without this, handing over someone else's `sessionKey`
|
|
// pushes a turn into their conversation and streams their agent's output back — the chat equivalent of
|
|
// resuming another account's shell, which `4d4a253f` refused for the pty sidecar.
|
|
//
|
|
// Throws rather than silently starting a fresh session under the same key: the caller asked to continue a
|
|
// specific conversation, and quietly giving them a different one is its own kind of wrong.
|
|
throw new Error('that chat session belongs to another account');
|
|
}
|
|
if (session) {
|
|
session.emit = onEvent; // adopt the latest emitter (equivalent across turns; keeps events flowing)
|
|
} else {
|
|
session = createSession(params, onEvent);
|
|
}
|
|
session.pushTurn(params.prompt, params.images);
|
|
}
|
|
|
|
/** Interrupt the current turn but KEEP the session alive (the "stop" button). */
|
|
export async function interruptClaudeSession(sessionKey: string, userId: number): Promise<boolean> {
|
|
const session = ownedSession(sessionKey, userId);
|
|
if (!session) return false;
|
|
// Set before the await: the failed `result` can arrive while interrupt() is still resolving, and the
|
|
// consumer loop reads this flag to tell a stop from a fault.
|
|
session.interrupted = true;
|
|
try {
|
|
await session.query.interrupt();
|
|
session.isGenerating = false;
|
|
if (session.stallTimer) clearTimeout(session.stallTimer);
|
|
session.stallTimer = undefined;
|
|
return true;
|
|
} catch {
|
|
session.interrupted = false;
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/** Fully tear the session down (the "disconnect" action / idle GC): abort the query + close input. */
|
|
export function killClaudeSession(sessionKey: string, userId: number): boolean {
|
|
const session = ownedSession(sessionKey, userId);
|
|
if (!session) return false;
|
|
if (session.idleTimer) clearTimeout(session.idleTimer);
|
|
if (session.stallTimer) clearTimeout(session.stallTimer);
|
|
try {
|
|
session.abort.abort();
|
|
} catch {
|
|
/* already aborted */
|
|
}
|
|
try {
|
|
session.closeInput();
|
|
} catch {
|
|
/* noop */
|
|
}
|
|
sessions.delete(sessionKey);
|
|
return true;
|
|
}
|
|
|
|
export function clearSession(sessionKey: string, userId: number): void {
|
|
// Only clears a mapping that is theirs. `getClaudeSession` already refuses a mismatch, so this asks it
|
|
// first rather than reimplementing the check.
|
|
if (!getClaudeSession(sessionKey, userId)) return;
|
|
clearClaudeSession(sessionKey);
|
|
}
|
|
|
|
/**
|
|
* The live session under this key, **only if it belongs to the caller**.
|
|
*
|
|
* Undefined for both "no such session" and "not yours", deliberately: every caller of this treats the two the
|
|
* same, and a distinct answer for the second would tell a guesser that a session exists under a key they do
|
|
* not own — which is the whole thing being defended against.
|
|
*/
|
|
function ownedSession(sessionKey: string, userId: number): PersistentSession | undefined {
|
|
const session = sessions.get(sessionKey);
|
|
if (!session || session.userId !== userId) return undefined;
|
|
return session;
|
|
}
|
|
|
|
/**
|
|
* Everything this process is holding, with the two facts that decide whether it is busy.
|
|
*
|
|
* Replaces a `getActiveSessionKeys` that returned bare keys and was never called by anything — the keys
|
|
* alone could not distinguish a session mid-turn from one merely open, which is the whole question a
|
|
* caller has. These are the same two fields `armIdle` consults before collecting a session.
|
|
*/
|
|
export function listSessions(userId: number): LiveClaudeSession[] {
|
|
// Filtered, not just labelled. Enumerating every live session is a disclosure on its own, before anyone
|
|
// acts on one: it names other accounts' conversations and says which are busy.
|
|
return Array.from(sessions.values())
|
|
.filter((session) => session.userId === userId)
|
|
.map((session) => ({
|
|
sessionKey: session.sessionKey,
|
|
// The only place this mapping exists. Without it a caller cannot find the transcript, because the
|
|
// key is officer's handle and the filename is Claude's id.
|
|
claudeSessionId: getClaudeSession(session.sessionKey, session.userId) ?? null,
|
|
isGenerating: session.isGenerating,
|
|
pendingTasks: session.pendingTasks.size,
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Is a turn actually in flight for this session, right now, in this process?
|
|
*
|
|
* This is the ground truth a reconnecting browser has no way to work out for itself. Officer's own view
|
|
* dies with `pm2 restart officer` while the turn keeps running here, so "officer doesn't remember" means
|
|
* nothing — and if *this* process was the one that restarted, the session is simply absent and the turn
|
|
* it was running is gone, however alive the client still believes it to be.
|
|
*/
|
|
export function isSessionGenerating(sessionKey: string, userId: number): boolean {
|
|
// "Not yours" answers the same as "no such session": false. The caller uses this to decide whether to end
|
|
// a turn it believes is running, and its own turn is the only one it can be right about.
|
|
return ownedSession(sessionKey, userId)?.isGenerating ?? false;
|
|
}
|