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.test.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.test.ts')
| -rw-r--r-- | packages/message-queue/src/service.test.ts | 101 |
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"); + }); +}); |
