Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
418 changes: 374 additions & 44 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts

Large diffs are not rendered by default.

196 changes: 144 additions & 52 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.testkit.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}),
close: Effect.void,
};
Expand Down Expand Up @@ -942,27 +956,93 @@ function replayQueryRunnerService(

function layerClaudeAgentSdkReplay(
transcript: ClaudeAgentSdkReplayTranscript,
configDir: string,
options: {
readonly replayGate?: ProviderReplayGate;
readonly queryRunner?: ClaudeQueryRunner;
} = {},
): Layer.Layer<ClaudeAdapterV2.ClaudeAgentSdkQueryRunner> {
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)),
),
});
}),
);
}
Expand Down Expand Up @@ -1018,40 +1098,47 @@ function layerClaudeProviderAdapterRegistryReplay(
options: {
readonly replayGate?: ProviderReplayGate;
readonly queryRunner?: ClaudeQueryRunner;
readonly configDir?: string;
} = {},
) {
const layerServerConfig = Layer.effect(
ServerConfig.ServerConfig,
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)),
);
}

Expand Down Expand Up @@ -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,
};
},
);
57 changes: 43 additions & 14 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand All @@ -11108,9 +11126,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
if (Option.isSome(conversationLine)) {
return false;
}
found = true;
}
return found;
return true;
},
);

Expand Down Expand Up @@ -11197,18 +11214,30 @@ 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 &&
nativeThreadId !== null &&
// 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,
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/orchestration-v2/ProviderSessionManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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),
),
Expand Down
Loading
Loading