authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-27 12:22:26-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-05 23:00:41-08:00
log58427d565d91c27f1f37b3e0d1a396348c8afe6a
treecfe7379e1046ead98ddcbf117ba69ffdb1837da4
parent9b3dbada17dcbd0b96ca7eb6f79af7f7e0a416db

Refactor ClaudeCodeAdapter session flow and stream handling

- Replace manual prompt async iterator with `Stream.fromQueue(...).toAsyncIterable` - Use `Ref` for shared session context in tool approval callbacks - Simplify timestamp/ID generation and close sessions via queue shutdown

1 files changed, 32 insertions(+), 62 deletions(-)

apps/server/src/provider/Layers/ClaudeCodeAdapter.ts+32-62
......@@ -31,7 +31,7 @@ import {
3131 ProviderThreadId,
3232 ProviderTurnId,
3333} from "@t3tools/contracts";
34import { Cause, DateTime, Deferred, Effect, Layer, Queue, Random, Stream } from "effect";
34import { Cause, DateTime, Deferred, Effect, Layer, Queue, Random, Ref, Stream } from "effect";
3535
3636import {
3737 ProviderAdapterProcessError,
......@@ -312,25 +312,12 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
312312 readonly options: ClaudeQueryOptions;
313313 }) => query({ prompt: input.prompt, options: input.options }) as ClaudeQueryRuntime);
314314
315 const sessions = new Map<
316 ReturnType<typeof ProviderSessionId.makeUnsafe>,
317 ClaudeSessionContext
318 >();
315 const sessions = new Map<ProviderSessionId, ClaudeSessionContext>();
319316 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
320317
321 const now = (): Effect.Effect<string> =>
322 DateTime.now.pipe(Effect.map(DateTime.formatIso));
323 const randomId = (): Effect.Effect<string> => Random.nextUUIDv4;
324 const makeEventId = (): Effect.Effect<ReturnType<typeof EventId.makeUnsafe>> =>
325 randomId().pipe(Effect.map((id) => EventId.makeUnsafe(id)));
326 const makeEventStamp = (): Effect.Effect<{
327 readonly eventId: ReturnType<typeof EventId.makeUnsafe>;
328 readonly createdAt: string;
329 }> =>
330 Effect.all({
331 eventId: makeEventId(),
332 createdAt: now(),
333 });
318 const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
319 const nextEventId = Effect.map(Random.nextUUIDv4, (id) => EventId.makeUnsafe(id));
320 const makeEventStamp = () => Effect.all({ eventId: nextEventId, createdAt: nowIso });
334321
335322 const offerRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
336323 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid);
......@@ -347,7 +334,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
347334 Effect.gen(function* () {
348335 const threadId =
349336 context.session.threadId ??
350 ProviderThreadId.makeUnsafe(`claude-thread-${yield* randomId()}`);
337 ProviderThreadId.makeUnsafe(`claude-thread-${yield* Random.nextUUIDv4}`);
351338 return {
352339 threadId,
353340 turns: context.turns.map((turn) => ({
......@@ -360,9 +347,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
360347 const updateResumeCursor = (context: ClaudeSessionContext): Effect.Effect<void> =>
361348 Effect.gen(function* () {
362349 const threadId = context.session.threadId;
363 if (!threadId) {
364 return;
365 }
350 if (!threadId) return;
366351
367352 const resumeCursor = {
368353 threadId,
......@@ -374,7 +359,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
374359 context.session = {
375360 ...context.session,
376361 resumeCursor,
377 updatedAt: yield* now(),
362 updatedAt: yield* nowIso,
378363 };
379364 });
380365
......@@ -387,7 +372,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
387372 const changed = context.session.threadId !== nextThreadId;
388373
389374 if (changed) {
390 const updatedAt = yield* now();
375 const updatedAt = yield* nowIso;
391376 context.session = {
392377 ...context.session,
393378 threadId: nextThreadId,
......@@ -487,7 +472,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
487472 ...(errorMessage ? { errorMessage } : {}),
488473 });
489474
490 const updatedAt = yield* now();
475 const updatedAt = yield* nowIso;
491476 context.turnState = undefined;
492477 context.session = {
493478 ...context.session,
......@@ -697,9 +682,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
697682 options?: { readonly emitExitEvent?: boolean },
698683 ): Effect.Effect<void> =>
699684 Effect.gen(function* () {
700 if (context.stopped) {
701 return;
702 }
685 if (context.stopped) return;
703686
704687 context.stopped = true;
705688
......@@ -725,11 +708,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
725708 yield* completeTurn(context, "interrupted", "Session stopped.");
726709 }
727710
728 yield* Queue.offer(context.promptQueue, { type: "terminate" }).pipe(Effect.ignore);
711 yield* Queue.shutdown(context.promptQueue);
729712
730 yield* Effect.sync(() => context.query.close()).pipe(Effect.ignoreCause);
713 context.query.close();
731714
732 const updatedAt = yield* now();
715 const updatedAt = yield* nowIso;
733716 context.session = {
734717 ...context.session,
735718 status: "closed",
......@@ -786,44 +769,31 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
786769 });
787770 }
788771
789 const startedAt = yield* now();
790 const sessionId = ProviderSessionId.makeUnsafe(`claude-session-${yield* randomId()}`);
772 const startedAt = yield* nowIso;
773 const sessionId = ProviderSessionId.makeUnsafe(
774 `claude-session-${yield* Random.nextUUIDv4}`,
775 );
791776 const resumeState = readClaudeResumeState(input.resumeCursor);
792777 const threadId =
793778 resumeState?.threadId ??
794 ProviderThreadId.makeUnsafe(`claude-thread-${yield* randomId()}`);
779 ProviderThreadId.makeUnsafe(`claude-thread-${yield* Random.nextUUIDv4}`);
795780
796781 const promptQueue = yield* Queue.unbounded<PromptQueueItem>();
797 const prompt: AsyncIterable<SDKUserMessage> = {
798 [Symbol.asyncIterator]() {
799 return {
800 next: async (): Promise<IteratorResult<SDKUserMessage>> => {
801 const item = await Effect.runPromise(Queue.take(promptQueue));
802 if (item.type === "terminate") {
803 return {
804 done: true,
805 value: undefined,
806 };
807 }
808
809 return {
810 done: false,
811 value: item.message,
812 };
813 },
814 };
815 },
816 };
782 const prompt = Stream.fromQueue(promptQueue).pipe(
783 Stream.filter((item) => item.type === "message"),
784 Stream.map((item) => item.message),
785 Stream.toAsyncIterable,
786 );
817787
818788 const pendingApprovals = new Map<ApprovalRequestId, PendingApproval>();
819789 const inFlightTools = new Map<number, ToolInFlight>();
820790
821 let contextRef: ClaudeSessionContext | undefined;
791 const contextRef = yield* Ref.make<ClaudeSessionContext | undefined>(undefined);
822792
823793 const canUseTool: CanUseTool = (toolName, toolInput, callbackOptions) =>
824794 Effect.runPromise(
825795 Effect.gen(function* () {
826 const context = contextRef;
796 const context = yield* Ref.get(contextRef);
827797 if (!context) {
828798 return {
829799 behavior: "deny",
......@@ -839,7 +809,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
839809 } satisfies PermissionResult;
840810 }
841811
842 const requestId = ApprovalRequestId.makeUnsafe(yield* randomId());
812 const requestId = ApprovalRequestId.makeUnsafe(yield* Random.nextUUIDv4);
843813 const requestKind = classifyRequestKind(toolName);
844814 const detail = summarizeToolRequest(toolName, toolInput);
845815 const decisionDeferred = yield* Deferred.make<ProviderApprovalDecision>();
......@@ -988,7 +958,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
988958 lastThreadStartedId: undefined,
989959 stopped: false,
990960 };
991 contextRef = context;
961 yield* Ref.set(contextRef, context);
992962 sessions.set(sessionId, context);
993963
994964 const sessionStartedStamp = yield* makeEventStamp();
......@@ -1027,16 +997,16 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
1027997 });
1028998 }
1029999
1030 const turnId = ProviderTurnId.makeUnsafe(`claude-turn-${yield* randomId()}`);
1000 const turnId = ProviderTurnId.makeUnsafe(`claude-turn-${yield* Random.nextUUIDv4}`);
10311001 const turnState: ClaudeTurnState = {
10321002 turnId,
1033 assistantItemId: `claude-message-${yield* randomId()}`,
1034 startedAt: yield* now(),
1003 assistantItemId: `claude-message-${yield* Random.nextUUIDv4}`,
1004 startedAt: yield* nowIso,
10351005 items: [],
10361006 messageCompleted: false,
10371007 };
10381008
1039 const updatedAt = yield* now();
1009 const updatedAt = yield* nowIso;
10401010 context.turnState = turnState;
10411011 context.session = {
10421012 ...context.session,