diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 56f488857f..310243f17f 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -33,6 +33,12 @@ import type {} from '@deepseek-ai/dsh-session-projection' // Type-only edges: resolve `ctx.get('commands')`, the `commands/change` event, and `ctx.get('skills')`. import type {} from '@deepseek-ai/dsh-commands' import type {} from '@deepseek-ai/dsh-skill' +import type { CallId } from '@deepseek-ai/dsh-llm/brand' +import type { ApprovalOutcome, ApprovalRequestId } from '@deepseek-ai/dsh-user-approval' +// Side-effect type import: resolves the `approval/request` waterfall and +// `ctx.get('approval')` without a value dependency on the seam (optional composition). +import type {} from '@deepseek-ai/dsh-user-approval' +import { approvalResponsePayloadSchema } from './api/approvals.schema.ts' import { questionResponsePayloadSchema } from './api/questions.schema.ts' import type { ClientResponse, RpcError, RpcReceipt, RpcRequest, RpcResponse } from './api/rpc.ts' import { RpcId } from './api/rpc.ts' @@ -123,9 +129,9 @@ class FrameQueue { } /** - * Server-side frame mint: pure pushes get a fresh rpcId per frame (stable ids - * for answerable frames belong to the approval/question registry, absent in - * this minimal version). + * Server-side frame mint: pure pushes get a fresh rpcId per frame (answerable + * frames — approval/question requested — mint their stable id in their + * pending registries instead). */ function frame(payload: F): RpcRequest { return { rpcId: RpcId(randomUUID()), payload } @@ -194,6 +200,36 @@ export interface ApiProxyDefaults { /** The tool/call payload fields the presenter path reads. */ interface ToolCallData { callId: string; name: string; arguments: string } +/** + * One outstanding approval question: the stable server-request id, the frame + * material replayed to late mux subscribers, and the resolver that settles the + * answerer's promise back into `ctx.approval`. + */ +interface PendingApproval { + rpcId: RpcId + sessionId: SessionId + approvalId: ApprovalRequestId + toolName: string + callId?: CallId + reason?: string + resolve(outcome: ApprovalOutcome): void +} + +/** Project a pending entry into its answerable mux frame (initial push and mux-open replay share it). */ +function requestedFrame(pending: PendingApproval): RpcRequest { + return { + rpcId: pending.rpcId, + payload: { + type: 'approval/requested', + sessionId: pending.sessionId, + approvalId: pending.approvalId, + toolName: pending.toolName, + ...pending.callId === undefined ? {} : { callId: pending.callId }, + ...pending.reason === undefined ? {} : { reason: pending.reason }, + }, + } +} + /** One host-owned question wait, addressed by the stable server-request id. */ interface PendingQuestion { rpcId: RpcId @@ -371,6 +407,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro /** Serializes path ownership checks with record creation across spellings. */ let workspaceCreationChain = Promise.resolve() const pendingQuestions = new Map() + const pendingApprovals = new Map() const muxQueues = new Set>>() /** @@ -519,6 +556,73 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro } }, 'api-proxy: user-interaction provider') + // --- Approval pending registry ------------------------------------------ + // The proxy is the approval channel for every agent this host owns: an ask + // through `ctx.approval` becomes an answerable server-request on the mux + // stream (stable rpcId), settled by POST /api/respond. The entry survives + // client disconnects — mux-open replays still-pending requested frames with + // the same rpcId (the refresh-recovery baseline) — and withdraws on the + // ask's own abort signal (turn cancel), pushing `cancelled` to subscribers. + if (ctx.get('approval') !== undefined) { + ctx.on('approval/request', (req, next) => { + // The audit pair `approval/asked` is already appended by the service + // before dispatch, but dispatch rides a microtask: parallel tool calls + // can append several asked events before any answerer runs. THIS + // request's event is therefore the newest asked event that is still + // undecided, unclaimed by another pending entry, and — when the ask + // names a call — carries the same callId. + const events = req.agent.session.events + const claimed = new Set() + for (const entry of pendingApprovals.values()) claimed.add(entry.approvalId) + const decided = new Set() + let approvalId: ApprovalRequestId | undefined + for (let i = events.length - 1; i >= 0; i -= 1) { + const event = events[i] as SessionEvent + if (event.type === 'approval/decided') { + decided.add(event.data.id) + } else if (event.type === 'approval/asked') { + if (decided.has(event.data.id) || claimed.has(event.data.id)) continue + if (req.callId !== undefined && event.data.callId !== req.callId) continue + approvalId = event.data.id + break + } + } + // No asked event means the request bypassed the service's audit path — + // not this channel's question; delegate to the fail-closed default. + if (approvalId === undefined) return next() + const id = approvalId + return new Promise((resolve) => { + const settle = (outcome: ApprovalOutcome): void => { + /* v8 ignore next 3 -- defensive double-settle guard: respond() routes + through the pending table (a settled id is not-pending before it can + re-settle) and the first settle removes the abort listener, so no + reachable path settles twice; kept against future settle callers. */ + if (!pendingApprovals.delete(pending.rpcId)) return + req.signal?.removeEventListener('abort', onAbort) + broadcast({ type: 'approval/resolved', sessionId: pending.sessionId, approvalId: id, outcome }) + // A cancelled ask was already settled by the service's own signal + // race, which discards this late resolution; resolving is a no-op + // there and keeps this promise from dangling forever. + resolve(outcome) + } + const onAbort = (): void => { settle('cancelled') } + const pending: PendingApproval = { + rpcId: RpcId(randomUUID()), + sessionId: req.agent.session.id, + approvalId: id, + toolName: req.toolName, + ...req.callId === undefined ? {} : { callId: req.callId }, + ...req.reason === undefined ? {} : { reason: req.reason }, + resolve: settle, + } + pendingApprovals.set(pending.rpcId, pending) + req.signal?.addEventListener('abort', onAbort, { once: true }) + const envelope = requestedFrame(pending) + for (const queue of muxQueues) queue.push(envelope) + }) + }) + } + /** * Gate the cold path on the store: an id absent from it, or naming a legacy * log without a cwd (pre-release stance: not served, no compatibility), is @@ -1150,6 +1254,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro }, }) } + // Refresh recovery: still-pending approval questions replay with their + // stable rpcId so a reconnecting client can still answer them. + for (const pending of pendingApprovals.values()) queue.push(requestedFrame(pending)) // Queue snapshot baseline (pendingQuestions precedent): frames replayed // in arrival order per session; a reconnecting client rebuilds its // queue view from these alone. @@ -1267,6 +1374,20 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro }, respond(message: ClientResponse): Promise { + // Route by the echoed rpcId (the wire correlation): approvals first, + // then questions — the two registries share one id space of UUIDs. + const approval = pendingApprovals.get(message.rpcId) + if (approval !== undefined) { + if (!message.result.ok) return Promise.resolve({ accepted: false, reason: 'bad-response' }) + const parsed = approvalResponsePayloadSchema.safeParse(message.result.value) + // The payload's audit correlation must match the entry the rpcId routed + // to — a mismatched answer is malformed, not merely late. + if (!parsed.success || parsed.data.approvalId !== approval.approvalId || parsed.data.sessionId !== approval.sessionId) { + return Promise.resolve({ accepted: false, reason: 'bad-response' }) + } + approval.resolve(parsed.data.outcome) + return Promise.resolve({ accepted: true }) + } const pending = pendingQuestions.get(message.rpcId) if (pending === undefined) return Promise.resolve({ accepted: false, reason: 'not-pending' }) if (!message.result.ok) {