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.

4 files changed, 618 insertions, 1 deletions

diff --git a/packages/session-orchestrator/src/extension.ts b/packages/session-orchestrator/src/extension.ts
index 781164a..6e56c2b 100644
--- a/packages/session-orchestrator/src/extension.ts
+++ b/packages/session-orchestrator/src/extension.ts
@@ -2,6 +2,7 @@ import { conversationStoreHandle } from "@dispatch/conversation-store";
 import { credentialStoreHandle } from "@dispatch/credential-store";
 import type { Extension, HostAPI, Manifest } from "@dispatch/kernel";
 import { runTurn } from "@dispatch/kernel";
+import { messageQueueHandle } from "@dispatch/message-queue";
 import {
 	cacheWarmHandle,
 	createSessionOrchestrator,
@@ -49,6 +50,15 @@ export function activate(host: HostAPI): void {
 		logger: host.logger,
 		now: () => Date.now(),
 		emit: (hook, payload) => host.emit(hook, payload),
+		resolveQueue: () => {
+			// Lazily resolve the message-queue service. Returns undefined when the
+			// extension isn't loaded (feature degrades off) — checked via the
+			// activated-manifests list so `host.getService` is only called when the
+			// service is registered. Lazy so activation order with message-queue
+			// doesn't matter; called per-turn / per-enqueue, not at activate time.
+			const loaded = host.getExtensions().some((m) => m.id === "message-queue");
+			return loaded ? host.getService(messageQueueHandle) : undefined;
+		},
 	});
 
 	host.provideService(sessionOrchestratorHandle, orchestrator);
