|
@@ -17,7 +17,9 @@ import {
|
|
|
PROTOCOL_VERSION,
|
|
PROTOCOL_VERSION,
|
|
|
RequestError,
|
|
RequestError,
|
|
|
type Agent as AcpAgent,
|
|
type Agent as AcpAgent,
|
|
|
|
|
+ type AnyMessage,
|
|
|
type AuthenticateRequest,
|
|
type AuthenticateRequest,
|
|
|
|
|
+ type AvailableCommand,
|
|
|
type CancelNotification,
|
|
type CancelNotification,
|
|
|
type ContentBlock as AcpContentBlock,
|
|
type ContentBlock as AcpContentBlock,
|
|
|
type CreateElicitationRequest,
|
|
type CreateElicitationRequest,
|
|
@@ -46,6 +48,7 @@ import type { ContentBlock, LlmCallConfig, LlmModelInfo, LlmProviderInfo } from
|
|
|
import { assertNever, CallId } from '@deepseek-ai/dsh-llm'
|
|
import { assertNever, CallId } from '@deepseek-ai/dsh-llm'
|
|
|
import type {} from '@deepseek-ai/dsh-llm-retry'
|
|
import type {} from '@deepseek-ai/dsh-llm-retry'
|
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
|
|
|
+import type {} from '@deepseek-ai/dsh-commands'
|
|
|
import { SessionId } from '@deepseek-ai/dsh-session'
|
|
import { SessionId } from '@deepseek-ai/dsh-session'
|
|
|
// Side-effect type import: resolves `ctx.get('permission')` to the service.
|
|
// Side-effect type import: resolves `ctx.get('permission')` to the service.
|
|
|
import type {} from '@deepseek-ai/dsh-permission'
|
|
import type {} from '@deepseek-ai/dsh-permission'
|
|
@@ -77,13 +80,50 @@ import {
|
|
|
|
|
|
|
|
export const name = 'acp'
|
|
export const name = 'acp'
|
|
|
// Interface services back loading, presentation, interaction, and prompt assembly.
|
|
// Interface services back loading, presentation, interaction, and prompt assembly.
|
|
|
-export const inject = ['agents', 'sessionPersistence', 'tools', 'userInteraction', 'llm', 'systemPrompt']
|
|
|
|
|
|
|
+export const inject = ['agents', 'commands', 'sessionPersistence', 'tools', 'userInteraction', 'llm', 'systemPrompt']
|
|
|
|
|
|
|
|
/** Preserve invalid-parameter detail in the SDK wire error message. */
|
|
/** Preserve invalid-parameter detail in the SDK wire error message. */
|
|
|
function invalidParams(detail: string): RequestError {
|
|
function invalidParams(detail: string): RequestError {
|
|
|
return RequestError.invalidParams(undefined, detail)
|
|
return RequestError.invalidParams(undefined, detail)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+/** Render arbitrary thrown values without trusting their string coercion. */
|
|
|
|
|
+function renderThrown(value: unknown): string {
|
|
|
|
|
+ try {
|
|
|
|
|
+ return String(value)
|
|
|
|
|
+ } catch {
|
|
|
|
|
+ return '<unrenderable thrown value>'
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Return a server-created session id carried by an outbound success response. */
|
|
|
|
|
+function responseSessionId(message: AnyMessage): SessionId | undefined {
|
|
|
|
|
+ if (!('result' in message) || typeof message.result !== 'object' || message.result === null
|
|
|
|
|
+ || !('sessionId' in message.result) || typeof message.result.sessionId !== 'string') {
|
|
|
|
|
+ return undefined
|
|
|
|
|
+ }
|
|
|
|
|
+ return SessionId(message.result.sessionId)
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/** Observe messages only after the wrapped ACP transport has written them. */
|
|
|
|
|
+function observeOutbound(stream: Stream, onWritten: (message: AnyMessage) => void): Stream {
|
|
|
|
|
+ const writer = stream.writable.getWriter()
|
|
|
|
|
+ return {
|
|
|
|
|
+ readable: stream.readable,
|
|
|
|
|
+ writable: new WritableStream<AnyMessage>({
|
|
|
|
|
+ async write(message) {
|
|
|
|
|
+ await writer.write(message)
|
|
|
|
|
+ onWritten(message)
|
|
|
|
|
+ },
|
|
|
|
|
+ /* v8 ignore start -- the ACP SDK never closes or aborts its outbound stream;
|
|
|
|
|
+ preserve the wrapped Stream contract for other consumers nonetheless */
|
|
|
|
|
+ close: () => writer.close(),
|
|
|
|
|
+ abort: (reason: unknown) => writer.abort(reason),
|
|
|
|
|
+ /* v8 ignore stop */
|
|
|
|
|
+ }),
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
/** Preserve failed-turn detail; plain handler errors become a generic wire internal error. */
|
|
/** Preserve failed-turn detail; plain handler errors become a generic wire internal error. */
|
|
|
function internalError(detail: string): RequestError {
|
|
function internalError(detail: string): RequestError {
|
|
|
return RequestError.internalError(undefined, detail)
|
|
return RequestError.internalError(undefined, detail)
|
|
@@ -260,6 +300,8 @@ interface SessionRecord {
|
|
|
reject: (error: Error) => void
|
|
reject: (error: Error) => void
|
|
|
turn: number | undefined
|
|
turn: number | undefined
|
|
|
} | undefined
|
|
} | undefined
|
|
|
|
|
+ /** Abort owner for a direct slash-command request, mutually exclusive with `inflight`. */
|
|
|
|
|
+ commandAbort: AbortController | undefined
|
|
|
/** Last idle switch per knob, anchored before the next prompt assembles. */
|
|
/** Last idle switch per knob, anchored before the next prompt assembles. */
|
|
|
pendingSwitches: { preset?: string }
|
|
pendingSwitches: { preset?: string }
|
|
|
}
|
|
}
|
|
@@ -274,6 +316,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
// ACP handlers execute outside this plugin's injection scope, so capture
|
|
// ACP handlers execute outside this plugin's injection scope, so capture
|
|
|
// injected services during apply(); lazy service reads in a handler fail.
|
|
// injected services during apply(); lazy service reads in a handler fail.
|
|
|
const agents = ctx.agents
|
|
const agents = ctx.agents
|
|
|
|
|
+ const commands = ctx.commands
|
|
|
const llm = ctx.llm
|
|
const llm = ctx.llm
|
|
|
const sessionPersistence = ctx.sessionPersistence
|
|
const sessionPersistence = ctx.sessionPersistence
|
|
|
const logger = ctx.logger
|
|
const logger = ctx.logger
|
|
@@ -380,6 +423,9 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
const sessions = new Map<SessionId, SessionRecord>()
|
|
const sessions = new Map<SessionId, SessionRecord>()
|
|
|
// Reserve an id before resume so pipelined load/new requests cannot duplicate it.
|
|
// Reserve an id before resume so pipelined load/new requests cannot duplicate it.
|
|
|
const loadingIds = new Set<SessionId>()
|
|
const loadingIds = new Set<SessionId>()
|
|
|
|
|
+ // A new-session response introduces its server-generated id to the client;
|
|
|
|
|
+ // keep its initial command snapshot pending until that response is written.
|
|
|
|
|
+ const pendingCommandSnapshots = new Map<SessionId, SessionRecord>()
|
|
|
// Async creation checks this after awaits to avoid publishing after teardown.
|
|
// Async creation checks this after awaits to avoid publishing after teardown.
|
|
|
let closed = false
|
|
let closed = false
|
|
|
// Each new or loaded session snapshots the latest connection capability.
|
|
// Each new or loaded session snapshots the latest connection capability.
|
|
@@ -468,6 +514,43 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
})
|
|
})
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ /** Project the effective registry view onto ACP discovery metadata. */
|
|
|
|
|
+ const availableCommands = (agent: Agent): AvailableCommand[] => commands.list(agent).map(command => ({
|
|
|
|
|
+ name: command.name,
|
|
|
|
|
+ description: command.description,
|
|
|
|
|
+ ...command.input === undefined ? {} : { input: { hint: command.input.hint } },
|
|
|
|
|
+ }))
|
|
|
|
|
+
|
|
|
|
|
+ /** Push the protocol's full-snapshot command catalog for one live session. */
|
|
|
|
|
+ const notifyCommands = (rec: SessionRecord): void => {
|
|
|
|
|
+ notify({
|
|
|
|
|
+ sessionId: rec.agent.session.id,
|
|
|
|
|
+ update: {
|
|
|
|
|
+ sessionUpdate: 'available_commands_update',
|
|
|
|
|
+ availableCommands: availableCommands(rec.agent),
|
|
|
|
|
+ },
|
|
|
|
|
+ })
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /** Enqueue a new session's first command snapshot behind its written RPC response. */
|
|
|
|
|
+ const announceInitialCommands = (message: AnyMessage): void => {
|
|
|
|
|
+ const sessionId = responseSessionId(message)
|
|
|
|
|
+ if (sessionId === undefined) return
|
|
|
|
|
+ const rec = pendingCommandSnapshots.get(sessionId)
|
|
|
|
|
+ if (rec === undefined) return
|
|
|
|
|
+ pendingCommandSnapshots.delete(sessionId)
|
|
|
|
|
+ notifyCommands(rec)
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Registration and HMR removal can affect global or one scoped view; refresh
|
|
|
|
|
+ // every announced bridge-owned session and let the registry resolve each
|
|
|
|
|
+ // exact agent. A pending new-session snapshot will read the latest registry.
|
|
|
|
|
+ ctx.on('commands/change', () => {
|
|
|
|
|
+ for (const rec of sessions.values()) {
|
|
|
|
|
+ if (!pendingCommandSnapshots.has(rec.agent.session.id)) notifyCommands(rec)
|
|
|
|
|
+ }
|
|
|
|
|
+ })
|
|
|
|
|
+
|
|
|
/** Settle the in-flight prompt with a stop reason, exactly once (no-op if none pending). */
|
|
/** Settle the in-flight prompt with a stop reason, exactly once (no-op if none pending). */
|
|
|
const settlePrompt = (rec: SessionRecord, reason: StopReason): void => {
|
|
const settlePrompt = (rec: SessionRecord, reason: StopReason): void => {
|
|
|
const inflight = rec.inflight
|
|
const inflight = rec.inflight
|
|
@@ -674,15 +757,18 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
await handle.dispose()
|
|
await handle.dispose()
|
|
|
throw internalError('connection closed during session/new')
|
|
throw internalError('connection closed during session/new')
|
|
|
}
|
|
}
|
|
|
- sessions.set(sessionId, {
|
|
|
|
|
|
|
+ const record: SessionRecord = {
|
|
|
agent: handle.agent,
|
|
agent: handle.agent,
|
|
|
dispose: () => handle.dispose(),
|
|
dispose: () => handle.dispose(),
|
|
|
presenter: makePresenter(handle.agent),
|
|
presenter: makePresenter(handle.agent),
|
|
|
terminalEnabled: terminalOutputCap,
|
|
terminalEnabled: terminalOutputCap,
|
|
|
target,
|
|
target,
|
|
|
inflight: undefined,
|
|
inflight: undefined,
|
|
|
|
|
+ commandAbort: undefined,
|
|
|
pendingSwitches: {},
|
|
pendingSwitches: {},
|
|
|
- })
|
|
|
|
|
|
|
+ }
|
|
|
|
|
+ sessions.set(sessionId, record)
|
|
|
|
|
+ pendingCommandSnapshots.set(sessionId, record)
|
|
|
const configOptions = configOptionsFor(handle.agent, directory)
|
|
const configOptions = configOptionsFor(handle.agent, directory)
|
|
|
return { sessionId, ...configOptions.length > 0 ? { configOptions } : {} }
|
|
return { sessionId, ...configOptions.length > 0 ? { configOptions } : {} }
|
|
|
},
|
|
},
|
|
@@ -763,6 +849,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
terminalEnabled,
|
|
terminalEnabled,
|
|
|
target,
|
|
target,
|
|
|
inflight: undefined,
|
|
inflight: undefined,
|
|
|
|
|
+ commandAbort: undefined,
|
|
|
pendingSwitches: {},
|
|
pendingSwitches: {},
|
|
|
}
|
|
}
|
|
|
sessions.set(sessionId, record)
|
|
sessions.set(sessionId, record)
|
|
@@ -787,6 +874,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
for (const event of agent.session.events) {
|
|
for (const event of agent.session.events) {
|
|
|
streamSessionEventUpdate(sessionId, event, notify, replayPresenter, replayTerminal)
|
|
streamSessionEventUpdate(sessionId, event, notify, replayPresenter, replayTerminal)
|
|
|
}
|
|
}
|
|
|
|
|
+ notifyCommands(record)
|
|
|
const configOptions = configOptionsFor(agent, directory)
|
|
const configOptions = configOptionsFor(agent, directory)
|
|
|
return configOptions.length > 0 ? { configOptions } : {}
|
|
return configOptions.length > 0 ? { configOptions } : {}
|
|
|
} finally {
|
|
} finally {
|
|
@@ -797,7 +885,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
async prompt(params: PromptRequest): Promise<PromptResponse> {
|
|
async prompt(params: PromptRequest): Promise<PromptResponse> {
|
|
|
assertOpen()
|
|
assertOpen()
|
|
|
const rec = requireSession(SessionId(params.sessionId))
|
|
const rec = requireSession(SessionId(params.sessionId))
|
|
|
- if (rec.inflight !== undefined) {
|
|
|
|
|
|
|
+ if (rec.inflight !== undefined || rec.commandAbort !== undefined) {
|
|
|
throw invalidParams('a prompt is already in flight for this session')
|
|
throw invalidParams('a prompt is already in flight for this session')
|
|
|
}
|
|
}
|
|
|
if (promptHasUnsupportedContent(params.prompt)) {
|
|
if (promptHasUnsupportedContent(params.prompt)) {
|
|
@@ -810,6 +898,52 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
// waiting for a settle that never comes.
|
|
// waiting for a settle that never comes.
|
|
|
throw invalidParams('empty prompt')
|
|
throw invalidParams('empty prompt')
|
|
|
}
|
|
}
|
|
|
|
|
+ // ACP command prompts may carry additional supported content blocks.
|
|
|
|
|
+ // The same lossless flattening used for model prompts supplies their
|
|
|
|
|
+ // unstructured command input; unsupported kinds were rejected above.
|
|
|
|
|
+ const commandLine = text.startsWith('/') ? text : undefined
|
|
|
|
|
+ if (commandLine !== undefined) {
|
|
|
|
|
+ const controller = new AbortController()
|
|
|
|
|
+ rec.commandAbort = controller
|
|
|
|
|
+ try {
|
|
|
|
|
+ const result = await commands.execute(rec.agent, commandLine, controller.signal)
|
|
|
|
|
+ if (result !== undefined && result.text !== undefined && result.text !== '') {
|
|
|
|
|
+ notify({
|
|
|
|
|
+ sessionId: rec.agent.session.id,
|
|
|
|
|
+ update: {
|
|
|
|
|
+ sessionUpdate: 'agent_message_chunk',
|
|
|
|
|
+ content: {
|
|
|
|
|
+ type: 'text',
|
|
|
|
|
+ text: result.kind === 'error' ? `Error: ${result.text}` : result.text,
|
|
|
|
|
+ },
|
|
|
|
|
+ },
|
|
|
|
|
+ })
|
|
|
|
|
+ } else if (result === undefined) {
|
|
|
|
|
+ notify({
|
|
|
|
|
+ sessionId: rec.agent.session.id,
|
|
|
|
|
+ update: {
|
|
|
|
|
+ sessionUpdate: 'agent_message_chunk',
|
|
|
|
|
+ content: { type: 'text', text: `Error: unknown command: ${commandLine}` },
|
|
|
|
|
+ },
|
|
|
|
|
+ })
|
|
|
|
|
+ }
|
|
|
|
|
+ return { stopReason: 'end_turn' }
|
|
|
|
|
+ } catch (error: unknown) {
|
|
|
|
|
+ if (controller.signal.aborted) return { stopReason: 'cancelled' }
|
|
|
|
|
+ const rendered = renderThrown(error)
|
|
|
|
|
+ logger.warn(`acp: command failed: ${rendered}`)
|
|
|
|
|
+ notify({
|
|
|
|
|
+ sessionId: rec.agent.session.id,
|
|
|
|
|
+ update: {
|
|
|
|
|
+ sessionUpdate: 'agent_message_chunk',
|
|
|
|
|
+ content: { type: 'text', text: `Error: command failed: ${rendered}` },
|
|
|
|
|
+ },
|
|
|
|
|
+ })
|
|
|
|
|
+ return { stopReason: 'end_turn' }
|
|
|
|
|
+ } finally {
|
|
|
|
|
+ rec.commandAbort = undefined
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
// Install the in-flight slot BEFORE send() (send does not synchronously
|
|
// Install the in-flight slot BEFORE send() (send does not synchronously
|
|
|
// flip status to running; the session/event listener records the turn
|
|
// flip status to running; the session/event listener records the turn
|
|
|
// number and settle/rejects it). Capture the log length now as the
|
|
// number and settle/rejects it). Capture the log length now as the
|
|
@@ -837,8 +971,12 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
// settle it, because cancel() may drop the turn before any turn/end is
|
|
// settle it, because cancel() may drop the turn before any turn/end is
|
|
|
// emitted, and removing this direct settle would move the RPC's
|
|
// emitted, and removing this direct settle would move the RPC's
|
|
|
// resolution onto a later observer path, changing its timing.
|
|
// resolution onto a later observer path, changing its timing.
|
|
|
- rec.agent.cancel('session/cancel')
|
|
|
|
|
- settlePrompt(rec, 'cancelled')
|
|
|
|
|
|
|
+ if (rec.commandAbort !== undefined) {
|
|
|
|
|
+ rec.commandAbort.abort(new Error('session/cancel'))
|
|
|
|
|
+ } else {
|
|
|
|
|
+ rec.agent.cancel('session/cancel')
|
|
|
|
|
+ settlePrompt(rec, 'cancelled')
|
|
|
|
|
+ }
|
|
|
return Promise.resolve()
|
|
return Promise.resolve()
|
|
|
},
|
|
},
|
|
|
|
|
|
|
@@ -908,7 +1046,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
Writable.toWeb(process.stdout) as WritableStream<Uint8Array>,
|
|
Writable.toWeb(process.stdout) as WritableStream<Uint8Array>,
|
|
|
Readable.toWeb(process.stdin) as ReadableStream<Uint8Array>,
|
|
Readable.toWeb(process.stdin) as ReadableStream<Uint8Array>,
|
|
|
)
|
|
)
|
|
|
- conn = new AgentSideConnection(makeAgent, stream)
|
|
|
|
|
|
|
+ conn = new AgentSideConnection(makeAgent, observeOutbound(stream, announceInitialCommands))
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* Tear ALL live sessions down to quiescence (docs/defensive-patterns.md "dispose must reach
|
|
* Tear ALL live sessions down to quiescence (docs/defensive-patterns.md "dispose must reach
|
|
@@ -946,12 +1084,14 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|
|
// installed yet) must observe this after its await and refuse to install a
|
|
// installed yet) must observe this after its await and refuse to install a
|
|
|
// post-teardown record. Set even when there are no live sessions.
|
|
// post-teardown record. Set even when there are no live sessions.
|
|
|
closed = true
|
|
closed = true
|
|
|
|
|
+ pendingCommandSnapshots.clear()
|
|
|
const recs = [...sessions.values()]
|
|
const recs = [...sessions.values()]
|
|
|
sessions.clear()
|
|
sessions.clear()
|
|
|
if (recs.length === 0) return Promise.resolve()
|
|
if (recs.length === 0) return Promise.resolve()
|
|
|
quiescing = (async () => {
|
|
quiescing = (async () => {
|
|
|
await Promise.all(recs.map(async (rec) => {
|
|
await Promise.all(recs.map(async (rec) => {
|
|
|
settlePrompt(rec, 'cancelled')
|
|
settlePrompt(rec, 'cancelled')
|
|
|
|
|
+ rec.commandAbort?.abort(new Error('ACP connection closed'))
|
|
|
// Per-agent dispose (the AgentHandle disposer): unregister this agent,
|
|
// Per-agent dispose (the AgentHandle disposer): unregister this agent,
|
|
|
// stop its loop (sets disposed + aborts the in-flight step), await
|
|
// stop its loop (sets disposed + aborts the in-flight step), await
|
|
|
// quiescence (the loop exit + final flush), and remove its session — so
|
|
// quiescence (the loop exit + final flush), and remove its session — so
|