From 75354bbd962d9fc38a1ba40ee78520822b349016 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 28 Feb 2026 12:27:45 -0800 Subject: [PATCH] Track runtime event sequence and order thread activities by sequence - add per-session `sessionSequence` on provider runtime events and persist activity `sequence` - migrate `projection_thread_activities` with nullable `sequence` column + index - sort server/web activity projections by sequence fallback to timestamp/id - allow provider sessions/turns to start before a real threadId is emitted --- .../Layers/ProjectionPipeline.ts | 3 + .../Layers/ProjectionSnapshotQuery.ts | 11 ++- .../Layers/ProviderRuntimeIngestion.ts | 6 ++ apps/server/src/orchestration/projector.ts | 19 +++- .../Layers/ProjectionThreadActivities.ts | 30 +++++- apps/server/src/persistence/Migrations.ts | 2 + .../008_ProjectionThreadActivitySequence.ts | 16 +++ .../Services/ProjectionThreadActivities.ts | 5 +- .../provider/Layers/ClaudeCodeAdapter.test.ts | 67 +++++++++++++ .../src/provider/Layers/ClaudeCodeAdapter.ts | 45 +++++---- .../provider/Layers/ProviderService.test.ts | 97 +++++++++++++++---- .../src/provider/Layers/ProviderService.ts | 52 +++++++--- apps/web/src/session-logic.test.ts | 24 +++++ apps/web/src/session-logic.ts | 25 +++-- packages/contracts/src/orchestration.ts | 1 + packages/contracts/src/provider.ts | 2 +- packages/contracts/src/providerRuntime.ts | 13 +++ 17 files changed, 349 insertions(+), 69 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/008_ProjectionThreadActivitySequence.ts diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index f4c8728f39fd53aac54b623c45631b1e161d72b2..6aba1b65b3b6bc8d04da93dd431599b4154dd168 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -598,6 +598,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { kind: event.payload.activity.kind, summary: event.payload.activity.summary, payload: event.payload.activity.payload, + ...(event.payload.activity.sequence !== undefined + ? { sequence: event.payload.activity.sequence } + : {}), createdAt: event.payload.activity.createdAt, }); return; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index a4f260118c43735096349d0726a51c7bcf9873fd..0c930a9b5a2075279713772358ef303f9900862e 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -2,6 +2,7 @@ import { ChatAttachment, IsoDateTime, MessageId, + NonNegativeInt, OrchestrationCheckpointFile, OrchestrationReadModel, ProjectScript, @@ -53,6 +54,7 @@ const ProjectionThreadDbRowSchema = ProjectionThread; const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( Struct.assign({ payload: Schema.fromJsonString(Schema.Unknown), + sequence: Schema.NullOr(NonNegativeInt), }), ); const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession; @@ -195,9 +197,15 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { kind, summary, payload_json AS "payload", + sequence, created_at AS "createdAt" FROM projection_thread_activities - ORDER BY thread_id ASC, created_at ASC, activity_id ASC + ORDER BY + thread_id ASC, + CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC, + sequence ASC, + created_at ASC, + activity_id ASC `, }); @@ -398,6 +406,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { summary: row.summary, payload: row.payload, turnId: row.turnId, + ...(row.sequence !== null ? { sequence: row.sequence } : {}), createdAt: row.createdAt, }); activitiesByThread.set(row.threadId, threadActivities); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 4451a6b83b526a4ac29eab7cd8bcc26779d03885..10b36ce76014f823956b2ec8db7a1597222c8791 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -79,6 +79,7 @@ function truncateDetail(value: string, limit = 180): string { function runtimeEventToActivities( event: ProviderRuntimeEvent, ): ReadonlyArray { + const maybeSequence = event.sessionSequence !== undefined ? { sequence: event.sessionSequence } : {}; switch (event.type) { case "approval.requested": { return [ @@ -97,6 +98,7 @@ function runtimeEventToActivities( ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), }, turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, }, ]; } @@ -115,6 +117,7 @@ function runtimeEventToActivities( ...(event.decision ? { decision: event.decision } : {}), }, turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, }, ]; } @@ -131,6 +134,7 @@ function runtimeEventToActivities( message: truncateDetail(event.message), }, turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, }, ]; } @@ -148,6 +152,7 @@ function runtimeEventToActivities( ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), }, turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, }, ]; } @@ -165,6 +170,7 @@ function runtimeEventToActivities( ...(event.detail ? { detail: truncateDetail(event.detail) } : {}), }, turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, }, ]; } diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index f92f7172a4cb88d2073cb733b382f44d3f6d5480..ac35b84118dfd368e6a1aaf38aa07551c7b57cf8 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -124,6 +124,23 @@ function retainThreadActivitiesAfterRevert( ); } +function compareThreadActivities( + left: OrchestrationThread["activities"][number], + right: OrchestrationThread["activities"][number], +): number { + if (left.sequence !== undefined && right.sequence !== undefined) { + if (left.sequence !== right.sequence) { + return left.sequence - right.sequence; + } + } else if (left.sequence !== undefined) { + return 1; + } else if (right.sequence !== undefined) { + return -1; + } + + return left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id); +} + export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { return { snapshotSequence: 0, @@ -503,7 +520,7 @@ export function projectEvent( ...thread.activities.filter((entry) => entry.id !== payload.activity.id), payload.activity, ] - .toSorted((left, right) => left.createdAt.localeCompare(right.createdAt)) + .toSorted(compareThreadActivities) .slice(-500); return { diff --git a/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts b/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts index c2285f95de7679058f72418e762165f2495936df..8e88cfa7853e94929f94ee70cb46f8d8d49d3c2f 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts @@ -1,5 +1,6 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import { NonNegativeInt } from "@t3tools/contracts"; import { Effect, Layer, Schema, Struct } from "effect"; import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts"; @@ -15,6 +16,7 @@ import { const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( Struct.assign({ payload: Schema.fromJsonString(Schema.Unknown), + sequence: Schema.NullOr(NonNegativeInt), }), ); @@ -29,7 +31,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const upsertProjectionThreadActivityRow = SqlSchema.void({ - Request: ProjectionThreadActivityDbRowSchema, + Request: ProjectionThreadActivity, execute: (row) => sql` INSERT INTO projection_thread_activities ( @@ -40,6 +42,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { kind, summary, payload_json, + sequence, created_at ) VALUES ( @@ -49,7 +52,8 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { ${row.tone}, ${row.kind}, ${row.summary}, - ${row.payload}, + ${JSON.stringify(row.payload)}, + ${row.sequence ?? null}, ${row.createdAt} ) ON CONFLICT (activity_id) @@ -60,6 +64,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { kind = excluded.kind, summary = excluded.summary, payload_json = excluded.payload_json, + sequence = excluded.sequence, created_at = excluded.created_at `, }); @@ -77,10 +82,15 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { kind, summary, payload_json AS "payload", + sequence, created_at AS "createdAt" FROM projection_thread_activities WHERE thread_id = ${threadId} - ORDER BY created_at ASC, activity_id ASC + ORDER BY + CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC, + sequence ASC, + created_at ASC, + activity_id ASC `, }); @@ -111,8 +121,18 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { "ProjectionThreadActivityRepository.listByThreadId:decodeRows", ), ), - Effect.map( - (rows) => rows as ReadonlyArray>, + Effect.map((rows) => + rows.map((row) => ({ + activityId: row.activityId, + threadId: row.threadId, + turnId: row.turnId, + tone: row.tone, + kind: row.kind, + summary: row.summary, + payload: row.payload, + ...(row.sequence !== null ? { sequence: row.sequence } : {}), + createdAt: row.createdAt, + })), ), ); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 3df831d0c76970909035382aefd422ce9cd576f4..0a0a8226df79031ec3940cb670f6e2460cf7666c 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -19,6 +19,7 @@ import Migration0004 from "./Migrations/004_ProviderSessionRuntime.ts"; import Migration0005 from "./Migrations/005_Projections.ts"; import Migration0006 from "./Migrations/006_ProjectionThreadSessionRuntimeModeColumns.ts"; import Migration0007 from "./Migrations/007_ProjectionThreadMessageAttachments.ts"; +import Migration0008 from "./Migrations/008_ProjectionThreadActivitySequence.ts"; /** * Migration loader with all migrations defined inline. @@ -38,6 +39,7 @@ const loader = Migrator.fromRecord({ "5_Projections": Migration0005, "6_ProjectionThreadSessionRuntimeModeColumns": Migration0006, "7_ProjectionThreadMessageAttachments": Migration0007, + "8_ProjectionThreadActivitySequence": Migration0008, }); /** diff --git a/apps/server/src/persistence/Migrations/008_ProjectionThreadActivitySequence.ts b/apps/server/src/persistence/Migrations/008_ProjectionThreadActivitySequence.ts new file mode 100644 index 0000000000000000000000000000000000000000..6000f0c62cb4b829ccaa7e2f40a43533aca6d7b4 --- /dev/null +++ b/apps/server/src/persistence/Migrations/008_ProjectionThreadActivitySequence.ts @@ -0,0 +1,16 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + ALTER TABLE projection_thread_activities + ADD COLUMN sequence INTEGER + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_thread_activities_thread_sequence + ON projection_thread_activities(thread_id, sequence) + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreadActivities.ts b/apps/server/src/persistence/Services/ProjectionThreadActivities.ts index e9e39b49d0704d3cb9b2853718b76ded29cc01b0..586ae3eb4a8499e81fd0458003f3392791334836 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadActivities.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadActivities.ts @@ -9,6 +9,7 @@ import { EventId, IsoDateTime, + NonNegativeInt, OrchestrationThreadActivityTone, ThreadId, TurnId, @@ -26,6 +27,7 @@ export const ProjectionThreadActivity = Schema.Struct({ kind: Schema.String, summary: Schema.String, payload: Schema.Unknown, + sequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, }); export type ProjectionThreadActivity = typeof ProjectionThreadActivity.Type; @@ -57,7 +59,8 @@ export interface ProjectionThreadActivityRepositoryShape { /** * List projected thread activity rows for a thread. * - * Returned in ascending creation order. + * Returned in ascending runtime sequence order (or creation order when + * sequence is unavailable). */ readonly listByThreadId: ( input: ListProjectionThreadActivitiesInput, diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts index b92cdec8f24b27509b4b86f7fa2d270ee9b136d7..5394dee2f86b1f861122513164516c3bd50d8690 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts @@ -505,6 +505,73 @@ describe("ClaudeCodeAdapterLive", () => { ); }); + it.effect("does not fabricate provider thread ids before first SDK session_id", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeCodeAdapter; + + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 5).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + provider: "claudeCode", + }); + assert.equal(session.threadId, undefined); + + const turn = yield* adapter.sendTurn({ + sessionId: session.sessionId, + input: "hello", + attachments: [], + }); + assert.equal(turn.threadId, undefined); + + harness.query.emit({ + type: "stream_event", + session_id: "sdk-thread-real", + uuid: "stream-thread-real", + parent_tool_use_id: null, + event: { + type: "message_start", + message: { + id: "msg-thread-real", + }, + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-thread-real", + uuid: "result-thread-real", + } as unknown as SDKMessage); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + assert.deepEqual( + runtimeEvents.map((event) => event.type), + ["session.started", "turn.started", "thread.started", "message.completed", "turn.completed"], + ); + + const sessionStarted = runtimeEvents[0]; + assert.equal(sessionStarted?.type, "session.started"); + if (sessionStarted?.type === "session.started") { + assert.equal("threadId" in sessionStarted, false); + } + + const threadStarted = runtimeEvents[2]; + assert.equal(threadStarted?.type, "thread.started"); + if (threadStarted?.type === "thread.started") { + assert.equal(threadStarted.threadId, "sdk-thread-real"); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("bridges approval request/response lifecycle through canUseTool", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts index 989cb73b1be21607dfd1a9b6938b78fa62d9736c..8289a3e40ea9a05da86267b7552a52b488ed60f1 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts @@ -126,6 +126,10 @@ function isUuid(value: string): boolean { ); } +function isSyntheticClaudeThreadId(value: string): boolean { + return value.startsWith("claude-thread-"); +} + function toMessage(cause: unknown, fallback: string): string { if (cause instanceof Error && cause.message.length > 0) { return cause.message; @@ -158,7 +162,11 @@ function readClaudeResumeState(resumeCursor: unknown): ClaudeResumeState | undef turnCount?: unknown; }; - const threadId = typeof cursor.threadId === "string" ? cursor.threadId : undefined; + const threadIdCandidate = typeof cursor.threadId === "string" ? cursor.threadId : undefined; + const threadId = + threadIdCandidate && !isSyntheticClaudeThreadId(threadIdCandidate) + ? threadIdCandidate + : undefined; const resumeCandidate = typeof cursor.resume === "string" ? cursor.resume @@ -464,9 +472,14 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { }>; }> => Effect.gen(function* () { - const threadId = - context.session.threadId ?? - ProviderThreadId.makeUnsafe(`claude-thread-${yield* Random.nextUUIDv4}`); + const threadId = context.session.threadId; + if (!threadId) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "readThread", + issue: "Session thread id is not initialized yet.", + }); + } return { threadId, turns: context.turns.map((turn) => ({ @@ -500,6 +513,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { + if (typeof message.session_id !== "string" || message.session_id.length === 0) { + return; + } const nextThreadId = ProviderThreadId.makeUnsafe(message.session_id); context.resumeSessionId = message.session_id; const changed = context.session.threadId !== nextThreadId; @@ -922,9 +938,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { `claude-session-${yield* Random.nextUUIDv4}`, ); const resumeState = readClaudeResumeState(input.resumeCursor); - const threadId = - resumeState?.threadId ?? - ProviderThreadId.makeUnsafe(`claude-thread-${yield* Random.nextUUIDv4}`); + const threadId = resumeState?.threadId; const promptQueue = yield* Queue.unbounded(); const prompt = Stream.fromQueue(promptQueue).pipe( @@ -1082,9 +1096,9 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { status: "ready", ...(input.cwd ? { cwd: input.cwd } : {}), ...(input.model ? { model: input.model } : {}), - threadId, + ...(threadId ? { threadId } : {}), resumeCursor: { - threadId, + ...(threadId ? { threadId } : {}), ...(resumeState?.resume ? { resume: resumeState.resume } : {}), ...(resumeState?.resumeSessionAt ? { resumeSessionAt: resumeState.resumeSessionAt } @@ -1119,7 +1133,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { provider: PROVIDER, sessionId, createdAt: sessionStartedStamp.createdAt, - threadId, + ...(threadId ? { threadId } : {}), }); Effect.runFork(runSdkStream(context)); @@ -1186,17 +1200,8 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { message, }).pipe(Effect.mapError((cause) => toRequestError(input.sessionId, "turn/start", cause))); - const threadId = context.session.threadId; - if (!threadId) { - return yield* new ProviderAdapterValidationError({ - provider: PROVIDER, - operation: "sendTurn", - issue: "Session thread id is not initialized.", - }); - } - return { - threadId, + ...(context.session.threadId ? { threadId: context.session.threadId } : {}), turnId, ...(context.session.resumeCursor !== undefined ? { resumeCursor: context.session.resumeCursor } diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 1ce721eaa71af28cbded4d11f342f5f20b6802c5..ffe8acf5f7cf7e2e5ab6ef00d85b8469efcb016f 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -28,7 +28,6 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { ProviderAdapterSessionNotFoundError, ProviderSessionNotFoundError, - ProviderValidationError, ProviderUnsupportedError, type ProviderAdapterError, } from "../Errors.ts"; @@ -648,6 +647,62 @@ fanout.layer("ProviderServiceLive fanout", (it) => { }), ); + it.effect("assigns monotonic per-session sequence numbers to canonical runtime events", () => + Effect.gen(function* () { + const provider = yield* ProviderService; + const session = yield* provider.startSession(asThreadId("thread-seq"), { + provider: "codex", + }); + + const receivedRef = yield* Ref.make>([]); + const consumer = yield* Stream.take(provider.streamEvents, 3).pipe( + Stream.runForEach((event) => Ref.update(receivedRef, (current) => [...current, event])), + Effect.forkChild, + ); + yield* sleep(20); + + fanout.codex.emit({ + type: "tool.started", + eventId: asEventId("evt-seq-1"), + provider: "codex", + sessionId: session.sessionId, + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("thread-1"), + turnId: asTurnId("turn-1"), + toolKind: "command", + title: "Command run", + }); + fanout.codex.emit({ + type: "tool.completed", + eventId: asEventId("evt-seq-2"), + provider: "codex", + sessionId: session.sessionId, + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("thread-1"), + turnId: asTurnId("turn-1"), + toolKind: "command", + title: "Command run", + }); + fanout.codex.emit({ + type: "turn.completed", + eventId: asEventId("evt-seq-3"), + provider: "codex", + sessionId: session.sessionId, + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("thread-1"), + turnId: asTurnId("turn-1"), + status: "completed", + }); + + yield* Fiber.join(consumer); + const received = yield* Ref.get(receivedRef); + assert.deepEqual( + received.map((event) => event.sessionSequence), + [1, 2, 3], + ); + }), + ); + it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () => Effect.gen(function* () { const provider = yield* ProviderService; @@ -746,9 +801,10 @@ validation.layer("ProviderServiceLive validation", (it) => { }), ); - it.effect("fails startSession when adapter returns no threadId", () => + it.effect("accepts startSession when adapter has not emitted provider thread id yet", () => Effect.gen(function* () { const provider = yield* ProviderService; + const runtimeRepository = yield* ProviderSessionRuntimeRepository; validation.codex.startSession.mockImplementationOnce((input: ProviderSessionStartInput) => Effect.sync(() => { @@ -757,27 +813,28 @@ validation.layer("ProviderServiceLive validation", (it) => { sessionId: asSessionId("sess-missing-thread"), provider: "codex", status: "ready", - cwd: input.cwd ?? process.cwd(), - createdAt: now, - updatedAt: now, - } satisfies ProviderSession; - }), + cwd: input.cwd ?? process.cwd(), + createdAt: now, + updatedAt: now, + } satisfies ProviderSession; + }), ); - const failure = yield* Effect.result( - provider.startSession(asThreadId("thread-missing"), { - provider: "codex", - cwd: "/tmp/project", - }), - ); + const session = yield* provider.startSession(asThreadId("thread-missing"), { + provider: "codex", + cwd: "/tmp/project", + }); - assertFailure( - failure, - new ProviderValidationError({ - operation: "ProviderService.startSession", - issue: "Provider 'codex' returned a session without threadId.", - }), - ); + assert.equal(session.sessionId, asSessionId("sess-missing-thread")); + assert.equal(session.threadId, undefined); + + const runtime = yield* runtimeRepository.getBySessionId({ + providerSessionId: session.sessionId, + }); + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + assert.equal(runtime.value.providerThreadId, null); + } }), ); }); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 06e97c8ab935b61ecf90f771940d83acb32906c6..465da54858929f225a1143b1dd15e8b15cbaa4e9 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -123,6 +123,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const routedSessionAliasesRef = yield* Ref.make>( new Map(), ); + const sessionRuntimeSequenceRef = yield* Ref.make>(new Map()); const canonicalizeRuntimeEventSession = ( event: ProviderRuntimeEvent, @@ -143,6 +144,21 @@ 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( @@ -165,23 +181,16 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const upsertSessionBinding = ( session: ProviderSession, - operation: string, threadId: ThreadId, ) => Effect.gen(function* () { const providerThreadId = session.threadId; - if (!providerThreadId) { - return yield* toValidationError( - operation, - `Provider '${session.provider}' returned a session without threadId.`, - ); - } yield* directory.upsert({ sessionId: session.sessionId, provider: session.provider, threadId, - providerThreadId, + ...(providerThreadId !== undefined ? { providerThreadId } : {}), status: toRuntimeStatus(session), ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}), runtimePayload: toRuntimePayloadFromSession(session), @@ -214,6 +223,16 @@ 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); @@ -262,14 +281,15 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => if (existing) { const existingProviderThreadId = yield* upsertSessionBinding( existing, - `${input.operation}:upsertExistingSession`, input.binding.threadId, ); yield* directory.upsert({ sessionId: input.staleSessionId, provider: existing.provider, threadId: input.binding.threadId, - providerThreadId: existingProviderThreadId, + ...(existingProviderThreadId !== undefined + ? { providerThreadId: existingProviderThreadId } + : {}), ...(existing.resumeCursor !== undefined ? { resumeCursor: existing.resumeCursor } : {}), }); if (existing.sessionId !== input.staleSessionId) { @@ -307,7 +327,6 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const resumedProviderThreadId = yield* upsertSessionBinding( resumed, - `${input.operation}:upsertRecoveredSession`, input.binding.threadId, ); @@ -315,7 +334,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => sessionId: input.staleSessionId, provider: resumed.provider, threadId: input.binding.threadId, - providerThreadId: resumedProviderThreadId, + ...(resumedProviderThreadId !== undefined + ? { providerThreadId: resumedProviderThreadId } + : {}), ...(resumed.resumeCursor !== undefined ? { resumeCursor: resumed.resumeCursor } : {}), }); @@ -428,7 +449,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ); } - yield* upsertSessionBinding(session, "ProviderService.startSession", threadId); + yield* upsertSessionBinding(session, threadId); return session; }); @@ -473,7 +494,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => sessionId: input.sessionId, provider: routed.adapter.provider, threadId, - providerThreadId: turn.threadId, + ...(turn.threadId !== undefined ? { providerThreadId: turn.threadId } : {}), status: "running", ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}), runtimePayload: { @@ -533,9 +554,11 @@ 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"] = () => @@ -584,6 +607,7 @@ 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/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index 17147adb3b42275c814d7835482c1d3a4e1c33ef..99d0a0b23581409b840dadeb3785e09ecfa3007d 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -17,6 +17,7 @@ function makeActivity(overrides: { tone?: OrchestrationThreadActivity["tone"]; payload?: Record; turnId?: string; + sequence?: number; }): OrchestrationThreadActivity { const payload = overrides.payload ?? {}; return { @@ -27,6 +28,7 @@ function makeActivity(overrides: { tone: overrides.tone ?? "tool", payload, turnId: overrides.turnId ? TurnId.makeUnsafe(overrides.turnId) : null, + ...(overrides.sequence !== undefined ? { sequence: overrides.sequence } : {}), }; } @@ -131,6 +133,28 @@ describe("deriveWorkLogEntries", () => { const entries = deriveWorkLogEntries(activities, undefined); expect(entries.map((entry) => entry.id)).toEqual(["tool-complete"]); }); + + it("orders work log by activity sequence when present", () => { + const activities: OrchestrationThreadActivity[] = [ + makeActivity({ + id: "second", + createdAt: "2026-02-23T00:00:03.000Z", + sequence: 2, + summary: "Tool call complete", + kind: "tool.completed", + }), + makeActivity({ + id: "first", + createdAt: "2026-02-23T00:00:04.000Z", + sequence: 1, + summary: "Tool call complete", + kind: "tool.completed", + }), + ]; + + const entries = deriveWorkLogEntries(activities, undefined); + expect(entries.map((entry) => entry.id)).toEqual(["first", "second"]); + }); }); describe("hasToolActivityForTurn", () => { diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 23b1480290f3b1b3baa3c90144b77a24be611f21..14ddeb4e4c9e6a0a141ecabbec3773da63d7792b 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -91,9 +91,7 @@ export function derivePendingApprovals( activities: ReadonlyArray, ): PendingApproval[] { const openByRequestId = new Map(); - const ordered = [...activities].toSorted((left, right) => - left.createdAt.localeCompare(right.createdAt), - ); + const ordered = [...activities].toSorted(compareActivitiesByOrder); for (const activity of ordered) { const payload = @@ -134,9 +132,7 @@ export function deriveWorkLogEntries( activities: ReadonlyArray, latestTurnId: TurnId | undefined, ): WorkLogEntry[] { - const ordered = [...activities].toSorted((left, right) => - left.createdAt.localeCompare(right.createdAt), - ); + const ordered = [...activities].toSorted(compareActivitiesByOrder); return ordered .filter((activity) => (latestTurnId ? activity.turnId === latestTurnId : true)) .filter((activity) => activity.kind !== "tool.started") @@ -159,6 +155,23 @@ export function deriveWorkLogEntries( }); } +function compareActivitiesByOrder( + left: OrchestrationThreadActivity, + right: OrchestrationThreadActivity, +): number { + if (left.sequence !== undefined && right.sequence !== undefined) { + if (left.sequence !== right.sequence) { + return left.sequence - right.sequence; + } + } else if (left.sequence !== undefined) { + return 1; + } else if (right.sequence !== undefined) { + return -1; + } + + return left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id); +} + export function hasToolActivityForTurn( activities: ReadonlyArray, turnId: TurnId | null | undefined, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 66180d28aa4427a6406bf0a85c65b632849e8104..cca0dde4c02989685b597feca643c3c33494d1a1 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -212,6 +212,7 @@ export const OrchestrationThreadActivity = Schema.Struct({ summary: TrimmedNonEmptyString, payload: Schema.Unknown, turnId: Schema.NullOr(TurnId), + sequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, }); export type OrchestrationThreadActivity = typeof OrchestrationThreadActivity.Type; diff --git a/packages/contracts/src/provider.ts b/packages/contracts/src/provider.ts index 2ead00d322ea8f09c36ed96db5dd1d160d5c0f5c..3307ad4dc0c5626682fd80000ef6996967ba2243 100644 --- a/packages/contracts/src/provider.ts +++ b/packages/contracts/src/provider.ts @@ -93,7 +93,7 @@ export const ProviderSendTurnInput = Schema.Struct({ export type ProviderSendTurnInput = typeof ProviderSendTurnInput.Type; export const ProviderTurnStartResult = Schema.Struct({ - threadId: ProviderThreadId, + threadId: Schema.optional(ProviderThreadId), turnId: ProviderTurnId, resumeCursor: Schema.optional(Schema.Unknown), }); diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index 1544fda1cc11ee01e3f5ff89bf4ce4955d2299fd..26910b54b89421b85b06d52090b12fe88d6c8306 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -31,6 +31,7 @@ export const ProviderRuntimeSessionStartedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), message: Schema.optional(TrimmedNonEmptyStringSchema), @@ -42,6 +43,7 @@ export const ProviderRuntimeSessionExitedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), message: Schema.optional(TrimmedNonEmptyStringSchema), @@ -53,6 +55,7 @@ export const ProviderRuntimeThreadStartedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: ProviderThreadId, }); @@ -63,6 +66,7 @@ export const ProviderRuntimeTurnStartedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: ProviderTurnId, @@ -74,6 +78,7 @@ export const ProviderRuntimeTurnCompletedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: Schema.optional(ProviderTurnId), @@ -87,6 +92,7 @@ export const ProviderRuntimeMessageDeltaEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: Schema.optional(ProviderTurnId), @@ -100,6 +106,7 @@ export const ProviderRuntimeMessageCompletedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, itemId: ProviderItemId, threadId: Schema.optional(ProviderThreadId), @@ -112,6 +119,7 @@ export const ProviderRuntimeToolStartedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: Schema.optional(ProviderTurnId), @@ -127,6 +135,7 @@ export const ProviderRuntimeToolCompletedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: Schema.optional(ProviderTurnId), @@ -142,6 +151,7 @@ export const ProviderRuntimeApprovalRequestedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: Schema.optional(ProviderTurnId), @@ -158,6 +168,7 @@ export const ProviderRuntimeApprovalResolvedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: Schema.optional(ProviderTurnId), @@ -173,6 +184,7 @@ export const ProviderRuntimeCheckpointCapturedEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: ProviderThreadId, turnId: Schema.optional(ProviderTurnId), @@ -187,6 +199,7 @@ export const ProviderRuntimeErrorEvent = Schema.Struct({ eventId: EventId, provider: ProviderKind, sessionId: ProviderSessionId, + sessionSequence: Schema.optional(NonNegativeInt), createdAt: IsoDateTime, threadId: Schema.optional(ProviderThreadId), turnId: Schema.optional(ProviderTurnId), -- 2.54.0