Merge branch 'feature/mid-run-steering'

Mid-run steering: Enter steers the live agent run (Pi session.steer()),
Alt+Enter queues a new run, Stop restores undelivered steers to the composer.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: TKassis

server/src/agent/events.ts

@@ -64,6 +64,24 @@ function resultText(s: unknown): string {
   return JSON.stringify(s ?? "");
 }
 
+/** Flatten a user message's content (string or content-part array) to plain
+ *  text. Image parts are dropped — the UI renders steered messages as text. */
+function userMessageText(message: unknown): string {
+  const content = (message as { content?: unknown }).content;
+  if (typeof content === "string") return content;
+  if (Array.isArray(content)) {
+    return content
+      .map((p) =>
+        p && typeof p === "object" && (p as { type?: string }).type === "text"
+          ? String((p as { text?: unknown }).text ?? "")
+          : "",
+      )
+      .filter(Boolean)
+      .join("\n");
+  }
+  return "";
+}
+
 function cap(s: unknown, max = 4000): string {
   const str = resultText(s);
   return str.length > max ? str.slice(0, max) + "…" : str;
@@ -87,8 +105,15 @@ export function toClientFrame(
       const usage = (ev.message as { usage?: unknown }).usage;
       return { type: "turn_end", usage };
     }
-    case "message_start":
-      return { type: "message_start", role: (ev.message as { role?: string }).role };
+    case "message_start": {
+      const role = (ev.message as { role?: string }).role;
+      // User content marks the exact point a steered message was delivered
+      // into the run, so the client can split the transcript there.
+      if (role === "user") {
+        return { type: "message_start", role, content: userMessageText(ev.message) };
+      }
+      return { type: "message_start", role };
+    }
     case "message_end":
       return { type: "message_end", role: (ev.message as { role?: string }).role };
     case "message_update": {

server/src/api/sessions.ts

@@ -183,10 +183,61 @@ export async function registerSessionRoutes(app: FastifyInstance): Promise<void>
 
   app.post<{ Params: { id: string } }>("/sessions/:id/abort", async (req) => {
     const session = await getSession(currentProjectId(), activePaths(), req.params.id);
-    if (session) await session.abort();
-    return { ok: true };
+    if (!session) return { ok: true, restored: [] };
+    // Clear BEFORE abort so a pending steer can't be delivered into the
+    // dying loop; the texts go back to the composer client-side.
+    const cleared = session.clearQueue();
+    await session.abort();
+    return { ok: true, restored: [...cleared.steering, ...cleared.followUp] };
   });
 
+  // Steering side-channel: queue a message into the LIVE run (delivered by Pi
+  // after the current tool calls, before the next LLM call). Never creates a
+  // run or an SSE stream — the /run stream carries the delivery + queue_update
+  // frames. 409 reason "not_streaming" tells the client to fall back to a
+  // normal run.
+  app.post<{ Params: { id: string }; Body: { message?: string } }>(
+    "/sessions/:id/steer",
+    async (req, reply) => {
+      const projectId = currentProjectId();
+      const session = await getSession(projectId, activePaths(), req.params.id);
+      if (!session) {
+        reply.code(404);
+        return { detail: "No such session" };
+      }
+      const message = req.body?.message;
+      if (!message || !message.trim()) {
+        reply.code(400);
+        return { detail: "message is required" };
+      }
+      if (!session.isStreaming) {
+        reply.code(409);
+        return { detail: "No run in flight", reason: "not_streaming" };
+      }
+      // A steer extends a live run's spend past what the run-start check
+      // gated, so re-check the cap here.
+      const budget = isBudgetExceeded(projectId);
+      if (budget.exceeded) {
+        reply.code(403);
+        return {
+          detail:
+            `Project spend limit reached ($${budget.totalUsd.toFixed(2)} / ` +
+            `$${(budget.limitUsd ?? 0).toFixed(2)}).`,
+          reason: "budget",
+        };
+      }
+      await session.steer(message);
+      // The run can end between the guard and the queue write; a steer left
+      // behind would silently deliver into the NEXT run, so pull it back out.
+      if (!session.isStreaming) {
+        session.clearQueue();
+        reply.code(409);
+        return { detail: "Run ended before the message was delivered", reason: "not_streaming" };
+      }
+      return { ok: true, pending: [...session.getSteeringMessages()] };
+    },
+  );
+
   app.post<{ Params: { id: string }; Body: RunBody }>(
     "/sessions/:id/run",
     async (req, reply) => {

server/test/backend.test.ts

@@ -306,6 +306,38 @@ describe("events → client frames", () => {
     );
     expect(frame).toMatchObject({ type: "tool_start", args: { path: "notes.md" } });
   });
+
+  it("includes content on user message_start (string and content-array forms)", () => {
+    expect(
+      toClientFrame({
+        type: "message_start",
+        message: { role: "user", content: "exclude sample 7" },
+      } as never),
+    ).toEqual({ type: "message_start", role: "user", content: "exclude sample 7" });
+
+    expect(
+      toClientFrame({
+        type: "message_start",
+        message: {
+          role: "user",
+          content: [
+            { type: "text", text: "look at" },
+            { type: "image", data: "…", mimeType: "image/png" },
+            { type: "text", text: "plot.png" },
+          ],
+        },
+      } as never),
+    ).toEqual({ type: "message_start", role: "user", content: "look at\nplot.png" });
+  });
+
+  it("omits content on assistant message_start", () => {
+    expect(
+      toClientFrame({
+        type: "message_start",
+        message: { role: "assistant", content: "internal" },
+      } as never),
+    ).toEqual({ type: "message_start", role: "assistant" });
+  });
 });
 
 describe("web access bridge", () => {

server/test/steer-abort.test.ts

@@ -0,0 +1,167 @@
+/**
+ * HTTP-level tests for the steering side-channel and abort queue restore.
+ * The session registry is mocked so no real Pi session (auth, model) is
+ * needed; the fake exposes exactly the surface the routes touch.
+ */
+import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
+import fs from "node:fs";
+
+const fakeSessions = new Map<string, FakeSession>();
+
+class FakeSession {
+  isStreaming = true;
+  steered: string[] = [];
+  aborted = false;
+  clearQueueCalls = 0;
+  calls: string[] = [];
+  /** Called by steer(); lets a test flip isStreaming mid-call. */
+  onSteer: (() => void) | null = null;
+
+  async steer(text: string): Promise<void> {
+    this.calls.push("steer");
+    this.steered.push(text);
+    this.onSteer?.();
+  }
+  getSteeringMessages(): readonly string[] {
+    return this.steered;
+  }
+  clearQueue(): { steering: string[]; followUp: string[] } {
+    this.calls.push("clearQueue");
+    this.clearQueueCalls += 1;
+    const steering = [...this.steered];
+    this.steered = [];
+    return { steering, followUp: [] };
+  }
+  async abort(): Promise<void> {
+    this.calls.push("abort");
+    this.aborted = true;
+  }
+}
+
+vi.mock("../src/agent/session-registry.ts", () => ({
+  getAuthStorage: vi.fn(),
+  getModelRegistry: vi.fn(),
+  createSession: vi.fn(),
+  getSession: vi.fn(async (_projectId: string, _paths: unknown, id: string) =>
+    fakeSessions.get(id) ?? null,
+  ),
+  listSessions: vi.fn(async () => []),
+  disposeSession: vi.fn(),
+}));
+
+import { buildApp } from "../src/index.ts";
+import { PROJECTS_ROOT } from "../src/config.ts";
+import { createProject } from "../src/projects.ts";
+import { recordRun } from "../src/cost/ledger.ts";
+
+const app = await buildApp();
+
+beforeEach(() => {
+  fakeSessions.clear();
+  fs.rmSync(PROJECTS_ROOT, { recursive: true, force: true });
+  fs.mkdirSync(PROJECTS_ROOT, { recursive: true });
+});
+
+afterAll(async () => {
+  await app.close();
+  fs.rmSync(PROJECTS_ROOT, { recursive: true, force: true });
+});
+
+function steer(id: string, body: unknown, projectId = "default") {
+  return app.inject({
+    method: "POST",
+    url: `/sessions/${id}/steer`,
+    headers: { "x-project-id": projectId, "content-type": "application/json" },
+    payload: body as Record<string, unknown>,
+  });
+}
+
+describe("POST /sessions/:id/abort", () => {
+  it("clears the queue before aborting and returns the texts", async () => {
+    const s = new FakeSession();
+    await s.steer("pending steer");
+    fakeSessions.set("s1", s);
+    const res = await app.inject({
+      method: "POST",
+      url: "/sessions/s1/abort",
+      headers: { "x-project-id": "default" },
+    });
+    expect(res.statusCode).toBe(200);
+    expect(res.json()).toEqual({ ok: true, restored: ["pending steer"] });
+    expect(s.aborted).toBe(true);
+    expect(s.clearQueueCalls).toBe(1);
+    expect(s.calls).toEqual(["steer", "clearQueue", "abort"]);
+  });
+
+  it("returns ok with empty restored for an unknown session", async () => {
+    const res = await app.inject({
+      method: "POST",
+      url: "/sessions/nope/abort",
+      headers: { "x-project-id": "default" },
+    });
+    expect(res.statusCode).toBe(200);
+    expect(res.json()).toEqual({ ok: true, restored: [] });
+  });
+});
+
+describe("POST /sessions/:id/steer", () => {
+  it("404s for an unknown session", async () => {
+    const res = await steer("nope", { message: "hi" });
+    expect(res.statusCode).toBe(404);
+  });
+
+  it("400s for an empty message", async () => {
+    fakeSessions.set("s1", new FakeSession());
+    const res = await steer("s1", { message: "   " });
+    expect(res.statusCode).toBe(400);
+  });
+
+  it("409s with reason not_streaming when no run is live", async () => {
+    const s = new FakeSession();
+    s.isStreaming = false;
+    fakeSessions.set("s1", s);
+    const res = await steer("s1", { message: "hi" });
+    expect(res.statusCode).toBe(409);
+    expect(res.json()).toMatchObject({ reason: "not_streaming" });
+    expect(s.steered).toEqual([]);
+  });
+
+  it("queues the message and returns the pending list", async () => {
+    const s = new FakeSession();
+    fakeSessions.set("s1", s);
+    const res = await steer("s1", { message: "exclude sample 7" });
+    expect(res.statusCode).toBe(200);
+    expect(res.json()).toEqual({ ok: true, pending: ["exclude sample 7"] });
+    expect(s.steered).toEqual(["exclude sample 7"]);
+  });
+
+  it("409s and clears the queue when the run ends while queueing", async () => {
+    const s = new FakeSession();
+    s.onSteer = () => {
+      s.isStreaming = false;
+    };
+    fakeSessions.set("s1", s);
+    const res = await steer("s1", { message: "too late" });
+    expect(res.statusCode).toBe(409);
+    expect(res.json()).toMatchObject({ reason: "not_streaming" });
+    // The stale steer must not leak into the next run.
+    expect(s.clearQueueCalls).toBe(1);
+    expect(s.steered).toEqual([]);
+  });
+
+  it("403s with reason budget when the project cap is reached", async () => {
+    const p = createProject({ name: "Capped", spendLimitUsd: 0.01 });
+    const zero = { costUsd: 0, input: 0, output: 0, cacheRead: 0, total: 0 };
+    recordRun({
+      sessionId: "s1",
+      projectId: p.id,
+      model: "m",
+      before: zero,
+      after: { costUsd: 0.02, input: 10, output: 10, cacheRead: 0, total: 20 },
+    });
+    fakeSessions.set("s1", new FakeSession());
+    const res = await steer("s1", { message: "hi" }, p.id);
+    expect(res.statusCode).toBe(403);
+    expect(res.json()).toMatchObject({ reason: "budget" });
+  });
+});