diff --git a/packages/api/session-controller/src/commands.ts b/packages/api/session-controller/src/commands.ts index 06edd32e87..2f40aa1972 100644 --- a/packages/api/session-controller/src/commands.ts +++ b/packages/api/session-controller/src/commands.ts @@ -206,6 +206,10 @@ export class SessionCommandController { * @returns the new Session identity. */ async fork(request: SessionForkRequest): Promise { + if (request.atSeq !== undefined + && (!Number.isInteger(request.atSeq) || request.atSeq < 0)) { + reject('bad-request', 'atSeq must be a non-negative integer', {}) + } let source: SessionReadState try { source = await this.readSessionState(request.sessionId) diff --git a/packages/api/session-controller/src/history.ts b/packages/api/session-controller/src/history.ts index 3e08b95815..3b04e0e506 100644 --- a/packages/api/session-controller/src/history.ts +++ b/packages/api/session-controller/src/history.ts @@ -44,8 +44,15 @@ interface BufferedEvent { /** Implements cold-safe history operations delegated by the Session Controller. */ export class SessionHistoryController { + private readonly closeFollowers = new Set<() => void>() + /** @param ctx - Host context carrying Session, persistence, presenter, and projection services. */ - constructor(private readonly ctx: Context) {} + constructor(private readonly ctx: Context) { + ctx.effect(() => () => { + for (const close of this.closeFollowers) close() + this.closeFollowers.clear() + }, 'session-controller.history') + } /** * Read one message-aligned history page without activating an Agent. @@ -101,6 +108,12 @@ export class SessionHistoryController { wake = undefined resume?.() } + const follower = { closed: false } + const close = (): void => { + follower.closed = true + notify() + } + this.closeFollowers.add(close) const disposeEvent = this.ctx.on('session/event', (session, event) => { if (session.id !== target) return buffered.push({ session, event }) @@ -138,7 +151,7 @@ export class SessionHistoryController { yield { type: 'event', ...entryFor(this.ctx, event, events, scope) } } } - while (!signal.aborted) { + while (!follower.closed && !signal.aborted) { const item = buffered.shift() if (item === undefined) { await new Promise((resolve) => { wake = resolve }) @@ -154,6 +167,7 @@ export class SessionHistoryController { yield { type: 'event', ...entry } } } finally { + this.closeFollowers.delete(close) signal.removeEventListener('abort', onAbort) disposeCreated() disposeEvent() diff --git a/packages/api/session-controller/src/list.ts b/packages/api/session-controller/src/list.ts index 04ee5a09bf..845e7fae86 100644 --- a/packages/api/session-controller/src/list.ts +++ b/packages/api/session-controller/src/list.ts @@ -25,6 +25,7 @@ export const DEFAULT_COLD_BLANK_PROBE_MAX_BYTES = 1024 const COLD_SUMMARY_BATCH_SIZE = 16 const SEARCH_PROVIDER_CALL_LIMIT = 100 +const SESSION_SEARCH_QUERY_MAX_CHARS = 500 const MESSAGE_TYPES = new Set(['user/message', 'assistant/message']) const sessionListMetadataSchema: z.ZodType = z.object({ @@ -190,6 +191,7 @@ export class ApiSessionList { * @returns authorized bounded Session search results. */ async search(query: string, signal: AbortSignal): Promise { + const normalizedQuery = normalizeSearchQuery(query) signal.throwIfAborted() const provider = this.ctx.get('sessionQuery') if (provider === undefined) { @@ -221,7 +223,7 @@ export class ApiSessionList { let page try { page = await provider.searchSessions({ - query, + query: normalizedQuery, eventFilters: [ { kind: 'type', values: ['user/message', 'assistant/message'] }, { kind: 'surface', values: ['current'] }, @@ -312,6 +314,24 @@ export class ApiSessionList { } } +function normalizeSearchQuery(query: string): string { + const normalized = query.trim() + if (normalized.length === 0) { + reject('bad-request', 'session search query must not be empty', {}) + } + if (normalized.length > SESSION_SEARCH_QUERY_MAX_CHARS) { + reject( + 'bad-request', + `session search query must contain at most ${SESSION_SEARCH_QUERY_MAX_CHARS} UTF-16 code units`, + {}, + ) + } + if (normalized.includes('\0')) { + reject('bad-request', 'session search query must not contain NUL', {}) + } + return normalized +} + function reject(code: string, message: string, details: object): never { throw new TypertRemoteFailure({ code, message, details }) } diff --git a/packages/api/session-controller/tests/session-fork.host.spec.ts b/packages/api/session-controller/tests/session-fork.host.spec.ts index f3bf6fe31b..481f51f825 100644 --- a/packages/api/session-controller/tests/session-fork.host.spec.ts +++ b/packages/api/session-controller/tests/session-fork.host.spec.ts @@ -218,6 +218,18 @@ describe('sessions.fork', () => { await ctx.fiber.dispose() }) + it('rejects invalid fork anchors before reading or creating a Session', async () => { + const ctx = await composed() + const proxy = remote(ctx) + + for (const atSeq of [-1, 0.5]) { + await expect(proxy.fork(request({ sessionId: sid('missing'), atSeq }))) + .resolves.toMatchObject({ ok: false, error: { code: 'bad-request' } }) + } + expect(ctx.sessions.list()).toEqual([]) + await ctx.fiber.dispose() + }) + it('cuts through an aborted turn: stopped is closed, not open', async () => { const ctx = await composed() const source = liveAgent(ctx, 'session-aborted', 1, 'aborted') diff --git a/packages/api/session-controller/tests/session-search.host.spec.ts b/packages/api/session-controller/tests/session-search.host.spec.ts index 0d4160b093..173d20cf9e 100644 --- a/packages/api/session-controller/tests/session-search.host.spec.ts +++ b/packages/api/session-controller/tests/session-search.host.spec.ts @@ -115,7 +115,7 @@ describe('session.search', () => { const remote = createSessionTestRemote(ctx, defaults) const signal = new AbortController().signal - const response = await remote.search(request('matching answer'), signal) + const response = await remote.search(request(' matching answer '), signal) expect(response).toEqual({ ok: true, @@ -143,6 +143,21 @@ describe('session.search', () => { expect(exec.signal).toBe(signal) }) + it('rejects invalid wire queries before invoking the search provider', async () => { + const ctx = await baseContext() + ctx.sessions.create(sid('visible'), { meta: header('visible') }) + const searchSessions = vi.fn() + ctx.provide('sessionQuery', { searchSessions } as never) + const remote = createSessionTestRemote(ctx, defaults) + + for (const query of ['', ' ', 'contains\0nul', 'x'.repeat(501)]) { + await expect(remote.search(request(query), new AbortController().signal)) + .resolves.toMatchObject({ ok: false, error: { code: 'bad-request' } }) + } + expect(searchSessions).not.toHaveBeenCalled() + await ctx.fiber.dispose() + }) + it('returns an empty page without invoking the index when no session is visible', async () => { const ctx = await baseContext() const searchSessions = vi.fn() diff --git a/packages/api/session-controller/tests/transport.host.spec.ts b/packages/api/session-controller/tests/transport.host.spec.ts index 775f846475..7bdb426e9d 100644 --- a/packages/api/session-controller/tests/transport.host.spec.ts +++ b/packages/api/session-controller/tests/transport.host.spec.ts @@ -82,6 +82,31 @@ describe('SessionHistoryController', () => { expect(await iterator.next()).toMatchObject({ done: true }) }) + it('ends active followers when the owning Controller unloads', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + let transport!: SessionHistoryController + const owner = ctx.plugin(Object.assign( + (inner: Context) => { transport = new SessionHistoryController(inner) }, + { inject: ['sessions'] }, + )) + await owner.await() + const session = ctx.sessions.create(SessionId('controller-unload'), { meta: { cwd: '/workspace' } }) + const iterator = transport.follow( + { address: { kind: 'session', sessionId: session.id } }, + new AbortController().signal, + )[Symbol.asyncIterator]() + + await expect(iterator.next()).resolves.toEqual({ + done: false, + value: { type: 'opened', cursor: -1 }, + }) + const pending = iterator.next() + await owner.dispose() + await expect(pending).resolves.toEqual({ done: true, value: undefined }) + await ctx.fiber.dispose() + }) + it('resumes from the last applied seq before delivering later live events', async () => { const { ctx, transport } = await setup() const session = ctx.sessions.create(SessionId('resume'), { meta: { cwd: '/workspace' } }) diff --git a/packages/interaction/user-questions/src/index.ts b/packages/interaction/user-questions/src/index.ts index 241ea6cef3..9c72db4c79 100644 --- a/packages/interaction/user-questions/src/index.ts +++ b/packages/interaction/user-questions/src/index.ts @@ -158,7 +158,10 @@ export class UserQuestionService extends Service { askProvider, )) } catch (error) { - if (request.signal?.aborted && !(error instanceof UserQuestionError)) { + if (error instanceof UserQuestionError) throw error + const restored = restoreUserQuestionError(error) + if (restored !== undefined) throw restored + if (request.signal?.aborted) { throw abortedQuestion(error) } throw error @@ -166,4 +169,12 @@ export class UserQuestionService extends Service { } } +function restoreUserQuestionError(reason: unknown): UserQuestionError | undefined { + if (!(reason instanceof Error) || reason.name !== 'UserQuestionError') return undefined + const code: unknown = (reason as Error & { readonly code?: unknown }).code + return typeof code === 'string' + ? new UserQuestionError(reason.message, code, { cause: reason }) + : undefined +} + export default UserQuestionService diff --git a/packages/interaction/user-questions/tests/user-questions.spec.ts b/packages/interaction/user-questions/tests/user-questions.spec.ts index 98b0c49375..a436d9bd7c 100644 --- a/packages/interaction/user-questions/tests/user-questions.spec.ts +++ b/packages/interaction/user-questions/tests/user-questions.spec.ts @@ -122,6 +122,46 @@ describe('UserQuestionService', () => { })).rejects.toBe(cancelled) }) + it('restores a transported provider rejection to UserQuestionError', async () => { + const ctx = new Context() + await ctx.plugin(UserQuestionService) + const transported = Object.assign(new Error('the user cancelled ask_user_question'), { + name: 'UserQuestionError', + code: 'ASK_CANCELLED', + }) + ctx.userQuestions.registerProvider({ ask: () => Promise.reject(transported) }) + + const rejection = await ctx.userQuestions.ask({ + questions: [{ id: 'confirm', question: 'Proceed?' }], + }).then( + () => undefined, + (error: unknown) => error, + ) + + expect(rejection).toBeInstanceOf(UserQuestionError) + expect(rejection).toMatchObject({ + name: 'UserQuestionError', + code: 'ASK_CANCELLED', + cause: transported, + }) + }) + + it.each([ + ['an ordinary Error', new Error('provider failed')], + ['a namesake Error without a string code', Object.assign(new Error('provider failed'), { + name: 'UserQuestionError', + })], + ['a non-Error rejection', { name: 'UserQuestionError', code: 'ASK_CANCELLED' }], + ])('preserves %s from the provider', async (_label, rejection) => { + const ctx = new Context() + await ctx.plugin(UserQuestionService) + ctx.userQuestions.registerProvider({ ask: vi.fn().mockRejectedValue(rejection) }) + + await expect(ctx.userQuestions.ask({ + questions: [{ id: 'confirm', question: 'Proceed?' }], + })).rejects.toBe(rejection) + }) + it('rejects empty question batches before reaching the provider', async () => { const ctx = new Context() await ctx.plugin(UserQuestionService) diff --git a/packages/plan/plan-mode/tests/plan-mode.spec.ts b/packages/plan/plan-mode/tests/plan-mode.spec.ts index eea2beddb0..79f57cb10c 100644 --- a/packages/plan/plan-mode/tests/plan-mode.spec.ts +++ b/packages/plan/plan-mode/tests/plan-mode.spec.ts @@ -997,8 +997,10 @@ describe('exit_plan_mode', () => { it('reads a dismissed review as the user taking the turn back, not as a failure', async () => { const { ctx, agent } = await setupWithReview() ctx.userQuestions.registerProvider({ - ask: () => Promise.reject(new UserQuestionError( - 'the user cancelled ask_user_question', 'ASK_CANCELLED')), + ask: () => Promise.reject(Object.assign( + new Error('the user cancelled ask_user_question'), + { name: 'UserQuestionError', code: 'ASK_CANCELLED' }, + )), }) const result = await callExit(ctx, agent) expect(result.isError).toBe(true)