From d13385701b34308cdff3cab208af2f59ea03b380 Mon Sep 17 00:00:00 2001 From: testikun Date: Thu, 3 Sep 2026 14:45:36 +0800 Subject: [PATCH] fix(runtime-host): lazily adopt legacy session connection identity Generated-by: OpenAI Codex --- .../canonical-session-projection.test.ts | 2 +- .../__tests__/execution-composition.test.ts | 38 ++++++++ .../src/__tests__/goal-root-authority.test.ts | 3 +- .../src/__tests__/message-coordinator.test.ts | 2 +- .../__tests__/root-turn-coordinator.test.ts | 9 +- .../session-catalog-coordinator.test.ts | 89 ++++++++++++++++++- .../src/server/execution-composition.ts | 35 +++++--- .../src/server/message-coordinator.ts | 19 ++-- .../src/server/root-turn-coordinator.ts | 58 +++++++++--- .../src/server/session-catalog-coordinator.ts | 50 +++++++++++ 10 files changed, 262 insertions(+), 43 deletions(-) diff --git a/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts b/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts index 06eff1cae3..e2c4ad41df 100644 --- a/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts +++ b/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts @@ -591,7 +591,7 @@ function createMessages( stores: ExecutionStoresWriter<'interactive'>, ): HostMessageCoordinator { const root: HostMessageRootPort = { - readSessionHeader: async () => ({ isArchived: false }), + readSessionAvailability: async () => ({ isArchived: false }), readRootState: () => ({ kind: 'active', sessionId, turnId: 'turn-1', runId: 'run-1' }), claimStopFence: async () => ({ ready: Promise.resolve(), diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 0f25236dc9..24c22ab827 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -470,6 +470,44 @@ test('production composition commits automatic titles through Host-owned Session }); }); +test('production composition lazily adopts an unlocked legacy Session before turn start', async () => { + await withCompositionRoot(async ({ root, owner }) => { + const connectionId = await configureFakeDefaultTarget(owner); + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + const session = await stores.sessionStore.create({ + cwd: root, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + }); + const { composition } = await createCapturedExecutionComposition(owner); + const context = { + hostEpoch: 'execution-composition-test', + connectionId: 'legacy-adoption-client', + principal: 'local_os_user' as const, + acquireResidency: () => ({ release() {} }), + }; + try { + const started = await composition.handlers['turn.start']( + { + sessionId: session.id, + turnId: 'legacy-adoption-turn', + content: { text: 'adopt this legacy Session' }, + }, + context, + ); + assert.equal(started.ok, true, JSON.stringify(started)); + await waitFor( + async () => + (await stores.sessionStore.readHeaderSnapshot(session.id)).llmConnectionId === + connectionId, + ); + } finally { + await composition.close(); + } + }); +}); + test('WorkHub creates new work through the production assignment composition', async () => { await withCompositionRoot(async ({ root, owner }) => { const connectionId = await configureFakeDefaultTarget(owner); diff --git a/packages/runtime-host/src/__tests__/goal-root-authority.test.ts b/packages/runtime-host/src/__tests__/goal-root-authority.test.ts index fcbc66a1a3..2e3c93bfa1 100644 --- a/packages/runtime-host/src/__tests__/goal-root-authority.test.ts +++ b/packages/runtime-host/src/__tests__/goal-root-authority.test.ts @@ -557,7 +557,8 @@ async function createFixture(options: { recoverAdmissions?: boolean } = {}): Pro let requestedDrain = false; const goalChangeListeners = new Set<() => void>(); const rootPort: HostMessageRootPort = { - readSessionHeader: (sessionId) => requireCoordinator(coordinator).readSessionHeader(sessionId), + readSessionAvailability: (sessionId, admission) => + requireCoordinator(coordinator).readSessionAvailability(sessionId, admission), readRootState: (sessionId) => requireCoordinator(coordinator).readRootState(sessionId), claimStopFence: (input, commitQueueFence, lease) => requireCoordinator(coordinator).claimStopFence(input, commitQueueFence, lease), diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index ae6c2e168f..da64a13385 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -3629,7 +3629,7 @@ function createFixture( const terminal = deferred(); let coordinator: HostMessageCoordinator; const root: HostMessageRootPort = { - readSessionHeader: async () => { + readSessionAvailability: async () => { return { isArchived: false }; }, readRootState: async () => { diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index 8ba15639e0..681b0d4651 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -1704,7 +1704,7 @@ test('worktree child Sessions reject roots outside managed child execution', asy }, ); assert.equal( - (await fixture.coordinator.readSessionHeader(child.id))?.unavailableReason, + (await fixture.coordinator.readSessionAvailability(child.id))?.unavailableReason, unavailableMessage, ); await assert.rejects( @@ -2582,8 +2582,8 @@ test('hosted linked child roots share admission, message, terminal, and stop aut let drainRequested = false; let stopClosureSignal: ReturnType> | undefined; const rootPort: HostMessageRootPort = { - readSessionHeader: (sessionId) => - requireCoordinator(coordinator).readSessionHeader(sessionId), + readSessionAvailability: (sessionId, admission) => + requireCoordinator(coordinator).readSessionAvailability(sessionId, admission), readRootState: (sessionId) => requireCoordinator(coordinator).readRootState(sessionId), claimStopFence: (input, commitQueueFence, admission) => requireCoordinator(coordinator).claimStopFence(input, commitQueueFence, admission), @@ -5078,7 +5078,8 @@ async function createFailureFixture(options: { let messages!: HostMessageCoordinator; let interactions: HostInteractionCoordinator | undefined; const rootPort: HostMessageRootPort = { - readSessionHeader: (sessionId) => requireCoordinator(coordinator).readSessionHeader(sessionId), + readSessionAvailability: (sessionId, admission) => + requireCoordinator(coordinator).readSessionAvailability(sessionId, admission), readRootState: (sessionId) => requireCoordinator(coordinator).readRootState(sessionId), claimStopFence: (input, commitQueueFence, admission) => requireCoordinator(coordinator).claimStopFence(input, commitQueueFence, admission), diff --git a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts index 614998f366..b13ae21c01 100644 --- a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts @@ -1075,6 +1075,91 @@ test('explicit recovery persists the selected Connection entity identity', async assert.equal(fixture.header().model, 'model-1'); }); +test('unlocked legacy Sessions lazily adopt the current slug owner with CAS revalidation', async () => { + const observed: unknown[] = []; + const fixture = createFixture({ + legacyConnectionIdentity: true, + header: { connectionLocked: false }, + connection: { + onResolve: (ref) => observed.push(ref), + }, + }); + + const adopted = await fixture.admission.run(fixture.sessionId, (lease) => + fixture.coordinator.adoptLegacySessionConnectionIdentity(fixture.sessionId, lease), + ); + + assert.ok(adopted); + assert.equal(fixture.header().llmConnectionId, 'connection-1'); + assert.deepEqual(observed, [ + { + kind: 'catalog_slug', + connectionSlug: 'test', + }, + ]); +}); + +test('locked legacy Sessions never guess an identity, even when the slug exists today', async () => { + const observed: unknown[] = []; + const fixture = createFixture({ + legacyConnectionIdentity: true, + connection: { + onResolve: (ref) => observed.push(ref), + }, + }); + + const adopted = await fixture.admission.run(fixture.sessionId, (lease) => + fixture.coordinator.adoptLegacySessionConnectionIdentity(fixture.sessionId, lease), + ); + + assert.equal(adopted, undefined); + assert.equal(fixture.header().llmConnectionId, undefined); + assert.deepEqual(observed, []); +}); + +test('legacy adoption leaves a Session unbound when its metadata CAS loses a race', async () => { + const fixture = createFixture({ + legacyConnectionIdentity: true, + header: { connectionLocked: false }, + stores: { + updateHeaderVersioned: async (_sessionId, _patch, expectedRevision) => { + throw new SessionMetadataVersionConflictError( + 'session-1', + expectedRevision, + expectedRevision + 1, + ); + }, + }, + }); + + const adopted = await fixture.admission.run(fixture.sessionId, (lease) => + fixture.coordinator.adoptLegacySessionConnectionIdentity(fixture.sessionId, lease), + ); + + assert.equal(adopted, undefined); + assert.equal(fixture.header().llmConnectionId, undefined); +}); + +for (const executionResolution of [ + { kind: 'not_found' as const }, + { kind: 'identity_mismatch' as const }, +]) { + test(`legacy adoption stays unbound when slug preflight is ${executionResolution.kind}`, async () => { + const fixture = createFixture({ + legacyConnectionIdentity: true, + header: { connectionLocked: false }, + connection: { executionResolution }, + }); + + const adopted = await fixture.admission.run(fixture.sessionId, (lease) => + fixture.coordinator.adoptLegacySessionConnectionIdentity(fixture.sessionId, lease), + ); + + assert.equal(adopted, undefined); + assert.equal(fixture.header().llmConnectionId, undefined); + }); +} + test('creation persists a canonical cwd while fingerprints retain exact target intent', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-session-create-cwd-')); const target = join(root, 'target'); @@ -1498,6 +1583,7 @@ function createFixture( }, ...options.manager, }; + const admission = new SessionAdmissionGate(); const continuity: SessionContinuity = { refreshCanonical: async () => undefined, ...options.continuity, @@ -1506,7 +1592,7 @@ function createFixture( stores, runtimePolicy, manager, - admission: new SessionAdmissionGate(), + admission, continuity, workspaceResolver: new HostWorkspaceResolver( options.projectCatalog ?? ({ list: async () => [] } as never), @@ -1519,6 +1605,7 @@ function createFixture( }); return { coordinator, + admission, sessionId, revision: () => revision, header: () => header, diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 2c390fed4f..2014e3218c 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -557,8 +557,8 @@ export async function createExecutionRuntimeHostComposition( let deepResearch: HostDeepResearchCoordinator | undefined; let dailyReview: HostDailyReviewCoordinator | undefined; const rootPort: HostMessageRootPort = { - readSessionHeader: (sessionId) => - requireRootCoordinator(rootCoordinator).readSessionHeader(sessionId), + readSessionAvailability: (sessionId, admission) => + requireRootCoordinator(rootCoordinator).readSessionAvailability(sessionId, admission), readRootState: (sessionId) => requireRootCoordinator(rootCoordinator).readRootState(sessionId), claimStopFence: (input, commitQueueFence, admission) => @@ -1163,6 +1163,18 @@ export async function createExecutionRuntimeHostComposition( Date.now, context.sessionAccessAuthority, ); + const sessionCatalog = new HostSessionCatalogCoordinator({ + stores: stores.sessionStore, + runtimePolicy: runtimePolicyStores, + manager, + admission: sessionAdmission, + continuity: continuityCoordinator, + workspaceResolver, + requestDrain: context.requestDrain, + ...(context.sessionAccessAuthority + ? { sessionAccessAuthority: context.sessionAccessAuthority } + : {}), + }); rootCoordinator = new RootTurnCoordinator( manager, stores, @@ -1209,6 +1221,13 @@ export async function createExecutionRuntimeHostComposition( }, (input) => sessionEffectCoordinator.nameSessionFromRootMessage(input), context.owner.capability.rootId, + async (sessionId, header, admissionLease) => { + const adopted = await sessionCatalog.adoptLegacySessionConnectionIdentity( + sessionId, + admissionLease, + ); + return adopted?.header ?? header; + }, ); const coordinator = rootCoordinator; const contextOperations = new HostContextCoordinator({ @@ -1327,18 +1346,6 @@ export async function createExecutionRuntimeHostComposition( oauthCredentials, onCommittedMutation: registerConfigurationMutation, }); - const sessionCatalog = new HostSessionCatalogCoordinator({ - stores: stores.sessionStore, - runtimePolicy: runtimePolicyStores, - manager, - admission: sessionAdmission, - continuity: continuityCoordinator, - workspaceResolver, - requestDrain: context.requestDrain, - ...(context.sessionAccessAuthority - ? { sessionAccessAuthority: context.sessionAccessAuthority } - : {}), - }); const workHubCoordination = new HostWorkHubCoordinationCoordinator({ stateRoot: context.owner.capability.canonicalPath, stores: stores.sessionStore, diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 2f483a7ee1..1078fd13e3 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -190,7 +190,10 @@ export type HostMessageExecutionDisposition = /** Root execution operations that must share the message coordinator's Session gate. */ export interface HostMessageRootPort { - readSessionHeader(sessionId: string): Promise; + readSessionAvailability( + sessionId: string, + admission?: SessionAdmissionLease, + ): Promise; readRootState(sessionId: string): Promise | HostMessageRootState; claimStopFence( input: Omit, @@ -1094,7 +1097,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { } | undefined; for (let attempt = 0; ; attempt++) { - const header = await this.#root.readSessionHeader(input.sessionId); + const header = await this.#root.readSessionAvailability(input.sessionId, admission); if (this.#failStopped) { return failure('host_draining', 'Runtime Host message authority has failed'); } @@ -1412,7 +1415,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { } async #retractAdmitted(input: QueueRetractInput): Promise> { - const header = await this.#root.readSessionHeader(input.sessionId); + const header = await this.#root.readSessionAvailability(input.sessionId); if (this.#failStopped) { return failure('host_draining', 'Runtime Host message authority has failed'); } @@ -1585,7 +1588,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { async #retractQueuedEntryAdmitted( input: QueueEntryRetractInput, ): Promise> { - const header = await this.#root.readSessionHeader(input.sessionId); + const header = await this.#root.readSessionAvailability(input.sessionId); if (this.#failStopped) { return failure('host_draining', 'Runtime Host message authority has failed'); } @@ -1621,7 +1624,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { async #promoteQueuedEntryAdmitted( input: QueueEntryPromoteInput, ): Promise> { - const header = await this.#root.readSessionHeader(input.sessionId); + const header = await this.#root.readSessionAvailability(input.sessionId); if (this.#failStopped) { return failure('host_draining', 'Runtime Host message authority has failed'); } @@ -1703,7 +1706,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { async #updateQueuedEntryAdmitted( input: QueueEntryUpdateInput, ): Promise> { - const header = await this.#root.readSessionHeader(input.sessionId); + const header = await this.#root.readSessionAvailability(input.sessionId); if (this.#failStopped) { return failure('host_draining', 'Runtime Host message authority has failed'); } @@ -1821,7 +1824,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { async #reorderQueuedEntriesAdmitted( input: QueueEntriesReorderInput, ): Promise> { - const header = await this.#root.readSessionHeader(input.sessionId); + const header = await this.#root.readSessionAvailability(input.sessionId); if (this.#failStopped) { return failure('host_draining', 'Runtime Host message authority has failed'); } @@ -1887,7 +1890,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { }; } - const header = await this.#root.readSessionHeader(input.sessionId); + const header = await this.#root.readSessionAvailability(input.sessionId); if (this.#failStopped) { return { kind: 'conflict' as const, diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index d246e0c4a6..3323a9f165 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -299,6 +299,12 @@ interface HostAgentGraphEpochAuthority { beginNextGraphEpoch(rootSessionId: string): Promise; } +type LegacySessionIdentityAdopter = ( + sessionId: string, + header: SessionHeader, + admissionLease: SessionAdmissionLease, +) => Promise; + export class RootTurnCoordinator implements HostedExecutionAuthority { readonly handlers: Pick = { 'turn.resume.query': (input, context) => this.queryTurnResume(input, context), @@ -337,6 +343,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { content: MessageContent; }) => void, private readonly directoryHostId?: string, + private readonly adoptLegacySessionIdentity?: LegacySessionIdentityAdopter, ) { this.stores = authenticateExecutionStoresWriter(stores, 'interactive'); this.executionProjection = new HostedExecutionProjectionReader(this.stores); @@ -347,6 +354,25 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { this.prepareSkillInvocation = prepareSkillInvocation; } + private async prepareSessionHeader( + header: SessionHeader, + admissionLease?: SessionAdmissionLease, + ): Promise { + return this.adoptLegacySessionIdentity && admissionLease + ? this.adoptLegacySessionIdentity(header.id, header, admissionLease) + : header; + } + + private async readAdmittedSessionHeader( + sessionId: string, + admissionLease: SessionAdmissionLease, + ): Promise { + return this.prepareSessionHeader( + await this.stores.sessionStore.readHeaderSnapshot(sessionId), + admissionLease, + ); + } + async prepareRecovery(): Promise { const plans = await prepareHostedExecutionRecovery({ stores: this.stores, @@ -397,7 +423,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { const input = activationInputForAdmission(admission); const disposition = await this.sessionAdmission.run(sessionId, async (lease) => { if (admission.execution.kind === 'safe_boundary_continuation') { - const header = await this.stores.sessionStore.readHeaderSnapshot(sessionId); + const header = await this.readAdmittedSessionHeader(sessionId, lease); if (runtimeHostSafeBoundaryContinuationUnavailableReason(header)) { this.parkContinuationAdmission(admission); return undefined; @@ -477,7 +503,10 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { throw new AggregateError(errors, 'Unable to close Runtime Host execution composition'); } - async readSessionHeader(sessionId: string): Promise { + async readSessionAvailability( + sessionId: string, + admissionLease?: SessionAdmissionLease, + ): Promise { if (isWorkHubCoordinationSessionId(sessionId)) { return { isArchived: false, @@ -485,7 +514,9 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { }; } try { - const header = await this.stores.sessionStore.readHeaderSnapshot(sessionId); + const header = admissionLease + ? await this.readAdmittedSessionHeader(sessionId, admissionLease) + : await this.stores.sessionStore.readHeaderSnapshot(sessionId); if (header.conversationCopy?.state === 'preparing') return null; return { isArchived: header.isArchived, @@ -761,7 +792,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { let header: SessionHeader; try { - header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId); + header = await this.readAdmittedSessionHeader(input.sessionId, lease); } catch (error) { if (isSessionNotFoundError(error)) { throw new RuntimeHostedRootUnavailableError( @@ -1057,7 +1088,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { const reservation = this.reserveRootTurn(input.sessionId); if (!reservation) return { error: 'Another root Turn is being admitted' }; try { - const header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId); + const header = await this.readAdmittedSessionHeader(input.sessionId, admissionLease); const unavailableReason = runtimeHostExternalTurnUnavailableReason(header); if (unavailableReason) return { error: unavailableReason }; const turnId = input.turnId ?? randomUUID(); @@ -1188,7 +1219,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { if (this.#executions.has(input.sessionId)) { return { error: 'A root Turn is still active' }; } - const header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId); + const header = await this.readAdmittedSessionHeader(input.sessionId, admissionLease); const unavailableReason = runtimeHostExternalTurnUnavailableReason(header); if (unavailableReason) return { error: unavailableReason }; const reservation = this.reserveRootTurn(input.sessionId); @@ -1539,7 +1570,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { } let header: SessionHeader; try { - header = await this.stores.sessionStore.readHeaderSnapshot(request.sessionId); + header = await this.readAdmittedSessionHeader(request.sessionId, lease); } catch (error) { if (isSessionNotFoundError(error)) { return completedStart(notFound('Session does not exist')); @@ -1694,10 +1725,10 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { if (isWorkHubCoordinationSessionId(input.sessionId)) { return operationUnavailable(WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON); } - return this.sessionAdmission.run(input.sessionId, async () => { + return this.sessionAdmission.run(input.sessionId, async (lease) => { let header: SessionHeader; try { - header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId); + header = await this.readAdmittedSessionHeader(input.sessionId, lease); } catch (error) { if (isSessionNotFoundError(error)) return notFound('Session does not exist'); throw error; @@ -1785,8 +1816,9 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { ), }; } - const header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId); - const unavailableReason = runtimeHostSafeBoundaryContinuationUnavailableReason(header); + const preparedHeader = await this.readAdmittedSessionHeader(input.sessionId, lease); + const unavailableReason = + runtimeHostSafeBoundaryContinuationUnavailableReason(preparedHeader); if (unavailableReason) { return { kind: 'complete', @@ -1876,7 +1908,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { let header: SessionHeader; try { - header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId); + header = await this.readAdmittedSessionHeader(input.sessionId, lease); } catch (error) { if (isSessionNotFoundError(error)) { return { @@ -2177,7 +2209,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { 'Root Turn admission payload does not match its input', ); } - const session = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId); + const session = await this.readAdmittedSessionHeader(input.sessionId, admissionLease); const unavailableReason = admission.execution.kind === 'safe_boundary_continuation' ? runtimeHostSafeBoundaryContinuationUnavailableReason(session) diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts index 3180a34226..3c3dac19c3 100644 --- a/packages/runtime-host/src/server/session-catalog-coordinator.ts +++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts @@ -226,6 +226,56 @@ export class HostSessionCatalogCoordinator { if (!outcome.ok) throw new Error(outcome.error.message); } + /** + * Compatibility migration for pre-identity Sessions whose + * `connectionLocked` flag is still false. The write is revision/CAS guarded + * and the resulting bound identity is resolved through the existing + * catalog-slug resolver before the metadata commit. + * The optional lease lets an already-admitted execution reuse its Session + * lane; callers without one are serialized here. Locked legacy Sessions stay + * untouched because the catalog cannot prove historical ownership after + * delete-and-reuse of a slug. + */ + async adoptLegacySessionConnectionIdentity( + sessionId: string, + admissionLease?: SessionAdmissionLease, + ): Promise { + return admissionLease + ? this.#admission.runAdmitted(sessionId, admissionLease, () => + this.#adoptLegacySessionConnectionIdentity(sessionId), + ) + : this.#admission.run(sessionId, () => this.#adoptLegacySessionConnectionIdentity(sessionId)); + } + + async #adoptLegacySessionConnectionIdentity( + sessionId: string, + ): Promise { + const current = await this.#stores.readHeaderRecordSnapshot(sessionId); + if ( + current.header.llmConnectionId !== undefined || + current.header.backend === 'fake' || + current.header.connectionLocked + ) { + return undefined; + } + const resolved = await this.#runtimePolicy.operations.resolveExecutionConnection({ + kind: 'catalog_slug', + connectionSlug: current.header.llmConnectionSlug, + }); + if (resolved.kind !== 'ready') return undefined; + const connection = resolved.connection; + try { + return await this.#stores.updateHeaderVersioned( + sessionId, + { llmConnectionId: connection.connectionId }, + current.revision, + ); + } catch (error) { + if (error instanceof SessionMetadataVersionConflictError) return undefined; + throw error; + } + } + /** WorkHub Action Gate path; callers cannot bypass the typed operation outcome. */ createForWorkHub(input: SessionCreateInput): Promise> { return this.#create(input);