feat(remote-dsh): queue next-run prompts and tolerate empty reasoning blocks

.changeset/queued-input-and-empty-reasoning.md

@@ -0,0 +1,7 @@
+---
+"@orbisapp/remote-dsh": minor
+---
+
+Add `next_run` prompt delivery and queued-input withdrawal. A prompt sent while the session is busy can be held by the plugin and started as a new run once the active run completes, and a held input can be withdrawn before it starts. The driver advertises both through the `prompt.next_run` and `prompt.queue.withdraw` capabilities.
+
+Fix sessions failing to open or sync when an assistant message contains an empty text or reasoning block. DeepSeek models at low reasoning effort return an empty reasoning block that carries only a replay signature; such blocks are now skipped instead of rejecting the whole session.

packages/orbis-agent-backend/src/backend.ts

@@ -177,12 +177,20 @@ export interface AgentSessionUpdateResult {
 }
 
 /**
- * A queued input is admitted to the current run rather than starting another
- * run. Drivers report which queue semantics they implement via capabilities.
+ * How a queued input reaches the agent. `follow_up` and `steer` are admitted to
+ * the current run; `next_run` is held by the driver and replayed as a new run.
+ * Drivers report which queue semantics they implement via capabilities.
  */
 export type AgentPromptDelivery =
   /** Deliver at the driver's next follow-up/continuation point. */
   | "follow_up"
+  /**
+   * Hold the input in the driver and start a new run with it once the active
+   * run completes. Unlike the other deliveries this one never conflicts with an
+   * active run: it means "send this, and hold it if the session is busy", so a
+   * client never has to guess the run state before submitting.
+   */
+  | "next_run"
   /** Deliver at the driver's next steering point within the active run. */
   | "steer";
 
