From 0090549188ddd7de373c0761133eb0ca78054ef1 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 30 Jul 2026 23:40:40 +0800 Subject: [PATCH] fix(agents): refresh Claude live system prompts (#116292) Co-authored-by: William Faris Chesnutt --- src/agents/cli-runner.spawn.test.ts | 404 +++++++++++++++++++ src/agents/cli-runner.test-helpers.ts | 3 +- src/agents/cli-runner/claude-live-session.ts | 302 +++++++++++++- 3 files changed, 696 insertions(+), 13 deletions(-) diff --git a/src/agents/cli-runner.spawn.test.ts b/src/agents/cli-runner.spawn.test.ts index 3109621ae997..032a2a2e89a7 100644 --- a/src/agents/cli-runner.spawn.test.ts +++ b/src/agents/cli-runner.spawn.test.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { SYSTEM_PROMPT_CACHE_BOUNDARY } from "@openclaw/ai/internal/shared"; import { expectDefined } from "@openclaw/normalization-core"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { createReplyOperation, replyRunRegistry } from "../auto-reply/reply/reply-run-registry.js"; @@ -2546,6 +2547,409 @@ describe("runCliAgent spawn path", () => { expect(supervisorSpawnMock).toHaveBeenCalledOnce(); }); + it("refreshes a reused Claude live session when only dynamic prompt context changes", async () => { + let userTurn = 0; + let controlRequest = 0; + const live = mockClaudeLiveRun(supervisorSpawnMock, { + onWrite: ({ data, emit }) => { + const parsed = JSON.parse(data) as { + type: string; + request_id?: string; + request?: { + subtype?: string; + model?: string; + system_prompt?: string; + }; + }; + if (parsed.type === "control_request") { + controlRequest += 1; + if (controlRequest === 1) { + expect(parsed.request).toEqual({ + subtype: "set_model", + model: "sonnet", + system_prompt: "", + }); + emit([ + { + type: "control_response", + response: { + subtype: "error", + request_id: parsed.request_id, + error: "set_model: system_prompt must be a non-empty string when present", + }, + }, + ]); + return; + } + expect(parsed.request).toEqual({ + subtype: "set_model", + model: "sonnet", + system_prompt: + "# OpenClaw\n\n## Stable Instructions\nKeep the operator informed.\nSecond-turn metadata", + }); + emit([ + { + type: "control_response", + response: { + subtype: "success", + request_id: parsed.request_id, + }, + }, + ]); + return; + } + userTurn += 1; + emit([ + { type: "system", subtype: "init", session_id: "live-dynamic-prompt" }, + { + type: "result", + session_id: "live-dynamic-prompt", + result: userTurn === 1 ? "one" : "two", + }, + ]); + }, + }); + const backend = { + resumeArgs: ["-p", "--output-format", "stream-json", "--resume={sessionId}"], + liveSession: "claude-stdio" as const, + systemPromptWhen: "always" as const, + }; + + const first = await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + prompt: "first", + systemPrompt: `# OpenClaw\n\n## Stable Instructions\nKeep the operator informed.${SYSTEM_PROMPT_CACHE_BOUNDARY}First-turn metadata`, + }), + ); + const second = await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + prompt: "second", + systemPrompt: `# OpenClaw\n\n## Stable Instructions\nKeep the operator informed.${SYSTEM_PROMPT_CACHE_BOUNDARY}Second-turn metadata`, + }), + "live-dynamic-prompt", + ); + + expect(first.text).toBe("one"); + expect(second.text).toBe("two"); + expect(supervisorSpawnMock).toHaveBeenCalledOnce(); + expect(live.writes.map((entry) => JSON.parse(entry).type)).toEqual([ + "user", + "control_request", + "control_request", + "user", + ]); + }); + + it("serializes direct live turns before refreshing their system prompts", async () => { + let userTurn = 0; + let releaseCapabilityProbe: (() => void) | undefined; + const live = mockClaudeLiveRun(supervisorSpawnMock, { + onWrite: ({ data, emit }) => { + const parsed = JSON.parse(data) as { + type: string; + request_id?: string; + request?: { system_prompt?: string }; + }; + if (parsed.type === "control_request") { + if (parsed.request?.system_prompt === "") { + releaseCapabilityProbe = () => { + emit([ + { + type: "control_response", + response: { + subtype: "error", + request_id: parsed.request_id, + error: "set_model: system_prompt must be a non-empty string when present", + }, + }, + ]); + }; + return; + } + emit([ + { + type: "control_response", + response: { + subtype: "success", + request_id: parsed.request_id, + }, + }, + ]); + return; + } + userTurn += 1; + emit([ + { type: "system", subtype: "init", session_id: "live-serialized-refresh" }, + { + type: "result", + session_id: "live-serialized-refresh", + result: `turn-${userTurn}`, + }, + ]); + }, + }); + const backend = { + args: ["-p", "--output-format", "stream-json"], + resumeArgs: ["-p", "--output-format", "stream-json", "--resume={sessionId}"], + liveSession: "claude-stdio" as const, + systemPromptWhen: "always" as const, + }; + const getProcessSupervisorForTest = () => ({ + spawn: (params: Parameters[0]) => + supervisorSpawnMock(params) as ReturnType, + cancel: vi.fn(), + cancelScope: vi.fn(), + getRecord: vi.fn(), + }); + const runTurn = ( + systemPrompt: string, + prompt: string, + useResume: boolean, + abortSignal?: AbortSignal, + cleanup: () => Promise = async () => {}, + ) => { + const context = buildPreparedCliRunContext({ backend, prompt, systemPrompt }); + context.params.abortSignal = abortSignal; + return runClaudeLiveSessionTurn({ + context, + args: context.preparedBackend.backend.args ?? [], + env: {}, + prompt, + useResume, + noOutputTimeoutMs: 1_000, + getProcessSupervisor: getProcessSupervisorForTest, + onAssistantDelta: () => {}, + cleanup, + }); + }; + + await expect( + runTurn(`Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}First metadata`, "first", false), + ).resolves.toMatchObject({ output: { text: "turn-1" } }); + + const second = runTurn( + `Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}Second metadata`, + "second", + true, + ); + await vi.waitFor(() => expect(releaseCapabilityProbe).toBeTypeOf("function")); + const queuedAbort = new AbortController(); + const abortedCleanup = vi.fn(async () => {}); + const third = runTurn( + `Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}Third metadata`, + "third", + true, + queuedAbort.signal, + abortedCleanup, + ); + await new Promise((resolve) => { + setImmediate(resolve); + }); + + expect(live.writes.map((entry) => JSON.parse(entry).type)).toEqual(["user", "control_request"]); + queuedAbort.abort(); + await expect(third).rejects.toMatchObject({ name: "AbortError" }); + expect(abortedCleanup).toHaveBeenCalledOnce(); + expect(live.writes.map((entry) => JSON.parse(entry).type)).toEqual(["user", "control_request"]); + releaseCapabilityProbe?.(); + + await expect(second).resolves.toMatchObject({ output: { text: "turn-2" } }); + await expect( + runTurn(`Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}Fourth metadata`, "fourth", true), + ).resolves.toMatchObject({ output: { text: "turn-3" } }); + expect(live.writes.map((entry) => JSON.parse(entry).type)).toEqual([ + "user", + "control_request", + "control_request", + "user", + "control_request", + "user", + ]); + expect(supervisorSpawnMock).toHaveBeenCalledOnce(); + }); + + it("restarts Claude live sessions when a multi-section stable prompt changes", async () => { + mockClaudeLiveRun(supervisorSpawnMock, { + events: [ + { type: "system", subtype: "init", session_id: "live-stable-prompt" }, + { type: "result", session_id: "live-stable-prompt", result: "one" }, + ], + cancelable: true, + }); + mockClaudeLiveRun(supervisorSpawnMock, { + events: [ + { type: "system", subtype: "init", session_id: "live-stable-prompt" }, + { type: "result", session_id: "live-stable-prompt", result: "two" }, + ], + }); + const backend = { + resumeArgs: ["-p", "--output-format", "stream-json", "--resume={sessionId}"], + liveSession: "claude-stdio" as const, + systemPromptWhen: "always" as const, + }; + + await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + systemPrompt: `# OpenClaw\n\n## Stable Instructions\nFirst instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}Metadata`, + }), + ); + const second = await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + systemPrompt: `# OpenClaw\n\n## Stable Instructions\nSecond instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}Metadata`, + }), + "live-stable-prompt", + ); + + expect(second.text).toBe("two"); + expect(supervisorSpawnMock).toHaveBeenCalledTimes(2); + }); + + it.each([ + { + name: "ignores the system_prompt field", + responses: [{ subtype: "success" }], + }, + { + name: "rejects the live refresh", + responses: [ + { + subtype: "error", + error: "set_model: system_prompt must be a non-empty string when present", + }, + { subtype: "error", error: "unsupported" }, + ], + }, + ])("restarts when Claude $name", async ({ responses }) => { + let controlRequest = 0; + mockClaudeLiveRun(supervisorSpawnMock, { + cancelable: true, + onWrite: ({ data, emit }) => { + const parsed = JSON.parse(data) as { type: string; request_id?: string }; + if (parsed.type === "control_request") { + const response = responses[controlRequest]; + controlRequest += 1; + emit([ + { + type: "control_response", + response: { + request_id: parsed.request_id, + ...response, + }, + }, + ]); + return; + } + emit([ + { type: "system", subtype: "init", session_id: "live-rejected-prompt" }, + { type: "result", session_id: "live-rejected-prompt", result: "one" }, + ]); + }, + }); + mockClaudeLiveRun(supervisorSpawnMock, { + events: [ + { type: "system", subtype: "init", session_id: "live-rejected-prompt" }, + { type: "result", session_id: "live-rejected-prompt", result: "two" }, + ], + }); + const backend = { + resumeArgs: ["-p", "--output-format", "stream-json", "--resume={sessionId}"], + liveSession: "claude-stdio" as const, + systemPromptWhen: "always" as const, + }; + + await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + systemPrompt: `Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}First metadata`, + }), + ); + const second = await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + systemPrompt: `Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}Second metadata`, + }), + "live-rejected-prompt", + ); + + expect(second.text).toBe("two"); + expect(supervisorSpawnMock).toHaveBeenCalledTimes(2); + expect(controlRequest).toBe(responses.length); + }); + + it("restarts on marker-free prompt changes instead of weakening prompt identity", async () => { + mockClaudeLiveRun(supervisorSpawnMock, { + events: [ + { type: "system", subtype: "init", session_id: "live-marker-free" }, + { type: "result", session_id: "live-marker-free", result: "one" }, + ], + cancelable: true, + }); + mockClaudeLiveRun(supervisorSpawnMock, { + events: [ + { type: "system", subtype: "init", session_id: "live-marker-free" }, + { type: "result", session_id: "live-marker-free", result: "two" }, + ], + }); + const backend = { + resumeArgs: ["-p", "--output-format", "stream-json", "--resume={sessionId}"], + liveSession: "claude-stdio" as const, + systemPromptWhen: "always" as const, + }; + + await executePreparedCliRun( + buildPreparedCliRunContext({ backend, systemPrompt: "First complete prompt" }), + ); + const second = await executePreparedCliRun( + buildPreparedCliRunContext({ backend, systemPrompt: "Second complete prompt" }), + "live-marker-free", + ); + + expect(second.text).toBe("two"); + expect(supervisorSpawnMock).toHaveBeenCalledTimes(2); + }); + + it("keeps legacy first-only system prompts on full-prompt restart identity", async () => { + mockClaudeLiveRun(supervisorSpawnMock, { + events: [ + { type: "system", subtype: "init", session_id: "live-first-only-prompt" }, + { type: "result", session_id: "live-first-only-prompt", result: "one" }, + ], + cancelable: true, + }); + mockClaudeLiveRun(supervisorSpawnMock, { + events: [ + { type: "system", subtype: "init", session_id: "live-first-only-prompt" }, + { type: "result", session_id: "live-first-only-prompt", result: "two" }, + ], + }); + const backend = { + resumeArgs: ["-p", "--output-format", "stream-json", "--resume={sessionId}"], + liveSession: "claude-stdio" as const, + systemPromptWhen: "first" as const, + }; + + await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + systemPrompt: `Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}First metadata`, + }), + ); + const second = await executePreparedCliRun( + buildPreparedCliRunContext({ + backend, + systemPrompt: `Stable instructions${SYSTEM_PROMPT_CACHE_BOUNDARY}Second metadata`, + }), + "live-first-only-prompt", + ); + + expect(second.text).toBe("two"); + expect(supervisorSpawnMock).toHaveBeenCalledTimes(2); + }); + it("serializes concurrent Claude live session creation for the same key", async () => { let releaseSpawn: (() => void) | undefined; let turn = 0; diff --git a/src/agents/cli-runner.test-helpers.ts b/src/agents/cli-runner.test-helpers.ts index bcef4099bcf3..95b97e0f8f80 100644 --- a/src/agents/cli-runner.test-helpers.ts +++ b/src/agents/cli-runner.test-helpers.ts @@ -134,6 +134,7 @@ export type PreparedCliRunContextOverrides = { cliToolAvailability?: PreparedCliRunContext["params"]["cliToolAvailability"]; emitCommentaryText?: boolean; workspaceDir?: string; + systemPrompt?: string; timeoutMs?: number; onSuccessfulAuthBinding?: PreparedCliRunContext["params"]["onSuccessfulAuthBinding"]; runtimeArtifact?: PreparedCliRunContext["backendResolved"]["runtimeArtifact"]; @@ -241,7 +242,7 @@ export function buildPreparedCliRunContext( contextEngineConfig: {}, modelId: model, normalizedModel: model, - systemPrompt: "You are a helpful assistant.", + systemPrompt: overrides.systemPrompt ?? "You are a helpful assistant.", systemPromptReport: {} as PreparedCliRunContext["systemPromptReport"], bootstrapPromptWarningLines: [], authEpochVersion: 2, diff --git a/src/agents/cli-runner/claude-live-session.ts b/src/agents/cli-runner/claude-live-session.ts index bcc5ff2fb753..7111b8b063e2 100644 --- a/src/agents/cli-runner/claude-live-session.ts +++ b/src/agents/cli-runner/claude-live-session.ts @@ -2,6 +2,10 @@ * Manages reusable Claude CLI stdio sessions for CLI-backed agent turns. */ import crypto from "node:crypto"; +import { + splitSystemPromptCacheBoundary, + stripSystemPromptCacheBoundary, +} from "@openclaw/ai/internal/shared"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import type { ReplyBackendHandle } from "../../auto-reply/reply/reply-run-registry.js"; import { createAbortError as createNamedAbortError } from "../../infra/abort-signal.js"; @@ -23,6 +27,7 @@ import { type ExecSecurity, } from "../../infra/exec-approvals.js"; import { BLOCKED_TOOL_CALL_ABORT_FLOOR_MS } from "../../logging/diagnostic-run-activity.js"; +import { KeyedAsyncQueue } from "../../plugin-sdk/keyed-async-queue.js"; import type { CliBackendConfig } from "../../plugins/cli-backend.types.js"; import { LEGACY_IMPLICIT_AGENT_ID, @@ -109,6 +114,8 @@ type ClaudeLiveSession = { key: string; generation: string; fingerprint: string; + systemPromptHash: string; + systemPromptSwitchCapability: "unknown" | "supported" | "unsupported"; managedRun: ManagedRun; providerId: string; modelId: string; @@ -121,6 +128,7 @@ type ClaudeLiveSession = { cleanup: () => Promise; cleanupPromise: Promise | null; closing: boolean; + pendingControlRequest: ClaudeLivePendingControlRequest | null; mcpCaptureKey?: string; /** * Native-tool allow-always grants are process-session scoped and in-memory only. @@ -138,6 +146,15 @@ type ClaudeLiveSessionCreate = { generation: string; promise: Promise; }; +type ClaudeLivePendingControlRequest = { + requestId: string; + timer: NodeJS.Timeout; + resolve: (response: ClaudeLiveControlResponse | null) => void; +}; +type ClaudeLiveControlResponse = { + subtype: string; + error?: string; +}; type ClaudeLiveRunResult = { output: CliOutput; }; @@ -163,6 +180,9 @@ type ClaudeLiveToolTerminalOutcome = | { outcome: "blocked"; deniedReason: string; reason?: string } | { outcome: "cancelled" | "failed" | "timed_out" | "unknown" }; const CLAUDE_LIVE_IDLE_TIMEOUT_MS = 10 * 60 * 1_000; +const CLAUDE_LIVE_CONTROL_TIMEOUT_MS = 3_000; +const CLAUDE_LIVE_SYSTEM_PROMPT_PROBE_ERROR = + "set_model: system_prompt must be a non-empty string when present"; const CLAUDE_LIVE_CLOSE_WAIT_TIMEOUT_MS = 5_000; // The observed queued-notification resume emits new process activity within // seconds. Cap this below the normal resumed no-output watchdog so terminal @@ -177,6 +197,7 @@ const CLAUDE_LIVE_PROVISIONAL_SYNTHETIC_PLACEHOLDERS = new Set([ ]); const liveSessions = new Map(); const liveSessionCreates = new Map(); +const liveSessionTurns = new KeyedAsyncQueue(); function sha256(value: string): string { return crypto.createHash("sha256").update(value).digest("hex"); @@ -375,6 +396,10 @@ function buildClaudeLiveFingerprint(params: { argv: string[]; env: Record; }): string { + const stableSystemPrompt = + (params.context.preparedBackend.backend.systemPromptWhen === "always" + ? splitSystemPromptCacheBoundary(params.context.systemPrompt)?.stablePrefix + : undefined) ?? params.context.systemPrompt; const normalizeMcpConfigPath = Boolean(params.context.preparedBackend.mcpConfigHash); const skillSnapshot = params.context.params.skillsSnapshot; const skillsFingerprint = skillSnapshot @@ -436,7 +461,7 @@ function buildClaudeLiveFingerprint(params: { cwdHash: params.context.cwdHash ?? sha256(params.context.cwd ?? params.context.workspaceDir), provider: params.context.params.provider, model: params.context.normalizedModel, - systemPromptHash: sha256(params.context.systemPrompt), + systemPromptHash: sha256(stableSystemPrompt), authProfileIdHash: params.context.effectiveAuthProfileId ? sha256(params.context.effectiveAuthProfileId) : undefined, @@ -494,6 +519,19 @@ function clearOutstandingBackgroundTasks(session: ClaudeLiveSession): void { session.outstandingBackgroundTaskIds.clear(); } +function settleClaudeLivePendingControlRequest( + session: ClaudeLiveSession, + response: ClaudeLiveControlResponse | null, +): void { + const pending = session.pendingControlRequest; + if (!pending) { + return; + } + clearTimeout(pending.timer); + session.pendingControlRequest = null; + pending.resolve(response); +} + function finishTurn(session: ClaudeLiveSession, output: CliOutput): void { const turn = session.currentTurn; if (!turn) { @@ -564,6 +602,7 @@ function closeLiveSession( if (liveSessions.get(session.key) === session) { liveSessions.delete(session.key); } + settleClaudeLivePendingControlRequest(session, null); if (error) { failTurn(session, error); } else { @@ -1070,6 +1109,129 @@ function writeClaudeLiveControlResponse(session: ClaudeLiveSession, response: un stdin.write(`${JSON.stringify(response)}\n`); } +function handleClaudeLiveControlResponse( + session: ClaudeLiveSession, + parsed: Record, +): boolean { + const pending = session.pendingControlRequest; + if (!pending || parsed.type !== "control_response" || !isRecord(parsed.response)) { + return false; + } + const response = parsed.response; + if (response.request_id !== pending.requestId) { + return false; + } + settleClaudeLivePendingControlRequest(session, { + subtype: typeof response.subtype === "string" ? response.subtype : "", + ...(typeof response.error === "string" ? { error: response.error } : {}), + }); + return true; +} + +async function requestClaudeLiveModelUpdate(params: { + session: ClaudeLiveSession; + model: string; + systemPrompt: string; +}): Promise { + if (params.session.pendingControlRequest) { + return null; + } + const requestId = crypto.randomUUID(); + const response = new Promise((resolve) => { + params.session.pendingControlRequest = { + requestId, + timer: setTimeout(() => { + settleClaudeLivePendingControlRequest(params.session, null); + }, CLAUDE_LIVE_CONTROL_TIMEOUT_MS), + resolve, + }; + }); + try { + await writeTurnInput( + params.session, + `${JSON.stringify({ + type: "control_request", + request_id: requestId, + request: { + subtype: "set_model", + model: params.model, + system_prompt: params.systemPrompt, + }, + })}\n`, + ); + } catch { + settleClaudeLivePendingControlRequest(params.session, null); + } + return response; +} + +async function supportsClaudeLiveSystemPromptSwitch(params: { + session: ClaudeLiveSession; + model: string; +}): Promise { + if (params.session.systemPromptSwitchCapability !== "unknown") { + return params.session.systemPromptSwitchCapability === "supported"; + } + // Older CLIs may accept set_model while ignoring unknown fields. The current + // prompt-switch contract rejects an empty field with this exact validation + // error, so only that response is strong enough to weaken process identity. + const response = await requestClaudeLiveModelUpdate({ + session: params.session, + model: params.model, + systemPrompt: "", + }); + const supported = + response?.subtype === "error" && response.error === CLAUDE_LIVE_SYSTEM_PROMPT_PROBE_ERROR; + params.session.systemPromptSwitchCapability = supported ? "supported" : "unsupported"; + return supported; +} + +async function updateClaudeLiveSystemPrompt(params: { + session: ClaudeLiveSession; + model: string; + systemPrompt: string; +}): Promise { + const systemPrompt = stripSystemPromptCacheBoundary(params.systemPrompt); + if ( + !systemPrompt.trim() || + !(await supportsClaudeLiveSystemPromptSwitch({ + session: params.session, + model: params.model, + })) + ) { + return false; + } + const response = await requestClaudeLiveModelUpdate({ + session: params.session, + model: params.model, + systemPrompt, + }); + return response?.subtype === "success"; +} + +async function refreshClaudeLiveSystemPromptForReuse(params: { + session: ClaudeLiveSession; + context: PreparedCliRunContext; + systemPromptHash: string; +}): Promise { + if (params.session.systemPromptHash === params.systemPromptHash) { + return true; + } + const updated = await updateClaudeLiveSystemPrompt({ + session: params.session, + model: params.context.normalizedModel, + systemPrompt: params.context.systemPrompt, + }); + if (updated) { + params.session.systemPromptHash = params.systemPromptHash; + return true; + } + // Older or unhealthy Claude CLIs may reject the control frame. Restart so + // the next process still receives the current prompt through argv. + closeLiveSession(params.session, "restart"); + return false; +} + function writeClaudeLiveToolControlResponse(params: { session: ClaudeLiveSession; requestId: string; @@ -1230,6 +1392,9 @@ function handleClaudeLiveLine(session: ClaudeLiveSession, line: string): void { if (parsedSessionId) { session.sessionId = parsedSessionId; } + if (handleClaudeLiveControlResponse(session, parsed)) { + return; + } if (!turn) { return; } @@ -1350,6 +1515,7 @@ function handleClaudeExit(session: ClaudeLiveSession, exitCode: number | null): if (liveSessions.get(session.key) === session) { liveSessions.delete(session.key); } + settleClaudeLivePendingControlRequest(session, null); void cleanupLiveSession(session); if (!session.currentTurn) { return; @@ -1437,6 +1603,7 @@ async function createClaudeLiveSession(params: { env: Record; generation: string; fingerprint: string; + systemPromptHash: string; key: string; mcpCaptureKey?: string; noOutputTimeoutMs: number; @@ -1496,6 +1663,8 @@ async function createClaudeLiveSession(params: { key: params.key, generation: params.generation, fingerprint: params.fingerprint, + systemPromptHash: params.systemPromptHash, + systemPromptSwitchCapability: "unknown", managedRun, providerId: params.context.params.provider, modelId: params.context.modelId, @@ -1510,6 +1679,7 @@ async function createClaudeLiveSession(params: { }, cleanupPromise: null, closing: false, + pendingControlRequest: null, mcpCaptureKey: params.mcpCaptureKey, nativeToolApprovalGrants: new Set(), outstandingBackgroundTaskIds: new Set(), @@ -1671,8 +1841,7 @@ function createRequiredLiveSessionError(params: { }); } -/** Runs one prompt through a reusable Claude CLI live session. */ -export async function runClaudeLiveSessionTurn(params: { +type RunClaudeLiveSessionTurnParams = { context: PreparedCliRunContext; args: string[]; executableCommand?: string; @@ -1701,8 +1870,90 @@ export async function runClaudeLiveSessionTurn(params: { onRequestPayload?: (payload: string) => void; onPhase?: (phase: "send" | "resolve") => void; cleanup: () => Promise; -}): Promise { +}; + +async function abortClaudeLiveTurnBeforeStart( + cleanup: () => Promise, + abortError: Error, +): Promise { + try { + await cleanup(); + } catch (cleanupError) { + throw new Error("Claude live turn aborted before start and cleanup failed", { + cause: cleanupError, + }); + } + throw abortError; +} + +/** Runs one prompt through a reusable Claude CLI live session. */ +export function runClaudeLiveSessionTurn( + params: RunClaudeLiveSessionTurnParams, +): Promise { const key = buildClaudeLiveKey(params.context); + // Keep prompt refresh, turn assignment, and stdin writes under one owner lock. + // Callers normally arrive through the outer CLI queue, but this owner enforces + // the invariant itself so alternate callers cannot mutate an active process. + let cleanupPromise: Promise | undefined; + const cleanup = () => (cleanupPromise ??= Promise.resolve().then(params.cleanup)); + const abortSignal = params.context.params.abortSignal; + if (!abortSignal) { + return liveSessionTurns.enqueue(key, () => + runSerializedClaudeLiveSessionTurn(params, key, cleanup), + ); + } + if (abortSignal.aborted) { + return abortClaudeLiveTurnBeforeStart(cleanup, createAbortError(abortSignal.reason)); + } + return new Promise((resolve, reject) => { + let started = false; + let settled = false; + const settle = ( + outcome: { kind: "resolve"; value: ClaudeLiveRunResult } | { kind: "reject"; error: unknown }, + ) => { + if (settled) { + return; + } + settled = true; + abortSignal.removeEventListener("abort", onAbort); + if (outcome.kind === "resolve") { + resolve(outcome.value); + } else { + reject( + outcome.error instanceof Error + ? outcome.error + : new Error(formatErrorMessage(outcome.error)), + ); + } + }; + const onAbort = () => { + if (!started) { + void abortClaudeLiveTurnBeforeStart(cleanup, createAbortError(abortSignal.reason)).catch( + (error: unknown) => settle({ kind: "reject", error }), + ); + } + }; + abortSignal.addEventListener("abort", onAbort, { once: true }); + const queued = liveSessionTurns.enqueue(key, async () => { + started = true; + abortSignal.removeEventListener("abort", onAbort); + if (abortSignal.aborted) { + return await abortClaudeLiveTurnBeforeStart(cleanup, createAbortError(abortSignal.reason)); + } + return await runSerializedClaudeLiveSessionTurn(params, key, cleanup); + }); + void queued.then( + (value) => settle({ kind: "resolve", value }), + (error: unknown) => settle({ kind: "reject", error }), + ); + }); +} + +async function runSerializedClaudeLiveSessionTurn( + params: RunClaudeLiveSessionTurnParams, + key: string, + cleanup: () => Promise, +): Promise { const resumeCapable = Boolean(params.context.preparedBackend.backend.resumeArgs?.length); const execPermission = resolveClaudeLiveExecPermission(params.context); const argv = [ @@ -1721,15 +1972,8 @@ export async function runClaudeLiveSessionTurn(params: { argv, env: params.env, }); - let cleanupDone = false; + const systemPromptHash = sha256(stripSystemPromptCacheBoundary(params.context.systemPrompt)); let createdSessionForTurn = false; - const cleanup = async () => { - if (cleanupDone) { - return; - } - cleanupDone = true; - await params.cleanup(); - }; let session = liveSessions.get(key) ?? null; if ( session && @@ -1763,6 +2007,23 @@ export async function runClaudeLiveSessionTurn(params: { closeLiveSession(session, "restart"); session = null; } + if ( + session && + !(await refreshClaudeLiveSystemPromptForReuse({ + session, + context: params.context, + systemPromptHash, + })) + ) { + if (params.requiredSessionGeneration) { + await cleanup(); + throw createRequiredLiveSessionError({ + context: params.context, + code: "cli_live_session_changed", + }); + } + session = null; + } if (!session && params.requiredSessionGeneration) { const pendingGeneration = liveSessionCreates.get(key)?.generation; if (pendingGeneration !== params.requiredSessionGeneration) { @@ -1831,6 +2092,22 @@ export async function runClaudeLiveSessionTurn(params: { closeLiveSession(session, "restart"); session = null; } else { + if ( + !(await refreshClaudeLiveSystemPromptForReuse({ + session, + context: params.context, + systemPromptHash, + })) + ) { + if (params.requiredSessionGeneration) { + await cleanup(); + throw createRequiredLiveSessionError({ + context: params.context, + code: "cli_live_session_changed", + }); + } + session = null; + } cleanupTurnArtifacts = true; } } @@ -1860,6 +2137,7 @@ export async function runClaudeLiveSessionTurn(params: { env: params.env, generation, fingerprint, + systemPromptHash, key, mcpCaptureKey, noOutputTimeoutMs: params.noOutputTimeoutMs,