Merge remote-tracking branch 'origin/master' into feat/py-types-code-mode
This commit is contained in:
Chinesezjc
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write packages/core/agent-loop/README.md
|
||||
README.md: 79d2865073c89bd88a4d39fafacb5cf60f1fc10c
|
||||
README.zh.md: 48c4f4900d25f524942abf53c1bc887e7d125cb3
|
||||
README.md: 1662b1076cc116888d048cb6af1be1c7ab8196f6
|
||||
README.zh.md: 2fca32a02fdd73961c912c988933e1cd1a1a5817
|
||||
|
||||
@@ -57,6 +57,8 @@ The concrete `ReactLoopAgent`, its queued input, outbox, and run controls are pa
|
||||
|
||||
The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. `reserveTurnAdmission()` can synchronously hold that idle boundary for a standalone durable operation: accepted waking work has right of way, later sends keep their ordinary queue identity and FIFO position, release re-arms the same driver path, and `whenIdle()` waits for the reservation without making teardown await it. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model.
|
||||
|
||||
`steer()` attaches a one-shot admission receipt to its exact accepted message. After `agent/step` and asynchronous prompt assembly succeed, the loop commits a stable pending batch as `steering/message`, snapshots derived history, and opens `step/start`; only then does each receipt resolve `admitted` with that turn and step. Later arrivals remain pending. Idle steering enters the ordinary FIFO and uses the first request of its eventual turn as the same admission boundary. A turn-concluding tool result, broad cancellation, disposal, or a claimed idle-steering turn that never reaches a request resolves affected receipts `rejected`; `cancel(..., { keepInbox: true })` and non-terminal routing preserve pending delivery. Open-turn `inject()` still commits after all tool results, including accepted context finalized during an interrupted batch, while steering remains provisional until a request admits it.
|
||||
|
||||
Every FIFO acceptance mints an `InboxItemId` and publishes `agent/inbox/enqueue` with the complete occurrence. `updateInbox()` owns the synchronous queued-item boundary: edit freezes replacement content without changing message identity or position, while remove publishes discard. Edit publishes `agent/inbox/update`; steering and claimed occurrences return `not-found`. Claim publishes `agent/inbox/dequeue` and irrevocably removes the live address before prompt admission, so a racing update cannot rewrite durable history; `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
|
||||
|
||||
### Loop lifecycle (`agent.ts`)
|
||||
|
||||
@@ -57,6 +57,8 @@ interface Config {
|
||||
|
||||
统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。`reserveTurnAdmission()` 可以为独立持久操作同步保留该空闲边界：已获接纳的唤醒工作拥有优先权，之后发送的项保留普通队列身份与 FIFO 位置，释放会重新启用同一驱动器路径，`whenIdle()` 会等待预留结束，但 teardown 不会等待它。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。
|
||||
|
||||
`steer()` 会把一次性准入回执附着到其准确的已接收消息。`agent/step` 和异步提示词组装成功后，循环把稳定的待处理批次提交为 `steering/message`、捕获派生历史并开启 `step/start`；只有此时，每个回执才会解析为 `admitted`，并附带轮次与步骤。之后到达的消息继续待处理。空闲 steering 会进入普通 FIFO，并以其最终轮次的首次请求作为相同准入边界。结束轮次的工具结果、广义取消、dispose（资源释放），或已领取 idle-steering 消息却从未到达请求的轮次，会把受影响回执解析为 `rejected`；`cancel(..., { keepInbox: true })` 和非终止型路由会保留待处理投递。活跃轮次内的 `inject()` 仍会在所有工具结果后提交，包括被中断批次中已最终确认的上下文；steering 则保持待准入，直到请求接纳它。
|
||||
|
||||
每次 FIFO 接受项时都会铸造一个 `InboxItemId`，并通过 `agent/inbox/enqueue` 发布完整的单次入队项。`updateInbox()` 持有同步 queued 项边界：编辑会冻结替换内容，但不改变消息标识或位置；移除会发布 discard。编辑会发布 `agent/inbox/update`；steering 项和已被认领的项会返回 `not-found`。认领操作会发布 `agent/inbox/dequeue`，并在提示词接纳前不可逆地移除实时寻址标识，因此竞态中的更新无法改写持久历史；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。
|
||||
|
||||
### 循环生命周期（`agent.ts`）
|
||||
|
||||
@@ -29,6 +29,8 @@ import type {
|
||||
RequestError,
|
||||
RequestErrorAction,
|
||||
SendOptions,
|
||||
SteeringOutcome,
|
||||
SteeringReceipt,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
import {
|
||||
BlockAssembler,
|
||||
@@ -56,6 +58,27 @@ type StepOutcome =
|
||||
| { kind: 'completed'; continueTurn: boolean; concluded: boolean; maxTokens: boolean }
|
||||
| { kind: 'request-failed'; error: RequestError; failure: LlmFailure; retryPolicy: ResolvedRetryPolicy | undefined }
|
||||
|
||||
/** Internal one-shot controller paired with a public steering receipt. */
|
||||
interface SteeringDelivery {
|
||||
readonly receipt: SteeringReceipt
|
||||
settle(outcome: SteeringOutcome): void
|
||||
}
|
||||
|
||||
/** Create one idempotent steering-admission controller. */
|
||||
function createSteeringDelivery(): SteeringDelivery {
|
||||
const { promise, resolve } = Promise.withResolvers<SteeringOutcome>()
|
||||
let settled = false
|
||||
return {
|
||||
receipt: { outcome: promise },
|
||||
settle(outcome): void {
|
||||
/* v8 ignore next -- each ownership transfer removes the delivery before another settlement path can reach it. */
|
||||
if (settled) return
|
||||
settled = true
|
||||
resolve(outcome)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
const RUNTIME_CONTEXT_SOURCE = '@deepseek-ai/dsh-system-prompt'
|
||||
/** Clearing marker kept distinct from every prefixed {@link renderContextSnapshot} result. */
|
||||
const CLEARED_RUNTIME_CONTEXT = 'Current runtime context: none. Earlier runtime-context snapshots no longer apply.'
|
||||
@@ -112,9 +135,13 @@ function requestProposal(header: EpochHeader): LlmCallConfig {
|
||||
*/
|
||||
export class ReactLoopAgent implements Agent {
|
||||
/** Prompts awaiting individual turns. */
|
||||
private queued: { item: InboxItem; wakeup: boolean }[] = []
|
||||
private queued: { item: InboxItem; wakeup: boolean; delivery?: SteeringDelivery }[] = []
|
||||
/** Input taken into the session log at step boundaries. */
|
||||
private outbox: { message: UserMessage; steering: boolean; item?: InboxItem }[] = []
|
||||
private outbox: { message: UserMessage; steering: boolean; item?: InboxItem; delivery?: SteeringDelivery }[] = []
|
||||
/** Steering already committed to the log but not yet captured by a request. */
|
||||
private pendingAdmissions: SteeringDelivery[] = []
|
||||
/** Whether the active cancellation preserves already committed pending delivery. */
|
||||
private preservePendingAdmissionsOnAbort = false
|
||||
|
||||
/** Whether observers see a running interval; consecutive turns share it. */
|
||||
private busy = false
|
||||
@@ -132,7 +159,6 @@ export class ReactLoopAgent implements Agent {
|
||||
private abort: AbortController | undefined
|
||||
/** Resolves when the current admission and turn exit. */
|
||||
done: Promise<void> = Promise.resolve()
|
||||
|
||||
/** The agent-scoped registration boundary; the lifecycle owner unwinds it after {@link done}. */
|
||||
readonly scope: Scope
|
||||
/** The agent's scoped composition context ({@link Agent.ctx}). */
|
||||
@@ -166,6 +192,15 @@ export class ReactLoopAgent implements Agent {
|
||||
send(
|
||||
message: UserMessage,
|
||||
options: SendOptions,
|
||||
): void {
|
||||
this.route(message, options)
|
||||
}
|
||||
|
||||
/** Route one accepted message, optionally tracking steering admission. */
|
||||
private route(
|
||||
message: UserMessage,
|
||||
options: SendOptions,
|
||||
delivery?: SteeringDelivery,
|
||||
): void {
|
||||
const { target, wakeup } = options
|
||||
if (target === 'next-step' && !wakeup) {
|
||||
@@ -184,9 +219,9 @@ export class ReactLoopAgent implements Agent {
|
||||
placement,
|
||||
})
|
||||
if (placement === 'steering') {
|
||||
this.outbox.push({ message, steering: true, item })
|
||||
this.outbox.push({ message, steering: true, item, ...delivery === undefined ? {} : { delivery } })
|
||||
} else {
|
||||
this.queued.push({ item, wakeup })
|
||||
this.queued.push({ item, wakeup, ...delivery === undefined ? {} : { delivery } })
|
||||
}
|
||||
// Preserve the routing decision for every send in this synchronous caller
|
||||
// stack, while installing quiescence ownership before enqueue observers
|
||||
@@ -217,6 +252,7 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
case 'remove': {
|
||||
this.queued.splice(queuedIndex, 1)
|
||||
pending.delivery?.settle({ status: 'rejected' })
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/inbox/discard', [pending.item])
|
||||
return 'applied'
|
||||
}
|
||||
@@ -234,12 +270,14 @@ export class ReactLoopAgent implements Agent {
|
||||
})
|
||||
}
|
||||
|
||||
/** Steer the open turn, falling back to a waking prompt while idle. */
|
||||
steer(input: UserMessage): void {
|
||||
this.send(input, {
|
||||
/** Steer the open turn, falling back to a tracked waking prompt while idle. */
|
||||
steer(input: UserMessage): SteeringReceipt {
|
||||
const delivery = createSteeringDelivery()
|
||||
this.route(input, {
|
||||
target: 'next-step',
|
||||
wakeup: true,
|
||||
})
|
||||
}, delivery)
|
||||
return delivery.receipt
|
||||
}
|
||||
|
||||
/** Append model-facing context without waking the driver. */
|
||||
@@ -293,11 +331,17 @@ export class ReactLoopAgent implements Agent {
|
||||
// inboxes clear; listener failures are contained by the dispatcher.
|
||||
if (cause.kind !== 'disposed') emitAgentEvent(this.loopCtx, this, 'agent/cancel-requested', cause)
|
||||
}
|
||||
if (options.keepInbox && this.abort !== undefined) this.preservePendingAdmissionsOnAbort = true
|
||||
if (!options.keepInbox) {
|
||||
const discarded = this.queued.map(item => item.item)
|
||||
for (const item of this.queued) item.delivery?.settle({ status: 'rejected' })
|
||||
for (const item of this.outbox) {
|
||||
if (item.steering && item.item !== undefined) discarded.push(item.item)
|
||||
if (item.steering && item.item !== undefined) {
|
||||
item.delivery?.settle({ status: 'rejected' })
|
||||
discarded.push(item.item)
|
||||
}
|
||||
}
|
||||
this.rejectPendingAdmissions()
|
||||
// Clear before abort observers run: replacement work belongs to the next turn.
|
||||
this.queued.length = 0
|
||||
this.outbox.length = 0
|
||||
@@ -362,7 +406,8 @@ export class ReactLoopAgent implements Agent {
|
||||
// The some() guard above proves the queue is non-empty; the non-null
|
||||
// assertion expresses that invariant.
|
||||
// oxlint-disable-next-line typescript/no-non-null-assertion
|
||||
const { item } = this.queued.shift()!
|
||||
const pending = this.queued.shift()!
|
||||
const { item, delivery } = pending
|
||||
const { message } = item
|
||||
const inheritedOutboxLength = this.outbox.length
|
||||
|
||||
@@ -412,6 +457,7 @@ export class ReactLoopAgent implements Agent {
|
||||
// still owns the slot here and releasing it unconditionally is exact.
|
||||
this.abort = undefined
|
||||
if (admitted === undefined) {
|
||||
delivery?.settle({ status: 'rejected' })
|
||||
this.acceptsNextStep = false
|
||||
try {
|
||||
this.flushRejectedAdmissionContexts()
|
||||
@@ -429,7 +475,7 @@ export class ReactLoopAgent implements Agent {
|
||||
this.continueOrIdle()
|
||||
return
|
||||
}
|
||||
await this.run(trigger, admitted, inheritedOutboxLength)
|
||||
await this.run(trigger, admitted, inheritedOutboxLength, Object.freeze([]), delivery)
|
||||
})
|
||||
// Published only after the abort owner and pending done are installed: a
|
||||
// dequeue listener that cancels or disposes must find live cancellation
|
||||
@@ -446,6 +492,7 @@ export class ReactLoopAgent implements Agent {
|
||||
admitted: UserMessage[] = [],
|
||||
inheritedOutboxLength = 0,
|
||||
priorFailures: readonly LlmFailure[] = Object.freeze([]),
|
||||
promptDelivery?: SteeringDelivery,
|
||||
): Promise<void> {
|
||||
// Both entries hold the invariant: kick() clears the admission slot before
|
||||
// awaiting run(), and a retry is entered only after the prior run clears it.
|
||||
@@ -453,6 +500,7 @@ export class ReactLoopAgent implements Agent {
|
||||
if (this.abort !== undefined) throw new Error(`agent "${this.id}" is already running`)
|
||||
const controller = new AbortController()
|
||||
this.abort = controller
|
||||
this.preservePendingAdmissionsOnAbort = false
|
||||
this.acceptsNextStep = true
|
||||
const signal = controller.signal
|
||||
const turn = this.lastTurn + 1
|
||||
@@ -476,13 +524,12 @@ export class ReactLoopAgent implements Agent {
|
||||
// Context or steering retained by an earlier rejected admission happened
|
||||
// before this prompt and must occupy the same order in durable history.
|
||||
this.drainOutbox(turn, inheritedOutboxLength)
|
||||
if (promptDelivery !== undefined) this.pendingAdmissions.push(promptDelivery)
|
||||
for (const input of admitted) {
|
||||
this.session.append('user/message', input, { surfaceOp: 'append' })
|
||||
}
|
||||
signal.throwIfAborted()
|
||||
|
||||
this.drainOutbox(turn)
|
||||
|
||||
steps: while (true) {
|
||||
step += 1
|
||||
const outcome = await this.step(turn, step, signal)
|
||||
@@ -490,11 +537,15 @@ export class ReactLoopAgent implements Agent {
|
||||
case 'completed':
|
||||
requestFailureHistory = Object.freeze([])
|
||||
if (outcome.maxTokens) reason = { kind: 'max-tokens' }
|
||||
// A concluding tool result is terminal: steering already in the
|
||||
// log waits for the next turn's request instead of reopening this
|
||||
// one, and the agent/turn-stopping drain below is skipped for the same
|
||||
// reason.
|
||||
if (outcome.concluded) break steps
|
||||
// A concluding tool result is terminal: reject steering that did
|
||||
// not enter a request, while retaining same-boundary context in
|
||||
// durable history before the turn closes.
|
||||
if (outcome.concluded) {
|
||||
this.discardOutboxSteering()
|
||||
this.drainOutbox(turn)
|
||||
break steps
|
||||
}
|
||||
/* v8 ignore next -- step() folded the same steering predicate into continueTurn immediately before returning. */
|
||||
if (outcome.continueTurn || this.outbox.some(item => item.steering)) continue
|
||||
break
|
||||
case 'request-failed': {
|
||||
@@ -530,7 +581,10 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
await this.loopCtx.serial(agentCarrier(this), 'agent/turn-stopping', this, turn, signal)
|
||||
signal.throwIfAborted()
|
||||
if (!this.drainOutbox(turn)) break
|
||||
this.drainOutboxContexts()
|
||||
if (!this.outbox.some(item => item.steering)) {
|
||||
break
|
||||
}
|
||||
}
|
||||
} catch (caught: unknown) {
|
||||
try {
|
||||
@@ -568,6 +622,10 @@ export class ReactLoopAgent implements Agent {
|
||||
// is still this run's controller here.
|
||||
this.abort = undefined
|
||||
signal.removeEventListener('abort', cancelRetry)
|
||||
const preservePending = signal.aborted && this.preservePendingAdmissionsOnAbort
|
||||
this.preservePendingAdmissionsOnAbort = false
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- keepInbox cancellation can set this while turn work is awaited.
|
||||
if (!preservePending) this.rejectPendingAdmissions()
|
||||
}
|
||||
|
||||
if (opened) {
|
||||
@@ -606,10 +664,6 @@ export class ReactLoopAgent implements Agent {
|
||||
await this.loopCtx.serial(agentCarrier(this), 'agent/step', this, turn, step, signal)
|
||||
signal.throwIfAborted()
|
||||
|
||||
// Take the outbox whole — same-boundary steering and context leave in
|
||||
// this request together.
|
||||
this.drainOutbox(turn)
|
||||
|
||||
// Assemble request-owned prompt inputs fresh each step. Dynamic context is
|
||||
// committed at the tail before deriving history once, preserving the stable
|
||||
// system/history cache prefix while keeping every model-visible byte logged.
|
||||
@@ -618,12 +672,18 @@ export class ReactLoopAgent implements Agent {
|
||||
const system = renderPrompt(assembly)
|
||||
materializeRuntimeContext(session, renderContextSnapshot(assembly))
|
||||
|
||||
// Commit the exact pending batch only after every asynchronous
|
||||
// pre-request contribution succeeded. Input accepted after this splice
|
||||
// remains pending for a later request.
|
||||
this.drainOutbox(turn)
|
||||
|
||||
// Snapshot the exact log prefix: the reconstruction boundary. Appends
|
||||
// after this synchronous snapshot join the next request.
|
||||
const boundaryMessages = session.deriveMessages()
|
||||
|
||||
session.append('step/start', { turn, step })
|
||||
this.stepOpen = true
|
||||
this.admitPendingAdmissions(turn, step)
|
||||
signal.throwIfAborted()
|
||||
|
||||
const { request, preparedCall } = await this.buildRequest(
|
||||
@@ -690,14 +750,14 @@ export class ReactLoopAgent implements Agent {
|
||||
))
|
||||
}
|
||||
|
||||
// Tool results stay adjacent to their calls; input accepted during the
|
||||
// request enters the log only after the complete result batch.
|
||||
const steered = this.drainOutbox(turn)
|
||||
// Ordinary context keeps the base loop's result-adjacent commit point.
|
||||
// Steering remains provisional until the next request snapshot admits it.
|
||||
this.drainOutboxContexts()
|
||||
session.append('step/end', { turn, step })
|
||||
this.stepOpen = false
|
||||
return {
|
||||
kind: 'completed',
|
||||
continueTurn: (toolCalls.length > 0 && !concluded) || steered,
|
||||
continueTurn: (toolCalls.length > 0 && !concluded) || this.outbox.some(item => item.steering),
|
||||
concluded,
|
||||
maxTokens: finish.kind === 'max-tokens',
|
||||
}
|
||||
@@ -802,25 +862,83 @@ export class ReactLoopAgent implements Agent {
|
||||
return { request, ...preparedCall === undefined ? {} : { preparedCall } }
|
||||
}
|
||||
|
||||
/** Commit the outbox and report whether it contained steering. */
|
||||
private drainOutbox(turn: number, limit = this.outbox.length): boolean {
|
||||
let steered = false
|
||||
for (const item of this.outbox.splice(0, limit)) {
|
||||
if (item.steering) {
|
||||
steered = true
|
||||
/* v8 ignore next -- only inbox-backed steer entries carry steering:true. */
|
||||
if (item.item === undefined) throw new Error(`agent "${this.id}" steering outbox item has no inbox identity`)
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.item)
|
||||
this.session.append(
|
||||
'steering/message',
|
||||
{ turn, message: item.message },
|
||||
{ surfaceOp: 'append' },
|
||||
)
|
||||
} else {
|
||||
this.session.append('user/message', item.message, { surfaceOp: 'append' })
|
||||
/** Commit one stable outbox prefix and retain tracked delivery until snapshot admission. */
|
||||
private drainOutbox(turn: number, limit = this.outbox.length): void {
|
||||
const batch = this.outbox.splice(0, limit)
|
||||
for (let index = 0; index < batch.length; index += 1) {
|
||||
const item = batch[index]
|
||||
/* v8 ignore next -- the index walks the exact array length. */
|
||||
if (item === undefined) throw new Error(`agent "${this.id}" outbox item disappeared during drain`)
|
||||
try {
|
||||
if (item.steering) {
|
||||
/* v8 ignore next -- only inbox-backed steer entries carry steering:true. */
|
||||
if (item.item === undefined) throw new Error(`agent "${this.id}" steering outbox item has no inbox identity`)
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.item)
|
||||
this.session.append(
|
||||
'steering/message',
|
||||
{ turn, message: item.message },
|
||||
{ surfaceOp: 'append' },
|
||||
)
|
||||
if (item.delivery !== undefined) this.pendingAdmissions.push(item.delivery)
|
||||
} else {
|
||||
this.session.append('user/message', item.message, { surfaceOp: 'append' })
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
item.delivery?.settle({ status: 'rejected' })
|
||||
this.outbox.unshift(...batch.slice(item.steering ? index + 1 : index))
|
||||
throw error
|
||||
}
|
||||
}
|
||||
return steered
|
||||
}
|
||||
|
||||
/** Commit ordinary context while retaining provisional steering in order. */
|
||||
private drainOutboxContexts(): void {
|
||||
const pending = this.outbox
|
||||
this.outbox = []
|
||||
for (let index = 0; index < pending.length; index += 1) {
|
||||
const item = pending[index]
|
||||
/* v8 ignore next -- the index walks the exact array length. */
|
||||
if (item === undefined) throw new Error(`agent "${this.id}" outbox item disappeared during context drain`)
|
||||
if (item.steering) {
|
||||
this.outbox.push(item)
|
||||
continue
|
||||
}
|
||||
try {
|
||||
this.session.append('user/message', item.message, { surfaceOp: 'append' })
|
||||
} catch (error: unknown) {
|
||||
this.outbox.push(...pending.slice(index))
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Settle every committed steering item captured by this immutable request. */
|
||||
private admitPendingAdmissions(turn: number, step: number): void {
|
||||
const outcome: SteeringOutcome = { status: 'admitted', turn, step }
|
||||
for (const delivery of this.pendingAdmissions.splice(0)) delivery.settle(outcome)
|
||||
}
|
||||
|
||||
/** Reject committed steering that left the inbox without reaching a request. */
|
||||
private rejectPendingAdmissions(): void {
|
||||
for (const delivery of this.pendingAdmissions.splice(0)) delivery.settle({ status: 'rejected' })
|
||||
}
|
||||
|
||||
/** Discard uncommitted steering while retaining same-boundary injected context. */
|
||||
private discardOutboxSteering(): void {
|
||||
const contexts: typeof this.outbox = []
|
||||
const discarded: InboxItem[] = []
|
||||
for (const item of this.outbox) {
|
||||
if (!item.steering) {
|
||||
contexts.push(item)
|
||||
continue
|
||||
}
|
||||
item.delivery?.settle({ status: 'rejected' })
|
||||
/* v8 ignore next -- only inbox-backed steer entries carry steering:true. */
|
||||
if (item.item === undefined) throw new Error(`agent "${this.id}" steering outbox item has no inbox identity`)
|
||||
discarded.push(item.item)
|
||||
}
|
||||
this.outbox = contexts
|
||||
if (discarded.length > 0) emitAgentEvent(this.loopCtx, this, 'agent/inbox/discard', discarded)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -50,10 +50,11 @@ describe('Agent', () => {
|
||||
}])).toBeUndefined()
|
||||
expect(call('inject', [message('context')])).toBeUndefined()
|
||||
expect(call('followup', [message('followup')])).toBeUndefined()
|
||||
expect(call('steer', [message('steering')])).toBeUndefined()
|
||||
const receipt = agent.steer(message('steering'))
|
||||
await agent.whenIdle()
|
||||
|
||||
expect(adapter.requests).toHaveLength(3)
|
||||
expect(await receipt.outcome).toEqual({ status: 'admitted', turn: 3, step: 1 })
|
||||
})
|
||||
|
||||
it('idle inject() appends context without opening a turn or requesting a flush', async () => {
|
||||
|
||||
@@ -470,7 +470,10 @@ describe('agent loop', () => {
|
||||
parameters: {},
|
||||
async execute() {
|
||||
// steer while the turn is running (during tool execution)
|
||||
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'change of plans' }], source: { kind: 'user' } }))
|
||||
agent.send(
|
||||
createUserMessage({ content: [{ type: 'text', text: 'change of plans' }], source: { kind: 'user' } }),
|
||||
{ target: 'next-step', wakeup: true },
|
||||
)
|
||||
return [{ type: 'text', text: 'tool done' }]
|
||||
},
|
||||
}))
|
||||
@@ -544,6 +547,120 @@ describe('agent loop', () => {
|
||||
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('pending steering')
|
||||
})
|
||||
|
||||
it('rejects failed steering commits while preserving later context', async () => {
|
||||
const adapter = new MockAdapter([textResponse('recovered')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('failed-steering-commit'), { provider: 'mock', model: 'mock' })
|
||||
let receipt: ReturnType<Agent['steer']> | undefined
|
||||
ctx.on('agent/step', (subject) => {
|
||||
if (subject !== agent || receipt !== undefined) return
|
||||
receipt = subject.steer(createUserMessage({
|
||||
content: [{ type: 'text', text: 'rejected steering' }],
|
||||
source: { kind: 'user' },
|
||||
}))
|
||||
subject.inject(createUserMessage({
|
||||
content: [{ type: 'text', text: 'preserved context' }],
|
||||
source: { kind: 'plugin', plugin: 'loop-test' },
|
||||
}))
|
||||
})
|
||||
let rejected = false
|
||||
ctx.on('internal/dispatch', (_mode, name, args) => {
|
||||
if (name !== 'session/event') return
|
||||
const event = args[1] as { type: string }
|
||||
if (event.type === 'steering/message' && !rejected) {
|
||||
rejected = true
|
||||
throw new Error('reject steering commit')
|
||||
}
|
||||
})
|
||||
|
||||
send(agent, 'first prompt')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
if (receipt === undefined) throw new Error('agent/step did not submit steering')
|
||||
expect(await receipt.outcome).toEqual({ status: 'rejected' })
|
||||
expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(false)
|
||||
|
||||
send(agent, 'recover')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
const request = JSON.stringify(adapter.requests[0]?.messages)
|
||||
expect(request).toContain('preserved context')
|
||||
expect(request).not.toContain('rejected steering')
|
||||
})
|
||||
|
||||
it('rejects committed steering when the step boundary fails', async () => {
|
||||
const adapter = new MockAdapter([])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('failed-step-boundary'), { provider: 'mock', model: 'mock' })
|
||||
let receipt: ReturnType<Agent['steer']> | undefined
|
||||
ctx.on('agent/step', (subject) => {
|
||||
if (subject !== agent || receipt !== undefined) return
|
||||
receipt = subject.steer(createUserMessage({
|
||||
content: [{ type: 'text', text: 'committed steering' }],
|
||||
source: { kind: 'user' },
|
||||
}))
|
||||
})
|
||||
ctx.on('internal/dispatch', (_mode, name, args) => {
|
||||
if (name !== 'session/event') return
|
||||
const event = args[1] as { type: string }
|
||||
if (event.type === 'step/start') throw new Error('reject step boundary')
|
||||
})
|
||||
|
||||
send(agent, 'prompt')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
if (receipt === undefined) throw new Error('agent/step did not submit steering')
|
||||
expect(await receipt.outcome).toEqual({ status: 'rejected' })
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(true)
|
||||
expect(agent.session.events.some(event => event.type === 'step/start')).toBe(false)
|
||||
})
|
||||
|
||||
it('retries context and steering after a context commit fails', async () => {
|
||||
const adapter = new MockAdapter([textResponse('recovered')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('failed-context-commit'), { provider: 'mock', model: 'mock' })
|
||||
let receipt: ReturnType<Agent['steer']> | undefined
|
||||
ctx.on('agent/step', (subject) => {
|
||||
if (subject !== agent || receipt !== undefined) return
|
||||
subject.inject(createUserMessage({
|
||||
content: [{ type: 'text', text: 'preserved context' }],
|
||||
source: { kind: 'plugin', plugin: 'loop-test' },
|
||||
}))
|
||||
receipt = subject.steer(createUserMessage({
|
||||
content: [{ type: 'text', text: 'preserved steering' }],
|
||||
source: { kind: 'user' },
|
||||
}))
|
||||
})
|
||||
let rejected = false
|
||||
ctx.on('internal/dispatch', (_mode, name, args) => {
|
||||
if (name !== 'session/event') return
|
||||
const event = args[1] as { type: string; data?: { source?: { kind: string } } }
|
||||
if (event.type === 'user/message' && event.data?.source?.kind === 'plugin' && !rejected) {
|
||||
rejected = true
|
||||
throw new Error('reject context commit')
|
||||
}
|
||||
})
|
||||
|
||||
send(agent, 'first prompt')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(false)
|
||||
|
||||
send(agent, 'recover')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
if (receipt === undefined) throw new Error('agent/step did not submit steering')
|
||||
expect(await receipt.outcome).toEqual({ status: 'admitted', turn: 2, step: 1 })
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
const request = JSON.stringify(adapter.requests[0]?.messages)
|
||||
expect(request).toContain('preserved context')
|
||||
expect(request).toContain('preserved steering')
|
||||
})
|
||||
|
||||
it('inject() while idle appends context without opening a turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
@@ -721,13 +838,24 @@ describe('agent loop', () => {
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
let receipt: ReturnType<Agent['steer']> | undefined
|
||||
let contextInjected = false
|
||||
ctx.on('session/event', (session, event) => {
|
||||
if (session !== agent.session || event.type !== 'step/end' || contextInjected) return
|
||||
contextInjected = true
|
||||
agent.inject(createUserMessage({
|
||||
content: [{ type: 'text', text: 'final context' }],
|
||||
source: { kind: 'plugin', plugin: 'finalize' },
|
||||
}))
|
||||
})
|
||||
ctx.tools.register(defineContentToolFixture({
|
||||
name: 'finalize',
|
||||
description: '',
|
||||
parameters: {},
|
||||
async execute(_args, exec) {
|
||||
// Steering lands while the concluding tool is still executing.
|
||||
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'late steering' }], source: { kind: 'user' } }))
|
||||
// Steering lands while the concluding tool is still executing; the
|
||||
// step/end listener adds ordinary context after the normal result drain.
|
||||
receipt = agent.steer(createUserMessage({ content: [{ type: 'text', text: 'late steering' }], source: { kind: 'user' } }))
|
||||
exec.concludeTurn()
|
||||
return [{ type: 'text', text: 'final' }]
|
||||
},
|
||||
@@ -740,9 +868,12 @@ describe('agent loop', () => {
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
const events = agent.session.events.map(event => event.type)
|
||||
expect(events.filter(type => type === 'turn/end')).toHaveLength(1)
|
||||
// The steering is durable inside the concluded turn and feeds the NEXT
|
||||
// turn's request instead of being dropped or re-queued.
|
||||
expect(events).toContain('steering/message')
|
||||
if (receipt === undefined) throw new Error('concluding tool did not submit steering')
|
||||
expect(await receipt.outcome).toEqual({ status: 'rejected' })
|
||||
expect(events).not.toContain('steering/message')
|
||||
expect(agent.session.events.some(event => event.type === 'user/message'
|
||||
&& event.data.source.kind === 'plugin'
|
||||
&& event.data.content.some(block => block.type === 'text' && block.text === 'final context'))).toBe(true)
|
||||
|
||||
send(agent, 'follow up')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -751,7 +882,8 @@ describe('agent loop', () => {
|
||||
.flatMap(message => message.content)
|
||||
.filter(block => block.type === 'text')
|
||||
.map(block => block.text)
|
||||
expect(texts).toContain('late steering')
|
||||
expect(texts).toContain('final context')
|
||||
expect(texts).not.toContain('late steering')
|
||||
})
|
||||
|
||||
it('agent/request waterfall switches models by returning a replacement config; the switch is logged', async () => {
|
||||
|
||||
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write packages/core/agent/README.md
|
||||
README.md: 8799bc3664b2137b386b752f905e1414fb770cb9
|
||||
README.zh.md: 851c174ba80bebbab8ee1255cb04be4bcec7eabd
|
||||
README.md: 8a6028352127c4638c0b5e0e3ee85964d1d7d734
|
||||
README.zh.md: ffa71ea987ab355ff2f30b6164376199cd5d0170
|
||||
|
||||
@@ -64,7 +64,7 @@ The handle every plugin programs against:
|
||||
- `agent.reserveTurnAdmission()` — synchronously reserve the idle boundary before any queued waking prompt can claim its turn. An accepted prompt, including a same-tick pending wake, has right of way and makes reservation return `undefined`. Later sends keep their ordinary IDs, FIFO placement, and wakeup facts while held; `acceptsNextStep` remains false, `inject()` is not withheld, `whenIdle()` counts the reservation as activity, and the returned release is idempotent. This narrow coordination capability lets standalone durable operations such as manual compaction finish and flush before queued prompts derive from the session.
|
||||
- `agent.updateInbox(itemId, action)` — synchronously edits or removes one still-pending queued occurrence. Edit keeps its `MessageId`, `InboxItemId`, source, and FIFO position while replacing frozen content; remove emits the occurrence's terminal discard. Steering and claimed occurrences return `not-found`.
|
||||
- `agent.followup(input)` — the `next-turn`/wakeup preset of `send()`: queue an ordinary follow-up turn and wake the driver.
|
||||
- `agent.steer(input)` — the `next-step`/wakeup preset: during prompt admission or an open turn, stage steering for the next safe boundary without dispatching `agent/prompt-submit`; outside that acceptance window, delegate to a woken follow-up. Admission failure leaves staged steering for retry or a later admitted prompt, while cancellation or disposal may discard it.
|
||||
- `agent.steer(input)` — the `next-step`/wakeup preset: submit one identified message and receive its `SteeringReceipt`. During prompt admission or an open turn, the message stages for the next safe request boundary without dispatching `agent/prompt-submit`; outside that acceptance window, it becomes a woken queued prompt. `receipt.outcome` resolves `admitted` with the turn and step only after the loop logs the message, captures it in immutable request history, and commits `step/start`. A turn-concluding tool result, broad cancellation, disposal, or pre-admission failure resolves it `rejected`; `cancel(..., { keepInbox: true })` and non-terminal routing preserve pending delivery. Reliable callers await the receipt, while best-effort UI steering may ignore it.
|
||||
- `agent.inject(input)` — the `next-step`/no-wakeup preset: append model-facing context without running the model; the next request sees a verbatim user-role message whose provenance is carried by the required `input.source`. During prompt admission or an open turn, injection waits in the outbox for the next safe boundary. Outside that acceptance window, it appends immediately without opening a turn; a context-only admission batch takes this fallback if admission closes without a turn, while context staged beside steering remains pending with it. Persistence reacts to `session/event` independently. Injection emits no `agent/inbox/*` event.
|
||||
- `agent.acceptsNextStep` — whether a `next-step` send would currently join prompt admission or the open turn. Use this narrower routing predicate when a caller must choose between steering and a fresh admitted prompt; `status === 'running'` also covers admission exit and turn settlement.
|
||||
- `agent.cancel(cause, options?)` — cancel the active turn and, unless `options.keepInbox`, ALL pending work. Callers must choose the `user | parent` cause explicitly; an active holder copies its discriminant into a detached frozen signal reason before aborting. An effective call emits `agent/cancel-requested` with the cause before clearing queued and steering work; dropped items are reported on `agent/inbox/discard`, and observers may synchronize state but cannot veto cancellation. `keepInbox: true` aborts the turn but preserves queued and steering items (no discard, and un-started work is not dropped). The same-process typed seam adds no runtime validation or compatibility fallback for untyped callers. Repeated active-turn cancellation is first-wins for the signal, and idle cancellation is a safe no-op with no notification. ACP maps to `user`, while in-process parent propagation maps to `parent`. The cause is runtime-only; durable `turn/end` stays coarse `aborted`.
|
||||
|
||||
@@ -64,7 +64,7 @@ Agent *创建* 由实现 `AgentFactory` 的插件（`dsh-agent-loop`）提供，
|
||||
- `agent.reserveTurnAdmission()`：在任何已排队唤醒提示词认领其轮次之前，同步预留空闲边界。已获接纳的提示词拥有优先权，包括同一 tick 内仍在等待唤醒的项，此时预留返回 `undefined`。预留期间，之后发送的项保留其普通 ID、FIFO 位置与唤醒信息；`acceptsNextStep` 保持 false，`inject()` 不受阻塞，`whenIdle()` 将该预留计为活动，返回的释放函数可幂等调用。这项范围有限的协调能力使手动压缩（compaction）等独立持久操作能够在排队提示词从会话派生内容前完成并 flush。
|
||||
- `agent.updateInbox(itemId, action)`：同步编辑或移除一个仍处于待处理状态的 queued 入队项。编辑会替换已冻结的内容，同时保留其 `MessageId`、`InboxItemId`、来源与 FIFO 位置；移除会发出该项的终态 discard。steering 项和已被认领的项会返回 `not-found`。
|
||||
- `agent.followup(input)`：`send()` 的 `next-turn`／wakeup 预设：排队一个普通后续轮次并唤醒驱动器。
|
||||
- `agent.steer(input)`：`next-step`／wakeup 预设：提示词接纳期间或轮次打开时，为下一个安全边界暂存 steering，且不分发 `agent/prompt-submit`；该接收窗口之外则委托给会唤醒的后续轮次。接纳失败会保留暂存的 steering，以供重试或之后获准的提示词使用，而取消或 dispose 可能丢弃它。
|
||||
- `agent.steer(input)`：`next-step`／wakeup 预设：提交一条已有标识的消息，并取得其 `SteeringReceipt`。提示词接纳期间或轮次打开时，消息会为下一个安全请求边界暂存，且不分发 `agent/prompt-submit`；该接收窗口之外则成为会唤醒驱动器的排队提示词。只有循环记录消息、将其捕获到不可变请求历史并提交 `step/start` 后，`receipt.outcome` 才会解析为 `admitted`，并附带轮次与步骤。结束轮次的工具结果、广义取消、dispose（资源释放）或准入前故障会使其解析为 `rejected`；`cancel(..., { keepInbox: true })` 和非终止型路由会保留待处理投递。需要可靠投递的调用方应等待回执；尽力执行的 UI steering 可以忽略它。
|
||||
- `agent.inject(input)`：`next-step`／不唤醒预设：追加面向模型的上下文而不运行模型；下一次请求会看到一条逐字的 user role 消息，其来源由必填的 `input.source` 携带。提示词接纳期间或轮次打开时，注入会在 outbox 中等待下一个安全边界。该接收窗口之外，它会立即追加而不开启轮次；如果接纳结束却未开启轮次，仅含上下文的接纳批次会采用这一回退，而与 steering 一同暂存的上下文则会随其继续待处理。持久化独立地响应 `session/event`。注入不发出 `agent/inbox/*` 事件。
|
||||
- `agent.acceptsNextStep`：当前发送 `next-step` 时，是否会加入提示词接纳或已打开的轮次。当调用方必须在 steering 与新接纳的提示词之间选择时，应使用这一更窄的路由判定；`status === 'running'` 还涵盖接纳收尾与轮次结算阶段。
|
||||
- `agent.cancel(cause, options?)`：取消活动轮次，并在未设置 `options.keepInbox` 时取消全部待处理工作。调用方必须显式选择 `user | parent` 原因；活动持有者会在中止前把其判别字段复制为已分离、冻结的信号原因。有效调用会在清除排队与 steering 工作前，随原因发出 `agent/cancel-requested`；丢弃项在 `agent/inbox/discard` 上报告，观察方可以同步状态，但不能 veto 取消。`keepInbox: true` 会中止轮次，但保留排队与 steering 项（不丢弃，且不删除尚未开始的工作）。同进程类型化 seam 不会为无类型调用方添加运行时校验或兼容回退。重复取消活动轮次时，首个信号生效；空闲取消是安全空操作，不发通知。ACP 映射到 `user`，进程内父传播映射到 `parent`。原因只存在于运行时；持久 `turn/end` 保持粗粒度的 `aborted`。
|
||||
|
||||
@@ -48,9 +48,9 @@ export interface CreateAgentOptions {
|
||||
readonly sessionId: SessionId
|
||||
/**
|
||||
* Session creation metadata: validated absolute `cwd`, `parentSession`
|
||||
* fork lineage, the `seedLength` seed boundary, and the `delegationDepth`
|
||||
* recursion budget. Mirrors the
|
||||
* `cwd`/`parentSession`/`seedLength`/`delegationDepth` fields of
|
||||
* fork lineage, the `seedLength` seed boundary, the coarse `origin`
|
||||
* classification, and the `delegationDepth` recursion budget. Mirrors the
|
||||
* `cwd`/`parentSession`/`seedLength`/`origin`/`delegationDepth` fields of
|
||||
* {@link CreateSessionOptions.meta} in dsh-session (the internal-only
|
||||
* `createdAt`, used when reconstructing a persisted session, is deliberately
|
||||
* excluded — a factory caller never sets it). This is durable session data,
|
||||
@@ -61,6 +61,7 @@ export interface CreateAgentOptions {
|
||||
readonly cwd?: string
|
||||
readonly parentSession?: SessionId
|
||||
readonly seedLength?: number
|
||||
readonly origin?: 'subagent'
|
||||
readonly delegationDepth?: number
|
||||
}
|
||||
/**
|
||||
|
||||
@@ -58,6 +58,20 @@ export type InboxAction =
|
||||
/** Result of applying an inbox action at the synchronous ownership boundary. */
|
||||
export type InboxActionResult = 'applied' | 'not-found'
|
||||
|
||||
/** Final admission outcome for one call to {@link Agent.steer}. */
|
||||
export type SteeringOutcome =
|
||||
| { readonly status: 'admitted'; readonly turn: number; readonly step: number }
|
||||
| { readonly status: 'rejected' }
|
||||
|
||||
/**
|
||||
* Message-owned steering admission receipt. The outcome promise always
|
||||
* resolves: synchronous input validation still throws from {@link Agent.steer},
|
||||
* while lifecycle policy reports non-admission as `rejected`.
|
||||
*/
|
||||
export interface SteeringReceipt {
|
||||
readonly outcome: Promise<SteeringOutcome>
|
||||
}
|
||||
|
||||
/**
|
||||
* Options for the unified {@link Agent.send} primitive over the
|
||||
* (`target` × `wakeup`) matrix. Named presets: {@link Agent.followup}
|
||||
@@ -225,16 +239,18 @@ export interface Agent {
|
||||
followup(message: UserMessage): void
|
||||
|
||||
/**
|
||||
* Submit steering during prompt admission or an open turn — the
|
||||
* `next-step`/wakeup preset of {@link send}. It stages for the next steering
|
||||
* checkpoint before a request or stop decision. If the activity fails before
|
||||
* that boundary, the remainder stays staged without waking the agent; retry
|
||||
* or a later prompt takes it. Outside that window steering falls back to a
|
||||
* woken follow-up turn, while cancellation or disposal may discard pending
|
||||
* steering.
|
||||
* Submit steering with a message-owned admission receipt — the
|
||||
* `next-step`/wakeup preset of {@link send}. During prompt admission or an
|
||||
* open turn, the message waits in the steering FIFO until a committed step
|
||||
* snapshots it; outside that window it enters the ordinary queued FIFO. The
|
||||
* receipt resolves `admitted` only after the message joins that step's
|
||||
* immutable request history, or `rejected` when terminal policy,
|
||||
* cancellation, or disposal discards it first. A non-terminal turn close may
|
||||
* leave it staged for a later admitted prompt without settling the receipt.
|
||||
* @param message - identified steering content and its producer provenance.
|
||||
* @returns the receipt for this exact message's eventual admission outcome.
|
||||
*/
|
||||
steer(message: UserMessage): void
|
||||
steer(message: UserMessage): SteeringReceipt
|
||||
|
||||
/**
|
||||
* Append model-facing context without running the model — the
|
||||
|
||||
@@ -26,7 +26,7 @@ function stubAgent(rawId: string, overrides: Partial<Agent> = {}): Agent {
|
||||
send: () => {},
|
||||
updateInbox: () => 'not-found',
|
||||
followup: () => {},
|
||||
steer: () => {},
|
||||
steer: () => ({ outcome: Promise.resolve({ status: 'rejected' as const }) }),
|
||||
inject: () => {},
|
||||
reserveTurnAdmission: () => undefined,
|
||||
cancel() {},
|
||||
|
||||
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write packages/core/session/README.md
|
||||
README.md: 9c7d41901e6fb0133fff0e210260e5310a025f75
|
||||
README.zh.md: ca1292289901a09b83f9b0a794fa4edc9754b1da
|
||||
README.md: d78dc5bcfe1df2edd01280208f3859eb1b2d6763
|
||||
README.zh.md: 40c58a539d5027f2619b5b2102b94e76f2c73e23
|
||||
|
||||
@@ -12,8 +12,8 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
|
||||
|
||||
### Public API
|
||||
|
||||
- `ctx.sessions.create(id?, { seed?, meta? }?)` validates and detaches durable seed/header data, fills the version and id, defaults `createdAt` to now, publishes the session, and binds it to the calling fiber. Persisted reconstruction supplies its original `createdAt`, `seedLength`, and `delegationDepth`.
|
||||
- `ctx.sessions.flush(session)` dispatches the awaited parallel durability checkpoint through the session's captured scope. Every listener starts and the call waits for all to settle before reporting failure; unpublished, detached, and stale objects reject.
|
||||
- `ctx.sessions.create(id?, { seed?, meta? }?)` validates and detaches durable seed/header data, fills the version and id, defaults `createdAt` to now, publishes the session, and binds it to the calling fiber. Persisted reconstruction supplies its original `createdAt`, `seedLength`, `origin`, and `delegationDepth`.
|
||||
- `ctx.sessions.flush(session)` dispatches the awaited parallel durability checkpoint through the session's captured scope. Every listener starts and the call waits for all to settle before reporting failure; it returns `true` when at least one listener participated and `false` for an empty snapshot, while unpublished, detached, and stale objects reject. A caller that requires durable storage rejects `false` at its own policy boundary.
|
||||
- `findLastMessageTurnEnd(events)` pairs message-triggered starts with their ends and returns the latest matched `turn/end`. Outcome consumers use this fold instead of the raw latest log event because between-turn records and non-message turns have no prompt outcome.
|
||||
- `ctx.sessions.fork(source, boundary?, childSessionId?): Session` — Resolve a live session object or id, select a seed through the inclusive `boundary` event seq (default: current last event), require that prefix to end outside an open turn, and create a live child session with lineage metadata.
|
||||
- `ctx.sessions.get(id: SessionId): Session | undefined`
|
||||
@@ -43,7 +43,7 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
|
||||
- `session.surface` exposes the readonly `SessionSurface` view owned by the session's single incremental surface manager; `replaceGeneration` changes on every committed rewrite.
|
||||
- `session.events` is a cached frozen snapshot invalidated by append; accepted events remain deeply frozen.
|
||||
- `session.seq`, `session.id` — current sequence and readonly typed identity.
|
||||
- `session.header: SessionHeader` — detached, deep-frozen creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`/`seedLength`/`delegationDepth`). Construction validates the durable record and requires its id to match `session.id`.
|
||||
- `session.header: SessionHeader` — detached, deep-frozen creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`/`seedLength`/`origin`/`delegationDepth`). `origin: 'subagent'` is a coarse product classification, not a continuation capability. Construction validates the durable record and requires its id to match `session.id`.
|
||||
|
||||
### Lossless JSON utilities
|
||||
|
||||
@@ -89,7 +89,7 @@ Every `SessionEvent` carries two optional top-level fields (structural metadata)
|
||||
|
||||
### Metadata types (`types.ts`)
|
||||
|
||||
- `SessionHeader` — session metadata written once when published as `Session.header`, where detachment and deep-freezing enforce immutability at runtime: `{ version, id, createdAt, cwd?, parentSession?, seedLength?, delegationDepth? }`. Persistence loaders may return mutable detached copies of the same data type. Owned here (beside `SessionId`) because `Session.header` is typed by it; persistence backends re-export it rather than own it (which would force a package cycle).
|
||||
- `SessionHeader` — session metadata written once when published as `Session.header`, where detachment and deep-freezing enforce immutability at runtime: `{ version, id, createdAt, cwd?, parentSession?, seedLength?, origin?, delegationDepth? }`. Persistence loaders may return mutable detached copies of the same data type. Owned here (beside `SessionId`) because `Session.header` is typed by it; persistence backends re-export it rather than own it (which would force a package cycle).
|
||||
|
||||
### Extension points
|
||||
|
||||
|
||||
@@ -12,8 +12,8 @@
|
||||
|
||||
### 公共 API
|
||||
|
||||
- `ctx.sessions.create(id?, { seed?, meta? }?)` 校验持久种子／头部数据并生成脱离副本，补齐版本和 id，在未提供 `createdAt` 时使用当前时间，发布会话并将其绑定到调用方 fiber。持久化重建会提供原始的 `createdAt`、`seedLength` 和 `delegationDepth`。
|
||||
- `ctx.sessions.flush(session)` 通过会话捕获的作用域分发受等待的并行持久性检查点。每个监听器都会启动；调用会等待全部结算后才报告失败。未发布、已脱离和陈旧的对象会被拒绝。
|
||||
- `ctx.sessions.create(id?, { seed?, meta? }?)` 校验持久种子／头部数据并生成脱离副本，补齐版本和 id，在未提供 `createdAt` 时使用当前时间，发布会话并将其绑定到调用方 fiber。持久化重建会提供原始的 `createdAt`、`seedLength`、`origin` 和 `delegationDepth`。
|
||||
- `ctx.sessions.flush(session)` 通过会话捕获的作用域分发受等待的并行持久性检查点。每个监听器都会启动；调用会等待全部结算后才报告失败；至少一个监听器参与时返回 `true`，监听器快照为空时返回 `false`，而未发布、已脱离和陈旧的对象会被拒绝。要求持久化存储的调用方应在自己的策略边界拒绝 `false`。
|
||||
- `findLastMessageTurnEnd(events)` 将由消息触发的开始与结束配对，并返回最近匹配的 `turn/end`。结果消费方使用该折叠逻辑，而不直接取日志中最近的事件，因为轮次间记录和非消息轮次没有提示词结果。
|
||||
- `ctx.sessions.fork(source, boundary?, childSessionId?): Session`：解析实时会话对象或 id，选取截至 `boundary` 事件序号（含该事件）的种子（默认为当前最后一个事件），要求所选前缀结束时没有开放轮次，再创建带谱系元数据的实时子会话。
|
||||
- `ctx.sessions.get(id: SessionId): Session | undefined`
|
||||
@@ -43,7 +43,7 @@
|
||||
- `session.surface` 暴露只读 `SessionSurface` 视图，由会话唯一的增量 surface 管理器所有；每次提交重写，`replaceGeneration` 都会变化。
|
||||
- `session.events` 是按追加失效的缓存冻结快照；已接受事件保持深度冻结。
|
||||
- `session.seq`、`session.id`：当前序号和只读类型化身份。
|
||||
- `session.header: SessionHeader`：脱离、深冻结的创建元数据（`version`、`id`、`createdAt`，以及可选的 `cwd`／`parentSession`／`seedLength`／`delegationDepth`）。构造时会校验持久记录，并要求其中的 id 与 `session.id` 一致。
|
||||
- `session.header: SessionHeader`：脱离、深冻结的创建元数据（`version`、`id`、`createdAt`，以及可选的 `cwd`／`parentSession`／`seedLength`／`origin`／`delegationDepth`）。`origin: 'subagent'` 是粗粒度产品分类，不代表具备继续执行能力。构造时会校验持久记录，并要求其中的 id 与 `session.id` 一致。
|
||||
|
||||
### 无损 JSON 工具
|
||||
|
||||
@@ -89,7 +89,7 @@
|
||||
|
||||
### 元数据类型（`types.ts`）
|
||||
|
||||
- `SessionHeader`：会话元数据，在发布为 `Session.header` 时写入一次；脱离和深冻结保证运行时不可变：`{ version, id, createdAt, cwd?, parentSession?, seedLength?, delegationDepth? }`。持久化 loader 可返回相同数据类型的可变脱离副本。该类型由此包与 `SessionId` 一同所有，因为 `Session.header` 以它为类型；持久化后端只是重新导出而不拥有它，否则会形成包循环依赖。
|
||||
- `SessionHeader`：会话元数据，在发布为 `Session.header` 时写入一次；脱离和深冻结保证运行时不可变：`{ version, id, createdAt, cwd?, parentSession?, seedLength?, origin?, delegationDepth? }`。持久化 loader 可返回相同数据类型的可变脱离副本。该类型由此包与 `SessionId` 一同所有，因为 `Session.header` 以它为类型；持久化后端只是重新导出而不拥有它，否则会形成包循环依赖。
|
||||
|
||||
### 扩展点
|
||||
|
||||
|
||||
@@ -93,8 +93,7 @@ declare module 'cordis' {
|
||||
'session/event'(this: Scoped<Session>, session: Session, event: SessionEvent): void
|
||||
/**
|
||||
* Awaited parallel durability checkpoint: every listener runs and the
|
||||
* caller awaits all of them, with no waterfall veto. Dispatch through
|
||||
* {@link SessionStore.flush}. Scope-filtered dispatch
|
||||
* caller awaits all of them, with no waterfall veto. Scope-filtered dispatch
|
||||
* (`@deepseek-ai/dsh-scope`) reuses the session's owner scope.
|
||||
* @param session - the session whose buffered events must reach durable storage.
|
||||
* @dshScopeScan unsupported
|
||||
@@ -139,6 +138,9 @@ function snapshotSessionHeader(id: SessionId, source?: SessionHeader): SessionHe
|
||||
&& (typeof record.seedLength !== 'number' || !Number.isSafeInteger(record.seedLength) || record.seedLength < 0)) {
|
||||
throw new Error('session header seedLength must be a non-negative safe integer')
|
||||
}
|
||||
if (record.origin !== undefined && record.origin !== 'subagent') {
|
||||
throw new Error('session header origin must be "subagent"')
|
||||
}
|
||||
if (record.delegationDepth !== undefined
|
||||
&& (typeof record.delegationDepth !== 'number' || !Number.isSafeInteger(record.delegationDepth) || record.delegationDepth < 0)) {
|
||||
throw new Error('session header delegationDepth must be a non-negative safe integer')
|
||||
@@ -838,6 +840,7 @@ export class SessionStore extends Service {
|
||||
...meta?.cwd === undefined ? {} : { cwd: meta.cwd },
|
||||
...meta?.parentSession === undefined ? {} : { parentSession: meta.parentSession },
|
||||
...meta?.seedLength === undefined ? {} : { seedLength: meta.seedLength },
|
||||
...meta?.origin === undefined ? {} : { origin: meta.origin },
|
||||
...meta?.delegationDepth === undefined ? {} : { delegationDepth: meta.delegationDepth },
|
||||
}
|
||||
return new Session(sessionId, seed, header)
|
||||
@@ -969,10 +972,11 @@ export class SessionStore extends Service {
|
||||
* raw `ctx.parallel('session/flush', …)` — one owner, one spelling, and the
|
||||
* scoped-dispatch invariant can pin it.
|
||||
* @param session - the session whose buffered events must reach durable storage.
|
||||
* @returns resolves when every flush listener has settled; after all settle,
|
||||
* rejects with the first registered listener failure if any listener failed.
|
||||
* @returns whether at least one durability listener participated, after every
|
||||
* listener has settled successfully.
|
||||
* @throws the first registered listener failure after every listener settles.
|
||||
*/
|
||||
async flush(session: Session): Promise<void> {
|
||||
async flush(session: Session): Promise<boolean> {
|
||||
const { carrier } = this.liveEntryFor(session)
|
||||
const callbackArgs: unknown[] = [session]
|
||||
const callbacks = collectSessionCallbacks(this.ctx, [carrier, 'session/flush', session])
|
||||
@@ -988,6 +992,7 @@ export class SessionStore extends Service {
|
||||
}))
|
||||
const failure = results.find((result): result is PromiseRejectedResult => result.status === 'rejected')
|
||||
if (failure !== undefined) throw failure.reason
|
||||
return callbacks.length > 0
|
||||
}
|
||||
|
||||
/** Return the exact live entry; detached/prepared objects reject. */
|
||||
|
||||
@@ -59,6 +59,11 @@ export interface SessionHeader {
|
||||
* boundary lets resume and replay distinguish parent history from child work.
|
||||
*/
|
||||
readonly seedLength?: number
|
||||
/**
|
||||
* Coarse product classification for a session created as a subagent child.
|
||||
* This is presentation metadata, not proof that the child is continuable.
|
||||
*/
|
||||
readonly origin?: 'subagent'
|
||||
/**
|
||||
* Delegation depth: absent (zero) for a top-level session, parent depth + 1
|
||||
* for a subagent child. Persisted so a recursion budget survives restart and
|
||||
@@ -84,6 +89,7 @@ export interface CreateSessionOptions {
|
||||
readonly parentSession?: SessionId
|
||||
readonly createdAt?: number
|
||||
readonly seedLength?: number
|
||||
readonly origin?: 'subagent'
|
||||
readonly delegationDepth?: number
|
||||
}
|
||||
}
|
||||
|
||||
@@ -80,6 +80,24 @@ describe('session dispatch carriers', () => {
|
||||
})
|
||||
|
||||
describe('sessions.flush()', () => {
|
||||
it('allows an ordinary flush with no listeners', async () => {
|
||||
const ctx = await mount()
|
||||
const session = ctx.sessions.create()
|
||||
|
||||
await expect(ctx.sessions.flush(session)).resolves.toBe(false)
|
||||
})
|
||||
|
||||
it('reports a participating listener after it succeeds', async () => {
|
||||
const ctx = await mount()
|
||||
const session = ctx.sessions.create()
|
||||
const flushed: Session[] = []
|
||||
ctx.on('session/flush', current => void flushed.push(current))
|
||||
|
||||
await expect(ctx.sessions.flush(session)).resolves.toBe(true)
|
||||
|
||||
expect(flushed).toEqual([session])
|
||||
})
|
||||
|
||||
it('dispatches session/flush with the owning carrier and awaits all listeners', async () => {
|
||||
const ctx = await mount()
|
||||
const scope = await mintScope(ctx, 'owner')
|
||||
|
||||
@@ -1251,15 +1251,16 @@ describe('SessionStore', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('attaches delegationDepth from meta to the header', async () => {
|
||||
it('attaches subagent origin and delegationDepth from meta to the header', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('delegated-child'), {
|
||||
meta: { parentSession: SessionId('parent'), delegationDepth: 2 },
|
||||
meta: { parentSession: SessionId('parent'), origin: 'subagent', delegationDepth: 2 },
|
||||
})
|
||||
expect(session.header).toMatchObject({
|
||||
id: 'delegated-child',
|
||||
parentSession: 'parent',
|
||||
origin: 'subagent',
|
||||
delegationDepth: 2,
|
||||
})
|
||||
})
|
||||
@@ -1278,6 +1279,7 @@ describe('SessionStore', () => {
|
||||
{ meta: { seedLength: '1' }, error: /seedLength must be a non-negative safe integer/ },
|
||||
{ meta: { seedLength: 0.5 }, error: /seedLength must be a non-negative safe integer/ },
|
||||
{ meta: { seedLength: -1 }, error: /seedLength must be a non-negative safe integer/ },
|
||||
{ meta: { origin: 'fork' }, error: /origin must be "subagent"/ },
|
||||
{ meta: { delegationDepth: '1' }, error: /delegationDepth must be a non-negative safe integer/ },
|
||||
{ meta: { delegationDepth: 0.5 }, error: /delegationDepth must be a non-negative safe integer/ },
|
||||
{ meta: { delegationDepth: -1 }, error: /delegationDepth must be a non-negative safe integer/ },
|
||||
|
||||
@@ -23,7 +23,7 @@ describe('gen-tool-catalog collectToolCatalog', () => {
|
||||
it('boots every shipped tool package and harvests its model-facing schemas', async () => {
|
||||
const catalog = await collectToolCatalog()
|
||||
const names = catalog.flatMap(entry => entry.schemas.map(s => s.name)).sort()
|
||||
expect(names).toEqual(['ask_user_question', 'bash', 'bash', 'cordis_inspect', 'cordis_mount', 'cordis_unmount', 'create_goal', 'edit', 'exit_plan_mode', 'get_goal', 'glob', 'grep', 'lsp', 'ralph', 'read', 'run_code', 'session_event_read', 'session_event_search', 'session_event_trace', 'session_search', 'session_trace', 'skill', 'str_replace_editor', 'subagent', 'task_kill', 'task_list', 'task_output', 'terminal_close', 'terminal_list', 'terminal_open', 'terminal_read', 'terminal_send', 'terminal_signal', 'todo_write', 'update_goal', 'web_fetch', 'web_search', 'workflow', 'write'])
|
||||
expect(names).toEqual(['ask_user_question', 'bash', 'bash', 'cordis_inspect', 'cordis_mount', 'cordis_unmount', 'create_goal', 'edit', 'exit_plan_mode', 'get_goal', 'glob', 'grep', 'list_agents', 'lsp', 'ralph', 'read', 'report', 'run_code', 'send_message', 'session_event_read', 'session_event_search', 'session_event_trace', 'session_search', 'session_trace', 'skill', 'str_replace_editor', 'subagent', 'task_kill', 'task_list', 'task_output', 'terminal_close', 'terminal_list', 'terminal_open', 'terminal_read', 'terminal_send', 'terminal_signal', 'todo_write', 'update_goal', 'web_fetch', 'web_search', 'workflow', 'write'])
|
||||
// Every tool carries a JSON-Schema `parameters` object (what the model sees).
|
||||
for (const entry of catalog) {
|
||||
for (const schema of entry.schemas) {
|
||||
@@ -43,10 +43,15 @@ describe('gen-tool-catalog collectToolCatalog', () => {
|
||||
expect(status?.enum).toEqual(['pending', 'in_progress', 'completed'])
|
||||
})
|
||||
|
||||
it('attributes each package with a source pointer that names its index', async () => {
|
||||
it('attributes each harvested tool with its registering plugin source', async () => {
|
||||
const catalog = await collectToolCatalog()
|
||||
const bash = catalog.find(entry => entry.pkg === '@deepseek-ai/dsh-tool-bash')
|
||||
expect(bash?.source).toBe('packages/bash/tool-bash/src/index.ts')
|
||||
expect(bash?.sources.bash).toBe('packages/bash/tool-bash/src/index.ts')
|
||||
const control = catalog.find(entry => entry.pkg === '@deepseek-ai/dsh-tool-subagent-control')
|
||||
expect(control?.sources).toEqual({
|
||||
list_agents: 'packages/subagent/tool-subagent-control/src/list-agents.ts',
|
||||
send_message: 'packages/subagent/tool-subagent-control/src/index.ts',
|
||||
})
|
||||
})
|
||||
|
||||
it('harvests search tools without depending on the generator process PATH', async () => {
|
||||
@@ -90,7 +95,7 @@ describe('gen-tool-catalog render', () => {
|
||||
const catalog: ToolCatalog = [
|
||||
{
|
||||
pkg: '@deepseek-ai/dsh-tool-demo',
|
||||
source: 'packages/demo/tool-demo/src/index.ts',
|
||||
sources: { demo: 'packages/demo/tool-demo/src/index.ts' },
|
||||
requires: ['ctx.tools'],
|
||||
writes: ['tool/result'],
|
||||
schemas: [{ name: 'demo', description: 'A demo tool.', parameters: { type: 'object', properties: {} } }],
|
||||
|
||||
Reference in New Issue
Block a user