summaryrefslogtreecommitdiffhomepage
path: root/packages/message-queue
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-26 22:03:19 +0900
committerAdam Malczewski <[email protected]>2026-06-26 22:23:39 +0900
commit727c98c9dae516a2070eb950410314380a20c974 (patch)
tree52aa1022c54f11770be7e4e2a324f0a8b8b8deec /packages/message-queue
parente59dc11f63b1df51142259bb2c406af8c9c8c2bb (diff)
downloaddispatch-727c98c9dae516a2070eb950410314380a20c974.tar.gz
dispatch-727c98c9dae516a2070eb950410314380a20c974.zip
style: switch from tabs to 2-space indentation
Diffstat (limited to 'packages/message-queue')
-rw-r--r--packages/message-queue/package.json24
-rw-r--r--packages/message-queue/src/extension.ts106
-rw-r--r--packages/message-queue/src/index.ts26
-rw-r--r--packages/message-queue/src/pure.test.ts238
-rw-r--r--packages/message-queue/src/pure.ts74
-rw-r--r--packages/message-queue/src/service.test.ts148
-rw-r--r--packages/message-queue/src/service.ts94
-rw-r--r--packages/message-queue/tsconfig.json18
8 files changed, 364 insertions, 364 deletions
diff --git a/packages/message-queue/package.json b/packages/message-queue/package.json
index a8fcc4b..75df09e 100644
--- a/packages/message-queue/package.json
+++ b/packages/message-queue/package.json
@@ -1,14 +1,14 @@
{
- "name": "@dispatch/message-queue",
- "version": "0.0.0",
- "type": "module",
- "private": true,
- "main": "dist/index.js",
- "types": "dist/index.d.ts",
- "dependencies": {
- "@dispatch/kernel": "workspace:*",
- "@dispatch/surface-registry": "workspace:*",
- "@dispatch/ui-contract": "workspace:*",
- "@dispatch/wire": "workspace:*"
- }
+ "name": "@dispatch/message-queue",
+ "version": "0.0.0",
+ "type": "module",
+ "private": true,
+ "main": "dist/index.js",
+ "types": "dist/index.d.ts",
+ "dependencies": {
+ "@dispatch/kernel": "workspace:*",
+ "@dispatch/surface-registry": "workspace:*",
+ "@dispatch/ui-contract": "workspace:*",
+ "@dispatch/wire": "workspace:*"
+ }
}
diff --git a/packages/message-queue/src/extension.ts b/packages/message-queue/src/extension.ts
index 26d19ca..29979be 100644
--- a/packages/message-queue/src/extension.ts
+++ b/packages/message-queue/src/extension.ts
@@ -12,71 +12,71 @@ import { buildQueueSpec, MESSAGE_QUEUE_SURFACE_ID } from "./pure.js";
import { createMessageQueueService, messageQueueHandle } from "./service.js";
export const manifest: Manifest = {
- id: "message-queue",
- name: "Message Queue",
- version: "0.0.0",
- apiVersion: "^0.1.0",
- trust: "bundled",
- activation: "eager",
- dependsOn: ["surface-registry"],
- capabilities: {},
- contributes: {
- services: ["message-queue"],
- },
+ id: "message-queue",
+ name: "Message Queue",
+ version: "0.0.0",
+ apiVersion: "^0.1.0",
+ trust: "bundled",
+ activation: "eager",
+ dependsOn: ["surface-registry"],
+ capabilities: {},
+ contributes: {
+ services: ["message-queue"],
+ },
};
export function activate(host: HostAPI): void {
- const registry = host.getService(surfaceRegistryHandle);
+ const registry = host.getService(surfaceRegistryHandle);
- const subscribers = new Set<() => void>();
+ const subscribers = new Set<() => void>();
- const service = createMessageQueueService({
- id: () => crypto.randomUUID(),
- now: () => Date.now(),
- notify: () => {
- for (const sub of subscribers) {
- sub();
- }
- },
- logger: host.logger,
- });
+ const service = createMessageQueueService({
+ id: () => crypto.randomUUID(),
+ now: () => Date.now(),
+ notify: () => {
+ for (const sub of subscribers) {
+ sub();
+ }
+ },
+ logger: host.logger,
+ });
- host.provideService(messageQueueHandle, service);
+ host.provideService(messageQueueHandle, service);
- function getSpec(context?: SurfaceContext): SurfaceSpec {
- const convId = context?.conversationId;
- const messages = convId === undefined ? [] : service.getQueue(convId);
- return buildQueueSpec(messages);
- }
+ function getSpec(context?: SurfaceContext): SurfaceSpec {
+ const convId = context?.conversationId;
+ const messages = convId === undefined ? [] : service.getQueue(convId);
+ return buildQueueSpec(messages);
+ }
- function invoke(_actionId: string, _payload?: unknown, _context?: SurfaceContext): void {
- // The message-queue surface is read-only: a client renders the queue and
- // the session-orchestrator drains it. No client-facing actions.
- }
+ function invoke(_actionId: string, _payload?: unknown, _context?: SurfaceContext): void {
+ // The message-queue surface is read-only: a client renders the queue and
+ // the session-orchestrator drains it. No client-facing actions.
+ }
- const provider: SurfaceProvider = {
- catalogEntry: {
- id: MESSAGE_QUEUE_SURFACE_ID,
- region: "side",
- title: "Message Queue",
- scope: "conversation",
- },
- getSpec,
- invoke,
- subscribe(onChange) {
- subscribers.add(onChange);
- return () => {
- subscribers.delete(onChange);
- };
- },
- };
+ const provider: SurfaceProvider = {
+ catalogEntry: {
+ id: MESSAGE_QUEUE_SURFACE_ID,
+ region: "side",
+ title: "Message Queue",
+ scope: "conversation",
+ },
+ getSpec,
+ invoke,
+ subscribe(onChange) {
+ subscribers.add(onChange);
+ return () => {
+ subscribers.delete(onChange);
+ };
+ },
+ };
- registry.register(provider);
+ registry.register(provider);
- host.logger.info("message-queue: registered");
+ host.logger.info("message-queue: registered");
}
export const extension: Extension = {
- manifest,
- activate,
+ manifest,
+ activate,
};
diff --git a/packages/message-queue/src/index.ts b/packages/message-queue/src/index.ts
index 79bbb25..11467e1 100644
--- a/packages/message-queue/src/index.ts
+++ b/packages/message-queue/src/index.ts
@@ -9,19 +9,19 @@
export type { QueuedMessage, QueuePayload } from "@dispatch/wire";
export { extension, manifest } from "./extension.js";
export {
- buildQueueSpec,
- combine,
- drain,
- enqueue,
- getQueue,
- MESSAGE_QUEUE_RENDERER_ID,
- MESSAGE_QUEUE_SURFACE_ID,
- type MessageQueueState,
- type QueueDeps,
+ buildQueueSpec,
+ combine,
+ drain,
+ enqueue,
+ getQueue,
+ MESSAGE_QUEUE_RENDERER_ID,
+ MESSAGE_QUEUE_SURFACE_ID,
+ type MessageQueueState,
+ type QueueDeps,
} from "./pure.js";
export {
- createMessageQueueService,
- type MessageQueueDeps,
- type MessageQueueService,
- messageQueueHandle,
+ createMessageQueueService,
+ type MessageQueueDeps,
+ type MessageQueueService,
+ messageQueueHandle,
} from "./service.js";
diff --git a/packages/message-queue/src/pure.test.ts b/packages/message-queue/src/pure.test.ts
index ff1a08f..3fd6039 100644
--- a/packages/message-queue/src/pure.test.ts
+++ b/packages/message-queue/src/pure.test.ts
@@ -1,144 +1,144 @@
import type { QueuedMessage } from "@dispatch/wire";
import { describe, expect, it } from "vitest";
import {
- buildQueueSpec,
- combine,
- drain,
- enqueue,
- getQueue,
- MESSAGE_QUEUE_RENDERER_ID,
- MESSAGE_QUEUE_SURFACE_ID,
- type MessageQueueState,
- type QueueDeps,
+ buildQueueSpec,
+ combine,
+ drain,
+ enqueue,
+ getQueue,
+ MESSAGE_QUEUE_RENDERER_ID,
+ MESSAGE_QUEUE_SURFACE_ID,
+ type MessageQueueState,
+ type QueueDeps,
} from "./pure.js";
function makeDeps(): QueueDeps {
- let idCounter = 0;
- let now = 1_000;
- return {
- id: () => `id-${++idCounter}`,
- now: () => (now += 10),
- };
+ let idCounter = 0;
+ let now = 1_000;
+ return {
+ id: () => `id-${++idCounter}`,
+ now: () => (now += 10),
+ };
}
describe("enqueue", () => {
- it("enqueue appends + returns snapshot with the new message (id unique, queuedAt set)", () => {
- const state: MessageQueueState = new Map();
- const deps = makeDeps();
-
- const first = enqueue(state, "c1", "first", deps);
- expect(first).toHaveLength(1);
- const firstMsg = first[0];
- if (firstMsg === undefined) throw new Error("expected a message");
- expect(firstMsg.id).toBe("id-1");
- expect(firstMsg.text).toBe("first");
- expect(firstMsg.queuedAt).toBe(1_010);
-
- const second = enqueue(state, "c1", "second", deps);
- expect(second).toHaveLength(2);
- const secondMsg = second[1];
- if (secondMsg === undefined) throw new Error("expected a message");
- expect(secondMsg.id).toBe("id-2"); // unique per message
- expect(secondMsg.text).toBe("second");
- expect(secondMsg.queuedAt).toBe(1_020);
-
- // a separate conversation does not share state
- const other = enqueue(state, "c2", "other", deps);
- expect(other).toHaveLength(1);
- expect(getQueue(state, "c2")).toHaveLength(1);
- });
+ it("enqueue appends + returns snapshot with the new message (id unique, queuedAt set)", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
+
+ const first = enqueue(state, "c1", "first", deps);
+ expect(first).toHaveLength(1);
+ const firstMsg = first[0];
+ if (firstMsg === undefined) throw new Error("expected a message");
+ expect(firstMsg.id).toBe("id-1");
+ expect(firstMsg.text).toBe("first");
+ expect(firstMsg.queuedAt).toBe(1_010);
+
+ const second = enqueue(state, "c1", "second", deps);
+ expect(second).toHaveLength(2);
+ const secondMsg = second[1];
+ if (secondMsg === undefined) throw new Error("expected a message");
+ expect(secondMsg.id).toBe("id-2"); // unique per message
+ expect(secondMsg.text).toBe("second");
+ expect(secondMsg.queuedAt).toBe(1_020);
+
+ // a separate conversation does not share state
+ const other = enqueue(state, "c2", "other", deps);
+ expect(other).toHaveLength(1);
+ expect(getQueue(state, "c2")).toHaveLength(1);
+ });
});
describe("getQueue", () => {
- it("getQueue returns current snapshot; empty array when none", () => {
- const state: MessageQueueState = new Map();
- const deps = makeDeps();
+ it("getQueue returns current snapshot; empty array when none", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
- expect(getQueue(state, "unknown")).toEqual([]);
+ expect(getQueue(state, "unknown")).toEqual([]);
- enqueue(state, "c1", "x", deps);
- enqueue(state, "c1", "y", deps);
- const snap = getQueue(state, "c1");
- expect(snap.map((m) => m.text)).toEqual(["x", "y"]);
+ enqueue(state, "c1", "x", deps);
+ enqueue(state, "c1", "y", deps);
+ const snap = getQueue(state, "c1");
+ expect(snap.map((m) => m.text)).toEqual(["x", "y"]);
- // returns a COPY — mutating it does not affect live state
- snap.push({ id: "evil", text: "mutate", queuedAt: 0 });
- expect(getQueue(state, "c1")).toHaveLength(2);
- });
+ // returns a COPY — mutating it does not affect live state
+ snap.push({ id: "evil", text: "mutate", queuedAt: 0 });
+ expect(getQueue(state, "c1")).toHaveLength(2);
+ });
});
describe("drain", () => {
- it("drain returns all + clears; second drain returns []", () => {
- const state: MessageQueueState = new Map();
- const deps = makeDeps();
- enqueue(state, "c1", "a", deps);
- enqueue(state, "c1", "b", deps);
-
- const drained = drain(state, "c1");
- expect(drained.map((m) => m.text)).toEqual(["a", "b"]);
-
- // cleared
- expect(getQueue(state, "c1")).toEqual([]);
- expect(drain(state, "c1")).toEqual([]);
- });
-
- it("drain on an empty queue returns [] and is a no-op", () => {
- const state: MessageQueueState = new Map();
-
- // never had a queue
- expect(drain(state, "c1")).toEqual([]);
- expect(getQueue(state, "c1")).toEqual([]);
-
- // and after a drain that emptied it, draining again is a no-op
- const deps = makeDeps();
- enqueue(state, "c1", "once", deps);
- drain(state, "c1");
- expect(drain(state, "c1")).toEqual([]);
- expect(getQueue(state, "c1")).toEqual([]);
- });
+ it("drain returns all + clears; second drain returns []", () => {
+ const state: MessageQueueState = new Map();
+ const deps = makeDeps();
+ enqueue(state, "c1", "a", deps);
+ enqueue(state, "c1", "b", deps);
+
+ const drained = drain(state, "c1");
+ expect(drained.map((m) => m.text)).toEqual(["a", "b"]);
+
+ // cleared
+ expect(getQueue(state, "c1")).toEqual([]);
+ expect(drain(state, "c1")).toEqual([]);
+ });
+
+ it("drain on an empty queue returns [] and is a no-op", () => {
+ const state: MessageQueueState = new Map();
+
+ // never had a queue
+ expect(drain(state, "c1")).toEqual([]);
+ expect(getQueue(state, "c1")).toEqual([]);
+
+ // and after a drain that emptied it, draining again is a no-op
+ const deps = makeDeps();
+ enqueue(state, "c1", "once", deps);
+ drain(state, "c1");
+ expect(drain(state, "c1")).toEqual([]);
+ expect(getQueue(state, "c1")).toEqual([]);
+ });
});
describe("combine", () => {
- it("combine joins texts with blank-line separator", () => {
- const msgs: QueuedMessage[] = [
- { id: "1", text: "alpha", queuedAt: 1 },
- { id: "2", text: "beta", queuedAt: 2 },
- ];
- expect(combine(msgs)).toBe("alpha\n\nbeta");
- expect(combine([])).toBe("");
- expect(combine([{ id: "1", text: "solo", queuedAt: 1 }])).toBe("solo");
- });
+ it("combine joins texts with blank-line separator", () => {
+ const msgs: QueuedMessage[] = [
+ { id: "1", text: "alpha", queuedAt: 1 },
+ { id: "2", text: "beta", queuedAt: 2 },
+ ];
+ expect(combine(msgs)).toBe("alpha\n\nbeta");
+ expect(combine([])).toBe("");
+ expect(combine([{ id: "1", text: "solo", queuedAt: 1 }])).toBe("solo");
+ });
});
describe("buildQueueSpec", () => {
- it("renders a custom field with the queue snapshot", () => {
- const msgs: QueuedMessage[] = [
- { id: "1", text: "a", queuedAt: 1 },
- { id: "2", text: "b", queuedAt: 2 },
- ];
- const spec = buildQueueSpec(msgs);
- expect(spec.id).toBe(MESSAGE_QUEUE_SURFACE_ID);
- expect(spec.region).toBe("side");
- expect(spec.title).toBe("Message Queue");
- expect(spec.fields).toHaveLength(1);
-
- const field = spec.fields[0];
- if (field === undefined || field.kind !== "custom") {
- throw new Error("expected a custom field");
- }
- expect(field.rendererId).toBe(MESSAGE_QUEUE_RENDERER_ID);
- const payload = field.payload as { messages: readonly QueuedMessage[] };
- expect(payload.messages).toHaveLength(2);
- expect(payload.messages.map((m) => m.text)).toEqual(["a", "b"]);
- });
-
- it("renders an empty list for an empty queue", () => {
- const spec = buildQueueSpec([]);
- const field = spec.fields[0];
- if (field === undefined || field.kind !== "custom") {
- throw new Error("expected a custom field");
- }
- const payload = field.payload as { messages: readonly QueuedMessage[] };
- expect(payload.messages).toEqual([]);
- });
+ it("renders a custom field with the queue snapshot", () => {
+ const msgs: QueuedMessage[] = [
+ { id: "1", text: "a", queuedAt: 1 },
+ { id: "2", text: "b", queuedAt: 2 },
+ ];
+ const spec = buildQueueSpec(msgs);
+ expect(spec.id).toBe(MESSAGE_QUEUE_SURFACE_ID);
+ expect(spec.region).toBe("side");
+ expect(spec.title).toBe("Message Queue");
+ expect(spec.fields).toHaveLength(1);
+
+ const field = spec.fields[0];
+ if (field === undefined || field.kind !== "custom") {
+ throw new Error("expected a custom field");
+ }
+ expect(field.rendererId).toBe(MESSAGE_QUEUE_RENDERER_ID);
+ const payload = field.payload as { messages: readonly QueuedMessage[] };
+ expect(payload.messages).toHaveLength(2);
+ expect(payload.messages.map((m) => m.text)).toEqual(["a", "b"]);
+ });
+
+ it("renders an empty list for an empty queue", () => {
+ const spec = buildQueueSpec([]);
+ const field = spec.fields[0];
+ if (field === undefined || field.kind !== "custom") {
+ throw new Error("expected a custom field");
+ }
+ const payload = field.payload as { messages: readonly QueuedMessage[] };
+ expect(payload.messages).toEqual([]);
+ });
});
diff --git a/packages/message-queue/src/pure.ts b/packages/message-queue/src/pure.ts
index 834018b..981e005 100644
--- a/packages/message-queue/src/pure.ts
+++ b/packages/message-queue/src/pure.ts
@@ -17,10 +17,10 @@ export type MessageQueueState = Map<string, QueuedMessage[]>;
/** Injected effectful factories kept out of the pure core. */
export interface QueueDeps {
- /** Stable (client-visible) id factory for UI keying + dedup. */
- readonly id: () => string;
- /** Clock returning epoch-ms for `queuedAt`. */
- readonly now: () => number;
+ /** Stable (client-visible) id factory for UI keying + dedup. */
+ readonly id: () => string;
+ /** Clock returning epoch-ms for `queuedAt`. */
+ readonly now: () => number;
}
/** Surface id this extension contributes (also the manifest + catalog id). */
@@ -34,19 +34,19 @@ export const MESSAGE_QUEUE_RENDERER_ID = "message-queue";
* live state through the returned value.
*/
export function enqueue(
- state: MessageQueueState,
- conversationId: string,
- text: string,
- deps: QueueDeps,
+ state: MessageQueueState,
+ conversationId: string,
+ text: string,
+ deps: QueueDeps,
): QueuedMessage[] {
- const message: QueuedMessage = { id: deps.id(), text, queuedAt: deps.now() };
- const existing = state.get(conversationId);
- if (existing === undefined) {
- state.set(conversationId, [message]);
- } else {
- existing.push(message);
- }
- return getQueue(state, conversationId);
+ const message: QueuedMessage = { id: deps.id(), text, queuedAt: deps.now() };
+ const existing = state.get(conversationId);
+ if (existing === undefined) {
+ state.set(conversationId, [message]);
+ } else {
+ existing.push(message);
+ }
+ return getQueue(state, conversationId);
}
/**
@@ -54,9 +54,9 @@ export function enqueue(
* if the conversation has no queue / is unknown.
*/
export function getQueue(state: MessageQueueState, conversationId: string): QueuedMessage[] {
- const existing = state.get(conversationId);
- if (existing === undefined) return [];
- return [...existing];
+ const existing = state.get(conversationId);
+ if (existing === undefined) return [];
+ return [...existing];
}
/**
@@ -67,11 +67,11 @@ export function getQueue(state: MessageQueueState, conversationId: string): Queu
* ChatMessage.
*/
export function drain(state: MessageQueueState, conversationId: string): QueuedMessage[] {
- const existing = state.get(conversationId);
- if (existing === undefined || existing.length === 0) return [];
- const drained = [...existing];
- state.delete(conversationId);
- return drained;
+ const existing = state.get(conversationId);
+ if (existing === undefined || existing.length === 0) return [];
+ const drained = [...existing];
+ state.delete(conversationId);
+ return drained;
}
/**
@@ -80,7 +80,7 @@ export function drain(state: MessageQueueState, conversationId: string): QueuedM
* ChatMessage from this.
*/
export function combine(messages: readonly QueuedMessage[]): string {
- return messages.map((m) => m.text).join("\n\n");
+ return messages.map((m) => m.text).join("\n\n");
}
/**
@@ -90,16 +90,16 @@ export function combine(messages: readonly QueuedMessage[]): string {
* I/O; the surface-registry re-fetches this on every notify.
*/
export function buildQueueSpec(messages: readonly QueuedMessage[]): SurfaceSpec {
- const payload: QueuePayload = { messages };
- const field: CustomField = {
- kind: "custom",
- rendererId: MESSAGE_QUEUE_RENDERER_ID,
- payload,
- };
- return {
- id: MESSAGE_QUEUE_SURFACE_ID,
- region: "side",
- title: "Message Queue",
- fields: [field],
- };
+ const payload: QueuePayload = { messages };
+ const field: CustomField = {
+ kind: "custom",
+ rendererId: MESSAGE_QUEUE_RENDERER_ID,
+ payload,
+ };
+ return {
+ id: MESSAGE_QUEUE_SURFACE_ID,
+ region: "side",
+ title: "Message Queue",
+ fields: [field],
+ };
}
diff --git a/packages/message-queue/src/service.test.ts b/packages/message-queue/src/service.test.ts
index 04cde91..aa59dd3 100644
--- a/packages/message-queue/src/service.test.ts
+++ b/packages/message-queue/src/service.test.ts
@@ -5,97 +5,97 @@ import { buildQueueSpec, combine } from "./pure.js";
import { createMessageQueueService, type MessageQueueDeps } from "./service.js";
interface Recorder extends MessageQueueDeps {
- readonly calls: { value: number };
+ readonly calls: { value: number };
}
function makeDeps(): Recorder {
- let idCounter = 0;
- let now = 1_000;
- const calls = { value: 0 };
- return {
- id: () => `q-${++idCounter}`,
- now: () => (now += 10),
- notify: () => {
- calls.value += 1;
- },
- calls,
- };
+ let idCounter = 0;
+ let now = 1_000;
+ const calls = { value: 0 };
+ return {
+ id: () => `q-${++idCounter}`,
+ now: () => (now += 10),
+ notify: () => {
+ calls.value += 1;
+ },
+ calls,
+ };
}
function queueMessages(spec: SurfaceSpec): readonly QueuedMessage[] {
- const field = spec.fields[0];
- if (field === undefined || field.kind !== "custom") {
- throw new Error("expected a custom field");
- }
- return (field.payload as QueuePayload).messages;
+ const field = spec.fields[0];
+ if (field === undefined || field.kind !== "custom") {
+ throw new Error("expected a custom field");
+ }
+ return (field.payload as QueuePayload).messages;
}
describe("message-queue service", () => {
- it("enqueue pushes a surface update (snapshot grew)", () => {
- const deps = makeDeps();
- const svc = createMessageQueueService(deps);
- expect(deps.calls.value).toBe(0);
+ it("enqueue pushes a surface update (snapshot grew)", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ expect(deps.calls.value).toBe(0);
- const snapshot = svc.enqueue("c1", "hello");
- expect(deps.calls.value).toBe(1); // surface update pushed
- expect(snapshot).toHaveLength(1);
- const first = snapshot[0];
- if (first === undefined) throw new Error("expected a message");
- expect(first.text).toBe("hello");
- expect(typeof first.id).toBe("string");
- expect(first.queuedAt).toBe(1_010);
+ const snapshot = svc.enqueue("c1", "hello");
+ expect(deps.calls.value).toBe(1); // surface update pushed
+ expect(snapshot).toHaveLength(1);
+ const first = snapshot[0];
+ if (first === undefined) throw new Error("expected a message");
+ expect(first.text).toBe("hello");
+ expect(typeof first.id).toBe("string");
+ expect(first.queuedAt).toBe(1_010);
- // the surface spec reflects the grown snapshot
- expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toHaveLength(1);
- });
+ // the surface spec reflects the grown snapshot
+ expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toHaveLength(1);
+ });
- it("drain pushes a surface update (empty)", () => {
- const deps = makeDeps();
- const svc = createMessageQueueService(deps);
- svc.enqueue("c1", "a");
- svc.enqueue("c1", "b");
- expect(deps.calls.value).toBe(2); // two enqueues notified
+ it("drain pushes a surface update (empty)", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ svc.enqueue("c1", "a");
+ svc.enqueue("c1", "b");
+ expect(deps.calls.value).toBe(2); // two enqueues notified
- const drained = svc.drain("c1");
- expect(deps.calls.value).toBe(3); // drain pushed a surface update
- expect(drained).toHaveLength(2);
- expect(drained.map((m) => m.text)).toEqual(["a", "b"]);
+ const drained = svc.drain("c1");
+ expect(deps.calls.value).toBe(3); // drain pushed a surface update
+ expect(drained).toHaveLength(2);
+ expect(drained.map((m) => m.text)).toEqual(["a", "b"]);
- // cleared + surface now shows empty
- expect(svc.getQueue("c1")).toEqual([]);
- expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toEqual([]);
- });
+ // cleared + surface now shows empty
+ expect(svc.getQueue("c1")).toEqual([]);
+ expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toEqual([]);
+ });
- it("drain on empty does NOT push a surface update (already empty)", () => {
- const deps = makeDeps();
- const svc = createMessageQueueService(deps);
+ it("drain on empty does NOT push a surface update (already empty)", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
- expect(svc.drain("c1")).toEqual([]);
- expect(deps.calls.value).toBe(0); // no notify — no change
+ expect(svc.drain("c1")).toEqual([]);
+ expect(deps.calls.value).toBe(0); // no notify — no change
- // after a real drain empties it, a further drain is a no-op (no notify)
- svc.enqueue("c1", "x");
- expect(deps.calls.value).toBe(1);
- svc.drain("c1");
- expect(deps.calls.value).toBe(2);
- svc.drain("c1");
- expect(deps.calls.value).toBe(2); // unchanged — queue was already empty
- });
+ // after a real drain empties it, a further drain is a no-op (no notify)
+ svc.enqueue("c1", "x");
+ expect(deps.calls.value).toBe(1);
+ svc.drain("c1");
+ expect(deps.calls.value).toBe(2);
+ svc.drain("c1");
+ expect(deps.calls.value).toBe(2); // unchanged — queue was already empty
+ });
- it("getQueue returns current snapshot; empty for unknown conversation", () => {
- const deps = makeDeps();
- const svc = createMessageQueueService(deps);
- expect(svc.getQueue("nope")).toEqual([]);
- svc.enqueue("c1", "hi");
- expect(svc.getQueue("c1")).toHaveLength(1);
- });
+ it("getQueue returns current snapshot; empty for unknown conversation", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ expect(svc.getQueue("nope")).toEqual([]);
+ svc.enqueue("c1", "hi");
+ expect(svc.getQueue("c1")).toHaveLength(1);
+ });
- it("drain returns the raw QueuedMessage[] the orchestrator combines", () => {
- const deps = makeDeps();
- const svc = createMessageQueueService(deps);
- svc.enqueue("c1", "alpha");
- svc.enqueue("c1", "beta");
- const drained = svc.drain("c1");
- expect(combine(drained)).toBe("alpha\n\nbeta");
- });
+ it("drain returns the raw QueuedMessage[] the orchestrator combines", () => {
+ const deps = makeDeps();
+ const svc = createMessageQueueService(deps);
+ svc.enqueue("c1", "alpha");
+ svc.enqueue("c1", "beta");
+ const drained = svc.drain("c1");
+ expect(combine(drained)).toBe("alpha\n\nbeta");
+ });
});
diff --git a/packages/message-queue/src/service.ts b/packages/message-queue/src/service.ts
index 91d0fc5..97e270d 100644
--- a/packages/message-queue/src/service.ts
+++ b/packages/message-queue/src/service.ts
@@ -18,17 +18,17 @@ import { drain as drainQueue, enqueue as enqueueMessage, getQueue as readQueue }
* `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[];
+ /** 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[];
}
/**
@@ -36,19 +36,19 @@ export interface MessageQueueService {
* session-orchestrator imports to reach the queue — no string-keyed lookup.
*/
export const messageQueueHandle: ServiceHandle<MessageQueueService> =
- defineService<MessageQueueService>("message-queue");
+ 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;
+ /**
+ * 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;
}
/**
@@ -58,31 +58,31 @@ export interface MessageQueueDeps extends QueueDeps {
* in this closure, reachable only through the returned service.
*/
export function createMessageQueueService(deps: MessageQueueDeps): MessageQueueService {
- const state: MessageQueueState = new Map();
+ 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;
- },
- };
+ 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;
+ },
+ };
}
diff --git a/packages/message-queue/tsconfig.json b/packages/message-queue/tsconfig.json
index da5dfb7..1c0d44b 100644
--- a/packages/message-queue/tsconfig.json
+++ b/packages/message-queue/tsconfig.json
@@ -1,11 +1,11 @@
{
- "extends": "../../tsconfig.base.json",
- "compilerOptions": { "rootDir": "src", "outDir": "dist", "composite": true },
- "include": ["src/**/*.ts"],
- "references": [
- { "path": "../kernel" },
- { "path": "../surface-registry" },
- { "path": "../ui-contract" },
- { "path": "../wire" }
- ]
+ "extends": "../../tsconfig.base.json",
+ "compilerOptions": { "rootDir": "src", "outDir": "dist", "composite": true },
+ "include": ["src/**/*.ts"],
+ "references": [
+ { "path": "../kernel" },
+ { "path": "../surface-registry" },
+ { "path": "../ui-contract" },
+ { "path": "../wire" }
+ ]
}