summaryrefslogtreecommitdiffhomepage
path: root/packages/message-queue/src/service.test.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.test.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.test.ts')
-rw-r--r--packages/message-queue/src/service.test.ts101
1 files changed, 101 insertions, 0 deletions
diff --git a/packages/message-queue/src/service.test.ts b/packages/message-queue/src/service.test.ts
new file mode 100644
index 0000000..04cde91
--- /dev/null
+++ b/packages/message-queue/src/service.test.ts
@@ -0,0 +1,101 @@
+import type { SurfaceSpec } from "@dispatch/ui-contract";
+import type { QueuedMessage, QueuePayload } from "@dispatch/wire";
+import { describe, expect, it } from "vitest";
+import { buildQueueSpec, combine } from "./pure.js";
+import { createMessageQueueService, type MessageQueueDeps } from "./service.js";
+
+interface Recorder extends MessageQueueDeps {
+ 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,
+ };
+}
+
+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;
+}
+
+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);
+
+ 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);
+ });
+
+ 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"]);
+
+ // 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);
+
+ 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
+ });
+
+ 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");
+ });
+});