From 193034c0230de0906a88019e70b4e91ffa3ede9a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 24 Jul 2026 00:03:24 +0200 Subject: [PATCH 1/2] feat(server): migrate v1 state into orchestrator v2 Co-authored-by: codex --- .../src/app/DesktopAppIdentity.test.ts | 6 +- .../app/DesktopConnectionCatalogStore.test.ts | 16 +- .../src/app/DesktopEnvironment.test.ts | 30 +- apps/desktop/src/app/DesktopEnvironment.ts | 6 +- .../DesktopBackendConfiguration.test.ts | 2 +- .../backend/DesktopBackendConfiguration.ts | 2 +- .../DesktopClientSettings.diagnostics.test.ts | 2 +- .../settings/DesktopSavedEnvironments.test.ts | 8 +- apps/server/src/cli/config.test.ts | 6 +- apps/server/src/config.ts | 2 +- .../ContextHandoffService.test.ts | 107 +++ .../orchestration-v2/ContextHandoffService.ts | 81 +++ .../LegacyV1ThreadImporter.test.ts | 238 +++++++ .../LegacyV1ThreadImporter.ts | 639 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 29 +- .../src/orchestration-v2/ProjectionStore.ts | 3 + .../ThreadManagementService.ts | 60 +- .../src/orchestration-v2/runtimeLayer.ts | 9 +- apps/server/src/persistence/Migrations.ts | 2 + .../035_036_OrchestrationV2.test.ts | 14 +- .../Migrations/043_LegacyV1ImportState.ts | 27 + apps/server/src/serverRuntimeStartup.test.ts | 3 + apps/server/src/serverRuntimeStartup.ts | 25 + apps/server/src/ws.ts | 1 + packages/contracts/src/orchestrationV2.ts | 5 + packages/ssh/src/tunnel.test.ts | 2 +- packages/ssh/src/tunnel.ts | 2 +- 27 files changed, 1273 insertions(+), 54 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ContextHandoffService.test.ts create mode 100644 apps/server/src/orchestration-v2/LegacyV1ThreadImporter.test.ts create mode 100644 apps/server/src/orchestration-v2/LegacyV1ThreadImporter.ts create mode 100644 apps/server/src/persistence/Migrations/043_LegacyV1ImportState.ts diff --git a/apps/desktop/src/app/DesktopAppIdentity.test.ts b/apps/desktop/src/app/DesktopAppIdentity.test.ts index 8629193b207..3c95b266bc1 100644 --- a/apps/desktop/src/app/DesktopAppIdentity.test.ts +++ b/apps/desktop/src/app/DesktopAppIdentity.test.ts @@ -126,7 +126,7 @@ const withIdentity = ( input.legacyPathProbeError ? Effect.fail(input.legacyPathProbeError) : Effect.succeed( - input.legacyPathExists === true && path.includes("T3 Code (Alpha v2)"), + input.legacyPathExists === true && path.includes("T3 Code (Alpha)"), ), readFileString: () => Effect.succeed(input.packageJson ?? '{"t3codeCommitHash":"abcdef1234567890"}'), @@ -147,14 +147,14 @@ describe("DesktopAppIdentity", () => { const identity = yield* DesktopAppIdentity.DesktopAppIdentity; const userDataPath = yield* identity.resolveUserDataPath; - assert.equal(userDataPath, "/Users/alice/Library/Application Support/T3 Code (Alpha v2)"); + assert.equal(userDataPath, "/Users/alice/Library/Application Support/T3 Code (Alpha)"); }), { legacyPathExists: true }, ), ); it.effect("preserves failures while inspecting the legacy userData path", () => { - const legacyPath = "/Users/alice/Library/Application Support/T3 Code (Alpha v2)"; + const legacyPath = "/Users/alice/Library/Application Support/T3 Code (Alpha)"; const cause = PlatformError.systemError({ _tag: "PermissionDenied", module: "FileSystem", diff --git a/apps/desktop/src/app/DesktopConnectionCatalogStore.test.ts b/apps/desktop/src/app/DesktopConnectionCatalogStore.test.ts index 1ee8817763e..7c7818994f6 100644 --- a/apps/desktop/src/app/DesktopConnectionCatalogStore.test.ts +++ b/apps/desktop/src/app/DesktopConnectionCatalogStore.test.ts @@ -254,7 +254,7 @@ describe("DesktopConnectionCatalogStore", () => { _tag: "PermissionDenied", module: "FileSystem", method: "readFileString", - pathOrDescriptor: `${baseDir}/userdata-v2/connection-catalog.json`, + pathOrDescriptor: `${baseDir}/userdata/connection-catalog.json`, }); const fileSystemLayer = Layer.succeed( FileSystem.FileSystem, @@ -271,11 +271,11 @@ describe("DesktopConnectionCatalogStore", () => { error, DesktopConnectionCatalogStore.DesktopConnectionCatalogStoreReadError, ); - assert.equal(error.catalogPath, `${baseDir}/userdata-v2/connection-catalog.json`); + assert.equal(error.catalogPath, `${baseDir}/userdata/connection-catalog.json`); assert.strictEqual(error.cause, permissionError); assert.equal( error.message, - `Failed to read the desktop connection catalog at ${baseDir}/userdata-v2/connection-catalog.json.`, + `Failed to read the desktop connection catalog at ${baseDir}/userdata/connection-catalog.json.`, ); assert.notEqual(error.message, permissionError.message); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), @@ -291,7 +291,7 @@ describe("DesktopConnectionCatalogStore", () => { _tag: "PermissionDenied", module: "FileSystem", method: "makeDirectory", - pathOrDescriptor: `${baseDir}/userdata-v2`, + pathOrDescriptor: `${baseDir}/userdata`, }); const fileSystemLayer = Layer.succeed( FileSystem.FileSystem, @@ -309,11 +309,11 @@ describe("DesktopConnectionCatalogStore", () => { DesktopConnectionCatalogStore.DesktopConnectionCatalogStoreWriteError, ); assert.equal(error.operation, "create-directory"); - assert.equal(error.path, `${baseDir}/userdata-v2`); + assert.equal(error.path, `${baseDir}/userdata`); assert.strictEqual(error.cause, permissionError); assert.equal( error.message, - `Desktop connection catalog write failed during create-directory at ${baseDir}/userdata-v2.`, + `Desktop connection catalog write failed during create-directory at ${baseDir}/userdata.`, ); assert.notEqual(error.message, permissionError.message); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), @@ -398,14 +398,14 @@ describe("DesktopConnectionCatalogStore", () => { DesktopConnectionCatalogStore.DesktopConnectionCatalogStoreProtectionError, ); assert.equal(error.operation, "decrypt-catalog"); - assert.equal(error.catalogPath, `${baseDir}/userdata-v2/connection-catalog.json`); + assert.equal(error.catalogPath, `${baseDir}/userdata/connection-catalog.json`); assert.instanceOf(error.cause, ElectronSafeStorage.ElectronSafeStorageDecryptError); const decryptError = error.cause as ElectronSafeStorage.ElectronSafeStorageDecryptError; assert.instanceOf(decryptError.cause, Error); assert.equal(decryptError.cause.message, "invalid encrypted catalog"); assert.equal( error.message, - `Desktop connection catalog protection failed during decrypt-catalog at ${baseDir}/userdata-v2/connection-catalog.json.`, + `Desktop connection catalog protection failed during decrypt-catalog at ${baseDir}/userdata/connection-catalog.json.`, ); assert.notEqual(error.message, decryptError.message); yield* Ref.set(failDecrypt, false); diff --git a/apps/desktop/src/app/DesktopEnvironment.test.ts b/apps/desktop/src/app/DesktopEnvironment.test.ts index 2f5dc218058..eb355a137f2 100644 --- a/apps/desktop/src/app/DesktopEnvironment.test.ts +++ b/apps/desktop/src/app/DesktopEnvironment.test.ts @@ -53,16 +53,16 @@ describe("DesktopEnvironment", () => { assert.equal(environment.isDevelopment, true); assert.equal(environment.appDataDirectory, "/Users/alice/Library/Application Support"); assert.equal(environment.baseDir, "/tmp/t3"); - assert.equal(environment.stateDir, "/tmp/t3/userdata-v2"); - assert.equal(environment.desktopSettingsPath, "/tmp/t3/userdata-v2/desktop-settings.json"); - assert.equal(environment.clientSettingsPath, "/tmp/t3/userdata-v2/client-settings.json"); + assert.equal(environment.stateDir, "/tmp/t3/userdata"); + assert.equal(environment.desktopSettingsPath, "/tmp/t3/userdata/desktop-settings.json"); + assert.equal(environment.clientSettingsPath, "/tmp/t3/userdata/client-settings.json"); assert.equal( environment.savedEnvironmentRegistryPath, - "/tmp/t3/userdata-v2/saved-environments.json", + "/tmp/t3/userdata/saved-environments.json", ); - assert.equal(environment.serverSettingsPath, "/tmp/t3/userdata-v2/settings.json"); - assert.equal(environment.logDir, "/tmp/t3/userdata-v2/logs"); - assert.equal(environment.browserArtifactsDir, "/tmp/t3/userdata-v2/browser-artifacts"); + assert.equal(environment.serverSettingsPath, "/tmp/t3/userdata/settings.json"); + assert.equal(environment.logDir, "/tmp/t3/userdata/logs"); + assert.equal(environment.browserArtifactsDir, "/tmp/t3/userdata/browser-artifacts"); assert.equal(environment.rootDir, "/repo"); assert.equal(environment.appRoot, "/repo"); assert.equal(environment.backendEntryPath, "/repo/apps/server/dist/bin.mjs"); @@ -81,7 +81,7 @@ describe("DesktopEnvironment", () => { }), ); - it.effect("stores production state under userdata-v2 in an explicit home", () => + it.effect("stores production state under userdata in an explicit home", () => Effect.gen(function* () { const environment = yield* makeEnvironment( {}, @@ -91,12 +91,12 @@ describe("DesktopEnvironment", () => { ); assert.equal(environment.isDevelopment, false); - assert.equal(environment.stateDir, "/tmp/t3/userdata-v2"); - assert.equal(environment.logDir, "/tmp/t3/userdata-v2/logs"); - assert.equal(environment.browserArtifactsDir, "/tmp/t3/userdata-v2/browser-artifacts"); - assert.equal(environment.serverSettingsPath, "/tmp/t3/userdata-v2/settings.json"); - assert.equal(environment.userDataDirName, "t3code-v2"); - assert.equal(environment.legacyUserDataDirName, "T3 Code (Alpha v2)"); + assert.equal(environment.stateDir, "/tmp/t3/userdata"); + assert.equal(environment.logDir, "/tmp/t3/userdata/logs"); + assert.equal(environment.browserArtifactsDir, "/tmp/t3/userdata/browser-artifacts"); + assert.equal(environment.serverSettingsPath, "/tmp/t3/userdata/settings.json"); + assert.equal(environment.userDataDirName, "t3code"); + assert.equal(environment.legacyUserDataDirName, "T3 Code (Alpha)"); }), ); @@ -109,7 +109,7 @@ describe("DesktopEnvironment", () => { const production = yield* makeEnvironment(); assert.equal(development.stateDir, "/Users/alice/.t3/dev"); - assert.equal(production.stateDir, "/Users/alice/.t3/userdata-v2"); + assert.equal(production.stateDir, "/Users/alice/.t3/userdata"); }), ); diff --git a/apps/desktop/src/app/DesktopEnvironment.ts b/apps/desktop/src/app/DesktopEnvironment.ts index 3887513b31d..c991f5b39d6 100644 --- a/apps/desktop/src/app/DesktopEnvironment.ts +++ b/apps/desktop/src/app/DesktopEnvironment.ts @@ -158,10 +158,10 @@ const make = Effect.fn("desktop.environment.make")(function* ( const displayName = branding.displayName; const stateDir = path.join( baseDir, - isDevelopment && Option.isNone(configuredBaseDir) ? "dev" : "userdata-v2", + isDevelopment && Option.isNone(configuredBaseDir) ? "dev" : "userdata", ); - const userDataDirName = isDevelopment ? "t3code-dev" : "t3code-v2"; - const legacyUserDataDirName = isDevelopment ? "T3 Code (Dev)" : "T3 Code (Alpha v2)"; + const userDataDirName = isDevelopment ? "t3code-dev" : "t3code"; + const legacyUserDataDirName = isDevelopment ? "T3 Code (Dev)" : "T3 Code (Alpha)"; const resourcesPath = input.resourcesPath; return DesktopEnvironment.of({ diff --git a/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts b/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts index 60df9db7977..24c90943ec3 100644 --- a/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts +++ b/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts @@ -358,7 +358,7 @@ describe("DesktopBackendConfiguration", () => { const baseDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-desktop-backend-config-test-", }); - const settingsPath = path.join(baseDir, "userdata-v2", "settings.json"); + const settingsPath = path.join(baseDir, "userdata", "settings.json"); const cause = PlatformError.systemError({ _tag: "PermissionDenied", module: "FileSystem", diff --git a/apps/desktop/src/backend/DesktopBackendConfiguration.ts b/apps/desktop/src/backend/DesktopBackendConfiguration.ts index 9dc196ef69d..7ae3ad6c912 100644 --- a/apps/desktop/src/backend/DesktopBackendConfiguration.ts +++ b/apps/desktop/src/backend/DesktopBackendConfiguration.ts @@ -503,7 +503,7 @@ const resolveWslStartConfig = Effect.fn("desktop.backendConfiguration.resolveWsl }; // Forward the dev-server URL as an explicit CLI flag so the WSL backend's - // config resolution lands in dev/ instead of userdata-v2/. Inheriting through + // config resolution lands in dev/ instead of userdata/. Inheriting through // WSLENV is unreliable in practice (URL-shaped values with colons / // slashes get translated unpredictably depending on flags), and the // packaged build leaves devServerUrl as None anyway. diff --git a/apps/desktop/src/settings/DesktopClientSettings.diagnostics.test.ts b/apps/desktop/src/settings/DesktopClientSettings.diagnostics.test.ts index db68a51b2d4..5034df44cf7 100644 --- a/apps/desktop/src/settings/DesktopClientSettings.diagnostics.test.ts +++ b/apps/desktop/src/settings/DesktopClientSettings.diagnostics.test.ts @@ -83,7 +83,7 @@ describe("DesktopClientSettings diagnostics", () => { _tag: "PermissionDenied", module: "FileSystem", method: "readFileString", - pathOrDescriptor: `${baseDir}/userdata-v2/client-settings.json`, + pathOrDescriptor: `${baseDir}/userdata/client-settings.json`, }); return Effect.gen(function* () { diff --git a/apps/desktop/src/settings/DesktopSavedEnvironments.test.ts b/apps/desktop/src/settings/DesktopSavedEnvironments.test.ts index 08cd158739c..ec70308b3d3 100644 --- a/apps/desktop/src/settings/DesktopSavedEnvironments.test.ts +++ b/apps/desktop/src/settings/DesktopSavedEnvironments.test.ts @@ -403,7 +403,7 @@ describe("DesktopSavedEnvironments", () => { const baseDir = yield* baseFileSystem.makeTempDirectoryScoped({ prefix: "t3-desktop-saved-environments-test-", }); - const registryPath = `${baseDir}/userdata-v2/saved-environments.json`; + const registryPath = `${baseDir}/userdata/saved-environments.json`; const permissionError = PlatformError.systemError({ _tag: "PermissionDenied", module: "FileSystem", @@ -439,7 +439,7 @@ describe("DesktopSavedEnvironments", () => { _tag: "PermissionDenied", module: "FileSystem", method: "makeDirectory", - pathOrDescriptor: `${baseDir}/userdata-v2`, + pathOrDescriptor: `${baseDir}/userdata`, }); const fileSystemLayer = Layer.succeed( FileSystem.FileSystem, @@ -455,11 +455,11 @@ describe("DesktopSavedEnvironments", () => { const error = yield* savedEnvironments.setRegistry([savedRegistryRecord]).pipe(Effect.flip); assert.instanceOf(error, DesktopSavedEnvironments.DesktopSavedEnvironmentsWriteError); assert.equal(error.operation, "create-directory"); - assert.equal(error.path, `${baseDir}/userdata-v2`); + assert.equal(error.path, `${baseDir}/userdata`); assert.strictEqual(error.cause, permissionError); assert.equal( error.message, - `Desktop saved-environment write failed during create-directory at ${baseDir}/userdata-v2.`, + `Desktop saved-environment write failed during create-directory at ${baseDir}/userdata.`, ); assert.notEqual(error.message, permissionError.message); }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), diff --git a/apps/server/src/cli/config.test.ts b/apps/server/src/cli/config.test.ts index 754bdd8a16a..f6c2a63e192 100644 --- a/apps/server/src/cli/config.test.ts +++ b/apps/server/src/cli/config.test.ts @@ -125,7 +125,7 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => { tailscaleServeEnabled: false, tailscaleServePort: 443, }); - assert.equal(resolved.stateDir, join(baseDir, "userdata-v2")); + assert.equal(resolved.stateDir, join(baseDir, "userdata")); }), ); @@ -195,7 +195,7 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => { tailscaleServeEnabled: true, tailscaleServePort: 8443, }); - assert.equal(resolved.dbPath, join(baseDir, "userdata-v2", "state.sqlite")); + assert.equal(resolved.dbPath, join(baseDir, "userdata", "state.sqlite")); }), ); @@ -342,7 +342,7 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => { tailscaleServeEnabled: false, tailscaleServePort: 443, }); - assert.equal(join(baseDir, "userdata-v2"), resolved.stateDir); + assert.equal(join(baseDir, "userdata"), resolved.stateDir); }), ); diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index 9dbb390bb27..3b081c95c34 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -100,7 +100,7 @@ export const deriveServerPaths = Effect.fn(function* ( const { join } = yield* Path.Path; const stateDir = join( baseDir, - devUrl !== undefined && !options.baseDirIsExplicit ? "dev" : "userdata-v2", + devUrl !== undefined && !options.baseDirIsExplicit ? "dev" : "userdata", ); const dbPath = join(stateDir, "state.sqlite"); const attachmentsDir = join(stateDir, "attachments"); diff --git a/apps/server/src/orchestration-v2/ContextHandoffService.test.ts b/apps/server/src/orchestration-v2/ContextHandoffService.test.ts new file mode 100644 index 00000000000..3031d0346cb --- /dev/null +++ b/apps/server/src/orchestration-v2/ContextHandoffService.test.ts @@ -0,0 +1,107 @@ +import { assert, it } from "@effect/vitest"; +import { + MessageId, + ProviderInstanceId, + ProviderThreadId, + RunId, + ThreadId, + TurnItemId, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { + ContextHandoffServiceV2, + layer as contextHandoffServiceLayer, + providerMessageWithContextHandoff, +} from "./ContextHandoffService.ts"; +import { layer as idAllocatorLayer } from "./IdAllocator.ts"; + +const TestLayer = contextHandoffServiceLayer.pipe(Layer.provide(idAllocatorLayer)); + +function importedItem( + input: + | { + readonly role: "user"; + readonly id: string; + readonly text: string; + readonly ordinal: number; + } + | { + readonly role: "assistant"; + readonly id: string; + readonly text: string; + readonly ordinal: number; + }, +): OrchestrationV2TurnItem { + const now = DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"); + const base = { + id: TurnItemId.make(`turn-item:${input.id}`), + threadId: ThreadId.make("thread:legacy-context"), + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: input.ordinal, + status: "completed" as const, + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + messageId: MessageId.make(`message:${input.id}`), + text: input.text, + }; + return input.role === "user" + ? { + ...base, + createdBy: "user", + creationSource: "server", + type: "user_message", + inputIntent: "turn_start", + attachments: [], + } + : { + ...base, + type: "assistant_message", + streaming: false, + }; +} + +it.layer(TestLayer)("ContextHandoffService legacy import", (it) => { + it.effect("prepares imported history for the first native v2 turn", () => + Effect.gen(function* () { + const service = yield* ContextHandoffServiceV2; + const handoff = yield* service.prepareLegacyImport({ + threadId: ThreadId.make("thread:legacy-context"), + targetRunId: RunId.make("run:first-v2"), + toProviderThreadId: ProviderThreadId.make("provider-thread:first-v2"), + toProviderInstanceId: ProviderInstanceId.make("codex"), + items: [ + importedItem({ role: "user", id: "one", text: "What did we decide?", ordinal: 1 }), + importedItem({ + role: "assistant", + id: "two", + text: "We decided to keep the migration lightweight.", + ordinal: 2, + }), + ], + createdAt: DateTime.makeUnsafe("2026-01-02T00:00:00.000Z"), + }); + + assert.equal(handoff.strategy, "manual_context"); + assert.deepStrictEqual(handoff.fromProviderThreadIds, []); + assert.include(handoff.summaryText, "What did we decide?"); + assert.include(handoff.summaryText, "keep the migration lightweight"); + const providerMessage = providerMessageWithContextHandoff({ + handoff, + userText: "Continue from there.", + }); + assert.include(providerMessage, handoff.summaryText); + assert.include(providerMessage, "User message:\nContinue from there."); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/ContextHandoffService.ts b/apps/server/src/orchestration-v2/ContextHandoffService.ts index 2f644fcf79d..9dd8b343d03 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffService.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffService.ts @@ -33,6 +33,14 @@ export const ContextHandoffServiceV2Error = Schema.Union([ContextHandoffPrepareE export type ContextHandoffServiceV2Error = typeof ContextHandoffServiceV2Error.Type; export interface ContextHandoffServiceV2Shape { + readonly prepareLegacyImport: (input: { + readonly threadId: ThreadId; + readonly targetRunId: RunId; + readonly toProviderThreadId: ProviderThreadId; + readonly toProviderInstanceId: ProviderInstanceId; + readonly items: ReadonlyArray; + readonly createdAt: DateTime.Utc; + }) => Effect.Effect; readonly prepare: (input: { readonly threadId: ThreadId; readonly targetRunId: RunId; @@ -153,6 +161,32 @@ function makeProviderHandoffSummary(input: { ].join("\n"); } +function makeLegacyImportSummary(items: ReadonlyArray): string { + const sections = items.flatMap((item) => { + switch (item.type) { + case "user_message": + return [`User:\n${item.text}`]; + case "assistant_message": + return [`Assistant:\n${item.text}`]; + default: + return []; + } + }); + const header = + "Imported conversation history from the previous T3 Code orchestrator. Use it as context; do not repeat it unless the user asks."; + const maxChars = 32_000; + const selected: Array = []; + let remaining = maxChars - header.length - 2; + for (const section of sections.toReversed()) { + if (remaining <= 0) break; + const suffix = + section.length <= remaining ? section : section.slice(section.length - remaining); + selected.unshift(suffix); + remaining -= suffix.length + 2; + } + return [header, ...selected].join("\n\n"); +} + export function providerMessageWithContextHandoff(input: { readonly handoff: OrchestrationV2ContextHandoff; readonly userText: string; @@ -181,6 +215,52 @@ const makeContextHandoffService = Effect.fn("orchestrationV2.ContextHandoffServi function* () { const idAllocator = yield* IdAllocatorV2; + const prepareLegacyImport = Effect.fn("orchestrationV2.contextHandoff.prepareLegacyImport")( + function* (input: { + readonly threadId: ThreadId; + readonly targetRunId: RunId; + readonly toProviderThreadId: ProviderThreadId; + readonly toProviderInstanceId: ProviderInstanceId; + readonly items: ReadonlyArray; + readonly createdAt: DateTime.Utc; + }) { + const handoffId = yield* idAllocator.allocate + .contextHandoff({ + threadId: input.threadId, + fromProviderInstanceId: ProviderInstanceId.make("legacy"), + toProviderInstanceId: input.toProviderInstanceId, + }) + .pipe( + Effect.mapError( + (cause) => + new ContextHandoffPrepareError({ + threadId: input.threadId, + targetRunId: input.targetRunId, + fromProviderThreadIds: [], + toProviderThreadId: input.toProviderThreadId, + cause, + }), + ), + ); + return { + id: handoffId, + transferId: null, + threadId: input.threadId, + targetRunId: input.targetRunId, + fromProviderThreadIds: [], + toProviderThreadId: input.toProviderThreadId, + coveredRunOrdinals: { from: 1, to: 1 }, + strategy: "manual_context", + status: "ready", + summaryMessageId: null, + summaryText: makeLegacyImportSummary(input.items), + createdByProviderInstanceId: null, + createdAt: input.createdAt, + updatedAt: input.createdAt, + } satisfies OrchestrationV2ContextHandoff; + }, + ); + const prepare = Effect.fn("orchestrationV2.contextHandoff.prepare")(function* (input: { readonly threadId: ThreadId; readonly targetRunId: RunId; @@ -330,6 +410,7 @@ const makeContextHandoffService = Effect.fn("orchestrationV2.ContextHandoffServi }); return ContextHandoffServiceV2.of({ + prepareLegacyImport, prepare, prepareForkDelta, prepareProviderHandoff, diff --git a/apps/server/src/orchestration-v2/LegacyV1ThreadImporter.test.ts b/apps/server/src/orchestration-v2/LegacyV1ThreadImporter.test.ts new file mode 100644 index 00000000000..20590ac2ce7 --- /dev/null +++ b/apps/server/src/orchestration-v2/LegacyV1ThreadImporter.test.ts @@ -0,0 +1,238 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import { layer as eventSinkLayer } from "./EventSink.ts"; +import { layer as eventStoreLayer } from "./EventStore.ts"; +import { + LegacyV1ThreadImporter, + layer as legacyV1ThreadImporterLayer, +} from "./LegacyV1ThreadImporter.ts"; +import { + ProjectionMaintenanceV2, + layer as projectionMaintenanceLayer, +} from "./ProjectionMaintenance.ts"; +import { ProjectionStoreV2, layer as projectionStoreLayer } from "./ProjectionStore.ts"; + +const databaseLayer = SqlitePersistenceMemory; +const eventStoreProvided = eventStoreLayer.pipe(Layer.provideMerge(databaseLayer)); +const projectionStoreProvided = projectionStoreLayer.pipe(Layer.provideMerge(databaseLayer)); +const storesProvided = Layer.mergeAll(databaseLayer, eventStoreProvided, projectionStoreProvided); +const eventSinkProvided = eventSinkLayer.pipe(Layer.provide(storesProvided)); +const importerProvided = legacyV1ThreadImporterLayer.pipe( + Layer.provide(Layer.mergeAll(storesProvided, eventSinkProvided)), +); +const projectionMaintenanceProvided = projectionMaintenanceLayer.pipe( + Layer.provide(storesProvided), +); +const TestLayer = Layer.mergeAll( + storesProvided, + eventSinkProvided, + importerProvided, + projectionMaintenanceProvided, +); + +it.layer(TestLayer)("LegacyV1ThreadImporter", (it) => { + it.effect("imports lightweight shells, hydrates transcripts, and remains idempotent", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const importer = yield* LegacyV1ThreadImporter; + const maintenance = yield* ProjectionMaintenanceV2; + const projections = yield* ProjectionStoreV2; + const threadId = ThreadId.make("thread:legacy-import"); + + yield* sql` + INSERT INTO projection_projects ( + project_id, + title, + workspace_root, + default_model_selection_json, + scripts_json, + created_at, + updated_at, + deleted_at + ) VALUES ( + 'project:legacy-import', + 'Legacy project', + '/tmp/legacy-project', + '{"instanceId":"codex","model":"gpt-5.4"}', + '[]', + '2026-01-01T00:00:00.000Z', + '2026-01-04T00:00:00.000Z', + NULL + ) + `; + yield* sql` + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + latest_turn_id, + created_at, + updated_at, + archived_at, + settled_override, + settled_at, + deleted_at + ) VALUES ( + ${threadId}, + 'project:legacy-import', + 'Migrated conversation', + '{"instanceId":"codex","model":"gpt-5.4"}', + 'full-access', + 'default', + 'main', + '/tmp/legacy-project', + NULL, + '2026-01-01T00:00:00.000Z', + '2026-01-04T00:00:00.000Z', + NULL, + NULL, + NULL, + NULL + ) + `; + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, + thread_id, + turn_id, + role, + text, + attachments_json, + is_streaming, + created_at, + updated_at + ) VALUES + ( + 'message:legacy:1', + ${threadId}, + NULL, + 'user', + 'First question', + '[]', + 0, + '2026-01-01T01:00:00.000Z', + '2026-01-01T01:00:00.000Z' + ), + ( + 'message:legacy:2', + ${threadId}, + NULL, + 'assistant', + 'First answer', + '[]', + 0, + '2026-01-02T01:00:00.000Z', + '2026-01-02T01:00:00.000Z' + ), + ( + 'message:legacy:3', + ${threadId}, + NULL, + 'user', + 'Follow-up question', + '[]', + 0, + '2026-01-03T01:00:00.000Z', + '2026-01-03T01:00:00.000Z' + ), + ( + 'message:legacy:4', + ${threadId}, + NULL, + 'assistant', + 'Partial answer', + '[]', + 1, + '2026-01-04T01:00:00.000Z', + '2026-01-04T01:00:00.000Z' + ) + `; + + const shellImport = yield* importer.reconcileShells; + assert.deepStrictEqual(shellImport, { + importedThreadCount: 1, + importedMessageCount: 2, + }); + const shellEventCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM orchestration_events + WHERE application_event_version = 2 + AND aggregate_kind = 'thread' + AND stream_id = ${threadId} + `; + assert.equal(shellEventCount[0]?.count, 6); + + const rebuilt = yield* maintenance.rebuild; + assert.isTrue(rebuilt.valid); + const shellProjection = yield* projections.getThreadProjection(threadId); + assert.equal(shellProjection.thread.historyOrigin, "v1_import"); + assert.deepStrictEqual( + shellProjection.messages.map((message) => message.id), + ["message:legacy:3", "message:legacy:4"], + ); + + const transcriptImport = yield* importer.ensureTranscript(threadId); + assert.deepStrictEqual(transcriptImport, { + importedThreadCount: 1, + importedMessageCount: 2, + }); + const projection = yield* projections.getThreadProjection(threadId); + assert.deepStrictEqual( + projection.messages.map((message) => message.id), + ["message:legacy:1", "message:legacy:2", "message:legacy:3", "message:legacy:4"], + ); + assert.deepStrictEqual( + projection.turnItems + .filter( + ( + item, + ): item is Extract< + (typeof projection.turnItems)[number], + { readonly type: "user_message" | "assistant_message" } + > => item.type === "user_message" || item.type === "assistant_message", + ) + .map((item) => [item.messageId, item.ordinal, item.status]), + [ + ["message:legacy:1", 1, "completed"], + ["message:legacy:2", 2, "completed"], + ["message:legacy:3", 3, "completed"], + ["message:legacy:4", 4, "interrupted"], + ], + ); + + const eventCountBeforeRetry = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM orchestration_events + WHERE application_event_version = 2 + AND aggregate_kind = 'thread' + AND stream_id = ${threadId} + `; + assert.deepStrictEqual(yield* importer.reconcileShells, { + importedThreadCount: 0, + importedMessageCount: 0, + }); + assert.deepStrictEqual(yield* importer.ensureTranscript(threadId), { + importedThreadCount: 0, + importedMessageCount: 0, + }); + const eventCountAfterRetry = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM orchestration_events + WHERE application_event_version = 2 + AND aggregate_kind = 'thread' + AND stream_id = ${threadId} + `; + assert.equal(eventCountAfterRetry[0]?.count, eventCountBeforeRetry[0]?.count); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/LegacyV1ThreadImporter.ts b/apps/server/src/orchestration-v2/LegacyV1ThreadImporter.ts new file mode 100644 index 00000000000..379d2c3cb15 --- /dev/null +++ b/apps/server/src/orchestration-v2/LegacyV1ThreadImporter.ts @@ -0,0 +1,639 @@ +import { + ChatAttachment, + DEFAULT_MODEL, + EventId, + MessageId, + ModelSelection, + type OrchestrationV2AppThread, + type OrchestrationV2ConversationMessage, + type OrchestrationV2DomainEvent, + type OrchestrationV2TurnItem, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { EventSinkV2 } from "./EventSink.ts"; +import { EventStoreV2 } from "./EventStore.ts"; +import { makeKeyedSerialExecutor } from "./KeyedSerialExecutor.ts"; + +const IMPORT_EVENT_PREFIX = "migration:v1"; +const TRANSCRIPT_EVENT_BATCH_SIZE = 100; + +interface LegacyThreadRow { + readonly thread_id: string; + readonly project_id: string; + readonly title: string; + readonly model_selection_json: string | null; + readonly runtime_mode: string; + readonly interaction_mode: string; + readonly branch: string | null; + readonly worktree_path: string | null; + readonly created_at: string; + readonly updated_at: string; + readonly archived_at: string | null; + readonly settled_override: string | null; + readonly settled_at: string | null; + readonly deleted_at: string | null; +} + +interface LegacyMessageRow { + readonly message_id: string; + readonly thread_id: string; + readonly role: "user" | "assistant"; + readonly text: string; + readonly attachments_json: string | null; + readonly is_streaming: number; + readonly created_at: string; + readonly updated_at: string; + readonly ordinal: number; +} + +interface LegacyImportRow { + readonly thread_id: string; + readonly transcript_imported_at: string | null; +} + +export interface LegacyV1ImportSummary { + readonly importedThreadCount: number; + readonly importedMessageCount: number; +} + +export class LegacyV1ThreadImportError extends Schema.TaggedErrorClass()( + "LegacyV1ThreadImportError", + { + operation: Schema.String, + threadId: Schema.optional(ThreadId), + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + return this.threadId === undefined + ? `Failed to ${this.operation} legacy v1 threads.` + : `Failed to ${this.operation} legacy v1 thread ${this.threadId}.`; + } +} + +export interface LegacyV1ThreadImporterShape { + readonly reconcileShells: Effect.Effect; + readonly ensureTranscript: ( + threadId: ThreadId, + ) => Effect.Effect; + readonly importPendingTranscripts: Effect.Effect; +} + +export class LegacyV1ThreadImporter extends Context.Service< + LegacyV1ThreadImporter, + LegacyV1ThreadImporterShape +>()("t3/orchestration-v2/LegacyV1ThreadImporter") {} + +const decodeModelSelection = Schema.decodeUnknownOption(ModelSelection); +const decodeAttachments = Schema.decodeUnknownOption(Schema.Array(ChatAttachment)); + +function parseJson(json: string): unknown { + try { + return JSON.parse(json); + } catch { + return undefined; + } +} + +function modelSelectionFor(row: LegacyThreadRow) { + const decoded = + row.model_selection_json === null + ? Option.none() + : decodeModelSelection(parseJson(row.model_selection_json)); + return Option.getOrElse(decoded, () => ({ + instanceId: ProviderInstanceId.make("codex"), + model: DEFAULT_MODEL, + })); +} + +function attachmentsFor(row: LegacyMessageRow) { + if (row.attachments_json === null) return []; + return Option.getOrElse(decodeAttachments(parseJson(row.attachments_json)), () => []); +} + +function runtimeModeFor(value: string): OrchestrationV2AppThread["runtimeMode"] { + return value === "approval-required" || + value === "auto-accept-edits" || + value === "auto" || + value === "full-access" + ? value + : "full-access"; +} + +function interactionModeFor(value: string): OrchestrationV2AppThread["interactionMode"] { + return value === "plan" ? "plan" : "default"; +} + +function settledOverrideFor(value: string | null): OrchestrationV2AppThread["settledOverride"] { + return value === "settled" || value === "active" ? value : null; +} + +function dateTime(value: string): DateTime.Utc { + return DateTime.makeUnsafe(value); +} + +function nullableDateTime(value: string | null): DateTime.Utc | null { + return value === null ? null : dateTime(value); +} + +function importedThread(row: LegacyThreadRow): OrchestrationV2AppThread { + const threadId = ThreadId.make(row.thread_id); + const modelSelection = modelSelectionFor(row); + return { + createdBy: "system", + creationSource: "server", + id: threadId, + projectId: ProjectId.make(row.project_id), + title: row.title.trim() === "" ? "Untitled thread" : row.title, + providerInstanceId: modelSelection.instanceId, + modelSelection, + runtimeMode: runtimeModeFor(row.runtime_mode), + interactionMode: interactionModeFor(row.interaction_mode), + branch: row.branch?.trim() ? row.branch : null, + worktreePath: row.worktree_path?.trim() ? row.worktree_path : null, + activeProviderThreadId: null, + historyOrigin: "v1_import", + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: dateTime(row.created_at), + updatedAt: dateTime(row.updated_at), + archivedAt: nullableDateTime(row.archived_at), + settledOverride: settledOverrideFor(row.settled_override), + settledAt: nullableDateTime(row.settled_at), + deletedAt: nullableDateTime(row.deleted_at), + }; +} + +function messageEvents(row: LegacyMessageRow): ReadonlyArray { + const threadId = ThreadId.make(row.thread_id); + const messageId = MessageId.make(row.message_id); + const createdAt = dateTime(row.created_at); + const updatedAt = dateTime(row.updated_at); + const attachments = attachmentsFor(row); + const message: OrchestrationV2ConversationMessage = { + createdBy: row.role === "user" ? "user" : "agent", + creationSource: "server", + id: messageId, + threadId, + runId: null, + nodeId: null, + role: row.role, + text: row.text, + attachments, + streaming: false, + createdAt, + updatedAt, + }; + const baseTurnItem = { + id: TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${row.message_id}`), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: row.ordinal, + status: row.is_streaming === 1 ? ("interrupted" as const) : ("completed" as const), + title: null, + startedAt: createdAt, + completedAt: updatedAt, + updatedAt, + }; + const turnItem: OrchestrationV2TurnItem = + row.role === "user" + ? { + ...baseTurnItem, + createdBy: "user", + creationSource: "server", + type: "user_message", + messageId, + inputIntent: "turn_start", + text: row.text, + attachments, + } + : { + ...baseTurnItem, + type: "assistant_message", + messageId, + text: row.text, + streaming: false, + }; + return [ + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:message:${row.message_id}`), + type: "message.updated", + threadId, + occurredAt: updatedAt, + payload: message, + }, + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${row.message_id}`), + type: "turn-item.updated", + threadId, + occurredAt: updatedAt, + payload: turnItem, + }, + ]; +} + +function chunks(items: ReadonlyArray, size: number): Array> { + const result: Array> = []; + for (let index = 0; index < items.length; index += size) { + result.push(items.slice(index, index + size)); + } + return result; +} + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const eventStore = yield* EventStoreV2; + const eventSink = yield* EventSinkV2; + const transcriptImports = yield* makeKeyedSerialExecutor(); + + const listMessages = (threadId: ThreadId) => + sql` + SELECT + message_id, + thread_id, + role, + text, + attachments_json, + is_streaming, + created_at, + updated_at, + ROW_NUMBER() OVER ( + PARTITION BY thread_id + ORDER BY created_at ASC, message_id ASC + ) AS ordinal + FROM projection_thread_messages + WHERE thread_id = ${threadId} + AND role IN ('user', 'assistant') + ORDER BY created_at ASC, message_id ASC + `; + + const listShellMessages = (threadId: ThreadId) => + Effect.gen(function* () { + const latest = yield* sql` + SELECT + message.message_id, + message.thread_id, + message.role, + message.text, + message.attachments_json, + message.is_streaming, + message.created_at, + message.updated_at, + ( + SELECT COUNT(*) + FROM projection_thread_messages AS earlier + WHERE earlier.thread_id = message.thread_id + AND earlier.role IN ('user', 'assistant') + AND ( + earlier.created_at < message.created_at + OR ( + earlier.created_at = message.created_at + AND earlier.message_id <= message.message_id + ) + ) + ) AS ordinal + FROM projection_thread_messages AS message + WHERE message.thread_id = ${threadId} + AND message.role IN ('user', 'assistant') + ORDER BY message.created_at DESC, message.message_id DESC + LIMIT 1 + `; + const latestUser = yield* sql` + SELECT + message.message_id, + message.thread_id, + message.role, + message.text, + message.attachments_json, + message.is_streaming, + message.created_at, + message.updated_at, + ( + SELECT COUNT(*) + FROM projection_thread_messages AS earlier + WHERE earlier.thread_id = message.thread_id + AND earlier.role IN ('user', 'assistant') + AND ( + earlier.created_at < message.created_at + OR ( + earlier.created_at = message.created_at + AND earlier.message_id <= message.message_id + ) + ) + ) AS ordinal + FROM projection_thread_messages AS message + WHERE message.thread_id = ${threadId} + AND message.role = 'user' + ORDER BY message.created_at DESC, message.message_id DESC + LIMIT 1 + `; + return [latestUser[0], latest[0]].filter( + (message, index, selected): message is LegacyMessageRow => + message !== undefined && + selected.findIndex((candidate) => candidate?.message_id === message.message_id) === index, + ); + }); + + const reconcileShellsBase = Effect.gen(function* () { + const now = DateTime.formatIso(yield* DateTime.now); + const rows = yield* sql` + SELECT + thread.thread_id, + thread.project_id, + thread.title, + thread.model_selection_json, + thread.runtime_mode, + thread.interaction_mode, + thread.branch, + thread.worktree_path, + thread.created_at, + thread.updated_at, + thread.archived_at, + thread.settled_override, + thread.settled_at, + thread.deleted_at + FROM projection_threads AS thread + WHERE NOT EXISTS ( + SELECT 1 + FROM orchestration_events AS event + WHERE event.application_event_version = 2 + AND event.aggregate_kind = 'thread' + AND event.stream_id = thread.thread_id + AND event.event_type = 'thread.created' + ) + ORDER BY thread.created_at ASC, thread.thread_id ASC + `; + let importedThreadCount = 0; + let importedMessageCount = 0; + for (const row of rows) { + const thread = importedThread(row); + const previews = yield* listShellMessages(thread.id); + const events: Array = [ + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:created`), + type: "thread.created", + threadId: thread.id, + providerInstanceId: thread.providerInstanceId, + occurredAt: thread.createdAt, + payload: thread, + }, + ...previews.flatMap(messageEvents), + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:shell`), + type: "thread.metadata-updated", + threadId: thread.id, + providerInstanceId: thread.providerInstanceId, + occurredAt: thread.updatedAt, + payload: thread, + }, + ]; + yield* sql.withTransaction( + Effect.gen(function* () { + yield* eventStore.append({ events }); + yield* Effect.forEach( + previews, + (message) => + sql` + INSERT INTO orchestration_v2_turn_item_positions ( + thread_id, + turn_item_id, + ordinal + ) + VALUES ( + ${thread.id}, + ${TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${message.message_id}`)}, + ${message.ordinal} + ) + ON CONFLICT(thread_id, turn_item_id) DO NOTHING + `, + { discard: true }, + ); + yield* sql` + INSERT INTO orchestration_v2_legacy_imports ( + thread_id, + source_updated_at, + shell_imported_at, + transcript_imported_at, + imported_message_count, + last_error + ) + VALUES ( + ${thread.id}, + ${row.updated_at}, + ${now}, + NULL, + ${previews.length}, + NULL + ) + ON CONFLICT(thread_id) DO NOTHING + `; + }), + ); + importedThreadCount += 1; + importedMessageCount += previews.length; + } + return { importedThreadCount, importedMessageCount }; + }); + + const reconcileShells = reconcileShellsBase.pipe( + Effect.mapError((cause) => new LegacyV1ThreadImportError({ operation: "import", cause })), + ); + + const ensureTranscriptBase = (threadId: ThreadId) => + transcriptImports.withLock( + threadId, + Effect.gen(function* () { + const imports = yield* sql` + SELECT thread_id, transcript_imported_at + FROM orchestration_v2_legacy_imports + WHERE thread_id = ${threadId} + LIMIT 1 + `; + const imported = imports[0]; + if (imported === undefined || imported.transcript_imported_at !== null) { + return { importedThreadCount: 0, importedMessageCount: 0 }; + } + const messages = yield* listMessages(threadId); + const existingRows = yield* sql<{ readonly event_id: string }>` + SELECT event_id + FROM orchestration_events + WHERE application_event_version = 2 + AND aggregate_kind = 'thread' + AND stream_id = ${threadId} + AND event_id LIKE ${`${IMPORT_EVENT_PREFIX}:message:%`} + `; + const existing = new Set(existingRows.map((row) => row.event_id)); + const missing = messages.filter( + (message) => !existing.has(`${IMPORT_EVENT_PREFIX}:message:${message.message_id}`), + ); + for (const batch of chunks(missing, TRANSCRIPT_EVENT_BATCH_SIZE / 2)) { + yield* Effect.forEach( + batch, + (message) => + sql` + INSERT INTO orchestration_v2_turn_item_positions ( + thread_id, + turn_item_id, + ordinal + ) + VALUES ( + ${threadId}, + ${TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${message.message_id}`)}, + ${message.ordinal} + ) + ON CONFLICT(thread_id, turn_item_id) DO NOTHING + `, + { discard: true }, + ); + yield* eventSink.write({ events: batch.flatMap(messageEvents) }); + yield* Effect.yieldNow; + } + const threadRows = yield* sql` + SELECT + thread_id, + project_id, + title, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + created_at, + updated_at, + archived_at, + settled_override, + settled_at, + deleted_at + FROM projection_threads + WHERE thread_id = ${threadId} + LIMIT 1 + `; + const source = threadRows[0]; + if (source !== undefined) { + const marker = `${IMPORT_EVENT_PREFIX}:thread:${threadId}:transcript`; + const markerRows = yield* sql<{ readonly event_id: string }>` + SELECT event_id + FROM orchestration_events + WHERE event_id = ${marker} + LIMIT 1 + `; + if (markerRows.length === 0) { + const thread = importedThread(source); + yield* eventSink.write({ + events: [ + { + id: EventId.make(marker), + type: "thread.metadata-updated", + threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: thread.updatedAt, + payload: thread, + }, + ], + }); + } + } + const now = DateTime.formatIso(yield* DateTime.now); + yield* sql` + UPDATE orchestration_v2_legacy_imports + SET + transcript_imported_at = ${now}, + imported_message_count = ${messages.length}, + last_error = NULL + WHERE thread_id = ${threadId} + `; + return { + importedThreadCount: 1, + importedMessageCount: missing.length, + }; + }), + ); + + const ensureTranscript = (threadId: ThreadId) => + ensureTranscriptBase(threadId).pipe( + Effect.mapError( + (cause) => + new LegacyV1ThreadImportError({ + operation: "hydrate transcript for", + threadId, + cause, + }), + ), + ); + + const importPendingTranscripts = Effect.gen(function* () { + const rows = yield* sql` + SELECT thread_id, transcript_imported_at + FROM orchestration_v2_legacy_imports + WHERE transcript_imported_at IS NULL + ORDER BY shell_imported_at ASC, thread_id ASC + `; + let importedThreadCount = 0; + let importedMessageCount = 0; + for (const row of rows) { + const result = yield* ensureTranscript(ThreadId.make(row.thread_id)).pipe( + Effect.tapError((error) => + Effect.logWarning("Failed to hydrate migrated v1 thread transcript", { + threadId: row.thread_id, + cause: error, + }), + ), + Effect.catch(() => + sql` + UPDATE orchestration_v2_legacy_imports + SET last_error = 'Transcript hydration failed; retry on next open.' + WHERE thread_id = ${row.thread_id} + `.pipe( + Effect.as({ importedThreadCount: 0, importedMessageCount: 0 }), + Effect.orElseSucceed(() => ({ + importedThreadCount: 0, + importedMessageCount: 0, + })), + ), + ), + ); + importedThreadCount += result.importedThreadCount; + importedMessageCount += result.importedMessageCount; + yield* Effect.yieldNow; + } + return { importedThreadCount, importedMessageCount }; + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Legacy v1 transcript background import stopped", { cause }).pipe( + Effect.as({ importedThreadCount: 0, importedMessageCount: 0 }), + ), + ), + ); + + return LegacyV1ThreadImporter.of({ + reconcileShells, + ensureTranscript, + importPendingTranscripts, + }); +}); + +export const layer: Layer.Layer< + LegacyV1ThreadImporter, + never, + EventSinkV2 | EventStoreV2 | SqlClient.SqlClient +> = Layer.effect(LegacyV1ThreadImporter, make); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3de91906479..f169a1e44b4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2505,6 +2505,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio driver: adapter.driver, nativeThreadId: `pending:${runId}`, }); + const legacyImportHandoff = + projection.thread.historyOrigin === "v1_import" && + projection.runs.length === 0 && + projection.turnItems.length > 0 + ? yield* contextHandoffService + .prepareLegacyImport({ + threadId: command.threadId, + targetRunId: runId, + toProviderThreadId: providerThreadId, + toProviderInstanceId: modelSelection.instanceId, + items: projection.turnItems, + createdAt: now, + }) + .pipe(mapDispatchError(command)) + : null; const providerThread: OrchestrationV2ProviderThread = activeProviderThread === undefined ? { @@ -2519,7 +2534,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio status: "not_loaded", firstRunOrdinal: ordinal, lastRunOrdinal: ordinal, - handoffIds: [], + handoffIds: legacyImportHandoff === null ? [] : [legacyImportHandoff.id], forkedFrom: null, createdAt: now, updatedAt: now, @@ -2573,7 +2588,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio startedAt: null, completedAt: null, checkpointId: null, - contextHandoffId: null, + contextHandoffId: legacyImportHandoff?.id ?? null, ...(command.sourcePlanRef === undefined ? {} : { sourcePlanRef: command.sourcePlanRef }), }; const attempt: OrchestrationV2RunAttempt = { @@ -2676,6 +2691,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: providerThread, }); + if (legacyImportHandoff !== null) { + yield* emitEvent({ + type: "context-handoff.updated", + threadId: command.threadId, + runId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: legacyImportHandoff, + }); + } yield* emitEvent({ type: "run.created", threadId: command.threadId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index c6a446039e1..50c3aa0c54d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -768,6 +768,9 @@ export function threadShellFromProjection( lineage: projection.thread.lineage, forkedFrom: projection.thread.forkedFrom, activeProviderThreadId: projection.thread.activeProviderThreadId, + ...(projection.thread.historyOrigin === undefined + ? {} + : { historyOrigin: projection.thread.historyOrigin }), latestRunId: latestRun?.id ?? null, activeRunId: activeRun?.id ?? null, status: latestRun?.status ?? "idle", diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 7a4301980e5..ae6c9fde8c3 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -29,6 +29,7 @@ import { type OrchestratorV2DispatchResult, type OrchestratorV2Error, } from "./Orchestrator.ts"; +import { LegacyV1ThreadImporter } from "./LegacyV1ThreadImporter.ts"; export type ThreadManagementSendMode = "auto" | "queue" | "steer" | "restart"; @@ -132,6 +133,7 @@ export class ThreadManagementError extends Schema.TaggedErrorClass Effect.Effect; readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; @@ -235,9 +237,41 @@ function managementError( const make = Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; + const legacyImporter = yield* LegacyV1ThreadImporter; + + const ensureLegacyTranscript = (threadId: ThreadId) => + legacyImporter.ensureTranscript(threadId).pipe( + Effect.tapError((cause) => + Effect.logWarning("Unable to hydrate migrated v1 thread transcript", { + threadId, + cause, + }), + ), + Effect.ignore, + ); + + const getThreadProjection: ThreadManagementServiceShape["getThreadProjection"] = (threadId) => + ensureLegacyTranscript(threadId).pipe( + Effect.andThen(orchestrator.getThreadProjection(threadId)), + ); + + const getThreadSnapshot: ThreadManagementServiceShape["getThreadSnapshot"] = (threadId) => + ensureLegacyTranscript(threadId).pipe(Effect.andThen(orchestrator.getThreadSnapshot(threadId))); + + const dispatch: ThreadManagementServiceShape["dispatch"] = (command) => { + const threadId = + command.type === "message.dispatch" + ? command.threadId + : command.type === "thread.fork" || command.type === "thread.merge_back" + ? command.sourceThreadId + : undefined; + return threadId === undefined + ? orchestrator.dispatch(command) + : ensureLegacyTranscript(threadId).pipe(Effect.andThen(orchestrator.dispatch(command))); + }; const getProjectThread: ThreadManagementServiceShape["getProjectThread"] = (input) => - orchestrator.getThreadProjection(input.threadId).pipe( + getThreadProjection(input.threadId).pipe( Effect.mapError((cause) => managementError( "thread_not_found", @@ -450,9 +484,10 @@ const make = Effect.gen(function* () { }); return ThreadManagementService.of({ - dispatch: orchestrator.dispatch, - getThreadProjection: orchestrator.getThreadProjection, - getThreadSnapshot: orchestrator.getThreadSnapshot, + ensureLegacyTranscript, + dispatch, + getThreadProjection, + getThreadSnapshot, getProjectThread, getShellSnapshot: orchestrator.getShellSnapshot, listProjectThreads, @@ -466,7 +501,22 @@ const make = Effect.gen(function* () { }); }); +const legacyV1ThreadImporterNoopLayer = Layer.succeed( + LegacyV1ThreadImporter, + LegacyV1ThreadImporter.of({ + reconcileShells: Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }), + ensureTranscript: () => Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }), + importPendingTranscripts: Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }), + }), +); + export const layer: Layer.Layer = Layer.effect( ThreadManagementService, make, -); +).pipe(Layer.provide(legacyV1ThreadImporterNoopLayer)); + +export const layerWithLegacyImporter: Layer.Layer< + ThreadManagementService, + never, + LegacyV1ThreadImporter | OrchestratorV2 +> = Layer.effect(ThreadManagementService, make); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index c2a4a3fd5a9..ed5a3bdd883 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -20,6 +20,7 @@ import { import { layerFromStores as eventSinkLayer } from "./EventSink.ts"; import { layerFromOrchestrationEventStore as eventStoreLayer } from "./EventStore.ts"; import { layer as idAllocatorLayer } from "./IdAllocator.ts"; +import { layer as legacyV1ThreadImporterLayer } from "./LegacyV1ThreadImporter.ts"; import { layer as orchestratorLayer } from "./Orchestrator.ts"; import { layer as projectionStoreLayer } from "./ProjectionStore.ts"; import { layer as projectionMaintenanceLayer } from "./ProjectionMaintenance.ts"; @@ -36,7 +37,7 @@ import { layer as runExecutionServiceLayer } from "./RunExecutionService.ts"; import { layer as runFinalizationServiceLayer } from "./RunFinalizationService.ts"; import { layerFromProjectRepository as runtimePolicyLayerFromProjectRepository } from "./RuntimePolicy.ts"; import { layer as runtimeRequestServiceLayer } from "./RuntimeRequestService.ts"; -import { layer as threadManagementServiceLayer } from "./ThreadManagementService.ts"; +import { layerWithLegacyImporter as threadManagementServiceLayer } from "./ThreadManagementService.ts"; import { layer as threadLaunchServiceLayer } from "./ThreadLaunchService.ts"; import { layer as threadLifecycleServiceLayer } from "./ThreadLifecycleService.ts"; import { layer as threadForkServiceLayer } from "./ThreadForkService.ts"; @@ -68,6 +69,9 @@ const storesLayer = Layer.mergeAll( const eventSinkProvided = eventSinkLayer.pipe(Layer.provide(storesLayer)); const projectionMaintenanceProvided = projectionMaintenanceLayer.pipe(Layer.provide(storesLayer)); +const legacyV1ThreadImporterProvided = legacyV1ThreadImporterLayer.pipe( + Layer.provide(Layer.mergeAll(eventSinkProvided, eventStoreProvided)), +); const providerEventIngestorProvided = providerEventIngestorLayer.pipe( Layer.provide(Layer.mergeAll(eventSinkProvided, idAllocatorLayer)), @@ -201,7 +205,7 @@ const orchestratorProvided = orchestratorLayer.pipe( ); const threadManagementProvided = threadManagementServiceLayer.pipe( - Layer.provide(orchestratorProvided), + Layer.provide(Layer.merge(orchestratorProvided, legacyV1ThreadImporterProvided)), ); export const ProjectSetupScriptRunnerLayerLive = projectSetupScriptRunnerLayer.pipe( Layer.provide(ProjectServiceLayerLive), @@ -236,6 +240,7 @@ export const OrchestrationV2LayerLive = Layer.mergeAll( providerSessionManagerProvided, providerRuntimeRecoveryProvided, projectionMaintenanceProvided, + legacyV1ThreadImporterProvided, ); export const OrchestrationV2ProductionLayerLive = Layer.mergeAll( diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index a5af2690dc3..05ff25d81a2 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -55,6 +55,7 @@ import Migration0039 from "./Migrations/039_OrchestrationV2ThreadLaunchWorkflows import Migration0040 from "./Migrations/040_ApplicationEventSource.ts"; import Migration0041 from "./Migrations/041_OrchestrationV2EffectCancellation.ts"; import Migration0042 from "./Migrations/042_ScheduledTasks.ts"; +import Migration0043 from "./Migrations/043_LegacyV1ImportState.ts"; /** * Migration loader with all migrations defined inline. @@ -109,6 +110,7 @@ export const migrationEntries = [ [40, "ApplicationEventSource", Migration0040], [41, "OrchestrationV2EffectCancellation", Migration0041], [42, "ScheduledTasks", Migration0042], + [43, "LegacyV1ImportState", Migration0043], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts index 43a46eef551..5b217fb9c70 100644 --- a/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts @@ -13,7 +13,7 @@ layer("035_036_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 42 }, (_, index) => index + 1), + Array.from({ length: 43 }, (_, index) => index + 1), ); }), ); @@ -122,7 +122,7 @@ it.effect("upgrades a database already at released main migration 034", () => assert.ok(snoozeColumns.some((column) => column.name === "snoozed_until")); assert.ok(snoozeColumns.some((column) => column.name === "snoozed_at")); - yield* runMigrations({ toMigrationInclusive: 42 }); + yield* runMigrations({ toMigrationInclusive: 43 }); const migrations = yield* sql<{ readonly migration_id: number; @@ -130,7 +130,7 @@ it.effect("upgrades a database already at released main migration 034", () => }>` SELECT migration_id, name FROM effect_sql_migrations - WHERE migration_id BETWEEN 34 AND 42 + WHERE migration_id BETWEEN 34 AND 43 ORDER BY migration_id `; assert.deepStrictEqual( @@ -145,6 +145,7 @@ it.effect("upgrades a database already at released main migration 034", () => [40, "ApplicationEventSource"], [41, "OrchestrationV2EffectCancellation"], [42, "ScheduledTasks"], + [43, "LegacyV1ImportState"], ], ); @@ -154,5 +155,12 @@ it.effect("upgrades a database already at released main migration 034", () => WHERE type = 'table' AND name = 'orchestration_v2_projection_threads' `; assert.strictEqual(v2Tables.length, 1); + + const legacyImportTables = yield* sql<{ readonly name: string }>` + SELECT name + FROM sqlite_master + WHERE type = 'table' AND name = 'orchestration_v2_legacy_imports' + `; + assert.strictEqual(legacyImportTables.length, 1); }).pipe(Effect.provide(NodeSqliteClient.layerMemory())), ); diff --git a/apps/server/src/persistence/Migrations/043_LegacyV1ImportState.ts b/apps/server/src/persistence/Migrations/043_LegacyV1ImportState.ts new file mode 100644 index 00000000000..ebd107ce78c --- /dev/null +++ b/apps/server/src/persistence/Migrations/043_LegacyV1ImportState.ts @@ -0,0 +1,27 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +/** + * Tracks the incremental import of v1 materialized thread state into the v2 + * event model. Shells are imported synchronously at startup; full transcripts + * are hydrated on demand and by a low-priority background pass. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE orchestration_v2_legacy_imports ( + thread_id TEXT PRIMARY KEY, + source_updated_at TEXT NOT NULL, + shell_imported_at TEXT NOT NULL, + transcript_imported_at TEXT, + imported_message_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT + ) + `; + + yield* sql` + CREATE INDEX orchestration_v2_legacy_imports_pending_transcript_idx + ON orchestration_v2_legacy_imports(transcript_imported_at, shell_imported_at, thread_id) + `; +}); diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index a584cf8cfa6..101c451a5fc 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -19,6 +19,7 @@ it.effect("runs projection repair, recovery, worker startup, and bootstrap in or const record = (label: string) => Ref.update(calls, (current) => [...current, label]); const result = yield* ServerRuntimeStartup.runOrderedV2StartupPhases({ + importLegacyShells: record("import"), verify: record("verify").pipe(Effect.as({ valid: false })), rebuild: record("rebuild").pipe(Effect.as({ valid: true })), recover: record("recover").pipe(Effect.as({ closedRequests: 2 })), @@ -27,6 +28,7 @@ it.effect("runs projection repair, recovery, worker startup, and bootstrap in or }); assert.deepEqual(yield* Ref.get(calls), [ + "import", "verify", "rebuild", "recover", @@ -44,6 +46,7 @@ it.effect("does not rebuild valid projections", () => Effect.gen(function* () { const rebuilt = yield* Ref.make(false); yield* ServerRuntimeStartup.runOrderedV2StartupPhases({ + importLegacyShells: Effect.void, verify: Effect.succeed({ valid: true }), rebuild: Ref.set(rebuilt, true).pipe(Effect.as({ valid: true })), recover: Effect.void, diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index df9d039ffa7..0fff2df2105 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -26,6 +26,7 @@ import * as ServerConfig from "./config.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as EffectWorker from "./orchestration-v2/EffectWorker.ts"; +import * as LegacyV1ThreadImporter from "./orchestration-v2/LegacyV1ThreadImporter.ts"; import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; import * as ProviderRuntimeRecovery from "./orchestration-v2/ProviderRuntimeRecoveryService.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; @@ -286,21 +287,25 @@ const runStartupPhase = (phase: string, effect: Effect.Effect) ); export function runOrderedV2StartupPhases< + Import, Verification extends { readonly valid: boolean }, RebuildVerification extends { readonly valid: boolean }, Recovery, Bootstrap, + ImportError, VerifyError, RebuildError, RecoveryError, WorkerError, BootstrapError, + ImportContext, VerifyContext, RebuildContext, RecoveryContext, WorkerContext, BootstrapContext, >(input: { + readonly importLegacyShells: Effect.Effect; readonly verify: Effect.Effect; readonly rebuild: Effect.Effect; readonly recover: Effect.Effect; @@ -308,6 +313,7 @@ export function runOrderedV2StartupPhases< readonly autoBootstrap: Effect.Effect; }) { return Effect.gen(function* () { + yield* input.importLegacyShells; const verification = yield* input.verify; if (!verification.valid) { const rebuilt = yield* input.rebuild; @@ -328,6 +334,7 @@ export const make = Effect.gen(function* () { const serverConfig = yield* ServerConfig.ServerConfig; const keybindings = yield* Keybindings.Keybindings; const projectionMaintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const legacyV1ThreadImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; const providerRuntimeRecovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; @@ -402,6 +409,16 @@ export const make = Effect.gen(function* () { const welcomeBase = yield* resolveWelcomeBase; const environment = yield* serverEnvironment.getDescriptor; const { recovery, bootstrap: bootstrapTargets } = yield* runOrderedV2StartupPhases({ + importLegacyShells: runStartupPhase( + "orchestration-v2.legacy-v1.import-shells", + legacyV1ThreadImporter.reconcileShells.pipe( + Effect.tap((summary) => + summary.importedThreadCount === 0 + ? Effect.void + : Effect.logInfo("Imported legacy v1 thread shells", summary), + ), + ), + ), verify: runStartupPhase( "orchestration-v2.projections.verify", projectionMaintenance.verify.pipe( @@ -447,6 +464,14 @@ export const make = Effect.gen(function* () { yield* Effect.logDebug("Accepting commands"); yield* commandGate.signalCommandReady; + yield* legacyV1ThreadImporter.importPendingTranscripts.pipe( + Effect.tap((summary) => + summary.importedThreadCount === 0 + ? Effect.void + : Effect.logInfo("Hydrated legacy v1 thread transcripts", summary), + ), + Effect.forkScoped, + ); yield* Effect.logDebug("startup phase: publishing welcome event", { environmentId: environment.environmentId, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1d51a3b6b9d..ee8f8986621 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -637,6 +637,7 @@ const makeWsRpcLayer = ( yield* Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId, }); + yield* threadManagement.ensureLegacyTranscript(input.threadId); const eventStreamFrom = (afterSequence: number) => threadManagement diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 186e0b36fb0..2c2805e0742 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -57,6 +57,9 @@ export const OrchestrationV2CreationSource = Schema.Literals([ ]); export type OrchestrationV2CreationSource = typeof OrchestrationV2CreationSource.Type; +export const OrchestrationV2ThreadHistoryOrigin = Schema.Literals(["native", "v1_import"]); +export type OrchestrationV2ThreadHistoryOrigin = typeof OrchestrationV2ThreadHistoryOrigin.Type; + const OrchestrationV2CreationFields = { createdBy: OrchestrationV2Actor, creationSource: OrchestrationV2CreationSource, @@ -297,6 +300,7 @@ export const OrchestrationV2AppThread = Schema.Struct({ branch: Schema.NullOr(TrimmedNonEmptyString), worktreePath: Schema.NullOr(TrimmedNonEmptyString), activeProviderThreadId: Schema.NullOr(ProviderThreadId), + historyOrigin: Schema.optional(OrchestrationV2ThreadHistoryOrigin), lineage: OrchestrationV2AppThreadLineage, forkedFrom: Schema.NullOr( Schema.Union([ @@ -1164,6 +1168,7 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ lineage: OrchestrationV2AppThreadLineage, forkedFrom: Schema.NullOr(OrchestrationV2AppThread.fields.forkedFrom), activeProviderThreadId: Schema.NullOr(ProviderThreadId), + historyOrigin: Schema.optional(OrchestrationV2ThreadHistoryOrigin), latestRunId: Schema.NullOr(RunId), activeRunId: Schema.NullOr(RunId), status: OrchestrationV2ShellThreadStatus, diff --git a/packages/ssh/src/tunnel.test.ts b/packages/ssh/src/tunnel.test.ts index d32b5fe3e94..7e7a5a54276 100644 --- a/packages/ssh/src/tunnel.test.ts +++ b/packages/ssh/src/tunnel.test.ts @@ -189,7 +189,7 @@ describe("ssh tunnel scripts", () => { assert.include(buildRemoteStopScript(target), 'rm -f "$PID_FILE" "$PORT_FILE" "$MANAGED_FILE"'); assert.include( buildRemoteLaunchScript(), - 'DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata-v2/server-runtime.json"', + 'DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata/server-runtime.json"', ); assert.include(buildRemoteLaunchScript(), "resolve_default_runtime_port()"); assert.include( diff --git a/packages/ssh/src/tunnel.ts b/packages/ssh/src/tunnel.ts index 25ea25a787b..016d5e9a854 100644 --- a/packages/ssh/src/tunnel.ts +++ b/packages/ssh/src/tunnel.ts @@ -440,7 +440,7 @@ export const REMOTE_LAUNCH_SCRIPT = `set -eu STATE_KEY="$1" STATE_DIR="$HOME/.t3/ssh-launch/$STATE_KEY" DEFAULT_SERVER_HOME="$HOME/.t3" -DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata-v2/server-runtime.json" +DEFAULT_RUNTIME_FILE="$DEFAULT_SERVER_HOME/userdata/server-runtime.json" PORT_FILE="$STATE_DIR/port" PID_FILE="$STATE_DIR/pid" MANAGED_FILE="$STATE_DIR/managed" From 80082809973557d1cec7f7107932a4db1d37f0e7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 24 Jul 2026 13:06:45 +0200 Subject: [PATCH 2/2] fix(server): preserve migrated context across provider failure Co-authored-by: codex --- .../src/orchestration-v2/Orchestrator.ts | 56 +++- .../ProviderSwitch.integration.test.ts | 278 ++++++++++++++++++ 2 files changed, 324 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index f169a1e44b4..11576fa57da 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2471,6 +2471,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const ordinal = nextRunOrdinal(projection); const runId = idAllocator.derive.run({ threadId: command.threadId, ordinal }); const latestCompletedRun = projection.runs.findLast((run) => run.status === "completed"); + const legacyImportItems = + projection.thread.historyOrigin === "v1_import" + ? projection.turnItems.filter((item) => item.runId === null) + : []; const isProviderSwitch = activeProviderThread !== undefined && activeProviderThread.providerInstanceId !== modelSelection.instanceId; @@ -2508,14 +2512,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const legacyImportHandoff = projection.thread.historyOrigin === "v1_import" && projection.runs.length === 0 && - projection.turnItems.length > 0 + legacyImportItems.length > 0 ? yield* contextHandoffService .prepareLegacyImport({ threadId: command.threadId, targetRunId: runId, toProviderThreadId: providerThreadId, toProviderInstanceId: modelSelection.instanceId, - items: projection.turnItems, + items: legacyImportItems, createdAt: now, }) .pipe(mapDispatchError(command)) @@ -3002,11 +3006,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const providerSwitchItems = providerSwitchCoveredRuns.length === 0 ? [] - : projection.turnItems.filter( - (item) => - item.runId !== null && - providerSwitchCoveredRuns.some((run) => run.id === item.runId), - ); + : [ + ...(targetProviderThread === undefined || requiresFullProviderSwitchContext + ? legacyImportItems + : []), + ...projection.turnItems.filter( + (item) => + item.runId !== null && + providerSwitchCoveredRuns.some((run) => run.id === item.runId), + ), + ]; const providerSwitchTransferId = providerSwitchCoveredRuns.length === 0 || latestCompletedRun === undefined ? null @@ -3070,6 +3079,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); + const legacyImportRecoveryHandoff = + isProviderSwitch && latestCompletedRun === undefined && legacyImportItems.length > 0 + ? yield* contextHandoffService + .prepareLegacyImport({ + threadId: command.threadId, + targetRunId: runId, + toProviderThreadId: ensuredProviderThread.id, + toProviderInstanceId: modelSelection.instanceId, + items: legacyImportItems, + createdAt: now, + }) + .pipe(mapDispatchError(command)) + : null; const providerThread: OrchestrationV2ProviderThread = { ...ensuredProviderThread, status: "active", @@ -3077,8 +3099,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio lastRunOrdinal: ordinal, handoffIds: [ ...ensuredProviderThread.handoffIds, - ...[portableForkHandoff, providerSwitchHandoff].flatMap((handoff) => - handoff === null ? [] : [handoff.id], + ...[portableForkHandoff, providerSwitchHandoff, legacyImportRecoveryHandoff].flatMap( + (handoff) => (handoff === null ? [] : [handoff.id]), ), ], updatedAt: now, @@ -3222,7 +3244,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completedAt: null, checkpointId: null, contextHandoffId: - portableForkHandoff?.id ?? providerSwitchHandoff?.id ?? mergeBackHandoff?.id ?? null, + portableForkHandoff?.id ?? + providerSwitchHandoff?.id ?? + mergeBackHandoff?.id ?? + legacyImportRecoveryHandoff?.id ?? + null, ...(command.sourcePlanRef === undefined ? {} : { sourcePlanRef: command.sourcePlanRef }), }; const attempt: OrchestrationV2RunAttempt = { @@ -3414,6 +3440,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: portableForkHandoff, }); } + if (legacyImportRecoveryHandoff !== null) { + yield* emitEvent({ + type: "context-handoff.updated", + threadId: command.threadId, + runId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: legacyImportRecoveryHandoff, + }); + } if ( providerSwitchTransferId !== null && providerSwitchHandoff !== null && diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 44d9853be94..9b009720d3a 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -17,20 +17,35 @@ import { } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { ClaudeProviderCapabilitiesV2 } from "../Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "../Adapters/CodexAdapterV2.ts"; import { CursorProviderCapabilitiesV2 } from "../Adapters/CursorAdapterV2.ts"; +import { layer as eventSinkLayer } from "../EventSink.ts"; +import { layer as eventStoreLayer } from "../EventStore.ts"; +import { + LegacyV1ThreadImporter, + layer as legacyV1ThreadImporterLayer, +} from "../LegacyV1ThreadImporter.ts"; import { OrchestratorV2 } from "../Orchestrator.ts"; +import { + ProjectionMaintenanceV2, + layer as projectionMaintenanceLayer, +} from "../ProjectionMaintenance.ts"; +import { layer as projectionStoreLayer } from "../ProjectionStore.ts"; import { type ProviderAdapterV2Event, ProviderAdapterProtocolError, type ProviderAdapterV2Shape, } from "../ProviderAdapter.ts"; import { makeLayer as makeProviderAdapterRegistryLayer } from "../ProviderAdapterRegistry.ts"; +import { makeProviderFailure } from "../ProviderFailure.ts"; import { CLAUDE_MODEL_SELECTION, CODEX_MODEL_SELECTION, @@ -68,6 +83,7 @@ function makeTestAdapter(input: { readonly responseByThreadId?: Readonly>>>; readonly capturedTurns: Ref.Ref>; readonly failResume?: boolean; + readonly failedRunOrdinals?: ReadonlySet; }): ProviderAdapterV2Shape { return { instanceId: input.instanceId, @@ -143,6 +159,43 @@ function makeTestAdapter(input: { const providerTurnId = ProviderTurnId.make( `provider-turn:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}`, ); + if (input.failedRunOrdinals?.has(turnInput.runOrdinal) === true) { + yield* PubSub.publish(events, { + type: "provider_turn.updated", + driver: input.driver, + providerTurn: { + id: providerTurnId, + providerThreadId: turnInput.providerThread.id, + nodeId: turnInput.rootNodeId, + runAttemptId: turnInput.attemptId, + nativeTurnRef: { + driver: input.driver, + nativeId: `native-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, + strength: "strong", + }, + ordinal: turnInput.runOrdinal, + status: "failed", + startedAt: eventTime, + completedAt: eventTime, + }, + }); + yield* PubSub.publish(events, { + type: "turn.terminal", + driver: input.driver, + providerThreadId: turnInput.providerThread.id, + providerTurnId, + runOrdinal: turnInput.runOrdinal, + failureItemOrdinal: turnInput.runOrdinal * 100 + 1, + status: "failed", + failure: makeProviderFailure({ + message: "Simulated provider failure.", + code: "simulated_failure", + class: "provider_error", + }), + threadDisposition: "reusable", + }); + return; + } const response = input.responseByThreadId?.[turnInput.threadId]?.[turnInput.runOrdinal] ?? input.responseByRunOrdinal[turnInput.runOrdinal] ?? @@ -243,6 +296,231 @@ const waitForIdle = Effect.fn("ProviderSwitchTest.waitForIdle")(function* ( }); describe("orchestration v2 provider switching", () => { + it.live("reissues imported v1 context when switching after the first provider fails", () => + Effect.scoped( + Effect.gen(function* () { + const importedThreadId = ThreadId.make("thread:provider-switch:legacy-import"); + const importedProjectId = ProjectId.make("project:provider-switch:legacy-import"); + const failedPrompt = "This first provider attempt should fail."; + const recoveryPrompt = "What was the imported release marker?"; + const cwd = yield* checkpointWorkspace("provider-switch-legacy-import"); + const capturedTurns = yield* Ref.make>([]); + const registryLayer = makeProviderAdapterRegistryLayer([ + makeTestAdapter({ + instanceId: ProviderInstanceId.make("codex"), + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + failedRunOrdinals: new Set([1]), + }), + makeTestAdapter({ + instanceId: ProviderInstanceId.make("claudeAgent"), + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: { 2: "The imported release marker is violet." }, + capturedTurns, + }), + ]); + const databaseLayer = SqlitePersistenceMemory; + const eventStoreProvided = eventStoreLayer.pipe(Layer.provideMerge(databaseLayer)); + const projectionStoreProvided = projectionStoreLayer.pipe( + Layer.provideMerge(databaseLayer), + ); + const storesProvided = Layer.mergeAll( + databaseLayer, + eventStoreProvided, + projectionStoreProvided, + ); + const eventSinkProvided = eventSinkLayer.pipe(Layer.provide(storesProvided)); + const importerProvided = legacyV1ThreadImporterLayer.pipe( + Layer.provide(Layer.mergeAll(storesProvided, eventSinkProvided)), + ); + const maintenanceProvided = projectionMaintenanceLayer.pipe(Layer.provide(storesProvided)); + const orchestratorProvided = makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "provider-switch-legacy-import", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registryLayer, + { databaseLayer }, + ); + const testLayer = Layer.mergeAll( + storesProvided, + importerProvided, + maintenanceProvided, + orchestratorProvided, + ); + + const projection = yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const importer = yield* LegacyV1ThreadImporter; + const maintenance = yield* ProjectionMaintenanceV2; + const orchestrator = yield* OrchestratorV2; + + yield* sql` + INSERT INTO projection_projects ( + project_id, + title, + workspace_root, + default_model_selection_json, + scripts_json, + created_at, + updated_at, + deleted_at + ) VALUES ( + ${importedProjectId}, + 'Imported provider switch project', + ${cwd}, + '{"instanceId":"codex","model":"gpt-5.4"}', + '[]', + '2026-01-01T00:00:00.000Z', + '2026-01-01T00:00:00.000Z', + NULL + ) + `; + yield* sql` + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + latest_turn_id, + created_at, + updated_at, + archived_at, + settled_override, + settled_at, + deleted_at + ) VALUES ( + ${importedThreadId}, + ${importedProjectId}, + 'Imported provider switch thread', + '{"instanceId":"codex","model":"gpt-5.4"}', + 'full-access', + 'default', + 'main', + ${cwd}, + NULL, + '2026-01-01T00:00:00.000Z', + '2026-01-01T00:00:00.000Z', + NULL, + NULL, + NULL, + NULL + ) + `; + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, + thread_id, + turn_id, + role, + text, + attachments_json, + is_streaming, + created_at, + updated_at + ) VALUES + ( + 'message:provider-switch:legacy-import:user', + ${importedThreadId}, + NULL, + 'user', + 'Remember that the imported release marker is violet.', + '[]', + 0, + '2026-01-01T01:00:00.000Z', + '2026-01-01T01:00:00.000Z' + ), + ( + 'message:provider-switch:legacy-import:assistant', + ${importedThreadId}, + NULL, + 'assistant', + 'I will remember violet.', + '[]', + 0, + '2026-01-01T01:01:00.000Z', + '2026-01-01T01:01:00.000Z' + ) + `; + + yield* importer.reconcileShells; + yield* maintenance.rebuild; + yield* importer.ensureTranscript(importedThreadId); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:provider-switch:legacy-import:failed"), + threadId: importedThreadId, + messageId: MessageId.make("message:provider-switch:legacy-import:failed"), + text: failedPrompt, + attachments: [], + modelSelection: CODEX_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* waitForIdle(importedThreadId); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:provider-switch:legacy-import:recovery"), + threadId: importedThreadId, + messageId: MessageId.make("message:provider-switch:legacy-import:recovery"), + text: recoveryPrompt, + attachments: [], + modelSelection: CLAUDE_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + return yield* waitForIdle(importedThreadId); + }).pipe(Effect.provide(testLayer)); + + const turns = yield* Ref.get(capturedTurns); + assert.deepEqual( + projection.runs.map((run) => [run.providerInstanceId, run.status]), + [ + ["codex", "failed"], + ["claudeAgent", "completed"], + ], + ); + assert.deepEqual( + projection.contextHandoffs.map((handoff) => [ + handoff.targetRunId, + handoff.strategy, + handoff.status, + ]), + [ + [projection.runs[0]?.id, "manual_context", "ready"], + [projection.runs[1]?.id, "manual_context", "ready"], + ], + ); + assert.equal(projection.runs[1]?.contextHandoffId, projection.contextHandoffs[1]?.id); + assert.include(turns[1]?.text ?? "", "Context handoff (manual_context):"); + assert.include(turns[1]?.text ?? "", "imported release marker is violet"); + assert.include(turns[1]?.text ?? "", "I will remember violet."); + assert.include(turns[1]?.text ?? "", recoveryPrompt); + assert.notInclude(turns[1]?.text ?? "", failedPrompt); + }), + ), + ); + it.live("uses portable fallback when native resume fails after a provider switch", () => Effect.scoped( Effect.gen(function* () {