fix(api): preserve migrated transport semantics

packages/api/session-controller/src/commands.ts

@@ -206,6 +206,10 @@ export class SessionCommandController {
    * @returns the new Session identity.
    */
   async fork(request: SessionForkRequest): Promise<SessionForkValue> {
+    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)

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<void>((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()

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<SessionListMetadata> = z.object({
@@ -190,6 +191,7 @@ export class ApiSessionList {
    * @returns authorized bounded Session search results.
    */
   async search(query: string, signal: AbortSignal): Promise<SessionSearchValue> {
+    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 })
 }

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')

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()

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' } })

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

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)

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)