diff --git a/app/api/chats/[id]/steer/route.ts b/app/api/chats/[id]/steer/route.ts new file mode 100644 index 0000000..ab41e60 --- /dev/null +++ b/app/api/chats/[id]/steer/route.ts @@ -0,0 +1,32 @@ +// POST /api/chats/[id]/steer — the composer's mid-turn steering rail +// (owner arc 2026-10-03). The learner's message for a RUNNING turn rides +// pi's steering queue: drained at the next response boundary INSIDE the +// same run (agent.steer + steeringMode "all"), forwarded verbatim on the +// live stream (the client upserts the card by the client-minted id), and +// persisted by the turn's own settlement. +// +// This is steering, NOT follow-ups: the route never starts a turn and +// never extends an ended one. No live turn → 409 and the client falls +// back to a plain text send (the settle race). Stranding safety: the +// registry copy survives a question termination or abort and rides the +// chat's next run. +import { ChatInputError, newUserSteer, parseUserSteerInput } from "@/src/core"; +import { getChat, steerLiveChat } from "@/src/server"; + +export async function POST(request: Request, ctx: { params: Promise<{ id: string }> }): Promise { + const { id } = await ctx.params; + if (!getChat(id)) return Response.json({ error: "unknown chat" }, { status: 404 }); + let input; + try { + const body = (await request.json()) as { input?: unknown }; + input = parseUserSteerInput(body?.input); + } catch (err) { + if (err instanceof ChatInputError) return Response.json({ error: err.message }, { status: 400 }); + throw err; + } + const message = newUserSteer(input.text, input.id); + if (!steerLiveChat(id, message)) { + return Response.json({ error: "no live turn to steer" }, { status: 409 }); + } + return Response.json({ ok: true }); +} diff --git a/components/chat.tsx b/components/chat.tsx index ff8b629..b284b84 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -2,6 +2,7 @@ import { createContext, Fragment, useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from "react"; import { useChatSession } from "@/components/session"; +import { visibleQueuedSteers } from "@/components/session-core"; import { SetupDialog } from "@/components/setup-dialog"; import { Hint } from "@/components/ui/hint"; import { useZeroKeys, ZeroKeysPill } from "@/components/setup-cta"; @@ -22,6 +23,7 @@ import { AnsweredCard } from "@/components/answered-card"; import { artifactRenderers } from "@/components/artifacts"; import { SubagentSteerCard } from "@/components/artifacts/steer-card"; import { Composer } from "@/components/composer"; +import { QueuedSteersBody, type QueuedSteerLine } from "@/components/queued-steers"; import { DictationBar, joinDictation, useDictation } from "@/components/dictation"; import { JobsContext } from "@/components/jobs-context"; import { StatsBar } from "@/components/stats-bar"; @@ -187,7 +189,7 @@ export function Chat({ // (core/view.ts projectView); sends are ONE explicit input event each // ({chatId, input} — the stash indirection died with the SDK), and the // turn stream is the VERBATIM AgentEvent passthrough the server emits. - const { messages, nativeMessages, send, submitAnswer, dismissQuestion, setMessages, adoptRecord, status, error, stop, clearError, resumeStream, setReattachSettled } = + const { messages, nativeMessages, send, steer, steerQueue, submitAnswer, dismissQuestion, setMessages, adoptRecord, status, error, stop, clearError, resumeStream, setReattachSettled } = useChatSession({ chatId, initialMessages, initiallyInFlight }); // The reattach settled terminal → the sync-and-continue seam (below). @@ -507,7 +509,12 @@ export function Chat({ // the spoken text is deferred in the run's buffer — Enter must be able // to trigger the finish-first send (see onSend), not be gated out by an // empty textarea. If nothing was said, the send is a harmless no-op. - const canSend = (text.trim().length > 0 || dictation.active) && !pending && !turnInFlight; + // Steering (2026-10-03): a turn in flight no longer gates the composer — + // Enter while a turn runs STEERS it (the quiet queue above the composer); + // a pending question still owns the interaction (the dock replaces the + // composer, and there is no live turn to steer anyway — questions end + // turns). + const canSend = (text.trim().length > 0 || dictation.active) && !pending; const syncFromServer = useCallback(async () => { const res = await fetch(`/api/chats/${chatId}`); @@ -893,16 +900,16 @@ export function Chat({ }); const onSend = () => { - if (busy || pending) return; + if (pending) return; /* FINISH DICTATION FIRST (owner, 2026-10-01): a learner who keeps * talking right through Enter must not lose the tail — stop the run, * let the last finals land in the composer, then send the REF value * (the flush lands via setState, invisible to this closure). */ if (!dictation.active) { - doSend(); + doSteerOrSend(); return; } - void dictation.finish().then(doSend); + void dictation.finish().then(doSteerOrSend); }; const doSend = () => { if (busy || pending) return; @@ -913,6 +920,24 @@ export function Chat({ void send({ type: "text", text: t }); }; + /* STEER VS SEND (owner arc 2026-10-03): Enter while a turn runs steers + * it — the quiet queue + the live agent's steering rail; the turn's own + * stream lands the card when pi drains it. Any other state (idle, + * submitted microtask window, a wake turn between our renders) takes the + * plain send — a steer POST with no live turn would 409 and fall back + * to exactly that anyway, so we only branch on the state we trust. */ + const doSteerOrSend = () => { + if (pending) return; + const t = textRef.current.trim(); + if (!t) return; + if (turnInFlight) { + applyText(""); + void steer(t); + return; + } + doSend(); + }; + return ( } dictationCancel={dictation.cancel} + queuedSteers={visibleQueuedSteers(steerQueue, nativeMessages)} setupCta={ zeroKeys ? ( setSetupOpen(true)} /> @@ -1380,6 +1406,7 @@ function DockOverlay({ dictationSlot, dictationCancel, setupCta, + queuedSteers, }: { pending: ReturnType; /** Teach-phase pending question: arrives collapsed behind the reveal button (W5). */ @@ -1406,6 +1433,9 @@ function DockOverlay({ /** Esc in a pending question's field stops dictation (the strip is the * stats bar's; the stop belongs to the page-owned hook, threaded down). */ dictationCancel?: () => void; + /** The queued-steer indicator's lines (composer steering, 2026-10-03): + * already filtered by visibleQueuedSteers — only undrained steers. */ + queuedSteers: QueuedSteerLine[]; /** The persistent setup CTA (BYOK, owner 2026-09-22): the same pill * home and /chat/new carry, above the composer while slots are unset. */ setupCta?: React.ReactNode; @@ -1692,6 +1722,11 @@ function DockOverlay({
{setupCta} + {/* The queued-steer indicator (owner spec 2026-10-03): quiet + one-liners above the shell - right-aligned to where its corner + rounding starts, truncated, no background, no container. A + line dies when its card lands (visibleQueuedSteers). */} +
{pending ? ( diff --git a/components/composer.tsx b/components/composer.tsx index 146d32a..861fc92 100644 --- a/components/composer.tsx +++ b/components/composer.tsx @@ -18,7 +18,11 @@ import { Hint } from "@/components/ui/hint"; * can pin the decision (key events can't be simulated in * renderToStaticMarkup). */ export function enterAction(state: { busy: boolean; canSend: boolean }): "stop" | "send" | "none" { - if (state.busy) return "stop"; + // Composer steering (owner arc 2026-10-03): while a turn runs, typed + // text makes Enter a STEER (the queue-and-redirect send); Enter with + // an empty composer stays the stop shortcut. The button itself keeps + // the square — Stop stays one click away at all times. + if (state.busy) return state.canSend ? "send" : "stop"; if (state.canSend) return "send"; return "none"; } diff --git a/components/queued-steers.tsx b/components/queued-steers.tsx new file mode 100644 index 0000000..a8685ec --- /dev/null +++ b/components/queued-steers.tsx @@ -0,0 +1,36 @@ +/** + * THE QUEUED-STEER INDICATOR (owner spec 2026-10-03): steers sent but not + * yet drained render as quiet one-liners ABOVE the composer shell — + * right-aligned to where the shell's corner rounding starts, truncated + * with an ellipsis, no background, no container, roughly half opacity. + * Visible but very quiet; layout tunable later. + * + * A line dies the moment its card lands in the transcript: the drain's + * verbatim message events upsert the real user card by the client-minted + * id, and visibleQueuedSteers (session-core, pure) filters the queue + * against the message ids. A steer parked server-side (stranded behind a + * question) keeps its line — an honest "waiting for the next run". + */ +export interface QueuedSteerLine { + id: string; + text: string; +} + +/** The pure body (SSR-pinnable): one right-aligned truncated line per + * queued steer, stacked. */ +export function QueuedSteersBody({ steers }: { steers: QueuedSteerLine[] }) { + if (steers.length === 0) return null; + return ( +
+ {steers.map((s) => ( + + {s.text} + + ))} +
+ ); +} diff --git a/components/session-core.ts b/components/session-core.ts index 29261aa..bd7fc58 100644 --- a/components/session-core.ts +++ b/components/session-core.ts @@ -140,6 +140,29 @@ export interface SessionState { nativeMessages: LhMessage[]; status: SessionStatus; error: { message: string } | null; + /** Steers sent but not yet drained (composer steering, 2026-10-03): + * the queued indicator's truth. An entry dies when its id lands in + * the messages (the drain's verbatim events upsert the real card) — + * visibleQueuedSteers is the pure fold. */ + steerQueue: SteerQueueItem[]; +} + +/** One queued composer steer: the client-minted id (the upsert key) + text. */ +export interface SteerQueueItem { + id: string; + text: string; +} + +/** The indicator's fold (pure): a queued steer stays visible until its id + * lands in the message list — from the live stream's upsert, the settle + * sync, or a reload. A steer parked server-side (stranded behind a + * question) keeps its line: honest "waiting" until the next run drains it. */ +export function visibleQueuedSteers( + queue: readonly SteerQueueItem[], + messages: readonly LhMessage[], +): SteerQueueItem[] { + const ids = new Set(messages.map((m) => (m as { id?: string }).id)); + return queue.filter((s) => !ids.has(s.id)); } export interface SessionModelOptions extends SessionModelDeps { @@ -156,6 +179,8 @@ export class SessionModel { readonly chatId: string; /** Injectable fetch (tests); defaults to the global. */ fetch: typeof globalThis.fetch; + /** Steers sent but not yet drained (the queued indicator's truth). */ + private steerQueueValue: SteerQueueItem[] = []; private readonly pollMs: number; private readonly waitMaxMs: number; private onReattachSettled?: () => void; @@ -209,6 +234,7 @@ export class SessionModel { nativeMessages: [...this.base, ...this.tail], status: this.statusValue, error: this.errorValue, + steerQueue: this.steerQueueValue, }; } @@ -506,6 +532,48 @@ export class SessionModel { } } + /** + * STEER (composer steering, owner arc 2026-10-03): the learner's + * mid-turn message for the RUNNING turn. NOT a turn — no optimistic + * transcript card, no status flip: the message joins the queued + * indicator (the quiet line above the composer) and POSTs to the + * steer route, which hands it to the live agent's steering queue. + * When pi drains it at the next response boundary, the verbatim + * message events upsert the real card (the client-minted id is the + * key) and the indicator's fold drops the line. A 409 means the turn + * settled mid-send (the race): fall back to the plain text send — + * the entry is dropped and send() pushes its own optimistic card. + */ + async steer(text: string): Promise { + const t = text.trim(); + if (!t) return; + const id = `steer_${localId()}`; + this.steerQueueValue = [...this.steerQueueValue, { id, text: t }]; + this.onChange?.(); + const dropSteer = () => { + this.steerQueueValue = this.steerQueueValue.filter((s) => s.id !== id); + this.onChange?.(); + }; + try { + const res = await this.fetch(`/api/chats/${this.chatId}/steer`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ input: { type: "steer", id, text: t } }), + }); + if (res.ok) return; // parked or delivered — the stream lands the card + if (res.status === 409) { + dropSteer(); + await this.send({ type: "text", text: t }); // the settle race: plain send + return; + } + dropSteer(); + this.setError((await res.text()) || `The steer failed (${res.status}).`); + } catch (err) { + dropSteer(); + this.setError((err as Error).message); + } + } + /** The answer path (the dock): the answer IS the question call's tool * result (fork 1), riding as its input event. */ submitAnswer(tool: string, toolCallId: string, output: QuestionAnswerData): Promise { diff --git a/components/session.ts b/components/session.ts index a2cca28..7adfbf8 100644 --- a/components/session.ts +++ b/components/session.ts @@ -88,6 +88,7 @@ export function useChatSession({ chatId, initialMessages, initiallyInFlight }: U ); const send = useCallback((input: ChatInput) => void model.send(input), [model]); + const steer = useCallback((text: string) => void model.steer(text), [model]); const submitAnswer = useCallback( (tool: string, toolCallId: string, output: QuestionAnswerData) => void model.submitAnswer(tool, toolCallId, output), @@ -108,7 +109,9 @@ export function useChatSession({ chatId, initialMessages, initiallyInFlight }: U nativeMessages: state.nativeMessages, status: state.status, error: state.error, + steerQueue: state.steerQueue, send, + steer, submitAnswer, dismissQuestion, setMessages, diff --git a/src/core/input-events.ts b/src/core/input-events.ts index fe264f4..318e811 100644 --- a/src/core/input-events.ts +++ b/src/core/input-events.ts @@ -76,6 +76,29 @@ export function parseDismissInput(input: unknown): DismissInput { return { type: "dismiss", toolCallId: i.toolCallId, answer }; } +/** A mid-turn steer (owner arc 2026-10-03): the learner's composer message + * for a RUNNING turn — queued on pi's steering rail (agent.steer, drained + * at the next response boundary INSIDE the same run). NOT a follow-up: + * it never extends an ended run, and it never composes a turn of its own + * (a send with no live turn is a plain text send, which already exists). + * The id is client-minted (steer_): the queued indicator keys on + * it, and the drain's message events upsert the real card by the same id. */ +export type UserSteerInput = { type: "steer"; id: string; text: string }; + +/** Parse and validate a steer event off a request body. */ +export function parseUserSteerInput(input: unknown): UserSteerInput { + if (typeof input !== "object" || input === null) throw new ChatInputError("input must be an object"); + const i = input as Record; + if (i.type !== "steer") throw new ChatInputError(`unknown input type: ${String(i.type)}`); + if (typeof i.id !== "string" || i.id.length === 0) { + throw new ChatInputError("steer input requires a client-minted id"); + } + if (typeof i.text !== "string" || i.text.trim().length === 0) { + throw new ChatInputError("steer input requires non-empty text"); + } + return { type: "steer", id: i.id, text: i.text }; +} + /** The native appends a dismissal makes (pure — same orphan discipline * as answers: the call must be a DANGLING question call). */ export function dismissAppends(record: { messages: LhMessage[] }, input: DismissInput): LhMessage[] { diff --git a/src/core/native-messages.ts b/src/core/native-messages.ts index 85234ee..d15aaf2 100644 --- a/src/core/native-messages.ts +++ b/src/core/native-messages.ts @@ -103,6 +103,15 @@ export function newUserMessage(text: string): LhMessage & UserMessage { }; } +/** A steered user message (mid-turn composer steering, owner 2026-10-03): + * a plain user message with the CLIENT-minted steer id — the same object + * rides agent.steer(), the live stream's message events, and the settled + * record, so the optimistic queued indicator and the transcript card + * upsert by one identity. */ +export function newUserSteer(text: string, id: string): LhMessage & UserMessage { + return { id, role: "user", content: text, timestamp: Date.now() }; +} + /** The learner's answer — the question call's tool result (fork 1). * TWO AUDIENCES (owner, tool-result diet 2026-10-02): the MODEL sees the * compact learner-only record (compactAnswerForModel — the call's own diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 9a921a3..13c71e3 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -8,10 +8,12 @@ import type { ModelRole } from "../../config.ts"; import { deliverJobResult, noteSteerVisible, + reconcileUserSteers, registerLiveAgent, setJobRespawn, setTurnRunner, settleSteers, + steerLiveChat, unregisterLiveAgent, } from "./jobs.ts"; import { piTurnResponse, runPiTurn } from "./pi/ui-stream.ts"; @@ -425,6 +427,10 @@ export async function handleChatInput( // the pump take any stragglers down the idle wake. unregisterLiveAgent(chatId); settleSteers(chatId, stamped); + // Composer steers (2026-10-03): delivered ids dropped, stranded + // ones stay parked for the chat's next run (the registry is the + // stranding safety net — the agent's undrained copy died above). + reconcileUserSteers(chatId, stamped); }, }); // The turn is now a live background object: register it so any view diff --git a/src/server/jobs.ts b/src/server/jobs.ts index 184c0e0..40d59ae 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -64,7 +64,7 @@ import { type SteerMessage, } from "../core/contracts.ts"; import type { TurnInput } from "../core/input-events.ts"; -import { danglingQuestionOf, endsInDismissal } from "../core/native-messages.ts"; +import { danglingQuestionOf, endsInDismissal, type LhMessage } from "../core/native-messages.ts"; import { fireJobAbort, isJobLive } from "../agent/subagents/aborts.ts"; import { stores } from "./stores.ts"; import { isTurnRunning } from "./turn-liveness.ts"; @@ -97,6 +97,12 @@ interface WakeState { * delivery registry (Wave C: the agent's own follow-up queue is a * per-run VEHICLE, not the registry — settlement reconciles here). */ queues: Map; + /** chatId → user-steered messages awaiting drain (composer steering, + * 2026-10-03). Same deliver-or-requeue lifecycle as job markers: the + * live agent's steering queue is the VEHICLE, this is the registry — + * a steer stranded behind a question or an abort stays parked here + * and rides the chat's NEXT run (handOffEligible). */ + userSteers: Map; /** chatId → the agent of the CURRENTLY streaming turn (cleared at * settlement, before settleSteers). The pump's live path hands * markers to this agent's follow-up queue. */ @@ -113,7 +119,7 @@ interface LhWakeGlobal { function state(): WakeState { const g = globalThis as typeof globalThis & LhWakeGlobal; - return (g.__lhWakePipeline ??= { queues: new Map(), liveAgents: new Map(), pumping: new Set(), lastSweepAt: 0 }); + return (g.__lhWakePipeline ??= { queues: new Map(), liveAgents: new Map(), userSteers: new Map(), pumping: new Set(), lastSweepAt: 0 }); } /** chat-handler wires this at module init: its adapter around handleChatRequest. */ @@ -206,11 +212,25 @@ export function deliverJobResult(steer: SteerMessage): Promise { * coalescing). */ function handOffEligible(chatId: string, agent: Agent): void { const queue = state().queues.get(chatId); - if (!queue || queue.length === 0) return; - for (const entry of queue) { - if (entry.handedTo !== agent && entry.steer.customType !== "probe-handoff") { - agent.steer(entry.steer); - entry.handedTo = agent; + if (queue && queue.length > 0) { + for (const entry of queue) { + if (entry.handedTo !== agent && entry.steer.customType !== "probe-handoff") { + agent.steer(entry.steer); + entry.handedTo = agent; + } + } + } + // User steers (2026-10-03): same hand-off, any live run — a steer + // stranded behind a question or abort rides the chat's next run's + // first boundary. No probe-handoff exclusion: these are learner + // words, owed to whatever run is live. + const userQueue = state().userSteers.get(chatId); + if (userQueue && userQueue.length > 0) { + for (const entry of userQueue) { + if (entry.handedTo !== agent) { + agent.steer(entry.message); + entry.handedTo = agent; + } } } } @@ -226,6 +246,47 @@ export function registerLiveAgent(chatId: string, agent: Agent): void { handOffEligible(chatId, agent); } +/** A parked user steer: the message plus the agent it was last handed to + * (the double-handoff guard — a reconcile keeps it only when its id + * never landed in the persisted record). */ +export interface UserSteerEntry { + message: LhMessage; + handedTo?: Agent; +} + +/** The steer route's live hand-off (composer steering, 2026-10-03). + * Synchronous by design — no interleaving between the live-agent check + * and the queue() means a settling turn can never race the hand-off. + * Returns false when no turn is live: the caller 409s and the client + * falls back to a plain text send (the settle race). The registry copy + * is the stranding safety net: if this agent dies undelivered (question + * termination, abort), reconcileUserSteers keeps it parked for the + * chat's next run. */ +export function steerLiveChat(chatId: string, message: LhMessage): boolean { + const agent = state().liveAgents.get(chatId); + if (!agent) return false; + const entry: UserSteerEntry = { message }; + const q = state().userSteers.get(chatId) ?? []; + q.push(entry); + state().userSteers.set(chatId, q); + agent.steer(message); + entry.handedTo = agent; + return true; +} + +/** TURN SETTLEMENT RECONCILIATION for user steers: delivered ids drop + * (the record holds them); undelivered stay parked for the next run. + * Memory-only by design (same tradeoff as job markers): a server + * restart drops parked steers — and the whole turn dies with it anyway. */ +export function reconcileUserSteers(chatId: string, persisted: readonly LhMessage[]): void { + const q = state().userSteers.get(chatId); + if (!q || q.length === 0) return; + const ids = new Set(persisted.map((m) => (m as { id?: string }).id)); + const kept = q.filter((e) => !ids.has(e.message.id)); + if (kept.length === 0) state().userSteers.delete(chatId); + else state().userSteers.set(chatId, kept); +} + /** Clear the live agent at settlement — MUST run before settleSteers so the * reconciliation's re-pump can never hand a marker to the dying agent. */ export function unregisterLiveAgent(chatId: string): void { diff --git a/tests/chat-flow.test.ts b/tests/chat-flow.test.ts index 4a3584a..d39d5fc 100644 --- a/tests/chat-flow.test.ts +++ b/tests/chat-flow.test.ts @@ -45,6 +45,7 @@ import type { LhMessage } from "../src/core/native-messages.ts"; // handler, driven directly. Env is the runner's (the same the static // store imports already resolved under). const dismissRoute = await import("../app/api/chats/[id]/dismiss/route.ts"); +const steerRoute = await import("../app/api/chats/[id]/steer/route.ts"); import { PATHS } from "../config.ts"; // The key-presence turn gate (provider-key interlude, 2026-09-25): scripted @@ -1387,3 +1388,80 @@ describe("parallel rounds: the LAST resolution decides the wake (owner ruling 20 assert.match(prompt2, /dismissed/, "the dismissal rode the wake too"); }); }); + +// ─── composer steering (owner arc 2026-10-03): the route + the live rail ──── + +describe("composer steering: POST /api/chats/[id]/steer", () => { + const steerPost = (chatId: string, body: unknown): Promise => + steerRoute.POST( + new Request(`http://test/api/chats/${chatId}/steer`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }), + { params: Promise.resolve({ id: chatId }) }, + ); + + it("honest rejections: 404 unknown chat, 400 bad input, 409 when no turn is live", async () => { + const chat = createConfiguredChat("Test: steer rejections"); + createdChats.push(chat.id); + assert.equal((await steerPost("chat_nope", { input: { type: "steer", id: "steer_1", text: "hi" } })).status, 404); + assert.equal((await steerPost(chat.id, { input: { type: "steer", id: "", text: "hi" } })).status, 400); + assert.equal((await steerPost(chat.id, { input: { type: "text", text: "hi" } })).status, 400); + assert.equal((await steerPost(chat.id, { input: { type: "steer", id: "steer_1", text: " " } })).status, 400); + assert.equal((await steerPost(chat.id, { input: { type: "steer", id: "steer_1", text: "hi" } })).status, 409); + }); + + it("200 + the live steering queue when a turn runs; the drain lands the card and the model sees it", async () => { + // A GATED engine (the scriptedTeacher seam): the first factory + // awaits our release, holding the turn open mid-generation so the + // steer POST lands while the run is genuinely live. + const calls: string[][] = []; + let release!: () => void; + const gate = new Promise((res) => { + release = res; + }); + const faux = createFauxCore({}); + faux.setResponses([ + async (context: { messages: unknown[] }) => { + calls.push(context.messages.map((m) => JSON.stringify((m as { content: unknown }).content))); + await gate; + return fauxAssistantMessage([fauxText("Teaching the solar system…")]); + }, + (context: { messages: unknown[] }) => { + calls.push(context.messages.map((m) => JSON.stringify((m as { content: unknown }).content))); + return fauxAssistantMessage([fauxText("Steered — covering Venus too.")]); + }, + ]); + const engine = { model: faux.getModel(), streamFn: faux.streamSimple }; + + const chat = createConfiguredChat("Test: steer live"); + createdChats.push(chat.id); + + // The turn starts and holds at the first model call. + const turnRes = await handleChatRequest(post(chat.id, { type: "text", text: "teach me" }), agentFactory(engine as never)); + await waitFor(() => calls.length === 1, 8000, "first model call started"); + + // The steer lands mid-turn: 200, handed to the live agent. + const res = await steerPost(chat.id, { input: { type: "steer", id: "steer_live1", text: "also cover Venus" } }); + assert.equal(res.status, 200); + assert.equal((await res.json() as { ok: boolean }).ok, true); + + release(); // the first response completes → the drain → the second call + const body = await drain(turnRes); + assert.match(body, /also cover Venus/, "the drained steer rode the live stream verbatim"); + + // ONE turn: the second model call's context contains the steer + // BETWEEN the user message and the continuing assistant text. + assert.equal(calls.length, 2, "still one run — the steer continued it, not a new turn"); + assert.match(calls[1]!.join("\n"), /also cover Venus/, "the model saw the steer in its context"); + + // The record persists the user steer in its true position. + const saved = chatStore.getChat(chat.id)!; + const roles = saved.messages.map((m) => (m as { role?: string }).role); + assert.deepEqual(roles, ["user", "assistant", "user", "assistant"], "text → teaching → STEER → steered reply"); + const steerMsg = saved.messages[2] as { id?: string; content?: unknown }; + assert.equal(steerMsg.id, "steer_live1", "the client-minted id survived to the record"); + assert.equal(steerMsg.content, "also cover Venus"); + }); +}); diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 1a04dc0..e84863d 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -51,11 +51,13 @@ import { abortJob, deliverJobResult, noteSteerVisible, + reconcileUserSteers, recoverStaleJobs, registerLiveAgent, setJobRespawn, setTurnRunner, settleSteers, + steerLiveChat, unregisterLiveAgent, } from "../src/server/jobs.ts"; import { registerJobAbort, unregisterJobAbort } from "../src/agent/subagents/aborts.ts"; @@ -1642,3 +1644,59 @@ describe("provider-key job refusal (key-presence interlude)", () => { assert.equal(result.sources.length, 0, "the failure result matches the research convention"); }); }); + +// ─── composer steering (owner arc 2026-10-03): the user-steer registry ────── + +describe("composer steering: steerLiveChat + the deliver-or-requeue lifecycle", () => { + /** A minimal stand-in for pi's Agent — only steer() is touched. */ + function fakeAgent(): { handed: unknown[]; agent: unknown } { + const handed: unknown[] = []; + return { handed, agent: { steer: (m: unknown) => handed.push(m) } }; + } + + it("no live turn → false, nothing parked (the client falls back to a plain send)", () => { + const chatId = chatStore.createChat("steer: idle").id; + const msg = { id: "steer_x1", role: "user", content: "hold on", timestamp: 1 }; + assert.equal(steerLiveChat(chatId, msg as never), false); + const { handed, agent } = fakeAgent(); + registerLiveAgent(chatId, agent as never); // a later run hands NOTHING + assert.equal(handed.length, 0, "nothing was parked by the refused steer"); + unregisterLiveAgent(chatId); + }); + + it("live agent → steered AND parked (the stranding safety net)", () => { + const chatId = chatStore.createChat("steer: live").id; + const msg = { id: "steer_x2", role: "user", content: "also cover Venus", timestamp: 1 }; + const { handed, agent } = fakeAgent(); + registerLiveAgent(chatId, agent as never); + assert.equal(steerLiveChat(chatId, msg as never), true); + assert.equal(handed.length, 1, "handed to the live agent's steering queue"); + // Re-registering the SAME agent never re-hands (double-handoff guard). + registerLiveAgent(chatId, agent as never); + assert.equal(handed.length, 1); + unregisterLiveAgent(chatId); + }); + + it("settlement reconciliation: delivered ids drop, stranded ones ride the NEXT run", () => { + const chatId = chatStore.createChat("steer: reconcile").id; + const delivered = { id: "steer_d1", role: "user", content: "delivered", timestamp: 1 }; + const stranded = { id: "steer_s1", role: "user", content: "stranded behind a question", timestamp: 2 }; + const { handed, agent } = fakeAgent(); + registerLiveAgent(chatId, agent as never); + assert.equal(steerLiveChat(chatId, delivered as never), true); + assert.equal(steerLiveChat(chatId, stranded as never), true); + + // Settlement: the run persisted ONLY the delivered one (the stranded + // copy died with the agent — question termination / abort). + unregisterLiveAgent(chatId); + reconcileUserSteers(chatId, [delivered] as never); + + // The chat's next run (the answer turn) picks up the stranded steer. + const next = fakeAgent(); + registerLiveAgent(chatId, next.agent as never); + assert.equal(next.handed.length, 1, "exactly the stranded steer rides the next run"); + assert.deepEqual((next.handed[0] as { id: string }).id, "steer_s1"); + unregisterLiveAgent(chatId); + reconcileUserSteers(chatId, [stranded, delivered] as never); // cleanup + }); +}); diff --git a/tests/render-composer.test.tsx b/tests/render-composer.test.tsx index 4df65d2..900079c 100644 --- a/tests/render-composer.test.tsx +++ b/tests/render-composer.test.tsx @@ -57,13 +57,19 @@ describe("Composer", () => { }); }); -// The Enter decision: stop a busy stream, otherwise send. (The plan-verdict -// accept branch died 2026-09-26 with the verdict itself — the post-plan -// reply is a natural text message.) Pure function so the static-render -// harness can pin it (key events can't be simulated in renderToStaticMarkup). +// The Enter decision (composer steering, 2026-10-03): while a turn runs, +// typed text makes Enter a STEER; Enter with an empty composer stays the +// stop shortcut. (The plan-verdict accept branch died 2026-09-26 with the +// verdict itself — the post-plan reply is a natural text message.) Pure +// function so the static-render harness can pin it (key events can't be +// simulated in renderToStaticMarkup). describe("enterAction", () => { - it("busy stream: Enter stops", () => { - assert.equal(enterAction({ busy: true, canSend: true }), "stop"); + it("busy stream with text: Enter steers (the queued send)", () => { + assert.equal(enterAction({ busy: true, canSend: true }), "send"); + }); + + it("busy stream, empty composer: Enter stops", () => { + assert.equal(enterAction({ busy: true, canSend: false }), "stop"); }); it("idle: send when there is text, nothing when there is not", () => { diff --git a/tests/render-queued-steers.test.tsx b/tests/render-queued-steers.test.tsx new file mode 100644 index 0000000..d005353 --- /dev/null +++ b/tests/render-queued-steers.test.tsx @@ -0,0 +1,51 @@ +/** + * The queued-steer indicator's pins (composer steering, owner arc + * 2026-10-03): the quiet one-liners above the composer shell, and the + * visibleQueuedSteers fold that decides which queued steers still show. + */ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import { renderToStaticMarkup } from "react-dom/server"; +import { QueuedSteersBody } from "../components/queued-steers.tsx"; +import { visibleQueuedSteers } from "../components/session-core.ts"; + +describe("QueuedSteersBody (the quiet queue above the composer)", () => { + it("renders one truncated right-aligned line per queued steer — no background, no container chrome", () => { + const html = renderToStaticMarkup( + , + ); + assert.match(html, /data-testid="queued-steers"/); + assert.equal(html.match(/ { + assert.equal(renderToStaticMarkup(), ""); + }); +}); + +describe("visibleQueuedSteers (the fold: a line dies when its card lands)", () => { + it("a queued steer stays visible until its id lands in the messages", () => { + const queue = [ + { id: "steer_a", text: "a" }, + { id: "steer_b", text: "b" }, + ]; + const before = visibleQueuedSteers(queue, [{ id: "msg_1" } as never]); + assert.equal(before.length, 2, "nothing landed yet — both lines show"); + const after = visibleQueuedSteers(queue, [ + { id: "msg_1" } as never, + { id: "steer_a", role: "user", content: "a" } as never, + ]); + assert.deepEqual(after.map((s) => s.id), ["steer_b"], "the drained steer's line dropped"); + }); +});