authorgravatar for 91240327+shivamhwp@users.noreply.github.comshivam <91240327+shivamhwp@users.noreply.github.com> 2026-10-13 08:31:25+05:30
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-13 08:31:25+05:30
log5a65169bfc853b8817c3e53e223ffa33ac7d5467
tree7902054e080ee0e50fd57ef6a8066e7e8a772a9f
parente24ea1d28ef9159cc6cc98a1e9d9de5f52ed385c
signature Signed by PGP key B5690EEEBB952194

fix(server): a thread with an event type from a newer build can still resync after a downgrade (#15937)

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> Co-authored-by: PR Batch Tester <agent@local.test>

5 files changed, 290 insertions(+), 33 deletions(-)

apps/server/src/persistence/OrchestrationEventStore.test.ts+95
......@@ -393,6 +393,101 @@ layer("OrchestrationEventStore", (it) => {
393393 );
394394 }),
395395 );
396
397 it.effect.each([
398 ["agent", "thread"],
399 ["application", "thread"],
400 ["application", "project"],
401 ] as const)(
402 "%s replay skips unknown %s types but rejects malformed known payloads",
403 ([reader, aggregateKind]) =>
404 Effect.gen(function* () {
405 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
406 const sql = yield* SqlClient.SqlClient;
407 const streamId = ThreadId.make(`skip-unknown:${reader}:${aggregateKind}`);
408 const now = "2026-01-05T00:00:00.000Z";
409 // Direct SQL models rows written by a newer build, which append cannot accept.
410 const insertRow = Effect.fnUntraced(function* (
411 version: number,
412 eventType: string,
413 payload: string,
414 ) {
415 const [row] = yield* sql<{ sequence: number }>`
416 INSERT INTO orchestration_events (
417 event_id, aggregate_kind, stream_id, stream_version, event_type,
418 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
419 ) VALUES (
420 ${`${streamId}:${version}`}, ${aggregateKind}, ${streamId}, ${version}, ${eventType},
421 ${now}, 'server', ${payload}, '{}', 2
422 ) RETURNING sequence
423 `;
424 return row!.sequence;
425 });
426 const read = (
427 afterSequence: number,
428 throughSequence: number,
429 skipUnknownEventTypes = false,
430 limit?: number,
431 ) => {
432 const input = { afterSequence, throughSequence, skipUnknownEventTypes };
433 return reader === "agent"
434 ? store
435 .readAgentEvents({
436 ...input,
437 threadId: streamId,
438 ...(limit === undefined ? {} : { limit }),
439 })
440 .pipe(
441 Stream.map((event) => event.sequence),
442 Stream.runCollect,
443 )
444 : store.readApplicationEvents(input).pipe(
445 Stream.map((event) => event.sequence),
446 Stream.runCollect,
447 );
448 };
449 const knownType =
450 aggregateKind === "thread" ? "provider-session.detached" : "project.meta-updated";
451 const knownPayload = JSON.stringify(
452 aggregateKind === "thread"
453 ? { providerSessionId: "session:known", detachedAt: now }
454 : { projectId: streamId, updatedAt: now },
455 );
456 const baseline = yield* store.latestApplicationSequence;
457 const before = yield* insertRow(0, knownType, knownPayload);
458 // More than one raw page ensures replay crosses an entirely filtered page.
459 yield* Effect.forEach(
460 Array.from({ length: 1_000 }, (_, i) => i + 1),
461 (version) => insertRow(version, "future.event", "not JSON"),
462 { discard: true },
463 );
464 // A type known for the other aggregate kind is unknown for this one.
465 yield* insertRow(
466 1_001,
467 aggregateKind === "thread" ? "project.meta-updated" : "provider-session.detached",
468 "{}",
469 );
470 const after = yield* insertRow(1_002, knownType, knownPayload);
471 const highWater = yield* insertRow(1_003, "future.trailing", "{}");
472 yield* insertRow(1_004, knownType, knownPayload);
473 assert.deepEqual(yield* read(baseline, highWater, true), [before, after]);
474 assert.isTrue(isPersistenceDecodeError(yield* read(baseline, highWater).pipe(Effect.flip)));
475 if (reader === "agent") {
476 // The work limit includes skipped rows, rather than fetching extra pages to fill it.
477 assert.deepEqual(yield* read(baseline, highWater, true, 2), [before]);
478 }
479 const beforeBroken = yield* store.latestApplicationSequence;
480 // Exercise a grouped V2 literal as well as the project type discriminator.
481 const broken = yield* insertRow(
482 1_005,
483 aggregateKind === "thread" ? "thread.metadata-updated" : "project.created",
484 "{}",
485 );
486 assert.isTrue(
487 isPersistenceDecodeError(yield* read(beforeBroken, broken, true).pipe(Effect.flip)),
488 );
489 }),
490 );
396491});
397492
398493it.effect.each(["high-water", "replay"] as const)(
apps/server/src/persistence/OrchestrationEventStore.ts+52-33
......@@ -17,6 +17,7 @@ import {
1717 CommandId,
1818 EventId,
1919 IsoDateTime,
20 isKnownOrchestrationV2EventType,
2021 NonNegativeInt,
2122 OrchestrationV2DomainEventJson,
2223 OrchestrationV2StoredEvent,
......@@ -69,7 +70,11 @@ export class OrchestrationEventStore extends Context.Service<
6970 * Read only V2 thread events from the application log.
7071 *
7172 * Reads in fixed-size sequence pages until the filtered range is exhausted;
72 * `limit` caps the total emitted events across pages.
73 * `limit` caps raw rows across pages, including skipped unknown types.
74 *
75 * A row whose `event_type` this build does not know fails the read unless
76 * `skipUnknownEventTypes` is set, which logs and drops it instead. Only
77 * client thread replay opts in; anything that builds server state stays strict.
7378 */
7479 readonly readAgentEvents: (input?: {
7580 readonly afterSequence?: number;
......@@ -78,6 +83,7 @@ export class OrchestrationEventStore extends Context.Service<
7883 readonly commandId?: CommandId;
7984 readonly eventType?: OrchestrationV2DomainEvent["type"];
8085 readonly limit?: number;
86 readonly skipUnknownEventTypes?: boolean;
8187 }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestrationEventStoreError>;
8288
8389 /** Measure one thread's bounded replay without loading or decoding its payloads. */
......@@ -118,10 +124,17 @@ export class OrchestrationEventStore extends Context.Service<
118124
119125 readonly latestApplicationSequence: Effect.Effect<number, OrchestrationEventStoreError>;
120126
121 /** Read the finite retained application-event range `(afterSequence, throughSequence]`. */
127 /**
128 * Read the finite retained application-event range `(afterSequence, throughSequence]`.
129 *
130 * A row whose `event_type` this build does not know fails the read unless
131 * `skipUnknownEventTypes` is set, which logs and drops it instead. Only
132 * client shell replay opts in; anything that builds server state stays strict.
133 */
122134 readonly readApplicationEvents: (input: {
123135 readonly afterSequence: number;
124136 readonly throughSequence: number;
137 readonly skipUnknownEventTypes?: boolean;
125138 }) => Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError>;
126139
127140 /** Publish only after the surrounding event/projection transaction commits. */
......@@ -266,6 +279,21 @@ function rowToApplicationStoredEvent(
266279 return row.aggregate_kind === "project" ? rowToProjectEvent(row) : rowToV2StoredEvent(row);
267280}
268281
282const isKnownProjectEventType = Schema.is(ProjectEventType);
283
284/** Filter only unknown discriminants; known types still undergo strict payload decoding. */
285const keepReplayRow = (row: ApplicationEventRow, skipUnknownEventTypes?: boolean) =>
286 skipUnknownEventTypes !== true ||
287 (row.aggregate_kind === "project"
288 ? isKnownProjectEventType(row.event_type)
289 : isKnownOrchestrationV2EventType(row.event_type))
290 ? Effect.succeed(true)
291 : Effect.logWarning("Skipping an application event row with an unknown event type", {
292 aggregateKind: row.aggregate_kind,
293 sequence: row.sequence,
294 eventTypeLength: row.event_type.length,
295 }).pipe(Effect.as(false));
296
269297function inferActorKind(event: UnsequencedProjectEvent): typeof ActorKind.Type {
270298 if (event.commandId !== null && event.commandId.startsWith("provider:")) {
271299 return "provider";
......@@ -534,6 +562,7 @@ const makeEventStore = Effect.gen(function* () {
534562 );
535563 },
536564 ).pipe(
565 Stream.filterEffect((row) => keepReplayRow(row, input?.skipUnknownEventTypes)),
537566 Stream.mapEffect((row) =>
538567 rowToV2StoredEvent(row).pipe(
539568 Effect.mapError(
......@@ -626,53 +655,43 @@ const makeEventStore = Effect.gen(function* () {
626655 ),
627656 );
628657
629 const readApplicationEventPage = (input: {
630 readonly afterSequence: number;
631 readonly throughSequence: number;
632 readonly limit: number;
633 }): Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError> =>
634 Stream.fromEffect(
635 readApplicationRows(input).pipe(
636 Effect.mapError(
637 toPersistenceSqlError("OrchestrationEventStore.readApplicationEvents:query"),
638 ),
639 ),
640 ).pipe(
641 Stream.flatMap(Stream.fromIterable),
642 Stream.mapEffect((row) =>
643 rowToApplicationStoredEvent(row).pipe(
644 Effect.mapError(
645 toPersistenceDecodeError("OrchestrationEventStore.readApplicationEvents:decode"),
646 ),
647 ),
648 ),
649 );
650
658 // Advance over raw rows so even a full page of unknown types cannot end replay early.
651659 const catchUpApplicationEvents = (input: {
652660 readonly afterSequence: number;
653661 readonly throughSequence: number;
654 }): Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError> => {
655 return Stream.paginate(input.afterSequence, (afterSequence) =>
656 readApplicationEventPage({
662 readonly skipUnknownEventTypes?: boolean;
663 }): Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError> =>
664 Stream.paginate(input.afterSequence, (afterSequence) =>
665 readApplicationRows({
657666 afterSequence,
658667 throughSequence: input.throughSequence,
659668 limit: READ_PAGE_SIZE,
660669 }).pipe(
661 Stream.runCollect,
662 Effect.map((events) => {
663 const last = events.at(-1);
670 Effect.mapError(
671 toPersistenceSqlError("OrchestrationEventStore.readApplicationEvents:query"),
672 ),
673 Effect.map((rows) => {
674 const last = rows.at(-1);
664675 return [
665 events,
676 rows,
666677 last === undefined ||
667 events.length < READ_PAGE_SIZE ||
678 rows.length < READ_PAGE_SIZE ||
668679 last.sequence >= input.throughSequence
669680 ? Option.none()
670681 : Option.some(last.sequence),
671682 ] as const;
672683 }),
673684 ),
685 ).pipe(
686 Stream.filterEffect((row) => keepReplayRow(row, input.skipUnknownEventTypes)),
687 Stream.mapEffect((row) =>
688 rowToApplicationStoredEvent(row).pipe(
689 Effect.mapError(
690 toPersistenceDecodeError("OrchestrationEventStore.readApplicationEvents:decode"),
691 ),
692 ),
693 ),
674694 );
675 };
676695
677696 const streamProjectedApplicationEvents: OrchestrationEventStore["Service"]["streamProjectedApplicationEvents"] =
678697 (input) =>
apps/server/src/ws.test.ts+130
......@@ -1,6 +1,10 @@
11import { assert, it } from "@effect/vitest";
22import {
33 ORCHESTRATION_PROTOCOL_VERSION,
4 ProjectId,
5 ThreadId,
6 type OrchestrationV2StoredEvent,
7 type OrchestrationProjectShell,
48 type ServerConfig,
59 type ServerConfigStreamEvent,
610} from "@t3tools/contracts";
......@@ -13,16 +17,26 @@ import * as Fiber from "effect/Fiber";
1317import * as FileSystem from "effect/FileSystem";
1418import * as Layer from "effect/Layer";
1519import * as Path from "effect/Path";
20import * as PubSub from "effect/PubSub";
1621import * as Queue from "effect/Queue";
1722import * as Stream from "effect/Stream";
1823import * as TestClock from "effect/testing/TestClock";
1924import { ChildProcessSpawner } from "effect/process";
25import * as SqlClient from "effect/sql/SqlClient";
2026
27import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts";
28import * as ProjectStore from "./orchestration-v2/ProjectStore.ts";
29import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts";
30import * as SqlitePersistence from "./persistence/Sqlite.ts";
2131import * as ExternalLauncher from "./process/externalLauncher.ts";
32import * as ProjectEnrichment from "./project/ProjectEnrichmentService.ts";
33import * as ProjectService from "./project/ProjectService.ts";
2234import {
2335 hasCompatibleOrchestrationProtocol,
2436 resolveAvailableEditorsForConfig,
2537 shouldUseBoundedThreadSnapshot,
38 subscribeOrchestrationV2Shell,
39 subscribeOrchestrationV2Thread,
2640 withLateEditorConfig,
2741} from "./ws.ts";
2842
......@@ -234,3 +248,119 @@ it.effect("recovers a reveal kind whose real probe outlasts the config timeout",
234248 }
235249 }).pipe(Effect.scoped),
236250);
251
252const ReplayTestLayer = OrchestrationEventStore.layer.pipe(
253 Layer.provideMerge(SqlitePersistence.layerMemory),
254);
255
256it.live.each(["thread", "shell"] as const)(
257 "%s resume skips unknown types, synchronizes, and receives the live tail",
258 (scope) =>
259 Effect.gen(function* () {
260 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
261 const sql = yield* SqlClient.SqlClient;
262 const now = "2026-01-05T00:00:00.000Z";
263 const threadId = ThreadId.make("thread:replay");
264 const projectId = ProjectId.make("project:replay");
265 const project = {
266 id: projectId,
267 title: "Replay",
268 workspaceRoot: "/tmp/replay",
269 defaultModelSelection: null,
270 scripts: [],
271 createdAt: now,
272 updatedAt: now,
273 } satisfies OrchestrationProjectShell;
274 const insert = (version: number, unknown = false) => sql<{ sequence: number }>`
275 INSERT INTO orchestration_events (
276 event_id, aggregate_kind, stream_id, stream_version, event_type,
277 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
278 ) VALUES (
279 ${`replay:${version}`}, ${scope === "thread" ? "thread" : "project"},
280 ${scope === "thread" ? threadId : projectId}, ${version},
281 ${unknown ? "future.event" : scope === "thread" ? "provider-session.detached" : "project.meta-updated"},
282 ${now}, 'server', ${JSON.stringify(
283 scope === "thread"
284 ? { providerSessionId: "session:replay", detachedAt: now }
285 : { projectId, updatedAt: now },
286 )}, '{}', 2
287 ) RETURNING sequence
288 `;
289 const [before] = yield* insert(0);
290 yield* insert(1, true);
291 const [after] = yield* insert(2);
292 const [highWater] = yield* insert(3, true);
293 const updates = yield* PubSub.unbounded();
294 let liveSequence = 0;
295 const consume = <A extends { readonly kind: string; readonly sequence?: number }, E, R>(
296 stream: Stream.Stream<A, E, R>,
297 ) =>
298 stream.pipe(
299 Stream.tap((item) =>
300 Effect.gen(function* () {
301 if (item.kind !== "synchronized") return;
302 const [live] = yield* insert(4);
303 liveSequence = live!.sequence;
304 const events = yield* store
305 .readApplicationEvents({
306 afterSequence: highWater!.sequence,
307 throughSequence: liveSequence,
308 })
309 .pipe(Stream.runCollect);
310 yield* store.publishCommitted(events);
311 }),
312 ),
313 Stream.takeUntil((item) => item.sequence === liveSequence),
314 Stream.map((item) => ({ kind: item.kind as string, sequence: item.sequence })),
315 Stream.runCollect,
316 );
317 const subscription = Effect.gen(function* () {
318 return scope === "thread"
319 ? yield* subscribeOrchestrationV2Thread({
320 threadId,
321 afterSequence: before!.sequence,
322 requestCompletionMarker: true,
323 }).pipe(Effect.flatMap(consume))
324 : yield* subscribeOrchestrationV2Shell({
325 afterSequence: before!.sequence,
326 requestCompletionMarker: true,
327 }).pipe(Effect.flatMap(consume));
328 });
329 const items = yield* subscription.pipe(
330 Effect.provideService(ThreadManagement.ThreadManagementService, {
331 ensureLegacyTranscript: () => Effect.void,
332 streamStoredEventsFrom: (
333 input: Parameters<
334 ThreadManagement.ThreadManagementService["Service"]["streamStoredEventsFrom"]
335 >[0],
336 ) => {
337 assert.equal(input?.afterSequence, highWater!.sequence);
338 return store
339 .streamApplicationEvents({ afterSequence: input?.afterSequence ?? 0 })
340 .pipe(
341 Stream.filter((event): event is OrchestrationV2StoredEvent => "event" in event),
342 );
343 },
344 } as unknown as ThreadManagement.ThreadManagementService["Service"]),
345 Effect.provideService(ProjectStore.ProjectStoreV2, {
346 listShells: () => Effect.succeed([]),
347 } as unknown as ProjectStore.ProjectStoreV2["Service"]),
348 Effect.provideService(ProjectService.ProjectService, {
349 getShell: () => Effect.succeedSome(project),
350 } as unknown as ProjectService.ProjectService["Service"]),
351 Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, {
352 subscribeChanges: PubSub.subscribe(updates),
353 } as ProjectEnrichment.ProjectEnrichmentService["Service"]),
354 );
355 assert.deepEqual(
356 items.map((item) => item.kind),
357 scope === "thread"
358 ? ["event", "synchronized", "event"]
359 : ["project.updated", "synchronized", "project.updated"],
360 );
361 assert.deepEqual(
362 items.flatMap((item) => (item.sequence === undefined ? [] : [item.sequence])),
363 [after!.sequence, liveSequence],
364 );
365 }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)),
366);
apps/server/src/ws.ts+8
......@@ -805,6 +805,10 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs
805805 afterSequence,
806806 throughSequence,
807807 limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1,
808 // A row a newer build wrote (e.g. after a downgrade) must not fail
809 // every resubscribe. The live tail resumes from `highWater`, so a
810 // dropped row never hides later events.
811 skipUnknownEventTypes: true,
808812 })
809813 .pipe(
810814 Stream.map((stored) => ({
......@@ -1159,6 +1163,10 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc
11591163 applicationEvents.readApplicationEvents({
11601164 afterSequence: input.afterSequence,
11611165 throughSequence: highWater,
1166 // A row a newer build wrote (e.g. after a downgrade) must not fail
1167 // every shell resubscribe. The live tail below resumes from
1168 // highWater, so a dropped row never hides later events.
1169 skipUnknownEventTypes: true,
11621170 }),
11631171 );
11641172 return composeShellStreamWithEnrichment({
packages/contracts/src/orchestrationV2.ts+5
......@@ -3438,6 +3438,11 @@ const knownDomainEventTypes: ReadonlySet<string> = new Set(
34383438 }),
34393439);
34403440
3441/** Whether `type` is an {@link OrchestrationV2DomainEvent} discriminant in this build. */
3442export function isKnownOrchestrationV2EventType(type: string): boolean {
3443 return knownDomainEventTypes.has(type);
3444}
3445
34413446/**
34423447 * A thread event whose type this build does not know, or a turn-item.updated
34433448 * carrying a turn item type it does not know. Newer servers add both; older