diff options
| author | Adam Malczewski <[email protected]> | 2026-06-21 02:08:44 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-21 02:08:44 +0900 |
| commit | ba47df37f0c89bff4f0c3dd7d0bc2ef6c8062b92 (patch) | |
| tree | 21d87eb847cd526a506cf274467fd1359f349705 /packages/message-queue/src/service.ts | |
| parent | 75032313a96856a932c109efbbe6b6a7eb782222 (diff) | |
| download | dispatch-ba47df37f0c89bff4f0c3dd7d0bc2ef6c8062b92.tar.gz dispatch-ba47df37f0c89bff4f0c3dd7d0bc2ef6c8062b92.zip | |
feat(message-queue): per-conversation queue + steering injection
A per-conversation message queue (new message-queue extension) holds user
messages enqueued while a turn generates; delivered mid-turn as steering at the
tool-result boundary (or carried to a new turn if no tool call fires).
- kernel: RunTurnInput.drainSteering callback (generic; kernel stays pure)
- wire 0.7.0->0.8.0: QueuedMessage, QueuePayload, TurnSteeringEvent (additive)
- transport-contract 0.11.0->0.12.0: POST /conversations/:id/queue + chat.queue WS op
- message-queue ext: queue state + per-conversation custom surface (rendererId message-queue)
- session-orchestrator: enqueue facade + drainSteering wiring + post-seal carry
- transport-http/ws: queue endpoint + chat.queue op (fixes WsClientMessage exhaustive switch)
- host-bin: register message-queue
1043 vitest + 199 transport bun pass; tsc/biome clean; boot smoke clean.
FE courier: frontend-message-queue-handoff.md.
Diffstat (limited to 'packages/message-queue/src/service.ts')
| -rw-r--r-- | packages/message-queue/src/service.ts | 88 |
1 files changed, 88 insertions, 0 deletions
diff --git a/packages/message-queue/src/service.ts b/packages/message-queue/src/service.ts new file mode 100644 index 0000000..91d0fc5 --- /dev/null +++ b/packages/message-queue/src/service.ts @@ -0,0 +1,88 @@ +/** + * Message-queue service — the typed handle the session-orchestrator obtains + * (lazily) to enqueue / read / drain a conversation's steering queue. + * + * The queue is transient + per-conversation (in-memory, only meaningful during + * generation). `drain` returns + clears AND pushes a surface update (empty). + * The orchestrator owns the delivery (drain → combine → inject → carry-to-new- + * turn); this service owns only the queue state, the surface, and the + * drain-and-clear. + */ +import { defineService, type Logger, type ServiceHandle } from "@dispatch/kernel"; +import type { QueuedMessage } from "@dispatch/wire"; +import type { MessageQueueState, QueueDeps } from "./pure.js"; +import { drain as drainQueue, enqueue as enqueueMessage, getQueue as readQueue } from "./pure.js"; + +/** + * The message-queue service interface. Obtained via + * `host.getService(messageQueueHandle)`. + */ +export interface MessageQueueService { + /** Append a message; return the CURRENT queue snapshot (post-append). */ + enqueue(conversationId: string, text: string): QueuedMessage[]; + /** Current queue snapshot (empty array if none / unknown conversation). */ + getQueue(conversationId: string): QueuedMessage[]; + /** + * Drain: return all queued messages, CLEAR the queue, and push a surface + * update (empty). Returns the raw `QueuedMessage[]` (NOT a ChatMessage — + * the caller combines + builds the ChatMessage). Empty array if the queue + * was empty (and then NO surface update is pushed — no change). + */ + drain(conversationId: string): QueuedMessage[]; +} + +/** + * Typed handle anchoring the message-queue service. The single symbol the + * session-orchestrator imports to reach the queue — no string-keyed lookup. + */ +export const messageQueueHandle: ServiceHandle<MessageQueueService> = + defineService<MessageQueueService>("message-queue"); + +/** Deps for the service shell — the pure core's deps plus the surface notifier. */ +export interface MessageQueueDeps extends QueueDeps { + /** + * Notify the surface-registry that the queue changed, so the transport + * re-fetches + pushes a full new SurfaceUpdate. Called ONLY on a real + * change (enqueue → grew; drain-non-empty → emptied); a no-op drain of an + * already-empty queue does NOT notify (no change → no surface push). + */ + readonly notify: () => void; + /** Injected host logger (optional in tests). Logs counts/lengths at debug, never bodies. */ + readonly logger?: Logger; +} + +/** + * Create a MessageQueueService backed by an in-memory per-conversation Map. + * Pure decision logic (enqueue/getQueue/drain) lives in ./pure.js; this shell + * owns the state + the notify edge. State is owned (not ambient): the Map lives + * in this closure, reachable only through the returned service. + */ +export function createMessageQueueService(deps: MessageQueueDeps): MessageQueueService { + const state: MessageQueueState = new Map(); + + return { + enqueue(conversationId, text) { + const snapshot = enqueueMessage(state, conversationId, text, deps); + deps.logger?.debug("message-queue: enqueued", { + conversationId, + queueLen: snapshot.length, + textLen: text.length, + }); + deps.notify(); + return snapshot; + }, + getQueue(conversationId) { + return readQueue(state, conversationId); + }, + drain(conversationId) { + const drained = drainQueue(state, conversationId); + if (drained.length === 0) return drained; + deps.logger?.debug("message-queue: drained", { + conversationId, + count: drained.length, + }); + deps.notify(); + return drained; + }, + }; +} |
