composer steering: the general steering machinery (owner arc 2026-10-03)

STEERING, NOT FOLLOW-UPS — pi-agent-core's two rails, used deliberately:
agent.steer() injects at the next response boundary INSIDE the live run
(one logical turn, the model changes course); followUp() would only extend
an ended run — not built. The same rail job markers already ride
(steeringMode 'all', the 2026-09-26 doctrine).

- POST /api/chats/[id]/steer: 404/400 honest guards; no live turn → 409
  and the client silently falls back to a plain text send (the settle
  race); live → steerLiveChat hands the message to the live agent's
  steering queue AND parks it in the userSteers registry — the stranding
  safety net. A steer stranded behind a question termination or abort
  stays parked and rides the chat's NEXT run (handOffEligible), the same
  deliver-or-requeue lifecycle as job markers. Reconciliation at settle:
  delivered ids drop (reconcileUserSteers).
- Client: steer(text) queues + POSTs without flipping any turn state;
  the drain's verbatim message events upsert the real user card by the
  client-minted id (steer_<localid>) and the indicator fold drops the
  line. canSend no longer gates on turnInFlight — Enter mid-turn steers;
  Enter with an empty composer still stops; the square button stays Stop.
- The queued-steer indicator (owner spec): quiet one-liners above the
  composer shell — right-aligned to where its corner rounding starts,
  truncated with an ellipsis, fixed max width, ~half opacity, no
  background, no container. Tunable later.
- Verified from pi-agent-core source: steering drains per completed turn
  inside the loop; injected messages emit message_start/end verbatim;
  finishTurn {action:'end'} (our question stopper) exits BEFORE the
  steering poll — hence the registry.
This commit is contained in:
amosblomqvist
2026-10-03 03:01:51 +03:00
parent a3a048f572
commit 5271ca0e4a
14 changed files with 489 additions and 19 deletions
+23
View File
@@ -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_<localid>): 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<string, unknown>;
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[] {
+9
View File
@@ -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
+6
View File
@@ -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
+68 -7
View File
@@ -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<string, QueuedSteer[]>;
/** 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<string, UserSteerEntry[]>;
/** 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<void> {
* 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 {