Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/codex/enforce-tool-cancellation' into worktree/explicit-turn-signal

Browse Source 
# Conflicts:
#	docs/architecture.md
#	docs/cordis-catalog/events.md
#	docs/core-data-structures/core.md
#	docs/event-producer-consumer.md
#	packages/core/agent-loop/README.md
#	packages/core/agent-loop/src/agent.ts
#	packages/core/agent/README.md
#	packages/core/agent/src/types.ts
#	packages/ui/acp/src/index.ts
#	packages/ui/tui/src/index.ts
#	packages/ui/tui/tests/harness.ts
This commit is contained in:
Tianyi Cui
2026-07-21 12:48:46 +08:00
parent c6e1d35a99 43d50a38b1
commit d6bebc18f8
267 changed files with 13024 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
packages/ui/acp/src/index.ts
View File

@@ -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'
@@ -77,13 +80,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)
@@ -260,6 +300,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 }
}
@@ -274,6 +316,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
@@ -380,6 +423,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.
@@ -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). */
const settlePrompt = (rec: SessionRecord, reason: StopReason): void => {
const inflight = rec.inflight
@@ -674,15 +757,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 } : {} }
},
@@ -763,6 +849,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
terminalEnabled,
target,
inflight: undefined,
commandAbort: undefined,
pendingSwitches: {},
}
sessions.set(sessionId, record)
@@ -787,6 +874,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 {
@@ -797,7 +885,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)) {
@@ -810,6 +898,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
@@ -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
// emitted, and removing this direct settle would move the RPC's
// resolution onto a later observer path, changing its timing.
rec.agent.cancel({ kind: 'user' })
settlePrompt(rec, 'cancelled')
if (rec.commandAbort !== undefined) {
rec.commandAbort.abort(new Error('session/cancel'))
} else {
rec.agent.cancel({ kind: 'user' })
settlePrompt(rec, 'cancelled')
}
return Promise.resolve()
},
@@ -908,7 +1046,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
@@ -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
// 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