diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index c969bb21cb2..0144337060f 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -22441,7 +22441,12 @@ describe('createAcpSessionBridge', () => { it('carries persisted source metadata into ACP session restore', async () => { for (const action of ['load', 'resume'] as const) { - const handle = makeChannel(); + const handle = makeChannel({ + extMethodImpl: async (method) => + method === SERVE_CONTROL_EXT_METHODS.sessionSource + ? { persisted: true } + : {}, + }); const bridge = makeBridge({ channelFactory: async () => handle.channel, }); @@ -22452,26 +22457,147 @@ describe('createAcpSessionBridge', () => { sourceId: 'dingtalk-main', }; - if (action === 'load') { - await bridge.loadSession(request); - } else { - await bridge.resumeSession(request); - } + const restored = + action === 'load' + ? await bridge.loadSession(request) + : await bridge.resumeSession(request); const call = action === 'load' ? handle.agent.loadSessionCalls[0] : handle.agent.resumeSessionCalls[0]; + expect(restored).toMatchObject({ + sourceType: 'channel', + sourceId: 'dingtalk-main', + sourcePersisted: true, + }); expect(call?._meta).toMatchObject({ [SESSION_SOURCE_META_KEY]: { sourceType: 'channel', sourceId: 'dingtalk-main', }, }); + expect(handle.agent.extMethodCalls).toContainEqual({ + method: SERVE_CONTROL_EXT_METHODS.sessionSource, + params: { + sessionId: request.sessionId, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }, + }); await bridge.shutdown(); } }); + it('backfills missing source metadata when restoring an existing entry', async () => { + const handle = makeChannel({ + extMethodImpl: async (method) => + method === SERVE_CONTROL_EXT_METHODS.sessionSource + ? { persisted: true } + : {}, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ + workspaceCwd: WS_A, + sessionScope: 'thread', + }); + + const restored = await bridge.resumeSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(restored).toMatchObject({ + sessionId: session.sessionId, + sourceType: 'channel', + sourceId: 'dingtalk-main', + sourcePersisted: true, + }); + expect(bridge.getSessionSummary(session.sessionId)).toMatchObject({ + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + expect(handle.agent.extMethodCalls).toContainEqual({ + method: SERVE_CONTROL_EXT_METHODS.sessionSource, + params: { + sessionId: session.sessionId, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }, + }); + + await bridge.shutdown(); + }); + + it('rejects an existing-entry restore when the channel exits during source persistence', async () => { + const sourceStarted = deferred(); + const sourceRelease = deferred>(); + const handle = makeChannel({ + extMethodImpl: async (method) => { + if (method === SERVE_CONTROL_EXT_METHODS.sessionSource) { + sourceStarted.resolve(); + return sourceRelease.promise; + } + return {}; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ + workspaceCwd: WS_A, + sessionScope: 'thread', + }); + + const restoring = bridge.resumeSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + await sourceStarted.promise; + handle.crash(); + + await expect(restoring).rejects.toThrow(SessionNotFoundError); + sourceRelease.resolve({ persisted: false }); + await bridge.shutdown(); + }); + + it('rejects a cold restore when the channel exits during source persistence', async () => { + const sourceStarted = deferred(); + const sourceRelease = deferred>(); + const handle = makeChannel({ + extMethodImpl: async (method) => { + if (method === SERVE_CONTROL_EXT_METHODS.sessionSource) { + sourceStarted.resolve(); + return sourceRelease.promise; + } + return {}; + }, + resumeSessionImpl: () => ({}), + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + + const restoring = bridge.resumeSession({ + sessionId: 'restore-source-channel-exit', + workspaceCwd: WS_A, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + await sourceStarted.promise; + handle.crash(); + + await expect(restoring).rejects.toThrow(SessionNotFoundError); + sourceRelease.resolve({ persisted: false }); + await bridge.shutdown(); + }); + it('does not replace the source when single scope attaches', async () => { const bridge = makeBridge({ channelFactory: async () => makeChannel().channel, @@ -22497,26 +22623,117 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); - it('passes source metadata in the resume request without a second control call', async () => { + it('does not replace an existing source when restore requests a new one', async () => { + const handle = makeChannel({ + extMethodImpl: async (method) => + method === SERVE_CONTROL_EXT_METHODS.sessionSource + ? { persisted: false } + : {}, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const first = await bridge.spawnOrAttach({ + workspaceCwd: WS_A, + sourceType: 'scheduled_task', + sourceId: 'task-1', + }); + + const restored = await bridge.resumeSession({ + sessionId: first.sessionId, + workspaceCwd: WS_A, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(restored).toMatchObject({ + sessionId: first.sessionId, + sourceType: 'scheduled_task', + sourceId: 'task-1', + }); + expect(bridge.getSessionSummary(first.sessionId)).toMatchObject({ + sourceType: 'scheduled_task', + sourceId: 'task-1', + }); + expect(handle.agent.extMethodCalls).not.toContainEqual({ + method: SERVE_CONTROL_EXT_METHODS.sessionSource, + params: { + sessionId: first.sessionId, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }, + }); + + await bridge.shutdown(); + }); + + it('keeps restore live-only when source persistence fails', async () => { const handle = makeChannel({ + extMethodImpl: async (method) => { + if (method === SERVE_CONTROL_EXT_METHODS.sessionSource) { + throw new Error('persist failed'); + } + return {}; + }, resumeSessionImpl: () => ({}), }); const bridge = makeBridge({ channelFactory: async () => handle.channel, }); - await bridge.resumeSession({ + const restored = await bridge.resumeSession({ + sessionId: 'restore-source-failed', + workspaceCwd: WS_A, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(restored).toMatchObject({ + sessionId: 'restore-source-failed', + sourceType: 'channel', + sourceId: 'dingtalk-main', + sourcePersisted: false, + }); + expect(bridge.getSessionSummary('restore-source-failed')).toMatchObject({ + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + await bridge.shutdown(); + }); + + it('persists source metadata after resume restores a session', async () => { + const handle = makeChannel({ + extMethodImpl: async (method) => + method === SERVE_CONTROL_EXT_METHODS.sessionSource + ? { persisted: true } + : {}, + resumeSessionImpl: () => ({}), + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + + const restored = await bridge.resumeSession({ sessionId: 'persisted-live-session', workspaceCwd: WS_A, sourceType: 'default', sourceId: 'realtime_voice:p1:h1:a1:call-1', }); - expect(handle.agent.extMethodCalls).not.toContainEqual( - expect.objectContaining({ - method: SERVE_CONTROL_EXT_METHODS.sessionSource, - }), - ); + expect(restored).toMatchObject({ + sourceType: 'default', + sourceId: 'realtime_voice:p1:h1:a1:call-1', + sourcePersisted: true, + }); + expect(handle.agent.extMethodCalls).toContainEqual({ + method: SERVE_CONTROL_EXT_METHODS.sessionSource, + params: { + sessionId: 'persisted-live-session', + sourceType: 'default', + sourceId: 'realtime_voice:p1:h1:a1:call-1', + }, + }); expect(handle.agent.resumeSessionCalls[0]?._meta).toEqual({ [SESSION_SOURCE_META_KEY]: { sourceType: 'default', diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 25ae83e8844..92ee4cf1fba 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -4799,38 +4799,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { let sourcePersisted: boolean | undefined; if (entry.sourceType) { - try { - const sourceResult = await Promise.race([ - withTimeout( - entry.connection.extMethod( - SERVE_CONTROL_EXT_METHODS.sessionSource, - { - sessionId: entry.sessionId, - sourceType: entry.sourceType, - ...(entry.sourceId !== undefined - ? { sourceId: entry.sourceId } - : {}), - ...(daemonOwnedStandaloneCreation - ? { [DAEMON_OWNED_STANDALONE_CREATION_KEY]: true } - : {}), - }, - ), - initTimeoutMs, - 'sessionSource', - ), - getTransportClosedReject(entry), - ]); - sourcePersisted = - (sourceResult as { persisted?: boolean } | undefined)?.persisted === - true; - } catch (err) { - sourcePersisted = false; - writeStderrLine( - `qwen serve: source metadata for ${entry.sessionId} was not persisted ` + - `(${err instanceof Error ? err.message : String(err)}) — the source is live-only ` + - `until restart (reported to the caller via sourcePersisted=false)`, - ); - } + sourcePersisted = await persistSessionSource( + entry, + entry.sessionId, + daemonOwnedStandaloneCreation, + ); } // ACP `newSession` doesn't take a model id; honor the caller's @@ -6169,6 +6142,63 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { return entry; }; + async function persistSessionSource( + entry: SessionEntry, + logContext: string, + daemonOwnedStandaloneCreation = false, + ): Promise { + try { + const sourceResult = await Promise.race([ + withTimeout( + entry.connection.extMethod(SERVE_CONTROL_EXT_METHODS.sessionSource, { + sessionId: entry.sessionId, + sourceType: entry.sourceType, + ...(entry.sourceId !== undefined + ? { sourceId: entry.sourceId } + : {}), + ...(daemonOwnedStandaloneCreation + ? { [DAEMON_OWNED_STANDALONE_CREATION_KEY]: true } + : {}), + }), + initTimeoutMs, + 'sessionSource', + ), + getTransportClosedReject(entry), + ]); + return ( + (sourceResult as { persisted?: boolean } | undefined)?.persisted === + true + ); + } catch (err) { + writeStderrLine( + `qwen serve: source metadata for ${logContext} was not persisted ` + + `(${err instanceof Error ? err.message : String(err)}) — the source is live-only ` + + `until restart (reported to the caller via sourcePersisted=false)`, + ); + return false; + } + } + + async function applyRestoreSourceIfMissing( + entry: SessionEntry, + req: BridgeRestoreSessionRequest, + ): Promise { + if (entry.sourceType !== undefined || req.sourceType === undefined) { + return undefined; + } + entry.sourceType = req.sourceType; + if (req.sourceId !== undefined) { + entry.sourceId = req.sourceId; + } else { + delete entry.sourceId; + } + markSessionCatalogChanged(); + return await persistSessionSource( + entry, + `${entry.sessionId} during session restore`, + ); + } + const prepareStandaloneArtifactWorkspace = async ( entry: SessionEntry, ): Promise => { @@ -6866,8 +6896,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { existing.attachCount++; const clientId = registerClient(existing, req.clientId); recordAttachRef(existing, clientId); + let previousApprovalMode: ApprovalMode | undefined; if (req.approvalMode) { - const previousApprovalMode = await applyApprovalModeForAttach( + previousApprovalMode = await applyApprovalModeForAttach( existing, req.approvalMode, clientId, @@ -6884,6 +6915,20 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { throw error; } } + const sourcePersisted = await applyRestoreSourceIfMissing(existing, req); + try { + assertAttachableSessionEntry(req.sessionId, existing); + } catch (error) { + if (previousApprovalMode !== undefined) { + await rollbackApprovalModeForRejectedAttach( + existing, + previousApprovalMode, + clientId, + ); + } + await rollbackAttachRegistration(existing, clientId); + throw error; + } return { sessionId: existing.sessionId, workspaceCwd: existing.workspaceCwd, @@ -6902,6 +6947,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { state: existing.restoreState ?? {}, hasActivePrompt: existing.promptActive || existing.goalTurnActive === true, + ...(sourcePersisted !== undefined ? { sourcePersisted } : {}), ...replayFields, ...(historyAnchorRecordId !== undefined ? { historyAnchorRecordId } @@ -7024,8 +7070,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // This coalescer's attachCount contribution was pre-folded via // `coalesceState.count`, so only the ledger is updated here. recordAttachRef(entry, clientId); + let previousApprovalMode: ApprovalMode | undefined; if (req.approvalMode) { - const previousApprovalMode = await applyApprovalModeForAttach( + previousApprovalMode = await applyApprovalModeForAttach( entry, req.approvalMode, clientId, @@ -7042,12 +7089,29 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { throw error; } } + const sourcePersisted = await applyRestoreSourceIfMissing(entry, req); + try { + assertAttachableSessionEntry(restored.sessionId, entry); + } catch (error) { + if (previousApprovalMode !== undefined) { + await rollbackApprovalModeForRejectedAttach( + entry, + previousApprovalMode, + clientId, + ); + } + await rollbackAttachRegistration(entry, clientId); + throw error; + } return { ...restored, attached: true, clientId, createdAt: entry.createdAt, hasActivePrompt: entry.promptActive || entry.goalTurnActive === true, + ...(entry.sourceType ? { sourceType: entry.sourceType } : {}), + ...(entry.sourceId !== undefined ? { sourceId: entry.sourceId } : {}), + ...(sourcePersisted !== undefined ? { sourcePersisted } : {}), ...(waiterReplayFields ?? {}), }; } @@ -7472,8 +7536,8 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { racedEntry.attachCount += 1 + coalesceState.count; const clientId = registerClient(racedEntry, req.clientId); recordAttachRef(racedEntry, clientId); + let previousApprovalMode: ApprovalMode | undefined; if (req.approvalMode) { - let previousApprovalMode: ApprovalMode; try { const result = await applyApprovalMode( racedEntry, @@ -7513,6 +7577,27 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { clientId, options, ); + const sourcePersisted = await applyRestoreSourceIfMissing( + racedEntry, + req, + ); + try { + assertAttachableSessionEntry(req.sessionId, racedEntry); + } catch (error) { + if (previousApprovalMode !== undefined) { + await rollbackApprovalModeForRejectedAttach( + racedEntry, + previousApprovalMode, + clientId, + ); + } + await rollbackAttachRegistration( + racedEntry, + clientId, + 1 + coalesceState.count, + ); + throw error; + } return { sessionId: racedEntry.sessionId, workspaceCwd: racedEntry.workspaceCwd, @@ -7533,6 +7618,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { restorePromptAdmitted || racedEntry.promptActive || racedEntry.goalTurnActive === true, + ...(sourcePersisted !== undefined ? { sourcePersisted } : {}), ...replayFieldsFor(racedEntry, action, liveReplayMode), }; } @@ -7627,8 +7713,13 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { } assertAttachableSessionEntry(req.sessionId, entry); const clientId = registerClient(entry, req.clientId); + let previousApprovalMode: ApprovalMode | undefined; if (req.approvalMode) { - await applyApprovalModeForAttach(entry, req.approvalMode, clientId); + previousApprovalMode = await applyApprovalModeForAttach( + entry, + req.approvalMode, + clientId, + ); assertAttachableSessionEntry(req.sessionId, entry); } // Fold synchronous coalesce reservations into the new entry's @@ -7646,6 +7737,26 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { clientId, options, ); + const sourcePersisted = entry.sourceType + ? await persistSessionSource( + entry, + `${entry.sessionId} during session restore`, + daemonOwnedStandaloneRestore, + ) + : undefined; + try { + assertAttachableSessionEntry(req.sessionId, entry); + } catch (error) { + if (previousApprovalMode !== undefined) { + await rollbackApprovalModeForRejectedAttach( + entry, + previousApprovalMode, + clientId, + ); + } + await rollbackAttachRegistration(entry, clientId); + throw error; + } // Explicit `session/load` / `session/resume` is "give me THIS // id"; it must NOT become the implicit attach target for // subsequent omitted-id `POST /session` callers under `single` @@ -7661,6 +7772,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { createdAt: entry.createdAt, ...(entry.sourceType ? { sourceType: entry.sourceType } : {}), ...(entry.sourceId !== undefined ? { sourceId: entry.sourceId } : {}), + ...(sourcePersisted !== undefined ? { sourcePersisted } : {}), state: publicState, ...(deferArtifactWorkspace || artifactRestoreWarnings.length > 0 ? { artifactWarnings: artifactRestoreWarnings } diff --git a/packages/channels/base/src/ChannelAgentBridge.ts b/packages/channels/base/src/ChannelAgentBridge.ts index ce915220849..ec586cf494f 100644 --- a/packages/channels/base/src/ChannelAgentBridge.ts +++ b/packages/channels/base/src/ChannelAgentBridge.ts @@ -102,9 +102,8 @@ export interface ChannelAgentBridgeSessionOptions { enableChannelLoops?: boolean; /** * Channel instance name (e.g. `feishu-main`) stamped as the daemon `sourceId` - * on **new** sessions — creation-time attribution paired with - * `sourceType: 'channel'`. Ignored by `loadSession`: loading an existing - * session never re-stamps its creation attribution. + * for new sessions and restore-time attribution for legacy sessions resumed + * through a channel. */ sourceId?: string; } diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index 1af66e2ecce..4bdf3416caf 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -3580,7 +3580,7 @@ describe('ChannelBase', () => { expect(recoveredBridge.loadSession).toHaveBeenCalledWith( 's-2', '/tmp', - undefined, + { sourceId: 'test-chan' }, expect.anything(), ), ); @@ -4142,7 +4142,7 @@ describe('ChannelBase', () => { expect(bridge.loadSession).toHaveBeenCalledWith( 's-1', '/tmp', - undefined, + { sourceId: 'test-chan' }, expect.anything(), ); diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index ec6f43ebbbb..69f18a42b95 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -662,7 +662,7 @@ describe('DaemonChannelBridge', () => { bridge.stop(); }); - it('forwards sourceId to the session factory only for new sessions', async () => { + it('forwards source IDs to the session factory for new and loaded sessions', async () => { const events = new EventQueue(); const session = createFakeSession(events); const factory = vi.fn().mockResolvedValue(session); @@ -672,22 +672,25 @@ describe('DaemonChannelBridge', () => { }); await bridge.start(); - await bridge.newSession('/repo', { sourceId: 'feishu-main' }); - await bridge.loadSession('session-1', '/repo', { sourceId: 'feishu-main' }); + await bridge.newSession('/repo', { + sourceId: 'feishu-main', + }); + await bridge.loadSession('session-1', '/repo', { + sourceId: 'feishu-main', + }); - // New session: sourceId is forwarded to the factory (creation attribution); expect(factory).toHaveBeenNthCalledWith(1, { workspaceCwd: '/repo', modelServiceId: undefined, sessionScope: 'thread', sourceId: 'feishu-main', }); - // Load: never forwarded even when provided — loads never re-stamp creation source. expect(factory).toHaveBeenNthCalledWith(2, { workspaceCwd: '/repo', modelServiceId: undefined, sessionId: 'session-1', sessionScope: 'thread', + sourceId: 'feishu-main', }); events.close(); diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index 4c0ef587354..db2e12fa5f2 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -76,7 +76,7 @@ export interface DaemonChannelSessionFactoryRequest { sessionId?: string; sessionScope?: SessionScope; approvalMode?: string; - /** Channel instance name stamped as daemon `sourceId` (new sessions only). */ + /** Channel instance name stamped as daemon `sourceId`. */ sourceId?: string; } @@ -416,6 +416,7 @@ export class DaemonChannelBridge sessionId, sessionScope: this.options.sessionScope ?? 'thread', ...(options?.approvalMode ? { approvalMode: options.approvalMode } : {}), + ...(options?.sourceId ? { sourceId: options.sourceId } : {}), }); if (lifecycleGeneration !== this.lifecycleGeneration) { await this.rejectStaleSession(session); diff --git a/packages/channels/base/src/SessionRouter.test.ts b/packages/channels/base/src/SessionRouter.test.ts index b4a480ad640..b4e79b108e8 100644 --- a/packages/channels/base/src/SessionRouter.test.ts +++ b/packages/channels/base/src/SessionRouter.test.ts @@ -627,7 +627,7 @@ describe('SessionRouter', () => { expect(bridge.loadSession).toHaveBeenCalledWith( 'dormant-session', '/tmp', - { enableChannelLoops: false }, + { enableChannelLoops: false, sourceId: 'ch' }, expect.anything(), ); }); @@ -708,7 +708,7 @@ describe('SessionRouter', () => { expect(replacementBridge.loadSession).toHaveBeenCalledWith( second, '/tmp', - undefined, + { sourceId: 'ch' }, expect.anything(), ); }); @@ -999,7 +999,27 @@ describe('SessionRouter', () => { expect(bridge.loadSession).toHaveBeenCalledWith( 'old-session', '/tmp', - { approvalMode: 'yolo' }, + { approvalMode: 'yolo', sourceId: 'ch' }, + expect.any(Object), + ); + }); + + it('stamps channel name as sourceId when restoring sessions', async () => { + const dir = mkdtempSync(join(tmpdir(), 'qwen-router-')); + tempDirs.push(dir); + const persistPath = join(dir, 'sessions.json'); + writePersistedSession(persistPath); + const router = new SessionRouter(bridge, '/tmp', 'user', persistPath); + + await expect(router.restoreSessions()).resolves.toEqual({ + restored: 1, + failed: 0, + }); + + expect(bridge.loadSession).toHaveBeenCalledWith( + 'old-session', + '/tmp', + { sourceId: 'ch' }, expect.any(Object), ); }); @@ -1168,7 +1188,7 @@ describe('SessionRouter', () => { expect(restartedBridge.loadSession).toHaveBeenCalledWith( aliceSession, '/tmp', - undefined, + { sourceId: 'ch' }, expect.any(Object), ); expect(router.getSession('ch', 'alice', 'chat1')).toBe(aliceSession); diff --git a/packages/channels/base/src/SessionRouter.ts b/packages/channels/base/src/SessionRouter.ts index 9ec20ebf053..51ee05f1b94 100644 --- a/packages/channels/base/src/SessionRouter.ts +++ b/packages/channels/base/src/SessionRouter.ts @@ -135,15 +135,14 @@ export class SessionRouter { private sessionOptions( channelName: string, - ): ChannelAgentBridgeSessionOptions | undefined { + ): ChannelAgentBridgeSessionOptions { const approvalMode = this.channelApprovalModes.get(channelName); const loopsDisabled = this.channelsWithoutLoops.has(channelName); - return approvalMode || loopsDisabled - ? { - ...(approvalMode ? { approvalMode } : {}), - ...(loopsDisabled ? { enableChannelLoops: false } : {}), - } - : undefined; + return { + ...(approvalMode ? { approvalMode } : {}), + ...(loopsDisabled ? { enableChannelLoops: false } : {}), + sourceId: channelName, + }; } async resolve( @@ -261,7 +260,6 @@ export class SessionRouter { key, this.sessionOptions(input.channelName), operation, - input.channelName, ); try { this.assertOperationCurrent(operation); @@ -345,7 +343,6 @@ export class SessionRouter { key, this.sessionOptions(input.channelName), operation, - input.channelName, ); try { this.assertOperationCurrent(operation); @@ -979,18 +976,13 @@ export class SessionRouter { cwd: string, loadWindow: SessionLoadWindow, routingKey: string, - options: { approvalMode?: string } | undefined, + options: ChannelAgentBridgeSessionOptions, operation: SessionOperation, - sourceId: string, ): Promise { const maxAttempts = 2; let lastDeadSessionId: string | undefined; for (let attempt = 0; attempt < maxAttempts; attempt++) { - const sessionId = await this.bridge.newSession( - cwd, - { ...options, sourceId }, - operation, - ); + const sessionId = await this.bridge.newSession(cwd, options, operation); try { this.assertOperationCurrent(operation); } catch (error) { diff --git a/packages/channels/base/src/named-session-manager.test.ts b/packages/channels/base/src/named-session-manager.test.ts index 9c0739fbb0d..02ad4ef39aa 100644 --- a/packages/channels/base/src/named-session-manager.test.ts +++ b/packages/channels/base/src/named-session-manager.test.ts @@ -527,7 +527,7 @@ describe('NamedSessionManager', () => { expect(bridge.loadSession).toHaveBeenCalledWith( review.sessionId, '/workspace', - undefined, + { sourceId: 'channel-a' }, expect.anything(), ); diff --git a/packages/cli/src/commands/channel/daemon-worker.test.ts b/packages/cli/src/commands/channel/daemon-worker.test.ts index 546884fcc96..7871cbef030 100644 --- a/packages/cli/src/commands/channel/daemon-worker.test.ts +++ b/packages/cli/src/commands/channel/daemon-worker.test.ts @@ -417,7 +417,7 @@ function stubProcessSend(send: NodeJS.Process['send'] | undefined): () => void { } describe('createDaemonSessionFactory', () => { - it('tags created channel sessions without changing loaded sessions', async () => { + it('tags created and resumed channel sessions', async () => { const sdk = createSdk(); const factory = createDaemonSessionFactory({ client: sdk.client, @@ -449,6 +449,7 @@ describe('createDaemonSessionFactory', () => { workspaceCwd: '/workspace', modelServiceId: 'qwen-plus', sessionScope: 'thread', + sourceType: 'channel', }, 'qwen-channel-worker', ); @@ -489,12 +490,13 @@ describe('createDaemonSessionFactory', () => { workspaceCwd: '/workspace', approvalMode: 'yolo', sessionScope: 'thread', + sourceType: 'channel', }, 'qwen-channel-worker', ); }); - it('stamps channel sourceId on created sessions only', async () => { + it('stamps channel sourceId on created and resumed sessions', async () => { const sdk = createSdk(); const factory = createDaemonSessionFactory({ client: sdk.client, @@ -519,14 +521,14 @@ describe('createDaemonSessionFactory', () => { }, 'qwen-channel-worker', ); - // The load branch never re-stamps creation attribution: no sourceId in the - // load request even when the factory request carried one. expect(sdk.DaemonSessionClient.resume).toHaveBeenCalledWith( sdk.client, 'existing-session', { workspaceCwd: '/workspace', sessionScope: 'thread', + sourceType: 'channel', + sourceId: 'dingtalk-main', }, 'qwen-channel-worker', ); diff --git a/packages/cli/src/commands/channel/daemon-worker.ts b/packages/cli/src/commands/channel/daemon-worker.ts index 794cc6164d7..bad80489543 100644 --- a/packages/cli/src/commands/channel/daemon-worker.ts +++ b/packages/cli/src/commands/channel/daemon-worker.ts @@ -146,6 +146,8 @@ interface DaemonSessionClientStaticLike { modelServiceId?: string; sessionScope: 'thread'; approvalMode?: string; + sourceType?: string; + sourceId?: string; }, clientId?: string, ): Promise; @@ -209,6 +211,11 @@ export function createDaemonSessionFactory({ // sessions remain thread-scoped so different channels never share the // daemon's default single session. sessionScope: 'thread' as const, + sourceType: 'channel', + // sourceId = channel instance name (e.g. feishu-main): distinguishes + // channel instances on the daemon data plane; the channel kind + // (dingtalk/feishu) is derivable from the name via the channel config. + ...(req.sourceId ? { sourceId: req.sourceId } : {}), }; if (req.sessionId) { return await DaemonSessionClient.resume( @@ -220,16 +227,7 @@ export function createDaemonSessionFactory({ } return await DaemonSessionClient.createOrAttach( client, - { - ...daemonReq, - sourceType: 'channel', - // sourceId = channel instance name (e.g. feishu-main): distinguishes - // channel instances on the daemon data plane; the channel kind - // (dingtalk/feishu) is derivable from the name via the channel config. - // The load branch above deliberately omits it: loading never re-stamps - // creation attribution. - ...(req.sourceId ? { sourceId: req.sourceId } : {}), - }, + daemonReq, clientId, ); }; diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index 3ff9d3e2e15..f0af68d339e 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -595,6 +595,42 @@ function parseOptionalApprovalMode( return rawApprovalMode as ApprovalMode; } +function parseRequestedSessionSource( + body: Record, + res: Response, +): { sourceType?: string; sourceId?: string } | null { + if ( + isReservedStandaloneSessionSource({ + sourceType: + typeof body['sourceType'] === 'string' ? body['sourceType'] : undefined, + }) + ) { + res.status(400).json({ + error: + 'The requested session source is reserved for daemon-owned standalone sessions.', + code: 'reserved_session_source', + }); + return null; + } + const source = parseSessionSource(body['sourceType'], body['sourceId']); + if ('error' in source) { + res.status(400).json({ + error: source.error, + code: 'invalid_session_source', + }); + return null; + } + if (isReservedLiveSessionSource(source)) { + res.status(400).json({ + error: + 'The requested session source is reserved for daemon-owned Live Voice sessions.', + code: 'reserved_session_source', + }); + return null; + } + return source; +} + export function registerSessionRoutes( app: Application, deps: RegisterSessionRoutesDeps, @@ -2669,37 +2705,8 @@ export function registerSessionRoutes( } const approvalMode = parseOptionalApprovalMode(body, res); if (approvalMode === null) return; - if ( - isReservedStandaloneSessionSource({ - sourceType: - typeof body['sourceType'] === 'string' - ? body['sourceType'] - : undefined, - }) - ) { - res.status(400).json({ - error: - 'The requested session source is reserved for daemon-owned standalone sessions.', - code: 'reserved_session_source', - }); - return; - } - const source = parseSessionSource(body['sourceType'], body['sourceId']); - if ('error' in source) { - res.status(400).json({ - error: source.error, - code: 'invalid_session_source', - }); - return; - } - if (isReservedLiveSessionSource(source)) { - res.status(400).json({ - error: - 'The requested session source is reserved for daemon-owned Live Voice sessions.', - code: 'reserved_session_source', - }); - return; - } + const source = parseRequestedSessionSource(body, res); + if (source === null) return; const clientId = parseClientIdHeader(req, res); if (clientId === null) return; @@ -3431,6 +3438,8 @@ export function registerSessionRoutes( if (historyPageSize === null) return; const liveReplayMode = parseLiveReplayMode(body ?? {}, res); if (liveReplayMode === null) return; + const restoreSource = parseRequestedSessionSource(body, res); + if (restoreSource === null) return; const clientId = parseClientIdHeader(req, res); if (clientId === null) return; if ( @@ -3610,6 +3619,15 @@ export function registerSessionRoutes( isReservedStandaloneSessionSource(metadata) ? metadataWithoutSource : metadata; + const hasPersistedSource = + 'sourceType' in restoreMetadata && + restoreMetadata.sourceType !== undefined; + const restoreRequestMetadata = { + ...restoreMetadata, + ...(hasPersistedSource || isInternalWorkspaceRuntime(runtime) + ? {} + : restoreSource), + }; assertRuntimeGenerationOpen?.(); if (isInternalWorkspaceRuntime(runtime)) { sessionIdReservation = requestedSessionIdAdmission.reserveRestore( @@ -3647,14 +3665,14 @@ export function registerSessionRoutes( ...(liveReplayMode !== undefined ? { liveReplayMode } : {}), ...(clientId !== undefined ? { clientId } : {}), ...(approvalMode !== undefined ? { approvalMode } : {}), - ...restoreMetadata, + ...restoreRequestMetadata, }) : await runtime.bridge.resumeSession({ sessionId, workspaceCwd, ...(clientId !== undefined ? { clientId } : {}), ...(approvalMode !== undefined ? { approvalMode } : {}), - ...restoreMetadata, + ...restoreRequestMetadata, }); // Every path that can register a Live entry relocates it before a // prompt can start. Re-queuing cd for an active entry would block diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index c81b09d3a51..8ddab1829bd 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -14237,6 +14237,157 @@ describe('createServeApp', () => { }); }); + it('forwards resume-time source metadata to the bridge', async () => { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const res = await request(app) + .post('/session/persisted-channel/resume') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(res.status).toBe(200); + expect(bridge.resumeCalls[0]).toMatchObject({ + sessionId: 'persisted-channel', + workspaceCwd: WS_BOUND, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + }); + + it('forwards load-time source metadata to the bridge', async () => { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const res = await request(app) + .post('/session/persisted-channel/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(res.status).toBe(200); + expect(bridge.loadCalls[0]).toMatchObject({ + sessionId: 'persisted-channel', + workspaceCwd: WS_BOUND, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + }); + + it('does not let restore-time source metadata replace persisted attribution', async () => { + const bridge = fakeBridge(); + const readCreationMetadata = vi + .spyOn(SessionService.prototype, 'readCreationMetadata') + .mockResolvedValue({ sourceType: 'channel', sourceId: 'feishu-main' }); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + try { + const res = await request(app) + .post('/session/persisted-channel/resume') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(res.status).toBe(200); + expect(bridge.resumeCalls[0]).toMatchObject({ + sessionId: 'persisted-channel', + workspaceCwd: WS_BOUND, + sourceType: 'channel', + sourceId: 'feishu-main', + }); + } finally { + readCreationMetadata.mockRestore(); + } + }); + + it.each(['load', 'resume'] as const)( + 'rejects reserved standalone source metadata on %s', + async (action) => { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const res = await request(app) + .post(`/session/persisted-channel/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ sourceType: 'standalone' }); + + expect(res.status).toBe(400); + expect(res.body.code).toBe('reserved_session_source'); + expect(bridge.loadCalls).toHaveLength(0); + expect(bridge.resumeCalls).toHaveLength(0); + }, + ); + + it.each(['load', 'resume'] as const)( + 'rejects sourceId without sourceType on %s', + async (action) => { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const res = await request(app) + .post(`/session/persisted-channel/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ sourceId: 'dingtalk-main' }); + + expect(res.status).toBe(400); + expect(res.body.code).toBe('invalid_session_source'); + expect(bridge.loadCalls).toHaveLength(0); + expect(bridge.resumeCalls).toHaveLength(0); + }, + ); + + it.each(['load', 'resume'] as const)( + 'rejects daemon-owned live source metadata on %s', + async (action) => { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const res = await request(app) + .post(`/session/persisted-channel/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + sourceType: 'default', + sourceId: 'realtime_voice:p1:h1:a1:forged', + }); + + expect(res.status).toBe(400); + expect(res.body.code).toBe('reserved_session_source'); + expect(bridge.loadCalls).toHaveLength(0); + expect(bridge.resumeCalls).toHaveLength(0); + }, + ); + // The restore handler's `!res.writable` cleanup branch (kill on // !attached, detach on attached) is line-for-line identical to // the matching branch on `POST /session`; routing-side @@ -19378,6 +19529,64 @@ describe('createServeApp', () => { ]); }); + it('filters a restored legacy session by live source metadata', async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440301'; + await writeStoredSession({ + sessionId, + cwd: WS_BOUND, + timestamp: '2026-05-17T12:00:00.000Z', + prompt: 'legacy channel session', + mtime: new Date('2026-05-17T12:00:00.000Z'), + }); + const bridge = fakeBridge({ + listImpl: () => [ + { + sessionId, + workspaceCwd: WS_BOUND, + createdAt: '2026-05-17T12:00:00.000Z', + updatedAt: '2026-05-17T12:03:00.000Z', + clientCount: 1, + hasActivePrompt: false, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }, + ], + }); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge, boundWorkspace: WS_BOUND }, + ); + + const filtered = await request(app) + .get( + `/workspace/${encodeURIComponent(WS_BOUND)}/sessions?sourceType=channel&sourceId=dingtalk-main`, + ) + .set('Host', `127.0.0.1:${baseOpts.port}`); + expect(filtered.status).toBe(200); + expect(filtered.body.sessions).toEqual([ + expect.objectContaining({ + sessionId, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }), + ]); + + const organized = await request(app) + .get( + `/workspace/${encodeURIComponent(WS_BOUND)}/sessions?view=organized&group=all&sourceType=channel&sourceId=dingtalk-main`, + ) + .set('Host', `127.0.0.1:${baseOpts.port}`); + expect(organized.status).toBe(200); + expect(organized.body.sessions).toEqual([ + expect.objectContaining({ + sessionId, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }), + ]); + }); + it('rejects sourceId without sourceType', async () => { const app = createServeApp( { ...baseOpts, workspace: WS_BOUND }, @@ -35520,6 +35729,54 @@ describe('Live conversation runtime lifecycle', () => { } }); + it('does not apply caller-supplied restore source metadata on the internal Live runtime', async () => { + const sessionId = 'legacy-live-channel'; + const setup = setupLiveRuntime(); + setup.registry.add(setup.liveRuntime); + const findSessionId = vi + .spyOn(SessionService.prototype, 'findSessionIdIgnoringCase') + .mockResolvedValue(sessionId); + const getLocation = vi + .spyOn(SessionService.prototype, 'getSessionLocation') + .mockResolvedValue('active'); + const readCreationMetadata = vi + .spyOn(SessionService.prototype, 'readCreationMetadata') + .mockResolvedValue({}); + const readCreationMetadataIfReadable = vi + .spyOn(SessionService.prototype, 'readCreationMetadataIfReadable') + .mockImplementation(async (id, _location) => readCreationMetadata(id)); + try { + const response = await request(setup.app) + .post(`/session/${sessionId}/resume`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + cwd: setup.root.canonicalRoot, + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(response.status).toBe(200); + expect(setup.liveBridge.resumeCalls).toContainEqual( + expect.objectContaining({ + sessionId, + workspaceCwd: setup.root.canonicalRoot, + }), + ); + expect(setup.liveBridge.resumeCalls[0]).not.toHaveProperty( + 'sourceType', + ); + expect(setup.liveBridge.resumeCalls[0]).not.toHaveProperty('sourceId'); + } finally { + readCreationMetadataIfReadable.mockRestore(); + readCreationMetadata.mockRestore(); + getLocation.mockRestore(); + findSessionId.mockRestore(); + await ( + setup.app.locals['sealAndWaitLiveCoordinator'] as () => Promise + )(); + } + }); + it('adopts a UUID legacy Conversations restore through the standalone service', async () => { const sessionId = '550e8400-e29b-41d4-a716-446655440010'; let restored = false; diff --git a/packages/core/src/services/chatRecordingService.test.ts b/packages/core/src/services/chatRecordingService.test.ts index 9240f2843de..69c073b6ffc 100644 --- a/packages/core/src/services/chatRecordingService.test.ts +++ b/packages/core/src/services/chatRecordingService.test.ts @@ -2873,6 +2873,47 @@ describe('ChatRecordingService', () => { }); describe('legacy recorder', () => { + it('reanchors session source after more than the tail window is appended', async () => { + await expect( + chatRecordingService.recordSessionSource('channel', 'channel-main'), + ).resolves.toBe(true); + + chatRecordingService.recordUserMessage([{ text: 'x'.repeat(65 * 1024) }]); + await chatRecordingService.flush(); + + const sourceRecords = vi + .mocked(mockLease.appendJsonLine) + .mock.calls.map((call) => call[0] as ChatRecord) + .filter((record) => record.subtype === 'session_source'); + expect(sourceRecords).toHaveLength(2); + expect(sourceRecords.at(-1)?.systemPayload).toEqual({ + sourceType: 'channel', + sourceId: 'channel-main', + }); + }); + + it('reanchors a restored session source on the next append', async () => { + const service = new ChatRecordingService(mockConfig); + service.activate(mockLease, undefined, undefined, { + lastCompletedUuid: 'projected-leaf', + turnParentUuids: [null], + sourceType: 'channel', + sourceId: 'channel-main', + }); + + service.recordUserMessage([{ text: 'next' }]); + await service.flush(); + + const sourceRecord = vi + .mocked(mockLease.appendJsonLine) + .mock.calls.map((call) => call[0] as ChatRecord) + .find((record) => record.subtype === 'session_source'); + expect(sourceRecord?.systemPayload).toEqual({ + sourceType: 'channel', + sourceId: 'channel-main', + }); + }); + it('restores reduced recorder state without the full conversation', async () => { const service = new ChatRecordingService(mockConfig, undefined, false, { lastCompletedUuid: 'projected-leaf', diff --git a/packages/core/src/services/chatRecordingService.ts b/packages/core/src/services/chatRecordingService.ts index ebab89473f7..b984a93ea50 100644 --- a/packages/core/src/services/chatRecordingService.ts +++ b/packages/core/src/services/chatRecordingService.ts @@ -78,15 +78,11 @@ const SESSION_FILE_DIFF_CHAR_LIMIT = 50_000; const SESSION_FILE_CONTENT_CHAR_LIMIT = 16_000; /** - * Re-append a fresh `custom_title` record to EOF once this many bytes - * of other JSONL content have been written since the last title - * anchor. Half of the picker's 64KB tail-read window so that even an - * oversized record landing right at the threshold keeps the title - * within scan range. Lifting this above 64KB would let the title - * fall out of the tail window between re-anchors; lowering it - * trades extra writes for a tighter safety margin. + * Re-append tail-readable metadata to EOF once this many bytes of other JSONL + * content have been written since its last anchor. Half of the reader's 64KB + * tail window leaves room for the anchor record itself. */ -const TITLE_REANCHOR_BYTES = 32 * 1024; +const METADATA_REANCHOR_BYTES = 32 * 1024; function isFileDiffDisplay(resultDisplay: unknown): resultDisplay is FileDiff { if ( @@ -993,6 +989,8 @@ export class ChatRecordingService { private pendingExplicitTitleWrites = 0; /** Title writes whose durable result and final cached value are unresolved. */ private pendingTitleWrites = 0; + /** Source writes whose durable result and cached value are unresolved. */ + private pendingSourceWrites = 0; /** * JSON-serialized form of the most recent attribution snapshot accepted for @@ -1022,6 +1020,7 @@ export class ChatRecordingService { */ private bytesSinceTitleAnchor = 0; private hasNonTitleContentSinceTitleAnchor = false; + private bytesSinceSourceAnchor = 0; constructor( config: Config, @@ -1188,7 +1187,10 @@ export class ChatRecordingService { this.currentTitleSource = persistedTitleInfo.source; } if (this.currentCustomTitle) { - this.bytesSinceTitleAnchor = TITLE_REANCHOR_BYTES; + this.bytesSinceTitleAnchor = METADATA_REANCHOR_BYTES; + } + if (this.currentSourceType) { + this.bytesSinceSourceAnchor = METADATA_REANCHOR_BYTES; } } @@ -1213,7 +1215,10 @@ export class ChatRecordingService { ? normalizeSessionModelPayload(state.sessionModel) : undefined; if (this.currentCustomTitle) { - this.bytesSinceTitleAnchor = TITLE_REANCHOR_BYTES; + this.bytesSinceTitleAnchor = METADATA_REANCHOR_BYTES; + } + if (this.currentSourceType) { + this.bytesSinceSourceAnchor = METADATA_REANCHOR_BYTES; } } @@ -1398,7 +1403,7 @@ export class ChatRecordingService { this.updateActiveBranch(record); } this.enqueueRecordWrite(record, legacyConversationFile, updateActiveTail); - this.updateTitleAnchorTracking(record); + this.updateMetadataAnchorTracking(record); } private async appendRecordStrict( @@ -1436,7 +1441,7 @@ export class ChatRecordingService { // Keep anchor accounting in logical queue order, matching appendRecord. // Once accepted, a failed write permanently stops this recorder, so no // rollback of this bookkeeping is needed on rejection. - this.updateTitleAnchorTracking(record); + this.updateMetadataAnchorTracking(record); await pendingWrite; } @@ -1461,17 +1466,13 @@ export class ChatRecordingService { } /** - * Maintain the "title is always in the tail window" invariant by - * counting bytes accepted since the last `custom_title` record and - * re-anchoring once enough non-title content has been written. + * Keep title and source metadata inside the reader's tail window by + * counting bytes accepted since each metadata record and re-anchoring + * independently before either can drift beyond that window. * - * - A `custom_title` record IS the new anchor — reset the counter. - * - Without a current or pending title, the counter is irrelevant. - * - Otherwise accumulate this record's serialized size; if the - * running total breaches the threshold, re-append a fresh - * `custom_title` to EOF. The recursive `appendRecord` call will - * land this branch's first arm (subtype === 'custom_title') and - * reset the counter to 0. + * Each metadata subtype resets only its own counter. Otherwise, active or + * pending metadata accumulates the serialized record size and is re-appended + * to EOF once its threshold is reached. * * Size estimate uses `JSON.stringify` for parity with the actual * write path (`jsonl.writeLine` serializes the same way). It's an @@ -1485,14 +1486,25 @@ export class ChatRecordingService { * actual on-disk distance from the last anchor blow past the 64KB * tail window before the threshold fires. */ - private updateTitleAnchorTracking(record: ChatRecord): void { - if (record.type === 'system' && record.subtype === 'custom_title') { + private updateMetadataAnchorTracking(record: ChatRecord): void { + const isTitleAnchor = + record.type === 'system' && record.subtype === 'custom_title'; + const isSourceAnchor = + record.type === 'system' && record.subtype === 'session_source'; + if (isTitleAnchor) { this.bytesSinceTitleAnchor = 0; this.hasNonTitleContentSinceTitleAnchor = false; - return; } - if (!this.currentCustomTitle && this.pendingTitleWrites === 0) return; - this.hasNonTitleContentSinceTitleAnchor = true; + if (isSourceAnchor) { + this.bytesSinceSourceAnchor = 0; + } + const trackTitle = + !isTitleAnchor && + (this.currentCustomTitle !== undefined || this.pendingTitleWrites > 0); + const trackSource = + !isSourceAnchor && + (this.currentSourceType !== undefined || this.pendingSourceWrites > 0); + if (!trackTitle && !trackSource) return; let serializedRecord: string; try { serializedRecord = JSON.stringify(record); @@ -1501,15 +1513,26 @@ export class ChatRecordingService { // The real serializer will surface the failure through writeChain. return; } - // +1 for the trailing newline jsonl.writeLine appends. - this.bytesSinceTitleAnchor += - Buffer.byteLength(serializedRecord, 'utf8') + 1; + const bytes = Buffer.byteLength(serializedRecord, 'utf8') + 1; + if (trackTitle) { + this.hasNonTitleContentSinceTitleAnchor = true; + this.bytesSinceTitleAnchor += bytes; + } + if (trackSource) { + this.bytesSinceSourceAnchor += bytes; + } if ( - this.bytesSinceTitleAnchor >= TITLE_REANCHOR_BYTES && + this.bytesSinceTitleAnchor >= METADATA_REANCHOR_BYTES && this.pendingTitleWrites === 0 ) { this.reanchorTitle(); } + if ( + this.bytesSinceSourceAnchor >= METADATA_REANCHOR_BYTES && + this.pendingSourceWrites === 0 + ) { + this.reanchorSessionSource(); + } } /** @@ -1520,7 +1543,13 @@ export class ChatRecordingService { * scanning the middle of the file. */ private reanchorTitle(): void { - if (!this.currentCustomTitle) return; + if ( + !this.currentCustomTitle || + this.bytesSinceTitleAnchor < METADATA_REANCHOR_BYTES + ) { + return; + } + this.bytesSinceTitleAnchor = 0; try { const record: ChatRecord = { ...this.createBaseRecord('system'), @@ -1545,6 +1574,32 @@ export class ChatRecordingService { } } + private reanchorSessionSource(): void { + if ( + !this.currentSourceType || + this.bytesSinceSourceAnchor < METADATA_REANCHOR_BYTES + ) { + return; + } + this.bytesSinceSourceAnchor = 0; + try { + const record: ChatRecord = { + ...this.createBaseRecord('system'), + type: 'system', + subtype: 'session_source', + systemPayload: { + sourceType: this.currentSourceType, + ...(this.currentSourceId !== undefined + ? { sourceId: this.currentSourceId } + : {}), + }, + }; + this.appendRecord(record, { updateActiveTail: false }); + } catch (error) { + debugLogger.error('Error re-anchoring session source:', error); + } + } + /** * Awaits all queued async writes. Call before process exit / session * teardown to ensure no records are dropped. @@ -2566,7 +2621,7 @@ export class ChatRecordingService { if ( persisted && this.pendingTitleWrites === 0 && - this.bytesSinceTitleAnchor >= TITLE_REANCHOR_BYTES && + this.bytesSinceTitleAnchor >= METADATA_REANCHOR_BYTES && !this.writeFailure ) { this.reanchorTitle(); @@ -2621,6 +2676,8 @@ export class ChatRecordingService { this.currentSourceId === sourceId ); } + this.pendingSourceWrites++; + let persisted = false; try { const record: ChatRecord = { ...this.createBaseRecord('system'), @@ -2634,12 +2691,23 @@ export class ChatRecordingService { await this.appendRecordStrict(record); this.currentSourceType = sourceType; this.currentSourceId = sourceId; + persisted = true; return true; } catch (error) { if (error !== this.writeFailure) { debugLogger.error('Error saving session source:', error); } return false; + } finally { + this.pendingSourceWrites--; + if ( + persisted && + this.pendingSourceWrites === 0 && + this.bytesSinceSourceAnchor >= METADATA_REANCHOR_BYTES && + !this.writeFailure + ) { + this.reanchorSessionSource(); + } } } diff --git a/packages/core/src/services/sessionService.test.ts b/packages/core/src/services/sessionService.test.ts index 5c902a1d95d..56ba7b67e9d 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -7024,6 +7024,32 @@ describe('SessionService', () => { }); }); + it('rehydrates source metadata appended after the head scan window', async () => { + const sessionId = '78777777-7777-4777-8777-777777777777'; + const lines: Array> = [ + userLine(sessionId, 'hello'), + ]; + for (let i = 0; i < 11; i++) { + lines.push({ + ...userLine(sessionId, `filler-${i}`), + uuid: `filler-${i}`, + }); + } + lines.push(sessionSourceLine(sessionId)); + writeSession(sessionId, lines); + + const result = await service.listSessions(); + + expect(findItem(result.items, sessionId)).toMatchObject({ + sourceType: 'scheduled_task', + sourceId: 'task-123', + }); + expect(await service.readCreationMetadata(sessionId)).toMatchObject({ + sourceType: 'scheduled_task', + sourceId: 'task-123', + }); + }); + it('reads one exact persisted summary without paging the catalog', async () => { const sessionId = '79777777-7777-4777-8777-777777777777'; writeSession(sessionId, [ diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index 49e90d99520..612390c63da 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -1041,7 +1041,7 @@ export class SessionService { ) { continue; } - return this.extractCreationMetadataFromRecords(records); + return this.extractCreationMetadataFromFile(filePath, records); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') continue; this.warn( @@ -1761,6 +1761,89 @@ export class SessionService { }; } + private extractCreationMetadataFromFile( + filePath: string, + records: ChatRecord[], + tailBuffer?: Buffer, + ): { + parentSessionId?: string; + sourceType?: string; + sourceId?: string; + } { + const metadata = this.extractCreationMetadataFromRecords(records); + if (metadata.sourceType !== undefined) return metadata; + + const tailSource = this.readSessionSourceFromTail(filePath, tailBuffer); + if (tailSource.sourceType === undefined) return metadata; + return { + ...metadata, + ...tailSource, + }; + } + + private readSessionSourceFromTail( + filePath: string, + scratchBuffer?: Buffer, + ): { sourceType?: string; sourceId?: string } { + let fd: number | undefined; + try { + const fileSize = fs.statSync(filePath).size; + if (fileSize === 0) return {}; + const readStart = Math.max(0, fileSize - TAIL_READ_SIZE); + const readLength = Math.min(fileSize, TAIL_READ_SIZE); + const buffer = + scratchBuffer && scratchBuffer.length >= readLength + ? scratchBuffer + : Buffer.alloc(readLength); + + fd = fs.openSync(filePath, 'r'); + const bytesRead = fs.readSync(fd, buffer, 0, readLength, readStart); + let firstSegmentIsPartial = false; + if (readStart > 0) { + const peek = Buffer.alloc(1); + fs.readSync(fd, peek, 0, 1, readStart - 1); + firstSegmentIsPartial = peek[0] !== 0x0a; + } + + const lines = buffer.toString('utf8', 0, bytesRead).split('\n'); + if (firstSegmentIsPartial) lines.shift(); + for (let i = lines.length - 1; i >= 0; i--) { + const trimmed = lines[i].trim(); + if (!trimmed) continue; + let record: ChatRecord; + try { + record = JSON.parse(trimmed) as ChatRecord; + } catch { + continue; + } + if (record.type !== 'system' || record.subtype !== 'session_source') { + continue; + } + const payload = record.systemPayload as + | { sourceType?: unknown; sourceId?: unknown } + | undefined; + if (typeof payload?.sourceType !== 'string') continue; + return { + sourceType: payload.sourceType, + ...(typeof payload.sourceId === 'string' + ? { sourceId: payload.sourceId } + : {}), + }; + } + return {}; + } catch { + return {}; + } finally { + if (fd !== undefined) { + try { + fs.closeSync(fd); + } catch { + // Best-effort after the metadata result has already been decided. + } + } + } + } + /** * Public accessor: returns both the current custom title and its source * for a given session. Used by `ChatRecordingService` on resume to @@ -2103,7 +2186,11 @@ export class SessionService { signal?.throwIfAborted(); const titleInfo = this.readSessionTitleInfoFromFile(filePath, tailBuffer); signal?.throwIfAborted(); - const source = this.extractCreationMetadataFromRecords(records); + const source = this.extractCreationMetadataFromFile( + filePath, + records, + tailBuffer, + ); items.push({ sessionId: firstRecord.sessionId, cwd: firstRecord.cwd, @@ -2168,7 +2255,7 @@ export class SessionService { return undefined; } const titleInfo = this.readSessionTitleInfoFromFile(filePath); - const source = this.extractCreationMetadataFromRecords(records); + const source = this.extractCreationMetadataFromFile(filePath, records); return { sessionId: firstRecord.sessionId, cwd: firstRecord.cwd, diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 515eba2efd8..f11f376a530 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -622,6 +622,10 @@ export interface RestoreSessionRequest { historyPageSize?: number; /** Load-only live-turn replay projection. Omit for the complete journal. */ liveReplayMode?: 'full' | 'summary'; + /** Restore-time attribution for legacy/unattributed sessions. */ + sourceType?: string; + /** Optional source-specific identifier. Requires `sourceType`. */ + sourceId?: string; /** * Client-side deadline for this restore request. `0` disables the client * timer and relies on the daemon's own restore deadline. @@ -3372,6 +3376,19 @@ export class DaemonClient { req: RestoreSessionRequest, clientId?: string, ): Promise { + let sourceType = req.sourceType; + let sourceId = req.sourceId; + if (req.sourceType !== undefined || req.sourceId !== undefined) { + try { + await this.requireCapability('session_source_metadata'); + } catch (error) { + if (!(error instanceof DaemonCapabilityMissingError)) { + throw error; + } + sourceType = undefined; + sourceId = undefined; + } + } const timeoutMs = this.resolveRestoreTimeoutMs(req.timeoutMs); return await this.fetchWithTimeout( `${this.baseUrl}/session/${urlEncode(sessionId)}/${action}`, @@ -3389,6 +3406,8 @@ export class DaemonClient { ...(action === 'load' && req.liveReplayMode !== undefined ? { liveReplayMode: req.liveReplayMode } : {}), + ...(sourceType !== undefined ? { sourceType } : {}), + ...(sourceId !== undefined ? { sourceId } : {}), }), }, async (res) => { diff --git a/packages/sdk-typescript/test/unit/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index 3373518feda..777dec781c2 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -2917,6 +2917,83 @@ describe('DaemonClient', () => { expect(JSON.parse(calls[0]!.body!)).toEqual({ cwd: '/w' }); }); + it.each(['load', 'resume'] as const)( + 'omits restore source metadata over %s when the daemon lacks the capability', + async (action) => { + const { fetch, calls } = recordingFetch((request) => + request.url.endsWith('/capabilities') + ? jsonResponse(200, { + v: 1, + mode: 'http-bridge', + features: ['session_restore'], + modelServices: [], + }) + : jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: false, + state: {}, + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + if (action === 'load') { + await client.loadSession('s-1', { + workspaceCwd: '/work/a', + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + } else { + await client.resumeSession('s-1', { + workspaceCwd: '/work/a', + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + } + + expect(calls.map((call) => call.url)).toEqual([ + 'http://daemon/capabilities', + `http://daemon/session/s-1/${action}`, + ]); + expect(JSON.parse(calls[1]!.body!)).toEqual({ cwd: '/work/a' }); + }, + ); + + it.each(['load', 'resume'] as const)( + 'fails %s closed when source capability probing fails', + async (action) => { + const { fetch, calls } = recordingFetch((request) => + request.url.endsWith('/capabilities') + ? jsonResponse(500, { error: 'capabilities unavailable' }) + : jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: false, + state: {}, + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + const restore = + action === 'load' + ? client.loadSession('s-1', { + sourceType: 'channel', + sourceId: 'dingtalk-main', + }) + : client.resumeSession('s-1', { + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + await expect(restore).rejects.toMatchObject({ + status: 500, + }); + expect(calls.map((call) => call.url)).toEqual([ + 'http://daemon/capabilities', + ]); + }, + ); + it('throws DaemonHttpError on restore failures', async () => { const { fetch } = recordingFetch(() => jsonResponse(404, { error: 'missing' }), diff --git a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts index 8f0f9c04ee1..6029747c96f 100644 --- a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts @@ -471,6 +471,100 @@ describe('DaemonSessionClient', () => { ]); }); + it('forwards source metadata through resume requests', async () => { + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/capabilities')) { + return jsonResponse(200, { features: ['session_source_metadata'] }); + } + if (req.url.endsWith('/session/s-1/resume')) { + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-1', + state: {}, + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + await DaemonSessionClient.resume(client, 's-1', { + workspaceCwd: '/work/a', + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(calls[1]?.url).toBe('http://daemon/session/s-1/resume'); + expect(JSON.parse(calls[1]!.body!)).toEqual({ + cwd: '/work/a', + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + }); + + it('omits source metadata before resuming against an old daemon', async () => { + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/capabilities')) { + return jsonResponse(200, { features: [] }); + } + if (req.url.endsWith('/session/s-1/resume')) { + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-1', + state: {}, + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + await DaemonSessionClient.resume(client, 's-1', { + workspaceCwd: '/work/a', + sourceType: 'channel', + sourceId: 'dingtalk-main', + }); + + expect(calls.map((c) => c.url)).toEqual([ + 'http://daemon/capabilities', + 'http://daemon/session/s-1/resume', + ]); + expect(JSON.parse(calls[1]!.body!)).toEqual({ + cwd: '/work/a', + }); + }); + + it('rejects source metadata resume when capability lookup fails', async () => { + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/capabilities')) { + return jsonResponse(500, { error: 'capability probe failed' }); + } + if (req.url.endsWith('/session/s-1/resume')) { + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-1', + state: {}, + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + await expect( + DaemonSessionClient.resume(client, 's-1', { + workspaceCwd: '/work/a', + sourceType: 'channel', + sourceId: 'dingtalk-main', + }), + ).rejects.toThrow('capability probe failed'); + + expect(calls.map((c) => c.url)).toEqual(['http://daemon/capabilities']); + }); + it('replays attach-time model switch events on first subscription', async () => { const { fetch, calls } = recordingFetch((req) => { if (req.url.endsWith('/session')) {