From 44dc64de74f40714002bb105b9df3689bd49284d Mon Sep 17 00:00:00 2001 From: RECTOR Date: Sun, 30 Aug 2026 15:16:08 +0700 Subject: [PATCH 1/2] fix: typed steer/abort rejection codes + status-list truncation marker (#84) SessionRejectionError (src/engine/session-rejection.ts) carries a reason discriminant (steer-unsupported | already-processing | already-aborted). The two deterministic throw sites (toLiveHandle, wrapPiSession) now throw it with UNCHANGED message text; RpcServer.steerVerb/abortVerb match on the type first and keep the message-substring checks as a back-compat fallback for third-party ChildSession implementations. Frozen error enum untouched. statusVerb adds "truncated": N to the reply ONLY when the registry exceeds LIST_CAP (additive field; absent when everything fits). Single- runId lookups never carry it. --- src/engine/session-rejection.ts | 23 ++++++++++++ src/engine/spawnSubagent.ts | 3 +- src/index.ts | 4 ++- src/rpc/rpc-server.ts | 14 ++++++-- test/live-handle.test.mts | 10 ++++-- test/rpc-server.test.mts | 64 +++++++++++++++++++++++++++++++++ 6 files changed, 112 insertions(+), 6 deletions(-) create mode 100644 src/engine/session-rejection.ts diff --git a/src/engine/session-rejection.ts b/src/engine/session-rejection.ts new file mode 100644 index 0000000..776650c --- /dev/null +++ b/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; +} diff --git a/src/engine/spawnSubagent.ts b/src/engine/spawnSubagent.ts index 0dc06de..69209eb 100644 --- a/src/engine/spawnSubagent.ts +++ b/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; }, diff --git a/src/index.ts b/src/index.ts index c091d48..d43eeac 100644 --- a/src/index.ts +++ b/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; }, }; } diff --git a/src/rpc/rpc-server.ts b/src/rpc/rpc-server.ts index da0500d..8c79c6b 100644 --- a/src/rpc/rpc-server.ts +++ b/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}`); diff --git a/test/live-handle.test.mts b/test/live-handle.test.mts index d579bb2..8b79524 100644 --- a/test/live-handle.test.mts +++ b/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 () => { diff --git a/test/rpc-server.test.mts b/test/rpc-server.test.mts index 8a112ad..c083b33 100644 --- a/test/rpc-server.test.mts +++ b/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[0]> = {}) { const dir = mkdtempSync(join(tmpdir(), "fleet-rpc-")); @@ -312,3 +313,66 @@ 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 }; + 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 }); } +}); From 42957b7246ba6169b72e2eedc55ae29a21c86721 Mon Sep 17 00:00:00 2001 From: RECTOR Date: Sun, 30 Aug 2026 15:19:40 +0700 Subject: [PATCH 2/2] test: pin the exactly-at-LIST_CAP truncation boundary (review NIT 5) --- test/rpc-server.test.mts | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/test/rpc-server.test.mts b/test/rpc-server.test.mts index c083b33..66cbbc7 100644 --- a/test/rpc-server.test.mts +++ b/test/rpc-server.test.mts @@ -376,3 +376,16 @@ test("#84: status list past LIST_CAP carries a truncated count; at/below cap doe 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 }); } +});