diff --git a/packages/agent-core-v2/src/approval/approval.ts b/packages/agent-core-v2/src/approval/approval.ts index c95dde685..6c942e49b 100644 --- a/packages/agent-core-v2/src/approval/approval.ts +++ b/packages/agent-core-v2/src/approval/approval.ts @@ -33,6 +33,12 @@ export interface ApprovalResponse { export interface IApprovalService { readonly _serviceBrand: undefined; request(req: ApprovalRequest): Promise; + /** + * Submit an approval request without blocking on the decision. Returns the + * request with its resolved `id`; the decision is delivered through the + * interaction `onDidResolve` stream. + */ + enqueue(req: ApprovalRequest): ApprovalRequest & { readonly id: string }; decide(id: string, response: ApprovalResponse): void; listPending(): readonly ApprovalRequest[]; } diff --git a/packages/agent-core-v2/src/approval/approvalService.ts b/packages/agent-core-v2/src/approval/approvalService.ts index b1c6acb21..d6d039a77 100644 --- a/packages/agent-core-v2/src/approval/approvalService.ts +++ b/packages/agent-core-v2/src/approval/approvalService.ts @@ -29,6 +29,17 @@ export class ApprovalService implements IApprovalService { }); } + enqueue(req: ApprovalRequest): ApprovalRequest & { readonly id: string } { + const id = requestId(req); + this.interaction.enqueue({ + id, + kind: 'approval', + payload: req, + origin: { agentId: req.agentId, turnId: req.turnId }, + }); + return { ...req, id }; + } + decide(id: string, response: ApprovalResponse): void { this.interaction.respond(id, response); } diff --git a/packages/agent-core-v2/src/interaction/interaction.ts b/packages/agent-core-v2/src/interaction/interaction.ts index 5a62748fb..fcbdb385d 100644 --- a/packages/agent-core-v2/src/interaction/interaction.ts +++ b/packages/agent-core-v2/src/interaction/interaction.ts @@ -4,10 +4,11 @@ * Defines the `Interaction` model and the `IInteractionService` kernel that * owns the session's pending interaction set: a unified, blocking request / * response primitive (`request` → `respond`) with change notification - * (`onDidChange`) and a `listPending` view. `approval` and `question` are - * typed specializations layered on top of this kernel; the kernel itself is - * domain-agnostic. Session-scoped — the pending set is keyed by session and - * dies with it. + * (`onDidChange`), a non-blocking enqueue (`enqueue`) for callers that observe + * the outcome through the `onDidResolve` stream, and a `listPending` view. + * `approval` and `question` are typed specializations layered on top of this + * kernel; the kernel itself is domain-agnostic. Session-scoped — the pending + * set is keyed by session and dies with it. */ import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; @@ -34,12 +35,27 @@ export interface Interaction { readonly origin: InteractionOrigin; } +/** Emitted by {@link IInteractionService.onDidResolve} when a request is responded to. */ +export interface InteractionResolution { + readonly id: string; + readonly response: unknown; +} + export interface IInteractionService { readonly _serviceBrand: undefined; request(req: InteractionRequest): Promise; + /** + * Park a request without blocking on its response. Returns the created + * `Interaction` (with its resolved `id`) immediately; the outcome is + * delivered through {@link onDidResolve}. Used by edge callers that stream + * the response rather than awaiting a Promise. + */ + enqueue(req: InteractionRequest): Interaction; respond(id: string, response: unknown): void; listPending(kind?: InteractionKind): readonly Interaction[]; readonly onDidChange: Event; + /** Fires when a pending request is responded to, carrying its id and response. */ + readonly onDidResolve: Event; } export const IInteractionService: ServiceIdentifier = diff --git a/packages/agent-core-v2/src/interaction/interactionService.ts b/packages/agent-core-v2/src/interaction/interactionService.ts index 0e6de88ea..3c82b4084 100644 --- a/packages/agent-core-v2/src/interaction/interactionService.ts +++ b/packages/agent-core-v2/src/interaction/interactionService.ts @@ -16,6 +16,7 @@ import { type InteractionKind, type InteractionOrigin, type InteractionRequest, + type InteractionResolution, IInteractionService, } from './interaction'; @@ -30,29 +31,27 @@ export class InteractionService extends Disposable implements IInteractionServic private readonly pending = new Map(); private readonly _onDidChange = this._register(new Emitter()); readonly onDidChange: Event = this._onDidChange.event; + private readonly _onDidResolve = this._register(new Emitter()); + readonly onDidResolve: Event = this._onDidResolve.event; private nextId = 0; request(req: InteractionRequest): Promise { - const id = req.id ?? this.generateId(); - const origin: InteractionOrigin = req.origin ?? {}; return new Promise((resolve) => { - const interaction: Interaction = { - id, - kind: req.kind, - payload: req.payload, - origin, - }; - this.pending.set(id, { interaction, resolve: resolve as (response: unknown) => void }); - this._onDidChange.fire(); + this.park(req, resolve as (response: unknown) => void); }); } + enqueue(req: InteractionRequest): Interaction { + return this.park(req, () => {}); + } + respond(id: string, response: unknown): void { const entry = this.pending.get(id); if (entry === undefined) return; this.pending.delete(id); entry.resolve(response); this._onDidChange.fire(); + this._onDidResolve.fire({ id, response }); } listPending(kind?: InteractionKind): readonly Interaction[] { @@ -60,6 +59,23 @@ export class InteractionService extends Disposable implements IInteractionServic return kind === undefined ? all : all.filter((i) => i.kind === kind); } + private park( + req: InteractionRequest, + resolve: (response: unknown) => void, + ): Interaction { + const id = req.id ?? this.generateId(); + const origin: InteractionOrigin = req.origin ?? {}; + const interaction: Interaction = { + id, + kind: req.kind, + payload: req.payload, + origin, + }; + this.pending.set(id, { interaction, resolve }); + this._onDidChange.fire(); + return interaction; + } + private generateId(): string { return `interaction-${this.nextId++}`; } diff --git a/packages/agent-core-v2/src/question/question.ts b/packages/agent-core-v2/src/question/question.ts index 7bb0f86d2..490acc102 100644 --- a/packages/agent-core-v2/src/question/question.ts +++ b/packages/agent-core-v2/src/question/question.ts @@ -16,6 +16,12 @@ export interface QuestionRequest { export interface IQuestionService { readonly _serviceBrand: undefined; request(req: QuestionRequest): Promise; + /** + * Post a question without blocking on the answer. Returns the request (its + * `id` is required, so it is already known); the answer is delivered through + * the interaction `onDidResolve` stream. + */ + enqueue(req: QuestionRequest): QuestionRequest; answer(id: string, answer: string): void; listPending(): readonly QuestionRequest[]; } diff --git a/packages/agent-core-v2/src/question/questionService.ts b/packages/agent-core-v2/src/question/questionService.ts index b5a8b1b80..92c63e565 100644 --- a/packages/agent-core-v2/src/question/questionService.ts +++ b/packages/agent-core-v2/src/question/questionService.ts @@ -24,6 +24,15 @@ export class QuestionService implements IQuestionService { }); } + enqueue(req: QuestionRequest): QuestionRequest { + this.interaction.enqueue({ + id: req.id, + kind: 'question', + payload: req, + }); + return req; + } + answer(id: string, answer: string): void { this.interaction.respond(id, answer); } diff --git a/packages/agent-core-v2/test/approval/approval.test.ts b/packages/agent-core-v2/test/approval/approval.test.ts index e20d6d405..c80f70b22 100644 --- a/packages/agent-core-v2/test/approval/approval.test.ts +++ b/packages/agent-core-v2/test/approval/approval.test.ts @@ -42,4 +42,13 @@ describe('ApprovalService', () => { const svc = ix.get(IApprovalService); expect(() => svc.decide('missing', { decision: 'rejected' })).not.toThrow(); }); + + it('enqueue parks a request and returns it with its id without blocking', () => { + const svc = ix.get(IApprovalService); + const enqueued = svc.enqueue(makeRequest('r1')); + expect(enqueued).toEqual({ ...makeRequest('r1'), id: 'r1' }); + expect(svc.listPending()).toEqual([makeRequest('r1')]); + svc.decide('r1', { decision: 'approved' }); + expect(svc.listPending()).toEqual([]); + }); }); diff --git a/packages/agent-core-v2/test/interaction/interaction.test.ts b/packages/agent-core-v2/test/interaction/interaction.test.ts index be100ebc7..34d48efb8 100644 --- a/packages/agent-core-v2/test/interaction/interaction.test.ts +++ b/packages/agent-core-v2/test/interaction/interaction.test.ts @@ -62,4 +62,42 @@ describe('InteractionService', () => { const svc = ix.get(IInteractionService); expect(() => svc.respond('nope', 'x')).not.toThrow(); }); + + it('enqueue parks a request and returns it without blocking', () => { + const svc = ix.get(IInteractionService); + const interaction = svc.enqueue({ id: 'e1', kind: 'approval', payload: { tool: 'bash' } }); + expect(interaction).toMatchObject({ + id: 'e1', + kind: 'approval', + payload: { tool: 'bash' }, + }); + expect(svc.listPending()).toHaveLength(1); + }); + + it('enqueue generates an id when none is provided', () => { + const svc = ix.get(IInteractionService); + const interaction = svc.enqueue({ kind: 'question', payload: {} }); + expect(interaction.id).toMatch(/^interaction-/); + expect(svc.listPending()[0]!.id).toBe(interaction.id); + }); + + it('onDidResolve fires with the id and response when responded to', () => { + const svc = ix.get(IInteractionService); + const seen: { id: string; response: unknown }[] = []; + disposables.add(svc.onDidResolve((r) => seen.push(r))); + + svc.enqueue({ id: 'e1', kind: 'approval', payload: {} }); + svc.respond('e1', { decision: 'approved' }); + + expect(seen).toEqual([{ id: 'e1', response: { decision: 'approved' } }]); + expect(svc.listPending()).toHaveLength(0); + }); + + it('onDidResolve does not fire for an unknown id', () => { + const svc = ix.get(IInteractionService); + let count = 0; + disposables.add(svc.onDidResolve(() => count++)); + svc.respond('nope', 'x'); + expect(count).toBe(0); + }); }); diff --git a/packages/agent-core-v2/test/permission/stubs.ts b/packages/agent-core-v2/test/permission/stubs.ts index 1df8c16fa..caf8a8e45 100644 --- a/packages/agent-core-v2/test/permission/stubs.ts +++ b/packages/agent-core-v2/test/permission/stubs.ts @@ -80,6 +80,7 @@ export function stubApprovalService( return { _serviceBrand: undefined, request: () => Promise.resolve(next()), + enqueue: (req) => ({ ...req, id: req.id ?? req.toolCallId ?? 'stub' }), decide: () => {}, listPending: () => [], }; diff --git a/packages/agent-core-v2/test/question/question.test.ts b/packages/agent-core-v2/test/question/question.test.ts index e0b0fe126..b4193ee45 100644 --- a/packages/agent-core-v2/test/question/question.test.ts +++ b/packages/agent-core-v2/test/question/question.test.ts @@ -28,4 +28,13 @@ describe('QuestionService', () => { await expect(p).resolves.toBe('kimi'); expect(svc.listPending()).toEqual([]); }); + + it('enqueue parks a question without blocking', () => { + const svc = ix.get(IQuestionService); + const enqueued = svc.enqueue({ id: 'q1', prompt: 'name?' }); + expect(enqueued).toEqual({ id: 'q1', prompt: 'name?' }); + expect(svc.listPending()).toEqual([{ id: 'q1', prompt: 'name?' }]); + svc.answer('q1', 'kimi'); + expect(svc.listPending()).toEqual([]); + }); }); diff --git a/packages/server-v2/src/transport/actionMap.ts b/packages/server-v2/src/transport/actionMap.ts index eb5c3bcfa..09b438607 100644 --- a/packages/server-v2/src/transport/actionMap.ts +++ b/packages/server-v2/src/transport/actionMap.ts @@ -117,9 +117,11 @@ export const actionMap: Record> = { 'session:archive': { service: ISessionService, method: 'archive' }, 'approvals:listPending': { service: IApprovalService, method: 'listPending', readonly: true }, + 'approvals:request': { service: IApprovalService, method: 'enqueue' }, 'approvals:decide': { service: IApprovalService, method: 'decide' }, 'questions:listPending': { service: IQuestionService, method: 'listPending', readonly: true }, + 'questions:ask': { service: IQuestionService, method: 'enqueue' }, 'questions:answer': { service: IQuestionService, method: 'answer' }, 'interactions:listPending': { @@ -127,6 +129,7 @@ export const actionMap: Record> = { method: 'listPending', readonly: true, }, + 'interactions:request': { service: IInteractionService, method: 'enqueue' }, 'interactions:respond': { service: IInteractionService, method: 'respond' }, 'workspace:resolve': { service: IWorkspaceContext, method: 'resolve', readonly: true }, diff --git a/packages/server-v2/src/transport/ws/eventMap.ts b/packages/server-v2/src/transport/ws/eventMap.ts index c3adc94c4..63247d366 100644 --- a/packages/server-v2/src/transport/ws/eventMap.ts +++ b/packages/server-v2/src/transport/ws/eventMap.ts @@ -4,13 +4,16 @@ * binds `event` → an `Event` subscription that yields an `IDisposable`. * * Only the high-value streams are exposed for now: - * Core `events` — process-wide `DomainEvent` bus (`IEventService`) - * Agent `events` — per-agent `AgentEvent` stream (`IEventSink`) + * Core `events` — process-wide `DomainEvent` bus (`IEventService`) + * Session `interactions` — pending human-in-the-loop requests (`IInteractionService.onDidChange`) + * Session `interactions:resolved` — request resolutions (`IInteractionService.onDidResolve`) + * Agent `events` — per-agent `AgentEvent` stream (`IEventSink`) */ import { IEventService, IEventSink, + IInteractionService, type DomainEvent, type IDisposable, type IScopeHandle, @@ -33,9 +36,25 @@ export const eventMap: Record> = { }, }, session: { - // Future: `metadata` (ISessionMetadata.onDidChange), `interactions` - // (IInteractionService.onDidChange). These carry no payload today, so they - // are not exposed until there is a concrete consumer. + // Pushes the full pending interaction set whenever it changes. Payload is + // the current `Interaction[]` (derived from `listPending`), so a client can + // render a pending approval/question without a follow-up `call`. Resync is + // out of band: `call interactions:listPending` before `listen`. + interactions: { + subscribe: (scope, listener) => { + const interaction = scope.accessor.get(IInteractionService); + return interaction.onDidChange(() => listener(interaction.listPending())); + }, + }, + // Pushes `{ id, response }` whenever a pending request is responded to. + // Paired with `interactions:request` (the non-blocking enqueue): a headless + // caller posts a request, then matches the resolution here by `id`. + 'interactions:resolved': { + subscribe: (scope, listener) => { + const interaction = scope.accessor.get(IInteractionService); + return interaction.onDidResolve((resolution) => listener(resolution)); + }, + }, }, agent: { events: { diff --git a/packages/server-v2/test/eventMap.test.ts b/packages/server-v2/test/eventMap.test.ts new file mode 100644 index 000000000..4921691e6 --- /dev/null +++ b/packages/server-v2/test/eventMap.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, it } from 'vitest'; + +import { IInteractionService, type IDisposable } from '@moonshot-ai/agent-core-v2'; + +import { resolveEventSource } from '../src/transport/ws/eventMap'; + +interface ManualEvent { + readonly event: (listener: (e: T) => void) => IDisposable; + readonly fire: (e: T) => void; +} + +function manualEvent(): ManualEvent { + const listeners = new Set<(e: T) => void>(); + return { + event: (listener) => { + listeners.add(listener); + return { dispose: () => listeners.delete(listener) }; + }, + fire: (e) => { + for (const l of listeners) l(e); + }, + }; +} + +describe('session `interactions` event source', () => { + it('forwards the current pending set whenever onDidChange fires', () => { + const { event, fire } = manualEvent(); + let pending: readonly unknown[] = []; + const interaction = { + onDidChange: event, + listPending: () => pending, + }; + const scope = { + accessor: { + get: (id: unknown) => (id === IInteractionService ? interaction : undefined), + }, + }; + + const source = resolveEventSource('session', 'interactions'); + expect(source).toBeDefined(); + + const seen: unknown[] = []; + const disposable = source!.subscribe(scope as never, (data) => seen.push(data)); + + pending = [{ id: 'a', kind: 'approval' }]; + fire(); + pending = []; + fire(); + + expect(seen).toEqual([[{ id: 'a', kind: 'approval' }], []]); + + disposable.dispose(); + fire(); + expect(seen).toHaveLength(2); + }); + + it('returns undefined for an unknown session event', () => { + expect(resolveEventSource('session', 'nope')).toBeUndefined(); + }); +}); + +describe('session `interactions:resolved` event source', () => { + it('forwards each resolution to the listener', () => { + const { event, fire } = manualEvent<{ id: string; response: unknown }>(); + const interaction = { + onDidResolve: event, + }; + const scope = { + accessor: { + get: (id: unknown) => (id === IInteractionService ? interaction : undefined), + }, + }; + + const source = resolveEventSource('session', 'interactions:resolved'); + expect(source).toBeDefined(); + + const seen: unknown[] = []; + const disposable = source!.subscribe(scope as never, (data) => seen.push(data)); + + fire({ id: 'e1', response: { decision: 'approved' } }); + fire({ id: 'e2', response: 'kimi' }); + + expect(seen).toEqual([ + { id: 'e1', response: { decision: 'approved' } }, + { id: 'e2', response: 'kimi' }, + ]); + + disposable.dispose(); + fire({ id: 'e3', response: null }); + expect(seen).toHaveLength(2); + }); +});