diff --git a/.changeset/fix-web-question-timeout.md b/.changeset/fix-web-question-timeout.md new file mode 100644 index 000000000..962ada6fe --- /dev/null +++ b/.changeset/fix-web-question-timeout.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Stop auto-dismissing questions in the web UI after 60 seconds so they wait for the user's answer. diff --git a/apps/kimi-web/src/api/daemon/agentEventProjector.ts b/apps/kimi-web/src/api/daemon/agentEventProjector.ts index 86d9e7224..31a01c1c6 100644 --- a/apps/kimi-web/src/api/daemon/agentEventProjector.ts +++ b/apps/kimi-web/src/api/daemon/agentEventProjector.ts @@ -1280,7 +1280,6 @@ const PROTOCOL_EVENT_NAMES = new Set([ 'question.requested', 'question.answered', 'question.dismissed', - 'question.expired', // Background tasks (projected) 'task.created', 'task.progress', diff --git a/apps/kimi-web/src/api/daemon/eventReducer.ts b/apps/kimi-web/src/api/daemon/eventReducer.ts index 44f43e709..77d8831f2 100644 --- a/apps/kimi-web/src/api/daemon/eventReducer.ts +++ b/apps/kimi-web/src/api/daemon/eventReducer.ts @@ -480,8 +480,7 @@ export function reduceAppEvent( // ------------------------------------------------------------------------- case 'questionAnswered': - case 'questionDismissed': - case 'questionExpired': { + case 'questionDismissed': { const sid = event.sessionId; const qid = event.questionId; const list = next.questionsBySession[sid] ?? []; diff --git a/apps/kimi-web/src/api/daemon/mappers.ts b/apps/kimi-web/src/api/daemon/mappers.ts index 2170bef43..a32697922 100644 --- a/apps/kimi-web/src/api/daemon/mappers.ts +++ b/apps/kimi-web/src/api/daemon/mappers.ts @@ -329,7 +329,6 @@ export function toAppQuestionRequest(wire: WireQuestionRequest): AppQuestionRequ turnId: wire.turn_id, toolCallId: wire.tool_call_id, questions: wire.questions.map(toAppQuestionItem), - expiresAt: wire.expires_at, createdAt: wire.created_at, }; } @@ -645,13 +644,6 @@ export function toAppEvent(wire: WireEvent): AppEvent { dismissedAt: w.payload.dismissed_at, }; - case 'event.question.expired': - return { - type: 'questionExpired', - sessionId: w.session_id, - questionId: w.payload.question_id, - }; - // ----- Background tasks ----- case 'event.task.created': return { diff --git a/apps/kimi-web/src/api/daemon/wire.ts b/apps/kimi-web/src/api/daemon/wire.ts index 314cb96cf..3433018d7 100644 --- a/apps/kimi-web/src/api/daemon/wire.ts +++ b/apps/kimi-web/src/api/daemon/wire.ts @@ -270,7 +270,6 @@ export interface WireQuestionRequest { turn_id?: number; tool_call_id?: string; questions: WireQuestionItem[]; - expires_at: string; created_at: string; } @@ -739,8 +738,6 @@ type WireEventQuestionDismissed = WireEventBase<'event.question.dismissed', { dismissed_by: string; dismissed_at: string; }>; -type WireEventQuestionExpired = WireEventBase<'event.question.expired', { question_id: string }>; - // Background tasks type WireEventTaskCreated = WireEventBase<'event.task.created', { task: WireBackgroundTask }>; type WireEventTaskProgress = WireEventBase<'event.task.progress', { @@ -802,7 +799,6 @@ export type WireEvent = | WireEventQuestionRequested | WireEventQuestionAnswered | WireEventQuestionDismissed - | WireEventQuestionExpired // Background tasks | WireEventTaskCreated | WireEventTaskProgress diff --git a/apps/kimi-web/src/api/types.ts b/apps/kimi-web/src/api/types.ts index ad52b6e0d..84cdd9ae4 100644 --- a/apps/kimi-web/src/api/types.ts +++ b/apps/kimi-web/src/api/types.ts @@ -272,7 +272,6 @@ export interface AppQuestionRequest { turnId?: number; toolCallId?: string; questions: QuestionItem[]; - expiresAt: string; createdAt: string; } @@ -415,7 +414,6 @@ export type AppEvent = | { type: 'questionRequested'; sessionId: string; question: AppQuestionRequest } | { type: 'questionAnswered'; sessionId: string; questionId: string; resolvedAt: string } | { type: 'questionDismissed'; sessionId: string; questionId: string; dismissedAt: string } - | { type: 'questionExpired'; sessionId: string; questionId: string } | { type: 'taskCreated'; sessionId: string; task: AppTask } | { type: 'taskProgress'; sessionId: string; taskId: string; outputChunk: string; stream: 'stdout' | 'stderr' } | { type: 'taskCompleted'; sessionId: string; taskId: string; status: AppTaskStatus; outputPreview?: string; outputBytes?: number } diff --git a/packages/agent-core/src/rpc/client.ts b/packages/agent-core/src/rpc/client.ts index d50cb5f3f..a4a0b0b28 100644 --- a/packages/agent-core/src/rpc/client.ts +++ b/packages/agent-core/src/rpc/client.ts @@ -55,7 +55,9 @@ export function createRPC, Right extends Record signal?.throwIfAborted(); let response: RpcResponse; try { - const value = await abortableRpc(Promise.resolve(fn(rpcPayload)), signal); + const handlerResult = + signal === undefined ? fn(rpcPayload) : fn(rpcPayload, { signal }); + const value = await abortableRpc(Promise.resolve(handlerResult), signal); response = { ok: true, value }; } catch (error) { signal?.throwIfAborted(); diff --git a/packages/agent-core/src/services/coreProcess/coreProcessClient.ts b/packages/agent-core/src/services/coreProcess/coreProcessClient.ts index 24e78e15c..11014b73f 100644 --- a/packages/agent-core/src/services/coreProcess/coreProcessClient.ts +++ b/packages/agent-core/src/services/coreProcess/coreProcessClient.ts @@ -53,8 +53,9 @@ export class BridgeClientAPI implements SDKAPI { async requestQuestion( request: QuestionRequest & { sessionId: string; agentId: string }, + options?: { signal?: AbortSignal }, ): Promise { - return this.deps.questionService.request(request); + return this.deps.questionService.request(request, options); } async toolCall( diff --git a/packages/agent-core/src/services/question/question.ts b/packages/agent-core/src/services/question/question.ts index 23fbc9185..da936ace4 100644 --- a/packages/agent-core/src/services/question/question.ts +++ b/packages/agent-core/src/services/question/question.ts @@ -66,7 +66,10 @@ export interface IQuestionService { * Resolves with the in-process `QuestionResult` (null = no handler / fully * dismissed). Concrete impls own timeout policy. */ - request(req: InProcessQuestionRequest & { sessionId: string; agentId: string }): Promise; + request( + req: InProcessQuestionRequest & { sessionId: string; agentId: string }, + options?: { signal?: AbortSignal }, + ): Promise; /** * Called by the answer-side (REST handler / TUI / mock) to settle a pending @@ -104,8 +107,6 @@ export interface QuestionToBrokerRequestParams { readonly sessionId: string; /** `createdAt` ISO string; broker passes `new Date().toISOString()`. */ readonly createdAt: string; - /** `expiresAt` ISO string; broker computes `createdAt + 60s`. */ - readonly expiresAt: string; } /** @@ -161,7 +162,6 @@ export function toBrokerRequest( session_id: params.sessionId, questions: req.questions.map((q, i) => buildItem(q, i)), created_at: params.createdAt, - expires_at: params.expiresAt, }; if (req.turnId !== undefined) out.turn_id = req.turnId; if (req.toolCallId !== undefined) out.tool_call_id = req.toolCallId; diff --git a/packages/agent-core/src/services/session/sessionService.ts b/packages/agent-core/src/services/session/sessionService.ts index cf98d3ee4..3af85e326 100644 --- a/packages/agent-core/src/services/session/sessionService.ts +++ b/packages/agent-core/src/services/session/sessionService.ts @@ -223,8 +223,7 @@ export class SessionService extends Disposable implements ISessionService { case 'event.approval.expired': case 'event.question.requested': case 'event.question.answered': - case 'event.question.dismissed': - case 'event.question.expired': { + case 'event.question.dismissed': { this._emitStatusChanged(sessionId); break; } diff --git a/packages/agent-core/test/services/question-adapter.test.ts b/packages/agent-core/test/services/question-adapter.test.ts index e5a6b5d4f..b74ceb883 100644 --- a/packages/agent-core/test/services/question-adapter.test.ts +++ b/packages/agent-core/test/services/question-adapter.test.ts @@ -48,7 +48,6 @@ describe('question-adapter · toBrokerRequest (in-process → protocol)', () => questionId: '01J_QUESTION', sessionId: 'sess_x', createdAt: '2026-06-04T10:30:00.000Z', - expiresAt: '2026-06-04T10:31:00.000Z', }); expect(protoReq.question_id).toBe('01J_QUESTION'); @@ -91,7 +90,6 @@ describe('question-adapter · toBrokerRequest (in-process → protocol)', () => questionId: 'q', sessionId: 's', createdAt: '2026-06-04T10:30:00.000Z', - expiresAt: '2026-06-04T10:31:00.000Z', }); expect(protoReq.turn_id).toBeUndefined(); expect(protoReq.tool_call_id).toBeUndefined(); diff --git a/packages/protocol/src/__tests__/question.test.ts b/packages/protocol/src/__tests__/question.test.ts index e4353e99a..95e8ab932 100644 --- a/packages/protocol/src/__tests__/question.test.ts +++ b/packages/protocol/src/__tests__/question.test.ts @@ -96,7 +96,6 @@ describe('questionRequestSchema (SCHEMAS §6.2)', () => { }, ], created_at: '2026-06-04T10:30:00Z', - expires_at: '2026-06-04T10:31:00Z', }; it('accepts a 1-question request', () => { @@ -262,7 +261,6 @@ describe('listPendingQuestionsResponseSchema (REST pending recovery)', () => { }, ], created_at: '2026-06-04T10:30:00Z', - expires_at: '2026-06-04T10:31:00Z', }; it('accepts status=pending query', () => { diff --git a/packages/protocol/src/question.ts b/packages/protocol/src/question.ts index 3bfdf8c9f..66afbd013 100644 --- a/packages/protocol/src/question.ts +++ b/packages/protocol/src/question.ts @@ -29,7 +29,6 @@ export const questionRequestSchema = z.object({ tool_call_id: z.string().min(1).optional(), questions: z.array(questionItemSchema).min(1).max(4), created_at: isoDateTimeSchema, - expires_at: isoDateTimeSchema, }); export type QuestionRequest = z.infer; diff --git a/packages/server-e2e/scenarios/08-pending-recovery.ts b/packages/server-e2e/scenarios/08-pending-recovery.ts index 584cb58b1..f69a860cc 100644 --- a/packages/server-e2e/scenarios/08-pending-recovery.ts +++ b/packages/server-e2e/scenarios/08-pending-recovery.ts @@ -37,7 +37,6 @@ interface QuestionRequestedPayload { options: Array<{ id: string; label: string }>; }>; created_at: string; - expires_at: string; } async function main() { diff --git a/packages/server/src/services/approval/approvalService.ts b/packages/server/src/services/approval/approvalService.ts index 29577acbd..92986e31e 100644 --- a/packages/server/src/services/approval/approvalService.ts +++ b/packages/server/src/services/approval/approvalService.ts @@ -16,13 +16,6 @@ export const APPROVAL_DEFAULT_TIMEOUT_MS = 60_000; export const APPROVAL_RECENTLY_RESOLVED_CAP = 1024; -export class ApprovalExpiredError extends Error { - constructor(public readonly approvalId: string, timeoutMs: number) { - super(`approval ${approvalId} expired after ${timeoutMs}ms`); - this.name = 'ApprovalExpiredError'; - } -} - class PendingApproval implements IDisposable { private _settled = false; @@ -35,13 +28,11 @@ class PendingApproval implements IDisposable { readonly protocolRequest: ProtocolApprovalRequest, private readonly _resolveFn: (r: ApprovalResponse) => void, private readonly _rejectFn: (e: Error) => void, - private readonly _timer: NodeJS.Timeout, ) {} markSettled(): void { if (this._settled) return; this._settled = true; - clearTimeout(this._timer); } resolve(r: ApprovalResponse): void { @@ -55,7 +46,6 @@ class PendingApproval implements IDisposable { dispose(): void { if (this._settled) return; this._settled = true; - clearTimeout(this._timer); try { this._rejectFn(new Error('server shutting down')); } catch { @@ -72,7 +62,6 @@ export class ApprovalService extends Disposable implements IApprovalService { private readonly _byToolCallId = new Map(); private readonly _recentlyResolved = new Set(); - private _timeoutMs = APPROVAL_DEFAULT_TIMEOUT_MS; private readonly _recentlyResolvedCap = APPROVAL_RECENTLY_RESOLVED_CAP; constructor( @@ -81,6 +70,39 @@ export class ApprovalService extends Disposable implements IApprovalService { ) { super(); this._pending = this._register(new DisposableMap()); + + // The turn's abort signal never reaches this broker: agent-core's + // `BridgeClientAPI.requestApproval` drops the `{ signal }` option, so an + // aborted turn would otherwise leave the approval in `_pending` forever + // (pinning the session in `awaiting_approval` and keeping the web panel + // open). Settle stale approvals when the in-process bus reports the turn + // ended for a cancellation reason. This is intentionally session-scoped: a + // turn has at most one pending approval, and on normal completion the + // approval is already resolved so this is a no-op. + this._register( + this.eventService.onDidPublish((event) => { + if ((event as { type?: string }).type !== 'turn.ended') return; + const reason = (event as { reason?: string }).reason; + if (reason !== 'cancelled' && reason !== 'failed' && reason !== 'filtered') return; + const sessionId = (event as { sessionId?: string }).sessionId; + if (sessionId === undefined || sessionId === '') return; + this.dismissForSession(sessionId); + }), + ); + } + + private dismissForSession(sessionId: string): void { + const ids: string[] = []; + for (const p of this._pending.values()) { + if (p.sessionId === sessionId) ids.push(p.approvalId); + } + for (const id of ids) { + // Reuse resolve(): clears `_pending` / `_byToolCallId` and publishes + // `event.approval.resolved` (decision: 'cancelled') so the web panel + // closes. The agent-core caller's promise is already rejected by the + // abort, so the resolved value is only observed by tests. + this.resolve(id, { decision: 'cancelled' }); + } } async request( @@ -92,7 +114,10 @@ export class ApprovalService extends Disposable implements IApprovalService { const approvalId = ulid(); const createdAt = new Date().toISOString(); - const expiresAt = new Date(Date.now() + this._timeoutMs).toISOString(); + // `expires_at` is still populated for the protocol/web contract, but the + // broker no longer enforces it — approvals wait until the user resolves + // them or the server shuts down. + const expiresAt = new Date(Date.now() + APPROVAL_DEFAULT_TIMEOUT_MS).toISOString(); const protocolRequest = approvalToBrokerRequest(req, { approvalId, @@ -121,8 +146,6 @@ export class ApprovalService extends Disposable implements IApprovalService { ); return new Promise((resolve, reject) => { - const timer = setTimeout(() => this._expire(approvalId), this._timeoutMs); - timer.unref?.(); this._pending.set( approvalId, new PendingApproval( @@ -134,7 +157,6 @@ export class ApprovalService extends Disposable implements IApprovalService { protocolRequest, resolve, reject, - timer, ), ); this._byToolCallId.set(req.toolCallId, approvalId); @@ -199,30 +221,6 @@ export class ApprovalService extends Disposable implements IApprovalService { return { sessionId: p.sessionId, toolCallId: p.toolCallId }; } - _setTimeoutMsForTests(ms: number): void { - this._timeoutMs = ms; - } - - private _expire(approvalId: string): void { - const p = this._pending.get(approvalId); - if (!p) return; - p.markSettled(); - this._pending.deleteAndLeak(approvalId); - this._byToolCallId.delete(p.toolCallId); - - this.markResolved(p.approvalId); - - const expiredEvent: Event = { - type: 'event.approval.expired', - sessionId: p.sessionId, - agentId: 'main', - approval_id: p.approvalId, - } as unknown as Event; - this.eventService.publish(expiredEvent); - - p.reject(new ApprovalExpiredError(p.approvalId, this._timeoutMs)); - } - override dispose(): void { if (this._store.isDisposed) return; this._byToolCallId.clear(); diff --git a/packages/server/src/services/approval/index.ts b/packages/server/src/services/approval/index.ts index 1978d7181..49ff011d8 100644 --- a/packages/server/src/services/approval/index.ts +++ b/packages/server/src/services/approval/index.ts @@ -1,6 +1,5 @@ export { ApprovalService, - ApprovalExpiredError, APPROVAL_DEFAULT_TIMEOUT_MS, APPROVAL_RECENTLY_RESOLVED_CAP, } from './approvalService'; diff --git a/packages/server/src/services/question/index.ts b/packages/server/src/services/question/index.ts index 95450a252..78b26f4a1 100644 --- a/packages/server/src/services/question/index.ts +++ b/packages/server/src/services/question/index.ts @@ -1,6 +1,4 @@ export { QuestionService, - QuestionExpiredError, - QUESTION_DEFAULT_TIMEOUT_MS, QUESTION_RECENTLY_RESOLVED_CAP, } from './questionService'; diff --git a/packages/server/src/services/question/questionService.ts b/packages/server/src/services/question/questionService.ts index 6b9b2d142..569f1e5d2 100644 --- a/packages/server/src/services/question/questionService.ts +++ b/packages/server/src/services/question/questionService.ts @@ -12,36 +12,31 @@ import type { // eslint-disable-next-line @typescript-eslint/no-unused-vars const _typeAnchor: typeof IQuestionService = IQuestionService; -export const QUESTION_DEFAULT_TIMEOUT_MS = 60_000; - export const QUESTION_RECENTLY_RESOLVED_CAP = 1024; -export class QuestionExpiredError extends Error { - constructor(public readonly questionId: string, timeoutMs: number) { - super(`question ${questionId} expired after ${timeoutMs}ms`); - this.name = 'QuestionExpiredError'; - } -} - class PendingQuestion implements IDisposable { private _settled = false; + private _abortCleanup: (() => void) | undefined; constructor( readonly questionId: string, readonly sessionId: string, readonly toolCallId: string | undefined, readonly createdAt: string, - readonly expiresAt: string, readonly protocolRequest: ProtocolQuestionRequest, private readonly _resolveFn: (r: QuestionResult) => void, private readonly _rejectFn: (e: Error) => void, - private readonly _timer: NodeJS.Timeout, ) {} + setAbortCleanup(cleanup: () => void): void { + this._abortCleanup = cleanup; + } + markSettled(): void { if (this._settled) return; this._settled = true; - clearTimeout(this._timer); + this._abortCleanup?.(); + this._abortCleanup = undefined; } resolve(r: QuestionResult): void { @@ -55,9 +50,10 @@ class PendingQuestion implements IDisposable { dispose(): void { if (this._settled) return; this._settled = true; - clearTimeout(this._timer); + this._abortCleanup?.(); + this._abortCleanup = undefined; try { - this._rejectFn(new Error('server shutting down')); + this.reject(new Error('server shutting down')); } catch { } @@ -70,7 +66,6 @@ export class QuestionService extends Disposable implements IQuestionService { private readonly _pending: DisposableMap; private readonly _recentlyResolved = new Set(); - private _timeoutMs = QUESTION_DEFAULT_TIMEOUT_MS; private readonly _recentlyResolvedCap = QUESTION_RECENTLY_RESOLVED_CAP; constructor( @@ -83,6 +78,7 @@ export class QuestionService extends Disposable implements IQuestionService { async request( req: QuestionRequest & { sessionId: string; agentId: string }, + options?: { signal?: AbortSignal }, ): Promise { if (this._store.isDisposed) { throw new Error('question service disposed'); @@ -90,13 +86,11 @@ export class QuestionService extends Disposable implements IQuestionService { const questionId = ulid(); const createdAt = new Date().toISOString(); - const expiresAt = new Date(Date.now() + this._timeoutMs).toISOString(); const protocolRequest = questionToBrokerRequest(req, { questionId, sessionId: req.sessionId, createdAt, - expiresAt, }); const event: Event = { @@ -119,22 +113,29 @@ export class QuestionService extends Disposable implements IQuestionService { ); return new Promise((resolve, reject) => { - const timer = setTimeout(() => this._expire(questionId), this._timeoutMs); - timer.unref?.(); - this._pending.set( + const pending = new PendingQuestion( questionId, - new PendingQuestion( - questionId, - req.sessionId, - req.toolCallId, - createdAt, - expiresAt, - protocolRequest, - resolve, - reject, - timer, - ), + req.sessionId, + req.toolCallId, + createdAt, + protocolRequest, + resolve, + reject, ); + this._pending.set(questionId, pending); + + // When the agent's turn is aborted, the broker entry must be settled so + // listPending()/session status don't stay stuck in awaiting_question. + const signal = options?.signal; + if (signal !== undefined) { + if (signal.aborted) { + this.dismiss(questionId); + } else { + const onAbort = () => this.dismiss(questionId); + signal.addEventListener('abort', onAbort, { once: true }); + pending.setAbortCleanup(() => signal.removeEventListener('abort', onAbort)); + } + } }); } @@ -214,28 +215,6 @@ export class QuestionService extends Disposable implements IQuestionService { return { sessionId: p.sessionId, toolCallId: p.toolCallId }; } - _setTimeoutMsForTests(ms: number): void { - this._timeoutMs = ms; - } - - private _expire(questionId: string): void { - const p = this._pending.get(questionId); - if (!p) return; - p.markSettled(); - this._pending.deleteAndLeak(questionId); - this.markResolved(p.questionId); - - const expiredEvent: Event = { - type: 'event.question.expired', - sessionId: p.sessionId, - agentId: 'main', - question_id: p.questionId, - } as unknown as Event; - this.eventService.publish(expiredEvent); - - p.reject(new QuestionExpiredError(p.questionId, this._timeoutMs)); - } - override dispose(): void { if (this._store.isDisposed) return; this._recentlyResolved.clear(); diff --git a/packages/server/test/approval.e2e.test.ts b/packages/server/test/approval.e2e.test.ts index bc347df66..cf9bb8f06 100644 --- a/packages/server/test/approval.e2e.test.ts +++ b/packages/server/test/approval.e2e.test.ts @@ -15,8 +15,6 @@ * - REST `POST` resolves → broker Promise settles → response converts back * to in-process SDK shape * - Idempotency (40902) + not-found (40404) - * - 60s timeout (override to 30ms) broadcasts `event.approval.expired` AND - * rejects the Promise with `ApprovalExpiredError`. */ import { mkdtempSync, rmSync } from 'node:fs'; @@ -29,14 +27,15 @@ import { WebSocket } from 'ws'; import { IApprovalService, + IEventService, type ApprovalRequest, type ApprovalResponse, + type Event, } from '@moonshot-ai/agent-core'; import { IRestGateway, startServer, type RunningServer } from '../src'; import { rawDataToString } from '../src/ws/rawData'; import { - ApprovalExpiredError, ApprovalService, } from '#/services/approval/approvalService'; @@ -267,6 +266,67 @@ describe('Approval reverse-RPC: WS broadcast → REST resolve → Promise settle ws.close(); }); + it('aborts the pending approval when the turn ends for a cancellation reason', async () => { + const r = await bootDaemon(); + const sid = await createSession(r); + const { ws, received } = await openSubscriber(r, sid); + + const broker = r.services.invokeFunction( + (a) => a.get(IApprovalService) as ApprovalService, + ); + const eventService = r.services.invokeFunction((a) => a.get(IEventService)); + + const pending = broker.request({ + sessionId: sid, + agentId: 'main', + turnId: 21, + toolCallId: 'tc_approval_abort', + toolName: 'shell.run', + action: 'Run `ls`', + display: { kind: 'command', command: 'ls', summary: 'ls' } as never, + }); + + const requested = await waitFor( + received, + (f) => f['type'] === 'event.approval.requested', + 2000, + ); + const payload = requested['payload'] as { approval_id: string }; + expect(broker.isPending(payload.approval_id)).toBe(true); + + // Simulate the agent's turn being aborted before the user resolves the + // approval. The broker subscribes to the in-process bus because the turn + // abort signal never reaches it through the RPC boundary. + eventService.publish({ + type: 'turn.ended', + sessionId: sid, + agentId: 'main', + turnId: 21, + reason: 'cancelled', + } as unknown as Event); + + const resolvedFrame = await waitFor( + received, + (f) => f['type'] === 'event.approval.resolved', + 2000, + ); + const resolvedPayload = resolvedFrame['payload'] as { + approval_id: string; + decision: string; + }; + expect(resolvedPayload.approval_id).toBe(payload.approval_id); + expect(resolvedPayload.decision).toBe('cancelled'); + + // Broker entry is cleaned up so listPending()/session status don't stick + // in awaiting_approval for a dead turn. + expect(broker.isPending(payload.approval_id)).toBe(false); + + const inProcResp: ApprovalResponse = await pending; + expect(inProcResp.decision).toBe('cancelled'); + + ws.close(); + }); + it('GET pending approvals lists recoverable requests and omits resolved ones', async () => { const r = await bootDaemon(); const sid = await createSession(r); @@ -359,52 +419,6 @@ describe('Approval reverse-RPC: WS broadcast → REST resolve → Promise settle expect(env.code).toBe(40001); }); - it('60s timeout broadcasts event.approval.expired + rejects with ApprovalExpiredError', async () => { - const r = await bootDaemon(); - const sid = await createSession(r); - const { ws, received } = await openSubscriber(r, sid); - - // Swap in a short-timeout broker for this session — clean by reaching into - // the container, since startServer doesn't expose a broker-options - // override yet. - const broker = r.services.invokeFunction( - (a) => a.get(IApprovalService) as ApprovalService, - ); - // Stamp the timeout via a private field hack — the test already - // co-owns the impl. (In a fuller world we'd thread a `brokerOptions` - // option through ServerStartOptions.) - (broker as unknown as { _timeoutMs: number })._timeoutMs = 40; - - const pending = broker.request({ - sessionId: sid, - agentId: 'main', - toolCallId: 'tc_timeout', - toolName: 'shell.run', - action: 'Run', - display: { kind: 'generic', summary: 'test' } as never, - turnId: 1, - }); - - // Expect a rejection AND an event.approval.expired frame. - let rejection: unknown; - try { - await pending; - } catch (err) { - rejection = err; - } - expect(rejection).toBeInstanceOf(ApprovalExpiredError); - - const expiredFrame = await waitFor( - received, - (f) => f['type'] === 'event.approval.expired', - 2000, - ); - const payload = expiredFrame['payload'] as { approval_id: string }; - expect(payload.approval_id).toMatch(/^[0-9A-HJKMNP-TV-Z]{26}$/); - - ws.close(); - }); - it('REST resolve on unknown approval_id returns 40404', async () => { const r = await bootDaemon(); const sid = await createSession(r); diff --git a/packages/server/test/question.e2e.test.ts b/packages/server/test/question.e2e.test.ts index 2c56760ed..833daf0f6 100644 --- a/packages/server/test/question.e2e.test.ts +++ b/packages/server/test/question.e2e.test.ts @@ -26,10 +26,7 @@ import { import { IRestGateway, startServer, type RunningServer } from '../src'; import { rawDataToString } from '../src/ws/rawData'; -import { - QuestionService, - QuestionExpiredError, -} from '#/services/question/questionService'; +import { QuestionService } from '#/services/question/questionService'; let tmpDir: string; let lockPath: string; @@ -297,7 +294,6 @@ describe('Question reverse-RPC: WS broadcast → REST resolve → Promise settle other_label?: string; }>; created_at: string; - expires_at: string; }>; }>(res.json()); expect(env.code).toBe(0); @@ -320,7 +316,6 @@ describe('Question reverse-RPC: WS broadcast → REST resolve → Promise settle { id: 'opt_0_1', label: 'No' }, ]); expect(item?.created_at).toMatch(/^\d{4}-\d{2}-\d{2}T/); - expect(item?.expires_at).toMatch(/^\d{4}-\d{2}-\d{2}T/); const dismissed = await appOf(r).inject({ method: 'POST', @@ -485,6 +480,52 @@ describe('Question reverse-RPC: WS broadcast → REST resolve → Promise settle ws.close(); }); + it('aborts the pending question when the request signal aborts', async () => { + const r = await bootDaemon(); + const sid = await createSession(r); + const { ws, received } = await openSubscriber(r, sid); + + const broker = r.services.invokeFunction( + (a) => a.get(IQuestionService) as QuestionService, + ); + + const controller = new AbortController(); + const pending = broker.request( + { + sessionId: sid, + agentId: 'main', + questions: [{ question: '?', options: [{ label: 'A' }, { label: 'B' }] }], + }, + { signal: controller.signal }, + ); + + const requested = await waitFor( + received, + (f) => f['type'] === 'event.question.requested', + 2000, + ); + const payload = requested['payload'] as { question_id: string }; + + // Simulate the turn being aborted before the user answers. + controller.abort(); + + const dismissedFrame = await waitFor( + received, + (f) => f['type'] === 'event.question.dismissed', + 2000, + ); + const dPayload = dismissedFrame['payload'] as { question_id: string }; + expect(dPayload.question_id).toBe(payload.question_id); + + // Broker entry is cleaned up so listPending/session status don't stick. + expect(broker.isPending(payload.question_id)).toBe(false); + + const result: QuestionResult = await pending; + expect(result).toBeNull(); + + ws.close(); + }); + it('REST resolve on unknown question_id returns 40405', async () => { const r = await bootDaemon(); const sid = await createSession(r); @@ -510,6 +551,8 @@ describe('Question reverse-RPC: WS broadcast → REST resolve → Promise settle }); it('REST re-resolve on already-resolved question returns 40902 with data:{resolved:false}', async () => { + + const r = await bootDaemon(); const sid = await createSession(r); @@ -586,41 +629,4 @@ describe('Question reverse-RPC: WS broadcast → REST resolve → Promise settle // Cleanup so the test doesn't leave a hanging Promise. broker.dismiss(questionId!); }); - - it('60s timeout broadcasts event.question.expired + rejects with QuestionExpiredError', async () => { - const r = await bootDaemon(); - const sid = await createSession(r); - const { ws, received } = await openSubscriber(r, sid); - - const broker = r.services.invokeFunction( - (a) => a.get(IQuestionService) as QuestionService, - ); - (broker as unknown as { _timeoutMs: number })._timeoutMs = 40; - - const pending = broker.request({ - sessionId: sid, - agentId: 'main', - questions: [ - { question: '?', options: [{ label: 'A' }, { label: 'B' }] }, - ], - }); - - let rejection: unknown; - try { - await pending; - } catch (err) { - rejection = err; - } - expect(rejection).toBeInstanceOf(QuestionExpiredError); - - const expiredFrame = await waitFor( - received, - (f) => f['type'] === 'event.question.expired', - 2000, - ); - const payload = expiredFrame['payload'] as { question_id: string }; - expect(payload.question_id).toMatch(/^[0-9A-HJKMNP-TV-Z]{26}$/); - - ws.close(); - }); }); diff --git a/packages/server/test/services.test.ts b/packages/server/test/services.test.ts index 5444f7667..f72e7cf6d 100644 --- a/packages/server/test/services.test.ts +++ b/packages/server/test/services.test.ts @@ -578,31 +578,6 @@ describe('ApprovalService (broadcasts + resolve-by-approval_id)', () => { bus.dispose(); }); - it('rejects with ApprovalExpiredError + broadcasts event.approval.expired after timeoutMs', async () => { - const { broker, bus, broadcast, conn } = makeBrokerWithBus(); - broker._setTimeoutMsForTests(30); - const pending = broker.request({ - sessionId: 'sess_1', - agentId: 'agent_1', - toolCallId: 'tc_timeout', - toolName: 'shell.run', - action: 'Run', - display: { kind: 'generic', summary: 'test' }, - } as Parameters[0]); - - await expect(pending).rejects.toMatchObject({ - name: 'ApprovalExpiredError', - }); - await broadcast._drainForTest('sess_1'); - const expiredFrame = conn.sent.find( - (f) => (f as { type: string }).type === 'event.approval.expired', - ); - expect(expiredFrame).toBeDefined(); - broker.dispose(); - broadcast.dispose(); - bus.dispose(); - }); - it('dispose rejects all pending requests with "server shutting down"', async () => { const { broker, bus, broadcast } = makeBrokerWithBus(); const p1 = broker.request({ @@ -733,29 +708,6 @@ describe('QuestionService (broadcasts + dismiss)', () => { bus.dispose(); }); - it('60s timeout broadcasts event.question.expired + rejects QuestionExpiredError', async () => { - const { broker, bus, broadcast, conn } = makeQuestionBroker(); - broker._setTimeoutMsForTests(30); - const pending = broker.request({ - sessionId: 's', - agentId: 'a', - questions: [ - { question: '?', options: [{ label: 'A' }, { label: 'B' }] }, - ], - } as Parameters[0]); - - await expect(pending).rejects.toMatchObject({ name: 'QuestionExpiredError' }); - await broadcast._drainForTest('s'); - const expiredFrame = conn.sent.find( - (f) => (f as { type: string }).type === 'event.question.expired', - ); - expect(expiredFrame).toBeDefined(); - - broker.dispose(); - broadcast.dispose(); - bus.dispose(); - }); - it('dispose rejects pending question Promises', async () => { const { broker, bus, broadcast } = makeQuestionBroker(); const pending = broker.request({