diff --git a/AGENTS.md b/AGENTS.md index 37a3c14991c3e33b5c2e34eb645a13420e87cdfa..8df851c2e8e551df511f6186812dbce8ed51c905 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,4 +1,4 @@ -# CLAUDE.md +# AGENTS.md ## Task Completion Requirements diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts index db4f9c683554413cb8a51d3c9b9f0b39f4b45b89..86f43d91358b96dd484ecb4ad5a9cde5db4d719a 100644 --- a/apps/server/integration/TestProviderAdapter.integration.ts +++ b/apps/server/integration/TestProviderAdapter.integration.ts @@ -5,11 +5,14 @@ import { EventId, ProviderApprovalDecision, ProviderRuntimeEvent, + RuntimeSessionId, ProviderSession, ProviderSessionId, ProviderThreadId, ProviderTurnId, ProviderTurnStartResult, + ThreadId, + TurnId, } from "@t3tools/contracts"; import { Effect, Queue, Stream } from "effect"; @@ -25,13 +28,30 @@ import type { } from "../src/provider/Services/ProviderAdapter.ts"; export interface TestTurnResponse { - readonly events: ReadonlyArray; + readonly events: ReadonlyArray; readonly mutateWorkspace?: (input: { readonly cwd: string; readonly turnCount: number; }) => Effect.Effect; } +export type FixtureProviderRuntimeEvent = { + readonly type: string; + readonly eventId: EventId; + readonly provider: "codex" | "claudeCode" | "cursor"; + readonly sessionId: string; + readonly createdAt: string; + readonly threadId?: string | undefined; + readonly turnId?: string | undefined; + readonly itemId?: string | undefined; + readonly requestId?: string | undefined; + readonly payload?: unknown | undefined; + readonly [key: string]: unknown; +}; + +// Temporary alias while fixtures migrate to the new name. +export type LegacyProviderRuntimeEvent = FixtureProviderRuntimeEvent; + interface SessionState { readonly session: ProviderSession; snapshot: ProviderThreadSnapshot; @@ -40,6 +60,126 @@ interface SessionState { readonly rollbackCalls: Array; } +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function normalizeTurnState(value: unknown): "completed" | "failed" | "interrupted" | "cancelled" { + if ( + value === "completed" || + value === "failed" || + value === "interrupted" || + value === "cancelled" + ) { + return value; + } + return "completed"; +} + +function mapRequestType(requestKind: unknown): + | "command_execution_approval" + | "file_change_approval" + | "unknown" { + if (requestKind === "command") { + return "command_execution_approval"; + } + if (requestKind === "file-change") { + return "file_change_approval"; + } + return "unknown"; +} + +function mapItemType(toolKind: unknown): "command_execution" | "file_change" | "unknown" { + if (toolKind === "command") { + return "command_execution"; + } + if (toolKind === "file-change") { + return "file_change"; + } + return "unknown"; +} + +function normalizeFixtureEvent(rawEvent: Record): ProviderRuntimeEvent { + const type = typeof rawEvent.type === "string" ? rawEvent.type : ""; + switch (type) { + case "turn.started": + return { + ...rawEvent, + type: "turn.started", + payload: isRecord(rawEvent.payload) ? rawEvent.payload : {}, + } as ProviderRuntimeEvent; + case "turn.completed": + return { + ...rawEvent, + type: "turn.completed", + payload: isRecord(rawEvent.payload) + ? rawEvent.payload + : { + state: normalizeTurnState(rawEvent.status), + }, + } as ProviderRuntimeEvent; + case "message.delta": + return { + ...rawEvent, + type: "content.delta", + payload: { + streamKind: "assistant_text", + delta: typeof rawEvent.delta === "string" ? rawEvent.delta : "", + }, + } as ProviderRuntimeEvent; + case "message.completed": + return { + ...rawEvent, + type: "item.completed", + payload: { + itemType: "assistant_message", + ...(typeof rawEvent.detail === "string" ? { detail: rawEvent.detail } : {}), + }, + } as ProviderRuntimeEvent; + case "tool.started": + return { + ...rawEvent, + type: "item.started", + payload: { + itemType: mapItemType(rawEvent.toolKind), + ...(typeof rawEvent.title === "string" ? { title: rawEvent.title } : {}), + ...(typeof rawEvent.detail === "string" ? { detail: rawEvent.detail } : {}), + }, + } as ProviderRuntimeEvent; + case "tool.completed": + return { + ...rawEvent, + type: "item.completed", + payload: { + itemType: mapItemType(rawEvent.toolKind), + status: "completed", + ...(typeof rawEvent.title === "string" ? { title: rawEvent.title } : {}), + ...(typeof rawEvent.detail === "string" ? { detail: rawEvent.detail } : {}), + }, + } as ProviderRuntimeEvent; + case "approval.requested": + return { + ...rawEvent, + type: "request.opened", + payload: { + requestType: mapRequestType(rawEvent.requestKind), + ...(typeof rawEvent.detail === "string" ? { detail: rawEvent.detail } : {}), + }, + } as ProviderRuntimeEvent; + case "approval.resolved": + return { + ...rawEvent, + type: "request.resolved", + payload: { + requestType: mapRequestType(rawEvent.requestKind), + ...(typeof rawEvent.decision === "string" ? { decision: rawEvent.decision } : {}), + }, + } as ProviderRuntimeEvent; + default: + return rawEvent as ProviderRuntimeEvent; + } +} + export interface TestProviderAdapterHarness { readonly adapter: ProviderAdapterShape; readonly provider: "codex" | "claudeCode"; @@ -172,19 +312,28 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter ...(fixtureEvent as Record), eventId: randomUUID(), provider, - sessionId: input.sessionId, + sessionId: RuntimeSessionId.makeUnsafe(input.sessionId), createdAt: nowIso(), }; if (Object.hasOwn(rawEvent, "threadId")) { - rawEvent.threadId = state.snapshot.threadId; + rawEvent.threadId = ThreadId.makeUnsafe(state.snapshot.threadId); } if (Object.hasOwn(rawEvent, "turnId")) { - rawEvent.turnId = turnId; + rawEvent.turnId = TurnId.makeUnsafe(turnId); } - const runtimeEvent = rawEvent as ProviderRuntimeEvent; - if (runtimeEvent.type === "message.delta") { - assistantDeltas.push(runtimeEvent.delta); + const runtimeEvent = normalizeFixtureEvent(rawEvent); + const runtimeType = (runtimeEvent as { type: string }).type; + if (runtimeType === "content.delta") { + const payload = runtimeEvent.payload as { delta?: unknown } | undefined; + if (typeof payload?.delta === "string") { + assistantDeltas.push(payload.delta); + } + } else if (runtimeType === "message.delta") { + const legacyDelta = (runtimeEvent as { delta?: unknown }).delta; + if (typeof legacyDelta === "string") { + assistantDeltas.push(legacyDelta); + } } if (runtimeEvent.type === "turn.completed") { deferredTurnCompletedEvents.push(runtimeEvent); @@ -223,11 +372,13 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter type: "turn.completed", eventId: EventId.makeUnsafe(randomUUID()), provider, - sessionId: input.sessionId, + sessionId: RuntimeSessionId.makeUnsafe(input.sessionId), createdAt: nowIso(), - threadId: state.snapshot.threadId, - turnId, - status: "completed", + threadId: ThreadId.makeUnsafe(state.snapshot.threadId), + turnId: TurnId.makeUnsafe(turnId), + payload: { + state: "completed", + }, }); } else { for (const completedEvent of deferredTurnCompletedEvents) { @@ -325,6 +476,9 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter const adapter: ProviderAdapterShape = { provider, + capabilities: { + sessionModelSwitch: "in-session", + }, startSession, sendTurn, interruptTurn, diff --git a/apps/server/integration/fixtures/providerRuntime.ts b/apps/server/integration/fixtures/providerRuntime.ts index c846fcb704b298e65575de9e32d8d6685f8f1db9..f56a587cb7cfb1c7229a504a0ec5d60087cb952b 100644 --- a/apps/server/integration/fixtures/providerRuntime.ts +++ b/apps/server/integration/fixtures/providerRuntime.ts @@ -1,22 +1,16 @@ -import { - ApprovalRequestId, - EventId, - ProviderSessionId, - ProviderThreadId, - ProviderTurnId, - type ProviderRuntimeEvent, -} from "@t3tools/contracts"; +import { EventId, RuntimeRequestId } from "@t3tools/contracts"; +import type { LegacyProviderRuntimeEvent } from "../TestProviderAdapter.integration.ts"; const PROVIDER = "codex" as const; -const SESSION_ID = ProviderSessionId.makeUnsafe("fixture-session"); -const THREAD_ID = ProviderThreadId.makeUnsafe("fixture-thread"); -const TURN_ID = ProviderTurnId.makeUnsafe("fixture-turn"); -const REQUEST_ID = ApprovalRequestId.makeUnsafe("req-1"); +const SESSION_ID = "fixture-session"; +const THREAD_ID = "fixture-thread"; +const TURN_ID = "fixture-turn"; +const REQUEST_ID = RuntimeRequestId.makeUnsafe("req-1"); function baseEvent( eventId: string, createdAt: string, -): Pick { +): Pick { return { eventId: EventId.makeUnsafe(eventId), provider: PROVIDER, @@ -31,29 +25,38 @@ export const codexTurnTextFixture = [ ...baseEvent("evt-1", "2026-02-23T00:00:00.000Z"), threadId: THREAD_ID, turnId: TURN_ID, + payload: {}, }, { - type: "message.delta", + type: "content.delta", ...baseEvent("evt-2", "2026-02-23T00:00:00.100Z"), threadId: THREAD_ID, turnId: TURN_ID, - delta: "I will make a small update.\n", + payload: { + streamKind: "assistant_text", + delta: "I will make a small update.\n", + }, }, { - type: "message.delta", + type: "content.delta", ...baseEvent("evt-3", "2026-02-23T00:00:00.200Z"), threadId: THREAD_ID, turnId: TURN_ID, - delta: "Done.\n", + payload: { + streamKind: "assistant_text", + delta: "Done.\n", + }, }, { type: "turn.completed", ...baseEvent("evt-4", "2026-02-23T00:00:00.300Z"), threadId: THREAD_ID, turnId: TURN_ID, - status: "completed", + payload: { + state: "completed", + }, }, -] satisfies ReadonlyArray; +] satisfies ReadonlyArray; export const codexTurnToolFixture = [ { @@ -61,40 +64,51 @@ export const codexTurnToolFixture = [ ...baseEvent("evt-11", "2026-02-23T00:01:00.000Z"), threadId: THREAD_ID, turnId: TURN_ID, + payload: {}, }, { - type: "tool.started", + type: "item.started", ...baseEvent("evt-12", "2026-02-23T00:01:00.100Z"), threadId: THREAD_ID, turnId: TURN_ID, - toolKind: "command", - title: "Command run", - detail: "echo integration", + payload: { + itemType: "command_execution", + title: "Command run", + detail: "echo integration", + }, }, { - type: "tool.completed", + type: "item.completed", ...baseEvent("evt-13", "2026-02-23T00:01:00.200Z"), threadId: THREAD_ID, turnId: TURN_ID, - toolKind: "command", - title: "Command run", - detail: "echo integration", + payload: { + itemType: "command_execution", + status: "completed", + title: "Command run", + detail: "echo integration", + }, }, { - type: "message.delta", + type: "content.delta", ...baseEvent("evt-14", "2026-02-23T00:01:00.300Z"), threadId: THREAD_ID, turnId: TURN_ID, - delta: "Applied the requested edit.\n", + payload: { + streamKind: "assistant_text", + delta: "Applied the requested edit.\n", + }, }, { type: "turn.completed", ...baseEvent("evt-15", "2026-02-23T00:01:00.400Z"), threadId: THREAD_ID, turnId: TURN_ID, - status: "completed", + payload: { + state: "completed", + }, }, -] satisfies ReadonlyArray; +] satisfies ReadonlyArray; export const codexTurnApprovalFixture = [ { @@ -102,37 +116,47 @@ export const codexTurnApprovalFixture = [ ...baseEvent("evt-21", "2026-02-23T00:02:00.000Z"), threadId: THREAD_ID, turnId: TURN_ID, + payload: {}, }, { - type: "approval.requested", + type: "request.opened", ...baseEvent("evt-22", "2026-02-23T00:02:00.100Z"), threadId: THREAD_ID, turnId: TURN_ID, requestId: REQUEST_ID, - requestKind: "command", - detail: "Please approve command", + payload: { + requestType: "command_execution_approval", + detail: "Please approve command", + }, }, { - type: "approval.resolved", + type: "request.resolved", ...baseEvent("evt-23", "2026-02-23T00:02:00.200Z"), threadId: THREAD_ID, turnId: TURN_ID, requestId: REQUEST_ID, - requestKind: "command", - decision: "accept", + payload: { + requestType: "command_execution_approval", + decision: "accept", + }, }, { - type: "message.delta", + type: "content.delta", ...baseEvent("evt-24", "2026-02-23T00:02:00.300Z"), threadId: THREAD_ID, turnId: TURN_ID, - delta: "Approval received and command executed.\n", + payload: { + streamKind: "assistant_text", + delta: "Approval received and command executed.\n", + }, }, { type: "turn.completed", ...baseEvent("evt-25", "2026-02-23T00:02:00.400Z"), threadId: THREAD_ID, turnId: TURN_ID, - status: "completed", + payload: { + state: "completed", + }, }, -] satisfies ReadonlyArray; +] satisfies ReadonlyArray; diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 44bc2deb4494d2b1efae74be4713b6602f70b83d..ae617428146473a336b038d3893a42e7a18b81bc 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -45,6 +45,20 @@ const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); +type LegacyProviderRuntimeEvent = { + readonly type: string; + readonly eventId: EventId; + readonly provider: "codex" | "claudeCode" | "cursor"; + readonly sessionId: string; + readonly createdAt: string; + readonly threadId?: string | undefined; + readonly turnId?: string | undefined; + readonly itemId?: string | undefined; + readonly requestId?: string | undefined; + readonly payload?: unknown | undefined; + readonly [key: string]: unknown; +}; + function createProviderServiceHarness( cwd: string, hasSession = true, @@ -86,8 +100,8 @@ function createProviderServiceHarness( streamEvents: Stream.fromPubSub(runtimeEventPubSub), }; - const emit = (event: ProviderRuntimeEvent): void => { - Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); + const emit = (event: LegacyProviderRuntimeEvent): void => { + Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent)); }; return { @@ -363,7 +377,7 @@ describe("CheckpointReactor", () => { createdAt: new Date().toISOString(), threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), turnId: asProviderTurnId("turn-1"), - status: "completed", + payload: { state: "completed" }, }); await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); @@ -443,7 +457,7 @@ describe("CheckpointReactor", () => { createdAt: new Date().toISOString(), threadId: ProviderThreadId.makeUnsafe("provider-thread-aux"), turnId: asProviderTurnId("turn-aux"), - status: "completed", + payload: { state: "completed" }, }); await Effect.runPromise(Effect.sleep("40 millis")); @@ -461,7 +475,7 @@ describe("CheckpointReactor", () => { createdAt: new Date().toISOString(), threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), turnId: asProviderTurnId("turn-main"), - status: "completed", + payload: { state: "completed" }, }); const thread = await waitForThread( @@ -522,7 +536,7 @@ describe("CheckpointReactor", () => { createdAt: new Date().toISOString(), threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), turnId: asProviderTurnId("turn-claude-1"), - status: "completed", + payload: { state: "completed" }, }); await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); @@ -570,7 +584,7 @@ describe("CheckpointReactor", () => { createdAt: new Date().toISOString(), threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), turnId: asProviderTurnId("turn-missing-baseline"), - status: "completed", + payload: { state: "completed" }, }); await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); @@ -662,7 +676,7 @@ describe("CheckpointReactor", () => { createdAt: new Date().toISOString(), threadId: ProviderThreadId.makeUnsafe("provider-thread-missing"), turnId: asProviderTurnId("turn-missing-cwd"), - status: "completed", + payload: { state: "completed" }, }); await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); @@ -678,7 +692,7 @@ describe("CheckpointReactor", () => { ).toBe("v2\n"); }); - it("maps runtime checkpoint.captured into thread.turn.diff.complete when missing in read model", async () => { + it("ignores non-v2 checkpoint.captured runtime events", async () => { const harness = await createHarness(); const createdAt = new Date().toISOString(); @@ -715,11 +729,11 @@ describe("CheckpointReactor", () => { status: "completed", }); - const thread = await waitForThread(harness.engine, (entry) => - entry.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 3), - ); - expect(thread.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 3)).toBe( - true, + await Effect.runPromise(Effect.sleep("40 millis")); + const readModel = await Effect.runPromise(harness.engine.getReadModel()); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(thread?.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 3)).toBe( + false, ); }); @@ -757,15 +771,14 @@ describe("CheckpointReactor", () => { ); harness.provider.emit({ - type: "checkpoint.captured", + type: "turn.completed", eventId: EventId.makeUnsafe("evt-runtime-capture-failure"), provider: "codex", sessionId: asSessionId("sess-1"), createdAt: new Date().toISOString(), threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), turnId: asProviderTurnId("turn-runtime-failure"), - turnCount: 1, - status: "completed", + payload: { state: "completed" }, }); harness.provider.emit({ diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 71b51ba737be24dc3e84d075efad07e09fe6e9f8..deb246fb6b00eeff492f3bfd457e908018344f78 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -8,6 +8,7 @@ import { TurnId, type OrchestrationEvent, type ProviderRuntimeEvent, + RuntimeSessionId, } from "@t3tools/contracts"; import { Cause, Effect, Layer, Option, Queue, Stream } from "effect"; @@ -34,11 +35,15 @@ type ReactorInput = }; function toTurnId(value: string | undefined): TurnId | null { - return value === undefined ? null : TurnId.makeUnsafe(value); + return value === undefined ? null : TurnId.makeUnsafe(String(value)); } function toProviderThreadId(value: string | undefined): ProviderThreadId | null { - return value === undefined ? null : ProviderThreadId.makeUnsafe(value); + return value === undefined ? null : ProviderThreadId.makeUnsafe(String(value)); +} + +function toProviderSessionId(value: RuntimeSessionId): ProviderSessionId { + return ProviderSessionId.makeUnsafe(String(value)); } function sameId(left: string | null | undefined, right: string | null | undefined): boolean { @@ -172,6 +177,7 @@ const make = Effect.gen(function* () { const captureCheckpointFromTurnCompletion = Effect.fnUntraced(function* ( event: Extract, ) { + const providerSessionId = toProviderSessionId(event.sessionId); const turnId = toTurnId(event.turnId); if (!turnId) { return; @@ -179,7 +185,7 @@ const make = Effect.gen(function* () { const readModel = yield* orchestrationEngine.getReadModel(); const thread = readModel.threads.find( - (entry) => entry.session?.providerSessionId === event.sessionId, + (entry) => entry.session?.providerSessionId === providerSessionId, ); if (!thread) { return; @@ -266,7 +272,7 @@ const make = Effect.gen(function* () { ), Effect.tapError((error) => appendCaptureFailureActivity({ - sessionId: event.sessionId, + sessionId: providerSessionId, turnId, detail: `Checkpoint captured, but turn diff summary is unavailable: ${error.message}`, createdAt: event.createdAt, @@ -296,7 +302,7 @@ const make = Effect.gen(function* () { turnId, completedAt: now, checkpointRef: targetCheckpointRef, - status: checkpointStatusFromRuntime(event.status), + status: checkpointStatusFromRuntime(event.payload.state), files, assistantMessageId, checkpointTurnCount: nextTurnCount, @@ -314,7 +320,7 @@ const make = Effect.gen(function* () { summary: "Checkpoint captured", payload: { turnCount: nextTurnCount, - ...(event.status !== undefined ? { status: event.status } : {}), + status: event.payload.state, }, turnId, createdAt: now, @@ -326,6 +332,7 @@ const make = Effect.gen(function* () { const ensurePreTurnBaselineFromTurnStart = Effect.fnUntraced(function* ( event: Extract, ) { + const providerSessionId = toProviderSessionId(event.sessionId); const turnId = toTurnId(event.turnId); if (!turnId) { return; @@ -333,7 +340,7 @@ const make = Effect.gen(function* () { const readModel = yield* orchestrationEngine.getReadModel(); const thread = readModel.threads.find( - (entry) => entry.session?.providerSessionId === event.sessionId, + (entry) => entry.session?.providerSessionId === providerSessionId, ); if (!thread) { return; @@ -581,7 +588,7 @@ const make = Effect.gen(function* () { yield* captureCheckpointFromTurnCompletion(event).pipe( Effect.catch((error) => appendCaptureFailureActivity({ - sessionId: event.sessionId, + sessionId: toProviderSessionId(event.sessionId), turnId, detail: error.message, createdAt: new Date().toISOString(), @@ -590,89 +597,6 @@ const make = Effect.gen(function* () { ); return; } - - if (event.type === "checkpoint.captured") { - const turnId = toTurnId(event.turnId); - if (!turnId) { - return; - } - - const readModel = yield* orchestrationEngine.getReadModel(); - const thread = readModel.threads.find( - (entry) => entry.session?.providerSessionId === event.sessionId, - ); - if (!thread) { - return; - } - - if ( - thread.checkpoints.some( - (checkpoint) => - checkpoint.turnId === turnId || checkpoint.checkpointTurnCount === event.turnCount, - ) - ) { - return; - } - - const sessionRuntime = yield* resolveSessionRuntimeForThread(thread.id); - if (Option.isNone(sessionRuntime)) { - return; - } - - const fromTurnCount = Math.max(0, event.turnCount - 1); - const fromCheckpointRef = checkpointRefForThreadTurn(thread.id, fromTurnCount); - const targetCheckpointRef = checkpointRefForThreadTurn(thread.id, event.turnCount); - - const targetExists = yield* checkpointStore.hasCheckpointRef({ - cwd: sessionRuntime.value.cwd, - checkpointRef: targetCheckpointRef, - }); - if (!targetExists) { - yield* checkpointStore.captureCheckpoint({ - cwd: sessionRuntime.value.cwd, - checkpointRef: targetCheckpointRef, - }); - } - - const files = yield* checkpointStore - .diffCheckpoints({ - cwd: sessionRuntime.value.cwd, - fromCheckpointRef, - toCheckpointRef: targetCheckpointRef, - fallbackFromToHead: false, - }) - .pipe( - Effect.map((diff) => - parseTurnDiffFilesFromUnifiedDiff(diff).map((file) => ({ - path: file.path, - kind: "modified" as const, - additions: file.additions, - deletions: file.deletions, - })), - ), - Effect.catch(() => Effect.succeed([])), - ); - - const assistantMessageId = - thread.messages - .toReversed() - .find((entry) => entry.role === "assistant" && entry.turnId === turnId)?.id ?? - MessageId.makeUnsafe(`assistant:${turnId}`); - - yield* orchestrationEngine.dispatch({ - type: "thread.turn.diff.complete", - commandId: serverCommandId("checkpoint-runtime-captured-diff-complete"), - threadId: thread.id, - turnId, - completedAt: event.createdAt, - checkpointRef: targetCheckpointRef, - status: checkpointStatusFromRuntime(event.status), - files, - assistantMessageId, - checkpointTurnCount: event.turnCount, - createdAt: event.createdAt, - }); - } }); const processInput = ( @@ -717,11 +641,7 @@ const make = Effect.gen(function* () { yield* Effect.forkScoped( Stream.runForEach(providerService.streamEvents, (event) => { - if ( - event.type !== "turn.started" && - event.type !== "turn.completed" && - event.type !== "checkpoint.captured" - ) { + if (event.type !== "turn.started" && event.type !== "turn.completed") { return Effect.void; } return Queue.offer(queue, { source: "runtime", event }).pipe(Effect.asVoid); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 711394061752659809383846fd230aa7f47a0503..f94cd183c918aa839991836c4004e9c37195099a 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -130,7 +130,7 @@ describe("ProviderCommandReactor", () => { }), ); const interruptTurn = vi.fn((_: unknown) => Effect.void); - const respondToRequest = vi.fn((_: unknown) => Effect.void); + const respondToRequest = vi.fn(() => Effect.void); const stopSession = vi.fn((input: unknown) => Effect.sync(() => { const sessionId = @@ -777,7 +777,7 @@ describe("ProviderCommandReactor", () => { }); }); - it("surfaces stale provider approval request failures and resolves stale pending approvals", async () => { + it("surfaces stale provider approval request failures without faking approval resolution", async () => { const harness = await createHarness(); const now = new Date().toISOString(); harness.respondToRequest.mockImplementation(() => @@ -869,7 +869,7 @@ describe("ProviderCommandReactor", () => { activity.payload !== null && (activity.payload as Record).requestId === "approval-request-1", ); - expect(resolvedActivity).toBeDefined(); + expect(resolvedActivity).toBeUndefined(); }); it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => { diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 2946bab3ac3d9fa866ac7f076a6afbfe0168ca84..33f22e4a1b4667e0b457773413083e1589c03012 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -12,7 +12,7 @@ import { type ProviderSession, type TurnId, } from "@t3tools/contracts"; -import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect"; +import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Schema, Stream } from "effect"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts"; import { GitCore } from "../../git/Services/GitCore.ts"; @@ -84,7 +84,7 @@ function toErrorMessage(error: unknown): string { } function isUnknownPendingApprovalRequestError(error: unknown): boolean { - if (error instanceof ProviderAdapterRequestError) { + if (Schema.is(ProviderAdapterRequestError)(error)) { const detail = error.detail.toLowerCase(); return ( detail.includes("unknown pending approval request") || @@ -175,33 +175,6 @@ const make = Effect.gen(function* () { createdAt: input.createdAt, }); - const appendApprovalResolvedActivity = (input: { - readonly threadId: ThreadId; - readonly requestId: string; - readonly decision: "accept" | "acceptForSession" | "decline" | "cancel"; - readonly detail: string; - readonly createdAt: string; - }) => - orchestrationEngine.dispatch({ - type: "thread.activity.append", - commandId: serverCommandId("approval-resolved"), - threadId: input.threadId, - activity: { - id: EventId.makeUnsafe(crypto.randomUUID()), - tone: "approval", - kind: "approval.resolved", - summary: "Approval resolved", - payload: { - requestId: input.requestId, - decision: input.decision, - detail: input.detail, - }, - turnId: null, - createdAt: input.createdAt, - }, - createdAt: input.createdAt, - }); - const setThreadSession = (input: { readonly threadId: ThreadId; readonly session: OrchestrationSession; @@ -569,17 +542,7 @@ const make = Effect.gen(function* () { requestId: event.payload.requestId, }); - if (!isUnknownPendingApprovalRequestError(error)) { - return; - } - - yield* appendApprovalResolvedActivity({ - threadId: event.payload.threadId, - requestId: event.payload.requestId, - decision: event.payload.decision, - detail: "Approval request is no longer pending in the provider session.", - createdAt: event.payload.createdAt, - }); + if (!isUnknownPendingApprovalRequestError(error)) return; }), ), ); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 701417900ea2078ea31172946255c852283ba5ff..97001e29582a768e49f30bfa80e6737fe4c407a9 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -40,6 +40,20 @@ const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(va const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); +type LegacyProviderRuntimeEvent = { + readonly type: string; + readonly eventId: EventId; + readonly provider: "codex" | "claudeCode" | "cursor"; + readonly sessionId: string; + readonly createdAt: string; + readonly threadId?: string | undefined; + readonly turnId?: string | undefined; + readonly itemId?: string | undefined; + readonly requestId?: string | undefined; + readonly payload?: unknown | undefined; + readonly [key: string]: unknown; +}; + function createProviderServiceHarness() { const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); @@ -57,8 +71,8 @@ function createProviderServiceHarness() { streamEvents: Stream.fromPubSub(runtimeEventPubSub), }; - const emit = (event: ProviderRuntimeEvent): void => { - Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); + const emit = (event: LegacyProviderRuntimeEvent): void => { + Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent)); }; return { diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index fd47b3ba8de17bfb5888e4f9178635913b016c0e..25d291cf319739dd122cd5a89dbfbcf8ac76619a 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1,4 +1,5 @@ import { + ApprovalRequestId, type AssistantDeliveryMode, CommandId, MessageId, @@ -10,7 +11,8 @@ import { TurnId, type OrchestrationThreadActivity, type ProviderRuntimeEvent, - type ProviderSessionId, + ProviderSessionId, + RuntimeSessionId, } from "@t3tools/contracts"; import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Ref, Stream } from "effect"; @@ -50,12 +52,20 @@ type RuntimeIngestionInput = event: TurnStartRequestedDomainEvent; }; -function toTurnId(value: string | undefined): TurnId | undefined { - return value === undefined ? undefined : TurnId.makeUnsafe(value); +function toTurnId(value: TurnId | string | undefined): TurnId | undefined { + return value === undefined ? undefined : TurnId.makeUnsafe(String(value)); } function toProviderThreadId(value: string | undefined): ProviderThreadId | null { - return value === undefined ? null : ProviderThreadId.makeUnsafe(value); + return value === undefined ? null : ProviderThreadId.makeUnsafe(String(value)); +} + +function toProviderSessionId(value: RuntimeSessionId): ProviderSessionId { + return ProviderSessionId.makeUnsafe(String(value)); +} + +function toApprovalRequestId(value: string | undefined): ApprovalRequestId | undefined { + return value === undefined ? undefined : ApprovalRequestId.makeUnsafe(value); } function sameId(left: string | null | undefined, right: string | null | undefined): boolean { @@ -163,7 +173,12 @@ function isToolLifecycleItemType(itemType: string): boolean { function runtimeEventToActivities( event: ProviderRuntimeEvent, ): ReadonlyArray { - const maybeSequence = event.sessionSequence !== undefined ? { sequence: event.sessionSequence } : {}; + const maybeSequence = (() => { + const eventWithSequence = event as ProviderRuntimeEvent & { sessionSequence?: number }; + return eventWithSequence.sessionSequence !== undefined + ? { sequence: eventWithSequence.sessionSequence } + : {}; + })(); switch (event.type) { case "request.opened": { const requestKind = requestKindFromCanonicalRequestType(event.payload.requestType); @@ -180,7 +195,7 @@ function runtimeEventToActivities( ? "File-change approval requested" : "Approval requested", payload: { - requestId: event.requestId, + requestId: toApprovalRequestId(event.requestId), ...(requestKind ? { requestKind } : {}), requestType: event.payload.requestType, ...(event.payload.detail ? { detail: truncateDetail(event.payload.detail) } : {}), @@ -201,7 +216,7 @@ function runtimeEventToActivities( kind: "approval.resolved", summary: "Approval resolved", payload: { - requestId: event.requestId, + requestId: toApprovalRequestId(event.requestId), ...(requestKind ? { requestKind } : {}), requestType: event.payload.requestType, ...(event.payload.decision ? { decision: event.payload.decision } : {}), @@ -296,7 +311,7 @@ function runtimeEventToActivities( ? "Command approval requested" : "File-change approval requested", payload: { - requestId: event.requestId, + requestId: toApprovalRequestId(event.requestId), ...(requestKind ? { requestKind } : {}), ...(detail ? { detail: truncateDetail(detail) } : {}), }, @@ -317,7 +332,7 @@ function runtimeEventToActivities( kind: "approval.resolved", summary: "Approval resolved", payload: { - requestId: event.requestId, + requestId: toApprovalRequestId(event.requestId), ...(requestKind ? { requestKind } : {}), ...(decision ? { decision } : {}), }, @@ -528,9 +543,10 @@ const make = Effect.gen(function* () { const processRuntimeEvent = (event: ProviderRuntimeEvent) => Effect.gen(function* () { + const providerSessionId = toProviderSessionId(event.sessionId); const readModel = yield* orchestrationEngine.getReadModel(); const thread = readModel.threads.find( - (entry) => entry.session?.providerSessionId === event.sessionId, + (entry) => entry.session?.providerSessionId === providerSessionId, ); if (!thread) return; @@ -614,10 +630,8 @@ const make = Effect.gen(function* () { : activeTurnId; const providerThreadIdFromEvent = event.type === "thread.started" - ? ProviderThreadId.makeUnsafe(event.threadId) - : event.threadId !== undefined - ? ProviderThreadId.makeUnsafe(event.threadId) - : null; + ? toProviderThreadId(event.threadId) + : toProviderThreadId(event.threadId); const providerThreadId = providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null; const status = (() => { @@ -651,7 +665,7 @@ const make = Effect.gen(function* () { threadId: thread.id, status, providerName: event.provider, - providerSessionId: event.sessionId, + providerSessionId, providerThreadId, approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY, sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE, @@ -677,7 +691,7 @@ const make = Effect.gen(function* () { ); const turnId = toTurnId(event.turnId); if (turnId) { - yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); + yield* rememberAssistantMessageId(providerSessionId, turnId, assistantMessageId); } const assistantDeliveryMode = yield* Ref.get(assistantDeliveryModeRef); @@ -724,7 +738,7 @@ const make = Effect.gen(function* () { const assistantMessageId = assistantCompletion.messageId; const turnId = toTurnId(event.turnId); if (turnId) { - yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); + yield* rememberAssistantMessageId(providerSessionId, turnId, assistantMessageId); } yield* finalizeAssistantMessage({ @@ -735,18 +749,20 @@ const make = Effect.gen(function* () { createdAt: now, commandTag: "assistant-complete", finalDeltaCommandTag: "assistant-delta-finalize", - fallbackText: assistantCompletion.fallbackText, + ...(assistantCompletion.fallbackText !== undefined + ? { fallbackText: assistantCompletion.fallbackText } + : {}), }); if (turnId) { - yield* forgetAssistantMessageId(event.sessionId, turnId, assistantMessageId); + yield* forgetAssistantMessageId(providerSessionId, turnId, assistantMessageId); } } if (event.type === "turn.completed") { const turnId = toTurnId(event.turnId); if (turnId) { - const assistantMessageIds = yield* getAssistantMessageIdsForTurn(event.sessionId, turnId); + const assistantMessageIds = yield* getAssistantMessageIdsForTurn(providerSessionId, turnId); yield* Effect.forEach( assistantMessageIds, (assistantMessageId) => @@ -761,12 +777,12 @@ const make = Effect.gen(function* () { }), { concurrency: 1 }, ).pipe(Effect.asVoid); - yield* clearAssistantMessageIdsForTurn(event.sessionId, turnId); + yield* clearAssistantMessageIdsForTurn(providerSessionId, turnId); } } if (event.type === "session.exited") { - yield* clearTurnStateForSession(event.sessionId); + yield* clearTurnStateForSession(providerSessionId); } if (event.type === "runtime.error") { @@ -793,7 +809,7 @@ const make = Effect.gen(function* () { threadId: thread.id, status: "error", providerName: event.provider, - providerSessionId: event.sessionId, + providerSessionId, providerThreadId, approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY, sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE, diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts index 5c9314de3f9983c6102a019a8414f4cb8d040338..d576c61f7a33d2fb3166fb92f57cdb6e6ff61059 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts @@ -331,14 +331,14 @@ describe("ClaudeCodeAdapterLive", () => { const turnStarted = runtimeEvents[3]; assert.equal(turnStarted?.type, "turn.started"); if (turnStarted?.type === "turn.started") { - assert.equal(turnStarted.turnId, turn.turnId); + assert.equal(String(turnStarted.turnId), String(turn.turnId)); } const deltaEvent = runtimeEvents.find((event) => event.type === "content.delta"); assert.equal(deltaEvent?.type, "content.delta"); if (deltaEvent?.type === "content.delta") { assert.equal(deltaEvent.payload.delta, "Hi"); - assert.equal(deltaEvent.turnId, turn.turnId); + assert.equal(String(deltaEvent.turnId), String(turn.turnId)); } const toolStarted = runtimeEvents.find((event) => event.type === "item.started"); @@ -350,7 +350,7 @@ describe("ClaudeCodeAdapterLive", () => { const turnCompleted = runtimeEvents[runtimeEvents.length - 1]; assert.equal(turnCompleted?.type, "turn.completed"); if (turnCompleted?.type === "turn.completed") { - assert.equal(turnCompleted.turnId, turn.turnId); + assert.equal(String(turnCompleted.turnId), String(turn.turnId)); assert.equal(turnCompleted.payload.state, "completed"); } }).pipe( @@ -438,7 +438,7 @@ describe("ClaudeCodeAdapterLive", () => { assert.equal(deltaEvent?.type, "content.delta"); if (deltaEvent?.type === "content.delta") { assert.equal(deltaEvent.payload.delta, "Late text"); - assert.equal(deltaEvent.turnId, turn.turnId); + assert.equal(String(deltaEvent.turnId), String(turn.turnId)); } }).pipe( Effect.provideService(Random.Random, makeDeterministicRandomService()), @@ -506,7 +506,7 @@ describe("ClaudeCodeAdapterLive", () => { assert.equal(deltaEvent?.type, "content.delta"); if (deltaEvent?.type === "content.delta") { assert.equal(deltaEvent.payload.delta, "Fallback hello"); - assert.equal(deltaEvent.turnId, turn.turnId); + assert.equal(String(deltaEvent.turnId), String(turn.turnId)); } }).pipe( Effect.provideService(Random.Random, makeDeterministicRandomService()), @@ -631,10 +631,15 @@ describe("ClaudeCodeAdapterLive", () => { if (requested.value.type !== "request.opened") { return; } + const runtimeRequestId = requested.value.requestId; + assert.equal(typeof runtimeRequestId, "string"); + if (runtimeRequestId === undefined) { + return; + } yield* adapter.respondToRequest( session.sessionId, - ApprovalRequestId.makeUnsafe(requested.value.requestId), + ApprovalRequestId.makeUnsafe(runtimeRequestId), "accept", ); @@ -744,7 +749,7 @@ describe("ClaudeCodeAdapterLive", () => { const firstCompleted = yield* Fiber.join(firstCompletedFiber); assert.equal(firstCompleted._tag, "Some"); if (firstCompleted._tag === "Some" && firstCompleted.value.type === "turn.completed") { - assert.equal(firstCompleted.value.turnId, firstTurn.turnId); + assert.equal(String(firstCompleted.value.turnId), String(firstTurn.turnId)); } const secondTurn = yield* adapter.sendTurn({ @@ -770,7 +775,7 @@ describe("ClaudeCodeAdapterLive", () => { const secondCompleted = yield* Fiber.join(secondCompletedFiber); assert.equal(secondCompleted._tag, "Some"); if (secondCompleted._tag === "Some" && secondCompleted.value.type === "turn.completed") { - assert.equal(secondCompleted.value.turnId, secondTurn.turnId); + assert.equal(String(secondCompleted.value.turnId), String(secondTurn.turnId)); } const threadBeforeRollback = yield* adapter.readThread(session.sessionId); @@ -875,7 +880,7 @@ describe("ClaudeCodeAdapterLive", () => { assert.equal(records.some((record) => record.event?.provider === "claudeCode"), true); assert.equal(records.some((record) => record.event?.sessionId === session.sessionId), true); - assert.equal(records.some((record) => record.event?.turnId === turn.turnId), true); + assert.equal(records.some((record) => String(record.event?.turnId) === String(turn.turnId)), true); assert.equal( records.some( (record) => record.event?.method === "claude/stream_event/content_block_delta/text_delta", diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts index 9d8f65f169ba54804647e1644b6c244c267d6201..4e777c8524bd53f6d8d62009c264c08c1d1a4b5e 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts @@ -31,6 +31,11 @@ import { type ProviderSession, ProviderThreadId, ProviderTurnId, + RuntimeItemId, + RuntimeRequestId, + RuntimeSessionId, + ThreadId, + TurnId, } from "@t3tools/contracts"; import { Cause, DateTime, Deferred, Effect, Layer, Queue, Random, Ref, Stream } from "effect"; @@ -123,9 +128,7 @@ export interface ClaudeCodeAdapterLiveOptions { } function isUuid(value: string): boolean { - return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test( - value, - ); + return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value); } function isSyntheticClaudeThreadId(value: string): boolean { @@ -139,6 +142,26 @@ function toMessage(cause: unknown, fallback: string): string { return fallback; } +function asRuntimeSessionId(value: ProviderSessionId): RuntimeSessionId { + return RuntimeSessionId.makeUnsafe(value); +} + +function asRuntimeThreadId(value: ProviderThreadId): ThreadId { + return ThreadId.makeUnsafe(value); +} + +function asRuntimeTurnId(value: ProviderTurnId): TurnId { + return TurnId.makeUnsafe(value); +} + +function asRuntimeItemId(value: string): RuntimeItemId { + return RuntimeItemId.makeUnsafe(value); +} + +function asRuntimeRequestId(value: ApprovalRequestId): RuntimeRequestId { + return RuntimeRequestId.makeUnsafe(value); +} + function toPermissionMode(value: unknown): PermissionMode | undefined { switch (value) { case "default": @@ -315,7 +338,11 @@ function extractAssistantText(message: SDKMessage): string { continue; } const candidate = block as { type?: unknown; text?: unknown }; - if (candidate.type === "text" && typeof candidate.text === "string" && candidate.text.length > 0) { + if ( + candidate.type === "text" && + typeof candidate.text === "string" && + candidate.text.length > 0 + ) { fragments.push(candidate.text); } } @@ -388,7 +415,9 @@ function sdkNativeMethod(message: SDKMessage): string { const streamType = sdkMessageType(message.event); if (streamType) { const deltaType = - streamType === "content_block_delta" ? sdkMessageType((message.event as { delta?: unknown }).delta) : undefined; + streamType === "content_block_delta" + ? sdkMessageType((message.event as { delta?: unknown }).delta) + : undefined; if (deltaType) { return `claude/${message.type}/${streamType}/${deltaType}`; } @@ -466,11 +495,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { : crypto.randomUUID(), kind: "notification", provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: observedAt, method: sdkNativeMethod(message), ...(typeof message.session_id === "string" ? { threadId: message.session_id } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}), payload: message, }, @@ -485,7 +514,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { id: ReturnType; items: ReadonlyArray; }>; - }> => + }, ProviderAdapterValidationError> => Effect.gen(function* () { const threadId = context.session.threadId; if (!threadId) { @@ -552,9 +581,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "thread.started", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - threadId: nextThreadId, + threadId: asRuntimeThreadId(nextThreadId), payload: { providerThreadId: nextThreadId, }, @@ -588,10 +617,10 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "runtime.error", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(turnState ? { turnId: turnState.turnId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...(turnState ? { turnId: asRuntimeTurnId(turnState.turnId) } : {}), payload: { message, class: "provider_error", @@ -617,10 +646,10 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "runtime.warning", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(turnState ? { turnId: turnState.turnId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...(turnState ? { turnId: asRuntimeTurnId(turnState.turnId) } : {}), payload: { message, ...(detail !== undefined ? { detail } : {}), @@ -647,9 +676,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "turn.completed", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), payload: { state: status, ...(result?.stop_reason !== undefined ? { stopReason: result.stop_reason } : {}), @@ -675,11 +704,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "content.delta", eventId: deltaStamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: deltaStamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: turnState.turnId, - itemId: ProviderItemId.makeUnsafe(turnState.assistantItemId), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(turnState.turnId), + itemId: asRuntimeItemId(turnState.assistantItemId), payload: { streamKind: "assistant_text", delta: turnState.fallbackAssistantText, @@ -698,11 +727,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "item.completed", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - itemId: ProviderItemId.makeUnsafe(turnState.assistantItemId), - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: turnState.turnId, + itemId: asRuntimeItemId(turnState.assistantItemId), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(turnState.turnId), payload: { itemType: "assistant_message", status: "completed", @@ -727,10 +756,10 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "turn.completed", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: turnState.turnId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(turnState.turnId), payload: { state: status, ...(result?.stop_reason !== undefined ? { stopReason: result.stop_reason } : {}), @@ -788,11 +817,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "content.delta", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: context.turnState.turnId, - itemId: ProviderItemId.makeUnsafe(context.turnState.assistantItemId), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(context.turnState.turnId), + itemId: asRuntimeItemId(context.turnState.assistantItemId), payload: { streamKind: streamKindFromDeltaType(event.delta.type), delta: event.delta.text, @@ -846,11 +875,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "item.started", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - itemId: ProviderItemId.makeUnsafe(tool.itemId), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + itemId: asRuntimeItemId(tool.itemId), payload: { itemType: tool.itemType, status: "inProgress", @@ -889,11 +918,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "item.completed", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - itemId: ProviderItemId.makeUnsafe(tool.itemId), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + itemId: asRuntimeItemId(tool.itemId), payload: { itemType: tool.itemType, status: "completed", @@ -942,11 +971,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "item.updated", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: context.turnState.turnId, - itemId: ProviderItemId.makeUnsafe(context.turnState.assistantItemId), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(context.turnState.turnId), + itemId: asRuntimeItemId(context.turnState.assistantItemId), payload: { itemType: "assistant_message", status: "inProgress", @@ -1003,10 +1032,10 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { const base = { eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), providerRefs: { providerSessionId: context.session.sessionId, ...(context.session.threadId ? { providerThreadId: context.session.threadId } : {}), @@ -1164,10 +1193,10 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { const base = { eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), providerRefs: { providerSessionId: context.session.sessionId, ...(context.session.threadId ? { providerThreadId: context.session.threadId } : {}), @@ -1303,11 +1332,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "request.resolved", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - requestId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + requestId: asRuntimeRequestId(requestId), payload: { requestType: pending.requestType, decision: "cancel", @@ -1344,9 +1373,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "session.exited", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), payload: { reason: "Session stopped", exitKind: "graceful", @@ -1450,11 +1479,11 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "request.opened", eventId: requestedStamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: requestedStamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - requestId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + requestId: asRuntimeRequestId(requestId), payload: { requestType, detail, @@ -1466,7 +1495,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { }, providerRefs: { providerSessionId: context.session.sessionId, - ...(context.session.threadId ? { providerThreadId: context.session.threadId } : {}), + ...(context.session.threadId + ? { providerThreadId: context.session.threadId } + : {}), ...(context.turnState ? { providerTurnId: context.turnState.turnId } : {}), providerRequestId: requestId, }, @@ -1502,18 +1533,20 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "request.resolved", eventId: resolvedStamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: resolvedStamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - requestId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + requestId: asRuntimeRequestId(requestId), payload: { requestType, decision, }, providerRefs: { providerSessionId: context.session.sessionId, - ...(context.session.threadId ? { providerThreadId: context.session.threadId } : {}), + ...(context.session.threadId + ? { providerThreadId: context.session.threadId } + : {}), ...(context.turnState ? { providerTurnId: context.turnState.turnId } : {}), providerRequestId: requestId, }, @@ -1628,9 +1661,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "session.started", eventId: sessionStartedStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: sessionStartedStamp.createdAt, - ...(threadId ? { threadId } : {}), + ...(threadId ? { threadId: asRuntimeThreadId(threadId) } : {}), payload: input.resumeCursor !== undefined ? { resume: input.resumeCursor } : {}, providerRefs: { providerSessionId: sessionId, @@ -1643,9 +1676,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "session.configured", eventId: configuredStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: configuredStamp.createdAt, - ...(threadId ? { threadId } : {}), + ...(threadId ? { threadId: asRuntimeThreadId(threadId) } : {}), payload: { config: { ...(input.model ? { model: input.model } : {}), @@ -1667,9 +1700,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "session.state.changed", eventId: readyStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: readyStamp.createdAt, - ...(threadId ? { threadId } : {}), + ...(threadId ? { threadId: asRuntimeThreadId(threadId) } : {}), payload: { state: "ready", }, @@ -1705,10 +1738,10 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { }); } - const turnId = ProviderTurnId.makeUnsafe(`claude-turn-${yield* Random.nextUUIDv4}`); + const turnId = ProviderTurnId.makeUnsafe(yield* Random.nextUUIDv4); const turnState: ClaudeTurnState = { turnId, - assistantItemId: `claude-message-${yield* Random.nextUUIDv4}`, + assistantItemId: yield* Random.nextUUIDv4, startedAt: yield* nowIso, items: [], messageCompleted: false, @@ -1730,10 +1763,10 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { type: "turn.started", eventId: turnStartedStamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: turnStartedStamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(turnId), payload: { ...(input.model ? { model: input.model } : {}), ...(input.effort ? { effort: input.effort } : {}), diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 9dd57bd2bc6f23394fbd64585983e36c95cdc06d..dfd5a52caafc4f764aeb10a9ea816c6b68d969cf 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -11,10 +11,16 @@ import { type CanonicalRequestType, type ProviderEvent, type ProviderRuntimeEvent, + RuntimeItemId, + RuntimeRequestId, + RuntimeSessionId, ProviderApprovalDecision, + ProviderItemId, ProviderSessionId, ProviderThreadId, ProviderTurnId, + ThreadId, + TurnId, } from "@t3tools/contracts"; import { Effect, FileSystem, Layer, Option, Queue, Schema, Stream } from "effect"; @@ -278,7 +284,27 @@ function contentStreamKindFromMethod( } } -function eventRawSource(event: ProviderEvent): ProviderRuntimeEvent["raw"]["source"] { +function asRuntimeSessionId(sessionId: ProviderSessionId): RuntimeSessionId { + return RuntimeSessionId.makeUnsafe(sessionId); +} + +function asRuntimeThreadId(threadId: ProviderThreadId): ThreadId { + return ThreadId.makeUnsafe(threadId); +} + +function asRuntimeTurnId(turnId: ProviderTurnId): TurnId { + return TurnId.makeUnsafe(turnId); +} + +function asRuntimeItemId(itemId: ProviderItemId): RuntimeItemId { + return RuntimeItemId.makeUnsafe(itemId); +} + +function asRuntimeRequestId(requestId: string): RuntimeRequestId { + return RuntimeRequestId.makeUnsafe(requestId); +} + +function eventRawSource(event: ProviderEvent): NonNullable["source"] { return event.kind === "request" ? "codex.app-server.request" : "codex.app-server.notification"; } @@ -301,12 +327,12 @@ function runtimeEventBase(event: ProviderEvent): Omit { const turnStarted = events[6]; assert.equal(turnStarted?.type, "turn.started"); if (turnStarted?.type === "turn.started") { - assert.equal(turnStarted.turnId, turn.turnId); + assert.equal(String(turnStarted.turnId), String(turn.turnId)); } const completion = events[12]; @@ -448,10 +448,15 @@ describe("CursorAdapterLive", () => { if (opened.value.type !== "request.opened") { return; } + const runtimeRequestId = opened.value.requestId; + assert.equal(typeof runtimeRequestId, "string"); + if (runtimeRequestId === undefined) { + return; + } yield* adapter.respondToRequest( session.sessionId, - ApprovalRequestId.makeUnsafe(opened.value.requestId), + ApprovalRequestId.makeUnsafe(runtimeRequestId), "acceptForSession", ); diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index 2b76b6c965e4c173f9a078ba677900971aa61d13..fc2d1c3da3fa6a3ff32830179bc01a4ff0b61b86 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -16,12 +16,18 @@ import { type CanonicalItemType, type CanonicalRequestType, type ProviderApprovalPolicy, + ProviderItemId, ProviderSessionId, type ProviderRuntimeEvent, type ProviderSession, ProviderThreadId, ProviderTurnId, type ProviderTurnStartResult, + RuntimeItemId, + RuntimeRequestId, + RuntimeSessionId, + ThreadId, + TurnId, } from "@t3tools/contracts"; import { DateTime, Effect, Layer, Queue, Random, Schema, Stream } from "effect"; @@ -67,7 +73,7 @@ interface PendingPermission { interface CursorTurnState { readonly turnId: ReturnType; - readonly assistantItemId: string; + readonly assistantItemId: ReturnType; readonly startedToolCalls: Set; readonly items: Array; } @@ -106,6 +112,30 @@ function toMessage(cause: unknown, fallback: string): string { return fallback; } +function asRuntimeSessionId(value: ProviderSessionId): RuntimeSessionId { + return RuntimeSessionId.makeUnsafe(value); +} + +function asRuntimeThreadId(value: ProviderThreadId): ThreadId { + return ThreadId.makeUnsafe(value); +} + +function asRuntimeTurnId(value: ProviderTurnId): TurnId { + return TurnId.makeUnsafe(value); +} + +function asRuntimeItemId(value: string): RuntimeItemId { + return RuntimeItemId.makeUnsafe(value); +} + +function asProviderItemId(value: string): ProviderItemId { + return ProviderItemId.makeUnsafe(value); +} + +function asRuntimeRequestId(value: ApprovalRequestId): RuntimeRequestId { + return RuntimeRequestId.makeUnsafe(value); +} + function toSessionError( sessionId: ReturnType, cause: unknown, @@ -252,11 +282,9 @@ function summarizeToolOutput(rawOutput: unknown): string | undefined { return summary.length > 400 ? `${summary.slice(0, 397)}...` : summary; } -function mapStopReasonToTurnState(stopReason: string | undefined): - | "completed" - | "failed" - | "interrupted" - | "cancelled" { +function mapStopReasonToTurnState( + stopReason: string | undefined, +): "completed" | "failed" | "interrupted" | "cancelled" { if (stopReason === "cancelled") return "cancelled"; if (stopReason === "interrupted") return "interrupted"; return "completed"; @@ -357,7 +385,10 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { }); }; - const resolvePendingRequest = (context: CursorSessionContext, message: Record) => { + const resolvePendingRequest = ( + context: CursorSessionContext, + message: Record, + ) => { const id = message.id; if (typeof id !== "string" && typeof id !== "number") { return; @@ -380,7 +411,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { }; const decodePermissionRequest = Schema.decodeUnknownSync(CursorAcpPermissionRequest); - const decodeSessionUpdateNotification = Schema.decodeUnknownSync(CursorAcpSessionUpdateNotification); + const decodeSessionUpdateNotification = Schema.decodeUnknownSync( + CursorAcpSessionUpdateNotification, + ); const emitRuntimeWarning = ( context: CursorSessionContext, @@ -393,10 +426,10 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "runtime.warning", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), payload: { message, ...(detail !== undefined ? { detail } : {}), @@ -426,11 +459,11 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "item.completed", eventId: itemStamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: itemStamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: turnState.turnId, - itemId: turnState.assistantItemId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(turnState.turnId), + itemId: asRuntimeItemId(turnState.assistantItemId), payload: { itemType: "assistant_message", status: "completed", @@ -454,10 +487,10 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "turn.completed", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: turnState.turnId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(turnState.turnId), payload: { state, ...(stopReason ? { stopReason } : {}), @@ -534,11 +567,11 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "request.resolved", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - requestId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + requestId: asRuntimeRequestId(requestId), payload: { requestType, decision: selection.decision, @@ -576,11 +609,11 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "request.opened", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - requestId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + requestId: asRuntimeRequestId(requestId), payload: { requestType, ...(detail ? { detail } : {}), @@ -621,9 +654,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { const base = { provider: PROVIDER, - sessionId: context.session.sessionId, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + sessionId: asRuntimeSessionId(context.session.sessionId), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), providerRefs: { providerSessionId: context.session.sessionId, ...(context.session.threadId ? { providerThreadId: context.session.threadId } : {}), @@ -663,8 +696,8 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "content.delta", eventId: stamp.eventId, createdAt: stamp.createdAt, - turnId: context.turnState.turnId, - itemId: context.turnState.assistantItemId, + turnId: asRuntimeTurnId(context.turnState.turnId), + itemId: asRuntimeItemId(context.turnState.assistantItemId), payload: { streamKind: "reasoning_text", delta: update.content.text, @@ -682,8 +715,8 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "content.delta", eventId: stamp.eventId, createdAt: stamp.createdAt, - turnId: context.turnState.turnId, - itemId: context.turnState.assistantItemId, + turnId: asRuntimeTurnId(context.turnState.turnId), + itemId: asRuntimeItemId(context.turnState.assistantItemId), payload: { streamKind: "assistant_text", delta: update.content.text, @@ -707,8 +740,8 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "item.started", eventId: stamp.eventId, createdAt: stamp.createdAt, - turnId: context.turnState.turnId, - itemId: update.toolCallId, + turnId: asRuntimeTurnId(context.turnState.turnId), + itemId: asRuntimeItemId(update.toolCallId), payload: { itemType, status: "inProgress", @@ -726,8 +759,8 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "item.updated", eventId: stamp.eventId, createdAt: stamp.createdAt, - turnId: context.turnState.turnId, - itemId: update.toolCallId, + turnId: asRuntimeTurnId(context.turnState.turnId), + itemId: asRuntimeItemId(update.toolCallId), payload: { itemType, status: "inProgress", @@ -749,8 +782,8 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: eventType, eventId: stamp.eventId, createdAt: stamp.createdAt, - turnId: context.turnState.turnId, - itemId: update.toolCallId, + turnId: asRuntimeTurnId(context.turnState.turnId), + itemId: asRuntimeItemId(update.toolCallId), payload: { itemType: "command_execution", status, @@ -792,7 +825,7 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { const nativeMethod = typeof message.method === "string" ? message.method - : (typeof message.id === "string" || typeof message.id === "number") + : typeof message.id === "string" || typeof message.id === "number" ? "cursor/acp/response" : "cursor/acp/message"; const nativeKind = @@ -808,11 +841,11 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { id: EventId.makeUnsafe(randomUUID()), kind: nativeKind, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: new Date().toISOString(), method: nativeMethod, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), payload: message, }, }); @@ -905,9 +938,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "session.exited", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), payload: { reason: "Session stopped", exitKind: "graceful", @@ -956,11 +989,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { } const startedAt = yield* nowIso; - const sessionId = ProviderSessionId.makeUnsafe(`cursor-session-${yield* Random.nextUUIDv4}`); + const sessionId = ProviderSessionId.makeUnsafe(yield* Random.nextUUIDv4); const cwd = input.cwd ?? process.cwd(); - const cursorOptions = input.providerOptions?.cursor as - | { binaryPath?: string } - | undefined; + const cursorOptions = input.providerOptions?.cursor as { binaryPath?: string } | undefined; const binaryPath = cursorOptions?.binaryPath ?? "agent"; const resumeState = readCursorResumeState(input.resumeCursor); @@ -1031,10 +1062,10 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "runtime.error", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), payload: { message: error.message || "Cursor ACP process error.", class: "transport_error", @@ -1064,9 +1095,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "session.exited", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), payload: { reason: `Cursor ACP exited (code=${code ?? "null"}, signal=${signal ?? "null"}).`, exitKind: code === 0 ? "graceful" : "error", @@ -1088,14 +1119,23 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { }), catch: (cause) => toRequestError(sessionId, "initialize", cause), }); - const decodedInitialize = Schema.decodeUnknownSync(CursorAcpInitializeResult)(initializeResult); + const decodedInitialize = yield* Effect.try({ + try: () => Schema.decodeUnknownSync(CursorAcpInitializeResult)(initializeResult), + catch: (cause) => + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: "Cursor initialize response did not match expected schema.", + cause, + }), + }); const initStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "session.configured", eventId: initStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: initStamp.createdAt, payload: { config: decodedInitialize, @@ -1116,7 +1156,7 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "auth.status", eventId: authStartStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: authStartStamp.createdAt, payload: { isAuthenticating: true, @@ -1133,19 +1173,24 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { const authenticateResult = yield* Effect.tryPromise({ try: async () => sendRequest(context, "authenticate", authenticateRequest), - catch: (cause) => cause, - }); + catch: (cause) => toRequestError(sessionId, "authenticate", cause), + }).pipe( + Effect.match({ + onFailure: (error) => ({ ok: false as const, error }), + onSuccess: (value) => ({ ok: true as const, value }), + }), + ); const authEndStamp = yield* makeEventStamp(); - if (authenticateResult instanceof Error) { + if (!authenticateResult.ok) { yield* offerRuntimeEvent({ type: "auth.status", eventId: authEndStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: authEndStamp.createdAt, payload: { isAuthenticating: false, - error: toMessage(authenticateResult, "Cursor authentication failed."), + error: toMessage(authenticateResult.error, "Cursor authentication failed."), }, providerRefs: { providerSessionId: sessionId, @@ -1154,7 +1199,7 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { source: "cursor.acp.response", method: "authenticate", payload: { - error: toMessage(authenticateResult, "Cursor authentication failed."), + error: toMessage(authenticateResult.error, "Cursor authentication failed."), }, }, }); @@ -1163,7 +1208,7 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "auth.status", eventId: authEndStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: authEndStamp.createdAt, payload: { isAuthenticating: false, @@ -1174,7 +1219,7 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { raw: { source: "cursor.acp.response", method: "authenticate", - payload: authenticateResult, + payload: authenticateResult.value, }, }); } @@ -1225,9 +1270,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "session.started", eventId: sessionStartedStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: sessionStartedStamp.createdAt, - threadId, + threadId: asRuntimeThreadId(threadId), payload: resumeState?.acpSessionId ? { resume: input.resumeCursor } : {}, providerRefs: { providerSessionId: sessionId, @@ -1240,9 +1285,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "thread.started", eventId: threadStartedStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: threadStartedStamp.createdAt, - threadId, + threadId: asRuntimeThreadId(threadId), payload: { providerThreadId: threadId, }, @@ -1257,9 +1302,9 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "session.state.changed", eventId: readyStamp.eventId, provider: PROVIDER, - sessionId, + sessionId: asRuntimeSessionId(sessionId), createdAt: readyStamp.createdAt, - threadId, + threadId: asRuntimeThreadId(threadId), payload: { state: "ready", }, @@ -1286,10 +1331,10 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { }); } - const turnId = ProviderTurnId.makeUnsafe(`cursor-turn-${yield* Random.nextUUIDv4}`); + const turnId = ProviderTurnId.makeUnsafe(yield* Random.nextUUIDv4); const turnState: CursorTurnState = { turnId, - assistantItemId: `cursor-assistant-${yield* Random.nextUUIDv4}`, + assistantItemId: asProviderItemId(yield* Random.nextUUIDv4), startedToolCalls: new Set(), items: [], }; @@ -1307,10 +1352,10 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "turn.started", eventId: startedStamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: startedStamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + turnId: asRuntimeTurnId(turnId), payload: { ...(input.model ? { model: input.model } : {}), ...(input.effort ? { effort: input.effort } : {}), @@ -1340,7 +1385,16 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { catch: (cause) => toRequestError(input.sessionId, "session/prompt", cause), }); - const promptResult = Schema.decodeUnknownSync(CursorAcpSessionPromptResult)(promptResultRaw); + const promptResult = yield* Effect.try({ + try: () => Schema.decodeUnknownSync(CursorAcpSessionPromptResult)(promptResultRaw), + catch: (cause) => + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Cursor session/prompt response did not match expected schema.", + cause, + }), + }); const turnStateValue = mapStopReasonToTurnState(promptResult.stopReason); yield* completeTurn( context, @@ -1376,15 +1430,21 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { } const cancelResult = yield* Effect.tryPromise({ - try: async () => sendRequest(context, "session/cancel", { sessionId: context.acpSessionId }, 15_000), - catch: (cause) => cause, - }); + try: async () => + sendRequest(context, "session/cancel", { sessionId: context.acpSessionId }, 15_000), + catch: (cause) => toRequestError(sessionId, "session/cancel", cause), + }).pipe( + Effect.match({ + onFailure: (error) => ({ ok: false as const, error }), + onSuccess: () => ({ ok: true as const }), + }), + ); - if (cancelResult instanceof Error) { + if (!cancelResult.ok) { yield* emitRuntimeWarning( context, "Cursor ACP session/cancel is unavailable; marking turn as interrupted.", - cancelResult, + cancelResult.error, ); } @@ -1456,11 +1516,11 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { type: "request.resolved", eventId: stamp.eventId, provider: PROVIDER, - sessionId: context.session.sessionId, + sessionId: asRuntimeSessionId(context.session.sessionId), createdAt: stamp.createdAt, - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - ...(context.turnState ? { turnId: context.turnState.turnId } : {}), - requestId, + ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + requestId: asRuntimeRequestId(requestId), payload: { requestType: pending.requestType, decision, diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 9445f9b455e4cb6009c86ebc76e856835b34cd60..bb3d4a65bd5e33432198d4dcedbf4084c00f670f 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -52,6 +52,20 @@ const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.make const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); +type LegacyProviderRuntimeEvent = { + readonly type: string; + readonly eventId: EventId; + readonly provider: "codex" | "claudeCode" | "cursor"; + readonly sessionId: string; + readonly createdAt: string; + readonly threadId?: string | undefined; + readonly turnId?: string | undefined; + readonly itemId?: string | undefined; + readonly requestId?: string | undefined; + readonly payload?: unknown | undefined; + readonly [key: string]: unknown; +}; + function makeFakeCodexAdapter(provider: ProviderKind = "codex") { const sessions = new Map(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); @@ -177,8 +191,8 @@ function makeFakeCodexAdapter(provider: ProviderKind = "codex") { streamEvents: Stream.fromPubSub(runtimeEventPubSub), }; - const emit = (event: ProviderRuntimeEvent): void => { - Effect.runSync(PubSub.publish(runtimeEventPubSub, event)); + const emit = (event: LegacyProviderRuntimeEvent): void => { + Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent)); }; return { @@ -383,16 +397,21 @@ it.effect( }); }).pipe(Effect.provide(secondProviderLayer)); - assert.deepEqual(secondCodex.startSession.mock.calls, [ - [ - { - provider: "codex", - cwd: "/tmp/project", - resumeThreadId: startedSession.threadId, - resumeCursor: startedSession.resumeCursor, - }, - ], - ]); + assert.equal(secondCodex.startSession.mock.calls.length, 1); + const resumedStartInput = secondCodex.startSession.mock.calls[0]?.[0]; + assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true); + if (resumedStartInput && typeof resumedStartInput === "object") { + const startPayload = resumedStartInput as { + provider?: string; + cwd?: string; + resumeCursor?: unknown; + resumeThreadId?: string; + }; + assert.equal(startPayload.provider, "codex"); + assert.equal(startPayload.cwd, "/tmp/project"); + assert.deepEqual(startPayload.resumeCursor, startedSession.resumeCursor); + assert.equal(startPayload.resumeThreadId, "thread-1"); + } assert.equal(secondCodex.rollbackThread.mock.calls.length, 1); const rollbackCall = secondCodex.rollbackThread.mock.calls[0]; assert.equal(typeof rollbackCall?.[0], "string"); @@ -493,16 +512,21 @@ routing.layer("ProviderServiceLive routing", (it) => { numTurns: 1, }); - assert.deepEqual(routing.codex.startSession.mock.calls, [ - [ - { - provider: "codex", - cwd: "/tmp/project", - resumeThreadId: initial.threadId, - resumeCursor: initial.resumeCursor, - }, - ], - ]); + assert.equal(routing.codex.startSession.mock.calls.length, 1); + const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0]; + assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true); + if (resumedStartInput && typeof resumedStartInput === "object") { + const startPayload = resumedStartInput as { + provider?: string; + cwd?: string; + resumeCursor?: unknown; + resumeThreadId?: string; + }; + assert.equal(startPayload.provider, "codex"); + assert.equal(startPayload.cwd, "/tmp/project"); + assert.deepEqual(startPayload.resumeCursor, initial.resumeCursor); + assert.equal(startPayload.resumeThreadId, "thread-2"); + } assert.equal(routing.codex.rollbackThread.mock.calls.length, 1); const rollbackCall = routing.codex.rollbackThread.mock.calls[0]; assert.equal(rollbackCall?.[1], 1); @@ -528,16 +552,21 @@ routing.layer("ProviderServiceLive routing", (it) => { attachments: [], }); - assert.deepEqual(routing.codex.startSession.mock.calls, [ - [ - { - provider: "codex", - cwd: "/tmp/project-send-turn", - resumeThreadId: initial.threadId, - resumeCursor: initial.resumeCursor, - }, - ], - ]); + assert.equal(routing.codex.startSession.mock.calls.length, 1); + const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0]; + assert.equal(typeof resumedStartInput === "object" && resumedStartInput !== null, true); + if (resumedStartInput && typeof resumedStartInput === "object") { + const startPayload = resumedStartInput as { + provider?: string; + cwd?: string; + resumeCursor?: unknown; + resumeThreadId?: string; + }; + assert.equal(startPayload.provider, "codex"); + assert.equal(startPayload.cwd, "/tmp/project-send-turn"); + assert.deepEqual(startPayload.resumeCursor, initial.resumeCursor); + assert.equal(startPayload.resumeThreadId, "thread-4"); + } assert.equal(routing.codex.sendTurn.mock.calls.length, 1); }), ); @@ -626,7 +655,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { ).pipe(Effect.forkChild); yield* sleep(20); - const completedEvent: ProviderRuntimeEvent = { + const completedEvent: LegacyProviderRuntimeEvent = { type: "turn.completed", eventId: asEventId("evt-1"), provider: "codex", @@ -650,7 +679,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { }), ); - it.effect("assigns monotonic per-session sequence numbers to canonical runtime events", () => + it.effect("fans out canonical runtime events in emission order", () => Effect.gen(function* () { const provider = yield* ProviderService; const session = yield* provider.startSession(asThreadId("thread-seq"), { @@ -700,8 +729,8 @@ fanout.layer("ProviderServiceLive fanout", (it) => { yield* Fiber.join(consumer); const received = yield* Ref.get(receivedRef); assert.deepEqual( - received.map((event) => event.sessionSequence), - [1, 2, 3], + received.map((event) => event.eventId), + [asEventId("evt-seq-1"), asEventId("evt-seq-2"), asEventId("evt-seq-3")], ); }), ); @@ -729,7 +758,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { ); yield* sleep(20); - const events: ReadonlyArray = [ + const events: ReadonlyArray = [ { type: "tool.completed", eventId: asEventId("evt-ordered-1"), diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index cf67b4a6ad40b8fcb01a893c3cefe67f5b88c6d9..c0e5bf921049054bc49ac3f4b00341c2cce8fe68 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -12,6 +12,7 @@ import { NonNegativeInt, ProviderSessionId, + RuntimeSessionId, ThreadId, ProviderInterruptTurnInput, ProviderRespondToRequestInput, @@ -123,7 +124,8 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const routedSessionAliasesRef = yield* Ref.make>( new Map(), ); - const sessionRuntimeSequenceRef = yield* Ref.make>(new Map()); + const asRuntimeSessionId = (sessionId: ProviderSessionId): RuntimeSessionId => + RuntimeSessionId.makeUnsafe(sessionId); const canonicalizeRuntimeEventSession = ( event: ProviderRuntimeEvent, @@ -131,10 +133,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => Ref.get(routedSessionAliasesRef).pipe( Effect.map((aliases) => { for (const [staleSessionId, liveSessionId] of aliases) { - if (liveSessionId === event.sessionId) { + if (asRuntimeSessionId(liveSessionId) === event.sessionId) { return { ...event, - sessionId: staleSessionId, + sessionId: asRuntimeSessionId(staleSessionId), } satisfies ProviderRuntimeEvent; } } @@ -144,21 +146,6 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => canonicalizeRuntimeEventSession(event).pipe( - Effect.flatMap((canonicalEvent) => - Ref.modify(sessionRuntimeSequenceRef, (current) => { - const next = new Map(current); - const currentSequence = next.get(canonicalEvent.sessionId) ?? 0; - const sessionSequence = currentSequence + 1; - next.set(canonicalEvent.sessionId, sessionSequence); - return [ - { - ...canonicalEvent, - sessionSequence, - } satisfies ProviderRuntimeEvent, - next, - ] as const; - }), - ), Effect.tap((canonicalEvent) => canonicalEventLogger ? Effect.flatMap( @@ -223,16 +210,6 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => return changed ? next : current; }); - const clearSessionSequence = (sessionId: ProviderSessionId) => - Ref.update(sessionRuntimeSequenceRef, (current) => { - if (!current.has(sessionId)) { - return current; - } - const next = new Map(current); - next.delete(sessionId); - return next; - }); - const setAlias = (staleSessionId: ProviderSessionId, liveSessionId: ProviderSessionId) => Ref.update(routedSessionAliasesRef, (current) => { const existing = current.get(staleSessionId); @@ -554,11 +531,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => if (routed.sessionId !== input.sessionId) { yield* directory.remove(routed.sessionId); yield* clearAliasesReferencing(routed.sessionId); - yield* clearSessionSequence(routed.sessionId); } yield* directory.remove(input.sessionId); yield* clearAliasesReferencing(input.sessionId); - yield* clearSessionSequence(input.sessionId); }); const listSessions: ProviderServiceShape["listSessions"] = () => @@ -610,7 +585,6 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => // Keep persisted session bindings so stale sessions can be resumed after // process restart via providerThreadId. yield* Ref.set(routedSessionAliasesRef, new Map()); - yield* Ref.set(sessionRuntimeSequenceRef, new Map()); }); return { diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index ce76f5b75fd0d50bfb5a9a2e1782d3a97e33a9bc..b0a4bc20d4a3ddb8b3731ca33221997257285928 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -1234,16 +1234,18 @@ describe("WebSocket Server", () => { return event.type === "thread.session-set"; }); - emitRuntimeEvent({ - type: "message.delta", - eventId: asEventId("evt-ws-runtime-message-delta"), - provider: "codex", - sessionId, - createdAt: new Date().toISOString(), - turnId: asProviderTurnId("turn-1"), - itemId: asProviderItemId("item-1"), - delta: "hello from runtime", - }); + emitRuntimeEvent( + { + type: "message.delta", + eventId: asEventId("evt-ws-runtime-message-delta"), + provider: "codex", + sessionId, + createdAt: new Date().toISOString(), + turnId: asProviderTurnId("turn-1"), + itemId: asProviderItemId("item-1"), + delta: "hello from runtime", + } as unknown as ProviderRuntimeEvent, + ); const domainPush = await waitForPush(ws, ORCHESTRATION_WS_CHANNELS.domainEvent, (push) => { const event = push.data as { type?: string; payload?: { messageId?: string; text?: string } }; diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index 4177a5d45cfe6591e499d4c371a91bdc3e480dd3..f78faf92f2e3bc7d424795f3bfe1165c1eec1963 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -100,6 +100,35 @@ describe("derivePendingApprovals", () => { }, ]); }); + + it("clears stale pending approvals when provider reports unknown pending request", () => { + const activities: OrchestrationThreadActivity[] = [ + makeActivity({ + id: "approval-open-stale", + createdAt: "2026-02-23T00:00:01.000Z", + kind: "approval.requested", + summary: "Command approval requested", + tone: "approval", + payload: { + requestId: "req-stale-1", + requestKind: "command", + }, + }), + makeActivity({ + id: "approval-failed-stale", + createdAt: "2026-02-23T00:00:02.000Z", + kind: "provider.approval.respond.failed", + summary: "Provider approval response failed", + tone: "error", + payload: { + requestId: "req-stale-1", + detail: "Unknown pending permission request: req-stale-1", + }, + }), + ]; + + expect(derivePendingApprovals(activities)).toEqual([]); + }); }); describe("deriveWorkLogEntries", () => { diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index e4bb7365116560410339bbb7415e7fad7f3ac6a4..1ffbe6c637eca83ff94684a430c1cf38ce98fe32 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -138,6 +138,7 @@ export function derivePendingApprovals( if (activity.kind === "approval.resolved" && requestId) { openByRequestId.delete(requestId); + continue; } } diff --git a/packages/contracts/src/model.test.ts b/packages/contracts/src/model.test.ts index 5b631bab295c51325e4a499c9b03d66184d684b9..508d1b53dfc3e44734683fe3ad9025e58124ff34 100644 --- a/packages/contracts/src/model.test.ts +++ b/packages/contracts/src/model.test.ts @@ -117,7 +117,7 @@ describe("cursor model selection", () => { thinking: false, }); expect(parseCursorModelSelection("gpt-5.2-codex")).toEqual({ - family: "gpt-5.2-codex", + family: "gpt-5.3-codex", reasoning: "normal", fast: false, thinking: false, @@ -142,11 +142,11 @@ describe("cursor model selection", () => { it("resolves codex family selections into concrete model ids", () => { expect( resolveCursorModelFromSelection({ - family: "gpt-5.2-codex", + family: "gpt-5.3-codex", reasoning: "xhigh", fast: true, }), - ).toBe("gpt-5.2-codex-xhigh-fast"); + ).toBe("gpt-5.3-codex-xhigh-fast"); }); });