From 6cee9137c36a408031a4fa96f7d72d84409bad9f Mon Sep 17 00:00:00 2001 From: ScottN-PV Date: Sat, 24 Oct 2026 23:21:53 -0400 Subject: [PATCH] fix(server): cache absent legacy transcripts for V2 threads --- .../legacy/LegacyV1ThreadImporter.test.ts | 65 +++++++++++++++++++ .../legacy/LegacyV1ThreadImporter.ts | 6 +- 2 files changed, 67 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts index 3b2351b28a1e..87059b7c1d0a 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts @@ -2,6 +2,8 @@ import { assert, it } from "@effect/vitest"; import { ComposerContextId, EventId, + ProjectId, + ProviderInstanceId, type OrchestrationMessageContext, ThreadId, } from "@t3tools/contracts"; @@ -150,6 +152,69 @@ it.effect("propagates interruption from transcript reads and retry-marker writes }).pipe(Effect.provide(layerTest)), ); +it.effect("caches the absent legacy transcript for a native V2 thread", () => + Effect.gen(function* () { + const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; + const eventSink = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make("thread:v2-born"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const now = DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"); + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:v2-born"), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + id: threadId, + projectId: ProjectId.make("project:v2-born"), + title: "Native V2 thread", + providerInstanceId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-6" }, + createdBy: "user", + creationSource: "web", + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }, + ], + }); + const before = yield* projections.getThreadProjection(threadId); + let legacyLookups = 0; + for (let read = 0; read < 3; read++) { + const summary = yield* importer.ensureTranscript(threadId).pipe( + Effect.provideService(Statement.CurrentTransformer, (statement) => { + const [query, parameters] = statement.compile(); + if ( + query.includes("FROM orchestration_v2_legacy_imports") && + parameters[0] === threadId + ) { + legacyLookups++; + } + return Effect.succeed(statement); + }), + ); + assert.deepStrictEqual(summary, { importedThreadCount: 0, importedMessageCount: 0 }); + } + assert.equal(legacyLookups, 1); + assert.deepStrictEqual(yield* projections.getThreadProjection(threadId), before); + }).pipe(Effect.provide(layerTest)), +); + it.layer(layerTest)("LegacyV1ThreadImporter", (it) => { it.effect("imports lightweight shells, hydrates transcripts, and remains idempotent", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts index ee4baf0abf83..d82a9e80a23b 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts @@ -677,7 +677,7 @@ const make = Effect.gen(function* () { ), ); - // transcript_imported_at never resets; cache confirmed imports for the process lifetime to skip repeated locks and reads. + // Both results are final for this process: transcript_imported_at never resets, and only startup shell reconciliation creates legacy rows. const confirmedTranscriptThreadIds = new Set(); const ensureTranscriptBase = (threadId: ThreadId) => @@ -692,9 +692,7 @@ const make = Effect.gen(function* () { `.pipe(NodeSqliteClient.readOnly); const imported = imports[0]; if (imported === undefined || imported.transcript_imported_at !== null) { - if (imported !== undefined) { - confirmedTranscriptThreadIds.add(threadId); - } + confirmedTranscriptThreadIds.add(threadId); return { importedThreadCount: 0, importedMessageCount: 0 }; } const messages = yield* listMessages(threadId);