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

feat(events): interception seams — the typed-Decision surface for hooks

Browse Source 
Reshape the agent's interception surface so every seam returns a small, typed
Decision union, and the set covers the hook points a CC/Codex bridge (and a
native plugin) needs. "Native hooks" are not a package — a native hook is just a
cordis plugin on these canonical events; the bridges (a later PR) only translate
an external protocol onto the same surface.

dsh-agent:
- NEW agent/session-start(agent, source) emit (once before turn 1; SessionStartSource
  startup|resume|clear|compact) — a pure notification, seeds context via inject().
- NEW agent/prompt-submit waterfall → PromptDecision (allow, optionally rewriting the
  prompt or attaching additionalContext, or block).
- RESHAPE agent/turn-continuation boolean → ContinuationDecision ({action:'stop'} |
  {action:'continue', reason?}; a continue reason is recorded as next-step steering).
- New HookContext envelope (required source — inject() would mislabel a missing one).

dsh-tools: split the single tools/execute waterfall into tools/pre-execute
(PreToolDecision allow/deny/ask gate) and tools/post-execute (PostToolDecision
accept/block, optionally replacing content or attaching additionalContext). Core
dispatch sits between as plain code; the tool body keeps its inner try/catch so a
thrown tool still reaches post-execute as an isError. ToolExecutionResult gains
additionalContext (ferried to the loop's per-step buffer). Input rewrite is
deliberately NOT offered (a proposed RFC designs it consistently).

dsh-session: new `rejected` TurnEndReason — a turn whose whole prompt batch was
blocked by prompt-submit.

agent-loop firing points: session-start emitted at create (source threaded —
startup for create/fork, resume for resume()); prompt-submit per drained message
with the always-open-turn rule (a fully-blocked batch is a zero-step rejected
turn); the continuation reshape; post-tool additionalContext buffered and appended
after all tool/results (adjacency). ACP codec maps rejected→cancelled.

A worked native-plugin example (interception.spec.ts) proves all four seams compose
end-to-end through the real loop with NO hook/* events (those belong to the bridge
lib). All existing tools/execute + turn-continuation tests migrated. The
tool-subagent abort test now aborts after a microtask so it still exercises the
live onAbort bridge (execute() awaits pre-execute before the body runs).

RFCs: implemented/feature/2026-06-30-interception-seams.md (the reshape) +
proposed/feature/2026-06-30-pre-tool-input-rewrite.md (the deferred rewrite design).
This commit is contained in:
Tianyi Cui
2026-06-30 17:11:18 +08:00
parent 13c6e847a2
commit dc95a7881d
28 changed files with 1255 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/agent-loop/src/index.ts
+28 -7
View File
@@ -10,7 +10,7 @@
import { Context, Service } from 'cordis'
import { randomUUID } from 'node:crypto'
import z from 'schemastery'
import type { AgentFactory, AgentHandle, AgentId, AgentOptions, CreateAgentOptions, ResumeAgentOptions } from '@deepseek-ai/dsh-agent'
import type { AgentFactory, AgentHandle, AgentId, AgentOptions, CreateAgentOptions, ResumeAgentOptions, SessionStartSource } from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
@@ -129,7 +129,7 @@ export class AgentLoop extends Service implements AgentFactory {
// session + agent down as one ordered chain, capturing the loop's closing
// flush). The whole effect is owned by THIS fiber; no AgentHandle is needed.
const session = this.ctx.sessions.prepare(SessionId(`${id}-session-${randomUUID()}`), { meta: {} })
const { agent } = this.start(id, options, session)
const { agent } = this.start(id, options, session, 'startup')
return agent
}
@@ -152,7 +152,9 @@ export class AgentLoop extends Service implements AgentFactory {
...options.seed !== undefined ? { seed: options.seed } : {},
meta: options.meta ?? {},
})
return this.startOwned(options.agentId, options.agentOptions ?? {}, session)
// A seeded (forked) create is still a fresh start, NOT a resume — `resume`
// is reserved for reloading a PERSISTED session via resume()/resumeWith().
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'startup')
}
/**
@@ -224,7 +226,7 @@ export class AgentLoop extends Service implements AgentFactory {
...meta.seedLength !== undefined ? { seedLength: meta.seedLength } : {},
},
})
return this.startOwned(options.agentId, options.agentOptions ?? {}, session)
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'resume')
}
/**
@@ -261,14 +263,33 @@ export class AgentLoop extends Service implements AgentFactory {
* so a throwing `session/created`/`agent/created` listener unwinds the
* already-yielded disposers instead of leaking.
*
* `source` says why the session began ({@link SessionStartSource}); it is
* emitted as `agent/session-start` once, AFTER the agent is registered (so a
* listener can resolve the agent via `ctx.agents.get(id)` and `inject()` into
* it) and BEFORE the loop starts its first turn. The emit is contained: a
* throwing session-start listener must not abort agent construction — it is
* logged, and the agent still starts. (Unlike a turn-boundary throw, there is
* no open turn here to balance; the durable evidence of a session-start hook
* is whatever it `inject()`ed.)
*
* Returns the agent plus the composite effect's disposer (`disposeAgent`).
*/
private start(id: AgentId, options: AgentOptions, session: Session): { agent: ReactLoopAgent; disposeAgent: () => Promise<void> } {
private start(
id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource,
): { agent: ReactLoopAgent; disposeAgent: () => Promise<void> } {
const agent = new ReactLoopAgent(this.ctx, id, options, session)
const dispose = this.ctx.effect(function* (this: AgentLoop) {
yield this.ctx.sessions.enter(session)
this.ctx.sessions.announce(session)
yield this.ctx.agents.register(agent)
// Fire AFTER register (a listener can ctx.agents.get(id) + inject()) and
// BEFORE the loop's first turn. Contained: a throwing listener is logged,
// never aborts construction (no open turn to balance here).
try {
this.ctx.emit('agent/session-start', agent, source)
} catch (error: unknown) {
this.ctx.logger.warn(`agent "${id}": agent/session-start listener threw: ${String(error)}`)
}
const stop = agent.start()
// Disposed FIRST (LIFO): request loop stop (sync), then AWAIT the loop's
// actual exit so its closing flush lands while onAppend (yielded above,
@@ -295,8 +316,8 @@ export class AgentLoop extends Service implements AgentFactory {
* `AgentHandle.dispose(): Promise<void>` contract (mirrors the ACP `quiesce()`
* helper).
*/
private startOwned(id: AgentId, options: AgentOptions, session: Session): AgentHandle {
const { agent, disposeAgent } = this.start(id, options, session)
private startOwned(id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource): AgentHandle {
const { agent, disposeAgent } = this.start(id, options, session, source)
let disposing: Promise<void> | undefined
return { agent, dispose: () => (disposing ??= disposeAgent()) }
}
packages/core/agent-loop/src/loop.ts
+74 -10
View File
@@ -10,6 +10,7 @@
import type { Context } from 'cordis'
import type { FinishReason, GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
import { BlockAssembler, HarnessError } from '@deepseek-ai/dsh-llm'
import type { ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
@@ -367,15 +368,48 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
// decides "owed" from the log via isTurnOpen, so even a throwing turn/start
// listener — append pushes before notifying — still gets its turn/end).
session.append('turn/start', { turn, trigger })
// Record the queued user messages INSIDE the turn (after turn/start), so
// every event in the log is turn-enclosed. turn/end is now owed, so a throw
// while appending these is caught below and the turn is still closed.
// Each drained queued message runs the `agent/prompt-submit` waterfall before
// it becomes a `user/message` — a hook can rewrite the prompt or block it.
// Recorded INSIDE the turn (after turn/start) so every event is turn-enclosed;
// turn/end is now owed, so a throwing prompt-submit listener (the waterfall
// throws) is caught below and the turn still closes.
let anyAllowed = false
// Seeded with a floor (only observable if the batch were empty, which
// runTurn never allows — it is called with ≥1 queued message); each `block`
// decision carries a required `reason` and overwrites it, so a fully-blocked
// batch always reports the last vetoing reason.
let lastBlockReason = 'prompt blocked by hook'
for (const message of queued) {
session.append('user/message', { content: message.content, source: message.source }, { surfaceOp: 'append' })
const decision = await ctx.waterfall(
'agent/prompt-submit', agent, message.content, message.source,
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
)
if (decision.kind === 'block') {
lastBlockReason = decision.reason
continue
}
anyAllowed = true
// `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
const content = decision.content ?? message.content
session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' })
// `allow.additionalContext` is a SEPARATE context/message the next request
// also sees. The turn is open, so inject() appends it into THIS turn.
if (decision.additionalContext) {
agent.inject(decision.additionalContext.content, { source: decision.additionalContext.source })
}
}
ctx.emit('agent/turn-start', agent, turn)
while (true) {
// A fully-blocked batch (every prompt vetoed by prompt-submit) opens a
// zero-step turn that ends `rejected`: break BEFORE the first step so the
// boundary stays balanced (turn/start → turn/end) and the block is a
// durable in-turn fact. `anyAllowed` never changes inside the loop, so this
// only ever fires on the first iteration.
if (!anyAllowed) {
reason = { kind: 'rejected', reason: lastBlockReason }
break
}
step += 1
// Steering from the previous round's continuation listeners (or
@@ -448,10 +482,10 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
if (closeStep()) break
const defaultDecision = stepOutcome.hadToolCalls || steered
let shouldContinue: boolean
const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
let decision: ContinuationDecision
try {
shouldContinue = await ctx.waterfall(
decision = await ctx.waterfall(
'agent/turn-continuation', agent, turn, defaultDecision,
() => Promise.resolve(defaultDecision),
)
@@ -461,9 +495,18 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
break
}
// A forced `continue` may carry model-facing context: record it as
// next-STEP steering (the steering channel), so the continued turn's next
// iteration drains it before its request — the typed twin of the /goal
// step/end-steer pattern.
if (decision.action === 'continue' && decision.reason) {
agent.inbox.steer({ content: decision.reason.content, source: decision.reason.source })
}
let shouldContinue = decision.action === 'continue'
// Steering from step/end session-event or continuation listeners (the
// /goal pattern) demands the model see it — it overrides a negative
// decision; the next iteration's drain records it.
// /goal pattern) demands the model see it — it overrides a stop decision;
// the next iteration's drain records it.
if (!shouldContinue && agent.inbox.hasSteering) shouldContinue = true
// A cancel that landed during the continuation window — after the step's
@@ -645,6 +688,12 @@ async function runStep(
// ToolRegistry.execute converts tool failures (including aborts) into
// isError results, so abort is re-checked around every call here.
const toolCalls = message.content.filter(block => block.type === 'tool-call')
// Per-step buffer of `additionalContext` attached by tools/post-execute
// listeners. Appended as context/message(s) only AFTER every tool/result for
// the step, so a multi-call step keeps tool-call/result adjacency
// (interleaving context between a call's result and the next call's would
// break the pairing the next model request relies on).
const pendingContext: HookContext[] = []
for (const call of toolCalls) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
@@ -655,6 +704,12 @@ async function runStep(
} catch {
parsedArguments = call.arguments
}
// TODO(pre-tool-input-rewrite): tools/pre-execute deliberately cannot rewrite
// `arguments` — tool/call (the audit record) and assistant/message (the
// model-history source) are logged BEFORE execute, and live consumers (ACP,
// tool-bash presentation) read the pre-execution args, so an execution-only
// rewrite would desync the UI from what ran. Designing that consistently is
// its own proposed RFC (docs/rfc/proposed/feature/…-pre-tool-input-rewrite.md).
const result = await ctx.tools.execute({
callId: call.id,
name: call.name,
@@ -666,7 +721,7 @@ async function runStep(
turn, step,
// The correlation id MUST be the loop's authoritative call.id (the
// model-transcript id that deriveMessages turns into toolCallId), NOT
// result.callId — a tools/execute waterfall listener returning a
// result.callId — a post-execute waterfall listener returning a
// mismatched id would otherwise orphan the call↔result pairing in the
// next model request. A listener-internal id, if ever needed, belongs in
// a separate diagnostic field, never overloaded onto callId.
@@ -675,6 +730,8 @@ async function runStep(
isError: result.isError,
...result.error ? { error: result.error } : {},
}, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
// Buffer (don't append yet) any post-execute additionalContext for this call.
if (result.additionalContext) pendingContext.push(result.additionalContext)
// signal CAN flip during the await above (abort() inside a tool);
// the analyzer can't see through the await boundary.
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
@@ -683,6 +740,13 @@ async function runStep(
/* v8 ignore stop */
}
// Append buffered post-execute context AFTER every tool/result, preserving
// tool-call/result adjacency across the whole batch. inject() appends into the
// open turn (a context/message at its chronological position).
for (const context of pendingContext) {
agent.inject(context.content, { source: context.source })
}
return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
}