@@ -194,13 +202,26 @@ export interface AgentPromptInput {
 
 export interface AgentPromptReceipt {
   readonly acceptedAt: AgentTimestamp;
+  /**
+   * The run this input belongs to: the run it started, or — for a queued
+   * delivery — the run it was admitted to or is waiting behind. An input held
+   * for `next_run` is identified by its `pendingInputs` entry, not by this.
+   */
   readonly runId: AgentRunId;
 }
 
 export interface AgentCancelInput {
   readonly keepInbox?: boolean;
 }
 
+export interface AgentQueuedInputWithdrawInput {
+  readonly id: string;
+}
+
+export interface AgentQueuedInputWithdrawResult {
+  readonly withdrawn: boolean;
+}
+
 export interface AgentCancelResult {
   readonly cancelled: boolean;
 }
@@ -209,12 +230,27 @@ export interface AgentSessionRuntime {
   readonly ref: AgentSessionRef;
 
   cancel(input?: AgentCancelInput): Promise<AgentCancelResult>;
+  /** Detach this connection lease. Does not cancel the backend-owned execution. */
   close(): Promise<void>;
+  /** Availability of this lease. error means a failed connection/runtime, not a failed run.
+   * ready/running are convenience summaries; revisioned events own execution state. */
   getStatus(): AgentRuntimeStatus;
   observeStatus(listener: AgentRuntimeStatusListener): () => void;
   prompt(input: AgentPromptInput): Promise<AgentPromptReceipt>;
   respondPermission(input: AgentPermissionResponseInput): Promise<AgentPermissionResponseResult>;
+  /**
+   * Removes one input the driver is holding. Only driver-held entries can be
+   * withdrawn — an agent-owned steering or follow-up queue has no removal API —
+   * so anything else reports `withdrawn: false` rather than failing.
+   */
+  withdrawQueuedInput(
+    input: AgentQueuedInputWithdrawInput,
+  ): Promise<AgentQueuedInputWithdrawResult>;
   respondQuestion(input: AgentQuestionResponseInput): Promise<AgentQuestionResponseResult>;
+  /**
+   * Establish revisioned execution state on subscription (state seed or snapshot),
+   * then deliver live events. Until that seed arrives execution state is unknown.
+   */
   subscribe(listener: AgentSessionEventListener): () => void;
 }
 

packages/orbis-agent-backend/src/capabilities.ts

@@ -7,6 +7,8 @@ export const AGENT_DRIVER_CAPABILITIES = [
   "permission.respond",
   "plan.select",
   "prompt.follow_up",
+  "prompt.next_run",
+  "prompt.queue.withdraw",
   "prompt.references.files",
   "prompt.references.sessions",
   "prompt.steer",

packages/orbis-agent-backend/src/contract.test.ts

@@ -105,19 +105,22 @@ test("validates subagent child/diagnostic rows and preserves pre-order", () => {
   const root = subagentRef("root");
   const child = subagentRef("child");
   const diagnostic = subagentRef("broken");
-  const entries = validateAgentSessionSubagentList([
-    {
-      activity: "running",
-      depth: 1,
-      hasChildren: true,
-      kind: "child",
-      mode: "continuable",
-      parentRef: root,
-      ref: child,
-      label: "Build helper",
-    },
-    { depth: 2, kind: "diagnostic", parentRef: child, reason: "corrupt", ref: diagnostic },
-  ], root);
+  const entries = validateAgentSessionSubagentList(
+    [
+      {
+        activity: "running",
+        depth: 1,
+        hasChildren: true,
+        kind: "child",
+        mode: "continuable",
+        parentRef: root,
+        ref: child,
+        label: "Build helper",
+      },
+      { depth: 2, kind: "diagnostic", parentRef: child, reason: "corrupt", ref: diagnostic },
+    ],
+    root,
+  );
 
   expect(entries.map((entry) => entry.ref.sessionId)).toEqual(["child", "broken"]);
   expect(Object.isFrozen(entries)).toBe(true);
@@ -149,22 +152,25 @@ test("rejects malformed subagent identities, duplicate rows, and invalid enums",
     }),
   ).toThrow(/depth/i);
   expect(() =>
-    validateAgentSessionSubagentList([
-      {
-        depth: 1,
-        kind: "diagnostic",
-        parentRef: root,
-        reason: "corrupt",
-        ref: child,
-      },
-      {
-        depth: 1,
-        kind: "diagnostic",
-        parentRef: root,
-        reason: "corrupt",
-        ref: child,
-      },
-    ], root),
+    validateAgentSessionSubagentList(
+      [
+        {
+          depth: 1,
+          kind: "diagnostic",
+          parentRef: root,
+          reason: "corrupt",
+          ref: child,
+        },
+        {
+          depth: 1,
+          kind: "diagnostic",
+          parentRef: root,
+          reason: "corrupt",
+          ref: child,
+        },
+      ],
+      root,
+    ),
   ).toThrow(/duplicate/i);
   expect(() =>
     validateAgentSessionSubagentEntry({
@@ -246,9 +252,9 @@ test("root-aware subagent validation rejects foreign, orphaned, and misnested ro
       root,
     ),
   ).toHaveLength(2);
-  expect(() =>
-    validateAgentSessionSubagentList([{ ...childRow, ref: foreign }], root),
-  ).toThrow(/foreign/i);
+  expect(() => validateAgentSessionSubagentList([{ ...childRow, ref: foreign }], root)).toThrow(
+    /foreign/i,
+  );
   expect(() =>
     validateAgentSessionSubagentList(
       [{ ...childRow, depth: 2, parentRef: orphan, ref: grandchild }],
@@ -392,7 +398,11 @@ describe("agent backend contract", () => {
 
     expect(piRuntime.getStatus()).toBe("running");
     expect(dshRuntime.getStatus()).toBe("running");
-    expect(piEvents.map((event) => event.type)).toEqual(["session.state.changed"]);
+    expect(piEvents[0]).toMatchObject({
+      payload: { patch: { activeRun: null }, revision: 0 },
+      type: "session.state.changed",
+    });
+    expect(piEvents.slice(1).map((event) => event.type)).toEqual(["session.state.changed"]);
     expect((await backend.readSession(piRecord.ref)).entries).toMatchObject([
       { content: [{ text: "Pi completed a durable response after the switch", type: "text" }] },
     ]);

packages/orbis-agent-backend/src/events.ts

@@ -190,7 +190,7 @@ export interface AgentQueuedInput {
   readonly content: readonly AgentContentBlock[];
   readonly id: string;
   readonly kind: "follow_up" | "next_run" | "steer";
-  readonly queuedAt: AgentTimestamp;
+  readonly queuedAt?: AgentTimestamp;
 }
 
 export type AgentGoalPhase = "active" | "blocked" | "complete" | "paused";
@@ -394,12 +394,45 @@ export interface AgentEntryDeltaEvent extends AgentTransientEventBase {
     readonly blockIndex: number;
     readonly chunkSeq: number;
     readonly delta: string;
+    /** Opaque transient identity. Tool state/baseline supplies its callId association. */
     readonly entryId: AgentEntryId;
     readonly part: "text" | "thinking" | "tool_input" | "tool_output";
   };
   readonly type: "entry.delta";
 }
 
+/** An atomic replacement baseline for a new connection or transport resynchronization.
+ * Each delivery has a fresh eventId. revision orders durable state; per-entry
+ * chunkSeq establishes the next incremental chunk. Equal revisions can advance chunks
+ * or repair a gap; consumers keep a baseline watermark separate from run transitions.
+ * Empty arrays clear old stream membership, but replay cannot undo durable settlement
+ * or reopen a terminal tool identity within the same run.
+ */
+export interface AgentSessionSnapshotEvent extends AgentTransientEventBase {
+  readonly type: "session.snapshot";
+  readonly payload: {
+    readonly revision: number;
+    readonly patch: AgentSessionStatePatch;
+    readonly stream: {
+      readonly runId?: AgentRunId;
+      readonly messages: readonly {
+        readonly entryId: AgentEntryId;
+        readonly chunkSeq: number;
+        readonly blocks: readonly {
+          readonly blockIndex: number;
+          readonly blockComplete?: true;
+          readonly content: Extract<AgentContentBlock, { type: "text" | "thinking" }>;
+        }[];
+      }[];
+      readonly tools: readonly (AgentToolStateChangedEvent["payload"]["tool"] & {
+        readonly chunkSeq: number;
+        readonly inputText: string;
+        readonly outputText: string;
+      })[];
+    };
+  };
+}
+
 export interface AgentToolStateChangedEvent extends AgentTransientEventBase {
   readonly payload: {
     readonly tool: {
@@ -439,6 +472,7 @@ export type AgentDurableSessionEvent = AgentEntryAppendedEvent;
 
 export type AgentTransientSessionEvent =
   | AgentEntryDeltaEvent
+  | AgentSessionSnapshotEvent
   | AgentPresenceChangedEvent
   | AgentRunActivityEvent
   | AgentSessionStateChangedEvent

packages/orbis-agent-backend/src/index.ts

@@ -21,6 +21,8 @@ export {
   type AgentPromptDelivery,
   type AgentPromptInput,
   type AgentPromptReceipt,
+  type AgentQueuedInputWithdrawInput,
+  type AgentQueuedInputWithdrawResult,
   type AgentRuntimeStatusListener,
   type AgentRuntimeStatus,
   type AgentSessionCatalog,
@@ -98,6 +100,7 @@ export type {
   AgentSessionMetadataPatch,
   AgentSessionRunState,
   AgentSessionStateChangedEvent,
+  AgentSessionSnapshotEvent,
   AgentSessionStatePatch,
   AgentToolStateChangedEvent,
   AgentToolEntry,

packages/orbis-agent-backend/src/prompt.ts

@@ -107,7 +107,9 @@ export function validateAgentPromptInput(value: unknown): AgentPromptInput {
     content,
     ...(input.delivery === undefined
       ? {}
-      : input.delivery === "steer" || input.delivery === "follow_up"
+      : input.delivery === "steer" ||
+          input.delivery === "follow_up" ||
+          input.delivery === "next_run"
         ? { delivery: input.delivery }
         : invalid("Prompt delivery is invalid")),
     ...(input.idempotencyKey === undefined

packages/orbis-agent-backend/src/testkit.ts

@@ -16,6 +16,8 @@ import type {
   AgentWorkspaceRegisterResult,
   AgentPromptInput,
   AgentPromptReceipt,
+  AgentQueuedInputWithdrawInput,
+  AgentQueuedInputWithdrawResult,
   AgentRuntimeStatusListener,
   AgentRuntimeStatus,
   AgentSessionCreateInput,
@@ -546,6 +548,14 @@ export class FakeAgentSessionRuntime implements AgentSessionRuntime {
     return { accepted: false };
   }
 
+  /** The fake holds no inputs of its own, so there is never one to withdraw. */
+  async withdrawQueuedInput(
+    _input: AgentQueuedInputWithdrawInput,
+  ): Promise<AgentQueuedInputWithdrawResult> {
+    this.assertOpen();
+    return { withdrawn: false };
+  }
+
   async close(): Promise<void> {
     if (this.status === "closed") return;
     this.setStatus("closed");
@@ -637,7 +647,7 @@ export class FakeAgentSessionRuntime implements AgentSessionRuntime {
     const previousStatus = this.status;
     this.activeRunId = undefined;
     this.activeRunStartedAt = undefined;
-    this.setStatus(outcome === "failed" ? "error" : "ready");
+    this.setStatus("ready");
     try {
       this.host.commit(this, event);
     } catch (error) {
@@ -707,6 +717,24 @@ export class FakeAgentSessionRuntime implements AgentSessionRuntime {
   subscribe(listener: AgentSessionEventListener): () => void {
     this.assertOpen();
     this.listeners.add(listener);
+    // Seed from the producer's revisioned execution state, not from lease status.
+    listener({
+      type: "session.state.changed",
+      durability: "transient",
+      eventId: agentEventId(this.host.nextEventId()),
+      occurredAt: this.host.now(),
+      sessionId: this.ref.sessionId,
+      source: { backendId: this.ref.backendId, driverId: this.ref.driverId },
+      payload: {
+        revision: this.stateRevision,
+        patch: {
+          activeRun:
+            this.activeRunId && this.activeRunStartedAt
+              ? { id: this.activeRunId, startedAt: this.activeRunStartedAt }
+              : null,
+        },
+      },
+    });
     return () => this.listeners.delete(listener);
   }