From a0968496b715f89d7e62b36d6164aee19d9c1b1d Mon Sep 17 00:00:00 2001
From: Yaowei Zheng
Date: Mon, 3 Aug 2026 15:35:32 +0800
Subject: [PATCH] fix(server,web): steering survives reloads with visible
content, and file attachments steer too (#157)
Co-authored-by: Claude Fable 5
---
packages/server/src/api/types.ts | 37 +++++-
packages/server/src/http/routes/sessions.ts | 59 +++++++--
.../server/src/runtime/session-manager.ts | 56 ++++++++-
.../server/src/services/task-attachments.ts | 17 ++-
packages/server/test/session-manager.test.ts | 58 ++++++++-
packages/server/test/steer.test.ts | 117 +++++++++++++++++-
packages/web/e2e/steer-reload.spec.mjs | 89 +++++++++++++
packages/web/src/features/chat/chat-input.tsx | 86 +++++++++----
packages/web/src/features/chat/chat-page.tsx | 26 +++-
.../web/src/features/chat/composer-send.ts | 10 +-
.../src/features/chat/use-session-draft.ts | 8 +-
.../src/features/chat/use-session-stream.ts | 13 +-
.../web/src/lib/omni/stream-controller.ts | 5 +
packages/web/src/lib/strings-en.ts | 4 +
packages/web/src/lib/strings.ts | 3 +
packages/web/test/composer-send.test.ts | 10 +-
packages/web/test/stream-controller.test.ts | 24 +++-
17 files changed, 554 insertions(+), 68 deletions(-)
create mode 100644 packages/web/e2e/steer-reload.spec.mjs
diff --git a/packages/server/src/api/types.ts b/packages/server/src/api/types.ts
index d3306cc..9efd359 100644
--- a/packages/server/src/api/types.ts
+++ b/packages/server/src/api/types.ts
@@ -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: ]` 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; 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. */
diff --git a/packages/server/src/http/routes/sessions.ts b/packages/server/src/http/routes/sessions.ts
index 369b4b8..822eef2 100644
--- a/packages/server/src/http/routes/sessions.ts
+++ b/packages/server/src/http/routes/sessions.ts
@@ -191,6 +191,26 @@ function parseSteerImages(body: Record): 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): 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, 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 {
// 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 {
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: ]` 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);
});
diff --git a/packages/server/src/runtime/session-manager.ts b/packages/server/src/runtime/session-manager.ts
index 952984a..4ff9361 100644
--- a/packages/server/src/runtime/session-manager.ts
+++ b/packages/server/src/runtime/session-manager.ts
@@ -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 {
diff --git a/packages/server/src/services/task-attachments.ts b/packages/server/src/services/task-attachments.ts
index 6e8efc2..d7d2f73 100644
--- a/packages/server/src/services/task-attachments.ts
+++ b/packages/server/src/services/task-attachments.ts
@@ -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, index: number): TaskAttachment {
+export function parseAttachmentPart(
+ part: Record,
+ 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, 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, 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(
diff --git a/packages/server/test/session-manager.test.ts b/packages/server/test/session-manager.test.ts
index 6eacf6e..00279c8 100644
--- a/packages/server/test/session-manager.test.ts
+++ b/packages/server/test/session-manager.test.ts
@@ -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[][] = [];
diff --git a/packages/server/test/steer.test.ts b/packages/server/test/steer.test.ts
index 12acd7a..9755bea 100644
--- a/packages/server/test/steer.test.ts
+++ b/packages/server/test/steer.test.ts
@@ -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: ]` 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");
+ });
});
diff --git a/packages/web/e2e/steer-reload.spec.mjs b/packages/web/e2e/steer-reload.spec.mjs
new file mode 100644
index 0000000..f2cc32d
--- /dev/null
+++ b/packages/web/e2e/steer-reload.spec.mjs
@@ -0,0 +1,89 @@
+/**
+ * Steering across a reload (#136/#140): a message steered into a running Task must
+ * (a) leave no composer draft behind — reloading must not resurrect the sent text (the
+ * old behavior revived it as a draft, and re-sending duplicated the steering);
+ * (b) keep its "queued" hint — content included — across the reload (the server's
+ * pending-steering mirror rides task_state events and the SSE subscribe snapshot);
+ * (c) land in the transcript exactly once, as a [user_steering] chip, once delivered.
+ *
+ * The LLM is mock-llm.mjs's "slow stream test": a ~8s exec_command keeps the Task busy,
+ * leaving a wide window to steer and reload before delivery (steering is delivered at the
+ * next input assembly, i.e. when the tool output returns).
+ */
+import { test, expect } from "@playwright/test";
+import { provisionAndLogin } from "./auth.mjs";
+
+const BASE = process.env.BASE_URL;
+const MOCK = process.env.MOCK_URL;
+const U = "steeruser";
+const P = "password123";
+const STEER_TEXT = "steer: also check the logs";
+
+/** Create a session for the user's auto-provisioned project (models PUT is idempotent). */
+async function createSession(page, approvalMode) {
+ const projects = await (await page.request.get(`${BASE}/api/projects`)).json();
+ const projectId = projects.projects[0].projectId;
+ const put = await page.request.put(`${BASE}/api/projects/${projectId}/models`, {
+ data: {
+ defaultModel: { provider: "custom", modelId: "claude-4-8" },
+ models: [
+ {
+ provider: "custom",
+ modelId: "claude-4-8",
+ apiKey: "sk-mock",
+ baseUrl: MOCK,
+ contextWindow: 200000,
+ },
+ ],
+ },
+ });
+ expect(put.ok(), "put models").toBeTruthy();
+ const res = await page.request.post(
+ `${BASE}/api/projects/${projectId}/agents/default_agent/sessions`,
+ { data: { provider: "custom", modelId: "claude-4-8", approvalMode } },
+ );
+ expect(res.ok(), `create session: ${await res.text()}`).toBeTruthy();
+ return (await res.json()).session.sessionId;
+}
+
+test("a steered message survives reload as a queued hint with its content — not a draft — and lands exactly once", async ({
+ page,
+}) => {
+ await provisionAndLogin(page.request, U, P);
+ const sessionId = await createSession(page, "always-ask");
+ await page.goto(`${BASE}/chat/${sessionId}`);
+ const ta = page.locator("textarea").first();
+ await ta.waitFor();
+ await ta.fill("slow stream test");
+ await page.getByRole("button", { name: "发送" }).click();
+ await expect(page.getByText("exec_command").first()).toBeVisible();
+ await page.getByRole("button", { name: "允许" }).click();
+
+ // Steer while the ~8s tool run keeps the Task busy (default mid-run mode is steer).
+ await ta.fill(STEER_TEXT);
+ await ta.press("Enter");
+ // The queued hint shows the content (the server mirror, not just the local flag), and
+ // the composer is cleared.
+ await expect(page.getByText(`插话已排队,将随下一轮送达:${STEER_TEXT}`)).toBeVisible();
+ await expect(ta).toHaveValue("");
+
+ // Reload mid-run: the sent text must NOT come back as a draft (#136 — that resurrection
+ // is what produced duplicates), while the queued hint returns with its content.
+ await page.reload();
+ const ta2 = page.locator("textarea").first();
+ await ta2.waitFor();
+ await expect(page.getByText(`插话已排队,将随下一轮送达:${STEER_TEXT}`)).toBeVisible({
+ timeout: 5000,
+ });
+ await expect(ta2).toHaveValue("");
+
+ // Delivered with the next turn: the [user_steering] chip carries the text exactly once
+ // (one paragraph holds the chip label and the text), and the queued hint retires — one
+ // paragraph total also proves the hint no longer repeats the content.
+ await expect(page.getByText("Command finished; the result looks as expected.")).toBeVisible({
+ timeout: 30_000,
+ });
+ await expect(page.getByText("用户插话").first()).toBeVisible();
+ await expect(page.locator("p", { hasText: STEER_TEXT })).toHaveCount(1);
+ await expect(page.getByText(/插话已排队/)).toHaveCount(0);
+});
diff --git a/packages/web/src/features/chat/chat-input.tsx b/packages/web/src/features/chat/chat-input.tsx
index 8e15d5e..78153d2 100644
--- a/packages/web/src/features/chat/chat-input.tsx
+++ b/packages/web/src/features/chat/chat-input.tsx
@@ -69,6 +69,7 @@ import type {
ApprovalMode,
ModelInfo,
ModelRefDto,
+ PendingSteeringInfo,
SessionStatus,
SkillMetadataItem,
TaskInputPart,
@@ -1116,6 +1117,16 @@ function readDataUrl(file: File): Promise {
* One place, because every send path submits the same draft: the normal send, the follow-up
* queue, the @ handoff and the `/model` switch.
*/
+/** One-line summary of a queued steering message: its text, then image/file counts for what the text cannot show. */
+function steeringSummary(p: PendingSteeringInfo): string {
+ const parts: string[] = [];
+ const line = p.text.replace(/\s+/g, " ").trim();
+ if (line) parts.push(line);
+ if (p.images > 0) parts.push(S.chat.imagesInMessage(p.images));
+ if (p.files > 0) parts.push(S.chat.filesInMessage(p.files));
+ return parts.join(" \u00b7 ");
+}
+
function appendAttachmentParts(
input: TaskInputPart[],
images: string[],
@@ -1132,6 +1143,7 @@ export function ChatInput({
onSend,
onSteer,
steeringDeliveredCount,
+ pendingSteering = [],
onQueueFollowUp,
queuedFollowUps = 0,
onStop,
@@ -1179,18 +1191,30 @@ export function ChatInput({
onSend: (input: TaskInputPart[], goal: { budget: number } | null) => Promise;
/**
* Mid-run steering (session state only): while a Task is running, Enter/send queues the
- * trimmed text **and any attached images** for the running agent — delivered between turns
- * as a standalone `[user_steering]` user message followed by its images. `"queued"` clears
- * the text and images and shows the queued hint; `"not_running"` (409 race with completion)
- * makes the input fall back to its full normal send path; `"failed"` keeps the draft. When
- * absent (draft state), the input stays send-disabled while running, as before.
+ * trimmed text **and any attached images and files** for the running agent — delivered
+ * between turns as a standalone `[user_steering]` user message followed by its images,
+ * with the files riding the text as `[attached file: ]` lines (#140: a file-only
+ * draft steers exactly like an image-only one). `"queued"` clears the text, images and
+ * files and shows the queued hint; `"not_running"` (409 race with completion) makes the
+ * input fall back to its full normal send path; `"failed"` keeps the draft. When absent
+ * (draft state), the input stays send-disabled while running, as before.
*/
- onSteer?: (text: string, images: string[]) => Promise<"queued" | "not_running" | "failed">;
+ onSteer?: (
+ text: string,
+ images: string[],
+ files: { fileName: string; dataUrl: string }[],
+ ) => Promise<"queued" | "not_running" | "failed">;
/**
* Count of steering messages already visible in the message stream: the queued hint stays
* up until this increases past its value at queue time (i.e. the message was delivered).
*/
steeringDeliveredCount?: number;
+ /**
+ * The server's undelivered-steering mirror (from task_state events): each entry renders as
+ * a "steering queued" line with its content, so the hint — and what was sent — survives
+ * reloads (#136). The local post-202 flag only bridges until the first event arrives.
+ */
+ pendingSteering?: PendingSteeringInfo[];
/**
* Follow-up queue (session state only): posts the full input with `queueIfBusy` — a busy
* session holds it server-side and auto-sends it as an ordinary next task once the current
@@ -1498,13 +1522,14 @@ export function ChatInput({
[onHandoffTargetChange, onPendingModelChange],
);
- // Mid-run steering: while running, Enter/send queues the text **and the attached images**
- // for the running agent (delivered between turns as a [user_steering] user message followed
- // by its images) — so an image with no caption is a complete steering message on its own.
- // File attachments and selected skills stay in the draft for a later normal send: a
- // [use_skills] block is task-level setup, not something to hand a turn already under way. A
- // staged /agent or /model chip also blocks steering: the text belongs to the conversation
- // that switch is about to open, not to the agent running here.
+ // Mid-run steering: while running, Enter/send queues the text **and the attached images
+ // and files** for the running agent (delivered between turns as a [user_steering] user
+ // message followed by its images; files ride the text as [attached file: ] lines) —
+ // so an image or a file with no caption is a complete steering message on its own (#140).
+ // Selected skills stay in the draft for a later normal send: a [use_skills] block is
+ // task-level setup, not something to hand a turn already under way. A staged /agent or
+ // /model chip also blocks steering: the text belongs to the conversation that switch is
+ // about to open, not to the agent running here.
// `!goalOn`: with the goal chip engaged the text is an OBJECTIVE — steering it into a run
// that happens to be active (e.g. a schedule fired) would silently repurpose it.
//
@@ -1535,6 +1560,7 @@ export function ChatInput({
hasPendingModel: pendingModel !== null,
hasText: text.trim().length > 0,
hasImages: images.length > 0,
+ hasFiles: attachments.length > 0,
hasContent: draftHasContent,
});
const steerAction = running && midRun === "steer";
@@ -1976,16 +2002,17 @@ export function ChatInput({
await sendNormal(onQueueFollowUp!);
return;
}
- // Steering branch: queue the trimmed text and the attached images for the running agent;
- // both are sent and cleared together — file attachments and selected skills stay for a
- // normal send (a staged switch chip blocks this branch outright, see midRunAction).
+ // Steering branch: queue the trimmed text, the attached images and the attached files
+ // for the running agent; all are sent and cleared together — selected skills stay for
+ // a normal send (a staged switch chip blocks this branch outright, see midRunAction).
if (!steerAction) return;
const steerText = text.trim();
const steerImages = images;
+ const steerFiles = attachments.map((f) => ({ fileName: f.name, dataUrl: f.dataUrl }));
setBusy(true);
let res: "queued" | "not_running" | "failed" = "failed";
try {
- res = await onSteer!(steerText, steerImages);
+ res = await onSteer!(steerText, steerImages, steerFiles);
if (res === "queued") {
// Show the "queued" hint until the steering message shows up in the stream
// (steeringDeliveredCount increases) — see the effect below.
@@ -1993,6 +2020,7 @@ export function ChatInput({
setSteerPending(true);
setText("");
setImages([]);
+ setAttachments([]);
}
} finally {
setBusy(false);
@@ -2275,12 +2303,24 @@ export function ChatInput({
)}
- {/* Mid-run steering queued: lightweight hint until the steering message appears in the
- stream (or the run ends). */}
- {steerPending && (
-
- {S.chat.steerQueuedIndicator}
-
+ {/* Mid-run steering queued: the server's undelivered-steering mirror, one line per
+ queued message with its content — task_state-fed, so it survives reloads (#136,
+ #140). The local steerPending flag only bridges the gap between the 202 and the
+ first task_state that carries the mirror. */}
+ {pendingSteering.length > 0 ? (
+
+ {pendingSteering.map((p, i) => (
+
+ {S.chat.steerQueuedItem(steeringSummary(p))}
+
+ ))}
+
+ ) : (
+ steerPending && (
+
+ {S.chat.steerQueuedIndicator}
+
+ )
)}
{/* Queued follow-ups (server-side, auto-sent once this run finishes): count from
diff --git a/packages/web/src/features/chat/chat-page.tsx b/packages/web/src/features/chat/chat-page.tsx
index 581ef98..29495e9 100644
--- a/packages/web/src/features/chat/chat-page.tsx
+++ b/packages/web/src/features/chat/chat-page.tsx
@@ -721,14 +721,27 @@ export function ChatPage() {
// Mid-run steering: the message is queued on the server and delivered between turns as a
// standalone `[user_steering]` user message followed by its images (visible once they
- // arrive over SSE / from the Trace). "not_running" (409) means no Task is in progress
- // anymore (race with completion): the input area then falls back to its **full** normal
- // send path — skills and the whole draft included — rather than a text+images task.
+ // arrive over SSE / from the Trace); file attachments land in the Session scratchpad and
+ // ride the steering text as `[attached file: ]` lines, exactly as a task's do. On
+ // "queued" the localStorage draft is discarded, like a successful send — without this a
+ // reload resurrects the already-sent text as a draft, and re-sending it duplicates the
+ // steering message (#136). "not_running" (409) means no Task is in progress anymore (race
+ // with completion): the input area then falls back to its **full** normal send path —
+ // skills and the whole draft included — rather than a text+images task.
const onSteer = useCallback(
- async (text: string, images: string[] = []): Promise<"queued" | "not_running" | "failed"> => {
+ async (
+ text: string,
+ images: string[] = [],
+ files: { fileName: string; dataUrl: string }[] = [],
+ ): Promise<"queued" | "not_running" | "failed"> => {
if (!selected) return "failed";
try {
- await api.postSteer(selected.sessionId, { text, ...(images.length > 0 ? { images } : {}) });
+ await api.postSteer(selected.sessionId, {
+ text,
+ ...(images.length > 0 ? { images } : {}),
+ ...(files.length > 0 ? { files } : {}),
+ });
+ discardSessionDraft();
return "queued";
} catch (e) {
if (e instanceof ApiError && e.status === 409) return "not_running";
@@ -736,7 +749,7 @@ export function ChatPage() {
return "failed";
}
},
- [selected],
+ [selected, discardSessionDraft],
);
const onApprove = useCallback(
@@ -881,6 +894,7 @@ export function ChatPage() {
// Count of steering messages already visible in the stream: the input area keeps its
// "queued" indicator up until this count increases (i.e. the steering message arrived).
steeringDeliveredCount={stream.model.items.filter((i) => i.kind === "user_steering").length}
+ pendingSteering={stream.pendingSteering}
onQueueFollowUp={onQueueFollowUp}
queuedFollowUps={stream.queuedFollowUps}
onStop={onStop}
diff --git a/packages/web/src/features/chat/composer-send.ts b/packages/web/src/features/chat/composer-send.ts
index 135e062..f88ed10 100644
--- a/packages/web/src/features/chat/composer-send.ts
+++ b/packages/web/src/features/chat/composer-send.ts
@@ -16,7 +16,7 @@
export type MidRunAction =
/** Abort the running Task — the button's face whenever this draft has no send channel. */
| "stop"
- /** Deliver text and images to the running agent as a `[user_steering]` message. */
+ /** Deliver text, images and file attachments to the running agent as a `[user_steering]` message. */
| "steer"
/** Hold the whole draft server-side and auto-send it as the next ordinary message. */
| "queue"
@@ -46,6 +46,8 @@ export interface MidRunComposerState {
hasText: boolean;
/** At least one image is attached. */
hasImages: boolean;
+ /** At least one file attachment is staged (steers as `[attached file: ]` lines, like a task's). */
+ hasFiles: boolean;
/**
* The draft carries anything at all — text, images, file attachments, a staged switch chip
* or selected skills. The queue takes a whole message, so this is its content rule.
@@ -58,8 +60,8 @@ export interface MidRunComposerState {
* the button is always Send there, gated by the ordinary `canSend`.
*
* Steering is preferred over the queue when both could carry the draft, since it reaches the
- * turn already under way; the queue picks up what steering cannot — a skills-only draft, file
- * attachments, a staged chip — so that "unsendable here" never costs the user Stop.
+ * turn already under way; the queue picks up what steering cannot — a skills-only draft, a
+ * staged chip — so that "unsendable here" never costs the user Stop.
*/
export function midRunAction(s: MidRunComposerState): MidRunAction {
if (s.sending) return "disabled";
@@ -71,7 +73,7 @@ export function midRunAction(s: MidRunComposerState): MidRunAction {
s.canSteerChannel &&
!s.hasHandoffTarget &&
!s.hasPendingModel &&
- (s.hasText || s.hasImages);
+ (s.hasText || s.hasImages || s.hasFiles);
if (canSteer && !s.followUpMode) return "steer";
const canQueue = open && s.canQueueChannel && s.stagedRoute !== "blocked" && s.hasContent;
if (canQueue) return "queue";
diff --git a/packages/web/src/features/chat/use-session-draft.ts b/packages/web/src/features/chat/use-session-draft.ts
index 2aa2d44..f3a7ca6 100644
--- a/packages/web/src/features/chat/use-session-draft.ts
+++ b/packages/web/src/features/chat/use-session-draft.ts
@@ -132,9 +132,11 @@ export function useSessionDraft(sessionId: string | null): {
const discard = useCallback(() => {
cancelPending();
- // Also clear the selected skills and both staged switch chips: ChatInput's clear after a
- // successful send doesn't fire a callback (same convention as onTextChange); without this,
- // a later text flush would resurrect the already-sent selection.
+ // Also clear the text, the selected skills and both staged switch chips: ChatInput's
+ // clear after a successful send doesn't fire a callback (same convention as
+ // onTextChange); without this, a later flush triggered by any discrete action (a skill
+ // toggle, a staged chip) would resurrect the already-sent text or selection.
+ textRef.current = "";
skillsRef.current = [];
handoffRef.current = null;
switchModelRef.current = null;
diff --git a/packages/web/src/features/chat/use-session-stream.ts b/packages/web/src/features/chat/use-session-stream.ts
index 5d26957..7ba5932 100644
--- a/packages/web/src/features/chat/use-session-stream.ts
+++ b/packages/web/src/features/chat/use-session-stream.ts
@@ -16,7 +16,11 @@
* server re-sends still-pending requests.
*/
import { useCallback, useEffect, useRef, useState } from "react";
-import type { GoalServerEvent, SessionStatus } from "@prismshadow/penguin-server/api";
+import type {
+ GoalServerEvent,
+ PendingSteeringInfo,
+ SessionStatus,
+} from "@prismshadow/penguin-server/api";
import { getGoal, getMe, getMessages } from "../../api/endpoints";
import { openSessionStream } from "../../api/sse";
import { createStreamController } from "../../lib/omni/stream-controller";
@@ -39,6 +43,8 @@ export interface SessionStreamState {
taskState: SessionStatus;
/** Queued follow-up count from the stream's task_state events (auto-sent once the session is idle). */
queuedFollowUps: number;
+ /** Steering messages queued on the server but not yet delivered (from task_state events): keeps the composer's hint and its content across reloads. */
+ pendingSteering: PendingSteeringInfo[];
/**
* Timestamp (ms) of the last main-session auth failure (request_end with status "auth"),
* or null. Derived from the model, so it survives history replay and resets when
@@ -86,6 +92,7 @@ export function useSessionStream(
const [loading, setLoading] = useState(true);
const [taskState, setTaskState] = useState(initialStatus);
const [queuedFollowUps, setQueuedFollowUps] = useState(0);
+ const [pendingSteering, setPendingSteering] = useState([]);
const [error, setError] = useState(null);
const [pendingTick, setPendingTick] = useState(0);
const [goal, setGoal] = useState(null);
@@ -163,6 +170,7 @@ export function useSessionStream(
controllerRef.current = null;
setTaskState("idle");
setQueuedFollowUps(0);
+ setPendingSteering([]);
setLoading(false);
setError(null);
setGoal(null);
@@ -178,6 +186,7 @@ export function useSessionStream(
setTaskState(initialStatus);
setGoal(null);
setQueuedFollowUps(0);
+ setPendingSteering([]);
setPendingTick((t) => t + 1);
// Restore an in-flight goal's banner (only when still active — a long-finished goal
@@ -203,6 +212,7 @@ export function useSessionStream(
loadMessages: () => getMessages(sessionId),
onTaskState: setTaskState,
onQueuedFollowUps: setQueuedFollowUps,
+ onPendingSteering: setPendingSteering,
onLoading: setLoading,
onError: setError,
onModelChange: bump,
@@ -276,6 +286,7 @@ export function useSessionStream(
loading,
taskState,
queuedFollowUps,
+ pendingSteering,
lastAuthFailureMs: (controllerRef.current?.model ?? placeholderRef.current).lastAuthFailureMs,
dismissModelAuthDead,
pendingApprovals: controllerRef.current?.pendingApprovals ?? EMPTY_PENDING,
diff --git a/packages/web/src/lib/omni/stream-controller.ts b/packages/web/src/lib/omni/stream-controller.ts
index dd6963e..f725e9f 100644
--- a/packages/web/src/lib/omni/stream-controller.ts
+++ b/packages/web/src/lib/omni/stream-controller.ts
@@ -34,6 +34,7 @@ import type { OmniMessage, ToolCallPayload } from "@prismshadow/penguin-core/omn
import type {
GoalServerEvent,
MessagesLiveTail,
+ PendingSteeringInfo,
ServerEvent,
SessionStatus,
} from "@prismshadow/penguin-server/api";
@@ -88,6 +89,8 @@ export interface StreamControllerDeps {
onTaskState: (state: SessionStatus) => void;
/** Queued follow-up count carried on task_state events (absent on old servers -> 0). */
onQueuedFollowUps?: (count: number) => void;
+ /** Undelivered steering messages carried on task_state events (absent = none): keeps the composer's "steering queued" hint alive across reloads. */
+ onPendingSteering?: (items: PendingSteeringInfo[]) => void;
onLoading: (loading: boolean) => void;
/** History load failure message (null = clear). */
onError: (message: string | null) => void;
@@ -190,6 +193,7 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro
streamStatus = ev.state;
deps.onTaskState(ev.state);
deps.onQueuedFollowUps?.(ev.queued ?? 0);
+ deps.onPendingSteering?.(ev.pendingSteering ?? []);
if (ev.state === "idle") {
// Task ended (or the snapshot confirms idle): finalize the current Task's stats; pending approvals have already converged server-side.
notifyTaskIdle(model);
@@ -379,6 +383,7 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro
streamStatus = ev.state;
deps.onTaskState(ev.state);
deps.onQueuedFollowUps?.(ev.queued ?? 0);
+ deps.onPendingSteering?.(ev.pendingSteering ?? []);
}
buffer.push({ kind: "server", ev, id: eventId });
return;
diff --git a/packages/web/src/lib/strings-en.ts b/packages/web/src/lib/strings-en.ts
index c3e2b6d..4ba4da2 100644
--- a/packages/web/src/lib/strings-en.ts
+++ b/packages/web/src/lib/strings-en.ts
@@ -708,6 +708,9 @@ Scenarios:
steerSend: "Send to the running agent",
/** Queued hint shown after a successful steer, until the steering message appears in the stream. */
steerQueuedIndicator: "Steering queued — delivered with the next turn",
+ /** Same hint, with the queued message's content (from the server's undelivered-steering mirror; survives reloads). */
+ steerQueuedItem: (content: string) =>
+ `Steering queued — delivered with the next turn: ${content}`,
/** Label of the [user_steering] chip (a mid-run user message delivered between turns). */
userSteering: "User steering",
/** Mid-run send-mode setting: steer (delivered mid-run) vs follow-up (queued until the run ends). */
@@ -797,6 +800,7 @@ Scenarios:
openWorkspace: "Open workspace",
openAgents: "Agents panel",
filesInMessage: (n: number) => `${n} ${n === 1 ? "file" : "files"}`,
+ imagesInMessage: (n: number) => `${n} ${n === 1 ? "image" : "images"}`,
openPreview: "Click to preview",
showMoreFiles: (n: number) => `Show ${n} more ${n === 1 ? "file" : "files"}`,
showLess: "Show less",
diff --git a/packages/web/src/lib/strings.ts b/packages/web/src/lib/strings.ts
index 2f52156..d06ee37 100644
--- a/packages/web/src/lib/strings.ts
+++ b/packages/web/src/lib/strings.ts
@@ -691,6 +691,8 @@ Benchmark:
steerSend: "发送给运行中的 Agent",
/** Queued hint shown after a successful steer, until the steering message appears in the stream. */
steerQueuedIndicator: "插话已排队,将随下一轮送达",
+ /** Same hint, with the queued message's content (from the server's undelivered-steering mirror; survives reloads). */
+ steerQueuedItem: (content: string) => `插话已排队,将随下一轮送达:${content}`,
/** Label of the [user_steering] chip (a mid-run user message delivered between turns). */
userSteering: "用户插话",
/** Mid-run send-mode setting: steer (delivered mid-run) vs follow-up (queued until the run ends). */
@@ -778,6 +780,7 @@ Benchmark:
openAgents: "智能体面板",
/** File summary card at the end of a message (Codex-style): title, inline preview action, and collapsed row. */
filesInMessage: (n: number) => `${n} 个文件`,
+ imagesInMessage: (n: number) => `${n} 张图片`,
openPreview: "点击预览",
showMoreFiles: (n: number) => `显示其余 ${n} 个文件`,
showLess: "收起",
diff --git a/packages/web/test/composer-send.test.ts b/packages/web/test/composer-send.test.ts
index 6af4325..28cd21d 100644
--- a/packages/web/test/composer-send.test.ts
+++ b/packages/web/test/composer-send.test.ts
@@ -24,23 +24,27 @@ const BASE: MidRunComposerState = {
hasPendingModel: false,
hasText: true,
hasImages: false,
+ hasFiles: false,
hasContent: true,
};
const act = (over: Partial = {}) => midRunAction({ ...BASE, ...over });
/** The draft shapes that carry nothing at all. */
-const EMPTY = { hasText: false, hasImages: false, hasContent: false } as const;
+const EMPTY = { hasText: false, hasImages: false, hasFiles: false, hasContent: false } as const;
describe("midRunAction — steering, the preferred channel", () => {
- it("takes text, images, or an image with no caption at all", () => {
+ it("takes text, images, files, or any of them with no caption at all", () => {
expect(act()).toBe("steer");
expect(act({ hasText: false, hasImages: true })).toBe("steer");
expect(act({ hasText: true, hasImages: true })).toBe("steer");
+ // A file-only draft steers exactly like an image-only one (#140) — no silent fallback
+ // to the queue, which used to answer it with the other channel's hint.
+ expect(act({ hasText: false, hasFiles: true })).toBe("steer");
});
it("is skipped for a draft it cannot carry, which the queue then takes", () => {
- // Skills or file attachments only: hasContent without text or images of its own.
+ // Skills only: hasContent without text, images or files of its own.
expect(act({ ...EMPTY, hasContent: true })).toBe("queue");
// A staged switch chip: the text belongs to the conversation that switch is about to open.
expect(act({ hasHandoffTarget: true, stagedRoute: "handoff" })).toBe("queue");
diff --git a/packages/web/test/stream-controller.test.ts b/packages/web/test/stream-controller.test.ts
index 7ac0805..ab8933c 100644
--- a/packages/web/test/stream-controller.test.ts
+++ b/packages/web/test/stream-controller.test.ts
@@ -17,7 +17,12 @@ import {
withOrigin,
} from "@prismshadow/penguin-core/omnimessage";
import type { OmniMessage, TokenCounts } from "@prismshadow/penguin-core/omnimessage";
-import type { MessagesLiveTail, ServerEvent, SessionStatus } from "@prismshadow/penguin-server/api";
+import type {
+ MessagesLiveTail,
+ PendingSteeringInfo,
+ ServerEvent,
+ SessionStatus,
+} from "@prismshadow/penguin-server/api";
import { createStreamController } from "../src/lib/omni/stream-controller";
import type { StreamController } from "../src/lib/omni/stream-controller";
import { approvalKey, findToolCard } from "../src/lib/omni/stream-model";
@@ -38,6 +43,7 @@ const flush = () => new Promise((resolve) => setTimeout(resolve, 0));
interface Harness {
controller: StreamController;
states: SessionStatus[];
+ pendingSteering: PendingSteeringInfo[][];
errors: Array;
loadings: boolean[];
loadCalls: () => number;
@@ -59,6 +65,7 @@ function createHarness(): Harness {
reject: (e: unknown) => void;
}> = [];
const states: SessionStatus[] = [];
+ const pendingSteering: PendingSteeringInfo[][] = [];
const errors: Array = [];
const loadings: boolean[] = [];
let calls = 0;
@@ -73,6 +80,7 @@ function createHarness(): Harness {
pendingLoads.push({ resolve, reject });
}),
onTaskState: (s) => states.push(s),
+ onPendingSteering: (items) => pendingSteering.push(items),
onLoading: (l) => loadings.push(l),
onError: (e) => errors.push(e),
onModelChange: () => {},
@@ -82,6 +90,7 @@ function createHarness(): Harness {
return {
controller,
states,
+ pendingSteering,
errors,
loadings,
loadCalls: () => calls,
@@ -173,6 +182,19 @@ describe("in-stream task_state is the authoritative running state (history-closi
expect(h.states).toEqual(["running"]);
});
+ it("reports the pending-steering mirror from task_state, and its absence as empty", async () => {
+ const h = createHarness();
+ void h.controller.load();
+ h.controller.handleServer({
+ type: "task_state",
+ state: "running",
+ pendingSteering: [{ text: "hold on", images: 0, files: 1 }],
+ });
+ // A later event without the field means "none left" — reported as empty, not skipped.
+ h.controller.handleServer({ type: "task_state", state: "running" });
+ expect(h.pendingSteering).toEqual([[{ text: "hold on", images: 0, files: 1 }], []]);
+ });
+
it("closes the current Task before an auto-started queued follow-up begins", async () => {
const h = createHarness();
const p = h.controller.load();