Merge remote-tracking branch 'origin/master' into codex/session-title
# Conflicts: # docs/architecture.md # docs/config-catalog.md # docs/event-producer-consumer.md # docs/module-graph.md # examples/acp-agent/tests/snapshots/advanced-toolchain/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/bash-spill/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/both-mode-turn/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/cancel-tool-calls/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/cancel/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/code-mode-turn/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/code-mode-workspace-context/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/error-finish/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/fs-edit/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/fs-policy-reject/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/fs-read-window/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/fs-read/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/fs-terminal-card/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/fs-write-overwrite/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/fs-write/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-cc-posttool-block/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-cc-posttool-context/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-cc-pretool-ask/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-cc-pretool-deny/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-cc-promptsubmit-context/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-cc-stop-continue/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-codex-posttool-block/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-codex-posttool-context/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-codex-pretool-block/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-codex-promptsubmit-context/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/hook-codex-stop-continue/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/model-switching/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/multi-turn/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/parallel-tool-calls/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/repeat-tool-guard/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/skill-load/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/subagent-depth-two-rejection/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/subagent-fork/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/subagent-mixed/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/subagent-multi/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/subagent-spawn/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/text-turn/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/todo-plan/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/tool-call-turn/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/workflow-run/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/workspace-context/stdout.expected.jsonl # examples/acp-agent/tests/snapshots/workspace-edit/stdout.expected.jsonl # examples/headless-agent/tests/headless.snapshot.ts # packages/examples/README.md # packages/examples/agent-spine-demo/README.md # packages/examples/agent-spine-demo/package.json # packages/ui/README.md # packages/ui/acp/README.md
This commit is contained in:
Tianyi Cui
@@ -17,7 +17,9 @@ import {
|
||||
PROTOCOL_VERSION,
|
||||
RequestError,
|
||||
type Agent as AcpAgent,
|
||||
type AnyMessage,
|
||||
type AuthenticateRequest,
|
||||
type AvailableCommand,
|
||||
type CancelNotification,
|
||||
type ContentBlock as AcpContentBlock,
|
||||
type CreateElicitationRequest,
|
||||
@@ -46,6 +48,7 @@ import type { ContentBlock, LlmCallConfig, LlmModelInfo, LlmProviderInfo } from
|
||||
import { assertNever, CallId } from '@deepseek-ai/dsh-llm'
|
||||
import type {} from '@deepseek-ai/dsh-llm-retry'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import type {} from '@deepseek-ai/dsh-commands'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
// Side-effect type import: resolves `ctx.get('permission')` to the service.
|
||||
import type {} from '@deepseek-ai/dsh-permission'
|
||||
@@ -79,13 +82,50 @@ import {
|
||||
|
||||
export const name = 'acp'
|
||||
// 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. */
|
||||
function invalidParams(detail: string): RequestError {
|
||||
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. */
|
||||
function internalError(detail: string): RequestError {
|
||||
return RequestError.internalError(undefined, detail)
|
||||
@@ -262,6 +302,8 @@ interface SessionRecord {
|
||||
reject: (error: Error) => void
|
||||
turn: number | 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. */
|
||||
pendingSwitches: { preset?: string }
|
||||
}
|
||||
@@ -276,6 +318,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
// ACP handlers execute outside this plugin's injection scope, so capture
|
||||
// injected services during apply(); lazy service reads in a handler fail.
|
||||
const agents = ctx.agents
|
||||
const commands = ctx.commands
|
||||
const llm = ctx.llm
|
||||
const sessionPersistence = ctx.sessionPersistence
|
||||
const logger = ctx.logger
|
||||
@@ -382,6 +425,9 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
const sessions = new Map<SessionId, SessionRecord>()
|
||||
// Reserve an id before resume so pipelined load/new requests cannot duplicate it.
|
||||
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.
|
||||
let closed = false
|
||||
// Each new or loaded session snapshots the latest connection capability.
|
||||
@@ -470,6 +516,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). */
|
||||
const settlePrompt = (rec: SessionRecord, reason: StopReason): void => {
|
||||
const inflight = rec.inflight
|
||||
@@ -676,15 +759,18 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
await handle.dispose()
|
||||
throw internalError('connection closed during session/new')
|
||||
}
|
||||
sessions.set(sessionId, {
|
||||
const record: SessionRecord = {
|
||||
agent: handle.agent,
|
||||
dispose: () => handle.dispose(),
|
||||
presenter: makePresenter(handle.agent),
|
||||
terminalEnabled: terminalOutputCap,
|
||||
target,
|
||||
inflight: undefined,
|
||||
commandAbort: undefined,
|
||||
pendingSwitches: {},
|
||||
})
|
||||
}
|
||||
sessions.set(sessionId, record)
|
||||
pendingCommandSnapshots.set(sessionId, record)
|
||||
const configOptions = configOptionsFor(handle.agent, directory)
|
||||
return { sessionId, ...configOptions.length > 0 ? { configOptions } : {} }
|
||||
},
|
||||
@@ -765,6 +851,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
terminalEnabled,
|
||||
target,
|
||||
inflight: undefined,
|
||||
commandAbort: undefined,
|
||||
pendingSwitches: {},
|
||||
}
|
||||
sessions.set(sessionId, record)
|
||||
@@ -789,6 +876,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
for (const event of agent.session.events) {
|
||||
streamSessionEventUpdate(sessionId, event, notify, replayPresenter, replayTerminal)
|
||||
}
|
||||
notifyCommands(record)
|
||||
const configOptions = configOptionsFor(agent, directory)
|
||||
return configOptions.length > 0 ? { configOptions } : {}
|
||||
} finally {
|
||||
@@ -799,7 +887,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
async prompt(params: PromptRequest): Promise<PromptResponse> {
|
||||
assertOpen()
|
||||
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')
|
||||
}
|
||||
if (promptHasUnsupportedContent(params.prompt)) {
|
||||
@@ -812,6 +900,52 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
// waiting for a settle that never comes.
|
||||
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
|
||||
// flip status to running; the session/event listener records the turn
|
||||
// number and settle/rejects it). Capture the log length now as the
|
||||
@@ -839,8 +973,12 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
// settle it, because cancel() may drop the turn before any turn/end is
|
||||
// emitted, and removing this direct settle would move the RPC's
|
||||
// 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()
|
||||
},
|
||||
|
||||
@@ -910,7 +1048,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
Writable.toWeb(process.stdout) as WritableStream<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
|
||||
@@ -948,12 +1086,14 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
// installed yet) must observe this after its await and refuse to install a
|
||||
// post-teardown record. Set even when there are no live sessions.
|
||||
closed = true
|
||||
pendingCommandSnapshots.clear()
|
||||
const recs = [...sessions.values()]
|
||||
sessions.clear()
|
||||
if (recs.length === 0) return Promise.resolve()
|
||||
quiescing = (async () => {
|
||||
await Promise.all(recs.map(async (rec) => {
|
||||
settlePrompt(rec, 'cancelled')
|
||||
rec.commandAbort?.abort(new Error('ACP connection closed'))
|
||||
// Per-agent dispose (the AgentHandle disposer): unregister this agent,
|
||||
// stop its loop (sets disposed + aborts the in-flight step), await
|
||||
// quiescence (the loop exit + final flush), and remove its session — so
|
||||
|
||||
Reference in New Issue
Block a user