diff --git a/packages/cli/src/serve/live/live-session-coordinator.test.ts b/packages/cli/src/serve/live/live-session-coordinator.test.ts index ec8fc7da7ce..7b226b7a859 100644 --- a/packages/cli/src/serve/live/live-session-coordinator.test.ts +++ b/packages/cli/src/serve/live/live-session-coordinator.test.ts @@ -1002,6 +1002,36 @@ describe('LiveSessionCoordinator', () => { await harness.finishTurn(0, [{ type: 'message', text: '继续完成。' }]); }); + it('registers a mixed-case resumed Live session by its canonical id', async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + const persistedSessionId = sessionId.toUpperCase(); + const sourceId = LIVE_SESSION_SOURCE_PREFIX + 'mixed-case'; + const harness = makeHarness({ + recent: [ + { + sessionId: persistedSessionId, + sourceType: 'default', + sourceId, + } as SessionListItem, + ], + }); + await harness.coordinator.start({ + epoch: 1, + callId: 'call-1', + mode: 'resume', + }); + + expect(harness.bridge.resumeSession).toHaveBeenCalledWith({ + sessionId, + workspaceCwd: '/conversations', + sourceType: 'default', + sourceId, + }); + expect(harness.bridge.resumeSession).not.toHaveBeenCalledWith( + expect.objectContaining({ sessionId: persistedSessionId }), + ); + }); + it('tracks a task session only from a completed built-in create_sub_session result', async () => { readPersistedParentSessionId.mockResolvedValue('live-new'); const harness = makeHarness(); diff --git a/packages/cli/src/serve/live/live-session-coordinator.ts b/packages/cli/src/serve/live/live-session-coordinator.ts index 7f566002400..3da56b2f7f5 100644 --- a/packages/cli/src/serve/live/live-session-coordinator.ts +++ b/packages/cli/src/serve/live/live-session-coordinator.ts @@ -43,6 +43,7 @@ import { isCompatibleLiveSessionSource, LIVE_SESSION_SOURCE_PREFIX, } from '../../runtime/live-session-source.js'; +import { normalizeSessionIdForLookup } from '../../config/session-id.js'; import type { LiveProviderReadiness, LiveSessionLocator } from './types.js'; export { LIVE_SESSION_SOURCE_PREFIX } from '../../runtime/live-session-source.js'; @@ -1407,7 +1408,7 @@ export class LiveSessionCoordinator { if (candidate) { try { const resumed = await runtime.bridge.resumeSession({ - sessionId: candidate.sessionId, + sessionId: normalizeSessionIdForLookup(candidate.sessionId), workspaceCwd: runtime.workspaceCwd, ...(candidate.parentSessionId ? { parentSessionId: candidate.parentSessionId } diff --git a/packages/cli/src/serve/live/live-task-service.test.ts b/packages/cli/src/serve/live/live-task-service.test.ts index 43787c968f7..bbab1f2ddaf 100644 --- a/packages/cli/src/serve/live/live-task-service.test.ts +++ b/packages/cli/src/serve/live/live-task-service.test.ts @@ -322,6 +322,7 @@ function makeHarness() { bridge, projectBridge, runtime, + projectRuntime, registry, summaries, resident, @@ -757,6 +758,110 @@ describe('LiveTaskService', () => { ).toEqual(['task-1', 'task-2']); }); + it('polls a mixed-case task through its canonical bridge entry', async () => { + const harness = makeHarness(); + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + const persistedSessionId = sessionId.toUpperCase(); + const liveSummary: BridgeSessionSummary = { + sessionId, + workspaceCwd: '/conversations', + createdAt: '2026-07-30T00:00:00.000Z', + displayName: 'Active task', + clientCount: 1, + hasActivePrompt: true, + }; + harness.summaries.set(sessionId, liveSummary); + harness.resident.add(sessionId); + persistedSessions.set(persistedSessionId, persisted(persistedSessionId)); + persistedSessionOwners.set(persistedSessionId, '/conversations'); + listWorkspaceSessionsForResponse.mockResolvedValue({ + sessions: [{ ...liveSummary, sessionId: persistedSessionId }], + }); + const resolveLiveSessionOwner = vi.spyOn( + harness.registry, + 'resolveLiveSessionOwner', + ); + const subscribeEvents = vi.spyOn(harness.bridge, 'subscribeEvents'); + + const result = await harness.service.handle({ + callerSessionId: 'live-root', + name: 'wait_threads', + arguments: { + targets: [{ threadId: persistedSessionId }], + timeoutMs: 10, + }, + }); + + expect(result).toMatchObject({ + timedOut: true, + polls: [{ thread: { id: persistedSessionId } }], + }); + expect(resolveLiveSessionOwner).toHaveBeenCalledWith(sessionId); + expect(resolveLiveSessionOwner).not.toHaveBeenCalledWith( + persistedSessionId, + ); + expect(harness.bridge.getSessionEventEpoch).toHaveBeenCalledWith(sessionId); + expect(harness.bridge.getSessionLastEventId).toHaveBeenCalledWith( + sessionId, + ); + expect(subscribeEvents).toHaveBeenCalledWith( + sessionId, + expect.objectContaining({ lastEventId: 7 }), + ); + expect(harness.bridge.getSessionEventEpoch).not.toHaveBeenCalledWith( + persistedSessionId, + ); + expect(harness.bridge.getSessionLastEventId).not.toHaveBeenCalledWith( + persistedSessionId, + ); + expect(subscribeEvents).not.toHaveBeenCalledWith( + persistedSessionId, + expect.any(Object), + ); + }); + + it('keeps the caller-visible id when a mixed-case task has no user turn', async () => { + const harness = makeHarness(); + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + const persistedSessionId = sessionId.toUpperCase(); + harness.summaries.set(sessionId, { + sessionId, + workspaceCwd: '/conversations', + createdAt: '2026-07-30T00:00:00.000Z', + displayName: 'Empty task', + clientCount: 0, + hasActivePrompt: false, + }); + harness.resident.add(sessionId); + persistedSessions.set(persistedSessionId, { + conversation: { + sessionId: persistedSessionId, + startTime: '2026-07-30T00:00:00.000Z', + lastUpdated: '2026-07-30T00:00:00.000Z', + messages: [], + }, + }); + persistedSessionOwners.set(persistedSessionId, '/conversations'); + + const result = await harness.service.handle({ + callerSessionId: 'live-root', + name: 'wait_threads', + arguments: { + targets: [{ threadId: persistedSessionId }], + timeoutMs: 0, + }, + }); + + expect(result).toMatchObject({ + polls: [ + { + thread: { id: persistedSessionId }, + latestTurn: { id: persistedSessionId }, + }, + ], + }); + }); + it('suppresses previously delivered text and markers for an unchanged cursor', async () => { const harness = makeHarness(); const summary: BridgeSessionSummary = { @@ -903,6 +1008,194 @@ describe('LiveTaskService', () => { expect(harness.sendPrompt).toHaveBeenCalledOnce(); }); + it('reuses the canonical bridge entry for a mixed-case persisted task', async () => { + const harness = makeHarness(); + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + const persistedSessionId = sessionId.toUpperCase(); + const liveSummary: BridgeSessionSummary = { + sessionId, + workspaceCwd: '/conversations', + createdAt: '2026-07-30T00:00:00.000Z', + displayName: 'Existing task', + clientCount: 0, + hasActivePrompt: false, + }; + const persistedSummary = { + ...liveSummary, + sessionId: persistedSessionId, + }; + harness.summaries.set(sessionId, liveSummary); + harness.resident.add(sessionId); + persistedSessions.set(persistedSessionId, persisted(persistedSessionId)); + persistedSessionOwners.set(persistedSessionId, '/conversations'); + listWorkspaceSessionsForResponse.mockResolvedValue({ + sessions: [persistedSummary], + }); + + const result = await harness.service.handle({ + callerSessionId: 'live-root', + name: 'send_message_to_thread', + arguments: { + threadId: persistedSessionId, + prompt: 'continue this task', + }, + }); + + expect(result).toEqual({ threadId: persistedSessionId }); + expect(harness.bridge.resumeSession).not.toHaveBeenCalled(); + expect(harness.bridge.changeSessionCwd).not.toHaveBeenCalled(); + expect(harness.sendPrompt).toHaveBeenCalledWith( + sessionId, + expect.objectContaining({ sessionId }), + undefined, + expect.any(Object), + ); + expect(harness.resident).not.toContain(persistedSessionId); + }); + + it('rejects a mixed-case task whose storage and live owners differ', async () => { + const harness = makeHarness(); + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + const persistedSessionId = sessionId.toUpperCase(); + harness.summaries.set(sessionId, { + sessionId, + workspaceCwd: '/project', + createdAt: '2026-07-30T00:00:00.000Z', + displayName: 'Other workspace task', + clientCount: 1, + hasActivePrompt: false, + }); + harness.resident.add(sessionId); + persistedSessions.set(persistedSessionId, persisted(persistedSessionId)); + persistedSessionOwners.set(persistedSessionId, '/conversations'); + vi.spyOn(harness.registry, 'resolveLiveSessionOwner').mockReturnValue({ + kind: 'found', + runtime: harness.projectRuntime, + }); + + await expect( + harness.service.handle({ + callerSessionId: 'live-root', + name: 'send_message_to_thread', + arguments: { + threadId: persistedSessionId, + prompt: 'continue this task', + }, + }), + ).rejects.toThrow(`Task id is ambiguous: ${persistedSessionId}`); + expect(harness.sendPrompt).not.toHaveBeenCalled(); + }); + + it('uses persisted metadata if a canonical live task disappears', async () => { + const harness = makeHarness(); + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + const persistedSessionId = sessionId.toUpperCase(); + const sourceId = `${LIVE_SESSION_SOURCE_PREFIX}mixed-case-race`; + harness.summaries.set(sessionId, { + sessionId, + workspaceCwd: '/conversations', + createdAt: '2026-07-30T00:00:00.000Z', + displayName: 'Disappearing task', + clientCount: 0, + hasActivePrompt: false, + }); + harness.resident.add(sessionId); + persistedSessions.set(persistedSessionId, persisted(persistedSessionId)); + persistedSessionOwners.set(persistedSessionId, '/conversations'); + sessionSources.set(persistedSessionId, { + sourceType: 'default', + sourceId, + }); + const getSessionSummary = harness.bridge.getSessionSummary.bind( + harness.bridge, + ); + vi.spyOn(harness.bridge, 'getSessionSummary').mockImplementation( + (requestedSessionId) => { + const summary = getSessionSummary(requestedSessionId); + if (requestedSessionId === sessionId) { + harness.resident.delete(sessionId); + } + return summary; + }, + ); + + await harness.service.handle({ + callerSessionId: 'live-root', + name: 'send_message_to_thread', + arguments: { + threadId: persistedSessionId, + prompt: 'continue this task', + }, + }); + + expect(harness.bridge.resumeSession).toHaveBeenCalledWith({ + sessionId, + workspaceCwd: '/conversations', + sourceType: 'default', + sourceId, + }); + expect(harness.sendPrompt).toHaveBeenCalledWith( + sessionId, + expect.objectContaining({ sessionId }), + undefined, + expect.any(Object), + ); + }); + + it('uses one canonical bridge id while resuming a mixed-case persisted task', async () => { + const harness = makeHarness(); + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + const persistedSessionId = sessionId.toUpperCase(); + const sourceId = `${LIVE_SESSION_SOURCE_PREFIX}mixed-case`; + const summary: BridgeSessionSummary = { + sessionId: persistedSessionId, + workspaceCwd: '/conversations', + createdAt: '2026-07-30T00:00:00.000Z', + displayName: 'Persisted task', + clientCount: 0, + hasActivePrompt: false, + }; + harness.summaries.set(persistedSessionId, summary); + persistedSessions.set(persistedSessionId, persisted(persistedSessionId)); + persistedSessionOwners.set(persistedSessionId, '/conversations'); + sessionSources.set(persistedSessionId, { + sourceType: 'default', + sourceId, + }); + listWorkspaceSessionsForResponse.mockResolvedValue({ sessions: [summary] }); + + await harness.service.handle({ + callerSessionId: 'live-root', + name: 'send_message_to_thread', + arguments: { + threadId: persistedSessionId, + prompt: 'continue this task', + }, + }); + + expect(harness.bridge.resumeSession).toHaveBeenCalledWith({ + sessionId, + workspaceCwd: '/conversations', + sourceType: 'default', + sourceId, + }); + expect(harness.materializeConversationDirectory).toHaveBeenCalledWith( + sessionId, + ); + expect(harness.bridge.changeSessionCwd).toHaveBeenCalledWith(sessionId, { + path: `/conversations/${sessionId}`, + allowedRoots: ['/conversations'], + managedRelocation: 'live-conversation', + }); + expect(harness.sendPrompt).toHaveBeenCalledWith( + sessionId, + expect.objectContaining({ sessionId }), + undefined, + expect.any(Object), + ); + expect(harness.resident).not.toContain(persistedSessionId); + }); + it('restores Live source identity before following a cold Live task', async () => { const harness = makeHarness(); const sourceId = `${LIVE_SESSION_SOURCE_PREFIX}call-2`; diff --git a/packages/cli/src/serve/live/live-task-service.ts b/packages/cli/src/serve/live/live-task-service.ts index f39209dfd85..7531cb99c76 100644 --- a/packages/cli/src/serve/live/live-task-service.ts +++ b/packages/cli/src/serve/live/live-task-service.ts @@ -41,6 +41,7 @@ import { readLoadableLiveConversationMetadata, } from '../../runtime/live-session-source.js'; import { conversationRuntimeUnavailableError } from '../conversations/conversation-runtime-errors.js'; +import { normalizeSessionIdForLookup } from '../../config/session-id.js'; const DEFAULT_LIST_LIMIT = 20; const DEFAULT_READ_TURN_LIMIT = 3; @@ -50,6 +51,8 @@ const DEFAULT_ITEM_TEXT_CHARS = 4_000; interface LocatedTask { runtime: WorkspaceRuntime; + threadId: string; + bridgeSessionId: string; persisted: Awaited>; summary: BridgeSessionSummary; } @@ -801,11 +804,11 @@ export class LiveTaskService { const { cursor } = decodeCursor(target.afterCursor, target.threadId); const lastEventId = cursor.eventEpoch === - task.runtime.bridge.getSessionEventEpoch(target.threadId) + task.runtime.bridge.getSessionEventEpoch(task.bridgeSessionId) ? cursor.eventId - : task.runtime.bridge.getSessionLastEventId(target.threadId); + : task.runtime.bridge.getSessionLastEventId(task.bridgeSessionId); for await (const event of task.runtime.bridge.subscribeEvents( - target.threadId, + task.bridgeSessionId, { lastEventId, signal }, )) { const reason = eventWakeReason(event); @@ -831,9 +834,11 @@ export class LiveTaskService { ...(task.summary.clientCount > 0 ? { eventEpoch: task.runtime.bridge.getSessionEventEpoch( - target.threadId, + task.bridgeSessionId, + ), + eventId: task.runtime.bridge.getSessionLastEventId( + task.bridgeSessionId, ), - eventId: task.runtime.bridge.getSessionLastEventId(target.threadId), } : {}), updatedAt: laterActivityTimestamp( @@ -861,7 +866,7 @@ export class LiveTaskService { const failed = task.summary.hasTurnError === true; const revision = task.summary.clientCount > 0 - ? task.runtime.bridge.getSessionLastEventId(target.threadId) + ? task.runtime.bridge.getSessionLastEventId(task.bridgeSessionId) : epochSeconds( laterActivityTimestamp( task.summary.updatedAt, @@ -931,8 +936,7 @@ export class LiveTaskService { return ( [...(task.persisted?.conversation.messages ?? [])] .reverse() - .find((record) => record.type === 'user')?.uuid ?? - task.summary.sessionId + .find((record) => record.type === 'user')?.uuid ?? task.threadId ); } @@ -944,7 +948,11 @@ export class LiveTaskService { localHost(args['hostId']); const located = await this.locateTask(threadId); await this.ensureResident(located); - await this.dispatchPrompt(located.runtime.bridge, threadId, prompt); + await this.dispatchPrompt( + located.runtime.bridge, + located.bridgeSessionId, + prompt, + ); return { threadId }; } @@ -1053,7 +1061,7 @@ export class LiveTaskService { private async ensureResident(task: LocatedTask): Promise { try { - task.runtime.bridge.getSessionSummary(task.summary.sessionId); + task.runtime.bridge.getSessionSummary(task.bridgeSessionId); return; } catch (error) { if (!(error instanceof SessionNotFoundError)) throw error; @@ -1061,25 +1069,22 @@ export class LiveTaskService { const service = createWorkspaceRuntimeSessionService(task.runtime); const metadata = task.runtime.provenance === 'live-conversation' - ? await readLoadableLiveConversationMetadata( - task.summary.sessionId, - service, - ) - : await service.readCreationMetadata(task.summary.sessionId); + ? await readLoadableLiveConversationMetadata(task.threadId, service) + : await service.readCreationMetadata(task.threadId); if (metadata === undefined) { - throw new SessionNotFoundError(task.summary.sessionId); + throw new SessionNotFoundError(task.bridgeSessionId); } await task.runtime.bridge.resumeSession({ - sessionId: task.summary.sessionId, + sessionId: task.bridgeSessionId, workspaceCwd: task.runtime.workspaceCwd, ...metadata, }); if (task.runtime.provenance === 'live-conversation') { const directory = await this.options.materializeConversationDirectory( - task.summary.sessionId, + task.bridgeSessionId, ); const changed = await task.runtime.bridge.changeSessionCwd( - task.summary.sessionId, + task.bridgeSessionId, { path: directory, allowedRoots: [task.runtime.workspaceCwd], @@ -1132,33 +1137,38 @@ export class LiveTaskService { } private async locateTask(threadId: string): Promise { + const bridgeSessionId = normalizeSessionIdForLookup(threadId); + const storedRuntimes = + bridgeSessionId === threadId + ? undefined + : await this.findStoredTaskRuntimes(threadId); + if (storedRuntimes && storedRuntimes.length > 1) { + throw new Error(`Task id is ambiguous: ${threadId}`); + } const live = - this.options.workspaceRegistry.resolveLiveSessionOwner(threadId); + this.options.workspaceRegistry.resolveLiveSessionOwner(bridgeSessionId); if (live.kind === 'ambiguous') { throw new Error(`Task id is ambiguous: ${threadId}`); } if (live.kind === 'unavailable') { throw conversationRuntimeUnavailableError(); } - const runtimes = - live.kind === 'found' - ? [live.runtime] - : ( - await Promise.all( - ( - this.options.workspaceRegistry.listAll?.() ?? - this.options.workspaceRegistry.list() - ).map(async (runtime) => ({ - runtime, - exists: - await createWorkspaceRuntimeSessionService( - runtime, - ).sessionExists(threadId), - })), - ) - ) - .filter((entry) => entry.exists) - .map((entry) => entry.runtime); + if ( + storedRuntimes?.length === 1 && + live.kind === 'found' && + storedRuntimes[0] !== live.runtime + ) { + throw new Error(`Task id is ambiguous: ${threadId}`); + } + let runtimes = storedRuntimes; + if (runtimes === undefined) { + runtimes = + live.kind === 'found' + ? [live.runtime] + : await this.findStoredTaskRuntimes(threadId); + } else if (runtimes.length === 0 && live.kind === 'found') { + runtimes = [live.runtime]; + } if (runtimes.length === 0) throw new SessionNotFoundError(threadId); if (runtimes.length > 1) throw new Error(`Task id is ambiguous: ${threadId}`); @@ -1167,7 +1177,7 @@ export class LiveTaskService { const persisted = await service.loadSession(threadId); let summary: BridgeSessionSummary; try { - summary = runtime.bridge.getSessionSummary(threadId); + summary = runtime.bridge.getSessionSummary(bridgeSessionId); } catch (error) { if ( !(error instanceof SessionNotFoundError) && @@ -1193,7 +1203,27 @@ export class LiveTaskService { if (!found) throw new SessionNotFoundError(threadId); summary = found; } - return { runtime, persisted, summary }; + return { runtime, threadId, bridgeSessionId, persisted, summary }; + } + + private async findStoredTaskRuntimes( + threadId: string, + ): Promise { + const entries = await Promise.all( + ( + this.options.workspaceRegistry.listAll?.() ?? + this.options.workspaceRegistry.list() + ).map(async (runtime) => ({ + runtime, + exists: + await createWorkspaceRuntimeSessionService(runtime).sessionExists( + threadId, + ), + })), + ); + return entries + .filter((entry) => entry.exists) + .map((entry) => entry.runtime); } }