Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@ import { assert, it } from "@effect/vitest";
import {
ComposerContextId,
EventId,
ProjectId,
ProviderInstanceId,
type OrchestrationMessageContext,
ThreadId,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -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* () {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<ThreadId>();

const ensureTranscriptBase = (threadId: ThreadId) =>
Expand All @@ -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);
Expand Down
Loading