370 lines
15 KiB
TypeScript
370 lines
15 KiB
TypeScript
/**
|
|
* Automation-only Agent Client Protocol server over JSON-RPC stdio.
|
|
*
|
|
* The bridge exposes fresh harness sessions to trusted programmatic clients. It
|
|
* carries prompt text, committed assistant text, cancellation, and one-shot
|
|
* permission decisions; presentation and human-interaction features stay with
|
|
* the harness's UI modules.
|
|
*
|
|
* @module @deepseek-ai/dsh-acp
|
|
*/
|
|
|
|
import type { Context } from 'cordis'
|
|
import { randomUUID } from 'node:crypto'
|
|
import { isAbsolute } from 'node:path'
|
|
import { Readable, Writable } from 'node:stream'
|
|
import Schema from 'schemastery'
|
|
import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
|
import {
|
|
AgentSideConnection,
|
|
ndJsonStream,
|
|
PROTOCOL_VERSION,
|
|
RequestError,
|
|
type Agent as AcpAgent,
|
|
type AuthenticateRequest,
|
|
type CancelNotification,
|
|
type InitializeRequest,
|
|
type InitializeResponse,
|
|
type NewSessionRequest,
|
|
type NewSessionResponse,
|
|
type PromptRequest,
|
|
type PromptResponse,
|
|
type SessionNotification,
|
|
type StopReason,
|
|
type Stream,
|
|
} from '@agentclientprotocol/sdk'
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
import { SessionId, type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
|
|
// Side-effect type import: declaration-merges the approval waterfall answered below.
|
|
import type {} from '@deepseek-ai/dsh-user-approval'
|
|
import { acpPromptToText, promptHasUnsupportedContent, turnEndToStopReason } from './codec.ts'
|
|
|
|
export const name = 'acp'
|
|
/** The bridge creates and owns agents; every other concern is carried by the agent composition. */
|
|
export const inject = ['agents']
|
|
|
|
/** Preserve invalid-parameter detail in the SDK wire error message. */
|
|
function invalidParams(detail: string): RequestError {
|
|
return RequestError.invalidParams(undefined, detail)
|
|
}
|
|
|
|
/** Preserve failed-turn detail; plain handler errors become a generic wire internal error. */
|
|
function internalError(detail: string): RequestError {
|
|
return RequestError.internalError(undefined, detail)
|
|
}
|
|
|
|
/** Plugin config: the provider/model target used for each ACP-created agent. */
|
|
export interface AcpConfig {
|
|
/** Provider route for created agents. */
|
|
provider?: string
|
|
/** Model name for created agents. */
|
|
model?: string
|
|
/** Runtime-only transport override; production uses stdio. */
|
|
stream?: Stream
|
|
}
|
|
|
|
export const Config: Schema<AcpConfig> = Schema.object({
|
|
provider: Schema.string(),
|
|
model: Schema.string(),
|
|
})
|
|
|
|
/** Per-session protocol state. */
|
|
interface SessionRecord {
|
|
agent: Agent
|
|
/** Exact owned-agent disposer; resolves after registry, loop, and session teardown. */
|
|
dispose: () => Promise<void>
|
|
/** In-flight prompt and its captured turn number for exact settlement. */
|
|
inflight: {
|
|
resolve: (reason: StopReason) => void
|
|
reject: (error: Error) => void
|
|
turn: number | undefined
|
|
/**
|
|
* A failed turn's terminal reason, held until quiescence: a retry action
|
|
* closes the failed turn and opens a successor that adopts the prompt, so
|
|
* rejecting at `turn/end` would race the recovery.
|
|
*/
|
|
pendingError: Extract<TurnEndReason, { kind: 'error' }> | undefined
|
|
} | undefined
|
|
}
|
|
|
|
/**
|
|
* Mount the automation-only ACP server.
|
|
* @param ctx - Cordis context carrying the agent factory and session events.
|
|
* @param config - Initial provider/model target and optional test transport.
|
|
*/
|
|
export function apply(ctx: Context, config: AcpConfig): void {
|
|
// ACP handlers execute outside this plugin's injection scope, so capture the
|
|
// injected service during apply rather than reading it lazily in a callback.
|
|
const agents = ctx.agents
|
|
const logger = ctx.logger
|
|
const sessions = new Map<SessionId, SessionRecord>()
|
|
let closed = false
|
|
let conn: AgentSideConnection
|
|
|
|
/** Return the bridge-owned record for an agent, rejecting same-id impostors. */
|
|
const ownedRecord = (agent: Agent): SessionRecord | undefined => {
|
|
const record = sessions.get(agent.session.id)
|
|
return record?.agent === agent ? record : undefined
|
|
}
|
|
|
|
const assertOpen = (): void => {
|
|
if (closed) throw internalError('the ACP bridge has been disposed')
|
|
}
|
|
|
|
const requireSession = (sessionId: SessionId): SessionRecord => {
|
|
const record = sessions.get(sessionId)
|
|
if (record === undefined) throw invalidParams(`unknown session: ${sessionId}`)
|
|
return record
|
|
}
|
|
|
|
/** Send a protocol update without letting a disconnected client fail an agent turn. */
|
|
const notify = (notification: SessionNotification): void => {
|
|
/* v8 ignore next 3 -- only a transport write failure reaches this guard. */
|
|
void conn.sessionUpdate(notification).catch((error: unknown) => {
|
|
logger.warn(`acp: session/update failed: ${String(error)}`)
|
|
})
|
|
}
|
|
|
|
const settlePrompt = (record: SessionRecord, reason: StopReason): void => {
|
|
const inflight = record.inflight
|
|
if (inflight === undefined) return
|
|
record.inflight = undefined
|
|
inflight.resolve(reason)
|
|
}
|
|
|
|
const rejectFromError = (
|
|
inflight: NonNullable<SessionRecord['inflight']>,
|
|
reason: Extract<TurnEndReason, { kind: 'error' }>,
|
|
): void => {
|
|
inflight.reject(internalError(`turn failed: ${'failure' in reason ? reason.failure.message : reason.message}`))
|
|
}
|
|
|
|
// Emit only committed assistant text. Raw chunks, reasoning, tools, plans,
|
|
// titles, and retry markers are presentation or trace data and stay off the
|
|
// automation wire.
|
|
ctx.on('session/event', (session, event: SessionEvent) => {
|
|
const record = sessions.get(session.header.id)
|
|
if (record === undefined || record.agent.session !== session) return
|
|
try {
|
|
if (event.type === 'assistant/message') {
|
|
for (const block of event.data.message.content) {
|
|
if (block.type === 'text' && block.text.length > 0) {
|
|
notify({
|
|
sessionId: record.agent.session.id,
|
|
update: {
|
|
sessionUpdate: 'agent_message_chunk',
|
|
content: { type: 'text', text: block.text },
|
|
},
|
|
})
|
|
}
|
|
}
|
|
}
|
|
} finally {
|
|
const inflight = record.inflight
|
|
if (inflight !== undefined && event.type === 'turn/start') {
|
|
if (inflight.turn === undefined && event.data.trigger.kind === 'message'
|
|
&& event.data.trigger.source.kind === 'user') {
|
|
inflight.turn = event.data.turn
|
|
} else if (inflight.pendingError !== undefined && event.data.trigger.kind === 'retry') {
|
|
// A recovery policy opened a retry turn on the failed history: the
|
|
// prompt rides it instead of rejecting on the failed turn's end.
|
|
inflight.turn = event.data.turn
|
|
inflight.pendingError = undefined
|
|
}
|
|
} else if (inflight !== undefined && event.type === 'turn/end' && inflight.turn === event.data.turn) {
|
|
if (event.data.reason.kind === 'error') {
|
|
// Hold the rejection: request recovery may adopt the prompt with a
|
|
// successor turn; quiescence without one delivers this error.
|
|
inflight.turn = undefined
|
|
inflight.pendingError = event.data.reason
|
|
} else {
|
|
record.inflight = undefined
|
|
inflight.resolve(turnEndToStopReason(event.data.reason))
|
|
}
|
|
}
|
|
}
|
|
})
|
|
|
|
// Permission requests are a machine policy channel for ACP clients such as
|
|
// dsh-subagent-acp. The bridge offers one-shot choices only and never infers a
|
|
// durable grant from an unknown client response.
|
|
ctx.on('approval/request', (request, next) => {
|
|
const record = ownedRecord(request.agent)
|
|
if (record === undefined || request.callId === undefined) return next()
|
|
return conn.requestPermission({
|
|
sessionId: record.agent.session.id,
|
|
toolCall: { toolCallId: request.callId },
|
|
options: [
|
|
{ optionId: 'allow-once', name: 'Allow once', kind: 'allow_once' },
|
|
{ optionId: 'reject-once', name: 'Reject', kind: 'reject_once' },
|
|
],
|
|
}).then(({ outcome }) => {
|
|
if (outcome.outcome === 'cancelled') return 'cancelled'
|
|
return outcome.optionId === 'allow-once' ? 'allowed-once' : 'rejected'
|
|
})
|
|
})
|
|
|
|
const makeAgent = (connection: AgentSideConnection): AcpAgent => {
|
|
conn = connection
|
|
return {
|
|
initialize(_params: InitializeRequest): Promise<InitializeResponse> {
|
|
// Single-version agent: the spec's "same version if supported, else
|
|
// the latest supported" both resolve to this server's one version.
|
|
return Promise.resolve({
|
|
protocolVersion: PROTOCOL_VERSION,
|
|
agentInfo: { name: 'deepseek-harness-acp', version: '0.0.1' },
|
|
agentCapabilities: {
|
|
promptCapabilities: { image: false, audio: false, embeddedContext: false },
|
|
},
|
|
authMethods: [],
|
|
})
|
|
},
|
|
|
|
authenticate(_params: AuthenticateRequest): Promise<void> {
|
|
return Promise.resolve()
|
|
},
|
|
|
|
async newSession(params: NewSessionRequest): Promise<NewSessionResponse> {
|
|
assertOpen()
|
|
validateSessionParams(params)
|
|
const sessionId = SessionId(randomUUID())
|
|
const handle = await agents.create({
|
|
sessionId,
|
|
meta: { cwd: params.cwd },
|
|
agentOptions: agentOptions(config),
|
|
})
|
|
/* v8 ignore next 4 -- a real stdio close can race an in-flight create. */
|
|
if (closed) {
|
|
await handle.dispose()
|
|
throw internalError('connection closed during session/new')
|
|
}
|
|
sessions.set(sessionId, {
|
|
agent: handle.agent,
|
|
dispose: () => handle.dispose(),
|
|
inflight: undefined,
|
|
})
|
|
return { sessionId }
|
|
},
|
|
|
|
async prompt(params: PromptRequest): Promise<PromptResponse> {
|
|
assertOpen()
|
|
const record = requireSession(SessionId(params.sessionId))
|
|
if (record.inflight !== undefined) {
|
|
throw invalidParams('a prompt is already in flight for this session')
|
|
}
|
|
if (promptHasUnsupportedContent(params.prompt)) {
|
|
throw invalidParams('only text and resource_link prompt content is supported')
|
|
}
|
|
const text = acpPromptToText(params.prompt)
|
|
if (text.trim().length === 0) throw invalidParams('empty prompt')
|
|
|
|
// Not driving a retired agent is this bridge's contract: an
|
|
// agent-loop-only reload disposes the loop's agents while the bridge
|
|
// record survives, so validate the record against the live registry
|
|
// before sending — a disposed machine would accept the item silently.
|
|
if (ctx.agents.get(record.agent.id) !== record.agent) {
|
|
throw internalError('prompt was not queued: the agent was disposed outside the bridge')
|
|
}
|
|
const stopReason = await new Promise<StopReason>((resolve, reject) => {
|
|
// Arm the slot before followup() so a listener-driven synchronous
|
|
// turn cannot slip past correlation; a synchronous followup()
|
|
// failure (invalid input) must free the slot again or the session
|
|
// would reject every later prompt as already in flight.
|
|
const inflight: NonNullable<SessionRecord['inflight']> = {
|
|
resolve, reject, turn: undefined, pendingError: undefined,
|
|
}
|
|
record.inflight = inflight
|
|
try {
|
|
record.agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
|
|
// The machine's send() contains listener failures and accepts
|
|
// any typed input; this guards a future synchronous throw so the
|
|
// slot cannot wedge.
|
|
/* v8 ignore start -- future-proofing guard, see above */
|
|
} catch (error: unknown) {
|
|
record.inflight = undefined
|
|
const detail = error instanceof Error ? error.message : String(error)
|
|
throw internalError(`prompt was not queued: ${detail}`)
|
|
}
|
|
/* v8 ignore stop */
|
|
// Admission is pre-turn and retries outlive their failed turn, so a
|
|
// turnless slot settles only at quiescence: a held failure rejects
|
|
// (no retry adopted the prompt); no turn at all means admission
|
|
// discarded the prompt — report cancelled.
|
|
void record.agent.whenIdle().then(() => {
|
|
if (record.inflight !== inflight || inflight.turn !== undefined) return
|
|
record.inflight = undefined
|
|
if (inflight.pendingError !== undefined) {
|
|
rejectFromError(inflight, inflight.pendingError)
|
|
return
|
|
}
|
|
inflight.resolve('cancelled')
|
|
})
|
|
})
|
|
return { stopReason }
|
|
},
|
|
|
|
cancel(params: CancelNotification): Promise<void> {
|
|
const record = sessions.get(SessionId(params.sessionId))
|
|
if (record === undefined) return Promise.resolve()
|
|
record.agent.cancel({ kind: 'user' })
|
|
settlePrompt(record, 'cancelled')
|
|
return Promise.resolve()
|
|
},
|
|
}
|
|
}
|
|
|
|
/* v8 ignore next 4 -- production stdio wiring; tests inject config.stream. */
|
|
const stream: Stream = config.stream ?? ndJsonStream(
|
|
Writable.toWeb(process.stdout) as WritableStream<Uint8Array>,
|
|
Readable.toWeb(process.stdin) as ReadableStream<Uint8Array>,
|
|
)
|
|
conn = new AgentSideConnection(makeAgent, stream)
|
|
|
|
let quiescing: Promise<void> | undefined
|
|
const quiesce = (): Promise<void> => {
|
|
if (quiescing !== undefined) return quiescing
|
|
closed = true
|
|
const records = [...sessions.values()]
|
|
sessions.clear()
|
|
quiescing = Promise.all(records.map(async (record) => {
|
|
settlePrompt(record, 'cancelled')
|
|
await record.dispose()
|
|
})).then(() => {})
|
|
return quiescing
|
|
}
|
|
|
|
/* v8 ignore start -- production transport rejection and teardown failure. */
|
|
void conn.closed
|
|
.catch((error: unknown) => {
|
|
logger.warn(`acp: connection closed with an error: ${String(error)}`)
|
|
})
|
|
.then(quiesce)
|
|
.catch((error: unknown) => {
|
|
logger.warn(`acp: connection-close teardown failed: ${String(error)}`)
|
|
})
|
|
/* v8 ignore stop */
|
|
|
|
ctx.effect(() => quiesce, 'acp.connection')
|
|
}
|
|
|
|
/**
|
|
* Build per-agent options from plugin config without assigning absent optional fields.
|
|
* @param config - ACP provider/model configuration.
|
|
* @returns the configured fields only.
|
|
*/
|
|
function agentOptions(config: AcpConfig): { provider?: string; model?: string } {
|
|
return {
|
|
...config.provider !== undefined ? { provider: config.provider } : {},
|
|
...config.model !== undefined ? { model: config.model } : {},
|
|
}
|
|
}
|
|
|
|
/** Reject session features outside the automation contract. */
|
|
function validateSessionParams(params: NewSessionRequest): void {
|
|
if (!isAbsolute(params.cwd)) throw invalidParams(`cwd must be an absolute path: ${params.cwd}`)
|
|
if (params.additionalDirectories !== undefined && params.additionalDirectories.length > 0) {
|
|
throw invalidParams('additionalDirectories is not supported')
|
|
}
|
|
if (params.mcpServers.length > 0) throw invalidParams('mcpServers is not supported')
|
|
}
|