fix: typed steer/abort rejection codes + status-list truncation marker (#84)

src/engine/session-rejection.ts

@@ -0,0 +1,23 @@
+// src/engine/session-rejection.ts
+// #84: typed rejections for the live-session control path (steer/abort). The RPC verbs
+// previously classified failures by fragile message-substring matching; the handles now
+// throw this class and the verbs match on `reason` first. Message TEXT is unchanged at
+// the existing throw sites — string matching survives as a back-compat fallback for
+// third-party ChildSession implementations that bubble bare Errors.
+export type SessionRejectionReason =
+  | "steer-unsupported"   // the backend has no steer (e.g. claude children)
+  | "already-processing"  // the session is mid-turn and cannot take the steer now
+  | "already-aborted";    // the session was already stopped
+
+export class SessionRejectionError extends Error {
+  readonly reason: SessionRejectionReason;
+  constructor(reason: SessionRejectionReason, message: string) {
+    super(message);
+    this.name = "SessionRejectionError";
+    this.reason = reason;
+  }
+}
+
+export function isSessionRejection(e: unknown): e is SessionRejectionError {
+  return e instanceof SessionRejectionError;
+}

src/engine/spawnSubagent.ts

@@ -7,6 +7,7 @@ import type { BackendRegistry } from "../backend/port.ts";
 import { genRunId, RunRegistry } from "./run-registry.ts";
 import type { RunRecord } from "./run-registry.ts";
 import { createTurnBudget, DEFAULT_MAX_TURNS } from "./turn-budget.ts";
+import { SessionRejectionError } from "./session-rejection.ts";
 import type { ForegroundLock } from "./concurrency-lock.ts";
 import type { RunLog } from "../runtime/run-log.ts";
 import { buildToolEvent } from "../runtime/run-log.ts";
@@ -83,7 +84,7 @@ export interface LiveSessionHandle {
  *  `supportsSteer` is derived from whether the backend implemented the optional `steer`. */
 export function toLiveHandle(session: ChildSession): LiveSessionHandle {
   return {
-    steer: (text) => session.steer ? session.steer(text) : Promise.reject(new Error("steer not supported on this backend")),
+    steer: (text) => session.steer ? session.steer(text) : Promise.reject(new SessionRejectionError("steer-unsupported", "steer not supported on this backend")),
     abort: () => session.abort(),
     subscribe: (h) => session.subscribe(h),
     get isStreaming() { return session.isStreaming ?? false; },

src/index.ts

@@ -12,6 +12,7 @@ import { createSubagentTool, mergeLifecycleSkills, resolveDispatchCwd, type Suba
 // SPEC-6-3: /fleet uses openWorkflowPanelLoop (Task 12) instead of the raw openFleetPanel factory.
 import { discoverAgents } from "./registry/discovery.ts";
 import { RunRegistry } from "./engine/run-registry.ts";
+import { SessionRejectionError } from "./engine/session-rejection.ts";
 import { createSingleSlotLock, createForegroundLock } from "./engine/concurrency-lock.ts";
 import { ArmoryTodoAdapter } from "./todo-sync/adapter.ts";
 import { ArmoryMemoryAdapter } from "./memory-hydrate/adapter.ts";
@@ -83,7 +84,8 @@ function wrapPiSession(inner: ChildSession, backendSessionId: string): ChildSess
       return inner.subscribe(handler);
     },
     // SPEC-5b-4: forward the native SDK steer + isStreaming to the real pi session.
-    steer: (t) => inner.steer ? inner.steer(t) : Promise.reject(new Error("pi session has no steer")),
+    // #84: typed rejection (message text unchanged — back-compat with string matching).
+    steer: (t) => inner.steer ? inner.steer(t) : Promise.reject(new SessionRejectionError("steer-unsupported", "pi session has no steer")),
     get isStreaming() { return inner.isStreaming ?? false; },
   };
 }

src/rpc/rpc-server.ts

@@ -7,6 +7,7 @@ import type { RunRegistry, RunRecord } from "../engine/run-registry.ts";
 import type { RunLog } from "../runtime/run-log.ts";
 import type { RunJournal } from "../runtime/run-journal.ts";
 import { resolveDispatchCwd } from "../tools/subagent.ts";
+import { isSessionRejection } from "../engine/session-rejection.ts";
 
 export type RpcErrorCode =
   | "E-CONTROL-DISABLED" | "E-RUN-NOT-FOUND" | "E-RUN-FINISHED" | "E-BAD-VERB"
@@ -183,8 +184,12 @@ export class RpcServer {
       if (!rec) return this.err(id, "E-RUN-NOT-FOUND", `no live run '${p.runId}' in the registry (finished runs older than the session are not listed)`);
       return { id, ok: true, data: { runs: [summarize(rec)] } };
     }
-    const runs = this.deps.runRegistry.list().slice(0, LIST_CAP).map(summarize);
-    return { id, ok: true, data: { runs } };
+    const runs = this.deps.runRegistry.list();
+    const capped = runs.slice(0, LIST_CAP).map(summarize);
+    // #84: surface the omitted count so RPC consumers know the list is partial. Absent
+    // when everything fit (additive field — consumers check presence, not falseness).
+    const truncated = runs.length - capped.length;
+    return { id, ok: true, data: truncated > 0 ? { runs: capped, truncated } : { runs: capped } };
   }
 
   private observeVerb(id: string, params: unknown): RpcReply {
@@ -251,6 +256,9 @@ export class RpcServer {
     try {
       await session.steer(p.message);
     } catch (e) {
+      // #84: typed rejections match by reason; string matching survives as a back-compat
+      // fallback for third-party ChildSession implementations that bubble bare Errors.
+      if (isSessionRejection(e) && e.reason === "steer-unsupported") return this.err(id, "E-STEER-UNSUPPORTED", e.message);
       const msg = (e as Error).message ?? "steer failed";
       if (msg.includes("not supported")) return this.err(id, "E-STEER-UNSUPPORTED", msg);
       return this.err(id, "E-INTERNAL", `steer failed: ${msg}`);
@@ -269,6 +277,8 @@ export class RpcServer {
     try {
       await session.abort();
     } catch (e) {
+      // #84: typed first (reason-based), string fallback for bare-Error handles.
+      if (isSessionRejection(e) && (e.reason === "already-aborted" || e.reason === "already-processing")) return this.err(id, "E-RUN-FINISHED", e.message);
       const msg = (e as Error).message ?? "abort failed";
       if (msg.includes("already")) return this.err(id, "E-RUN-FINISHED", msg);
       return this.err(id, "E-INTERNAL", `abort failed: ${msg}`);

test/live-handle.test.mts

@@ -2,6 +2,7 @@
 import { test } from "node:test";
 import assert from "node:assert/strict";
 import { toLiveHandle, type ChildSession, type LiveSessionHandle } from "../src/engine/spawnSubagent.ts";
+import { SessionRejectionError } from "../src/engine/session-rejection.ts";
 
 /** A fake ChildSession that HAS steer + isStreaming (stands in for the pi backend). */
 function steerableChild(streaming = true): ChildSession {
@@ -40,11 +41,16 @@ test("toLiveHandle: steerable child → supportsSteer true, steer forwards, isSt
   assert.equal(h.isStreaming, false, "isStreaming is live (getter), not captured");
 });
 
-test("toLiveHandle: bare child → supportsSteer false, steer rejects, isStreaming defaults false", async () => {
+test("toLiveHandle: bare child → supportsSteer false, steer rejects TYPED (#84), isStreaming defaults false", async () => {
   const h = toLiveHandle(bareChild());
   assert.equal(h.supportsSteer, false);
   assert.equal(h.isStreaming, false);
-  await assert.rejects(() => h.steer("x"), /steer not supported/);
+  await assert.rejects(() => h.steer("x"), (e: unknown) => {
+    assert.ok(e instanceof SessionRejectionError, "rejection must be typed, not a bare Error");
+    assert.equal((e as SessionRejectionError).reason, "steer-unsupported");
+    assert.match((e as Error).message, /steer not supported/); // message text unchanged (back-compat)
+    return true;
+  });
 });
 
 test("toLiveHandle: abort + subscribe forward to the wrapped child", async () => {

test/rpc-server.test.mts

@@ -9,6 +9,7 @@ import { RunJournal } from "../src/runtime/run-journal.ts";
 import { RunRegistry, type RunRecord } from "../src/engine/run-registry.ts";
 import { FleetEventBus } from "../src/rpc/event-bus.ts";
 import { RpcServer, rpcControlEnabled } from "../src/rpc/rpc-server.ts";
+import { SessionRejectionError } from "../src/engine/session-rejection.ts";
 
 function harness(over: Partial<ConstructorParameters<typeof RpcServer>[0]> = {}) {
   const dir = mkdtempSync(join(tmpdir(), "fleet-rpc-"));
@@ -312,3 +313,79 @@ test("a handler exception → E-INTERNAL, never a thrown reply (one reply per re
     assert.equal((r as { error: { code: string } }).error.code, "E-INTERNAL");
   } finally { rmSync(h.dir, { recursive: true, force: true }); }
 });
+
+// #84: typed steer/abort rejections — the verb maps on the error TYPE first (the string
+// matching stays as a back-compat fallback for third-party session handles).
+test("#84: typed SessionRejectionError maps by reason, message text irrelevant", async () => {
+  const h = harness();
+  try {
+    h.registry.add(record({ runId: "fl-typed-unsup", session: {
+      steer: async () => { throw new SessionRejectionError("steer-unsupported", "completely different wording"); },
+      abort: async () => {},
+      get supportsSteer() { return true; }, // flag true but the call rejects — the race fallback
+    } as never }));
+    h.registry.add(record({ runId: "fl-typed-already", session: {
+      steer: async () => {},
+      abort: async () => { throw new SessionRejectionError("already-aborted", "also different wording"); },
+      get supportsSteer() { return true; },
+    } as never }));
+    const unsup = await h.server.handle({ id: "t1", verb: "steer", params: { runId: "fl-typed-unsup", message: "m" } });
+    assert.equal((unsup as { error: { code: string } }).error.code, "E-STEER-UNSUPPORTED", "typed reason wins over message text");
+    const already = await h.server.handle({ id: "t2", verb: "abort", params: { runId: "fl-typed-already" } });
+    assert.equal((already as { error: { code: string } }).error.code, "E-RUN-FINISHED", "typed already-aborted → E-RUN-FINISHED");
+  } finally { rmSync(h.dir, { recursive: true, force: true }); }
+});
+
+test("#84: string-matched rejections (third-party handles) still map — back-compat fallback", async () => {
+  const h = harness();
+  try {
+    h.registry.add(record({ runId: "fl-str-unsup", session: {
+      steer: async () => { throw new Error("steer not supported on this backend"); },
+      abort: async () => {},
+      get supportsSteer() { return true; },
+    } as never }));
+    h.registry.add(record({ runId: "fl-str-already", session: {
+      steer: async () => {},
+      abort: async () => { throw new Error("run already aborted"); },
+      get supportsSteer() { return true; },
+    } as never }));
+    const unsup = await h.server.handle({ id: "s1", verb: "steer", params: { runId: "fl-str-unsup", message: "m" } });
+    assert.equal((unsup as { error: { code: string } }).error.code, "E-STEER-UNSUPPORTED");
+    const already = await h.server.handle({ id: "s2", verb: "abort", params: { runId: "fl-str-already" } });
+    assert.equal((already as { error: { code: string } }).error.code, "E-RUN-FINISHED");
+  } finally { rmSync(h.dir, { recursive: true, force: true }); }
+});
+
+test("#84: status list past LIST_CAP carries a truncated count; at/below cap does not", async () => {
+  const h = harness();
+  try {
+    for (let i = 0; i < 28; i++) {
+      h.registry.add(record({ runId: `fl-cap-${i}`, task: `t${i}` }));
+    }
+    const r = await h.server.handle({ id: "c1", verb: "status", params: {} }) as { ok: true; data: { runs: unknown[]; truncated?: number } };
+    assert.equal(r.ok, true);
+    assert.equal(r.data.runs.length, 25, "LIST_CAP still caps the list");
+    assert.equal(r.data.truncated, 3, "truncated names the omitted count");
+
+    h.registry.add(record({ runId: "fl-cap-99", task: "one more" })); // 29 total → 25 + 4
+    const r2 = await h.server.handle({ id: "c2", verb: "status", params: {} }) as { ok: true; data: { truncated?: number } };
+    assert.equal(r2.data.truncated, 4);
+
+    const empty = await h.server.handle({ id: "c3", verb: "status", params: { runId: "fl-cap-0" } }) as { ok: true; data: Record<string, unknown> };
+    assert.equal(empty.ok, true);
+    assert.equal("truncated" in empty.data, false, "single-run lookup never carries truncated");
+  } finally { rmSync(h.dir, { recursive: true, force: true }); }
+});
+
+test("#84: exactly-at-LIST_CAP → truncated absent (boundary pin, review NIT 5)", async () => {
+  const h = harness();
+  try {
+    for (let i = 0; i < 25; i++) h.registry.add(record({ runId: `fl-edge-${i}` }));
+    const r = await h.server.handle({ id: "e1", verb: "status", params: {} }) as { ok: true; data: { runs: unknown[]; truncated?: number } };
+    assert.equal(r.data.runs.length, 25);
+    assert.equal("truncated" in r.data, false, "at-cap is NOT partial — no marker");
+    h.registry.add(record({ runId: "fl-edge-99" })); // 26th
+    const r2 = await h.server.handle({ id: "e2", verb: "status", params: {} }) as { ok: true; data: { truncated?: number } };
+    assert.equal(r2.data.truncated, 1, "one past the cap → truncated: 1");
+  } finally { rmSync(h.dir, { recursive: true, force: true }); }
+});