From 7f2164d24f99c94a35e284ee432d532ec24e6c48 Mon Sep 17 00:00:00 2001 From: omsd512-W Date: Sun, 30 Aug 2026 03:48:41 +0800 Subject: [PATCH 1/2] feat: expose dsh.compact as a registered Harness command Register a DeepSeek Harness command catalog (dsh.compact, argumentMode none) and route execution through the native prompt command seam: sessions.prompt with a single /compact text block, requiring the command slot in the response and publishing a temporary projection Turn lifecycle that is not persisted into ordinary history. --- .../src/deepseek-harness-adapter.ts | 93 ++++++++++++++ .../test/deepseek-harness-adapter.test.ts | 114 ++++++++++++++++++ 2 files changed, 207 insertions(+) diff --git a/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts b/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts index 1d66da17..87b94c99 100644 --- a/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts +++ b/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts @@ -15,6 +15,9 @@ import { validateHostApprovalResponse, validateHostQuestionResponse, type HarnessAdapter, + type HarnessCommandAccepted, + type HarnessCommandCapability, + type HarnessCommandInvocation, type HarnessError, type HarnessInspection, type HarnessModelRef, @@ -51,6 +54,7 @@ import { type TurnStartCommand, } from "@codexhost/harness-adapter"; import { + harnessCommandCatalogSchema, harnessIdSchema, hostInteractionIdSchema, hostItemIdSchema, @@ -143,6 +147,18 @@ interface ActiveTurn { } const deepSeekHarnessId = harnessIdSchema.parse("deepseek-harness"); +const deepSeekCommandCatalog = harnessCommandCatalogSchema.parse({ + commands: [ + { + id: "dsh.compact", + invocation: "/compact", + label: "Compact context", + description: + "Compact the current conversation context through the DeepSeek Harness command registry", + argumentMode: "none", + }, + ], +}); const DEFAULT_TOOL_OUTPUT_LIMIT = 64_000; const HISTORY_PAGE_MESSAGES = 100; const HISTORY_PAGE_LIMIT = 10_000; @@ -221,6 +237,10 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { }, history: { fork: false, forkAcrossCwd: false, rollbackLastTurn: false }, }; + readonly commands: HarnessCommandCapability = { + list: () => Promise.resolve({ ok: true, value: deepSeekCommandCatalog }), + execute: (command) => this.#executeHarnessCommand(command), + }; readonly initialState: HarnessSessionState; readonly initialUsage: HostUsage | null; @@ -547,6 +567,79 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { } } + async #executeHarnessCommand( + command: HarnessCommandInvocation, + ): Promise> { + if (command.commandId !== "dsh.compact") { + return { + ok: false, + error: { + code: "unsupported", + message: `DeepSeek Harness does not expose command '${command.commandId}'`, + retryable: false, + }, + }; + } + if (this.#active || this.#configuring || this.#reading) { + return { + ok: false, + error: { + code: "sessionBusy", + message: + "DeepSeek Harness Session cannot execute a command while another operation is active", + retryable: true, + }, + }; + } + if (command.arguments && Object.keys(command.arguments).length > 0) { + return { + ok: false, + error: { + code: "invalidRequest", + message: "DeepSeek Harness compact command does not accept arguments", + retryable: false, + }, + }; + } + + this.#configuring = true; + try { + try { + const response = unwrapRpc( + await this.#client.sessions.prompt({ + sessionId: this.#nativeRef.nativeSessionId as SessionId, + mode: "queue", + content: [{ type: "text", text: "/compact" }], + }), + "session.prompt", + ); + if (!response.command) { + return { + ok: false, + error: { + code: "nativeFailure", + message: "DeepSeek Harness did not treat the invocation as a command", + retryable: false, + }, + }; + } + // Temporary command projection Turn: lifecycle events only, never persisted + // into the ordinary conversation history and carrying no native turn identity. + this.#emit({ type: "turn.started", turnId: command.turnId }); + this.#emit({ + type: "turn.completed", + turnId: command.turnId, + outcome: { status: "succeeded" }, + }); + return { ok: true, value: { turnId: command.turnId } }; + } catch (error) { + return { ok: false, error: normalizedError(error, "nativeFailure") }; + } + } finally { + this.#configuring = false; + } + } + async #cancel(command: TurnCancelCommand): Promise> { const active = this.#active; if (!active || active.command.turnId !== command.turnId) { diff --git a/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts b/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts index 819603cc..447476bc 100644 --- a/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts +++ b/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts @@ -342,6 +342,120 @@ describe("DeepSeekHarnessAdapter local Host", () => { await adapter.close(); }); + it("exposes the command catalog and executes dsh.compact through the native prompt command seam", async () => { + const { adapter, connection } = fixture(); + const opened = await adapter.open({ + kind: "resume", + cwd: "/workspace", + nativeRef: { + harnessId: adapter.harnessId, + nativeSessionId: SESSION_ID, + formatVersion: 1, + }, + }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const commands = session.commands; + if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); + + await expect(commands.list()).resolves.toMatchObject({ + ok: true, + value: { commands: [{ id: "dsh.compact", invocation: "/compact" }] }, + }); + connection.calls.prompt.mockResolvedValueOnce( + success({ accepted: true, command: { kind: "success", text: "compacted" } }), + ); + const iterator = session.outputs[Symbol.asyncIterator](); + const executing = commands.execute({ + turnId: hostTurnIdSchema.parse("manual-compact"), + commandId: "dsh.compact", + }); + await expect(iterator.next()).resolves.toMatchObject({ + done: false, + value: { kind: "event", event: { type: "turn.started", turnId: "manual-compact" } }, + }); + await expect(iterator.next()).resolves.toMatchObject({ + done: false, + value: { + kind: "event", + event: { + type: "turn.completed", + turnId: "manual-compact", + outcome: { status: "succeeded" }, + }, + }, + }); + await expect(executing).resolves.toEqual({ + ok: true, + value: { turnId: "manual-compact" }, + }); + expect(connection.calls.prompt).toHaveBeenCalledWith({ + sessionId: SESSION_ID, + mode: "queue", + content: [{ type: "text", text: "/compact" }], + }); + await adapter.close(); + }); + + it("rejects unknown Harness commands and command arguments without touching the Host", async () => { + const { adapter, connection } = fixture(); + const opened = await adapter.open({ + kind: "resume", + cwd: "/workspace", + nativeRef: { + harnessId: adapter.harnessId, + nativeSessionId: SESSION_ID, + formatVersion: 1, + }, + }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const commands = session.commands; + if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); + + await expect( + commands.execute({ + turnId: hostTurnIdSchema.parse("manual-x"), + commandId: "dsh.unknown", + }), + ).resolves.toMatchObject({ ok: false, error: { code: "unsupported" } }); + await expect( + commands.execute({ + turnId: hostTurnIdSchema.parse("manual-compact"), + commandId: "dsh.compact", + arguments: { text: "keep details" }, + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidRequest" } }); + expect(connection.calls.prompt).not.toHaveBeenCalled(); + await adapter.close(); + }); + + it("fails closed when DeepSeek Harness does not treat the invocation as a command", async () => { + const { adapter, connection } = fixture(); + const opened = await adapter.open({ + kind: "resume", + cwd: "/workspace", + nativeRef: { + harnessId: adapter.harnessId, + nativeSessionId: SESSION_ID, + formatVersion: 1, + }, + }); + if (!opened.ok) throw new Error(opened.error.message); + const session = opened.value; + const commands = session.commands; + if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); + + await expect( + commands.execute({ + turnId: hostTurnIdSchema.parse("manual-compact"), + commandId: "dsh.compact", + }), + ).resolves.toMatchObject({ ok: false, error: { code: "nativeFailure" } }); + expect(connection.calls.prompt).toHaveBeenCalledTimes(1); + await adapter.close(); + }); + it("preserves the confirmed Model when native selection fails", async () => { const { adapter, connection } = fixture(); const session = await openCreated(adapter); From 39cd5af245fc49ff79ec678be49ed6e5396292e1 Mon Sep 17 00:00:00 2001 From: omsd512-W Date: Sun, 30 Aug 2026 05:54:01 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(deepseek-harness):=20=E6=94=B9=E7=94=A8?= =?UTF-8?q?=E5=8E=9F=E7=94=9F=E5=91=BD=E4=BB=A4=20Remote?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复 /compact 被当作普通 prompt 提交给模型的问题，并补齐异步压缩投影、取消、原生目录校验及真实 wire 测试。 --- .../src/deepseek-harness-adapter.ts | 191 +++++++++++--- .../deepseek-harness/src/host-client.ts | 190 +++++++++++++- .../test/deepseek-harness-adapter.test.ts | 238 ++++++++++++++---- .../deepseek-harness/test/host-client.test.ts | 92 +++++++ 4 files changed, 618 insertions(+), 93 deletions(-) diff --git a/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts b/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts index 87b94c99..abb9dd45 100644 --- a/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts +++ b/packages/adapters/deepseek-harness/src/deepseek-harness-adapter.ts @@ -29,6 +29,7 @@ import { type HostAgentMessageItem, type HostApprovalInteraction, type HostCommand, + type HostContextCompactionItem, type HostFileChangeItem, type HostItem, type HostItemOutcome, @@ -71,6 +72,7 @@ import { import { DeepSeekHarnessTransportError, DeepSeekHostConnection, + type DeepSeekCommandClient, type DeepSeekHostClient, type DeepSeekHostConnectionOptions, type DeepSeekHostSubscriber, @@ -146,6 +148,13 @@ interface ActiveTurn { snapshots: HostItemSnapshot[]; } +interface ActiveCommand { + command: HarnessCommandInvocation; + abort: AbortController; + cancellationRequested: boolean; + item: HostContextCompactionItem; +} + const deepSeekHarnessId = harnessIdSchema.parse("deepseek-harness"); const deepSeekCommandCatalog = harnessCommandCatalogSchema.parse({ commands: [ @@ -159,6 +168,10 @@ const deepSeekCommandCatalog = harnessCommandCatalogSchema.parse({ }, ], }); +const emptyDeepSeekCommandCatalog = harnessCommandCatalogSchema.parse({ commands: [] }); +const DSH_COMPACT_BUSY = + "Compaction is unavailable because this process has an active compaction, or the agent is not idle."; +const DSH_COMPACT_CANCELLED = "Compaction cancelled."; const DEFAULT_TOOL_OUTPUT_LIMIT = 64_000; const HISTORY_PAGE_MESSAGES = 100; const HISTORY_PAGE_LIMIT = 10_000; @@ -186,6 +199,15 @@ function unsupported(message: string): HarnessError { return { code: "unsupported", message, retryable: false }; } +function commandFailure(operation: string, error: { code: string; message: string }): HarnessError { + const code = error.code === "session-not-found" ? "unavailable" : "nativeFailure"; + return { + code, + message: `DeepSeek Harness '${operation}' failed: ${error.message}`, + retryable: code === "unavailable" || error.code === "internal", + }; +} + function unwrapRpc(response: RpcResponse, operation: string): T { if (response.result.ok) return response.result.value; const error = response.result.error; @@ -238,7 +260,7 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { history: { fork: false, forkAcrossCwd: false, rollbackLastTurn: false }, }; readonly commands: HarnessCommandCapability = { - list: () => Promise.resolve({ ok: true, value: deepSeekCommandCatalog }), + list: () => this.#listHarnessCommands(), execute: (command) => this.#executeHarnessCommand(command), }; readonly initialState: HarnessSessionState; @@ -247,11 +269,13 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { readonly outputs: AsyncIterable; readonly #channel = new HarnessOutputChannel(); readonly #client: DeepSeekHostClient; + readonly #commandClient: DeepSeekCommandClient; readonly #nativeRef: NativeSessionRef; readonly #onClosed: () => void; readonly #toolOutputLimit: number; readonly #unsubscribe: () => void; #active: ActiveTurn | null = null; + #activeCommand: ActiveCommand | null = null; #closePromise: Promise | null = null; #closed = false; #configuring = false; @@ -280,6 +304,7 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { unsubscribe(): void; }) { this.#client = input.client; + this.#commandClient = input.client.commands; this.#model = input.model; this.#onClosed = input.onClosed; this.#toolOutputLimit = input.toolOutputLimit; @@ -303,7 +328,7 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { if (this.#closed) { return { ok: false, error: invalidState("DeepSeek Harness Session is closed") }; } - if (this.#active || this.#configuring || this.#reading) { + if (this.#active || this.#activeCommand || this.#configuring || this.#reading) { return { ok: false, error: { @@ -383,7 +408,7 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { error: unsupported(`DeepSeek Harness does not support '${command.type}'`), }; } - if (this.#active || this.#configuring || this.#reading) { + if (this.#active || this.#activeCommand || this.#configuring || this.#reading) { return { ok: false, error: { @@ -483,6 +508,15 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { async #performClose(): Promise { if (this.#closed) return; + const activeCommand = this.#activeCommand; + if (activeCommand) { + activeCommand.cancellationRequested = true; + activeCommand.abort.abort(new Error("codexhost Session closed")); + this.#finishCommand(activeCommand, { + status: "cancelled", + reason: "DeepSeek Harness command was cancelled because the Session closed", + }); + } const active = this.#active; if (active) { for (const pending of active.interactions.values()) { @@ -505,7 +539,7 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { } async #selectModel(command: ModelSelectCommand): Promise> { - if (this.#active || this.#configuring || this.#reading) { + if (this.#active || this.#activeCommand || this.#configuring || this.#reading) { return { ok: false, error: { @@ -567,9 +601,28 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { } } + async #listHarnessCommands(): Promise> { + if (this.#closed) { + return { ok: false, error: invalidState("DeepSeek Harness Session is closed") }; + } + try { + const result = await this.#commandClient.list(this.#nativeRef.nativeSessionId as SessionId); + if (!result.ok) return { ok: false, error: commandFailure("commands/list", result.error) }; + const available = result.value.some( + (descriptor) => descriptor.name === "compact" && descriptor.input === undefined, + ); + return { ok: true, value: available ? deepSeekCommandCatalog : emptyDeepSeekCommandCatalog }; + } catch (error) { + return { ok: false, error: normalizedError(error, "unavailable") }; + } + } + async #executeHarnessCommand( command: HarnessCommandInvocation, ): Promise> { + if (this.#closed) { + return { ok: false, error: invalidState("DeepSeek Harness Session is closed") }; + } if (command.commandId !== "dsh.compact") { return { ok: false, @@ -580,7 +633,7 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { }, }; } - if (this.#active || this.#configuring || this.#reading) { + if (this.#active || this.#activeCommand || this.#configuring || this.#reading) { return { ok: false, error: { @@ -602,45 +655,106 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { }; } - this.#configuring = true; + const active: ActiveCommand = { + command, + abort: new AbortController(), + cancellationRequested: false, + item: { type: "contextCompaction", itemId: this.#newItemId() }, + }; + this.#activeCommand = active; + this.#emit({ type: "turn.started", turnId: command.turnId }); + this.#emit({ type: "item.started", turnId: command.turnId, item: active.item }); + void this.#runHarnessCommand(active); + return { ok: true, value: { turnId: command.turnId } }; + } + + async #runHarnessCommand(active: ActiveCommand): Promise { try { - try { - const response = unwrapRpc( - await this.#client.sessions.prompt({ - sessionId: this.#nativeRef.nativeSessionId as SessionId, - mode: "queue", - content: [{ type: "text", text: "/compact" }], - }), - "session.prompt", - ); - if (!response.command) { - return { - ok: false, - error: { - code: "nativeFailure", - message: "DeepSeek Harness did not treat the invocation as a command", - retryable: false, - }, - }; + const response = await this.#commandClient.execute( + this.#nativeRef.nativeSessionId as SessionId, + "/compact", + active.abort.signal, + ); + if (this.#activeCommand !== active) return; + if (!response.ok) { + if (active.cancellationRequested || response.error.code === "cancelled") { + this.#finishCommand(active, { + status: "cancelled", + reason: "DeepSeek Harness context compaction was cancelled", + }); + } else { + this.#finishCommand(active, { + status: "failed", + error: commandFailure("commands/execute", response.error), + }); } - // Temporary command projection Turn: lifecycle events only, never persisted - // into the ordinary conversation history and carrying no native turn identity. - this.#emit({ type: "turn.started", turnId: command.turnId }); - this.#emit({ - type: "turn.completed", - turnId: command.turnId, - outcome: { status: "succeeded" }, + return; + } + const execution = response.value; + if (!execution) { + this.#finishCommand(active, { + status: "failed", + error: { + code: "nativeFailure", + message: "DeepSeek Harness did not resolve the registered /compact command", + retryable: false, + }, + }); + return; + } + if (execution.result.kind === "success") { + this.#finishCommand(active, { status: "succeeded" }); + } else if (active.cancellationRequested || execution.result.text === DSH_COMPACT_CANCELLED) { + this.#finishCommand(active, { + status: "cancelled", + reason: execution.result.text, + }); + } else { + this.#finishCommand(active, { + status: "failed", + error: { + code: execution.result.text === DSH_COMPACT_BUSY ? "sessionBusy" : "nativeFailure", + message: execution.result.text, + retryable: true, + }, + }); + } + } catch (error) { + if (this.#activeCommand !== active) return; + if (active.cancellationRequested || active.abort.signal.aborted) { + this.#finishCommand(active, { + status: "cancelled", + reason: "DeepSeek Harness context compaction was cancelled", + }); + } else { + this.#finishCommand(active, { + status: "failed", + error: normalizedError(error, "nativeFailure"), }); - return { ok: true, value: { turnId: command.turnId } }; - } catch (error) { - return { ok: false, error: normalizedError(error, "nativeFailure") }; } - } finally { - this.#configuring = false; } } + #finishCommand(active: ActiveCommand, outcome: HostItemOutcome): void { + if (this.#activeCommand !== active) return; + this.#activeCommand = null; + this.#emit({ + type: "item.completed", + turnId: active.command.turnId, + snapshot: { item: active.item, outcome }, + }); + this.#emit({ type: "turn.completed", turnId: active.command.turnId, outcome }); + } + async #cancel(command: TurnCancelCommand): Promise> { + const activeCommand = this.#activeCommand; + if (activeCommand?.command.turnId === command.turnId) { + if (!activeCommand.cancellationRequested) { + activeCommand.cancellationRequested = true; + activeCommand.abort.abort(new Error("DeepSeek Harness command cancelled by user")); + } + return { ok: true, value: { cancellationRequested: true } }; + } const active = this.#active; if (!active || active.command.turnId !== command.turnId) { return { ok: false, error: invalidState("DeepSeek Harness cancel requires the active Turn") }; @@ -1141,6 +1255,11 @@ class DeepSeekHarnessSession implements HarnessSession, DeepSeekHostSubscriber { #fault(error: HarnessError): void { if (this.#closed) return; + const activeCommand = this.#activeCommand; + if (activeCommand) { + activeCommand.abort.abort(new Error(error.message)); + this.#finishCommand(activeCommand, { status: "failed", error }); + } const active = this.#active; if (active) { const outcome: HostItemOutcome = { status: "failed", error }; diff --git a/packages/adapters/deepseek-harness/src/host-client.ts b/packages/adapters/deepseek-harness/src/host-client.ts index e4edcb4b..adab95a6 100644 --- a/packages/adapters/deepseek-harness/src/host-client.ts +++ b/packages/adapters/deepseek-harness/src/host-client.ts @@ -1,17 +1,50 @@ import { spawn, type ChildProcess } from "node:child_process"; +import { randomUUID } from "node:crypto"; import { accessSync, constants, statSync } from "node:fs"; import path from "node:path"; import { sanitizeDiagnosticTail } from "@codexhost/harness-adapter"; -import type { HostFrame, MuxFrame, RpcRequest } from "@deepseek-ai/dsh-host-apiproxy/api"; +import type { HostFrame, MuxFrame, RpcError, RpcRequest } from "@deepseek-ai/dsh-host-apiproxy/api"; import { hostFrameSchema, muxFrameSchema } from "@deepseek-ai/dsh-host-apiproxy/api/events.schema"; -import { serverRequestSchema } from "@deepseek-ai/dsh-host-apiproxy/api/rpc.schema"; +import { + serverRequestSchema, + serverResponseSchema, +} from "@deepseek-ai/dsh-host-apiproxy/api/rpc.schema"; import { AbstractApiClient, type IApiClient } from "@deepseek-ai/dsh-host-apiproxy/client"; +import type { SessionId } from "@deepseek-ai/dsh-session/types"; -export type DeepSeekHostClient = IApiClient; +export type DeepSeekHostClient = IApiClient & { readonly commands: DeepSeekCommandClient }; export type DeepSeekMuxEnvelope = RpcRequest; export type DeepSeekHostEnvelope = RpcRequest; +export interface DeepSeekCommandDescriptor { + readonly name: string; + readonly description: string; + readonly input?: { readonly hint: string; readonly images?: boolean }; +} + +export interface DeepSeekCommandExecution { + readonly commandId: string; + readonly result: + | { readonly kind: "success"; readonly text?: string; readonly sourceEventSeq?: number } + | { readonly kind: "error"; readonly text: string }; +} + +export type DeepSeekCommandResult = + { readonly ok: true; readonly value: T } | { readonly ok: false; readonly error: RpcError }; + +export interface DeepSeekCommandClient { + list( + sessionId: SessionId, + signal?: AbortSignal, + ): Promise>; + execute( + sessionId: SessionId, + line: string, + signal?: AbortSignal, + ): Promise>; +} + export type DeepSeekHarnessTransportErrorCode = "notInstalled" | "unavailable" | "protocolError" | "processExited"; @@ -28,13 +61,72 @@ export class DeepSeekHarnessTransportError extends Error { type StreamFrame = MuxFrame | HostFrame; type StreamItem = { type: "frame"; envelope: RpcRequest } | { type: "end" }; type FrameSchema = { parse(value: unknown): F }; +const MISSING_COMMAND_IMAGES = + 'typert gateway: commands/execute: args fields do not match the descriptor: missing "images"'; + +function commandProtocolError(method: string, detail: string): DeepSeekHarnessTransportError { + return new DeepSeekHarnessTransportError( + "protocolError", + `DeepSeek Harness '${method}' returned ${detail}`, + ); +} + +function record(value: unknown): Record | null { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : null; +} + +function parseCommandDescriptors(value: unknown): DeepSeekCommandDescriptor[] { + const valid = + Array.isArray(value) && + value.every((entry) => { + const descriptor = record(entry); + if ( + !descriptor || + typeof descriptor.name !== "string" || + typeof descriptor.description !== "string" + ) { + return false; + } + if (descriptor.input === undefined) return true; + const input = record(descriptor.input); + return ( + !!input && + typeof input.hint === "string" && + (input.images === undefined || typeof input.images === "boolean") + ); + }); + if (!valid) throw new TypeError("an invalid command catalog"); + return value as DeepSeekCommandDescriptor[]; +} + +function parseCommandExecution(value: unknown): DeepSeekCommandExecution | undefined { + if (value === undefined) return undefined; + const execution = record(value); + const result = record(execution?.result); + const validResult = + !!result && + ((result.kind === "error" && typeof result.text === "string") || + (result.kind === "success" && + (result.text === undefined || typeof result.text === "string") && + (result.sourceEventSeq === undefined || + (Number.isSafeInteger(result.sourceEventSeq) && + (result.sourceEventSeq as number) >= 0)))); + if (!execution || typeof execution.commandId !== "string" || !validResult) { + throw new TypeError("an invalid command execution"); + } + return value as DeepSeekCommandExecution; +} export class NodeDeepSeekHostClient extends AbstractApiClient { + readonly commands: DeepSeekCommandClient; readonly #endpoint: URL; constructor(endpoint: string, timeoutMs?: number) { super(timeoutMs); this.#endpoint = parseLoopbackEndpoint(endpoint); + this.commands = new NodeDeepSeekCommandClient(endpoint, timeoutMs); } protected override resolveBase(): string { @@ -133,6 +225,98 @@ export class NodeDeepSeekHostClient extends AbstractApiClient { } } +export class NodeDeepSeekCommandClient implements DeepSeekCommandClient { + readonly #endpoint: URL; + readonly #timeoutMs: number; + + constructor(endpoint: string, timeoutMs = 5_000) { + this.#endpoint = parseLoopbackEndpoint(endpoint); + this.#timeoutMs = timeoutMs; + } + + list( + sessionId: SessionId, + signal?: AbortSignal, + ): Promise> { + const timeout = AbortSignal.timeout(this.#timeoutMs); + return this.#call( + "commands/list", + { agentId: sessionId }, + parseCommandDescriptors, + signal ? AbortSignal.any([signal, timeout]) : timeout, + ); + } + + async execute( + sessionId: SessionId, + line: string, + signal?: AbortSignal, + ): Promise> { + const result = await this.#call( + "commands/execute", + { agentId: sessionId, line }, + parseCommandExecution, + signal, + ); + if ( + !result.ok && + result.error.code === "internal" && + result.error.message === MISSING_COMMAND_IMAGES + ) { + // rc.6 has no images parameter; newer DSH requires an explicit empty batch. + return this.#call( + "commands/execute", + { agentId: sessionId, line, images: [] }, + parseCommandExecution, + signal, + ); + } + return result; + } + + async #call( + method: string, + args: Record, + parse: (value: unknown) => T, + signal?: AbortSignal, + ): Promise> { + const rpcId = randomUUID(); + const response = await globalThis.fetch(new URL(`/api/${method}`, this.#endpoint), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ type: "client-request", rpcId, method, payload: { args } }), + ...(signal ? { signal } : {}), + }); + if (!response.ok) { + throw new DeepSeekHarnessTransportError( + "unavailable", + `DeepSeek Harness '${method}' transport failed with HTTP ${response.status}`, + ); + } + let envelope: ReturnType; + try { + envelope = serverResponseSchema.parse(await response.json()); + } catch (error) { + throw commandProtocolError( + method, + `an invalid response: ${error instanceof Error ? error.message : String(error)}`, + ); + } + if (envelope.rpcId !== rpcId) { + throw new DeepSeekHarnessTransportError( + "protocolError", + `DeepSeek Harness '${method}' returned an unexpected RPC identity`, + ); + } + if (!envelope.result.ok) return envelope.result; + try { + return { ok: true, value: parse(envelope.result.value) }; + } catch (error) { + throw commandProtocolError(method, error instanceof Error ? error.message : String(error)); + } + } +} + export interface DeepSeekHostConnectionOptions { command?: string; endpoint?: string; diff --git a/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts b/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts index 447476bc..5b36ba9e 100644 --- a/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts +++ b/packages/adapters/deepseek-harness/test/deepseek-harness-adapter.test.ts @@ -20,6 +20,7 @@ import { type DeepSeekHostConnectionLike, } from "../src/deepseek-harness-adapter.js"; import type { + DeepSeekCommandExecution, DeepSeekHostClient, DeepSeekHostSubscriber, DeepSeekMuxEnvelope, @@ -47,6 +48,10 @@ function success(value: T) { return { rpcId: RpcId("response"), result: { ok: true as const, value } }; } +function commandSuccess(value: T) { + return { ok: true as const, value }; +} + function event(seq: number, type: string, data: Record): HistoryEntry { return { event: { type, seq, time: seq, data } as HistoryEntry["event"], @@ -63,6 +68,8 @@ class FakeConnection implements DeepSeekHostConnectionLike { models: vi.fn(), selectModel: vi.fn(), prompt: vi.fn(), + commandList: vi.fn(), + commandExecute: vi.fn(), cancel: vi.fn(), respond: vi.fn(), }; @@ -104,10 +111,19 @@ class FakeConnection implements DeepSeekHostConnectionLike { }, ); this.calls.prompt.mockResolvedValue(success({ accepted: true })); + this.calls.commandList.mockResolvedValue( + commandSuccess([{ name: "compact", description: "Compact older conversation history" }]), + ); + this.calls.commandExecute.mockResolvedValue(commandSuccess(undefined)); this.calls.cancel.mockResolvedValue(success({ accepted: true })); this.calls.respond.mockResolvedValue({ accepted: true }); + const commands = { + list: this.calls.commandList, + execute: this.calls.commandExecute, + }; this.client = { sessions, + commands, host: { describe: vi.fn().mockResolvedValue( success({ @@ -179,6 +195,12 @@ async function collectUntilTurn(session: Awaited> throw new Error("Output stream ended before Turn completion"); } +async function nextEvent(iterator: AsyncIterator) { + const next = await iterator.next(); + if (next.done || next.value.kind !== "event") throw new Error("Expected a Harness event"); + return next.value.event; +} + async function openCreated(adapter: DeepSeekHarnessAdapter) { const opened = await adapter.open({ kind: "create", cwd: "/workspace" }); if (!opened.ok) throw new Error(opened.error.message); @@ -342,7 +364,7 @@ describe("DeepSeekHarnessAdapter local Host", () => { await adapter.close(); }); - it("exposes the command catalog and executes dsh.compact through the native prompt command seam", async () => { + it("discovers and executes dsh.compact through the native Remote command service", async () => { const { adapter, connection } = fixture(); const opened = await adapter.open({ kind: "resume", @@ -362,54 +384,90 @@ describe("DeepSeekHarnessAdapter local Host", () => { ok: true, value: { commands: [{ id: "dsh.compact", invocation: "/compact" }] }, }); - connection.calls.prompt.mockResolvedValueOnce( - success({ accepted: true, command: { kind: "success", text: "compacted" } }), + expect(connection.calls.commandList).toHaveBeenCalledWith(SESSION_ID); + let resolveExecution: + ((value: { ok: true; value: DeepSeekCommandExecution }) => void) | undefined; + connection.calls.commandExecute.mockImplementationOnce( + () => + new Promise<{ ok: true; value: DeepSeekCommandExecution }>((resolve) => { + resolveExecution = resolve; + }), ); const iterator = session.outputs[Symbol.asyncIterator](); - const executing = commands.execute({ - turnId: hostTurnIdSchema.parse("manual-compact"), - commandId: "dsh.compact", + const turnId = hostTurnIdSchema.parse("manual-compact"); + await expect( + commands.execute({ + turnId, + commandId: "dsh.compact", + }), + ).resolves.toEqual({ ok: true, value: { turnId } }); + expect(await nextEvent(iterator)).toEqual({ type: "turn.started", turnId }); + const started = await nextEvent(iterator); + if (started.type !== "item.started") throw new Error("Expected compaction Item start"); + expect(started).toMatchObject({ + type: "item.started", + turnId, + item: { type: "contextCompaction" }, }); - await expect(iterator.next()).resolves.toMatchObject({ - done: false, - value: { kind: "event", event: { type: "turn.started", turnId: "manual-compact" } }, + await expect(session.readSnapshot()).resolves.toMatchObject({ + ok: false, + error: { code: "sessionBusy" }, }); - await expect(iterator.next()).resolves.toMatchObject({ - done: false, - value: { - kind: "event", - event: { - type: "turn.completed", - turnId: "manual-compact", - outcome: { status: "succeeded" }, - }, - }, + + resolveExecution?.( + commandSuccess({ + commandId: "native-command-1", + result: { kind: "success", text: "compacted" }, + }), + ); + expect(await nextEvent(iterator)).toEqual({ + type: "item.completed", + turnId, + snapshot: { item: started.item, outcome: { status: "succeeded" } }, + }); + expect(await nextEvent(iterator)).toEqual({ + type: "turn.completed", + turnId, + outcome: { status: "succeeded" }, }); - await expect(executing).resolves.toEqual({ + expect(connection.calls.commandExecute).toHaveBeenCalledWith( + SESSION_ID, + "/compact", + expect.any(AbortSignal), + ); + expect(connection.calls.prompt).not.toHaveBeenCalled(); + await expect(session.readSnapshot()).resolves.toMatchObject({ ok: true, - value: { turnId: "manual-compact" }, + value: { turns: [] }, }); - expect(connection.calls.prompt).toHaveBeenCalledWith({ - sessionId: SESSION_ID, - mode: "queue", - content: [{ type: "text", text: "/compact" }], + await adapter.close(); + }); + + it("hides dsh.compact when the native deployment does not advertise the argument-free command", async () => { + const { adapter, connection } = fixture(); + connection.calls.commandList.mockResolvedValueOnce( + commandSuccess([ + { + name: "compact", + description: "Different deployment contract", + input: { hint: "" }, + }, + ]), + ); + const session = await openCreated(adapter); + const commands = session.commands; + if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); + + await expect(commands.list()).resolves.toEqual({ + ok: true, + value: { commands: [] }, }); await adapter.close(); }); it("rejects unknown Harness commands and command arguments without touching the Host", async () => { const { adapter, connection } = fixture(); - const opened = await adapter.open({ - kind: "resume", - cwd: "/workspace", - nativeRef: { - harnessId: adapter.harnessId, - nativeSessionId: SESSION_ID, - formatVersion: 1, - }, - }); - if (!opened.ok) throw new Error(opened.error.message); - const session = opened.value; + const session = await openCreated(adapter); const commands = session.commands; if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); @@ -426,33 +484,105 @@ describe("DeepSeekHarnessAdapter local Host", () => { arguments: { text: "keep details" }, }), ).resolves.toMatchObject({ ok: false, error: { code: "invalidRequest" } }); + expect(connection.calls.commandExecute).not.toHaveBeenCalled(); expect(connection.calls.prompt).not.toHaveBeenCalled(); await adapter.close(); }); - it("fails closed when DeepSeek Harness does not treat the invocation as a command", async () => { + it("fails the temporary Turn when the native command does not resolve", async () => { const { adapter, connection } = fixture(); - const opened = await adapter.open({ - kind: "resume", - cwd: "/workspace", - nativeRef: { - harnessId: adapter.harnessId, - nativeSessionId: SESSION_ID, - formatVersion: 1, - }, - }); - if (!opened.ok) throw new Error(opened.error.message); - const session = opened.value; + const session = await openCreated(adapter); const commands = session.commands; if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); + const iterator = session.outputs[Symbol.asyncIterator](); + const turnId = hostTurnIdSchema.parse("missing-compact"); - await expect( - commands.execute({ - turnId: hostTurnIdSchema.parse("manual-compact"), - commandId: "dsh.compact", + await expect(commands.execute({ turnId, commandId: "dsh.compact" })).resolves.toEqual({ + ok: true, + value: { turnId }, + }); + expect((await nextEvent(iterator)).type).toBe("turn.started"); + const started = await nextEvent(iterator); + expect(started.type).toBe("item.started"); + await expect(nextEvent(iterator)).resolves.toMatchObject({ + type: "item.completed", + snapshot: { outcome: { status: "failed", error: { code: "nativeFailure" } } }, + }); + await expect(nextEvent(iterator)).resolves.toMatchObject({ + type: "turn.completed", + outcome: { status: "failed", error: { code: "nativeFailure" } }, + }); + expect(connection.calls.prompt).not.toHaveBeenCalled(); + await adapter.close(); + }); + + it("projects the native compact busy result as a failed temporary Turn", async () => { + const { adapter, connection } = fixture(); + const session = await openCreated(adapter); + const commands = session.commands; + if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); + connection.calls.commandExecute.mockResolvedValueOnce( + commandSuccess({ + commandId: "native-command-1", + result: { + kind: "error", + text: "Compaction is unavailable because this process has an active compaction, or the agent is not idle.", + }, }), - ).resolves.toMatchObject({ ok: false, error: { code: "nativeFailure" } }); - expect(connection.calls.prompt).toHaveBeenCalledTimes(1); + ); + const iterator = session.outputs[Symbol.asyncIterator](); + const turnId = hostTurnIdSchema.parse("busy-compact"); + + await commands.execute({ turnId, commandId: "dsh.compact" }); + await nextEvent(iterator); + await nextEvent(iterator); + await expect(nextEvent(iterator)).resolves.toMatchObject({ + type: "item.completed", + snapshot: { outcome: { status: "failed", error: { code: "sessionBusy" } } }, + }); + await expect(nextEvent(iterator)).resolves.toMatchObject({ + type: "turn.completed", + outcome: { status: "failed", error: { code: "sessionBusy" } }, + }); + await expect(session.readSnapshot()).resolves.toMatchObject({ ok: true }); + await adapter.close(); + }); + + it("cancels a running native compact command through its AbortSignal", async () => { + const { adapter, connection } = fixture(); + const session = await openCreated(adapter); + const commands = session.commands; + if (!commands) throw new Error("DeepSeek Harness Session did not expose commands"); + let commandSignal: AbortSignal | undefined; + connection.calls.commandExecute.mockImplementationOnce( + (_sessionId: SessionId, _line: string, signal: AbortSignal) => + new Promise((_resolve, reject) => { + commandSignal = signal; + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }), + ); + const iterator = session.outputs[Symbol.asyncIterator](); + const turnId = hostTurnIdSchema.parse("cancel-compact"); + + await commands.execute({ + turnId, + commandId: "dsh.compact", + }); + await nextEvent(iterator); + await nextEvent(iterator); + await expect(session.execute({ type: "turn.cancel", turnId })).resolves.toEqual({ + ok: true, + value: { cancellationRequested: true }, + }); + expect(commandSignal?.aborted).toBe(true); + await expect(nextEvent(iterator)).resolves.toMatchObject({ + type: "item.completed", + snapshot: { outcome: { status: "cancelled" } }, + }); + await expect(nextEvent(iterator)).resolves.toMatchObject({ + type: "turn.completed", + outcome: { status: "cancelled" }, + }); await adapter.close(); }); diff --git a/packages/adapters/deepseek-harness/test/host-client.test.ts b/packages/adapters/deepseek-harness/test/host-client.test.ts index c94bf99b..20234a30 100644 --- a/packages/adapters/deepseek-harness/test/host-client.test.ts +++ b/packages/adapters/deepseek-harness/test/host-client.test.ts @@ -9,6 +9,7 @@ import { describe, expect, it, vi } from "vitest"; import type { DeepSeekHostClient } from "../src/host-client.js"; import { DeepSeekHostConnection, + NodeDeepSeekCommandClient, NodeDeepSeekHostClient, deepSeekProcessInvocation, resolveDeepSeekCommand, @@ -47,6 +48,97 @@ function childProcess(): ChildProcess { } describe("DeepSeek local Host connection", () => { + it("calls the Typert Remote command wire and validates its catalog", async () => { + const requests: Array<{ url: string; body: Record }> = []; + vi.stubGlobal( + "fetch", + vi.fn((input: URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as Record; + requests.push({ url: input.href, body }); + return Promise.resolve( + new Response( + JSON.stringify({ + type: "server-response", + rpcId: body.rpcId, + result: { + ok: true, + value: [{ name: "compact", description: "Compact older conversation history" }], + }, + }), + ), + ); + }), + ); + try { + const client = new NodeDeepSeekCommandClient("http://127.0.0.1:43123"); + + await expect(client.list("session-1" as never)).resolves.toEqual({ + ok: true, + value: [{ name: "compact", description: "Compact older conversation history" }], + }); + expect(requests).toHaveLength(1); + expect(requests[0]?.url).toBe("http://127.0.0.1:43123/api/commands/list"); + expect(requests[0]?.body).toMatchObject({ + type: "client-request", + method: "commands/list", + payload: { args: { agentId: "session-1" } }, + }); + } finally { + vi.unstubAllGlobals(); + } + }); + + it("retries commands/execute with the newer empty images field only when requested", async () => { + const payloads: Array> = []; + vi.stubGlobal( + "fetch", + vi.fn((input: URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as Record; + payloads.push(body.payload as Record); + const first = payloads.length === 1; + return Promise.resolve( + new Response( + JSON.stringify({ + type: "server-response", + rpcId: body.rpcId, + result: first + ? { + ok: false, + error: { + code: "internal", + message: + 'typert gateway: commands/execute: args fields do not match the descriptor: missing "images"', + details: {}, + }, + } + : { + ok: true, + value: { + commandId: "command-1", + result: { kind: "success", text: "compacted" }, + }, + }, + }), + ), + ); + }), + ); + try { + const client = new NodeDeepSeekCommandClient("http://127.0.0.1:43123"); + + await expect(client.execute("session-1" as never, "/compact")).resolves.toEqual({ + ok: true, + value: { commandId: "command-1", result: { kind: "success", text: "compacted" } }, + }); + expect(payloads).toEqual([ + { args: { agentId: "session-1", line: "/compact" } }, + { args: { agentId: "session-1", line: "/compact", images: [] } }, + ]); + } finally { + vi.unstubAllGlobals(); + } + }); + it("connects to an existing compatible Host without spawning or stopping it", async () => { const spawn = vi.fn(); const dependencies: DeepSeekHostConnectionDependencies = {