diff --git a/.changeset/queued-input-and-empty-reasoning.md b/.changeset/queued-input-and-empty-reasoning.md new file mode 100644 index 0000000..e11619f --- /dev/null +++ b/.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. diff --git a/packages/orbis-agent-backend/src/backend.ts b/packages/orbis-agent-backend/src/backend.ts index 83cd4c2..7967014 100644 --- a/packages/orbis-agent-backend/src/backend.ts +++ b/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,6 +202,11 @@ 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; } @@ -201,6 +214,14 @@ 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; + /** Detach this connection lease. Does not cancel the backend-owned execution. */ close(): Promise; + /** 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; respondPermission(input: AgentPermissionResponseInput): Promise; + /** + * 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; respondQuestion(input: AgentQuestionResponseInput): Promise; + /** + * 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; } diff --git a/packages/orbis-agent-backend/src/capabilities.ts b/packages/orbis-agent-backend/src/capabilities.ts index 9846ef7..aae99fd 100644 --- a/packages/orbis-agent-backend/src/capabilities.ts +++ b/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", diff --git a/packages/orbis-agent-backend/src/contract.test.ts b/packages/orbis-agent-backend/src/contract.test.ts index 1379e99..79995cb 100644 --- a/packages/orbis-agent-backend/src/contract.test.ts +++ b/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" }] }, ]); diff --git a/packages/orbis-agent-backend/src/events.ts b/packages/orbis-agent-backend/src/events.ts index 964a26c..ba421f8 100644 --- a/packages/orbis-agent-backend/src/events.ts +++ b/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; + }[]; + }[]; + 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 diff --git a/packages/orbis-agent-backend/src/index.ts b/packages/orbis-agent-backend/src/index.ts index 05e2b07..5728234 100644 --- a/packages/orbis-agent-backend/src/index.ts +++ b/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, diff --git a/packages/orbis-agent-backend/src/prompt.ts b/packages/orbis-agent-backend/src/prompt.ts index eb07f5f..d76068f 100644 --- a/packages/orbis-agent-backend/src/prompt.ts +++ b/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 diff --git a/packages/orbis-agent-backend/src/testkit.ts b/packages/orbis-agent-backend/src/testkit.ts index e2f931f..2dc4974 100644 --- a/packages/orbis-agent-backend/src/testkit.ts +++ b/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 { + this.assertOpen(); + return { withdrawn: false }; + } + async close(): Promise { 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); } diff --git a/packages/orbis-remote-agent-protocol/src/v2-connection.test.ts b/packages/orbis-remote-agent-protocol/src/v2-connection.test.ts index 0ddc40d..2e7d9c5 100644 --- a/packages/orbis-remote-agent-protocol/src/v2-connection.test.ts +++ b/packages/orbis-remote-agent-protocol/src/v2-connection.test.ts @@ -12,6 +12,7 @@ import { ORBIS_REMOTE_AGENT_V2_METHODS, ORBIS_REMOTE_AGENT_V2_METHOD_LIST, } from "./v2-constants"; +import { v2QueuedInputSchema } from "./v2-schemas"; const ref = createAgentSessionRef({ backendId: "remote:host-a", @@ -637,7 +638,8 @@ test("v2 connection reconstructs bounded attachment reads and rejects unstable m }); const connection = new OrbisRemoteAgentV2Connection(transport); await connection.hello({ device: { name: "Test phone" }, supportedVersions: [2] }); - await expect(connection.readAttachment(ref, "attachment-1")).resolves.toEqual({ + const controller = new AbortController(); + await expect(connection.readAttachment(ref, "attachment-1", controller.signal)).resolves.toEqual({ attachmentId: "attachment-1", bytes: 6, data: "AQIDBAUG", @@ -646,6 +648,7 @@ test("v2 connection reconstructs bounded attachment reads and rejects unstable m name: "image.png", width: 3, }); + expect(transport.requestSignals.slice(1)).toEqual([controller.signal, controller.signal]); const oversized = new FakeV2Transport(); oversized.respond(ORBIS_REMOTE_AGENT_V2_METHODS.hello, attachmentHelloResult()); @@ -931,7 +934,7 @@ test("v2 sync validates the cursor and entry identity pair", async () => { expect(transport.requests).toHaveLength(1); }); -test("v2 sync returns the host revision used for cache reconciliation", async () => { +test("v2 sync returns the host revision and native queue inputs without admission timestamps", async () => { const transport = new FakeV2Transport(); transport.respond(ORBIS_REMOTE_AGENT_V2_METHODS.hello, helloResult()); transport.respond(ORBIS_REMOTE_AGENT_V2_METHODS.sessionsSync, { @@ -961,7 +964,15 @@ test("v2 sync returns the host revision used for cache reconciliation", async () leafEntryId: null, mode: null, model: null, - pendingInputs: [], + pendingInputs: [ + { id: "42", kind: "next_run", content: [{ type: "text", text: "later" }] }, + { + id: "43", + kind: "steer", + content: [{ type: "text", text: "adjust" }], + queuedAt: "2026-08-11T00:00:00.000Z", + }, + ], pendingPermissions: [], pendingQuestions: [], ref: { @@ -983,6 +994,12 @@ test("v2 sync returns the host revision used for cache reconciliation", async () const result = await connection.sync({ mode: "once", ref }); + expect(result.state?.pendingInputs).toHaveLength(2); + expect(result.state?.pendingInputs[0]).not.toHaveProperty("queuedAt"); + expect(result.state?.pendingInputs[1]?.queuedAt).toBe("2026-08-11T00:00:00.000Z"); + for (const input of result.state!.pendingInputs) { + expect(v2QueuedInputSchema.parse(input)).toEqual(input); + } expect(result).toMatchObject({ baseline: true, hostRevision: "revision-b", diff --git a/packages/orbis-remote-agent-protocol/src/v2-connection.ts b/packages/orbis-remote-agent-protocol/src/v2-connection.ts index 0a63857..ea27720 100644 --- a/packages/orbis-remote-agent-protocol/src/v2-connection.ts +++ b/packages/orbis-remote-agent-protocol/src/v2-connection.ts @@ -66,6 +66,7 @@ import { } from "./v2-schemas"; import type { RemoteAgentV2CancelInput, + RemoteAgentV2WithdrawQueuedInputInput, RemoteAgentV2ContentBlock, RemoteAgentV2CreateInput, RemoteAgentV2Delivery, @@ -117,6 +118,9 @@ export interface RemoteAgentV2Connection { input: RemoteAgentV2WorkspaceCreateFolderInput, ): Promise; cancel(input: RemoteAgentV2CancelInput): Promise<{ readonly cancelled: boolean }>; + withdrawQueuedInput( + input: RemoteAgentV2WithdrawQueuedInputInput, + ): Promise<{ readonly withdrawn: boolean }>; close(): void; createSession(input: RemoteAgentV2CreateInput): Promise; completePromptReferences( @@ -169,9 +173,14 @@ export interface RemoteAgentV2Connection { abortAttachment( uploadId: string, ): Promise<{ readonly aborted: boolean; readonly uploadId: string }>; + /** + * Reads a complete attachment through paged RPCs. The signal is passed to + * every page request so aborting a Query cancels the active transport call. + */ readAttachment( ref: AgentSessionRef, attachmentId: string, + signal?: AbortSignal, ): Promise; respondPermission(input: { readonly ref: AgentSessionRef; @@ -449,7 +458,9 @@ function parseQueuedInput(value: unknown): RemoteAgentV2SessionState["pendingInp id: string(input, "id", "Queued input id"), kind, content: array(input, "content", "Queued input content").map(parseContent), - queuedAt: agentTimestamp(string(input, "queuedAt", "Queued input time")), + ...(input.queuedAt === undefined + ? {} + : { queuedAt: agentTimestamp(string(input, "queuedAt", "Queued input time")) }), }; } @@ -1312,6 +1323,7 @@ export class OrbisRemoteAgentV2Connection implements RemoteAgentV2Connection { async readAttachment( ref: AgentSessionRef, attachmentId: string, + signal?: AbortSignal, ): Promise { if (this.helloResult?.capabilities.attachments === false || this.helloResult === undefined) { throw new AgentBackendError("unsupported", "The remote host does not support attachments"); @@ -1355,6 +1367,8 @@ export class OrbisRemoteAgentV2Connection implements RemoteAgentV2Connection { : { height: integer(output, "height", "Attachment height", 1) }), }; }, + false, + signal, ); if (result.attachmentId !== attachmentId) protocolError("Attachment id changed during read"); const chunk = decodeBase64(result.data); @@ -1429,6 +1443,17 @@ export class OrbisRemoteAgentV2Connection implements RemoteAgentV2Connection { ); } + withdrawQueuedInput(input: RemoteAgentV2WithdrawQueuedInputInput) { + return this.request( + ORBIS_REMOTE_AGENT_V2_METHODS.sessionsWithdrawQueuedInput, + { ...input, ref: refPayload(input.ref) }, + (value) => { + const output = record(value, "Withdraw queued input result"); + return { withdrawn: boolean(output, "withdrawn", "Withdrawn flag") }; + }, + ); + } + update(input: RemoteAgentV2UpdateInput): Promise<{ readonly revision: number }> { return this.request<{ readonly revision: number }>( ORBIS_REMOTE_AGENT_V2_METHODS.sessionsUpdate, diff --git a/packages/orbis-remote-agent-protocol/src/v2-constants.ts b/packages/orbis-remote-agent-protocol/src/v2-constants.ts index 78fba0d..d445b78 100644 --- a/packages/orbis-remote-agent-protocol/src/v2-constants.ts +++ b/packages/orbis-remote-agent-protocol/src/v2-constants.ts @@ -20,6 +20,7 @@ export const ORBIS_REMOTE_AGENT_V2_METHODS = { sessionsPrompt: "orbis.sessions.prompt", sessionsCancel: "orbis.sessions.cancel", sessionsUpdate: "orbis.sessions.update", + sessionsWithdrawQueuedInput: "orbis.sessions.withdrawQueuedInput", sessionsRespondPermission: "orbis.sessions.respondPermission", sessionsRespondQuestion: "orbis.sessions.respondQuestion", attachmentsUploadBegin: "orbis.attachments.upload.begin", @@ -53,6 +54,7 @@ export const ORBIS_REMOTE_AGENT_V2_METHOD_SCOPES = { [ORBIS_REMOTE_AGENT_V2_METHODS.sessionsEntries]: [ORBIS_REMOTE_SCOPES.agentRead], [ORBIS_REMOTE_AGENT_V2_METHODS.sessionsPrompt]: [ORBIS_REMOTE_SCOPES.agentWrite], [ORBIS_REMOTE_AGENT_V2_METHODS.sessionsCancel]: [ORBIS_REMOTE_SCOPES.agentWrite], + [ORBIS_REMOTE_AGENT_V2_METHODS.sessionsWithdrawQueuedInput]: [ORBIS_REMOTE_SCOPES.agentWrite], [ORBIS_REMOTE_AGENT_V2_METHODS.sessionsUpdate]: [ORBIS_REMOTE_SCOPES.agentWrite], [ORBIS_REMOTE_AGENT_V2_METHODS.sessionsRespondPermission]: [ORBIS_REMOTE_SCOPES.agentWrite], [ORBIS_REMOTE_AGENT_V2_METHODS.sessionsRespondQuestion]: [ORBIS_REMOTE_SCOPES.agentWrite], diff --git a/packages/orbis-remote-agent-protocol/src/v2-host.test.ts b/packages/orbis-remote-agent-protocol/src/v2-host.test.ts index 68ed68c..b586336 100644 --- a/packages/orbis-remote-agent-protocol/src/v2-host.test.ts +++ b/packages/orbis-remote-agent-protocol/src/v2-host.test.ts @@ -172,6 +172,7 @@ function presenceBackend( ): RemoteAgentV2Backend { const runtime: RemoteAgentV2Runtime = { cancel: async () => ({ cancelled: false }), + withdrawQueuedInput: async () => ({ withdrawn: false }), close: async () => undefined, prompt: async (input) => { options.onPrompt?.(input.content); @@ -295,6 +296,7 @@ test("v2 host replays native entries from its cursor index without ACK state", a let updates = 0; const runtime: RemoteAgentV2Runtime = { cancel: async () => ({ cancelled: false }), + withdrawQueuedInput: async () => ({ withdrawn: false }), close: async () => undefined, prompt: async () => { prompts += 1; @@ -1245,11 +1247,7 @@ test("v2 host does not replay transient backlog to a peer joining live sync", as }, }, }); - const emit = ( - chunkSeq: number, - sessionId = nativeRef.sessionId, - blockComplete = false, - ): void => { + const emit = (chunkSeq: number, sessionId = nativeRef.sessionId, blockComplete = false): void => { const event: RemoteAgentV2SessionEvent = { blockIndex: 0, channel: "transient", @@ -2015,6 +2013,7 @@ test("v2 host validates staged uploads, consumes successful prompts, and bounds const prompted: AgentPromptContentBlock[][] = []; let failPrompt = false; let readData = "AQIDBAUG"; + let readSignal: AbortSignal | undefined; const peerB: RemoteAgentHostPeer = { deviceId: "device-b", deviceName: "Test Device B", @@ -2030,8 +2029,9 @@ test("v2 host validates staged uploads, consumes successful prompts, and bounds } prompted.push([...content]); }, - readAttachment: async (native, attachmentId) => { + readAttachment: async (native, attachmentId, signal) => { expect(native.sessionId).toBe("session-a"); + readSignal = signal; if (attachmentId !== "att-1" && attachmentId !== "att-too-large") { throw new AgentBackendError("not_found", "attachment not found"); } @@ -2068,10 +2068,13 @@ test("v2 host validates staged uploads, consumes successful prompts, and bounds send: async () => undefined, }, }); - const requestContext = (target: RemoteAgentHostPeer = peer) => ({ + const requestContext = ( + target: RemoteAgentHostPeer = peer, + signal: AbortSignal = new AbortController().signal, + ) => ({ maxResponseBytes: 1024 * 1024, peer: target, - signal: new AbortController().signal, + signal, }); const begin = (target: RemoteAgentHostPeer, uploadId: string, totalBytes: number) => host.handleRequest( @@ -2204,11 +2207,13 @@ test("v2 host validates staged uploads, consumes successful prompts, and bounds serverCode: "not_found", }); + const readController = new AbortController(); const firstRead = await host.handleRequest( ORBIS_REMOTE_AGENT_V2_METHODS.attachmentsRead, params({ attachmentId: "att-1", offset: 0, ref: publicRef }), - requestContext(), + requestContext(peer, readController.signal), ); + expect(readSignal).toBe(readController.signal); expect(firstRead).toMatchObject({ attachmentId: "att-1", bytes: 6, diff --git a/packages/orbis-remote-agent-protocol/src/v2-host.ts b/packages/orbis-remote-agent-protocol/src/v2-host.ts index 03ad62c..4ea7b1c 100644 --- a/packages/orbis-remote-agent-protocol/src/v2-host.ts +++ b/packages/orbis-remote-agent-protocol/src/v2-host.ts @@ -45,6 +45,7 @@ import { } from "./v2-constants"; import { v2CancelInputSchema, + v2WithdrawQueuedInputInputSchema, v2ContentBlockSchema, v2PromptContentBlockSchema, v2CreateInputSchema, @@ -76,6 +77,7 @@ import { import type { RemoteAgentV2Backend, RemoteAgentV2CancelInput, + RemoteAgentV2WithdrawQueuedInputInput, RemoteAgentV2ContentBlock, RemoteAgentV2DeviceDescriptor, RemoteAgentV2Entry, @@ -622,6 +624,15 @@ function parseCancelInput(value: JsonValue): RemoteAgentV2CancelInput { }; } +function parseWithdrawQueuedInputInput(value: JsonValue): RemoteAgentV2WithdrawQueuedInputInput { + const input = parseSchema(v2WithdrawQueuedInputInputSchema, value, "Withdraw queued input input"); + return { + idempotencyKey: input.idempotencyKey, + queuedInputId: input.queuedInputId, + ref: parseRef(input.ref), + }; +} + function parseUpdateInput(value: JsonValue): RemoteAgentV2UpdateInput { const input = parseSchema(v2UpdateInputSchema, value, "Session update input"); return { @@ -1085,6 +1096,8 @@ export class OrbisRemoteAgentV2Host { return await this.prompt(parsePromptInput(params), context); case ORBIS_REMOTE_AGENT_V2_METHODS.sessionsCancel: return await this.cancel(parseCancelInput(params), context); + case ORBIS_REMOTE_AGENT_V2_METHODS.sessionsWithdrawQueuedInput: + return await this.withdrawQueuedInput(parseWithdrawQueuedInputInput(params), context); case ORBIS_REMOTE_AGENT_V2_METHODS.sessionsRespondPermission: return await this.respondPermission(parsePermissionResponseInput(params), context); case ORBIS_REMOTE_AGENT_V2_METHODS.sessionsRespondQuestion: @@ -1886,6 +1899,26 @@ export class OrbisRemoteAgentV2Host { }); } + /** + * Withdrawing is idempotent by nature — a second attempt reports + * `withdrawn: false` — so it needs no idempotency claim of its own. + */ + private async withdrawQueuedInput( + input: RemoteAgentV2WithdrawQueuedInputInput, + context: RemoteAgentHostRequestContext, + ): Promise { + const owner = await this.ownerFor(input.ref); + return this.enqueue(owner, async () => { + this.assertRequestActive(context); + await this.assertDriverCapability(owner.nativeRef.driverId, "prompt.queue.withdraw"); + return toJsonResult( + await this.requiredRuntime(owner).withdrawQueuedInput({ + queuedInputId: input.queuedInputId, + }), + ); + }); + } + private async respondPermission( input: RemoteAgentV2PermissionResponseInput, context: RemoteAgentHostRequestContext, @@ -2470,6 +2503,7 @@ export class OrbisRemoteAgentV2Host { capability: | "permission.respond" | "plan.select" + | "prompt.queue.withdraw" | "prompt.references.files" | "prompt.references.sessions" | "question.respond" @@ -2489,13 +2523,15 @@ export class OrbisRemoteAgentV2Host { ? "The driver cannot respond to questions" : capability === "plan.select" ? "The driver cannot select plan mode" - : capability === "prompt.references.files" - ? "The driver cannot complete file references" - : capability === "prompt.references.sessions" - ? "The driver cannot complete session references" - : capability === "session.subagents.list" - ? "The driver cannot list session subagents" - : "The driver cannot open server workspaces", + : capability === "prompt.queue.withdraw" + ? "The driver cannot withdraw a queued input" + : capability === "prompt.references.files" + ? "The driver cannot complete file references" + : capability === "prompt.references.sessions" + ? "The driver cannot complete session references" + : capability === "session.subagents.list" + ? "The driver cannot list session subagents" + : "The driver cannot open server workspaces", ); } } diff --git a/packages/orbis-remote-agent-protocol/src/v2-schemas.ts b/packages/orbis-remote-agent-protocol/src/v2-schemas.ts index 1360c2a..489c9ce 100644 --- a/packages/orbis-remote-agent-protocol/src/v2-schemas.ts +++ b/packages/orbis-remote-agent-protocol/src/v2-schemas.ts @@ -316,7 +316,7 @@ export const v2QueuedInputSchema = z content: z.array(v2ContentBlockSchema), id: nonEmptyString, kind: z.enum(["steer", "follow_up", "next_run"]), - queuedAt: timestamp, + queuedAt: timestamp.optional(), }) .passthrough(); @@ -806,7 +806,7 @@ export const v2CreateInputSchema = z export const v2PromptInputSchema = z .object({ content: z.array(v2PromptContentBlockSchema).min(1), - delivery: z.enum(["steer", "follow_up"]).optional(), + delivery: z.enum(["steer", "follow_up", "next_run"]).optional(), expectedRevision: nonNegativeInteger.optional(), idempotencyKey: nonEmptyString, ref: v2RefSchema, @@ -898,6 +898,14 @@ export const v2CancelInputSchema = z }) .passthrough(); +export const v2WithdrawQueuedInputInputSchema = z + .object({ + idempotencyKey: nonEmptyString, + queuedInputId: nonEmptyString, + ref: v2RefSchema, + }) + .passthrough(); + export const v2UpdateInputSchema = z .object({ expectedRevision: nonNegativeInteger.optional(), diff --git a/packages/orbis-remote-agent-protocol/src/v2-types.ts b/packages/orbis-remote-agent-protocol/src/v2-types.ts index 6f9d044..04623db 100644 --- a/packages/orbis-remote-agent-protocol/src/v2-types.ts +++ b/packages/orbis-remote-agent-protocol/src/v2-types.ts @@ -76,7 +76,7 @@ export interface RemoteAgentV2QueuedInput { readonly id: string; readonly kind: "steer" | "follow_up" | "next_run"; readonly content: readonly RemoteAgentV2ContentBlock[]; - readonly queuedAt: AgentTimestamp; + readonly queuedAt?: AgentTimestamp; } export interface RemoteAgentV2PermissionRequest { @@ -290,11 +290,17 @@ export interface RemoteAgentV2WorkspaceCreateFolderInput { export interface RemoteAgentV2PromptInput { readonly ref: AgentSessionRef; readonly content: readonly RemoteAgentV2PromptContentBlock[]; - readonly delivery?: "steer" | "follow_up"; + readonly delivery?: "steer" | "follow_up" | "next_run"; readonly expectedRevision?: number; readonly idempotencyKey: string; } +export interface RemoteAgentV2WithdrawQueuedInputInput { + readonly ref: AgentSessionRef; + readonly queuedInputId: string; + readonly idempotencyKey: string; +} + export interface RemoteAgentV2PromptReceipt { readonly runId: AgentRunId; readonly acceptedAt: AgentTimestamp; @@ -334,6 +340,11 @@ export interface RemoteAgentV2Runtime { readonly idempotencyKey?: string; }, ): Promise; + withdrawQueuedInput( + input: Omit & { + readonly idempotencyKey?: string; + }, + ): Promise<{ readonly withdrawn: boolean }>; respondPermission( input: Omit & { readonly idempotencyKey?: string; @@ -402,6 +413,11 @@ export interface RemoteAgentV2Backend { */ observeCatalog?(listener: () => void): () => void; readSession(ref: AgentSessionRef): Promise; + /** + * Reads attachment bytes for one page request. Implementations must observe + * the signal so a disconnected or cancelled client does not keep the read + * resource alive. + */ readAttachment( ref: AgentSessionRef, attachmentId: string, diff --git a/packages/orbis-remote-dsh/README.md b/packages/orbis-remote-dsh/README.md index 8cac38a..37369b2 100644 --- a/packages/orbis-remote-dsh/README.md +++ b/packages/orbis-remote-dsh/README.md @@ -26,7 +26,7 @@ Plan, goal, todos, and inbox state use native DSH projections. The plugin keeps plan/todo read units registered on the host so a cold read works before a Web preset opens an Agent. Todo's native plugin runs with private tool and prompt services; preset tools and policies keep their own ownership. Orbis retains only -its enqueue timestamps and protocol revisions. Observation +its enqueue timestamps, `next_run` queue, and protocol revisions. Observation leases are released after each read, and projection-only reads never expand the transcript. diff --git a/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.test.ts b/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.test.ts index de3f297..af06a0e 100644 --- a/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.test.ts +++ b/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.test.ts @@ -1125,7 +1125,19 @@ describe("DSH local backend", () => { model: "test-model-2", provider: "test-provider", }); - expect(events).toMatchObject([ + expect(events[0]).toMatchObject({ + payload: { + patch: { + pendingInputs: [], + pendingPermissions: [], + pendingQuestions: [], + runState: "idle", + }, + revision: 0, + }, + type: "session.state.changed", + }); + expect(events.slice(1)).toMatchObject([ { payload: { patch: { model: { modelId: "test-model-2", provider: "test-provider" } }, @@ -2605,7 +2617,19 @@ describe("DSH local backend", () => { expect(firstRuntime.getStatus()).toBe("closed"); expect(testDsh.disposeCalls.get("created-session") ?? 0).toBe(0); - expect(firstEvents.map((native) => native.type)).toEqual(["session.state.changed"]); + expect(firstEvents[0]).toMatchObject({ + payload: { + patch: { + pendingInputs: [], + pendingPermissions: [], + pendingQuestions: [], + runState: "idle", + }, + revision: 0, + }, + type: "session.state.changed", + }); + expect(firstEvents.slice(1).map((native) => native.type)).toEqual(["session.state.changed"]); const reconnected = await backend.connectRuntime(record.ref); const reconnectedEvents: AgentSessionEvent[] = []; @@ -2649,12 +2673,24 @@ describe("DSH local backend", () => { ); expect(reconnected.getStatus()).toBe("ready"); - expect(reconnectedEvents.map((native) => native.type)).toEqual([ + expect(reconnectedEvents[0]).toMatchObject({ + payload: { + patch: { + pendingInputs: [], + pendingPermissions: [], + pendingQuestions: [], + runState: "running", + }, + revision: 1, + }, + type: "session.state.changed", + }); + expect(reconnectedEvents.slice(1).map((native) => native.type)).toEqual([ "entry.delta", "entry.appended", "session.state.changed", ]); - expect(reconnectedEvents[0]).toMatchObject({ payload: { blockIndex: 2 } }); + expect(reconnectedEvents[1]).toMatchObject({ payload: { blockIndex: 2 } }); expect(await backend.listSessions()).toHaveLength(1); }); @@ -2697,7 +2733,19 @@ describe("DSH local backend", () => { queuedAt: FIXED_TIME, }, ]); - expect(events).toMatchObject([ + expect(events[0]).toMatchObject({ + payload: { + patch: { + pendingInputs: [], + pendingPermissions: [], + pendingQuestions: [], + runState: "idle", + }, + revision: 0, + }, + type: "session.state.changed", + }); + expect(events.slice(1)).toMatchObject([ { payload: { patch: { pendingInputs: [{ id: "steer-1" }] } }, type: "session.state.changed" }, { payload: { patch: { pendingInputs: [{ id: "steer-1" }, { id: "follow-up-1" }] } }, diff --git a/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.ts b/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.ts index 8a8edef..2b24115 100644 --- a/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.ts +++ b/packages/orbis-remote-dsh/src/adapter/dsh-local-backend.ts @@ -50,6 +50,9 @@ import { type AgentPromptInput, type AgentPromptContentBlock, type AgentPromptReceipt, + type AgentQueuedInputWithdrawInput, + type AgentQueuedInputWithdrawResult, + type AgentRunOutcome, type AgentPromptReferenceCompletionInput, type AgentPromptReferenceCompletionResult, type AgentRuntimeStatusListener, @@ -129,6 +132,8 @@ export const DSH_LOCAL_DRIVER_ID = "dsh"; const DSH_LOCAL_CAPABILITIES = [ "prompt.follow_up", + "prompt.next_run", + "prompt.queue.withdraw", "prompt.steer", "run.cancel", "session.create", @@ -565,6 +570,18 @@ function publicUnavailable(message = "The local DSH backend is unavailable"): Ag return new AgentBackendError("unavailable", message, { retryable: true }); } +/** + * One input this driver holds for a future run. DSH's inbox admits messages to + * the *active* run and exposes no per-message removal, so an input that must + * stay withdrawable waits here instead of being handed over. It is already a + * `DshUserMessage`, so the queue entry and the eventually delivered message + * share one identity. + */ +interface DshHeldInput { + readonly message: DshUserMessage; + readonly queuedAt: AgentTimestamp; +} + function queuedInput( message: DshUserMessage, kind: AgentQueuedInput["kind"], @@ -1854,6 +1871,9 @@ export class DshLocalHarnessDriver implements AgentHarnessDriver { } } +// Delivery identity outlives one controller, just like a consumer's deduplication window. +let executionSeedSequence = 0; + /** A detachable observer façade; closing it never cancels the DSH agent. */ export class DshLocalSessionRuntime implements AgentSessionRuntime { readonly ref: AgentSessionRef; @@ -1925,9 +1945,21 @@ export class DshLocalSessionRuntime implements AgentSessionRuntime { return this.controller.respondQuestion(this, input); } + async withdrawQueuedInput( + input: AgentQueuedInputWithdrawInput, + ): Promise { + this.assertOpen(); + return this.controller.withdrawQueuedInput(this, input); + } + subscribe(listener: AgentSessionEventListener): () => void { this.assertOpen(); this.listeners.add(listener); + try { + listener(this.controller.executionStateEvent()); + } catch { + // Passive observers cannot reject an already established subscription. + } const pending = this.controller.permissionSnapshot(); if (pending.length > 0) this.controller.publishPermissionState(pending); const questions = this.controller.questionSnapshot(); @@ -2005,6 +2037,8 @@ class DshLocalSessionController { private stream: { attemptId: string; turn: number; step: number; index: number } | undefined; private readonly deltaCoalescer: DshDeltaCoalescer; private readonly pendingInputTimes = new Map(); + /** Inputs this driver holds for a future run; see {@link DshHeldInput}. */ + private heldInputs: readonly DshHeldInput[] = []; private readonly permissionBridge: DshPermissionInteraction | undefined; private readonly questionBridge: DshQuestionBridge | undefined; private readonly overlayTools = new Map< @@ -2125,6 +2159,12 @@ class DshLocalSessionController { input: AgentCancelInput, ): Promise { this.assertAttached(runtime); + // Only this driver's held inputs can be kept deliberately; the agent's own + // inbox is governed by the same flag inside DSH. + if (input.keepInbox !== true && this.heldInputs.length > 0) { + this.heldInputs = []; + this.publishPendingInputs(); + } if (this.agent.status !== "running") return { cancelled: false }; try { this.agent.cancel( @@ -2211,6 +2251,8 @@ class DshLocalSessionController { return [ ...this.agent.inbox.nextStep.map((message) => queuedInputFor(message, "steer")), ...this.agent.inbox.nextTurn.map((message) => queuedInputFor(message, "follow_up")), + // Held inputs run after everything the active run will consume. + ...this.heldInputs.map((held) => queuedInput(held.message, "next_run", held.queuedAt)), ]; } @@ -2222,7 +2264,7 @@ class DshLocalSessionController { const canonical = validateAgentPromptInput(input); const inspection = await this.host.inspect(this.ref); const running = this.agent.status === "running"; - if (input.delivery !== undefined && !running) { + if (input.delivery !== undefined && input.delivery !== "next_run" && !running) { throw new AgentBackendError( "conflict", "DSH steering and follow-up require an active run for this session", @@ -2261,6 +2303,14 @@ class DshLocalSessionController { const runId = running ? (activeRunId(inspection) ?? nextDshRunId(inspection.events)) : nextDshRunId(inspection.events); + // `next_run` never conflicts with an active run: it means "send this, and + // hold it if the session is busy". Deciding that here, against this + // controller's own status, spares every caller a run-state guess. + if (input.delivery === "next_run" && running) { + this.heldInputs = [...this.heldInputs, { message, queuedAt: this.host.now() }]; + this.publishPendingInputs(); + return { acceptedAt: this.host.now(), runId }; + } try { if (input.delivery === "steer") this.agent.steer(message); else this.agent.followup(message); @@ -2271,6 +2321,22 @@ class DshLocalSessionController { return { acceptedAt: this.host.now(), runId }; } + /** + * Held inputs are this driver's own state, so they are withdrawable. An input + * already handed to DSH's inbox is not: that inbox has no removal API. + */ + async withdrawQueuedInput( + runtime: DshLocalSessionRuntime, + input: AgentQueuedInputWithdrawInput, + ): Promise { + this.assertAttached(runtime); + const remaining = this.heldInputs.filter((held) => String(held.message.id).trim() !== input.id); + if (remaining.length === this.heldInputs.length) return { withdrawn: false }; + this.heldInputs = remaining; + this.publishPendingInputs(); + return { withdrawn: true }; + } + async respondPermission( runtime: DshLocalSessionRuntime, input: AgentPermissionResponseInput, @@ -2406,6 +2472,8 @@ class DshLocalSessionController { }, runState: finished.outcome === "failed" ? "error" : "idle", }); + // The settled run is fully published before its successor starts. + this.drainHeldInputs(finished.outcome); } } catch (error) { this.host.report( @@ -2417,6 +2485,11 @@ class DshLocalSessionController { } receiveInboxChanged(): void { + this.publishPendingInputs(); + } + + /** Publishes the whole queued-input set, inbox-owned and driver-held alike. */ + private publishPendingInputs(): void { if (this.disposed) return; const native = lastDshSessionEvent(this.agent.session) ?? @@ -2434,6 +2507,55 @@ class DshLocalSessionController { } } + /** + * Starts the oldest held input as its own run. Only a completed run drains: + * after a failure or a cancellation the inputs stay visible in + * `pendingInputs` for the user to edit, withdraw or send, because starting + * the next run automatically would compound whatever just went wrong. + */ + private drainHeldInputs(outcome: AgentRunOutcome): void { + if (outcome !== "completed" || this.agent.status === "running") return; + const [next, ...rest] = this.heldInputs; + if (next === undefined) return; + + this.heldInputs = rest; + this.publishPendingInputs(); + try { + this.agent.followup(next.message); + } catch (error) { + // The run never started, and this input was the oldest with nothing + // consumed in between, so its original slot is still the head: put it + // back rather than losing what the user wrote. + this.heldInputs = [next, ...this.heldInputs]; + this.publishPendingInputs(); + this.host.reportUpstreamError(error); + this.host.report(publicUnavailable("The DSH run could not accept the queued input")); + } + } + + /** Deliver current execution state at its existing revision; subscribing is not a mutation. */ + executionStateEvent(): AgentSessionEvent { + return { + type: "session.state.changed", + durability: "transient", + eventId: agentEventId( + `orbis:${this.ref.sessionId}:execution-seed:${++executionSeedSequence}`, + ), + occurredAt: this.host.now(), + sessionId: this.ref.sessionId, + source: { backendId: this.ref.backendId, driverId: this.ref.driverId }, + payload: { + revision: this.stateRevision, + patch: { + runState: this.agent.status === "running" ? "running" : "idle", + pendingInputs: this.pendingInputs(), + pendingPermissions: this.permissionSnapshot(), + pendingQuestions: this.questionSnapshot(), + }, + }, + }; + } + runtimeStatus(): AgentRuntimeStatus { if (this.disposed) return "closed"; return this.agent.status === "running" ? "running" : "ready"; diff --git a/packages/orbis-remote-dsh/src/adapter/dsh-projection.ts b/packages/orbis-remote-dsh/src/adapter/dsh-projection.ts index 807abe3..dc3dc6c 100644 --- a/packages/orbis-remote-dsh/src/adapter/dsh-projection.ts +++ b/packages/orbis-remote-dsh/src/adapter/dsh-projection.ts @@ -173,13 +173,20 @@ function contentBlocks(value: unknown, toolCalls: Map): Age const blocks: AgentContentBlock[] = []; for (const rawBlock of value) { const block = record(rawBlock, "message content block"); - switch (requiredString(block, "type", "message content block type")) { + const type = requiredString(block, "type", "message content block type"); + switch (type) { case "text": - blocks.push({ text: requiredString(block, "text", "text block"), type: "text" }); - break; - case "reasoning": - blocks.push({ text: requiredString(block, "text", "reasoning block"), type: "thinking" }); + case "reasoning": { + // DSH keeps empty blocks so `replayState.blocks` stays index-aligned; a + // reasoning block can carry only a provider signature. Neither has a + // canonical presentation. + const text = block.text; + if (typeof text !== "string") { + throw new AgentBackendError("protocol", `DSH ${type} block is invalid`); + } + if (text.length > 0) blocks.push({ text, type: type === "text" ? "text" : "thinking" }); break; + } case "image": blocks.push(imageReference(block)); break; diff --git a/packages/orbis-remote-dsh/src/adapter/dsh-v4-projection.test.ts b/packages/orbis-remote-dsh/src/adapter/dsh-v4-projection.test.ts index c87c42d..12adb2a 100644 --- a/packages/orbis-remote-dsh/src/adapter/dsh-v4-projection.test.ts +++ b/packages/orbis-remote-dsh/src/adapter/dsh-v4-projection.test.ts @@ -113,6 +113,60 @@ describe("DSH V4 projection contract", () => { }); }); + test("drops empty text and signature-only reasoning blocks", () => { + // Recorded from deepseek-v4-flash at low reasoning effort. + const entry = new DshSessionEntryProjector().project({ + seq: 21, + time: 10, + type: "assistant/message", + data: { + turn: 1, + step: 1, + message: { + id: "assistant-1", + role: "assistant", + content: [ + { type: "reasoning", text: "" }, + { type: "text", text: "" }, + { type: "text", text: "Hi! What can I help you with today?" }, + ], + source: { + kind: "model", + provider: "deepseek-official", + model: "deepseek-v4-flash", + replayState: { + response: { kind: "deepseek-messages", version: 1, model: "deepseek-v4-flash" }, + blocks: [ + { type: "reasoning", signature: "signature-1" }, + { type: "text" }, + { type: "text" }, + ], + }, + }, + }, + }, + }); + expect(entry).toMatchObject({ + content: [{ type: "text", text: "Hi! What can I help you with today?" }], + }); + expect(entry && "content" in entry ? entry.content : undefined).toHaveLength(1); + }); + + test("rejects a reasoning block without text", () => { + expect(() => + new DshSessionEntryProjector().project({ + seq: 1, + time: 10, + type: "assistant/message", + data: { + turn: 1, + step: 1, + message: { id: "assistant-1", role: "assistant", content: [{ type: "reasoning" }] }, + }, + }), + ).toThrow("DSH reasoning block is invalid"); + }); + test("reads a fork-closed turn without inventing a failed run", () => { expect( runFinishForDshEvent({ diff --git a/packages/orbis-remote-dsh/src/host/dsh-v2-backend.ts b/packages/orbis-remote-dsh/src/host/dsh-v2-backend.ts index cd83bc9..fa58f21 100644 --- a/packages/orbis-remote-dsh/src/host/dsh-v2-backend.ts +++ b/packages/orbis-remote-dsh/src/host/dsh-v2-backend.ts @@ -340,6 +340,15 @@ class DshV2Runtime implements RemoteAgentV2Runtime { return this.native.cancel({ keepInbox: input.keepInbox }); } + async withdrawQueuedInput(input: { + readonly queuedInputId: string; + readonly idempotencyKey?: string; + }): Promise<{ readonly withdrawn: boolean }> { + this.assertOpen(); + void input.idempotencyKey; + return this.native.withdrawQueuedInput({ id: input.queuedInputId }); + } + async close(): Promise { if (this.closed) return; this.closed = true; @@ -355,7 +364,7 @@ class DshV2Runtime implements RemoteAgentV2Runtime { async prompt(input: { readonly content: readonly AgentPromptContentBlock[]; - readonly delivery?: "steer" | "follow_up"; + readonly delivery?: "steer" | "follow_up" | "next_run"; readonly idempotencyKey?: string; }): Promise<{ readonly runId: ReturnType;