fix(server,web): steering survives reloads with visible content, and file attachments steer too (#157)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Yaowei ZhengandClaude Fable 5 authored and GitHub committed 2026-08-03 15:35:32 +08:00
1 parent 835dfb5c5a
commit a0968496b7
17 files changed
+554 -68

No files matched your search

+33 -4
View File
@@ -689,16 +689,39 @@ export interface TaskCreateResponse {
* task POST).
*/
export interface SteerRequest {
/** Message text (trimmed server-side); may be empty when `images` carries the message. */
/** Message text (trimmed server-side); may be empty when `images` or `files` carries the message. */
text: string;
/**
* Images sent with the steering message (`data:` or http(s) URLs, same rule as
* `TaskInputPart.image_url`): delivered as user image messages right behind the
* `[user_steering]` text. A model without vision receives them as scratchpad path lines
* instead, exactly as it would a Prompt's images. At least one of `text` / `images` must
* be non-empty.
* instead, exactly as it would a Prompt's images. At least one of `text` / `images` /
* `files` must be non-empty.
*/
images?: string[];
/**
* File attachments riding the steering message — the same shape, caps and handling as a
* task input's `{type:"file"}` parts: written into the Session scratchpad and delivered
* as `[attached file: <path>]` lines on the `[user_steering]` text, so a file-only draft
* steers exactly like an image-only one instead of falling back to the follow-up queue.
*/
files?: { fileName: string; dataUrl: string }[];
}
/**
* One steering message queued on the server but not yet delivered to the model (delivery
* happens at the next input assembly between turns). Carried on `task_state` events and the
* SSE subscribe snapshot so the composer's "steering queued" hint — including what was sent —
* survives reloads; entries leave the list as their `[user_steering]` message appears on the
* stream, and the whole list drops when the run exits (core discards undelivered steering).
*/
export interface PendingSteeringInfo {
/** The message text as accepted (trimmed); may be empty when images/files carry the message. */
text: string;
/** Number of images that rode along. */
images: number;
/** Number of file attachments that rode along. */
files: number;
}
export interface ApprovalDecisionRequest {
@@ -727,7 +750,13 @@ export type ServerEvent =
*/
| { type: "approval_request"; toolCall: OmniMessage<ToolCallPayload>; origin?: string[] }
/** Session run status flip (for toggling the input area and list); `queued` = queued follow-up count (see TaskCreateRequest.queueIfBusy). */
| { type: "task_state"; state: SessionStatus; queued?: number }
| {
type: "task_state";
state: SessionStatus;
queued?: number;
/** Steering messages queued but not yet delivered (absent = none): lets the composer's hint and its content survive reloads. */
pendingSteering?: PendingSteeringInfo[];
}
/** The model-generated title after the first turn has been persisted (for in-place list updates). */
| { type: "session_title"; sessionId: string; title: string }
/** Last-Event-ID has been evicted from the buffer: the frontend should re-fetch the history endpoint before continuing to consume this connection. */
+50 -9
View File
@@ -191,6 +191,26 @@ function parseSteerImages(body: Record<string, unknown>): string[] {
return images.map((url, i) => requireImageUrl(url, `images[${i}]`));
}
/**
* Validate the optional `files` field of a steer request: the same shape and caps as a task
* input's `{type:"file"}` parts (parseAttachmentPart, budget re-checked per item), absent or
* empty = no attachments.
*/
function parseSteerFiles(body: Record<string, unknown>): TaskAttachment[] {
const files = body.files;
if (files === undefined) return [];
if (!Array.isArray(files)) throw badRequest("files must be an array.");
const attachments: TaskAttachment[] = [];
files.forEach((item, i) => {
if (item === null || typeof item !== "object" || Array.isArray(item)) {
throw badRequest(`files[${i}] must be an object.`);
}
attachments.push(parseAttachmentPart(item as Record<string, unknown>, i, "files"));
assertAttachmentBudget(attachments);
});
return attachments;
}
/**
* Validate the optional `goal` field of a task request: absent = a regular task (null);
* present = goal mode with a token budget (a positive integer, or -1/omitted = unlimited).
@@ -506,11 +526,15 @@ export function sessionsRoutes(deps: AppDeps): Hono<AppEnv> {
// this as authoritative, eliminating input-area lockup or premature Task closure
// caused by a stale running/idle in the list; followed by replaying all still-pending
// approval requests.
const pendingSteering = deps.manager.pendingSteeringOf(row.sessionId);
const initialEvents: ServerEvent[] = [
{
type: "task_state",
state: deps.manager.statusOf(row.sessionId),
queued: deps.manager.pendingFollowUpCount(row.sessionId),
// Undelivered steering rides the snapshot too, so the composer's "steering queued"
// hint (and what it says) survives a reload.
...(pendingSteering.length > 0 ? { pendingSteering } : {}),
},
...deps.manager.pendingApprovals(row.sessionId).map((p) => ({
type: "approval_request" as const,
@@ -602,18 +626,35 @@ export function sessionsRoutes(deps: AppDeps): Hono<AppEnv> {
const body = await readJson(c);
const text = typeof body.text === "string" ? body.text.trim() : "";
const images = parseSteerImages(body);
// Either half can carry the message on its own: an image with no caption is a complete
// steering message, and so is plain text.
if (!text && images.length === 0) {
throw badRequest("text or images must carry the steering message.");
const files = parseSteerFiles(body);
// Any part can carry the message on its own: an image or a file with no caption is a
// complete steering message, and so is plain text.
if (!text && images.length === 0 && files.length === 0) {
throw badRequest("text, images or files must carry the steering message.");
}
// The wire shape becomes core's: a user text message (omitted when the images are the
// whole message, so the fold's path lines aren't preceded by a blank one) plus one image
// message each — the same input a normal task would carry.
deps.manager.steer(row.sessionId, [
...(text ? [userText(text)] : []),
...images.map((url) => imageUrlMessage(url)),
]);
// message each — the same input a normal task would carry. File attachments land in the
// Session scratchpad exactly as a task's do and ride as `[attached file: <path>]` lines
// on the steering text (a files-only input becomes a line-only text message).
const { input, written } = await attachFilesToInput(
[...(text ? [userText(text)] : []), ...images.map((url) => imageUrlMessage(url))],
files,
scratchpadDir(deps.config.root, row.projectId, row.agentId),
row.sessionId,
);
try {
deps.manager.steer(row.sessionId, input, {
text,
images: images.length,
files: files.length,
});
} catch (err) {
// 409 (not running) or any other refusal: the files must not stay behind — the
// frontend falls back to a normal task POST, which writes its own copies.
await removeAttachments(written);
throw err;
}
return c.body(null, 202);
});
+51 -5
View File
@@ -34,6 +34,7 @@ import {
goalTokenDelta,
isGoalRoundInput,
isSessionMeta,
parseUserSteeringText,
stripLeadingMarkerBlocks,
tracesDir,
} from "@prismshadow/penguin-core";
@@ -46,7 +47,7 @@ import type {
TextPayload,
ThinkingLevelName,
} from "@prismshadow/penguin-core";
import type { ServerEvent, SessionStatus } from "../api/types.js";
import type { PendingSteeringInfo, ServerEvent, SessionStatus } from "../api/types.js";
import { HttpError, isMissingCredential, modelCredentialMissing } from "../http/errors.js";
import type { GoalsRepo } from "../db/repos/goals.js";
import type { SessionRow, SessionsRepo } from "../db/repos/sessions.js";
@@ -241,6 +242,14 @@ interface RuntimeEntry {
* Deliberately NOT discarded on abort: they are future tasks the user explicitly queued.
*/
followUps: QueuedFollowUp[];
/**
* Steering messages queued on core but not yet delivered to the model — a display mirror
* of core's steering queue (same FIFO order), so the composer's "steering queued" hint and
* its content survive reloads: pushed on `steer`, shifted as each `[user_steering]`
* message appears on the drive stream, dropped wholesale when the run exits (core
* discards undelivered steering on abort/completion). Broadcast on every `task_state`.
*/
pendingSteering: PendingSteeringInfo[];
/** Timestamp of last activity (refreshed on load / status flip / drive completion), used for idle-eviction checks. */
lastActivityMs: number;
}
@@ -265,6 +274,13 @@ function agentKey(projectId: string, agentId: string): string {
}
/** If msg is a run_subagent tool call carrying a `prompt`, return its id and prompt (for use as the subagent's title); otherwise null. */
/** Whether this main-stream message is a delivered `[user_steering]` user text (one per queued steering entry, see core's steeringMessages). */
function isDeliveredSteering(msg: OmniMessage): boolean {
const p = msg.payload as { type?: string; role?: string; text?: string };
if (msg.type !== "model_msg" || p.type !== "text" || p.role !== "user") return false;
return typeof p.text === "string" && parseUserSteeringText(p.text) !== null;
}
function runSubagentCall(msg: OmniMessage): { toolCallId: string; prompt: string } | null {
const p = msg.payload as {
type?: string;
@@ -372,6 +388,11 @@ export class SessionManager {
return this.entries.get(sessionId)?.followUps.length ?? 0;
}
/** Steering messages queued but not yet delivered to the model (display mirror; see RuntimeEntry.pendingSteering). */
pendingSteeringOf(sessionId: string): PendingSteeringInfo[] {
return this.entries.get(sessionId)?.pendingSteering ?? [];
}
/**
* Live tail of a running session: one synthetic `partial_* start` OmniMessage per open
* streaming fragment, carrying the full accumulated content so far (see live-tail.ts).
@@ -406,6 +427,7 @@ export class SessionManager {
running: null,
generation: this.generationOf(row.projectId, row.agentId),
followUps: [],
pendingSteering: [],
lastActivityMs: Date.now(),
});
}
@@ -783,8 +805,13 @@ export class SessionManager {
* publishes). 409 when the Session isn't running a Task (idle / compacting / not loaded)
* or the run finished in the race window — the caller falls back to submitting a normal
* task, which carries the same text and images.
*
* `info` is the queued message's display summary: mirrored on the entry and broadcast via
* `task_state` (and the SSE subscribe snapshot) until the delivered `[user_steering]`
* message is observed on the stream — that is what keeps the composer's "steering queued"
* hint, content included, alive across reloads.
*/
steer(sessionId: string, input: OmniMessage[]): void {
steer(sessionId: string, input: OmniMessage[], info: PendingSteeringInfo): void {
const entry = this.entries.get(sessionId);
if (!entry || entry.status !== "running" || !entry.session.steer(input)) {
throw new HttpError(
@@ -793,7 +820,9 @@ export class SessionManager {
"This Session has no Task in progress; send the message as a new task instead.",
);
}
entry.pendingSteering.push(info);
entry.lastActivityMs = Date.now();
this.publishState(entry, entry.status);
}
/**
@@ -1039,6 +1068,7 @@ export class SessionManager {
running: null,
generation,
followUps: [],
pendingSteering: [],
lastActivityMs: Date.now(),
};
this.entries.set(currentId, entry);
@@ -1112,6 +1142,13 @@ export class SessionManager {
if (denied) subagentPrompts.delete(denied);
const settled = settledToolCallId(msg);
if (settled) subagentPrompts.delete(settled);
// Steering delivery: core emits exactly one `[user_steering]` user text per queued
// entry, in queue order — shift the display mirror and re-broadcast so the
// composer's "steering queued" hint retires the moment the message is on stream.
if (entry.pendingSteering.length > 0 && isDeliveredSteering(msg)) {
entry.pendingSteering.shift();
this.publishState(entry, entry.status);
}
} else if (isSessionMeta(msg)) {
// Subagent registration is only a "side effect" — it must never interrupt the
// main run flow on error: wrap the whole thing in a defensive try/catch.
@@ -1201,6 +1238,10 @@ export class SessionManager {
entry.status = "idle";
entry.abort = null;
entry.running = null;
// The run is over, so core has discarded any undelivered steering (see ContextEngine's
// steeringQueue) — drop the mirror with it; the idle publish below broadcasts the
// now-empty state.
entry.pendingSteering = [];
entry.lastActivityMs = Date.now();
this.publishState(entry, "idle");
if (titleSource && titleSource.userExcerpt.trim()) {
@@ -1305,9 +1346,14 @@ export class SessionManager {
}
private publishState(entry: RuntimeEntry, state: SessionStatus): void {
// Every state flip also reports the queued follow-up count, so subscribers can render
// the "N queued" hint without a dedicated event type.
this.publishEvent(entry, { type: "task_state", state, queued: entry.followUps.length });
// Every state flip also reports the queued follow-up count and the undelivered steering
// mirror, so subscribers can render both hints without a dedicated event type.
this.publishEvent(entry, {
type: "task_state",
state,
queued: entry.followUps.length,
...(entry.pendingSteering.length > 0 ? { pendingSteering: entry.pendingSteering } : {}),
});
}
private publishEvent(entry: RuntimeEntry, event: ServerEvent): void {
@@ -105,10 +105,15 @@ export interface TaskAttachment {
/**
* Validate one `{type:"file"}` input part. Shape problems are 400s in the same style as the
* neighbouring text/image checks; only the size cap answers 413 (`file_too_large`, the code
* the Web App already has copy for). `index` is the part's position in `input`, so the message
* points at the offending item like the other input errors do.
* the Web App already has copy for). `index` is the part's position in the request array
* named by `field` (default `input`; the steer route passes `files`), so the message points
* at the offending item like the other input errors do.
*/
export function parseAttachmentPart(part: Record<string, unknown>, index: number): TaskAttachment {
export function parseAttachmentPart(
part: Record<string, unknown>,
index: number,
field = "input",
): TaskAttachment {
const fileName = part.fileName;
// Path separators and `..` are rejected rather than sanitized away: the name is the user's,
// and a name that looks like a path means the caller is confused about the contract (the
@@ -122,7 +127,7 @@ export function parseAttachmentPart(part: Record<string, unknown>, index: number
fileName.includes("\0")
) {
throw badRequest(
`input[${index}].fileName must be a non-empty file name without path separators or "..".`,
`${field}[${index}].fileName must be a non-empty file name without path separators or "..".`,
);
}
const dataUrl = part.dataUrl;
@@ -133,11 +138,11 @@ export function parseAttachmentPart(part: Record<string, unknown>, index: number
const match =
typeof dataUrl === "string" ? /^data:[^,]*;base64,([A-Za-z0-9+/=\s]+)$/.exec(dataUrl) : null;
if (!match) {
throw badRequest(`input[${index}].dataUrl must be a base64 data: URL of the file's bytes.`);
throw badRequest(`${field}[${index}].dataUrl must be a base64 data: URL of the file's bytes.`);
}
const bytes = Buffer.from(match[1]!, "base64");
if (bytes.length === 0) {
throw badRequest(`input[${index}].dataUrl decodes to an empty file.`);
throw badRequest(`${field}[${index}].dataUrl decodes to an empty file.`);
}
if (bytes.length > MAX_ATTACHMENT_BYTES) {
throw new HttpError(
+57 -1
View File
@@ -20,6 +20,7 @@ import {
thinkingMessage,
toolCall,
toolCallOutput,
userSteeringText,
userText,
withOrigin,
} from "@prismshadow/penguin-core";
@@ -248,7 +249,7 @@ describe("session-manager", () => {
const manager = makeManager(loaderOf(fake));
const steerErr = (text: string): unknown => {
try {
manager.steer("session-1", [userText(text)]);
manager.steer("session-1", [userText(text)], { text, images: 0, files: 0 });
return null;
} catch (e) {
return e;
@@ -277,6 +278,61 @@ describe("session-manager", () => {
expect((steerErr("post") as HttpError).code).toBe("not_running");
});
it("pendingSteering mirror: broadcast with task_state on steer, shifted at delivery, dropped at run end", async () => {
// Two approval parks, with core's delivery shape — one `[user_steering]` user text —
// yielded between them, so the mid-run shift is observable while the run keeps going.
const fake: RuntimeSession = {
...approvalFakeSession("session-1"),
steer: () => true,
async *run(_input: OmniMessage[], opts: { approve: ApproveFn; signal: AbortSignal }) {
const tc1 = toolCall({ name: "write_file", arguments: "{}", toolCallId: "tc-1" });
yield tc1;
yield approvalDecision(await opts.approve(tc1), "tc-1");
yield userText(userSteeringText("focus on tests"));
const tc2 = toolCall({ name: "write_file", arguments: "{}", toolCallId: "tc-2" });
yield tc2;
yield approvalDecision(await opts.approve(tc2), "tc-2");
yield assistantText("done");
},
};
const manager = makeManager(loaderOf(fake));
const events = capture("session-1");
await manager.startTask("session-1", [userText("go")]);
await waitFor(() => manager.pendingApprovalCount("session-1") === 1);
// Two queued steering messages: the mirror keeps both, in queue order.
manager.steer("session-1", [userText("a")], { text: "focus on tests", images: 0, files: 0 });
manager.steer("session-1", [userText("b")], { text: "later", images: 1, files: 2 });
expect(manager.pendingSteeringOf("session-1")).toEqual([
{ text: "focus on tests", images: 0, files: 0 },
{ text: "later", images: 1, files: 2 },
]);
// First delivery observed on the stream: the mirror shifts while the run is still going.
manager.decideApproval("session-1", "tc-1", "allow");
await waitFor(() => manager.pendingSteeringOf("session-1").length === 1);
expect(manager.statusOf("session-1")).toBe("running");
expect(manager.pendingSteeringOf("session-1")).toEqual([
{ text: "later", images: 1, files: 2 },
]);
// Run end: core discards undelivered steering, and the mirror goes with it.
await waitFor(() => manager.pendingApprovalCount("session-1") === 1);
manager.decideApproval("session-1", "tc-2", "allow");
await waitFor(() => manager.statusOf("session-1") === "idle");
expect(manager.pendingSteeringOf("session-1")).toEqual([]);
// task_state broadcasts traced the whole lifecycle: grow to 1, 2, shift to 1, gone at idle.
const states = serverEvents(events).filter((e) => e.type === "task_state");
const mirrored = states
.filter((e) => e.pendingSteering !== undefined)
.map((e) => (e.pendingSteering as unknown[]).length);
expect(mirrored).toEqual([1, 2, 1]);
const last = states[states.length - 1]!;
expect(last.state).toBe("idle");
expect(last.pendingSteering).toBeUndefined();
});
it("queueIfBusy: enqueues while running, auto-starts in order after each finish; abort keeps the queue", async () => {
sessions.updateApprovalMode("session-1", "always-ask");
const runInputs: string[][] = [];
+115 -2
View File
@@ -1,13 +1,26 @@
/**
* Integration tests for POST /api/sessions/:id/steer (mid-run steering):
* - 202 while a Task is running, forwarding the trimmed text and its images to the core session;
* - 400 when neither text nor images carry a message, and for malformed image URLs;
* - 400 when neither text nor images nor files carry a message, and for malformed
* image URLs / file parts;
* - file attachments land in the Session scratchpad and ride the steering text as
* `[attached file: <path>]` lines (a 409 cleans them up again);
* - the SSE subscribe snapshot carries the pending-steering mirror (task_state), which is
* what keeps the composer's "steering queued" hint alive across reloads;
* - 409 not_running when the Session is idle (the frontend then falls back to a
* normal task POST);
* - 404 for foreign/unknown sessions (via the shared resolveSession lookup).
*/
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { approvalDecision, assistantText, toolCall, userText } from "@prismshadow/penguin-core";
import { readFile, readdir, realpath } from "node:fs/promises";
import path from "node:path";
import {
approvalDecision,
assistantText,
scratchpadDir,
toolCall,
userText,
} from "@prismshadow/penguin-core";
import type { ApproveFn, OmniMessage } from "@prismshadow/penguin-core";
import type { SessionRow } from "../src/db/repos/sessions.js";
import type { RuntimeSession } from "../src/runtime/session-manager.js";
@@ -154,4 +167,104 @@ describe("steer route", () => {
const res = await api.post(`/api/sessions/session-ghost/steer`, { text: "x" });
expect(res.status).toBe(404);
});
it("files ride the steering text as [attached file] lines — and carry the message alone", async () => {
await t.deps.manager.startTask(SID, [userText("go")]);
await waitFor(() => t.deps.manager.pendingApprovalCount(SID) === 1);
const data = `data:text/plain;base64,${Buffer.from("hello notes").toString("base64")}`;
const captioned = await api.post(`/api/sessions/${SID}/steer`, {
text: " read this ",
files: [{ fileName: "notes.txt", dataUrl: data }],
});
expect(captioned.status).toBe(202);
// A file with no caption is a complete steering message: the attachment line becomes a
// line-only text message (same shared rule as a task's attachments-only input).
const bare = await api.post(`/api/sessions/${SID}/steer`, {
text: "",
files: [{ fileName: "solo.txt", dataUrl: data }],
});
expect(bare.status).toBe(202);
expect(steered).toHaveLength(2);
const first = shape(steered[0]!);
expect(first).toHaveLength(1);
expect(first[0]).toMatch(/^text:read this\n\n\[attached file: .*notes\.txt\]$/);
const second = shape(steered[1]!);
expect(second).toHaveLength(1);
expect(second[0]).toMatch(/^text:\[attached file: .*solo\.txt\]$/);
// The bytes really landed in this Session's scratchpad. Directories are compared via
// realpath, not string prefixes: on the Windows CI runner the temp root mixes 8.3
// short and long name forms, so two spellings of the same directory are expected.
const written = /\[attached file: (.*)\]/.exec(first[0]!)![1]!;
const expectedDir = path.join(
scratchpadDir(t.root, "steerer-default_project", "default_agent"),
SID,
);
expect(await realpath(path.dirname(written))).toBe(await realpath(expectedDir));
expect(path.basename(written)).toBe("notes.txt");
expect(await readFile(written, "utf8")).toBe("hello notes");
// Same validation as a task input's file parts, under the steer request's own field name.
expect(
(await api.post(`/api/sessions/${SID}/steer`, { text: "x", files: "nope" })).status,
).toBe(400);
const evil = await api.post(`/api/sessions/${SID}/steer`, {
text: "x",
files: [{ fileName: "../evil.txt", dataUrl: data }],
});
expect(evil.status).toBe(400);
expect(((await evil.json()) as { error: { message: string } }).error.message).toContain(
"files[0]",
);
expect(
(
await api.post(`/api/sessions/${SID}/steer`, {
text: "x",
files: [{ fileName: "a.txt", dataUrl: "nope" }],
})
).status,
).toBe(400);
expect(steered).toHaveLength(2);
t.deps.manager.decideApproval(SID, "tc-steer", "allow");
await waitFor(() => t.deps.manager.statusOf(SID) === "idle");
});
it("a 409 steer leaves no attachment behind (the fallback normal send writes its own copy)", async () => {
const data = `data:text/plain;base64,${Buffer.from("orphan?").toString("base64")}`;
const res = await api.post(`/api/sessions/${SID}/steer`, {
text: "",
files: [{ fileName: "orphan.txt", dataUrl: data }],
});
expect(res.status).toBe(409);
const dir = path.join(scratchpadDir(t.root, "steerer-default_project", "default_agent"), SID);
expect(await readdir(dir).catch(() => [])).toEqual([]);
});
it("the SSE subscribe snapshot carries the pending-steering mirror (what makes the hint survive reloads)", async () => {
await t.deps.manager.startTask(SID, [userText("go")]);
await waitFor(() => t.deps.manager.pendingApprovalCount(SID) === 1);
await api.post(`/api/sessions/${SID}/steer`, { text: "hold on" });
expect(t.deps.manager.pendingSteeringOf(SID)).toEqual([
{ text: "hold on", images: 0, files: 0 },
]);
// The first SSE frames are the initial task_state snapshot: it must carry the mirror.
const res = await api.get(`/api/sessions/${SID}/stream`);
const reader = res.body!.getReader();
let seen = "";
for (let i = 0; i < 5 && !seen.includes("task_state"); i += 1) {
const { value, done } = await reader.read();
if (done) break;
seen += new TextDecoder().decode(value);
}
await reader.cancel();
expect(seen).toContain('"task_state"');
expect(seen).toContain('"pendingSteering"');
expect(seen).toContain("hold on");
t.deps.manager.decideApproval(SID, "tc-steer", "allow");
await waitFor(() => t.deps.manager.statusOf(SID) === "idle");
});
});