diff --git a/apps/server/src/persistence/OrchestrationEventStore.test.ts b/apps/server/src/persistence/OrchestrationEventStore.test.ts index abe4b66392d2c17c26fedc4dfa5e4a1dafce14c5..6a075e19c46eed02018c4e4e3e1e3d4e150409f7 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.test.ts @@ -393,6 +393,101 @@ layer("OrchestrationEventStore", (it) => { ); }), ); + + it.effect.each([ + ["agent", "thread"], + ["application", "thread"], + ["application", "project"], + ] as const)( + "%s replay skips unknown %s types but rejects malformed known payloads", + ([reader, aggregateKind]) => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const streamId = ThreadId.make(`skip-unknown:${reader}:${aggregateKind}`); + const now = "2026-01-05T00:00:00.000Z"; + // Direct SQL models rows written by a newer build, which append cannot accept. + const insertRow = Effect.fnUntraced(function* ( + version: number, + eventType: string, + payload: string, + ) { + const [row] = yield* sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`${streamId}:${version}`}, ${aggregateKind}, ${streamId}, ${version}, ${eventType}, + ${now}, 'server', ${payload}, '{}', 2 + ) RETURNING sequence + `; + return row!.sequence; + }); + const read = ( + afterSequence: number, + throughSequence: number, + skipUnknownEventTypes = false, + limit?: number, + ) => { + const input = { afterSequence, throughSequence, skipUnknownEventTypes }; + return reader === "agent" + ? store + .readAgentEvents({ + ...input, + threadId: streamId, + ...(limit === undefined ? {} : { limit }), + }) + .pipe( + Stream.map((event) => event.sequence), + Stream.runCollect, + ) + : store.readApplicationEvents(input).pipe( + Stream.map((event) => event.sequence), + Stream.runCollect, + ); + }; + const knownType = + aggregateKind === "thread" ? "provider-session.detached" : "project.meta-updated"; + const knownPayload = JSON.stringify( + aggregateKind === "thread" + ? { providerSessionId: "session:known", detachedAt: now } + : { projectId: streamId, updatedAt: now }, + ); + const baseline = yield* store.latestApplicationSequence; + const before = yield* insertRow(0, knownType, knownPayload); + // More than one raw page ensures replay crosses an entirely filtered page. + yield* Effect.forEach( + Array.from({ length: 1_000 }, (_, i) => i + 1), + (version) => insertRow(version, "future.event", "not JSON"), + { discard: true }, + ); + // A type known for the other aggregate kind is unknown for this one. + yield* insertRow( + 1_001, + aggregateKind === "thread" ? "project.meta-updated" : "provider-session.detached", + "{}", + ); + const after = yield* insertRow(1_002, knownType, knownPayload); + const highWater = yield* insertRow(1_003, "future.trailing", "{}"); + yield* insertRow(1_004, knownType, knownPayload); + assert.deepEqual(yield* read(baseline, highWater, true), [before, after]); + assert.isTrue(isPersistenceDecodeError(yield* read(baseline, highWater).pipe(Effect.flip))); + if (reader === "agent") { + // The work limit includes skipped rows, rather than fetching extra pages to fill it. + assert.deepEqual(yield* read(baseline, highWater, true, 2), [before]); + } + const beforeBroken = yield* store.latestApplicationSequence; + // Exercise a grouped V2 literal as well as the project type discriminator. + const broken = yield* insertRow( + 1_005, + aggregateKind === "thread" ? "thread.metadata-updated" : "project.created", + "{}", + ); + assert.isTrue( + isPersistenceDecodeError(yield* read(beforeBroken, broken, true).pipe(Effect.flip)), + ); + }), + ); }); it.effect.each(["high-water", "replay"] as const)( diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index ef8b49c2c53f2fcabda7c4ed92af800cb3dfab41..16cb8208e66ccf37d23228bcbfbc00d9a91a2bb3 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -17,6 +17,7 @@ import { CommandId, EventId, IsoDateTime, + isKnownOrchestrationV2EventType, NonNegativeInt, OrchestrationV2DomainEventJson, OrchestrationV2StoredEvent, @@ -69,7 +70,11 @@ export class OrchestrationEventStore extends Context.Service< * Read only V2 thread events from the application log. * * Reads in fixed-size sequence pages until the filtered range is exhausted; - * `limit` caps the total emitted events across pages. + * `limit` caps raw rows across pages, including skipped unknown types. + * + * A row whose `event_type` this build does not know fails the read unless + * `skipUnknownEventTypes` is set, which logs and drops it instead. Only + * client thread replay opts in; anything that builds server state stays strict. */ readonly readAgentEvents: (input?: { readonly afterSequence?: number; @@ -78,6 +83,7 @@ export class OrchestrationEventStore extends Context.Service< readonly commandId?: CommandId; readonly eventType?: OrchestrationV2DomainEvent["type"]; readonly limit?: number; + readonly skipUnknownEventTypes?: boolean; }) => Stream.Stream; /** Measure one thread's bounded replay without loading or decoding its payloads. */ @@ -118,10 +124,17 @@ export class OrchestrationEventStore extends Context.Service< readonly latestApplicationSequence: Effect.Effect; - /** Read the finite retained application-event range `(afterSequence, throughSequence]`. */ + /** + * Read the finite retained application-event range `(afterSequence, throughSequence]`. + * + * A row whose `event_type` this build does not know fails the read unless + * `skipUnknownEventTypes` is set, which logs and drops it instead. Only + * client shell replay opts in; anything that builds server state stays strict. + */ readonly readApplicationEvents: (input: { readonly afterSequence: number; readonly throughSequence: number; + readonly skipUnknownEventTypes?: boolean; }) => Stream.Stream; /** Publish only after the surrounding event/projection transaction commits. */ @@ -266,6 +279,21 @@ function rowToApplicationStoredEvent( return row.aggregate_kind === "project" ? rowToProjectEvent(row) : rowToV2StoredEvent(row); } +const isKnownProjectEventType = Schema.is(ProjectEventType); + +/** Filter only unknown discriminants; known types still undergo strict payload decoding. */ +const keepReplayRow = (row: ApplicationEventRow, skipUnknownEventTypes?: boolean) => + skipUnknownEventTypes !== true || + (row.aggregate_kind === "project" + ? isKnownProjectEventType(row.event_type) + : isKnownOrchestrationV2EventType(row.event_type)) + ? Effect.succeed(true) + : Effect.logWarning("Skipping an application event row with an unknown event type", { + aggregateKind: row.aggregate_kind, + sequence: row.sequence, + eventTypeLength: row.event_type.length, + }).pipe(Effect.as(false)); + function inferActorKind(event: UnsequencedProjectEvent): typeof ActorKind.Type { if (event.commandId !== null && event.commandId.startsWith("provider:")) { return "provider"; @@ -534,6 +562,7 @@ const makeEventStore = Effect.gen(function* () { ); }, ).pipe( + Stream.filterEffect((row) => keepReplayRow(row, input?.skipUnknownEventTypes)), Stream.mapEffect((row) => rowToV2StoredEvent(row).pipe( Effect.mapError( @@ -626,53 +655,43 @@ const makeEventStore = Effect.gen(function* () { ), ); - const readApplicationEventPage = (input: { - readonly afterSequence: number; - readonly throughSequence: number; - readonly limit: number; - }): Stream.Stream => - Stream.fromEffect( - readApplicationRows(input).pipe( - Effect.mapError( - toPersistenceSqlError("OrchestrationEventStore.readApplicationEvents:query"), - ), - ), - ).pipe( - Stream.flatMap(Stream.fromIterable), - Stream.mapEffect((row) => - rowToApplicationStoredEvent(row).pipe( - Effect.mapError( - toPersistenceDecodeError("OrchestrationEventStore.readApplicationEvents:decode"), - ), - ), - ), - ); - + // Advance over raw rows so even a full page of unknown types cannot end replay early. const catchUpApplicationEvents = (input: { readonly afterSequence: number; readonly throughSequence: number; - }): Stream.Stream => { - return Stream.paginate(input.afterSequence, (afterSequence) => - readApplicationEventPage({ + readonly skipUnknownEventTypes?: boolean; + }): Stream.Stream => + Stream.paginate(input.afterSequence, (afterSequence) => + readApplicationRows({ afterSequence, throughSequence: input.throughSequence, limit: READ_PAGE_SIZE, }).pipe( - Stream.runCollect, - Effect.map((events) => { - const last = events.at(-1); + Effect.mapError( + toPersistenceSqlError("OrchestrationEventStore.readApplicationEvents:query"), + ), + Effect.map((rows) => { + const last = rows.at(-1); return [ - events, + rows, last === undefined || - events.length < READ_PAGE_SIZE || + rows.length < READ_PAGE_SIZE || last.sequence >= input.throughSequence ? Option.none() : Option.some(last.sequence), ] as const; }), ), + ).pipe( + Stream.filterEffect((row) => keepReplayRow(row, input.skipUnknownEventTypes)), + Stream.mapEffect((row) => + rowToApplicationStoredEvent(row).pipe( + Effect.mapError( + toPersistenceDecodeError("OrchestrationEventStore.readApplicationEvents:decode"), + ), + ), + ), ); - }; const streamProjectedApplicationEvents: OrchestrationEventStore["Service"]["streamProjectedApplicationEvents"] = (input) => diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 0f735efd31c705686e014ddab76dffd6876c5829..476224cb2cd44cc2b49eda16dc52813c825b95be 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1,6 +1,10 @@ import { assert, it } from "@effect/vitest"; import { ORCHESTRATION_PROTOCOL_VERSION, + ProjectId, + ThreadId, + type OrchestrationV2StoredEvent, + type OrchestrationProjectShell, type ServerConfig, type ServerConfigStreamEvent, } from "@t3tools/contracts"; @@ -13,16 +17,26 @@ import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { ChildProcessSpawner } from "effect/process"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; +import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; +import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; +import * as SqlitePersistence from "./persistence/Sqlite.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; +import * as ProjectEnrichment from "./project/ProjectEnrichmentService.ts"; +import * as ProjectService from "./project/ProjectService.ts"; import { hasCompatibleOrchestrationProtocol, resolveAvailableEditorsForConfig, shouldUseBoundedThreadSnapshot, + subscribeOrchestrationV2Shell, + subscribeOrchestrationV2Thread, withLateEditorConfig, } from "./ws.ts"; @@ -234,3 +248,119 @@ it.effect("recovers a reveal kind whose real probe outlasts the config timeout", } }).pipe(Effect.scoped), ); + +const ReplayTestLayer = OrchestrationEventStore.layer.pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), +); + +it.live.each(["thread", "shell"] as const)( + "%s resume skips unknown types, synchronizes, and receives the live tail", + (scope) => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const now = "2026-01-05T00:00:00.000Z"; + const threadId = ThreadId.make("thread:replay"); + const projectId = ProjectId.make("project:replay"); + const project = { + id: projectId, + title: "Replay", + workspaceRoot: "/tmp/replay", + defaultModelSelection: null, + scripts: [], + createdAt: now, + updatedAt: now, + } satisfies OrchestrationProjectShell; + const insert = (version: number, unknown = false) => sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`replay:${version}`}, ${scope === "thread" ? "thread" : "project"}, + ${scope === "thread" ? threadId : projectId}, ${version}, + ${unknown ? "future.event" : scope === "thread" ? "provider-session.detached" : "project.meta-updated"}, + ${now}, 'server', ${JSON.stringify( + scope === "thread" + ? { providerSessionId: "session:replay", detachedAt: now } + : { projectId, updatedAt: now }, + )}, '{}', 2 + ) RETURNING sequence + `; + const [before] = yield* insert(0); + yield* insert(1, true); + const [after] = yield* insert(2); + const [highWater] = yield* insert(3, true); + const updates = yield* PubSub.unbounded(); + let liveSequence = 0; + const consume = ( + stream: Stream.Stream, + ) => + stream.pipe( + Stream.tap((item) => + Effect.gen(function* () { + if (item.kind !== "synchronized") return; + const [live] = yield* insert(4); + liveSequence = live!.sequence; + const events = yield* store + .readApplicationEvents({ + afterSequence: highWater!.sequence, + throughSequence: liveSequence, + }) + .pipe(Stream.runCollect); + yield* store.publishCommitted(events); + }), + ), + Stream.takeUntil((item) => item.sequence === liveSequence), + Stream.map((item) => ({ kind: item.kind as string, sequence: item.sequence })), + Stream.runCollect, + ); + const subscription = Effect.gen(function* () { + return scope === "thread" + ? yield* subscribeOrchestrationV2Thread({ + threadId, + afterSequence: before!.sequence, + requestCompletionMarker: true, + }).pipe(Effect.flatMap(consume)) + : yield* subscribeOrchestrationV2Shell({ + afterSequence: before!.sequence, + requestCompletionMarker: true, + }).pipe(Effect.flatMap(consume)); + }); + const items = yield* subscription.pipe( + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + streamStoredEventsFrom: ( + input: Parameters< + ThreadManagement.ThreadManagementService["Service"]["streamStoredEventsFrom"] + >[0], + ) => { + assert.equal(input?.afterSequence, highWater!.sequence); + return store + .streamApplicationEvents({ afterSequence: input?.afterSequence ?? 0 }) + .pipe( + Stream.filter((event): event is OrchestrationV2StoredEvent => "event" in event), + ); + }, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(ProjectStore.ProjectStoreV2, { + listShells: () => Effect.succeed([]), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService(ProjectService.ProjectService, { + getShell: () => Effect.succeedSome(project), + } as unknown as ProjectService.ProjectService["Service"]), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(updates), + } as ProjectEnrichment.ProjectEnrichmentService["Service"]), + ); + assert.deepEqual( + items.map((item) => item.kind), + scope === "thread" + ? ["event", "synchronized", "event"] + : ["project.updated", "synchronized", "project.updated"], + ); + assert.deepEqual( + items.flatMap((item) => (item.sequence === undefined ? [] : [item.sequence])), + [after!.sequence, liveSequence], + ); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 8ee055e2a1e44ac57e06f7aa97efbf1e4b47e17b..e6bb4cab906aa9b324932dfcb0cf7090130f52b2 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -805,6 +805,10 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs afterSequence, throughSequence, limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1, + // A row a newer build wrote (e.g. after a downgrade) must not fail + // every resubscribe. The live tail resumes from `highWater`, so a + // dropped row never hides later events. + skipUnknownEventTypes: true, }) .pipe( Stream.map((stored) => ({ @@ -1159,6 +1163,10 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc applicationEvents.readApplicationEvents({ afterSequence: input.afterSequence, throughSequence: highWater, + // A row a newer build wrote (e.g. after a downgrade) must not fail + // every shell resubscribe. The live tail below resumes from + // highWater, so a dropped row never hides later events. + skipUnknownEventTypes: true, }), ); return composeShellStreamWithEnrichment({ diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index b4db1038f4ebf0761304f60603edfdf023eb2040..8228e289a47351a598a7d4f2018441dfc7ef03c8 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3438,6 +3438,11 @@ const knownDomainEventTypes: ReadonlySet = new Set( }), ); +/** Whether `type` is an {@link OrchestrationV2DomainEvent} discriminant in this build. */ +export function isKnownOrchestrationV2EventType(type: string): boolean { + return knownDomainEventTypes.has(type); +} + /** * A thread event whose type this build does not know, or a turn-item.updated * carrying a turn item type it does not know. Newer servers add both; older