diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts index 30570029ac5f471f660798adcdccf86f2346a4e8..7ea84f9bda7d99756e1837a4eb0bbb384d6dec7d 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts @@ -31,7 +31,7 @@ import { ProviderThreadId, ProviderTurnId, } from "@t3tools/contracts"; -import { Cause, DateTime, Deferred, Effect, Layer, Queue, Random, Stream } from "effect"; +import { Cause, DateTime, Deferred, Effect, Layer, Queue, Random, Ref, Stream } from "effect"; import { ProviderAdapterProcessError, @@ -312,25 +312,12 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { readonly options: ClaudeQueryOptions; }) => query({ prompt: input.prompt, options: input.options }) as ClaudeQueryRuntime); - const sessions = new Map< - ReturnType, - ClaudeSessionContext - >(); + const sessions = new Map(); const runtimeEventQueue = yield* Queue.unbounded(); - const now = (): Effect.Effect => - DateTime.now.pipe(Effect.map(DateTime.formatIso)); - const randomId = (): Effect.Effect => Random.nextUUIDv4; - const makeEventId = (): Effect.Effect> => - randomId().pipe(Effect.map((id) => EventId.makeUnsafe(id))); - const makeEventStamp = (): Effect.Effect<{ - readonly eventId: ReturnType; - readonly createdAt: string; - }> => - Effect.all({ - eventId: makeEventId(), - createdAt: now(), - }); + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + const nextEventId = Effect.map(Random.nextUUIDv4, (id) => EventId.makeUnsafe(id)); + const makeEventStamp = () => Effect.all({ eventId: nextEventId, createdAt: nowIso }); const offerRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid); @@ -347,7 +334,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { Effect.gen(function* () { const threadId = context.session.threadId ?? - ProviderThreadId.makeUnsafe(`claude-thread-${yield* randomId()}`); + ProviderThreadId.makeUnsafe(`claude-thread-${yield* Random.nextUUIDv4}`); return { threadId, turns: context.turns.map((turn) => ({ @@ -360,9 +347,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { const updateResumeCursor = (context: ClaudeSessionContext): Effect.Effect => Effect.gen(function* () { const threadId = context.session.threadId; - if (!threadId) { - return; - } + if (!threadId) return; const resumeCursor = { threadId, @@ -374,7 +359,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { context.session = { ...context.session, resumeCursor, - updatedAt: yield* now(), + updatedAt: yield* nowIso, }; }); @@ -387,7 +372,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { const changed = context.session.threadId !== nextThreadId; if (changed) { - const updatedAt = yield* now(); + const updatedAt = yield* nowIso; context.session = { ...context.session, threadId: nextThreadId, @@ -487,7 +472,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { ...(errorMessage ? { errorMessage } : {}), }); - const updatedAt = yield* now(); + const updatedAt = yield* nowIso; context.turnState = undefined; context.session = { ...context.session, @@ -697,9 +682,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { options?: { readonly emitExitEvent?: boolean }, ): Effect.Effect => Effect.gen(function* () { - if (context.stopped) { - return; - } + if (context.stopped) return; context.stopped = true; @@ -725,11 +708,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { yield* completeTurn(context, "interrupted", "Session stopped."); } - yield* Queue.offer(context.promptQueue, { type: "terminate" }).pipe(Effect.ignore); + yield* Queue.shutdown(context.promptQueue); - yield* Effect.sync(() => context.query.close()).pipe(Effect.ignoreCause); + context.query.close(); - const updatedAt = yield* now(); + const updatedAt = yield* nowIso; context.session = { ...context.session, status: "closed", @@ -786,44 +769,31 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { }); } - const startedAt = yield* now(); - const sessionId = ProviderSessionId.makeUnsafe(`claude-session-${yield* randomId()}`); + const startedAt = yield* nowIso; + const sessionId = ProviderSessionId.makeUnsafe( + `claude-session-${yield* Random.nextUUIDv4}`, + ); const resumeState = readClaudeResumeState(input.resumeCursor); const threadId = resumeState?.threadId ?? - ProviderThreadId.makeUnsafe(`claude-thread-${yield* randomId()}`); + ProviderThreadId.makeUnsafe(`claude-thread-${yield* Random.nextUUIDv4}`); const promptQueue = yield* Queue.unbounded(); - const prompt: AsyncIterable = { - [Symbol.asyncIterator]() { - return { - next: async (): Promise> => { - const item = await Effect.runPromise(Queue.take(promptQueue)); - if (item.type === "terminate") { - return { - done: true, - value: undefined, - }; - } - - return { - done: false, - value: item.message, - }; - }, - }; - }, - }; + const prompt = Stream.fromQueue(promptQueue).pipe( + Stream.filter((item) => item.type === "message"), + Stream.map((item) => item.message), + Stream.toAsyncIterable, + ); const pendingApprovals = new Map(); const inFlightTools = new Map(); - let contextRef: ClaudeSessionContext | undefined; + const contextRef = yield* Ref.make(undefined); const canUseTool: CanUseTool = (toolName, toolInput, callbackOptions) => Effect.runPromise( Effect.gen(function* () { - const context = contextRef; + const context = yield* Ref.get(contextRef); if (!context) { return { behavior: "deny", @@ -839,7 +809,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { } satisfies PermissionResult; } - const requestId = ApprovalRequestId.makeUnsafe(yield* randomId()); + const requestId = ApprovalRequestId.makeUnsafe(yield* Random.nextUUIDv4); const requestKind = classifyRequestKind(toolName); const detail = summarizeToolRequest(toolName, toolInput); const decisionDeferred = yield* Deferred.make(); @@ -988,7 +958,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { lastThreadStartedId: undefined, stopped: false, }; - contextRef = context; + yield* Ref.set(contextRef, context); sessions.set(sessionId, context); const sessionStartedStamp = yield* makeEventStamp(); @@ -1027,16 +997,16 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { }); } - const turnId = ProviderTurnId.makeUnsafe(`claude-turn-${yield* randomId()}`); + const turnId = ProviderTurnId.makeUnsafe(`claude-turn-${yield* Random.nextUUIDv4}`); const turnState: ClaudeTurnState = { turnId, - assistantItemId: `claude-message-${yield* randomId()}`, - startedAt: yield* now(), + assistantItemId: `claude-message-${yield* Random.nextUUIDv4}`, + startedAt: yield* nowIso, items: [], messageCompleted: false, }; - const updatedAt = yield* now(); + const updatedAt = yield* nowIso; context.turnState = turnState; context.session = { ...context.session,