diff --git a/packages/session-orchestrator/src/index.ts b/packages/session-orchestrator/src/index.ts
index 711fa5a..afec2b4 100644
--- a/packages/session-orchestrator/src/index.ts
+++ b/packages/session-orchestrator/src/index.ts
@@ -5,6 +5,8 @@ export {
 	conversationClosed,
 	createSessionOrchestrator,
 	createWarmService,
+	type EnqueueInput,
+	type EnqueueResult,
 	type SessionOrchestrator,
 	type SessionOrchestratorBundle,
 	type SessionOrchestratorDeps,
diff --git a/packages/session-orchestrator/src/orchestrator.ts b/packages/session-orchestrator/src/orchestrator.ts
index 5b2f264..3a74c2d 100644
--- a/packages/session-orchestrator/src/orchestrator.ts
+++ b/packages/session-orchestrator/src/orchestrator.ts
@@ -15,6 +15,7 @@ import type {
 	UsageEvent,
 } from "@dispatch/kernel";
 import { defineEventHook, defineService, type ServiceHandle } from "@dispatch/kernel";
+import type { MessageQueueService, QueuedMessage } from "@dispatch/message-queue";
 import { createMetricsAccumulator } from "./metrics.js";
 import {
 	buildUserMessage,
@@ -38,6 +39,25 @@ export type StartTurnResult =
 	| { readonly started: true; readonly turnId: string }
 	| { readonly started: false; readonly reason: "already-active" };
 
+/** Input to `SessionOrchestrator.enqueue` — the single entry transports call. */
+export interface EnqueueInput {
+	readonly conversationId: string;
+	readonly text: string;
+}
+
+/**
+ * Result of `SessionOrchestrator.enqueue`. When `startedTurn` is true the
+ * conversation was idle and a turn was started (the message is the opening
+ * prompt — nothing queued). When false the conversation was active: the message
+ * was enqueued onto the steering queue and `queue` is the post-enqueue snapshot
+ * (empty when the message-queue extension isn't loaded — degraded: the message
+ * is dropped, see `enqueue` docs).
+ */
+export interface EnqueueResult {
+	readonly startedTurn: boolean;
+	readonly queue: readonly QueuedMessage[];
+}
+
 export type TurnEventListener = (event: AgentEvent) => void;
 
 interface ActiveTurn {
@@ -109,6 +129,16 @@ export const cacheWarmHandle: ServiceHandle<WarmService> = defineService<WarmSer
 
 export interface SessionOrchestrator {
 	startTurn(input: StartTurnInput): StartTurnResult;
+	/**
+	 * The single entry transports call to deliver a user message. Owns the
+	 * idle→startTurn vs active→queue decision (no separate `isActive` race —
+	 * `startTurn`'s single-flight guard is authoritative). When the conversation
+	 * is idle, starts a turn (the message is the opening prompt). When active,
+	 * enqueues onto the steering queue (if the message-queue extension is
+	 * loaded); with no queue extension loaded the message is dropped and the
+	 * returned snapshot is empty (degraded — feature off).
+	 */
+	enqueue(input: EnqueueInput): EnqueueResult;
 	subscribe(conversationId: string, listener: TurnEventListener): () => void;
 	isActive(conversationId: string): boolean;
 	/**
@@ -143,6 +173,16 @@ export interface SessionOrchestratorDeps {
 		modelName: string,
 	) => { provider: ProviderContract; model: string } | undefined;
 	readonly runTurn: (input: RunTurnInput) => Promise<RunTurnResult>;
+	/**
+	 * Lazily resolves the message-queue service (the steering queue), or
+	 * `undefined` when the message-queue extension isn't loaded (the feature
+	 * degrades off: no `drainSteering`, no post-seal carry, `enqueue` drops
+	 * messages when active). host-bin wires this via `host.getService`; the
+	 * orchestrator calls it per-turn / per-enqueue so activation order with the
+	 * message-queue extension doesn't matter. Injected (not ambient) so a turn
+	 * stays reproducible from its inputs and tests use a fake queue.
+	 */
+	readonly resolveQueue?: () => MessageQueueService | undefined;
 	/** Apply the per-turn tools filter chain. Injected for testability. */
 	readonly applyToolsFilter: (assembly: ToolAssembly) => Promise<ToolAssembly>;
 	/** Base logger (auto-scoped to this extension); childed per turn for span capture. */
@@ -184,6 +224,25 @@ export function createSessionOrchestrator(
 		}
 	}
 
+	/**
+	 * Post-seal carry: if a steering queue is available and non-empty, drain it,
+	 * combine, and start a NEW detached turn whose opening `user-message` carries
+	 * the combined text (no `steering` event — that's only for mid-turn drain).
+	 * Returns true iff a new turn was started. Called from `runTurnDetached`'s
+	 * finally AFTER `activeTurns.delete` (so the new turn's single-flight guard
+	 * passes) and BEFORE `activeConversations.delete` (skipped when carried, since
+	 * the new turn re-adds it). May chain — the new turn's own finally re-checks.
+	 */
+	function tryCarryQueue(conversationId: string): boolean {
+		const queue = deps.resolveQueue?.();
+		if (queue === undefined) return false;
+		if (queue.getQueue(conversationId).length === 0) return false;
+		const drained = queue.drain(conversationId);
+		const combined = drained.map((q) => q.text).join("\n\n");
+		const result = orchestrator.startTurn({ conversationId, text: combined });
+		return result.started;
+	}
+
 	function runTurnDetached(
 		conversationId: string,
 		text: string,
@@ -218,6 +277,7 @@ export function createSessionOrchestrator(
 		});
 
 		void (async () => {
+			let sealed = false;
 			try {
 				const [effectiveCwd, storedEffort] = await Promise.all([
 					effectiveCwdPromise,
@@ -273,6 +333,30 @@ export function createSessionOrchestrator(
 					...(modelOverride !== undefined ? { model: modelOverride } : {}),
 				};
 
+				// Resolve the steering queue once for this turn. When present, wire
+				// `drainSteering`: the kernel calls it at the tool-result boundary and
+				// appends whatever it returns as user-role messages alongside the tool
+				// results (mid-turn steering). The wrapper emits a `steering` AgentEvent
+				// into the hub (buffered for late-join like `user-message`) so a
+				// frontend can place a user bubble in the transcript live; the kernel
+				// only appends the returned messages — it does NOT emit the event.
+				const queue = deps.resolveQueue?.();
+				const drainSteering =
+					queue === undefined
+						? undefined
+						: (): readonly ChatMessage[] => {
+								const queued = queue.drain(conversationId);
+								if (queued.length === 0) return [];
+								const steerText = queued.map((q) => q.text).join("\n\n");
+								emitToHub(conversationId, {
+									type: "steering",
+									conversationId,
+									turnId,
+									text: steerText,
+								});
+								return [{ role: "user", chunks: [{ type: "text", text: steerText }] }];
+							};
+
 				const opts: RunTurnInput = {
 					provider,
 					messages: [...history, userMsg],
@@ -286,6 +370,7 @@ export function createSessionOrchestrator(
 					...(turnLogger !== undefined ? { logger: turnLogger } : {}),
 					...(effectiveCwd !== undefined ? { cwd: effectiveCwd } : {}),
 					...(deps.now !== undefined ? { now: deps.now } : {}),
+					...(drainSteering !== undefined ? { drainSteering } : {}),
 				};
 
 				const result = await deps.runTurn(opts);
@@ -297,6 +382,7 @@ export function createSessionOrchestrator(
 				await deps.conversationStore.appendMetrics(conversationId, turnMetrics);
 
 				emitToHub(conversationId, { type: "turn-sealed", conversationId, turnId });
+				sealed = true;
 			} catch (err) {
 				const message = err instanceof Error ? err.message : String(err);
 				emitToHub(conversationId, {
@@ -307,7 +393,16 @@ export function createSessionOrchestrator(
 				});
 			} finally {
 				activeTurns.delete(conversationId);
-				activeConversations.delete(conversationId);
+				// Post-seal carry: if the turn sealed with a non-empty steering queue
+				// (no tool call fired → drainSteering never drained it), start a NEW
+				// detached turn whose opening user-message carries the combined text.
+				// The new turn re-adds to activeTurns + activeConversations, so skip
+				// the activeConversations.delete when carried. May chain (user keeps
+				// steering) — each carried turn's own finally re-checks the queue.
+				const carried = sealed && tryCarryQueue(conversationId);
+				if (!carried) {
+					activeConversations.delete(conversationId);
+				}
 				void payloadPromise.then((payload) => {
 					deps.emit?.(turnSettled, payload);
 				});
@@ -326,6 +421,19 @@ export function createSessionOrchestrator(
 			return { started: true, turnId };
 		},
 
+		enqueue({ conversationId, text }) {
+			const result = orchestrator.startTurn({ conversationId, text });
+			if (result.started) {
+				return { startedTurn: true, queue: [] };
+			}
+			// Already active → enqueue onto the steering queue. When the
+			// message-queue extension isn't loaded this degrades: the message is
+			// dropped and the snapshot is empty (feature off).
+			const queue = deps.resolveQueue?.();
+			const snapshot = queue !== undefined ? queue.enqueue(conversationId, text) : [];
+			return { startedTurn: false, queue: snapshot };
+		},
+
 		subscribe(conversationId, listener) {
 			let listeners = subscribers.get(conversationId);
 			if (listeners === undefined) {
diff --git a/packages/session-orchestrator/src/queue.test.ts b/packages/session-orchestrator/src/queue.test.ts
new file mode 100644
index 0000000..c1f12da
--- /dev/null
+++ b/packages/session-orchestrator/src/queue.test.ts
@@ -0,0 +1,497 @@
+import type { ConversationStore } from "@dispatch/conversation-store";
+import type {
+	AgentEvent,
+	ChatMessage,
+	ProviderContract,
+	ProviderEvent,
+	ReasoningEffort,
+	RunTurnInput,
+	RunTurnResult,
+	StoredChunk,
+	ToolContract,
+	TurnMetrics,
+} from "@dispatch/kernel";
+import { runTurn } from "@dispatch/kernel";
+import { createMessageQueueService } from "@dispatch/message-queue";
+import { describe, expect, it } from "vitest";
+import { createSessionOrchestrator } from "./orchestrator.js";
+import type { ToolAssembly } from "./tools-filter.js";
+
+// --- Shared test helpers (duplicated from orchestrator.test.ts per isolation-over-dRY;
+//     a shared test-helper module wired between test files is a coupling smell) ---
+
+function createInMemoryStore(): ConversationStore & {
+	readonly data: Map<string, ChatMessage[]>;
+	readonly metricsData: Map<string, TurnMetrics[]>;
+	readonly cwdData: Map<string, string>;
+	readonly effortData: Map<string, ReasoningEffort>;
+} {
+	const data = new Map<string, ChatMessage[]>();
+	const metricsData = new Map<string, TurnMetrics[]>();
+	const cwdData = new Map<string, string>();
+	const effortData = new Map<string, ReasoningEffort>();
+	return {
+		data,
+		metricsData,
+		cwdData,
+		effortData,
+		async append(conversationId, messages) {
+			const existing = data.get(conversationId) ?? [];
+			data.set(conversationId, [...existing, ...messages]);
+		},
+		async load(conversationId) {
+			return [...(data.get(conversationId) ?? [])];
+		},
+		async loadSince(conversationId, sinceSeq) {
+			const messages = data.get(conversationId) ?? [];
+			const result: StoredChunk[] = [];
+			let seq = 1;
+			for (const msg of messages) {
+				for (const chunk of msg.chunks) {
+					if (sinceSeq === undefined || seq > sinceSeq) {
+						result.push({ seq, role: msg.role, chunk });
+					}
+					seq++;
+				}
+			}
+			return result;
+		},
+		async appendMetrics(conversationId, metrics) {
+			const existing = metricsData.get(conversationId) ?? [];
+			metricsData.set(conversationId, [...existing, metrics]);
+		},
+		async loadMetrics(conversationId) {
+			return [...(metricsData.get(conversationId) ?? [])];
+		},
+		async getCwd(conversationId) {
+			return cwdData.get(conversationId) ?? null;
+		},
+		async setCwd(conversationId, cwd) {
+			cwdData.set(conversationId, cwd);
+		},
+		async getReasoningEffort(conversationId) {
+			return effortData.get(conversationId) ?? null;
+		},
+		async setReasoningEffort(conversationId, effort) {
+			effortData.set(conversationId, effort);
+		},
+	};
+}
+
+function identityApplyToolsFilter(assembly: ToolAssembly): Promise<ToolAssembly> {
+	return Promise.resolve(assembly);
+}
+
+function noTools(): readonly ToolContract[] {
+	return [];
+}
+
+function simpleProvider(): ProviderContract {
+	return {
+		id: "fake",
+		stream: async function* () {
+			yield { type: "text-delta", delta: "ok" } as ProviderEvent;
+			yield { type: "finish", reason: "stop" } as ProviderEvent;
+		},
+	};
+}
+
+/**
+ * A capturing runTurn that simulates the kernel calling `drainSteering` at the
+ * tool-result boundary. It records the RunTurnInput (so the test can assert
+ * drainSteering was wired) and collects what drainSteering returned. NOT a mock
+ * of @dispatch/* — it's a plain fake of the outermost runTurn edge.
+ */
+function createDrainingCaptureRunTurn(): {
+	captured: RunTurnInput[];
+	drainedMessages: ChatMessage[];
+	wasDrainCalled: () => boolean;
+	runTurn: (input: RunTurnInput) => Promise<RunTurnResult>;
+} {
+	const captured: RunTurnInput[] = [];
+	const drainedMessages: ChatMessage[] = [];
+	let drainCalled = false;
+	return {
+		captured,
+		drainedMessages,
+		wasDrainCalled: () => drainCalled,
+		runTurn: async (input) => {
+			captured.push(input);
+			if (input.drainSteering !== undefined) {
+				drainCalled = true;
+				const drained = input.drainSteering();
+				drainedMessages.push(...drained);
+			}
+			return {
+				messages: [{ role: "assistant", chunks: [{ type: "text", text: "ok" }] }],
+				usage: { inputTokens: 1, outputTokens: 1 },
+				finishReason: "stop",
+			};
+		},
+	};
+}
+
+function waitForSealed(
+	orchestrator: ReturnType<typeof createSessionOrchestrator>["orchestrator"],
+	conversationId: string,
+): Promise<void> {
+	return new Promise((resolve) => {
+		const unsub = orchestrator.subscribe(conversationId, (e) => {
+			if (e.type === "turn-sealed") {
+				unsub();
+				resolve();
+			}
+		});
+	});
+}
+
+function waitForSealedCount(
+	orchestrator: ReturnType<typeof createSessionOrchestrator>["orchestrator"],
+	conversationId: string,
+	count: number,
+): Promise<void> {
+	return new Promise((resolve) => {
+		let seen = 0;
+		const unsub = orchestrator.subscribe(conversationId, (e) => {
+			if (e.type === "turn-sealed") {
+				seen++;
+				if (seen >= count) {
+					unsub();
+					resolve();
+				}
+			}
+		});
+	});
+}
+
+function isSteering(e: AgentEvent): e is Extract<AgentEvent, { type: "steering" }> {
+	return e.type === "steering";
+}
+
+function isUserMessage(e: AgentEvent): e is Extract<AgentEvent, { type: "user-message" }> {
+	return e.type === "user-message";
+}
+
+function createTestQueue() {
+	return createMessageQueueService({
+		id: () => `q-${Math.random().toString(36).slice(2, 8)}`,
+		now: () => 1000,
+		notify: () => {},
+	});
+}
+
+// --- drainSteering (mid-turn, at the tool-result boundary) ---
+
+describe("drainSteering", () => {
+	it("drainSteering drains the queue + emits a steering event + returns one combined user message", async () => {
+		const store = createInMemoryStore();
+		const queue = createTestQueue();
+		queue.enqueue("conv-drain", "first");
+		queue.enqueue("conv-drain", "second");
+
+		const { captured, drainedMessages, runTurn: captureRunTurn } = createDrainingCaptureRunTurn();
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => ({ id: "p", stream: async function* () {} }),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn: captureRunTurn,
+			resolveQueue: () => queue,
+		});
+
+		const events: AgentEvent[] = [];
+		const unsub = orchestrator.subscribe("conv-drain", (e) => events.push(e));
+
+		orchestrator.startTurn({ conversationId: "conv-drain", text: "go" });
+		await waitForSealed(orchestrator, "conv-drain");
+		unsub();
+
+		// drainSteering was wired on the RunTurnInput
+		expect(captured).toHaveLength(1);
+		expect(captured[0]?.drainSteering).toBeDefined();
+		expect(typeof captured[0]?.drainSteering).toBe("function");
+
+		// The fake runTurn called drainSteering → returned one combined user message
+		expect(drainedMessages).toHaveLength(1);
+		const steerMsg = drainedMessages[0];
+		if (steerMsg === undefined) throw new Error("expected drained message");
+		expect(steerMsg.role).toBe("user");
+		expect(steerMsg.chunks).toHaveLength(1);
+		const chunk = steerMsg.chunks[0];
+		if (chunk === undefined) throw new Error("expected chunk");
+		expect(chunk.type).toBe("text");
+		if (chunk.type === "text") {
+			expect(chunk.text).toBe("first\n\nsecond");
+		}
+
+		// The queue was drained (cleared)
+		expect(queue.getQueue("conv-drain")).toHaveLength(0);
+
+		// A steering event was emitted into the hub with the combined text
+		const steering = events.find(isSteering);
+		expect(steering).toBeDefined();
+		expect(steering?.conversationId).toBe("conv-drain");
+		expect(steering?.text).toBe("first\n\nsecond");
+		expect(steering?.turnId).toMatch(/^turn-/);
+	});
+
+	it("drainSteering on an empty queue returns [] and emits nothing", async () => {
+		const store = createInMemoryStore();
+		const queue = createTestQueue();
+
+		const {
+			drainedMessages,
+			wasDrainCalled,
+			runTurn: captureRunTurn,
+		} = createDrainingCaptureRunTurn();
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => ({ id: "p", stream: async function* () {} }),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn: captureRunTurn,
+			resolveQueue: () => queue,
+		});
+
+		const events: AgentEvent[] = [];
+		const unsub = orchestrator.subscribe("conv-empty", (e) => events.push(e));
+
+		orchestrator.startTurn({ conversationId: "conv-empty", text: "go" });
+		await waitForSealed(orchestrator, "conv-empty");
+		unsub();
+
+		// drainSteering was wired and called, but returned []
+		expect(wasDrainCalled()).toBe(true);
+		expect(drainedMessages).toHaveLength(0);
+
+		// No steering event was emitted
+		expect(events.filter(isSteering)).toHaveLength(0);
+	});
+
+	it("no queue ext (resolveQueue undefined) → drainSteering omitted; turn unchanged", async () => {
+		const store = createInMemoryStore();
+
+		const { captured, wasDrainCalled, runTurn: captureRunTurn } = createDrainingCaptureRunTurn();
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => ({ id: "p", stream: async function* () {} }),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn: captureRunTurn,
+			// resolveQueue intentionally omitted — feature degrades off
+		});
+
+		const events: AgentEvent[] = [];
+		const unsub = orchestrator.subscribe("conv-noqueue", (e) => events.push(e));
+
+		orchestrator.startTurn({ conversationId: "conv-noqueue", text: "go" });
+		await waitForSealed(orchestrator, "conv-noqueue");
+		unsub();
+
+		// drainSteering is absent from the RunTurnInput (not undefined — omitted)
+		expect(captured).toHaveLength(1);
+		expect(captured[0]?.drainSteering).toBeUndefined();
+		expect(wasDrainCalled()).toBe(false);
+
+		// No steering event; turn sealed normally
+		expect(events.filter(isSteering)).toHaveLength(0);
+		expect(events.filter((e) => e.type === "turn-sealed")).toHaveLength(1);
+	});
+});
+
+// --- Post-seal carry (turn ended with a non-empty queue → new turn) ---
+
+describe("post-seal carry", () => {
+	it("post-seal: non-empty queue → a new turn starts with the combined message", async () => {
+		const store = createInMemoryStore();
+		const queue = createTestQueue();
+		queue.enqueue("conv-carry", "queued-a");
+		queue.enqueue("conv-carry", "queued-b");
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => simpleProvider(),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn,
+			resolveQueue: () => queue,
+		});
+
+		const events: AgentEvent[] = [];
+		const unsub = orchestrator.subscribe("conv-carry", (e) => events.push(e));
+
+		orchestrator.startTurn({ conversationId: "conv-carry", text: "original" });
+		// Wait for the original turn + the carried turn to both seal.
+		await waitForSealedCount(orchestrator, "conv-carry", 2);
+		unsub();
+
+		// Two user-message events: the original prompt + the carried combined text.
+		const userMessages = events.filter(isUserMessage);
+		expect(userMessages).toHaveLength(2);
+		expect(userMessages[0]?.text).toBe("original");
+		expect(userMessages[1]?.text).toBe("queued-a\n\nqueued-b");
+
+		// No steering event — the carry case emits user-message, not steering.
+		expect(events.filter(isSteering)).toHaveLength(0);
+
+		// The queue was drained by the carry.
+		expect(queue.getQueue("conv-carry")).toHaveLength(0);
+
+		// Both turns persisted (original + carry).
+		expect(store.data.get("conv-carry")?.length).toBeGreaterThanOrEqual(4);
+	});
+
+	it("post-seal: empty queue → no new turn", async () => {
+		const store = createInMemoryStore();
+		const queue = createTestQueue();
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => simpleProvider(),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn,
+			resolveQueue: () => queue,
+		});
+
+		const events: AgentEvent[] = [];
+		const unsub = orchestrator.subscribe("conv-no-carry", (e) => events.push(e));
+
+		orchestrator.startTurn({ conversationId: "conv-no-carry", text: "original" });
+		await waitForSealed(orchestrator, "conv-no-carry");
+		// Give the carry check a chance to run (it's in the finally, synchronous
+		// after turn-sealed, but await yields first).
+		await new Promise<void>((resolve) => setTimeout(resolve, 10));
+		unsub();
+
+		// Only one user-message (the original) — no carry turn.
+		expect(events.filter(isUserMessage)).toHaveLength(1);
+		expect(events.filter((e) => e.type === "turn-sealed")).toHaveLength(1);
+	});
+});
+
+// --- enqueue facade (the single entry transports call) ---
+
+describe("enqueue", () => {
+	it("enqueue when idle → starts a turn (startedTurn:true)", async () => {
+		const store = createInMemoryStore();
+		const queue = createTestQueue();
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => simpleProvider(),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn,
+			resolveQueue: () => queue,
+		});
+
+		const result = orchestrator.enqueue({ conversationId: "conv-idle", text: "hello" });
+		expect(result.startedTurn).toBe(true);
+		expect(result.queue).toHaveLength(0);
+
+		await waitForSealed(orchestrator, "conv-idle");
+
+		// The turn ran and persisted.
+		expect(store.data.get("conv-idle")).toBeDefined();
+		expect(store.data.get("conv-idle")?.length).toBeGreaterThanOrEqual(2);
+	});
+
+	it("enqueue when active → queues (startedTurn:false, snapshot with the message)", async () => {
+		const store = createInMemoryStore();
+		const queue = createTestQueue();
+
+		let resolveFirst: (() => void) | undefined;
+		const firstBlocker = new Promise<void>((resolve) => {
+			resolveFirst = resolve;
+		});
+		let callCount = 0;
+		const blockingFirstRunTurn = async (_input: RunTurnInput): Promise<RunTurnResult> => {
+			callCount++;
+			if (callCount === 1) {
+				await firstBlocker;
+			}
+			return {
+				messages: [{ role: "assistant", chunks: [{ type: "text", text: "done" }] }],
+				usage: { inputTokens: 1, outputTokens: 1 },
+				finishReason: "stop",
+			};
+		};
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => simpleProvider(),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn: blockingFirstRunTurn,
+			resolveQueue: () => queue,
+		});
+
+		// Start the original turn (it blocks in runTurn).
+		orchestrator.startTurn({ conversationId: "conv-active", text: "first" });
+		// Let the turn reach the blocked runTurn call.
+		await new Promise<void>((resolve) => setTimeout(resolve, 10));
+
+		// Enqueue while active.
+		const result = orchestrator.enqueue({ conversationId: "conv-active", text: "second" });
+		expect(result.startedTurn).toBe(false);
+		expect(result.queue).toHaveLength(1);
+		expect(result.queue[0]?.text).toBe("second");
+
+		// The queue holds the enqueued message.
+		expect(queue.getQueue("conv-active")).toHaveLength(1);
+
+		// Release the original turn → it seals → post-seal carry starts a new
+		// turn with the enqueued message. Subscribe before releasing to catch
+		// both turn-sealed events.
+		const sealed = waitForSealedCount(orchestrator, "conv-active", 2);
+		resolveFirst?.();
+		await sealed;
+	});
+
+	it("enqueue when active + no queue ext → startedTurn:false, empty queue (degraded)", async () => {
+		const store = createInMemoryStore();
+
+		let resolveFirst: (() => void) | undefined;
+		const firstBlocker = new Promise<void>((resolve) => {
+			resolveFirst = resolve;
+		});
+		let callCount = 0;
+		const blockingFirstRunTurn = async (_input: RunTurnInput): Promise<RunTurnResult> => {
+			callCount++;
+			if (callCount === 1) {
+				await firstBlocker;
+			}
+			return {
+				messages: [{ role: "assistant", chunks: [{ type: "text", text: "done" }] }],
+				usage: { inputTokens: 1, outputTokens: 1 },
+				finishReason: "stop",
+			};
+		};
+
+		const { orchestrator } = createSessionOrchestrator({
+			conversationStore: store,
+			resolveProvider: () => simpleProvider(),
+			resolveTools: noTools,
+			applyToolsFilter: identityApplyToolsFilter,
+			runTurn: blockingFirstRunTurn,
+			// resolveQueue omitted — no queue extension loaded (degraded)
+		});
+
+		orchestrator.startTurn({ conversationId: "conv-degraded", text: "first" });
+		await new Promise<void>((resolve) => setTimeout(resolve, 10));
+
+		// Enqueue while active, but no queue ext → message dropped, empty snapshot.
+		const result = orchestrator.enqueue({ conversationId: "conv-degraded", text: "second" });
+		expect(result.startedTurn).toBe(false);
+		expect(result.queue).toHaveLength(0);
+
+		// Release the original turn; no carry (no queue ext).
+		const sealed = waitForSealed(orchestrator, "conv-degraded");
+		resolveFirst?.();
+		await sealed;
+	});
+});