diff --git a/packages/agent-core-v2/scripts/check-domain-layers.mjs b/packages/agent-core-v2/scripts/check-domain-layers.mjs index c66ae8b4a..12a6909d9 100644 --- a/packages/agent-core-v2/scripts/check-domain-layers.mjs +++ b/packages/agent-core-v2/scripts/check-domain-layers.mjs @@ -91,6 +91,11 @@ const DOMAIN_LAYER = new Map([ // create-source/close-reason vocabulary; a pure contract with no IO, so it // sits in L1 beside `sessionContext`. ['sessionLifecycleHooks', 1], + // `sessionLifetime` is the per-session liveness seed (`AbortSignal` + // aborted by the Workspace-scope `workspaceHandler` when the session's + // close begins); a pure contract with no IO, so it sits in L1 beside + // `sessionLifecycleHooks`. + ['sessionLifetime', 1], // `scopeContext` is the Agent-scope seeded immutable facts value // (`agentId` plus a persistence scope helper); a pure seed with no IO, so it // sits in L1 beside `sessionContext`. diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index 8746eebad..3e1591a2a 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -366,6 +366,7 @@ export * from '#/workspace/workspaceHandler/workspaceHandler'; export * from '#/workspace/workspaceHandler/workspaceHandlerService'; export * from '#/workspace/workspaceHandler/addressing'; export * from '#/session/sessionLifecycleHooks/sessionLifecycleHooks'; +export * from '#/session/sessionLifetime/sessionLifetime'; export * from '#/session/externalHooks/externalHooks'; export * from '#/session/externalHooks/externalHooksService'; import '#/app/sessionExport/errors'; diff --git a/packages/agent-core-v2/src/session/sessionLifetime/sessionLifetime.ts b/packages/agent-core-v2/src/session/sessionLifetime/sessionLifetime.ts new file mode 100644 index 000000000..b209420b2 --- /dev/null +++ b/packages/agent-core-v2/src/session/sessionLifetime/sessionLifetime.ts @@ -0,0 +1,28 @@ +/** + * `sessionLifetime` domain (L1) — per-session liveness seed. + * + * Defines the `ISessionLifetime` carrying the session scope's liveness + * `AbortSignal`: the Workspace-scope `workspaceHandler` creates one + * `AbortController` per materialized session, seeds the signal into the + * Session scope, and aborts it synchronously when the session's close + * begins — before any async `onWillCloseSession` hook runs. Session-scope + * consumers with in-flight async work (e.g. `sessionTitle`) treat an + * aborted signal as "this scope is being torn down": cancel the work and + * drop its write-back. Pure contract — no IO, no store. Session-scoped. + */ + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { ScopeSeed } from '#/_base/di/scope'; + +export interface ISessionLifetime { + readonly _serviceBrand: undefined; + + readonly signal: AbortSignal; +} + +export const ISessionLifetime: ServiceIdentifier = + createDecorator('sessionLifetime'); + +export function sessionLifetimeSeed(lifetime: ISessionLifetime): ScopeSeed { + return [[ISessionLifetime as ServiceIdentifier, lifetime]]; +} diff --git a/packages/agent-core-v2/src/session/sessionTitle/sessionTitleService.ts b/packages/agent-core-v2/src/session/sessionTitle/sessionTitleService.ts index a95234636..a186b069b 100644 --- a/packages/agent-core-v2/src/session/sessionTitle/sessionTitleService.ts +++ b/packages/agent-core-v2/src/session/sessionTitle/sessionTitleService.ts @@ -9,8 +9,11 @@ * (the kap-server route), gated by a managed Kimi Code OAuth login; any * failure degrades to keeping the current title, and a custom title set by * the user is never overwritten. An already-generated title is not - * regenerated unless forced, and the write-back is dropped when this scope - * is no longer the live session in `workspaceHandler`. Provider config comes + * regenerated unless forced, and the write-back is dropped once this + * scope's `sessionLifetime` signal fires — the in-flight request carries + * the signal, and a close drains the pending generation through the + * `sessionLifecycleHooks` `onWillCloseSession` slot before the scope is + * disposed. Provider config comes * from `provider`, the bearer token from `auth`, host identity headers from * `model`, prompt history from `agentLifecycle`/`sessionTitle`, and logs * through `log`. Bound at Session scope. @@ -29,13 +32,18 @@ import { LifecycleScope, ScopeActivation, registerScopedService } from '#/_base/ import { ILogService } from '#/_base/log/log'; import { IOAuthService } from '#/app/auth/auth'; import { IEventService } from '#/app/event/event'; +import type { Hooks } from '#/hooks'; import { IAgentLifecycleService, MAIN_AGENT_ID } from '#/session/agentLifecycle/agentLifecycle'; import { IHostRequestHeaders } from '#/kosong/model/hostRequestHeaders'; import { IProviderService } from '#/kosong/provider/provider'; import { isOAuthCatalogVendor } from '#/kosong/provider/providerDefinition'; import { ISessionContext } from '#/session/sessionContext/sessionContext'; +import { + ISessionLifecycleHooks, + type SessionLifecycleHookSlots, +} from '#/session/sessionLifecycleHooks/sessionLifecycleHooks'; +import { ISessionLifetime } from '#/session/sessionLifetime/sessionLifetime'; import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata'; -import { IWorkspaceHandlerService } from '#/workspace/workspaceHandler/workspaceHandler'; import { IAgentTitlePromptSource } from './agentTitlePromptSource'; import { ISessionTitleService } from './sessionTitle'; @@ -54,16 +62,27 @@ export class SessionTitleService implements ISessionTitleService { constructor( @ISessionContext private readonly ctx: ISessionContext, @ISessionMetadata private readonly metadata: ISessionMetadata, + @ISessionLifetime private readonly lifetime: ISessionLifetime, + @ISessionLifecycleHooks + lifecycleHooks: Hooks, @IAgentLifecycleService private readonly agentLifecycle: IAgentLifecycleService, @IEventService private readonly eventService: IEventService, - @IWorkspaceHandlerService private readonly workspaceHandler: IWorkspaceHandlerService, @IProviderService private readonly providers: IProviderService, @IOAuthService private readonly oauth: IOAuthService, @IHostRequestHeaders private readonly hostHeaders: IHostRequestHeaders, @ILogService private readonly log: ILogService, - ) {} + ) { + // A close aborts the lifetime signal first, so the in-flight generation + // settles fast; draining it here keeps its metadata write ordered before + // the scope's disposal (and any later resume re-reading the document). + lifecycleHooks.onWillCloseSession.register('sessionTitle', async (_event, next) => { + await this._generation?.catch(() => undefined); + await next(); + }); + } async generateTitle(options?: { readonly force?: boolean }): Promise { + if (this.lifetime.signal.aborted) return undefined; const current = await this.metadata.read(); if (current.titleKind === 'custom') return undefined; if (current.titleKind === 'generated' && options?.force !== true) return undefined; @@ -122,9 +141,10 @@ export class SessionTitleService implements ISessionTitleService { ...this.hostHeaders.headers, ...provider.customHeaders, }, + signal: this.lifetime.signal, }); let result = await requestTitle(token); - if (result.kind === 'error' && result.status === 401) { + if (result.kind === 'error' && result.status === 401 && !this.lifetime.signal.aborted) { try { token = await tokenProvider.getAccessToken({ force: true }); } catch (error) { @@ -138,10 +158,7 @@ export class SessionTitleService implements ISessionTitleService { this.log.debug(`chat_title request failed: ${result.message}`); return undefined; } - const live = this.workspaceHandler.get(this.ctx.sessionId); - if (live === undefined || live.accessor.get(ISessionMetadata) !== this.metadata) { - return undefined; - } + if (this.lifetime.signal.aborted) return undefined; const title = result.title.slice(0, MAX_GENERATED_TITLE_LENGTH); const applied = await this.metadata.setGeneratedTitleIfUncustomized(title); if (!applied) return undefined; diff --git a/packages/agent-core-v2/src/workspace/workspaceHandler/workspaceHandlerService.ts b/packages/agent-core-v2/src/workspace/workspaceHandler/workspaceHandlerService.ts index 9004bdd9f..4968b1a8e 100644 --- a/packages/agent-core-v2/src/workspace/workspaceHandler/workspaceHandlerService.ts +++ b/packages/agent-core-v2/src/workspace/workspaceHandler/workspaceHandlerService.ts @@ -4,8 +4,11 @@ * Owns the registry of THIS handler's open Session child scopes, creating * them through the DI scope tree (children of the handler's Workspace * scope) and seeding each with its identity, storage addressing derived - * from the handler's `workspaceContext.persistenceScope`, and a per-session - * `sessionLifecycleHooks` slots instance it runs around create/close, + * from the handler's `workspaceContext.persistenceScope`, a per-session + * `sessionLifecycleHooks` slots instance it runs around create/close, and + * a per-session `sessionLifetime` abort signal it fires synchronously when + * a close/archive begins — before any async close hook — so in-flight + * session work can cancel itself and drop its write-back, * tearing sessions down on close/archive — archiving flags the session's * `sessionMetadata`, removes its `agentLifecycle` agents, restoring clears * the archived flag, and broadcasts through `event`; session start and @@ -106,6 +109,7 @@ import { sessionLifecycleHooksSeed, type SessionLifecycleHookSlots, } from '#/session/sessionLifecycleHooks/sessionLifecycleHooks'; +import { sessionLifetimeSeed } from '#/session/sessionLifetime/sessionLifetime'; import { ISessionMetadata, type SessionMeta } from '#/session/sessionMetadata/sessionMetadata'; import { ISessionProcessRunner } from '#/session/process/processRunner'; import { sessionSkillCatalogDataSeed } from '#/session/sessionSkillCatalog/skillCatalogData'; @@ -165,6 +169,7 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan private readonly _onDidForkSession = this._register(new Emitter()); readonly onDidForkSession: Event = this._onDidForkSession.event; private readonly resuming = new Map>(); + private readonly sessionLifetimes = new Map(); constructor( @IInstantiationService private readonly instantiation: IInstantiationService, @@ -230,6 +235,7 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan // Roll back ONLY the session directory — the handler stays live. const sessionDir = handle.accessor.get(ISessionContext).sessionDir; this.sessions.delete(sessionId); + this.invalidateSessionLifetime(sessionId); await this.drainAgents(handle).catch(() => {}); handle.dispose(); await this.hostFs.remove(sessionDir).catch(() => {}); @@ -266,6 +272,8 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan 'onDidCreateSession', 'onWillCloseSession', ]); + const lifetime = new AbortController(); + this.sessionLifetimes.set(opts.sessionId, lifetime); await this.hostEnv.ready; const handle = createScopedChildHandle( this.instantiation, @@ -275,6 +283,7 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan extra: [ ...sessionContextSeed(ctx), ...sessionLifecycleHooksSeed(hooks), + ...sessionLifetimeSeed({ _serviceBrand: undefined, signal: lifetime.signal }), [ITelemetryService, this.telemetry.withContext({ sessionId: opts.sessionId })], // Workspace resource seeds (the §3.5 injection contracts): the // handler's shared skill / agent-profile catalogs, AGENTS.md @@ -309,6 +318,7 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan ]); await this.mcp.ready; } catch (error) { + this.invalidateSessionLifetime(opts.sessionId); handle.dispose(); // Re-arm the explicit agent-profile loader after a fatal rejection: // its `ready` tracks the latest load pass, so without a reload the @@ -409,6 +419,7 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan async close(sessionId: string): Promise { const handle = this.sessions.get(sessionId); if (handle === undefined) return; + this.invalidateSessionLifetime(sessionId); await this.announceWillClose({ sessionId, handle, reason: 'exit' }); this.sessions.delete(sessionId); await this.drainAgents(handle); @@ -419,6 +430,7 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan async archive(sessionId: string): Promise { const handle = this.sessions.get(sessionId); if (handle === undefined) return; + this.invalidateSessionLifetime(sessionId); const meta = handle.accessor.get(ISessionMetadata); await meta.setArchived(true); await this.drainAgents(handle); @@ -445,6 +457,18 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan .onWillCloseSession.run({ reason: event.reason }); } + /** + * Synchronously fires the session's liveness abort signal. Runs before any + * async close work (hooks, agent drain) so in-flight session services see + * the teardown immediately; idempotent — later calls are no-ops. + */ + private invalidateSessionLifetime(sessionId: string): void { + const lifetime = this.sessionLifetimes.get(sessionId); + if (lifetime === undefined) return; + this.sessionLifetimes.delete(sessionId); + lifetime.abort(); + } + private async drainAgents(handle: ISessionScopeHandle): Promise { const agentLifecycle = handle.accessor.get(IAgentLifecycleService); for (const agent of agentLifecycle.list()) { @@ -545,6 +569,7 @@ export class WorkspaceHandlerService extends Disposable implements IWorkspaceHan } catch (error) { if (targetId !== undefined) { this.sessions.delete(targetId); + this.invalidateSessionLifetime(targetId); } if (target !== undefined) { try { diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index 1faa14ee2..dc364a6b6 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -153,6 +153,7 @@ import { ISessionLifecycleHooks, type SessionLifecycleHookSlots, } from '#/session/sessionLifecycleHooks/sessionLifecycleHooks'; +import { ISessionLifetime } from '#/session/sessionLifetime/sessionLifetime'; import { IEventBus } from '#/app/event/eventBus'; import { IWireService } from '#/wire/wire'; import { WireService } from '#/wire/wireService'; @@ -1177,6 +1178,10 @@ export class AgentTestContext { 'onWillCloseSession', ]), ); + reg.defineInstance(ISessionLifetime, { + _serviceBrand: undefined, + signal: new AbortController().signal, + }); reg.defineInstance(ISessionInteractionService, this.createInteractionService()); reg.defineInstance(ISessionApprovalService, this.createApprovalService()); reg.defineInstance(ISessionQuestionService, this.createQuestionService()); diff --git a/packages/agent-core-v2/test/session/sessionTitle/sessionTitleService.test.ts b/packages/agent-core-v2/test/session/sessionTitle/sessionTitleService.test.ts index bd74dd8bd..7ca4a6f84 100644 --- a/packages/agent-core-v2/test/session/sessionTitle/sessionTitleService.test.ts +++ b/packages/agent-core-v2/test/session/sessionTitle/sessionTitleService.test.ts @@ -12,17 +12,15 @@ import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vite import { OAuthConnectionError, OAuthUnauthorizedError } from '@moonshot-ai/kimi-code-oauth'; import { DisposableStore, type IDisposable } from '#/_base/di/lifecycle'; -import type { ServiceIdentifier } from '#/_base/di/instantiation'; import { LifecycleScope, type IAgentScopeHandle, - type ISessionScopeHandle, } from '#/_base/di/scope'; import { createServices, type TestInstantiationService } from '#/_base/di/test'; import { Emitter } from '#/_base/event'; import { IOAuthService } from '#/app/auth/auth'; import { type DomainEvent, IEventService } from '#/app/event/event'; -import { IWorkspaceHandlerService } from '#/workspace/workspaceHandler/workspaceHandler'; +import { createHooks, type Hooks } from '#/hooks'; import { HostRequestHeaders, IHostRequestHeaders } from '#/kosong/model/hostRequestHeaders'; import { IProviderService, @@ -34,6 +32,11 @@ import { IAgentLifecycleService, MAIN_AGENT_ID, } from '#/session/agentLifecycle/agentLifecycle'; +import { + ISessionLifecycleHooks, + type SessionLifecycleHookSlots, +} from '#/session/sessionLifecycleHooks/sessionLifecycleHooks'; +import { ISessionLifetime } from '#/session/sessionLifetime/sessionLifetime'; import { IAgentTitlePromptSource, } from '#/session/sessionTitle/agentTitlePromptSource'; @@ -137,20 +140,6 @@ function createPendingFetch() { }; } -function makeLiveSessionHandle(meta: ISessionMetadata): ISessionScopeHandle { - return { - id: SESSION_ID, - kind: LifecycleScope.Session, - accessor: { - get: (id: ServiceIdentifier) => { - if (id === ISessionMetadata) return meta as T; - throw new Error(`unexpected service ${String(id)}`); - }, - }, - dispose: () => undefined, - }; -} - describe('SessionTitleService', () => { let disposables: DisposableStore; let ix: TestInstantiationService; @@ -162,7 +151,8 @@ describe('SessionTitleService', () => { let forceTokenError: Error | undefined; let resolvedOAuthRefs: Array; let titlePrompts: readonly string[]; - let liveSession: ISessionScopeHandle | undefined; + let lifetimeController: AbortController; + let lifecycleHooks: Hooks; let tokenCalls: boolean[]; beforeEach(() => { @@ -173,7 +163,11 @@ describe('SessionTitleService', () => { tokenCalls = []; providers = { 'managed:kimi-code': MANAGED_PROVIDER }; metadata = new FakeSessionMetadata(); - liveSession = makeLiveSessionHandle(metadata); + lifetimeController = new AbortController(); + lifecycleHooks = createHooks([ + 'onDidCreateSession', + 'onWillCloseSession', + ]); events = new FakeEventService(); fetchMock = vi.fn<(url: string, init?: RequestInit) => Promise>( async () => @@ -213,9 +207,11 @@ describe('SessionTitleService', () => { get: () => mainAgent, }); reg.defineInstance(IEventService, events); - reg.definePartialInstance(IWorkspaceHandlerService, { - get: () => liveSession, + reg.defineInstance(ISessionLifetime, { + _serviceBrand: undefined, + signal: lifetimeController.signal, }); + reg.defineInstance(ISessionLifecycleHooks, lifecycleHooks); reg.defineInstance(IProviderService, stubProviderService(providers)); reg.definePartialInstance(IOAuthService, { resolveTokenProvider: (_provider, oauthRef) => { @@ -362,27 +358,85 @@ describe('SessionTitleService', () => { expect(metadata.meta.title).toBe('user 取的标题'); }); - it('drops the write-back when the session scope was superseded mid-flight', async () => { + it('drops the write-back when the lifetime signal aborts after the response arrived', async () => { + const pendingFetch = createPendingFetch(); + fetchMock.mockImplementationOnce(pendingFetch.fetch); const applySpy = vi.spyOn(metadata, 'setGeneratedTitleIfUncustomized'); titlePrompts = ['hello']; - liveSession = makeLiveSessionHandle(new FakeSessionMetadata()); - await expect(ix.get(ISessionTitleService).generateTitle()).resolves.toBeUndefined(); - expect(fetchMock).toHaveBeenCalledTimes(1); + const generation = ix.get(ISessionTitleService).generateTitle(); + await pendingFetch.started; + lifetimeController.abort(); + pendingFetch.resolve( + new Response(JSON.stringify({ title: '生成的标题' }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }), + ); + + await expect(generation).resolves.toBeUndefined(); expect(applySpy).not.toHaveBeenCalled(); expect(metadata.meta.title).toBeUndefined(); }); - it('drops the write-back while the session is still resuming', async () => { + it('cancels the in-flight request when the lifetime signal aborts', async () => { + fetchMock.mockImplementationOnce( + async (_url, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => { + reject(new DOMException('The operation was aborted.', 'AbortError')); + }); + }), + ); const applySpy = vi.spyOn(metadata, 'setGeneratedTitleIfUncustomized'); titlePrompts = ['hello']; - liveSession = undefined; - await expect(ix.get(ISessionTitleService).generateTitle()).resolves.toBeUndefined(); + const generation = ix.get(ISessionTitleService).generateTitle(); + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + lifetimeController.abort(); + + await expect(generation).resolves.toBeUndefined(); expect(applySpy).not.toHaveBeenCalled(); expect(metadata.meta.title).toBeUndefined(); }); + it('returns unavailable without fetching when the scope is already closing', async () => { + lifetimeController.abort(); + titlePrompts = ['hello']; + + await expect(ix.get(ISessionTitleService).generateTitle()).resolves.toBeUndefined(); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it('drains an in-flight generation through onWillCloseSession', async () => { + const pendingFetch = createPendingFetch(); + fetchMock.mockImplementationOnce(pendingFetch.fetch); + titlePrompts = ['hello']; + + const generation = ix.get(ISessionTitleService).generateTitle(); + await pendingFetch.started; + + let drained = false; + const closeRun = lifecycleHooks.onWillCloseSession.run({ reason: 'exit' }).then(() => { + drained = true; + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(drained).toBe(false); + + pendingFetch.resolve( + new Response(JSON.stringify({ title: '生成的标题' }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }), + ); + await closeRun; + expect(drained).toBe(true); + await expect(generation).resolves.toBe('生成的标题'); + expect(metadata.meta.title).toBe('生成的标题'); + }); + it('keeps the current title when the backend request fails', async () => { fetchMock.mockImplementationOnce(async () => new Response('', { status: 500 })); titlePrompts = ['hello']; diff --git a/packages/oauth/src/managed-tools.ts b/packages/oauth/src/managed-tools.ts index c859239c8..5c6b63b37 100644 --- a/packages/oauth/src/managed-tools.ts +++ b/packages/oauth/src/managed-tools.ts @@ -34,9 +34,14 @@ export async function fetchChatTitle( url: string, accessToken: string, chatContent: string, - opts: { timeoutMs?: number; headers?: Record } = {}, + opts: { timeoutMs?: number; headers?: Record; signal?: AbortSignal } = {}, ): Promise { const controller = new AbortController(); + if (opts.signal !== undefined) { + const external = opts.signal; + if (external.aborted) controller.abort(); + else external.addEventListener('abort', () => { controller.abort(); }, { once: true }); + } const timer = setTimeout(() => { controller.abort(); }, opts.timeoutMs ?? 8000); @@ -71,7 +76,9 @@ export async function fetchChatTitle( return { kind: 'ok', title }; } catch (error) { if (error instanceof Error && error.name === 'AbortError') { - return { kind: 'error', message: 'Failed to generate session title: request timed out.' }; + const reason = + opts.signal?.aborted === true ? 'request aborted.' : 'request timed out.'; + return { kind: 'error', message: `Failed to generate session title: ${reason}` }; } const msg = error instanceof Error ? error.message : String(error); return { kind: 'error', message: `Failed to generate session title: ${msg}` };