diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 06b62e0e03b09..567f6fe306eef 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -32,6 +32,7 @@ import { ChatImageAttachment, CheckpointId, ClaudeSettings, + CommandId, EnvironmentId, MessageId, type ModelSelection, @@ -63,6 +64,7 @@ import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as PlatformError from "effect/PlatformError"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; @@ -99,7 +101,11 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import { ClaudeOrchestratorReplayHarness } from "./ClaudeAdapterV2.testkit.ts"; import { provideDeterministicTestRuntime } from "../testkit/DeterministicRuntime.ts"; import { runOrchestratorV2Scenario } from "../testkit/OrchestratorScenario.ts"; -import { layerProviderReplay } from "../testkit/ProviderReplayHarness.ts"; +import { layerProviderReplay, layerWithRegistry } from "../testkit/ProviderReplayHarness.ts"; +import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; +import * as Orchestrator from "../Orchestrator.ts"; +import * as EffectWorker from "../EffectWorker.ts"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; import { assertBaseProjection, assertSemanticProjectionIntegrity, @@ -1382,7 +1388,61 @@ describe("ClaudeAdapterV2 executable path", () => { }); describe("ClaudeAdapterV2 resumeThread", () => { - it.effect("fails for a transcript without messages so a fresh session replaces it", () => + const openRuntime = Effect.fnUntraced(function* ( + configDir: string, + fileSystem: FileSystem.FileSystem, + environment: NodeJS.ProcessEnv = { CLAUDE_CONFIG_DIR: configDir }, + ) { + const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ + instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, + settings: DEFAULT_CLAUDE_SETTINGS, + environment, + attachmentsDir: configDir, + fileSystem, + path: yield* Path.Path, + crypto: yield* Crypto.Crypto, + idAllocator: yield* IdAllocator.IdAllocatorV2, + queryRunner: { + allocateSessionId: Effect.succeed("native-thread-empty"), + open: () => + Effect.succeed({ + messages: Stream.never, + offer: () => Effect.void, + setModel: () => Effect.void, + applyFlagSettings: () => Effect.void, + setPermissionMode: () => Effect.void, + backgroundTasks: () => Effect.succeed(false), + stopTask: () => Effect.void, + interrupt: Effect.void, + close: Effect.void, + }), + forkSession: () => Effect.die("unused"), + subagentLaunchToolUseId: () => Effect.succeed(null), + assertComplete: Effect.void, + }, + }); + const threadId = ThreadId.make("thread-claude-empty-session"); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-claude-empty-session"), + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + return { runtime, providerThread, threadId }; + }); + + const testLayer = Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + ); + + it.effect("fails for missing or empty transcripts and retains saved conversations", () => Effect.scoped( Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1392,6 +1452,7 @@ describe("ClaudeAdapterV2 resumeThread", () => { }); const projectDir = path.join(configDir, "projects", "-workspace"); yield* fileSystem.makeDirectory(projectDir, { recursive: true }); + yield* fileSystem.writeFileString(path.join(configDir, "projects", "stray-file"), ""); yield* fileSystem.writeFileString( path.join(projectDir, "native-thread-empty.jsonl"), [ @@ -1405,8 +1466,92 @@ describe("ClaudeAdapterV2 resumeThread", () => { path.join(projectDir, "native-thread-saved.jsonl"), `{"type":"user","message":{"role":"user","content":"hello"},"sessionId":"native-thread-saved"}\n`, ); + const { runtime, providerThread } = yield* openRuntime(configDir, fileSystem); + + const error = yield* Effect.flip(runtime.resumeThread({ providerThread })); + assert.equal(error._tag, "ProviderAdapterResumeThreadError"); + + const saved = yield* runtime.resumeThread({ + providerThread: { + ...providerThread, + nativeThreadRef: { + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeId: "native-thread-saved", + strength: "strong", + }, + }, + }); + assert.equal(saved.nativeThreadRef?.nativeId, "native-thread-saved"); + + const missing = yield* Effect.flip( + runtime.resumeThread({ + providerThread: { + ...providerThread, + nativeThreadRef: { + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeId: "native-thread-missing", + strength: "strong", + }, + }, + }), + ); + assert.equal(missing._tag, "ProviderAdapterResumeThreadError"); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect.each(["HOME", "relative CLAUDE_CONFIG_DIR"] as const)( + "retains a saved conversation using %s in the thread runtime cwd", + (location) => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-resume-location-", + }); + const cwd = path.join(root, "thread"); + const configDir = path.join(cwd, ".claude"); + const projectDir = path.join(configDir, "projects", "-workspace"); + yield* fileSystem.makeDirectory(projectDir, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(projectDir, "native-thread-empty.jsonl"), + '{"type":"user","message":{"role":"user","content":"saved conversation"}}\n', + ); + const { runtime, providerThread } = yield* openRuntime( + configDir, + fileSystem, + location === "HOME" + ? { HOME: cwd, USERPROFILE: cwd } + : { CLAUDE_CONFIG_DIR: ".claude" }, + ); + const resumed = yield* runtime + .resumeThread({ + providerThread, + runtimePolicy: { ...CLAUDE_TEST_RUNTIME_POLICY, cwd }, + }) + .pipe(Effect.provideService(HostProcess.HomeDirectory, path.join(root, "server-home"))); + + assert.deepEqual(resumed.nativeThreadRef, providerThread.nativeThreadRef); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("replaces an unpersisted native session after warm Stop and immediate retry", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cwd = yield* checkpointWorkspace("claude-warm-resume"); + const configDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-warm-resume-", + }); + const opens: ClaudeAdapterV2.ClaudeAgentSdkQueryOpenInput[] = []; + const offers: SDKUserMessage[] = []; + let nativeId = 0; + let interrupts = 0; const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ - instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, + instanceId: CLAUDE_TEST_MODEL_SELECTION.instanceId, settings: DEFAULT_CLAUDE_SETTINGS, environment: { CLAUDE_CONFIG_DIR: configDir }, attachmentsDir: configDir, @@ -1415,58 +1560,223 @@ describe("ClaudeAdapterV2 resumeThread", () => { crypto: yield* Crypto.Crypto, idAllocator: yield* IdAllocator.IdAllocatorV2, queryRunner: { - allocateSessionId: Effect.succeed("native-thread-empty"), - open: () => Effect.die("unused"), + allocateSessionId: Effect.sync(() => `native-warm-${++nativeId}`), + open: (input) => + Effect.gen(function* () { + opens.push(input); + const messages = yield* Queue.unbounded(); + return { + messages: Stream.fromQueue(messages), + offer: (message) => + Effect.sync(() => { + offers.push(message); + }), + setModel: () => Effect.void, + applyFlagSettings: () => Effect.void, + setPermissionMode: () => Effect.void, + backgroundTasks: () => Effect.succeed(false), + stopTask: () => Effect.void, + interrupt: Effect.sync(() => { + interrupts++; + }), + close: Queue.shutdown(messages), + }; + }), forkSession: () => Effect.die("unused"), subagentLaunchToolUseId: () => Effect.succeed(null), assertComplete: Effect.void, }, }); - const threadId = ThreadId.make("thread-claude-empty-session"); - const runtime = yield* adapter.openSession({ - threadId, - providerSessionId: ProviderSessionId.make("provider-session-claude-empty-session"), - modelSelection: CLAUDE_TEST_MODEL_SELECTION, - runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, - }); - const providerThread = yield* runtime.ensureThread({ - threadId, - modelSelection: CLAUDE_TEST_MODEL_SELECTION, - runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread-claude-warm-resume"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("warm-create"), + threadId, + projectId: ProjectId.make("project-claude-warm-resume"), + title: "Warm resume", + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const dispatch = Effect.fnUntraced(function* (id: string, text: string) { + const running = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(id), + text, + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + return yield* orchestrator.getThreadProjection(threadId); + }); + const first = yield* dispatch("warm-first", "Remember the launch code is apricot."); + assert.equal(first.providerTurns[0]?.status, "running"); + assert.equal(opens[0]?.options.sessionId, "native-warm-1"); + assert.isUndefined(opens[0]?.options.resume); + assert.lengthOf(offers, 1); + assert.isFalse(yield* fileSystem.exists(path.join(configDir, "projects"))); + + const stopped = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === first.runs[0]?.id && + event.payload.status === "interrupted", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("warm-stop"), + threadId, + }); + yield* worker.drain(); + yield* Fiber.join(stopped); + yield* worker.drain(); + assert.equal(interrupts, 1); + assert.isFalse(yield* fileSystem.exists(path.join(configDir, "projects"))); + + // No clock advance, detach, policy change, or adapter resume call between turns. + const second = yield* dispatch("warm-retry", "What is the launch code?"); + assert.equal(second.providerTurns.at(-1)?.status, "running"); + assert.lengthOf(opens, 2); + assert.equal(opens[1]?.providerSessionId, opens[0]?.providerSessionId); + assert.equal(opens[1]?.options.cwd, opens[0]?.options.cwd); + assert.equal(opens[1]?.options.model, opens[0]?.options.model); + assert.equal(opens[1]?.options.permissionMode, opens[0]?.options.permissionMode); + assert.deepEqual( + opens.map(({ options }) => ({ sessionId: options.sessionId, resume: options.resume })), + [ + { sessionId: "native-warm-1", resume: undefined }, + { sessionId: "native-warm-2", resume: undefined }, + ], + ); + assert.equal(second.providerThreads[0]?.nativeThreadRef?.nativeId, "native-warm-2"); + assert.lengthOf(second.contextHandoffs, 1); + assert.equal(second.contextHandoffs[0]?.strategy, "full_thread_summary"); + assert.lengthOf(second.contextTransfers, 1); + assert.equal(second.contextTransfers[0]?.type, "provider_handoff"); + assert.equal(second.contextTransfers[0]?.targetRunId, second.runs[1]?.id); + assert.include(JSON.stringify(offers[1]), "apricot"); + }).pipe( + Effect.provide( + layerWithRegistry( + { name: "claude-warm-resume" }, + ProviderAdapterRegistry.layerSingle(adapter), + { runEffectWorker: false }, + ), + ), + ); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("fails when the Claude home has no projects directory", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const configDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-missing-projects-", }); + const { runtime, providerThread } = yield* openRuntime(configDir, fileSystem); const error = yield* Effect.flip(runtime.resumeThread({ providerThread })); + assert.equal(error._tag, "ProviderAdapterResumeThreadError"); + }), + ).pipe(Effect.provide(testLayer)), + ); - const saved = yield* runtime.resumeThread({ - providerThread: { - ...providerThread, - nativeThreadRef: { - driver: ClaudeAdapterV2.CLAUDE_PROVIDER, - nativeId: "native-thread-saved", - strength: "strong", - }, - }, - }); - assert.equal(saved.nativeThreadRef?.nativeId, "native-thread-saved"); + it.effect.each(["readDirectory", "exists", "stream"] as const)( + "retains the native session when transcript storage fails at %s", + (method) => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const configDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-unreadable-session-", + }); + const projectDir = path.join(configDir, "projects", "-workspace"); + yield* fileSystem.makeDirectory(projectDir, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(projectDir, "native-thread-empty.jsonl"), + '{"type":"user","message":{"role":"user","content":"hello"}}\n', + ); + const storageError = PlatformError.systemError({ + _tag: "PermissionDenied", + module: "FileSystem", + method, + pathOrDescriptor: projectDir, + }); + const { runtime, providerThread } = yield* openRuntime(configDir, { + ...fileSystem, + readDirectory: + method === "readDirectory" + ? () => Effect.fail(storageError) + : fileSystem.readDirectory, + exists: method === "exists" ? () => Effect.fail(storageError) : fileSystem.exists, + stream: method === "stream" ? () => Stream.fail(storageError) : fileSystem.stream, + }); - const missing = yield* runtime.resumeThread({ - providerThread: { - ...providerThread, - nativeThreadRef: { - driver: ClaudeAdapterV2.CLAUDE_PROVIDER, - nativeId: "native-thread-missing", - strength: "strong", - }, - }, + const resumed = yield* runtime.resumeThread({ providerThread }); + + assert.deepEqual(resumed.nativeThreadRef, providerThread.nativeThreadRef); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("retains a live query before its first transcript is persisted", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const configDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-live-session-", + }); + yield* fileSystem.makeDirectory(path.join(configDir, "projects", "-workspace"), { + recursive: true, }); - assert.equal(missing.nativeThreadRef?.nativeId, "native-thread-missing"); + const { runtime, providerThread, threadId } = yield* openRuntime(configDir, fileSystem); + yield* runtime.startTurn( + makeClaudeTestTurnInput({ + threadId, + providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-resume-live"), + text: "hello", + attachments: [], + }), + ); + + const resumed = yield* runtime.resumeThread({ providerThread }); + + assert.deepEqual(resumed.nativeThreadRef, providerThread.nativeThreadRef); }), - ).pipe( - Effect.provide( - Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), - ), - ), + ).pipe(Effect.provide(testLayer)), ); }); @@ -3213,6 +3523,21 @@ describe("ClaudeAdapterV2 background wake turns", () => { }; }); const makeWakeHarness = makeWakeHarnessWithOptions(); + // resumeThread refuses a native session the CLI never saved. + const makeResumableWakeHarness = Effect.fnUntraced(function* (nativeSessionId: string) { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const configDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-v2-wake-home-", + }); + const projectDir = path.join(configDir, "projects", "-workspace"); + yield* fileSystem.makeDirectory(projectDir, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(projectDir, `${nativeSessionId}.jsonl`), + `{"type":"user","message":{"role":"user","content":"hello"},"sessionId":"${nativeSessionId}"}\n`, + ); + return yield* makeWakeHarnessWithOptions({ environment: { CLAUDE_CONFIG_DIR: configDir } }); + }); it.effect.each(["release-after", "cancel-after", "release-before", "resume-pending"] as const)( "keeps attachment reservations across compaction: %s boundary", @@ -3339,7 +3664,10 @@ describe("ClaudeAdapterV2 background wake turns", () => { assert.equal(usage?.pagesUsed, undelivered ? 0 : 60); if (undelivered) assert.equal(usage?.bytesUsed, 0); else assert.isAbove(usage?.bytesUsed ?? 0, 25 * 1024 * 1024); - const next = outcome === "resume-pending" ? yield* makeWakeHarness : harness; + const next = + outcome === "resume-pending" + ? yield* makeResumableWakeHarness(WAKE_NATIVE_SESSION) + : harness; const providerThread = outcome === "resume-pending" ? yield* next.runtime.resumeThread({ providerThread: persisted }) @@ -3555,7 +3883,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { }), ) : persisted; - const resumed = yield* makeWakeHarness; + const resumed = yield* makeResumableWakeHarness( + fork ? "forked-native-session" : WAKE_NATIVE_SESSION, + ); const providerThread = yield* resumed.runtime.resumeThread({ providerThread: inherited }); yield* fileSystem.writeFileString( path.join(resumed.attachmentsDir, attachmentRelativePath(document)!), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts index c975e022b37df..cc54017569735 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts @@ -844,8 +844,22 @@ function makeReplayQueryRunner( replayEffect(() => { assertNextOutboundFrame({ type: "query.background_tasks", toolUseId }); }).pipe(Effect.as(true)), - interrupt: replayEffect(() => { - assertNextOutboundFrame({ type: "query.interrupt" }); + interrupt: Effect.tryPromise({ + try: async (signal) => { + // Persisting inbound frames can yield before the recorded interrupt point. + while (true) { + const entry = transcript.entries[cursor]; + if (entry?.type !== "emit_inbound") break; + throwIfFailed(); + if (entry.label !== undefined) { + replayOptions.replayGate?.release(entry.label); + } + await waitForCursorAdvance(cursorAdvanced.promise, signal); + if (signal.aborted) return; + } + assertNextOutboundFrame({ type: "query.interrupt" }); + }, + catch: (cause) => replayQueryRunnerError(transcript, cause), }), close: Effect.void, }; @@ -942,27 +956,93 @@ function replayQueryRunnerService( function layerClaudeAgentSdkReplay( transcript: ClaudeAgentSdkReplayTranscript, + configDir: string, options: { readonly replayGate?: ProviderReplayGate; readonly queryRunner?: ClaudeQueryRunner; } = {}, -): Layer.Layer { - if (options.queryRunner !== undefined) { - return Layer.succeed( - ClaudeAdapterV2.ClaudeAgentSdkQueryRunner, - replayQueryRunnerService(transcript, options.queryRunner), - ); - } - const queryRunner = makeReplayQueryRunner(transcript, options); +) { + const queryRunner = options.queryRunner ?? makeReplayQueryRunner(transcript, options); return Layer.effect( ClaudeAdapterV2.ClaudeAgentSdkQueryRunner, Effect.gen(function* () { - yield* Effect.addFinalizer(() => - Effect.sync(() => { - queryRunner.assertComplete(); - }), + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const projectDir = path.join(configDir, "projects", "replay-project"); + const transcriptPath = (sessionId: string) => path.join(projectDir, `${sessionId}.jsonl`); + const recordSchema = Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)); + const encodeRecord = Schema.encodeEffect(recordSchema); + const decodeRecord = Schema.decodeEffect(recordSchema); + if (options.queryRunner === undefined) { + yield* Effect.addFinalizer(() => + Effect.sync(() => { + queryRunner.assertComplete(); + }), + ); + } + const replay = replayQueryRunnerService(transcript, queryRunner); + + // The real CLI saves conversation frames before the adapter can resume them. + const persistMessage = Effect.fnUntraced( + function* (message: SDKMessage) { + if ( + (message.type !== "user" && message.type !== "assistant") || + message.parent_tool_use_id !== null || + typeof message.session_id !== "string" + ) { + return; + } + yield* fileSystem.makeDirectory(projectDir, { recursive: true }); + const record = yield* encodeRecord({ ...message, sessionId: message.session_id }); + yield* fileSystem.writeFileString(transcriptPath(message.session_id), `${record}\n`, { + flag: "a", + }); + }, + Effect.mapError((cause) => replayQueryRunnerError(transcript, cause)), ); - return replayQueryRunnerService(transcript, queryRunner); + + return ClaudeAdapterV2.ClaudeAgentSdkQueryRunner.of({ + ...replay, + open: (input) => + replay.open(input).pipe( + Effect.map((query) => ({ + ...query, + messages: query.messages.pipe(Stream.tap(persistMessage)), + })), + ), + forkSession: Effect.fnUntraced( + function* (input) { + const forked = yield* replay.forkSession(input); + const source = yield* fileSystem.readFileString(transcriptPath(input.sessionId)); + const records = yield* Effect.forEach(source.trimEnd().split("\n"), (line) => + decodeRecord(line), + ); + const boundary = input.options.upToMessageId; + const end = + boundary === undefined + ? records.length - 1 + : records.findIndex((record) => record.uuid === boundary); + if (end < 0) { + return yield* Effect.die( + new Error("Claude replay fork cursor is not in saved history."), + ); + } + const forkRecords = yield* Effect.forEach(records.slice(0, end + 1), (record) => + encodeRecord({ + ...record, + sessionId: forked.sessionId, + session_id: forked.sessionId, + }), + ); + yield* fileSystem.writeFileString( + transcriptPath(forked.sessionId), + `${forkRecords.join("\n")}\n`, + ); + return forked; + }, + Effect.mapError((cause) => replayQueryRunnerError(transcript, cause)), + ), + }); }), ); } @@ -1018,6 +1098,7 @@ function layerClaudeProviderAdapterRegistryReplay( options: { readonly replayGate?: ProviderReplayGate; readonly queryRunner?: ClaudeQueryRunner; + readonly configDir?: string; } = {}, ) { const layerServerConfig = Layer.effect( @@ -1025,33 +1106,39 @@ function layerClaudeProviderAdapterRegistryReplay( makeReplayServerConfig(transcript.scenario).pipe(Effect.orDie), ).pipe(Layer.provide(NodeServices.layer)); return Layer.unwrap( - writeReplayTaskOutputFiles(transcript).pipe( - Effect.orDie, - Effect.map((tmpRoot) => - ProviderAdapterRegistry.layerFromAdaptersEffect( - ClaudeAdapterV2.createClaudeAdapterV2( - { - instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, - displayName: undefined, - environment: [{ name: "CLAUDE_CODE_TMPDIR", value: tmpRoot, sensitive: false }], - enabled: true, - config: DEFAULT_CLAUDE_SETTINGS, - }, - {}, - ).pipe(Effect.map((adapter) => [adapter])), + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const tmpRoot = yield* writeReplayTaskOutputFiles(transcript); + const configDir = + options.configDir ?? + (yield* fileSystem.makeTempDirectoryScoped({ prefix: "claude-replay-home-" })); + return ProviderAdapterRegistry.layerFromAdaptersEffect( + ClaudeAdapterV2.createClaudeAdapterV2( + { + instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, + displayName: undefined, + environment: [ + { name: "CLAUDE_CODE_TMPDIR", value: tmpRoot, sensitive: false }, + { name: "CLAUDE_CONFIG_DIR", value: configDir, sensitive: false }, + ], + enabled: true, + config: DEFAULT_CLAUDE_SETTINGS, + }, + {}, + ).pipe(Effect.map((adapter) => [adapter])), + ).pipe( + Layer.provide( + Layer.mergeAll( + layerClaudeAgentSdkReplay(transcript, configDir, options).pipe( + Layer.provide(NodeServices.layer), + ), + IdAllocator.layer, + NodeServices.layer, + layerServerConfig, + ), ), - ), - Effect.provide(NodeServices.layer), - ), - ).pipe( - Layer.provide( - Layer.mergeAll( - layerClaudeAgentSdkReplay(transcript, options), - IdAllocator.layer, - NodeServices.layer, - layerServerConfig, - ), - ), + ); + }).pipe(Effect.orDie, Effect.provide(NodeServices.layer)), ); } @@ -2832,14 +2919,19 @@ export const ClaudeOrchestratorReplayHarness: OrchestratorV2ProviderReplayHarnes layerClaudeProviderAdapterRegistryReplay(transcript, options), }; -export function makeClaudeRestartReplayHarness(transcript: ClaudeAgentSdkReplayTranscript) { - const queryRunner = makeReplayQueryRunner(transcript); - return { - harness: { - ...ClaudeOrchestratorReplayHarness, - makeProviderAdapterRegistryLayer: (replayed) => - layerClaudeProviderAdapterRegistryReplay(replayed, { queryRunner }), - } satisfies typeof ClaudeOrchestratorReplayHarness, - assertComplete: replayQueryRunnerService(transcript, queryRunner).assertComplete, - }; -} +export const makeClaudeRestartReplayHarness = Effect.fn("makeClaudeRestartReplayHarness")( + function* (transcript: ClaudeAgentSdkReplayTranscript) { + const fileSystem = yield* FileSystem.FileSystem; + // Native storage outlives each simulated server's adapter registry. + const configDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "claude-replay-home-" }); + const queryRunner = makeReplayQueryRunner(transcript); + return { + harness: { + ...ClaudeOrchestratorReplayHarness, + makeProviderAdapterRegistryLayer: (replayed) => + layerClaudeProviderAdapterRegistryReplay(replayed, { queryRunner, configDir }), + } satisfies typeof ClaudeOrchestratorReplayHarness, + assertComplete: replayQueryRunnerService(transcript, queryRunner).assertComplete, + }; + }, +); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 4f26e50e57645..d0c1b5c3b971c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -11077,23 +11077,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); yield* Effect.addFinalizer(() => closeSession()); - // Metadata-only transcripts cannot be resumed or recreated; fail resumeThread to trigger a fresh session and handoff. + // A CLI stopped before it saved the first prompt can leave no transcript + // or only metadata lines. Reject those ids so the caller binds a fresh + // native session with a context handoff. const hasEmptyClaudeTranscript = Effect.fn("ClaudeAdapterV2.hasEmptyClaudeTranscript")( - function* (nativeThreadId: string) { + function* (nativeThreadId: string, cwd: string) { const configuredDir = adapterOptions.environment.CLAUDE_CONFIG_DIR?.trim() ?? ""; + const instanceHome = + (yield* HostProcess.Platform) === "win32" + ? adapterOptions.environment.USERPROFILE + : adapterOptions.environment.HOME; const projectsDir = path.join( configuredDir.length > 0 - ? path.resolve(session.cwd, configuredDir) - : path.join(yield* HostProcess.HomeDirectory, ".claude"), + ? path.resolve(cwd, configuredDir) + : path.join(instanceHome || (yield* HostProcess.HomeDirectory), ".claude"), "projects", ); - // Any message in any transcript copy means Claude may still resume it. - let found = false; - for (const project of yield* fileSystem.readDirectory(projectsDir)) { + const projects = yield* fileSystem.readDirectory(projectsDir).pipe( + Effect.catchTags({ + PlatformError: (error) => + error.reason._tag === "NotFound" ? Effect.succeed([]) : Effect.fail(error), + }), + ); + // Any saved conversation means the CLI may still resume the id. + for (const project of projects) { const transcriptPath = path.join(projectsDir, project, `${nativeThreadId}.jsonl`); // Stray files beside the project dirs fail with ENOTDIR. if ( - !(yield* fileSystem.exists(transcriptPath).pipe(Effect.orElseSucceed(() => false))) + !(yield* fileSystem.exists(transcriptPath).pipe( + Effect.catchTags({ + PlatformError: (error) => + field(error.reason.cause, "code") === "ENOTDIR" + ? Effect.succeed(false) + : Effect.fail(error), + }), + )) ) { continue; } @@ -11108,9 +11126,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (Option.isSome(conversationLine)) { return false; } - found = true; } - return found; + return true; }, ); @@ -11197,8 +11214,19 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ), ), ), + isThreadLoaded: (providerThread) => + Ref.get(queryContext).pipe( + Effect.map( + (query) => + query !== null && + query.nativeThreadId === providerThread.nativeThreadRef?.nativeId, + ), + ), resumeThread: Effect.fn("ClaudeAdapterV2.resumeThread")( - function* (threadInput: { readonly providerThread: OrchestrationV2ProviderThread }) { + function* (threadInput: { + readonly providerThread: OrchestrationV2ProviderThread; + readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy; + }) { const nativeThreadId = threadInput.providerThread.nativeThreadRef?.nativeId; if ( nativeThreadId !== undefined && @@ -11206,9 +11234,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // A live process holds the conversation even before saving it. (yield* Ref.get(queryContext))?.nativeThreadId !== nativeThreadId && // An unreadable transcript leaves the resume to the CLI. - (yield* hasEmptyClaudeTranscript(nativeThreadId).pipe( - Effect.orElseSucceed(() => false), - )) + (yield* hasEmptyClaudeTranscript( + nativeThreadId, + threadInput.runtimePolicy?.cwd ?? session.cwd, + ).pipe(Effect.orElseSucceed(() => false))) ) { return yield* new ProviderAdapter.ProviderAdapterProtocolError({ driver: CLAUDE_PROVIDER, diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 03a7278774a64..0f340e344dac2 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1766,6 +1766,10 @@ export const layerWithOptions = ( Effect.andThen( isProviderThreadLoaded({ providerSessionId, threadId, providerThreadKey }), ), + Effect.filterOrElse( + (loaded) => !loaded, + () => runtime.isThreadLoaded?.(input.providerThread) ?? Effect.succeed(true), + ), Effect.flatMap((loaded) => loaded ? Effect.succeed(input.providerThread) : runtime.resumeThread(input), ), diff --git a/apps/server/src/orchestration-v2/testkit/ClaudeReplayFixtures.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ClaudeReplayFixtures.integration.test.ts index 8886daad3e017..75627b7fb1d03 100644 --- a/apps/server/src/orchestration-v2/testkit/ClaudeReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ClaudeReplayFixtures.integration.test.ts @@ -12,9 +12,11 @@ import * as Effect from "effect/Effect"; import { ClaudeOrchestratorReplayHarness } from "../Adapters/ClaudeAdapterV2.testkit.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as Orchestrator from "../Orchestrator.ts"; +import * as EffectWorker from "../EffectWorker.ts"; import { userFacingDispatchErrorMessage } from "../UserFacingErrors.ts"; import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts"; import { subagentInput } from "./fixtures/subagent/input.ts"; +import { turnInterruptMidToolInput } from "./fixtures/turn_interrupt_mid_tool/input.ts"; import { runOrchestratorV2Scenario } from "./OrchestratorScenario.ts"; import * as ProviderReplayHarness from "./ProviderReplayHarness.ts"; import { materializeReplayTranscriptRuntimeInstructions } from "./ReplayRuntimeInstructions.ts"; @@ -30,6 +32,8 @@ import { type ReplayFixtureCase, } from "./ProviderReplayFixtures.ts"; import { readProviderReplayTranscript } from "@t3tools/provider-testing/replayTranscript"; +import { makeProviderReplayGate } from "@t3tools/provider-testing/replayGate"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; const readTranscript = Effect.fn("readClaudeReplayFixture")(function* (file: URL) { return yield* readProviderReplayTranscript(file); @@ -46,6 +50,71 @@ function runClaudeReplayFixture(input: ReplayFixtureCase) { describe("Claude Agent SDK replay fixtures", () => { itRunsReplayFixtures("claudeAgent", runClaudeReplayFixture); + it.effect("interrupts a replay while an inbound frame is held by a gate", () => + Effect.gen(function* () { + const recorded = yield* readClaudeTranscriptFixture("turn_interrupt_mid_tool"); + const interruptIndex = recorded.entries.findIndex( + (entry) => entry.type === "expect_outbound" && entry.label === "query.interrupt:1", + ); + assert.equal(recorded.entries[interruptIndex - 1]?.type, "emit_inbound"); + const transcript = yield* ClaudeOrchestratorReplayHarness.decodeTranscript({ + ...recorded, + entries: recorded.entries.map((entry, index) => + index === interruptIndex - 1 ? { ...entry, label: "before-interrupt" } : entry, + ), + }); + const materialized = yield* materializeFixtureInput({ + scenario: transcript.scenario, + fixtureInput: turnInterruptMidToolInput(), + driver: ProviderDriverKind.make("claudeAgent"), + modelSelection: CLAUDE_MODEL_SELECTION, + }).pipe(Effect.provide(IdAllocator.layer)); + const workspace = yield* checkpointWorkspace(transcript.scenario); + const replayGate = makeProviderReplayGate(["before-interrupt"]); + const scenario = { + name: `${transcript.scenario}:gated-interrupt`, + transcript, + ...materialized, + runtimePolicyOverride: { cwd: workspace }, + }; + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + for (const command of materialized.commands) { + if (command.type === "run.interrupt") { + assert.isTrue( + yield* Effect.promise(() => replayGate.waitForReached("before-interrupt")), + ); + const active = yield* orchestrator.getThreadProjection(command.threadId); + assert.equal(active.providerTurns[0]?.status, "running"); + } + yield* orchestrator.dispatch(command); + yield* worker.drain(); + } + const projection = yield* orchestrator.getThreadProjection( + materialized.projectionThreadIds[0]!, + ); + assert.deepEqual( + projection.runs.map((run) => run.status), + ["interrupted"], + ); + assert.equal(projection.providerTurns[0]?.status, "interrupted"); + assert.equal(projection.providerThreads[0]?.status, "idle"); + assert.equal( + projection.turnItems.find((item) => item.type === "run_interrupt_result")?.status, + "interrupted", + ); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerProviderReplay(scenario, ClaudeOrchestratorReplayHarness, { + replayGate, + runEffectWorker: false, + }), + ), + ); + }).pipe(provideDeterministicTestRuntime, Effect.provide(NodeServices.layer), Effect.scoped), + ); + it.effect("refuses messages to a native subagent thread without touching it", () => Effect.gen(function* () { const raw = yield* readClaudeTranscriptFixture("subagent"); diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts index d17b407506018..4706a571bc957 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts @@ -405,7 +405,7 @@ describe("orchestrator replay recovery", () => { }); const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } = splitAfterFirstIdle(materialized); - const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript); + const { harness, assertComplete } = yield* makeClaudeRestartReplayHarness(transcript); const options = { databaseLayer: SqlitePersistence.layerFromPath(path.join(tempDir, "state.sqlite")).pipe( Layer.provide(NodeServices.layer), @@ -517,7 +517,7 @@ describe("orchestrator replay recovery", () => { }); const { phase1Commands, phase1Steps, phase2Commands, phase2Steps } = splitAfterFirstIdle(materialized); - const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript); + const { harness, assertComplete } = yield* makeClaudeRestartReplayHarness(transcript); const options = { databaseLayer: SqlitePersistence.layerFromPath(path.join(tempDir, "state.sqlite")).pipe( Layer.provide(NodeServices.layer), diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRestartBackgroundNote.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRestartBackgroundNote.integration.test.ts index 4adc098ba6d7b..fb312da718e84 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRestartBackgroundNote.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRestartBackgroundNote.integration.test.ts @@ -156,7 +156,7 @@ const runRestart = Effect.fn("runRestart")(function* (input: { const firstIdle = materialized.steps.findIndex((step) => step.type === "await_thread_idle"); const phase1Steps = materialized.steps.slice(0, firstIdle + 1); const phase2Steps = materialized.steps.slice(firstIdle + 1); - const { harness, assertComplete } = makeClaudeRestartReplayHarness(transcript); + const { harness, assertComplete } = yield* makeClaudeRestartReplayHarness(transcript); const layerDatabase = SqlitePersistence.layerFromPath(path.join(tempDir, "state.sqlite")).pipe( Layer.provide(NodeServices.layer), ); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_result_is_error/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_result_is_error/output.ts index fe3167b22cbcd..75e8ed46fdd1f 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_result_is_error/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_result_is_error/output.ts @@ -46,7 +46,12 @@ export function assertClaudeResultIsErrorOutput( ); assert.isDefined(errorItem); if (errorItem?.type !== "error") throw new Error("expected error item"); - assert.equal(errorItem.failure.message, AUTH_FAILURE_TEXT); + const replayLocation = + / from "[^"]+", with CLAUDE_CONFIG_DIR set to "[^"]+[/\\]claude-replay-home-[^"]+"/.exec( + errorItem.failure.message, + ); + assert.isNotNull(replayLocation); + assert.equal(errorItem.failure.message.replace(replayLocation[0], ""), AUTH_FAILURE_TEXT); assert.equal(errorItem.failure.code, "api_error_401"); assert.equal(errorItem.failure.class, "provider_error"); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index a92256d7dd791..e703ac155835e 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -595,6 +595,10 @@ export interface ProviderAdapterV2SessionRuntime { readonly ensureThread: ( input: ProviderAdapterV2EnsureThreadInput, ) => Effect.Effect; + /** Verifies a cached load against the runtime's current native process state. */ + readonly isThreadLoaded?: ( + providerThread: OrchestrationV2ProviderThread, + ) => Effect.Effect; readonly resumeThread: (input: { readonly providerThread: OrchestrationV2ProviderThread; readonly threadId?: ThreadId;