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

Merge remote-tracking branch 'origin/master' into codex/agent-session-jsonl-location

Browse Source 
# Conflicts:
#	docs/module-graph.md
#	docs/rfc/INDEX.md
#	packages/core/agent-core/tests/agent-core.spec.ts
#	pnpm-lock.yaml
This commit is contained in:
Yichen Jiang
2026-07-13 16:18:44 +08:00
parent d4b5227071 b3e1a4154b
commit ba329b16ce
241 changed files with 14508 additions and 4772 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
packages/ui/acp-agent/package.json
View File

@@ -46,6 +46,7 @@
"@cordisjs/plugin-loader": "workspace:^",
"@deepseek-ai/dsh-app-boot": "workspace:^",
"@deepseek-ai/dsh-acp": "workspace:^",
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-agent-core": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",

8
packages/ui/acp-agent/tests/acp-agent.spec.ts
View File

@@ -4,6 +4,7 @@ import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import { agentEvents, type Agent } from '@deepseek-ai/dsh-agent'
import { TOOL_ORDER_REST } from '@deepseek-ai/dsh-system-prompt'
import type { Message } from '@deepseek-ai/dsh-llm'
import * as acpAgent from '../src/index.ts'
@@ -36,10 +37,11 @@ async function isolatedSkillsConfig(catalogDescriptionMaxLength?: number): Promi
}
async function composePrefix(ctx: Context): Promise<Message[]> {
const agent = { session: { header: { cwd: '/tmp' } } } as unknown as Agent
const empty: Message[] = []
return await ctx.waterfall(
'agent/session-prefix', { session: { header: { cwd: '/tmp' } } } as never,
empty, new AbortController().signal, () => Promise.resolve(empty),
return await agentEvents(ctx, agent).waterfall(
'agent/session-prefix', empty, new AbortController().signal,
() => Promise.resolve(empty),
)
}

3
packages/ui/acp-agent/tsconfig.json
View File

@@ -23,6 +23,9 @@
{
"path": "../acp"
},
{
"path": "../../core/agent"
},
{
"path": "../../core/agent-core"
},

6
packages/ui/acp/README.md
View File

@@ -16,7 +16,7 @@ It is a **client-driver / UI plugin**, the structured analogue of the readline `
|---|---|---|
| `model` | — | Model name for created agents (must have a registered adapter). |
(No persona key: the deployment persona is `dsh-system-prompt`'s own `persona` config — a context-wide section, so ACP-created agents render it without the bridge carrying prompt text.)
(No persona key: `dsh-system-prompt`'s own `persona` config supplies the global default section, so ACP-created agents render it without the bridge carrying prompt text. An agent-scoped same-name section may still shadow that default.)
The `initialize` handshake reports a fixed server identity (`agentInfo: { name: 'deepseek-harness-acp', version: '0.0.1' }`) — branding is a literal at the `initialize` site, not config.
@@ -36,7 +36,7 @@ The `initialize` handshake reports a fixed server identity (`agentInfo: { name:
## Multi-session
The bridge multiplexes N sessions over one connection. Live sessions are held in a `Map<sessionId, SessionRecord>` (forward) with a `WeakMap<Agent, sessionId>` reverse map so `agent/*` events — which carry only the `Agent` — demux in O(1). Every `session/event` and `agent/status` is routed strictly to its owning record, so concurrent sessions never cross-settle or interleave their `session/update` notifications. State is per session: one in-flight prompt each, `session/cancel` aborts and settles only its own agent/prompt, and disposal drains every live session in parallel to quiescence. Permission prompts follow the same ownership: the `approval/request` answerer resolves the owning session through the reverse map and prompts only there.
The bridge multiplexes N sessions over one connection. Live sessions are held in a `Map<sessionId, SessionRecord>` (forward) with a `WeakMap<Agent, sessionId>` reverse map so agent-scoped approval events demultiplex in O(1). Every `session/event` is routed strictly to its owning record, so concurrent sessions never cross-settle or interleave their `session/update` notifications. State is per session: one in-flight prompt each, `session/cancel` aborts and settles only its own agent/prompt, and disposal drains every live session in parallel to quiescence. Permission prompts follow the same ownership: the `approval/request` answerer resolves the owning session through the reverse map and prompts only there.
## Session config options
@@ -71,7 +71,7 @@ When the client does NOT advertise the capability, none of the `_meta`/terminal
## Settle-exactly-once
A `session/prompt` resolves (or rejects) exactly once, keyed off the canonical session log (the `session/event` stream). One listener captures the prompt's owning turn from the log's `turn/start` and settles on the matching `turn/end` — the durable boundary event (`closeTurn` appends it unconditionally; there is no `agent/*` turn mirror). A prompt settles only on ITS OWN turn (`inflight.turn === turn/end.turn`), so a stale `turn/end` for a previously-cancelled turn whose end arrives late can never settle the wrong prompt. A turn that ends `error` REJECTS the RPC with an internal error carrying the failure message (ACP has no error stop reason); every other reason resolves via the codec. As a fallback, when the agent settles to `idle`/`disposed` with a prompt still pending — e.g. a peer `session/event` listener registered before the bridge threw and starved the bridge's listener — an `agent/status` handler reconciles the prompt from the log (the owning turn's `turn/end`, or `cancelled` if the turn was torn down without one). An empty/whitespace prompt is rejected up front — it would queue no work, so no turn would start and the RPC would hang.
A `session/prompt` resolves or rejects exactly once from the canonical `session/event` stream. The listener captures the prompt's owning turn from `turn/start` and settles in a `finally` block when the matching `turn/end` is appended, so a presentation/streaming failure cannot strand the RPC after the durable terminal event exists. Correlation by turn id prevents a late end from a cancelled prompt from settling its successor. A turn ending in `error` rejects the RPC with an internal error carrying the failure message because ACP has no error stop reason; every other reason resolves through the codec. An empty or whitespace-only prompt is rejected before enqueue because it would start no turn and otherwise leave the RPC pending.
## Permission prompts

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

@@ -71,7 +71,7 @@ import {
} from '@agentclientprotocol/sdk'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import { assertNever, CallId } from '@deepseek-ai/dsh-llm'
import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { AgentId } from '@deepseek-ai/dsh-agent'
import { SessionId } from '@deepseek-ai/dsh-session'
import { SANDBOX_MODES, effectiveSandboxMode, setSandboxMode } from '@deepseek-ai/dsh-bash'
@@ -297,7 +297,8 @@ interface SessionRecord {
/**
* The in-flight `session/prompt`, or `undefined` when none is pending. A
* prompt resolves with a {@link StopReason} or rejects with an Error (a
* turn that ended in failure). Settled exactly once via {@link settlePrompt}.
* turn that ended in failure). Settled exactly once by its matching
* `turn/end`, direct cancellation, or teardown.
*
* `turn` is the loop turn number this prompt owns, captured from the log's
* `turn/start` after `send()`. Until then it is `undefined` (the turn has not
@@ -307,18 +308,11 @@ interface SessionRecord {
* wrong prompt. A direct cancel/dispose settle clears the whole in-flight slot,
* so a later stale `turn/end` finds no pending prompt.
*
* `logWatermark` is the session log length at the moment the prompt was
* installed (before `send()`). The settle-from-log fallback uses it to infer
* the owning `turn/start` from the canonical log even when the live
* `session/event` capture was starved (a peer listener that throws on
* `turn/start` — see `settleFromLog`): the prompt owns the FIRST `turn/start`
* appended at or after this watermark.
*/
inflight: {
resolve: (reason: StopReason) => void
reject: (error: Error) => void
turn: number | undefined
logWatermark: number
} | undefined
/**
* Config switches accepted while the session was IDLE, not yet anchored in
@@ -337,13 +331,9 @@ interface SessionRecord {
/**
* Drive the in-flight prompt's settle from the harness event stream. The bridge
* settles off the durable log: the `turn/end` session event on the
* `session/event` feed for the prompt's own turn, with the agent
* erroring/settling to idle as a fallback (docs/defensive-patterns.md "honor
* cross-seam contracts on BOTH sides") for the case where a throwing peer `session/event` listener
* starved the bridge's listener before it saw the boundary. The first of these
* to fire settles the prompt; `settle` is then cleared so the others are no-ops
* (settle-exactly-once).
* settles off the durable `turn/end` event for the prompt's own turn. Session
* contains post-commit observers independently, and this listener performs
* correlation in a `finally` so presentation failure cannot starve settlement.
*/
export function apply(ctx: Context, config: AcpConfig): void {
// Capture the injected services NOW, during apply(), while we are inside this
@@ -360,7 +350,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
const userInteraction = ctx.userInteraction
// A new ToolPresenter per session (and a throwaway per load replay), each given
// this warn sink so a throwing tool presenter is logged, not propagated.
const makePresenter = (): ToolPresenter => new ToolPresenter(tools, (message) => { logger.warn(message) })
const makePresenter = (agent?: Agent): ToolPresenter => new ToolPresenter(tools, (message) => { logger.warn(message) }, agent)
// Live sessions keyed by id (RFC 011 multi-session), plus an agent→sessionId
// reverse map so `agent/*` events (which carry only the Agent) demux in O(1).
@@ -501,83 +491,24 @@ export function apply(ctx: Context, config: AcpConfig): void {
ctx.on('session/event', (session, event: SessionEvent) => {
const rec = sessions.get(session.header.id)
if (rec === undefined) return
streamSessionEventUpdate(rec.sessionId, event, notify, rec.presenter, {
enabled: rec.terminalEnabled,
cwd: session.header.cwd,
}, { includeUserMessages: false })
const inflight = rec.inflight
if (inflight === undefined) return
if (event.type === 'turn/start') {
// Tag the in-flight prompt with its owning turn — but ONLY a
// `message`-triggered turn (the kind a `send()` prompt produces). A turn
// a plugin opens between prompt-install and the prompt's own turn (an idle
// `agent.inject()` writes a one-shot `injection`-triggered turn) must NOT
// be mistaken for the prompt's turn, or its turn/end would settle the RPC
// early. The first message turn at/after install owns the prompt
// (`turn === undefined` guard); the loop batches queued messages into one
// turn, so there is exactly one.
if (inflight.turn === undefined && event.data.trigger.kind === 'message') {
inflight.turn = event.data.turn
try {
streamSessionEventUpdate(rec.sessionId, event, notify, rec.presenter, {
enabled: rec.terminalEnabled,
cwd: session.header.cwd,
}, { includeUserMessages: false })
} finally {
const inflight = rec.inflight
if (inflight !== undefined && event.type === 'turn/start') {
// The first message-triggered turn after prompt installation owns the
// prompt; injection-triggered turns must not settle it early.
if (inflight.turn === undefined && event.data.trigger.kind === 'message') {
inflight.turn = event.data.turn
}
} else if (inflight !== undefined && event.type === 'turn/end' && inflight.turn === event.data.turn) {
rec.inflight = undefined
settleFromTurnEnd(inflight, event.data.reason)
}
return
}
// Settle only on the OWNING turn's end.
if (event.type !== 'turn/end' || inflight.turn !== event.data.turn) return
rec.inflight = undefined
settleFromTurnEnd(inflight, event.data.reason)
})
// Settle fallback: a `session/event` listener registered BEFORE ACP that
// throws (on `turn/start` OR `turn/end`) would, via cordis `emit`'s
// stop-on-throw, starve ACP's listener above — the prompt would hang or, if
// only the turn number was missed, settle as the wrong outcome. So when the
// agent settles to `idle` (or is disposed), reconcile against the canonical
// log: determine the prompt's owning turn (the captured `turn`, or — if the
// live capture was starved — the FIRST `turn/start` appended at/after the
// install-time `logWatermark`), then settle from that turn's `turn/end`
// (reject on error, resolve via codec), or `cancelled` if no owning turn ever
// started. Never double-settles — clears `inflight` first.
const settleFromLog = (rec: SessionRecord): void => {
const inflight = rec.inflight
if (inflight === undefined) return
const events = rec.agent.session.events
// The owning turn number: the captured one, or — if the live capture was
// starved — inferred from the log as the first MESSAGE-triggered turn opened
// at/after the watermark. The message-trigger filter matches the live
// capture: a one-shot `injection` turn a plugin may open between
// prompt-install and the prompt's turn is NOT the prompt's turn. Undefined
// only if no message turn ever started for this prompt.
const owningTurn = inflight.turn ?? events.slice(inflight.logWatermark).find(
(e): e is Extract<SessionEvent, { type: 'turn/start' }> =>
e.type === 'turn/start' && e.data.trigger.kind === 'message',
)?.data.turn
// The owning turn's end in the log. If `owningTurn` is undefined (no turn
// ever started for this prompt — a torn-down-before-turn case that quiesce's
// direct settle normally pre-empts), no `turn/end` matches (turn numbers are
// >= 1) and `findLast` returns undefined, falling through to cancelled.
const end = events.findLast(
(e): e is Extract<SessionEvent, { type: 'turn/end' }> =>
e.type === 'turn/end' && e.data.turn === owningTurn,
)
rec.inflight = undefined
if (end === undefined) {
// No owning turn / no clean turn/end (torn down mid-turn) → cancelled.
inflight.resolve('cancelled')
return
}
settleFromTurnEnd(inflight, end.data.reason)
}
// On a settle to idle/disposed, reconcile any still-pending prompt from the
// log (covers a starved `session/event` listener — see settleFromLog). A mid-
// step disposal that never appended a clean turn/end resolves `cancelled`.
// Demux via the agent→sessionId reverse map.
ctx.on('agent/status', (agent, status: AgentStatus) => {
const sessionId = bySession.get(agent)
if (sessionId === undefined) return
const rec = sessions.get(sessionId)
if (rec === undefined) return
if (status === 'idle' || status === 'disposed') settleFromLog(rec)
})
// --- Approval answerer -----------------------------------------------------
@@ -744,29 +675,39 @@ export function apply(ctx: Context, config: AcpConfig): void {
return Promise.resolve()
},
newSession(params: NewSessionRequest): Promise<NewSessionResponse> {
async newSession(params: NewSessionRequest): Promise<NewSessionResponse> {
assertOpen()
validateWorkspaceParams(params)
validateMcpServers(params)
const sessionId = SessionId(randomUUID())
const handle = agents.create({
const handle = await agents.create({
agentId: AgentId(sessionId),
sessionId,
meta: { cwd: params.cwd },
agentOptions: agentOptions(config),
})
// Creation is now asynchronous because it awaits the unpublished setup
// transaction. A client disconnect can therefore close this bridge
// after the entry check but before the handle resolves; never install a
// post-close record that quiesce() could not have seen.
/* v8 ignore next 4 -- the in-memory transport rejects the in-flight RPC
immediately on close; real stdio may let the handler resume */
if (closed) {
await handle.dispose()
throw internalError('connection closed during session/new')
}
bySession.set(handle.agent, sessionId)
sessions.set(sessionId, {
sessionId,
agent: handle.agent,
dispose: () => handle.dispose(),
presenter: makePresenter(),
presenter: makePresenter(handle.agent),
terminalEnabled: terminalOutputCap,
inflight: undefined,
pendingSwitches: {},
})
const configOptions = configOptionsFor(handle.agent)
return Promise.resolve({ sessionId, ...configOptions.length > 0 ? { configOptions } : {} })
return { sessionId, ...configOptions.length > 0 ? { configOptions } : {} }
},
async loadSession(params: LoadSessionRequest): Promise<LoadSessionResponse> {
@@ -839,7 +780,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
sessionId,
agent,
dispose: () => handle.dispose(),
presenter: makePresenter(),
presenter: makePresenter(agent),
terminalEnabled,
inflight: undefined,
pendingSwitches: {},
@@ -858,7 +799,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
// future live events for this session. The throwaway pairs call→result
// as the log replays in order (same as live) and is discarded after,
// so the record's presenter starts clean for the post-load live stream.
const replayPresenter = makePresenter()
const replayPresenter = makePresenter(agent)
const replayTerminal: TerminalRendering = {
enabled: terminalEnabled,
cwd: agent.session.header.cwd,
@@ -892,12 +833,10 @@ export function apply(ctx: Context, config: AcpConfig): void {
// 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
// watermark: the settle-from-log fallback infers the owning turn/start
// as the first one appended at/after it, surviving a starved live
// capture. A turn that ends in error rejects this promise (the codec
// never produces an error stop reason).
// A turn that ends in error rejects this promise (the codec never
// produces an error stop reason).
const stopReason = await new Promise<StopReason>((resolve, reject) => {
rec.inflight = { resolve, reject, turn: undefined, logWatermark: rec.agent.session.events.length }
rec.inflight = { resolve, reject, turn: undefined }
rec.agent.send([{ type: 'text', text }])
})
return { stopReason }
@@ -916,8 +855,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
// as cancelled directly here: do NOT rely on the resulting turn/end to
// 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 the settleFromLog/agent-status path, changing its
// timing.
// resolution onto a later observer path, changing its timing.
rec.agent.cancel('session/cancel')
settlePrompt(rec, 'cancelled')
return Promise.resolve()
@@ -1000,15 +938,15 @@ export function apply(ctx: Context, config: AcpConfig): void {
* quiescence"): for each session settle any pending prompt `cancelled`, then
* run that session's {@link AgentHandle} `dispose()` — which stops the loop
* (sets `disposed`, aborts the in-flight step), AWAITS the loop's exit (the
* final `turn/end` + `session/flush` are captured while `onAppend` is still
* final `turn/end` + `session/flush` are captured while the store-owned publication hooks are still
* attached), unregisters the agent, and removes its session from the store.
* The per-session disposes run in parallel. Idempotent — clears the `sessions`
* map first and memoizes, so a second call (close racing dispose) is a no-op.
* Shared by Cordis disposal AND client disconnect (`conn.closed`).
*
* Per-agent disposal closes the former pre-step best-effort window — but via
* the DISPOSED path, not `cancel()`: the start-disposer resolves `handle.disposed`,
* which wakes the parked loop, and `isDisposed()` breaks the loop before a
* Per-agent disposal closes the queued-before-run window through the DISPOSED
* path, not `cancel()`: the start-disposer resolves `handle.disposed`, which
* wakes the parked loop, and `isDisposed()` breaks the loop before a
* queued-but-not-yet-running turn can start (a turn cut off mid-flight ends
* with reason `disposed`, not `aborted`). A bare client disconnect (resolves
* `conn.closed` WITHOUT disposing the fiber) thus leaves NO registered agent
@@ -1268,6 +1206,13 @@ export class ToolPresenter {
constructor(
private readonly tools: Pick<ToolRegistry, 'get'>,
private readonly onError: (message: string) => void = () => {},
/**
* The agent whose view resolves tool presentations: a scoped/shadowed
* tool presents with ITS OWN presentCall/presentResult — the same
* definition that executed — not a same-named global's. Absent (a replay
* with no live agent) the global view presents.
*/
private readonly agent?: Agent,
) {}
/**
@@ -1284,7 +1229,7 @@ export class ToolPresenter {
const args = parseToolArguments(argsJson)
let present: ToolCallView | undefined
try {
present = this.tools.get(name)?.presentCall?.(args)
present = this.tools.get(name, this.agent)?.presentCall?.(args)
} catch (error: unknown) {
// A throwing presentCall must not break streaming: log and fall back.
this.onError(`acp: tool "${name}" presentCall threw, using generic presentation: ${String(error)}`)
@@ -1318,7 +1263,8 @@ export class ToolPresenter {
if (call === undefined) return { card: 'generic', content }
let present: ToolResultView | undefined
try {
present = this.tools.get(call.name)?.presentResult?.(call.args, { content, isError, ...meta !== undefined ? { meta } : {} })
present = this.tools.get(call.name, this.agent)
?.presentResult?.(call.args, { content, isError, ...meta !== undefined ? { meta } : {} })
} catch (error: unknown) {
// A throwing presentResult must not break streaming/replay: log + fall back.
this.onError(`acp: tool "${call.name}" presentResult threw, using raw result: ${String(error)}`)

16
packages/ui/acp/tests/dispose.spec.ts
View File

@@ -159,8 +159,8 @@ describe('acp bridge — disposal & HMR safety', () => {
it('the final turn closing events are persisted across an AgentHandle dispose (durability)', async () => {
// The teardown-ORDER guarantee: a per-agent dispose must stop the loop,
// AWAIT its exit (so the loop's final `turn/end` + `session/flush` fire
// through the still-attached `session.onAppend` → `session/event`), and only
// THEN detach onAppend + remove the session. If the order were inverted
// through the still-attached store observer → `session/event`), and only
// THEN remove its publication hooks and session entry. If the order were inverted
// (detach first), the closing events would never reach persistence. Drive a
// CLEAN turn to completion, dispose JUST the bridge, then re-load the
// persisted log from disk and assert the closing turn/end is on disk — the
@@ -190,7 +190,7 @@ describe('acp bridge — disposal & HMR safety', () => {
// produced BY the dispose itself. Here the model stream HANGS, so the turn is
// still open when teardown runs: the composite agent effect stops the loop,
// the loop unwinds and appends `turn/end {disposed}` + runs its final
// `session/flush` — all while `onAppend` is still attached (the session
// `session/flush` — all while the store-owned publication hooks are still attached (the session
// detach is the LAST disposer in the same effect's LIFO chain) — and only
// THEN is the session detached. If the order were inverted (or the session
// were a racing SIBLING effect), the abort-produced `turn/end` would never
@@ -229,10 +229,10 @@ describe('acp bridge — disposal & HMR safety', () => {
// dispose one handle, and assert the other survives, registered and
// queryable, with its session still in the store.
const harness = await makeBridgeHarness({ storageDir, script: [] })
const handleA = harness.ctx.agents.create({
const handleA = await harness.ctx.agents.create({
agentId: AgentId('sib-a'), sessionId: SessionId('sib-a'), agentOptions: { model: 'mock' },
})
const handleB = harness.ctx.agents.create({
const handleB = await harness.ctx.agents.create({
agentId: AgentId('sib-b'), sessionId: SessionId('sib-b'), agentOptions: { model: 'mock' },
})
expect(harness.ctx.agents.get(AgentId('sib-a'))).toBe(handleA.agent)
@@ -255,13 +255,13 @@ describe('acp bridge — disposal & HMR safety', () => {
// into ONE composite effect whose disposers run as a `.then()` chain. The
// register disposer emits `agent/disposed`; if a listener throws and the
// emit is UNCONTAINED, the rejected chain skips the LATER session-detach
// disposer — stranding the session in the store with `onAppend` attached (a
// disposer — stranding the session in the store with its publication hooks attached (a
// leak AND a durability hole, since the new design relies on detach
// running). The emit must be contained. Register a throwing listener, drive
// a clean turn, dispose, and assert the session was STILL removed.
const harness = await makeBridgeHarness({ storageDir, script: [textResponse('ok')] })
harness.ctx.on('agent/disposed', () => { throw new Error('boom disposed listener') })
const handle = harness.ctx.agents.create({
const handle = await harness.ctx.agents.create({
agentId: AgentId('guard-a'), sessionId: SessionId('guard-a'), agentOptions: { model: 'mock' },
})
handle.agent.send([{ type: 'text', text: 'go' }])
@@ -282,7 +282,7 @@ describe('acp bridge — disposal & HMR safety', () => {
// first call's await agent.done + final flush finished. Every caller must
// observe the same quiescence boundary.
const harness = await makeBridgeHarness({ storageDir, script: ['hang'] })
const handle = harness.ctx.agents.create({
const handle = await harness.ctx.agents.create({
agentId: AgentId('conc-a'), sessionId: SessionId('conc-a'), agentOptions: { model: 'mock' },
})
// Drive a turn that hangs in the model stream, so the loop is mid-turn when

4
packages/ui/acp/tests/edges.spec.ts
View File

@@ -20,14 +20,14 @@ describe('acp bridge — demux & config edges', () => {
it('ignores events from an agent the bridge does not own (strict id demux)', async () => {
// A second agent created directly on the registry (NOT via the bridge) runs
// a turn. Its session/event + agent/status must NOT produce ACP updates and
// a turn. Its session events must NOT produce ACP updates and
// must not settle anything — the bridge demuxes strictly by its own id.
harness = await makeBridgeHarness({ storageDir, script: [textResponse('foreign')] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
const before = harness.updates.length
const { agent: foreign } = harness.ctx.agents.create({ agentId: AgentId('foreign'), sessionId: SessionId('foreign-session'), agentOptions: { model: 'mock' } })
const { agent: foreign } = await harness.ctx.agents.create({ agentId: AgentId('foreign'), sessionId: SessionId('foreign-session'), agentOptions: { model: 'mock' } })
foreign.send([{ type: 'text', text: 'hi' }])
await foreign.whenIdle()
await new Promise(r => setTimeout(r, 10))

30
packages/ui/acp/tests/turns.spec.ts
View File

@@ -235,12 +235,9 @@ describe('acp bridge — turn outcomes', () => {
expect(failed).toHaveLength(1)
})
it('settles via the log fallback when a prior session/event listener throws (starvation)', async () => {
// A peer session/event listener that runs BEFORE the bridge's listener
// throws on turn/end (prepend: true puts it first). cordis emit stops at the
// throw, so the bridge's session/event listener never sees turn/end and
// cannot settle there. The agent/status idle-fallback must reconcile the
// prompt from the log so the RPC settles instead of hanging.
it('settles successfully when an earlier turn/end observer throws', async () => {
// Session contains each post-commit observer failure, so a prepended peer
// cannot starve the bridge's live turn/end delivery.
harness = await makeBridgeHarness({ storageDir, script: [textResponse('answer')] })
harness.ctx.on('session/event', (_s, event) => {
if (event.type === 'turn/end') throw new Error('peer listener boom')
@@ -250,9 +247,7 @@ describe('acp bridge — turn outcomes', () => {
expect(res.stopReason).toBe('end_turn')
})
it('log fallback REJECTS when the starved turn ended in error', async () => {
// Same starvation as above, but the turn fails: the idle-fallback must
// reject the RPC from the logged turn/end{error}, not resolve.
it('still rejects a failed turn when an earlier turn/end observer throws', async () => {
harness = await makeBridgeHarness({ storageDir, script: [errorResponse('starved boom')] })
harness.ctx.on('session/event', (_s, event) => {
if (event.type === 'turn/end') throw new Error('peer listener boom')
@@ -262,21 +257,18 @@ describe('acp bridge — turn outcomes', () => {
.rejects.toThrow(/turn failed: starved boom/)
})
it('log fallback infers the owning turn when turn/START capture is starved', async () => {
// A peer listener throws on turn/START (not turn/end): the bridge never
it('captures and settles the owning turn when an earlier turn-start observer throws', async () => {
// Turn correlation still reaches the bridge after the throwing peer and
// captures inflight.turn via the live stream. A throwing turn/start listener
// also FAILS the turn (the throw is recorded as the turn's error). Without
// the watermark inference the fallback would resolve `cancelled` (the bug);
// with it, it infers the owning turn from the log and REJECTS from that
// turn's error turn/end. (The model's own error is never reached — the turn
// failed at start — so the rejection carries the listener's failure.)
harness = await makeBridgeHarness({ storageDir, script: [textResponse('never runs')] })
// Session contains post-commit callbacks independently.
// The model request and normal turn outcome therefore still occur.
harness = await makeBridgeHarness({ storageDir, script: [textResponse('answer')] })
harness.ctx.on('session/event', (_s, event) => {
if (event.type === 'turn/start') throw new Error('peer listener boom on start')
}, { prepend: true })
const sessionId = await newSession(harness)
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
.rejects.toThrow(/turn failed:/)
const result = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
expect(result.stopReason).toBe('end_turn')
})
it('a between-turn injection does not settle the prompt early (message-trigger correlation)', async () => {

9
packages/ui/stdio-agent/tests/stdio-agent.spec.ts
View File

@@ -4,7 +4,7 @@ import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import { AgentId } from '@deepseek-ai/dsh-agent'
import { AgentId, agentEvents, type Agent } from '@deepseek-ai/dsh-agent'
import type { Message } from '@deepseek-ai/dsh-llm'
import { TOOL_ORDER_REST } from '@deepseek-ai/dsh-system-prompt'
import * as stdioAgent from '../src/index.ts'
@@ -43,10 +43,11 @@ async function isolatedSkillsConfig(catalogDescriptionMaxLength?: number): Promi
}
async function composePrefix(ctx: Context): Promise<Message[]> {
const agent = { session: { header: { cwd: '/tmp' } } } as unknown as Agent
const empty: Message[] = []
return await ctx.waterfall(
'agent/session-prefix', { session: { header: { cwd: '/tmp' } } } as never,
empty, new AbortController().signal, () => Promise.resolve(empty),
return await agentEvents(ctx, agent).waterfall(
'agent/session-prefix', empty, new AbortController().signal,
() => Promise.resolve(empty),
)
}

6
packages/ui/user-approval/README.md
View File

@@ -2,11 +2,11 @@
User-approval seam. Owns the `ctx.approval` service ([`ApprovalService`](src/index.ts)) and the one-shot permission vocabulary the harness shares: `ApprovalRequest` (agent + tool identity + reason + abort signal), the closed `ApprovalOutcome` union (`allowed-once` / `rejected` / `cancelled` / `unavailable`), the `ApprovalRequestId` brand pairing the two log-only audit events (`approval/asked` / `approval/decided`), and the `approval/request` waterfall the answerers listen on. It lives in the UI group because its purpose is human permission, while remaining channel-neutral: it depends only on Cordis and core vocabulary packages, never on a concrete UI.
The contract in one line: `ctx.approval.request(req)` puts exactly one question — "may this specific action proceed?" — to whatever answerers the deployment composed, and always resolves to an outcome, never rejects: an aborted signal yields `cancelled`, a throwing or missing answerer yields `unavailable`, and `allowed-once` is a grant for the single asked-about action, never a class of future ones. The one precondition: ask from inside an open turn — the audit pair is turn-enclosed by contract (the turn is the durable log's commit/replay boundary; a bare event between turns is crash-tail garbage on reload), so an idle ask throws before appending anything.
The contract in one line: `ctx.approval.request(req)` puts exactly one question — "may this specific action proceed?" — to whatever answerers the deployment composed, and its answerer phase always produces an outcome: an aborted signal yields `cancelled`, a throwing or missing answerer yields `unavailable`, and `allowed-once` is a grant for the single asked-about action, never a class of future ones. `ApprovalRequest` is a readonly same-process contract: the service borrows the exact request, agent, session, and abort signal rather than cloning or freezing them. The request requires an open turn because the audit pair is turn-enclosed by contract (the turn is the durable log's commit/replay boundary; a bare event between turns is crash-tail garbage on reload), so an idle ask rejects before appending. Either audit append may reject before commit because returning an unlogged decision would violate the pair. Session contains post-commit observer failures, so an authoritative audit append cannot reject the request or suppress its matching event.
The service is the mechanism, answerers are the policy. Answerers are `approval/request` waterfall listeners occupying a single decision slot: answer for an agent you own by returning an outcome without calling `next()`, or delegate an agent you don't recognize by calling `next()` — the chain's built-in default is `unavailable`, so a deployment with no answerer (headless, CI) fails closed with zero configuration. Registration order across sibling plugins is not load-order deterministic; compose one terminal answerer per deployment and use `prepend` listeners only for decide-or-delegate gates.
The service is the mechanism, answerers are the policy. Answerers are `approval/request` waterfall listeners occupying a single decision slot: answer for an agent you own by returning an outcome without calling `next()`, or delegate an agent you don't recognize by calling `next()` — the chain's built-in default is `unavailable`, so a deployment with no answerer (headless, CI) fails closed with zero configuration. Dispatch is keyed by `req.agent`: a listener registered through `agent.ctx` receives only that agent's questions, while a plain-context listener receives every agent's. Registration order across sibling plugins is not load-order deterministic; compose one terminal answerer per deployment and use `prepend` listeners only for decide-or-delegate gates.
The seam also owns the per-session POLICY tier ([the sandbox RFC § Per-session mode switching](../../../docs/rfc/implemented/feature/2026-07-06-sandbox.md)): `ApprovalPolicy` is `'ask'` (delegate to the answerers) or `'never'` (deterministically reject without prompting anyone; the strict CI/unattended stance), with `effective = fold(the session's 'approval/policy' events, last one wins) ?? Config.policy` — the session log is the store, written only through `setApprovalPolicy(session, policy)`. The service decides `'never'` inside `request()` itself, before dispatching the waterfall (`'never'` → `'rejected'` with the audit pair still landing; no listener registration, including a later `prepend`, can precede it), states `'never'` — and only `'never'` in prose — in a per-agent prompt section, records either value with a source-owned header marker, and narrates a policy switch to the model in at most one coalesced `agent/pre-step` notice. The restart fallback reads the marker rather than deployment-controlled persona prose; attribution is positional (an override event after the last `request/header*` reads `changed by the user`, otherwise `changed by the operator/config`).
The seam also owns the per-session POLICY tier ([the sandbox RFC § Per-session mode switching](../../../docs/rfc/implemented/feature/2026-07-06-sandbox.md)): `ApprovalPolicy` is `'ask'` (delegate to the answerers) or `'never'` (deterministically reject without prompting anyone; the strict CI/unattended stance), with `effective = fold(the session's 'approval/policy' events, last one wins) ?? Config.policy` — the session log is the store, written only through `setApprovalPolicy(session, policy)`, which rejects any value outside that closed vocabulary before appending. The service decides `'never'` inside `request()` itself, before dispatching the waterfall (`'never'` → `'rejected'` with the audit pair still landing; no listener registration, including a later `prepend`, can precede it), states `'never'` — and only `'never'` in prose — in a per-agent prompt section, records either value with a source-owned header marker, and narrates a policy switch to the model in at most one coalesced `agent/pre-step` notice. The restart fallback reads the marker rather than deployment-controlled persona prose; attribution is positional (an override event after the last `request/header*` reads `changed by the user`, otherwise `changed by the operator/config`).
One seam serves both ask paths of [the sandbox RFC](../../../docs/rfc/implemented/feature/2026-07-06-sandbox.md): the `tools/pre-execute` `ask` decision (routed by [`@deepseek-ai/dsh-tools`](../../core/tools/) when this service is mounted; degrading to deny when it is not), and the sandbox post-denial escalated retry (the bash tool's `sandbox_permissions` gate in [`@deepseek-ai/dsh-tool-bash`](../../bash/tool-bash/) — [the sandbox RFC § Escalation](../../../docs/rfc/implemented/feature/2026-07-06-sandbox.md)). The full design: [the approval-seam RFC](../../../docs/rfc/implemented/feature/2026-07-06-approval-seam.md).

2
packages/ui/user-approval/package.json
View File

@@ -25,6 +25,7 @@
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-brand": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-scope": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
"cordis": "^4.0.0-rc.6"
@@ -36,6 +37,7 @@
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-scope": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"cordis": "^4.0.0-rc.6"

101
packages/ui/user-approval/src/index.ts
View File

@@ -39,6 +39,8 @@ import z from 'schemastery'
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { Agent } from '@deepseek-ai/dsh-agent'
import type { CallId } from '@deepseek-ai/dsh-llm'
import { scopeTarget } from '@deepseek-ai/dsh-scope'
import type { Scoped } from '@deepseek-ai/dsh-scope'
import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-system-prompt'
@@ -58,10 +60,14 @@ declare module 'cordis' {
* not recognize the agent MUST call `next()` so another answerer — or the
* fail-closed default `'unavailable'` — gets the question. Throwing is
* contained by the service and yields `'unavailable'`.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`) keys the carrier by `req.agent`: a
* listener registered through `agent.ctx` receives only that agent's
* questions, while a plain-context listener receives every agent's.
* `req` is a readonly same-process value borrowed from the caller.
* @param req - the pending decision (agent, tool identity, reason, signal).
* @mode waterfall
*/
'approval/request'(this: ApprovalService, req: ApprovalRequest, next: () => Promise<ApprovalOutcome>): Promise<ApprovalOutcome>
'approval/request'(this: Scoped<ApprovalService>, req: ApprovalRequest, next: () => Promise<ApprovalOutcome>): Promise<ApprovalOutcome>
}
}
@@ -215,12 +221,16 @@ function hasOpenTurn(events: readonly SessionEvent[]): boolean {
* THE write path for a session's approval-policy override: appends exactly
* one `approval/policy` event — the switch IS its event; nothing mutates
* policy state out of band. Takes effect on the session's next ask and next
* prompt assembly (the consumers fold on every read).
* prompt assembly (the consumers fold on every read). Rejects a value outside
* {@link APPROVAL_POLICIES} before appending anything.
* @param session - the session the override belongs to.
* @param policy - the policy every subsequent ask for this session resolves
* under (until the next switch).
*/
export function setApprovalPolicy(session: Session, policy: ApprovalPolicy): void {
if (!APPROVAL_POLICIES.includes(policy)) {
throw new TypeError('approval policy must be one of "ask" or "never"')
}
session.append('approval/policy', { policy })
}
@@ -229,7 +239,9 @@ export function setApprovalPolicy(session: Session, policy: ApprovalPolicy): voi
* for an answerer to present it and for the audit events to reconstruct what
* was asked — it deliberately does NOT carry tool arguments: a UI answerer
* attaches the prompt to the already-streamed tool call via `callId` instead
* of re-rendering the call.
* of re-rendering the call. This is a readonly same-process contract:
* `request()` borrows the request and its `agent` and `signal` capabilities
* directly rather than treating them as serialized input.
*/
export interface ApprovalRequest {
/**
@@ -237,21 +249,21 @@ export interface ApprovalRequest {
* UI answerer only answers for agents it owns) and receives the audit
* events on its session log.
*/
agent: Agent
readonly agent: Agent
/** The tool the question is about (presentation and audit). */
toolName: string
readonly toolName: string
/**
* The exact tool call being decided, when the asker has one — lets a UI
* attach the prompt to the tool call it already streamed.
*/
callId?: CallId
readonly callId?: CallId
/** The asker's human-readable explanation of WHY it is asking. */
reason?: string
readonly reason?: string
/**
* Aborting withdraws the question: the request settles `'cancelled'`
* immediately and a late answer from a still-pending answerer is discarded.
*/
signal?: AbortSignal
readonly signal?: AbortSignal
}
/** Plugin config. All optional — `static Config` supplies the defaults. */
@@ -262,7 +274,7 @@ export interface Config {
* (fail-closed with none); `'never'` auto-rejects every ask without
* prompting (the deterministic CI/unattended stance).
*/
policy?: ApprovalPolicy
readonly policy?: ApprovalPolicy
}
/**
@@ -287,7 +299,7 @@ export class ApprovalService extends Service {
constructor(ctx: Context, public config: Config) {
super(ctx, 'approval')
const effective = (agent: Agent): ApprovalPolicy => this.effectivePolicy(agent)
const effective = (agent: Agent): ApprovalPolicy => this.effectivePolicy(agent.session)
// Visibility layer 1, scoped on the prompt registry so headless
// compositions mount the seam without it: state the one deterministic
@@ -335,7 +347,7 @@ export class ApprovalService extends Service {
}
// Same fold effectivePolicy performs — override is scanned here anyway
// for POSITIONAL attribution; the default lives once, in the method.
const current = this.effectivePolicy(agent)
const current = this.effectivePolicy(session)
const header = session.requestHeader()
const told = narrated.get(session) ?? toldApprovalPolicy(header?.system)
narrated.set(session, current)
@@ -351,22 +363,26 @@ export class ApprovalService extends Service {
}
/**
* Ask the composed answerers to decide one request. Requires an open turn
* on the requesting agent's session — the audit pair below is turn-enclosed
* by contract (the turn is the log's commit/replay boundary; an idle append
* would be dropped as crash tail on reload) — and throws before appending
* anything when called idle; asking outside a turn is a deferred design.
* Within that precondition it always resolves to an outcome, never rejects:
* an aborted signal yields `'cancelled'`, a missing or throwing answerer
* yields `'unavailable'` (fail closed), and a rogue non-vocabulary return
* value is normalized to `'unavailable'`. Appends the
* `approval/asked`/`approval/decided` audit pair (log-only) around the
* decision regardless of outcome.
* Ask the composed answerers to decide one readonly same-process request.
* The service borrows the request, agent, session, and live signal directly.
* The request requires an open turn because the audit pair must be enclosed
* by the durable log's commit/replay boundary; an idle ask rejects before
* appending anything. The answerer phase always produces an outcome: an
* aborted signal yields `'cancelled'`, a missing or throwing answerer yields
* `'unavailable'` (fail closed), and a rogue non-vocabulary return value is
* normalized to `'unavailable'`. A failure that prevents either audit append
* from committing still rejects because returning an unlogged decision would
* violate the pair. Session contains post-commit observer failures, so an
* authoritative append cannot reject the request or suppress its matching
* audit event.
* @param req - the pending decision (agent, tool identity, reason, signal).
* @returns the closed outcome; `'allowed-once'` is the only grant.
* @throws when no turn is open or either audit event fails before the session
* append commit point.
*/
async request(req: ApprovalRequest): Promise<ApprovalOutcome> {
if (!hasOpenTurn(req.agent.session.events)) {
const session = req.agent.session
if (!hasOpenTurn(session.events)) {
throw new Error(
'approval.request() outside an open turn: the approval/asked + approval/decided audit pair '
+ 'must be turn-enclosed (a bare event between turns is crash-tail garbage on reload). '
@@ -374,14 +390,14 @@ export class ApprovalService extends Service {
)
}
const id = ApprovalRequestId(randomUUID())
req.agent.session.append('approval/asked', {
session.append('approval/asked', {
id,
toolName: req.toolName,
...req.callId !== undefined ? { callId: req.callId } : {},
...req.reason !== undefined ? { reason: req.reason } : {},
})
const outcome = await this.decide(req)
req.agent.session.append('approval/decided', { id, outcome })
const outcome = await this.decide(req, session)
session.append('approval/decided', { id, outcome })
return outcome
}
@@ -389,28 +405,37 @@ export class ApprovalService extends Service {
* The session's effective policy: its own `approval/policy` fold, else the
* configured default (the schema already defaulted an omitted policy to
* `'ask'`; the `??` only narrows the optional-input TYPE).
* @param agent - the agent whose session's policy applies.
* @returns the policy every ask for this agent resolves under right now.
* @param session - the exact accepted session whose policy applies.
* @returns the policy every ask for this session resolves under right now.
*/
private effectivePolicy(agent: Agent): ApprovalPolicy {
return effectiveApprovalPolicy(agent.session.events) ?? this.config.policy ?? 'ask'
private effectivePolicy(session: Session): ApprovalPolicy {
return effectiveApprovalPolicy(session.events) ?? this.config.policy ?? 'ask'
}
/** Dispatch the waterfall, contained and raced against `req.signal`. */
private async decide(req: ApprovalRequest): Promise<ApprovalOutcome> {
if (req.signal?.aborted) return 'cancelled'
/**
* Dispatch the waterfall, contained and raced against the request signal.
* @param req - the borrowed public request.
* @param session - the request agent's session used for policy lookup.
* @returns the normalized closed outcome.
*/
private async decide(req: ApprovalRequest, session: Session): Promise<ApprovalOutcome> {
const signal = req.signal
if (signal?.aborted) return 'cancelled'
// The 'never' policy is decided HERE, before any dispatch: a listener
// registered with `prepend: true` after this service mounts would sit
// ahead of any gate LISTENER, so a listener-shaped gate cannot keep the
// documented promise that 'never' rejects deterministically regardless
// of registration order — only the service's own request path can.
if (this.effectivePolicy(req.agent) === 'never') return 'rejected'
if (this.effectivePolicy(session) === 'never') return 'rejected'
// Enter the promise chain BEFORE dispatching: a listener that throws
// SYNCHRONOUSLY (before its first await) must land in the same rejection
// path as an async one — `Promise.resolve(call())` would let it escape
// the containment into the caller.
const answer: Promise<ApprovalOutcome> = Promise.resolve().then(
() => this.ctx.waterfall(this, 'approval/request', req, () => Promise.resolve<ApprovalOutcome>('unavailable')),
() => this.ctx.waterfall(
scopeTarget(this, req.agent), 'approval/request', req,
() => Promise.resolve<ApprovalOutcome>('unavailable'),
),
).then(
// Normalize a rogue (non-vocabulary) answerer return to the fail-closed
// outcome instead of leaking it into callers' closed-union switches.
@@ -419,10 +444,12 @@ export class ApprovalService extends Service {
// tool call open — the seam contains its callbacks.
() => 'unavailable',
)
const signal = req.signal
if (signal === undefined) return answer
return await new Promise<ApprovalOutcome>((resolve) => {
const onAbort = () => { resolve('cancelled') }
const onAbort = () => {
signal.removeEventListener('abort', onAbort)
resolve('cancelled')
}
signal.addEventListener('abort', onAbort, { once: true })
void answer.then((outcome) => {
signal.removeEventListener('abort', onAbort)

157
packages/ui/user-approval/tests/approval.spec.ts
View File

@@ -2,7 +2,9 @@ import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { CallId } from '@deepseek-ai/dsh-llm'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import { carrierKeyOf, createScope } from '@deepseek-ai/dsh-scope'
import type { Scope } from '@deepseek-ai/dsh-scope'
import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ApprovalService, { ApprovalOutcome, ApprovalRequest, effectiveApprovalPolicy, setApprovalPolicy } from '@deepseek-ai/dsh-user-approval'
@@ -77,6 +79,99 @@ describe('ApprovalService.request', () => {
expect(Object.keys(appended[0]?.data ?? {}).sort()).toEqual(['id', 'toolName'])
})
it('borrows the exact readonly request for scoped dispatch and audit', async () => {
const ctx = await mounted()
const { agent, appended } = fakeAgent()
let scope!: Scope
const scopeFiber = await ctx.plugin(Object.assign((inner: Context) => {
scope = createScope(inner, agent)
}, { inject: ['approval'] }))
let received: ApprovalRequest | undefined
let carrier: unknown
scope.ctx.on('approval/request', function (req) {
received = req
carrier = carrierKeyOf(this)
return Promise.resolve<ApprovalOutcome>('allowed-once')
})
const request = requestOf(agent, {
toolName: 'scoped-tool',
callId: CallId('scoped-call'),
reason: 'scoped reason',
})
await expect(ctx.approval.request(request)).resolves.toBe('allowed-once')
expect(carrier).toBe(agent)
expect(received).toBe(request)
expect(appended).toHaveLength(2)
expect(appended[0]?.data).toMatchObject({
toolName: 'scoped-tool',
callId: 'scoped-call',
reason: 'scoped reason',
})
expect(appended[1]?.data).toMatchObject({ outcome: 'allowed-once' })
expect(appended[1]?.data['id']).toBe(appended[0]?.data['id'])
await scopeFiber.dispose()
})
it('contains an approval/asked observer throw after append and still completes the pair', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(ApprovalService)
const session = ctx.sessions.create(SessionId('asked-observer-throw'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
const agent = { session } as unknown as Agent
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
ctx.on('session/event', (_session, event) => {
if (event.type === 'approval/asked') throw new Error('observer failed after asked append')
})
ctx.on('approval/request', () => Promise.resolve<ApprovalOutcome>('allowed-once'))
await expect(ctx.approval.request(requestOf(agent))).resolves.toBe('allowed-once')
const audit = session.events.filter(event => event.type.startsWith('approval/'))
const asked = session.events.find((event): event is SessionEvent<'approval/asked'> => event.type === 'approval/asked')
const decided = session.events.find((event): event is SessionEvent<'approval/decided'> => event.type === 'approval/decided')
expect(audit.map(event => event.type)).toEqual(['approval/asked', 'approval/decided'])
expect(decided?.data.id).toBe(asked?.data.id)
expect(warn).toHaveBeenCalledWith(expect.stringContaining('session/event listener threw: Error: observer failed after asked append'))
})
it('contains an approval/decided observer throw after append and still resolves', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(ApprovalService)
const session = ctx.sessions.create(SessionId('decided-observer-throw'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
const agent = { session } as unknown as Agent
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
ctx.on('session/event', (_session, event) => {
if (event.type === 'approval/decided') throw new Error('observer failed after decided append')
})
ctx.on('approval/request', () => Promise.resolve<ApprovalOutcome>('rejected'))
await expect(ctx.approval.request(requestOf(agent))).resolves.toBe('rejected')
const audit = session.events.filter(event => event.type.startsWith('approval/'))
const asked = session.events.find((event): event is SessionEvent<'approval/asked'> => event.type === 'approval/asked')
const decided = session.events.find((event): event is SessionEvent<'approval/decided'> => event.type === 'approval/decided')
expect(audit.map(event => event.type)).toEqual(['approval/asked', 'approval/decided'])
expect(decided?.data).toMatchObject({ id: asked?.data.id, outcome: 'rejected' })
expect(warn).toHaveBeenCalledWith(expect.stringContaining('session/event listener threw: Error: observer failed after decided append'))
})
it('propagates an append failure that prevented audit log growth', async () => {
const ctx = await mounted()
const failure = new Error('append failed before log growth')
const agent = {
session: {
events: [{ type: 'turn/start' }],
append: () => { throw failure },
},
} as unknown as Agent
await expect(ctx.approval.request(requestOf(agent))).rejects.toBe(failure)
})
it('returns the first answering listener outcome (single decision slot)', async () => {
const ctx = await mounted()
const { agent } = fakeAgent()
@@ -99,6 +194,57 @@ describe('ApprovalService.request', () => {
await expect(ctx.approval.request(requestOf(agent))).resolves.toBe('unavailable')
})
it('dispatches to global and matching agent-scoped listeners, never a foreign scope', async () => {
const ctx = await mounted()
const { agent: agentA } = fakeAgent()
const { agent: agentB } = fakeAgent()
let scopeA!: Scope
let scopeB!: Scope
const scopesFiber = await ctx.plugin(Object.assign((inner: Context) => {
scopeA = createScope(inner, agentA)
scopeB = createScope(inner, agentB)
}, { inject: ['approval'] }))
const heard: string[] = []
ctx.on('approval/request', (req, next) => {
heard.push(req.agent === agentA ? 'global:A' : 'global:B')
return next()
})
scopeA.ctx.on('approval/request', (_req, next) => {
heard.push('scoped:A')
return next()
})
scopeB.ctx.on('approval/request', (_req, next) => {
heard.push('scoped:B')
return next()
})
await expect(ctx.approval.request(requestOf(agentA))).resolves.toBe('unavailable')
await expect(ctx.approval.request(requestOf(agentB))).resolves.toBe('unavailable')
expect(heard).toEqual(['global:A', 'scoped:A', 'global:B', 'scoped:B'])
await scopesFiber.dispose()
})
it('keys the scoped dispatch carrier to the exact request agent', async () => {
const ctx = await mounted()
const { agent } = fakeAgent()
let scope!: Scope
const scopeFiber = await ctx.plugin(Object.assign((inner: Context) => {
scope = createScope(inner, agent)
}, { inject: ['approval'] }))
let seenKey: object | undefined
scope.ctx.on('approval/request', function (req, next) {
seenKey = carrierKeyOf(this)
expect(req.agent).toBe(agent)
return next()
})
await expect(ctx.approval.request(requestOf(agent))).resolves.toBe('unavailable')
expect(seenKey).toBe(agent)
await scopeFiber.dispose()
})
it('contains a throwing answerer as unavailable', async () => {
const ctx = await mounted()
const { agent, appended } = fakeAgent()
@@ -241,6 +387,15 @@ describe('approval policy (the approval/policy fold)', () => {
expect(session.events.at(-1)).toMatchObject({ type: 'approval/policy', data: { policy: 'ask' } })
})
it('rejects a policy outside the closed vocabulary before appending', () => {
const append = vi.fn()
const session = { append } as unknown as Session
expect(() => { setApprovalPolicy(session, 'sometimes' as Parameters<typeof setApprovalPolicy>[1]) })
.toThrow('approval policy must be one of "ask" or "never"')
expect(append).not.toHaveBeenCalled()
})
it('defaults a schema-less construction to ask (the ?? narrows the optional TYPE)', async () => {
// Direct construction bypasses the plugin schema (the SystemPrompt-test
// precedent for covering a defaulted Config field's type-narrowing ??).

3
packages/ui/user-approval/tsconfig.json
View File

@@ -23,6 +23,9 @@
{
"path": "../../llm/llm"
},
{
"path": "../../core/scope"
},
{
"path": "../../core/session"
},