summaryrefslogtreecommitdiffhomepage
path: root/packages/message-queue/src/service.ts
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-21 02:08:44 +0900
committerAdam Malczewski <[email protected]>2026-06-21 02:08:44 +0900
commitba47df37f0c89bff4f0c3dd7d0bc2ef6c8062b92 (patch)
tree21d87eb847cd526a506cf274467fd1359f349705 /packages/message-queue/src/service.ts
parent75032313a96856a932c109efbbe6b6a7eb782222 (diff)
downloaddispatch-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.ts88
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;
+ },
+ };
+}