mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
fix(schedule): close concurrent durability gaps
This commit is contained in:
pku-xht
Tianyi Cui
@@ -29,7 +29,7 @@ The version-1 `schedule/change` stream is the only durable Schedule authority. A
|
||||
|
||||
The current rule accepts a non-empty prompt and exactly one positive safe-integer `after_seconds`. Its record is `{ id, kind: 'after', prompt, afterSeconds, scheduledAt }`; dispatch stores only the id because the record already fixes its occurrence. `at`, `every_seconds`, `cron`, and `time_zone` are rejected rather than hidden in unused fields. Tool values derive `scheduled` or `overdue` and always include `deliveryMode: 'session-local'`.
|
||||
|
||||
Every tool operation that reads or decides from the fold first awaits `ctx.sessions.flush(session)`. Create may reject input-shape failures before this preflight; after a successful preflight it allocates an id, appends create, and waits for a second barrier. Delete preflights before deciding whether an id is active and waits for a second barrier only when it appends. List and unknown or finished delete never answer from an unconfirmed live suffix. A failed barrier returns `persistence_uncertain` rather than guessing whether an eager write committed.
|
||||
An Agent-scoped FIFO serializes each accepted management transaction and the live owner's due transaction from preflight through any post-append barrier. Every tool operation that reads or decides from the fold first awaits `ctx.sessions.flush(session)`. Create may reject input-shape failures before entering the FIFO; after a successful preflight it allocates an id, appends create, and waits for a second barrier. Delete validates its id before the FIFO, then preflights before deciding whether the id is active and waits for a second barrier only when it appends. List and unknown or finished delete never answer from an unconfirmed live suffix or observe a dispatch before its own barrier. A failed barrier returns `persistence_uncertain` rather than guessing whether an eager write committed.
|
||||
|
||||
Every successful management preflight also asks the live owner to recompute. This closes the recovery path where create appended successfully but its post-append barrier rejected: a later list can confirm the coordinator's retained batch, return the active record, and arm its timer without a Schedule-specific retry loop.
|
||||
|
||||
@@ -41,7 +41,7 @@ The persistence coordinator supplies that acknowledgement only after its write p
|
||||
|
||||
### Live delivery lifecycle
|
||||
|
||||
The Agent-scoped owner derives its earliest target from the durable fold. Long targets use bounded timer segments, and every wake reads the wall clock again, so a rollback cannot fire early and a forward jump becomes overdue. If a turn or another maintenance task already owns the Agent, `runMaintenance()` rejects the claim; the record stays active and one `whenIdle()` wait triggers a later retry. A rejected persistence preflight also leaves the record active, but no private retry timer runs; later Agent activity reaching idle or a successful Schedule management preflight asks the owner to try again.
|
||||
The Agent-scoped owner derives its earliest target from the durable fold. Long targets use bounded timer segments, and every wake reads the wall clock again, so a rollback cannot fire early and a forward jump becomes overdue. If a turn or another maintenance task already owns the Agent, `runMaintenance()` rejects the claim; the record stays active and one `whenIdle()` wait triggers a later retry. A rejected persistence preflight or contained framing/synchronous-enqueue failure also leaves the record active, but no private retry timer runs; later Agent activity reaching idle or a successful Schedule management preflight asks the owner to try again.
|
||||
|
||||
The accepted path first clears pending persistence and claims the true idle phase through `runMaintenance()`. Inside that task it refolds the exact Session suffix so a direct management mutation that won the claim race cannot be followed by a stale dispatch, samples the decision clock once, constructs the complete fixed reminder frame with JSON-escaped id and prompt, synchronously queues one `followup()`, and appends the id-only dispatch. Waking input remains parked until maintenance settles, so the driver cannot claim the message before dispatch enters the log; only after the task releases the phase does the owner wait for the dispatch barrier. A framing or synchronous enqueue failure is contained and appends no dispatch. An append failure faults that owner because the message may already be queued. A later prompt-admission, request-checkpoint, or model failure cannot retract a dispatch.
|
||||
|
||||
|
||||
@@ -29,7 +29,7 @@ Status: implemented
|
||||
|
||||
当前规则接受非空 prompt 与恰好一个正 safe-integer `after_seconds`。record 形状是 `{ id, kind: 'after', prompt, afterSeconds, scheduledAt }`；dispatch 只保存 id，因为 record 已经唯一确定 occurrence。`at`、`every_seconds`、`cron` 与 `time_zone` 会被拒绝，不会作为未使用字段隐藏在协议中。工具 value 派生 `scheduled` 或 `overdue`，并始终包含 `deliveryMode: 'session-local'`。
|
||||
|
||||
每项从 fold 读取或作出判断的工具操作都会先等待 `ctx.sessions.flush(session)`。create 可以在这次 preflight 前拒绝只依赖输入 shape 的失败；preflight 成功后才分配 id、追加 create，并等待第二个 barrier。delete 在判断 id 是否活动前先 preflight，只有实际追加时才等待第二个 barrier。list 与未知或已终结 delete 绝不会从未确认的 live 后缀作答。barrier 失败会返回 `persistence_uncertain`，而不是猜测 eager write 是否已经提交。
|
||||
一个 Agent-scoped FIFO 会将每项已接纳的管理事务与 live owner 的到期事务从 preflight 到任何 post-append barrier 全程串行化。每项从 fold 读取或作出判断的工具操作都会先等待 `ctx.sessions.flush(session)`。create 可以在进入 FIFO 前拒绝只依赖输入 shape 的失败；preflight 成功后才分配 id、追加 create，并等待第二个 barrier。delete 在进入 FIFO 前验证其 id，随后在判断 id 是否活动前先 preflight，只有实际追加时才等待第二个 barrier。list 与未知或已终结 delete 绝不会从未确认的 live 后缀作答，也不会在自身的 barrier 前观察到 dispatch。barrier 失败会返回 `persistence_uncertain`，而不是猜测 eager write 是否已经提交。
|
||||
|
||||
每次成功的管理 preflight 也会要求 live owner 重新计算。这闭合了 create 已成功追加、但 post-append barrier 拒绝时的恢复路径：后续 list 可以确认 coordinator 保留的 batch、返回活动 record，并在没有 Schedule 私有重试循环的情况下 arm timer。
|
||||
|
||||
@@ -41,7 +41,7 @@ persistence coordinator 只有在写路径完全停稳后才给出该确认。li
|
||||
|
||||
### Live 交付生命周期
|
||||
|
||||
Agent-scoped owner 从持久 fold 派生最早目标。超长目标使用有界 timer 分段，每次 wake 都重新读取墙钟，因此回拨不会提前触发，前跳则会形成 overdue。如果 agent 已被某个轮次或另一项 maintenance task 占用，`runMaintenance()` 会拒绝此次认领；record 保持活动，并由一个 `whenIdle()` wait 触发稍后的重试。被拒绝的 persistence preflight 同样会让 record 保持活动，但不会运行私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 会要求 owner 再次尝试。
|
||||
Agent-scoped owner 从持久 fold 派生最早目标。超长目标使用有界 timer 分段，每次 wake 都重新读取墙钟，因此回拨不会提前触发，前跳则会形成 overdue。如果 agent 已被某个轮次或另一项 maintenance task 占用，`runMaintenance()` 会拒绝此次认领；record 保持活动，并由一个 `whenIdle()` wait 触发稍后的重试。被拒绝的 persistence preflight 或被收容的 framing／同步入队失败同样会让 record 保持活动，但不会运行私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 会要求 owner 再次尝试。
|
||||
|
||||
获得准入的路径会先清空 pending persistence，并通过 `runMaintenance()` 认领真正的 idle phase。该任务会重新折叠确切的 Session 后缀，从而确保在认领竞态中胜出的直接管理变更之后不会跟随陈旧 dispatch；随后只采样一次 decision clock，使用 JSON-escaped id 与 prompt 构造完整固定 reminder frame，同步排入一次 `followup()`，再追加只含 id 的 dispatch。触发唤醒的 input 会保持 parked，直到 maintenance 结束，因此 driver 无法在 dispatch 进入 log 前认领消息；只有该任务释放 phase 后，owner 才会等待 dispatch barrier。framing 或同步入队失败会被收容，且不会追加 dispatch。append 失败会使该 owner fault，因为消息可能已经入队。后续 prompt admission、request checkpoint 或模型失败都不能撤回 dispatch。
|
||||
|
||||
|
||||
@@ -403,6 +403,9 @@ export class Session implements SessionFace {
|
||||
try {
|
||||
const { result } = await this.history({ beforeSeq: loading.beforeSeq, maxMessages: PAGE_MESSAGES })
|
||||
if (this.loadingOlder !== loading) return
|
||||
// A concurrent gap repair may replace the window with a newer tail page.
|
||||
// The captured older page no longer adjoins that window and must be dropped.
|
||||
if (this.baseSeq !== loading.beforeSeq) return
|
||||
if (!result.ok) return // keep the window as-is; do not overwrite openError (open already succeeded)
|
||||
const older = result.value.events
|
||||
if (older.length === 0) {
|
||||
|
||||
@@ -286,6 +286,40 @@ describe('late event views', () => {
|
||||
])
|
||||
})
|
||||
|
||||
it('drops an older page after a concurrent gap repair advances the window base', async () => {
|
||||
const { api, session } = makeSession()
|
||||
api.onHistory = () => histResponse(logRange(50, 100), true)
|
||||
await session.open()
|
||||
|
||||
const repair = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
|
||||
const page = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
|
||||
api.onHistory = payload => payload.beforeSeq === undefined ? repair.promise : page.promise
|
||||
const gapTail = ev.user(200, '修复后的新窗口')
|
||||
session.handleMuxEnvelope('gap' as never, {
|
||||
type: 'session/event', sessionId: SID, event: gapTail,
|
||||
})
|
||||
await vi.waitFor(() => {
|
||||
expect(api.callsOf('session.history')).toHaveLength(2)
|
||||
})
|
||||
|
||||
const loading = session.loadOlder()
|
||||
repair.resolve(ok({
|
||||
events: entries([...logRange(150, 200), gapTail]) as never[],
|
||||
hasMore: true,
|
||||
}))
|
||||
await vi.waitFor(() => {
|
||||
expect(session.getSnapshot().nodes.map(node => node.seq)).toEqual([200])
|
||||
})
|
||||
|
||||
page.resolve(ok({
|
||||
events: entries([...logRange(0, 44), ...plainTurn(44, 0, '陈旧问题', '陈旧回答')]) as never[],
|
||||
hasMore: false,
|
||||
}))
|
||||
await loading
|
||||
expect(session.getSnapshot()).toMatchObject({ hasMore: true })
|
||||
expect(session.getSnapshot().nodes.map(node => node.seq)).toEqual([200])
|
||||
})
|
||||
|
||||
it('resyncs when repeated older-page late views disagree on event identity', async () => {
|
||||
const { api, session } = makeSession()
|
||||
const newer = logRange(6, 12)
|
||||
|
||||
@@ -22,7 +22,7 @@ Replay rejects unknown versions, extra fields, reused ids, and delete or dispatc
|
||||
|
||||
The generated [tool catalog](../../../docs/tool-catalog.md) owns the argument and output schemas for `schedule_create`, `schedule_list`, and `schedule_delete`. Their canonical values use camelCase record fields even though model input uses `after_seconds`.
|
||||
|
||||
`schedule_create` validates shape-only failures before persistence, then checkpoints, allocates a never-reused id, appends the create, and checkpoints again. `schedule_list` returns every active record in create order with `state: "scheduled" | "overdue"` and `deliveryMode: "session-local"`. `schedule_delete` rejects an empty or whitespace-padded id before persistence and appends only for an active id; an unknown or terminal id returns `{ id, deleted: false, code: "schedule_not_found" }` after its preflight.
|
||||
One Agent-scoped queue serializes each accepted management transaction and the live owner's due transaction from preflight through any post-append barrier. Direct callers therefore cannot interleave a fold with another Schedule mutation or observe a dispatch before its own barrier. `schedule_create` validates shape-only failures before entering that queue, then checkpoints, allocates a never-reused id, appends the create, and checkpoints again. `schedule_list` returns every active record in create order with `state: "scheduled" | "overdue"` and `deliveryMode: "session-local"`. `schedule_delete` rejects an empty or whitespace-padded id before entering the queue and appends only for an active id; an unknown or terminal id returns `{ id, deleted: false, code: "schedule_not_found" }` after its preflight.
|
||||
|
||||
Every successful management preflight also asks the live owner to recompute. This matters after a create or delete barrier returned `persistence_uncertain`: a later list or mutation can confirm the retained batch and immediately arm or retire the now-durable record without a private persistence-retry timer.
|
||||
|
||||
@@ -79,7 +79,7 @@ The reminder appends after existing history and preserves its reusable prefix. I
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Session-local delivery only** — a reminder runs on time only while its original session is live; a cold session receives no external notification and processes an overdue record only after resume.
|
||||
- **Activity-driven persistence retry** — a rejected due preflight leaves the overdue record active but starts no private retry timer; the owner retries after later Agent activity reaches idle or a successful Schedule management preflight asks it to recompute.
|
||||
- **Activity-driven retry** — a rejected due preflight or contained framing/enqueue failure leaves the overdue record active but starts no private retry timer; the owner retries after later Agent activity reaches idle or a successful Schedule management preflight asks it to recompute.
|
||||
- **After-only protocol** — version 1 rejects `at`, `every_seconds`, `cron`, and `time_zone`; those rules require later protocol variants rather than hidden compatibility fields.
|
||||
- **Narrow crash duplicate window** — a crash after synchronous followup admission but before the dispatch checkpoint can repeat the reminder after recovery; the package does not claim model completion, user acknowledgement, or exactly-once external effects.
|
||||
- **Load-order boundary** — the plugin does not scan or adopt agents that were already live when it loaded.
|
||||
|
||||
@@ -22,7 +22,7 @@
|
||||
|
||||
生成的[工具目录](../../../docs/tool-catalog.md)负责 `schedule_create`、`schedule_list` 和 `schedule_delete` 的参数与输出 schema。虽然模型输入使用 `after_seconds`，但其规范值中的记录字段使用 camelCase。
|
||||
|
||||
`schedule_create` 会在持久化前验证只依赖输入形状的失败，随后执行检查点、分配永不复用的 id、追加 create，再次执行检查点。`schedule_list` 按创建顺序返回所有活动记录，其中包含 `state: "scheduled" | "overdue"` 与 `deliveryMode: "session-local"`。`schedule_delete` 会在持久化前拒绝空 id 或前后带空白的 id，并只为活动 id 追加事件；未知或已终结的 id 会在 preflight（预检）后返回 `{ id, deleted: false, code: "schedule_not_found" }`。
|
||||
一条 Agent-scoped 队列会将每项已接纳的管理事务与 live owner 的到期事务从 preflight 到任何 post-append barrier 全程串行化。因此，直接调用方无法让一次 fold 与另一项 Schedule 变更交错，也无法在自身的 barrier 前观察到 dispatch。`schedule_create` 会在进入该队列前验证只依赖输入形状的失败，随后执行检查点、分配永不复用的 id、追加 create，再次执行检查点。`schedule_list` 按创建顺序返回所有活动记录，其中包含 `state: "scheduled" | "overdue"` 与 `deliveryMode: "session-local"`。`schedule_delete` 会在进入该队列前拒绝空 id 或前后带空白的 id，并只为活动 id 追加事件；未知或已终结的 id 会在 preflight（预检）后返回 `{ id, deleted: false, code: "schedule_not_found" }`。
|
||||
|
||||
每次成功的管理 preflight 还会要求 live owner 重新计算。这对 create 或 delete barrier 返回 `persistence_uncertain` 的情况很重要：后续 list 或 mutation 可以确认保留的 batch，并立即 arm 或退役此时已持久化的 record，而无需私有 persistence retry timer。
|
||||
|
||||
@@ -79,7 +79,7 @@ reminder_prompt_json: <JSON.stringify(prompt)>
|
||||
## 已知限制与暂缓事项
|
||||
|
||||
- **仅限会话本地交付**：提醒只有在原会话 live 时才能准时运行；cold 会话不会收到外部通知，只有恢复后才会处理 overdue 记录。
|
||||
- **活动驱动的持久化重试**：到期 preflight 被拒绝后，overdue 记录仍保持活动，但不会启动私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 要求 owner 重新计算后，owner 会重试。
|
||||
- **活动驱动的重试**：到期 preflight 被拒绝或 framing／入队失败被收容后，overdue 记录仍保持活动，但不会启动私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 要求 owner 重新计算后，owner 会重试。
|
||||
- **仅支持 after 协议**：版本 1 拒绝 `at`、`every_seconds`、`cron` 和 `time_zone`；这些规则需要后续协议变体，而不是隐藏的兼容字段。
|
||||
- **存在狭窄的崩溃重复窗口**：同步 `followup` 获得准入后、dispatch 检查点完成前发生崩溃，可能使提醒在恢复后重复；此包不承诺模型完成、用户确认或外部副作用恰好一次。
|
||||
- **加载顺序边界**：插件不会扫描或接管加载时已经 live 的 agent。
|
||||
|
||||
@@ -9,6 +9,7 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
||||
import type { AfterScheduleRecord } from './types.ts'
|
||||
import { foldScheduleEvents, renderReminderFraming, ScheduleLogError } from './domain.ts'
|
||||
import { flushSchedulePersistence } from './persistence.ts'
|
||||
import { runScheduleTransaction } from './transaction.ts'
|
||||
|
||||
/** Largest delay that Node timers represent without clamping. */
|
||||
export const MAX_TIMER_DELAY_MS = 2_147_483_647
|
||||
@@ -102,7 +103,7 @@ export class ScheduleOwner {
|
||||
private async runRequested(): Promise<void> {
|
||||
while (this.requested && !this.stopping && !this.faulted) {
|
||||
this.requested = false
|
||||
await this.driveOnce()
|
||||
await runScheduleTransaction(this.agent, () => this.driveOnce())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -18,6 +18,7 @@ import {
|
||||
scheduleView,
|
||||
} from './domain.ts'
|
||||
import { flushSchedulePersistence } from './persistence.ts'
|
||||
import { runScheduleTransaction } from './transaction.ts'
|
||||
import type {
|
||||
AfterScheduleRecord,
|
||||
PersistenceUncertainError,
|
||||
@@ -255,31 +256,33 @@ export function registerScheduleTools(
|
||||
if (exec.agent !== agent) return internalError()
|
||||
const invalid = validateCreateArgs(args)
|
||||
if (invalid !== undefined) return invalid
|
||||
const uncertain = await preflight(rootCtx, agent, 'create')
|
||||
if (uncertain !== undefined) return uncertain
|
||||
notifyDurableChange()
|
||||
const folded = foldForTool(agent)
|
||||
if (isToolError(folded)) return folded
|
||||
const id = allocateScheduleId(folded)
|
||||
let record: AfterScheduleRecord
|
||||
try {
|
||||
record = createAfterScheduleRecord(id, args.prompt, args.after_seconds, Date.now())
|
||||
} catch (error: unknown) {
|
||||
return error instanceof ScheduleInputError ? inputError(error) : internalError()
|
||||
}
|
||||
try {
|
||||
agent.session.append('schedule/change', {
|
||||
version: 1,
|
||||
operation: 'create',
|
||||
schedule: record,
|
||||
})
|
||||
} catch {
|
||||
return internalError()
|
||||
}
|
||||
const barrier = await preflight(rootCtx, agent, 'create', id)
|
||||
if (barrier !== undefined) return barrier
|
||||
notifyDurableChange()
|
||||
return scheduleView(record, Date.now())
|
||||
return runScheduleTransaction(agent, async () => {
|
||||
const uncertain = await preflight(rootCtx, agent, 'create')
|
||||
if (uncertain !== undefined) return uncertain
|
||||
notifyDurableChange()
|
||||
const folded = foldForTool(agent)
|
||||
if (isToolError(folded)) return folded
|
||||
const id = allocateScheduleId(folded)
|
||||
let record: AfterScheduleRecord
|
||||
try {
|
||||
record = createAfterScheduleRecord(id, args.prompt, args.after_seconds, Date.now())
|
||||
} catch (error: unknown) {
|
||||
return error instanceof ScheduleInputError ? inputError(error) : internalError()
|
||||
}
|
||||
try {
|
||||
agent.session.append('schedule/change', {
|
||||
version: 1,
|
||||
operation: 'create',
|
||||
schedule: record,
|
||||
})
|
||||
} catch {
|
||||
return internalError()
|
||||
}
|
||||
const barrier = await preflight(rootCtx, agent, 'create', id)
|
||||
if (barrier !== undefined) return barrier
|
||||
notifyDurableChange()
|
||||
return scheduleView(record, Date.now())
|
||||
})
|
||||
},
|
||||
presentCall: args => present('Create reminder', 'other', args.prompt),
|
||||
})))
|
||||
@@ -291,13 +294,15 @@ export function registerScheduleTools(
|
||||
output: { schema: LIST_OUTPUT_SCHEMA, render: renderValue },
|
||||
async execute(_args, exec): Promise<ScheduleListValue> {
|
||||
if (exec.agent !== agent) return internalError()
|
||||
const uncertain = await preflight(rootCtx, agent, 'list')
|
||||
if (uncertain !== undefined) return uncertain
|
||||
notifyDurableChange()
|
||||
const folded = foldForTool(agent)
|
||||
if (isToolError(folded)) return folded
|
||||
const now = Date.now()
|
||||
return folded.active.map(record => scheduleView(record, now))
|
||||
return runScheduleTransaction(agent, async () => {
|
||||
const uncertain = await preflight(rootCtx, agent, 'list')
|
||||
if (uncertain !== undefined) return uncertain
|
||||
notifyDurableChange()
|
||||
const folded = foldForTool(agent)
|
||||
if (isToolError(folded)) return folded
|
||||
const now = Date.now()
|
||||
return folded.active.map(record => scheduleView(record, now))
|
||||
})
|
||||
},
|
||||
presentCall: () => present('List reminders', 'read'),
|
||||
})))
|
||||
@@ -315,23 +320,25 @@ export function registerScheduleTools(
|
||||
}
|
||||
const id = ScheduleId(args.id)
|
||||
if (exec.agent !== agent) return internalError()
|
||||
const uncertain = await preflight(rootCtx, agent, 'delete', id)
|
||||
if (uncertain !== undefined) return uncertain
|
||||
notifyDurableChange()
|
||||
const folded = foldForTool(agent)
|
||||
if (isToolError(folded)) return folded
|
||||
if (!folded.active.some(record => record.id === id)) {
|
||||
return { id, deleted: false, code: 'schedule_not_found' }
|
||||
}
|
||||
try {
|
||||
agent.session.append('schedule/change', { version: 1, operation: 'delete', id })
|
||||
} catch {
|
||||
return internalError()
|
||||
}
|
||||
const barrier = await preflight(rootCtx, agent, 'delete', id)
|
||||
if (barrier !== undefined) return barrier
|
||||
notifyDurableChange()
|
||||
return { id, deleted: true }
|
||||
return runScheduleTransaction(agent, async () => {
|
||||
const uncertain = await preflight(rootCtx, agent, 'delete', id)
|
||||
if (uncertain !== undefined) return uncertain
|
||||
notifyDurableChange()
|
||||
const folded = foldForTool(agent)
|
||||
if (isToolError(folded)) return folded
|
||||
if (!folded.active.some(record => record.id === id)) {
|
||||
return { id, deleted: false, code: 'schedule_not_found' }
|
||||
}
|
||||
try {
|
||||
agent.session.append('schedule/change', { version: 1, operation: 'delete', id })
|
||||
} catch {
|
||||
return internalError()
|
||||
}
|
||||
const barrier = await preflight(rootCtx, agent, 'delete', id)
|
||||
if (barrier !== undefined) return barrier
|
||||
notifyDurableChange()
|
||||
return { id, deleted: true }
|
||||
})
|
||||
},
|
||||
presentCall: args => present('Delete reminder', 'other', args.id),
|
||||
})))
|
||||
|
||||
23
packages/schedule/tool-schedule/src/transaction.ts
Normal file
23
packages/schedule/tool-schedule/src/transaction.ts
Normal file
@@ -0,0 +1,23 @@
|
||||
/** Agent-scoped serialization for Schedule reads and durable mutations. */
|
||||
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
|
||||
const tails = new WeakMap<Agent, Promise<void>>()
|
||||
|
||||
/**
|
||||
* Run one complete Schedule transaction after its exact Agent's prior transaction.
|
||||
* @param agent - Exact Schedule owner and serialization key.
|
||||
* @param operation - Complete preflight, fold, mutation, and postflight operation.
|
||||
* @returns The operation result after exclusive execution.
|
||||
*/
|
||||
export async function runScheduleTransaction<T>(agent: Agent, operation: () => Promise<T>): Promise<T> {
|
||||
const prior = tails.get(agent) ?? Promise.resolve()
|
||||
const run = prior.then(operation)
|
||||
const tail = run.then(() => undefined, () => undefined)
|
||||
tails.set(agent, tail)
|
||||
try {
|
||||
return await run
|
||||
} finally {
|
||||
if (tails.get(agent) === tail) tails.delete(agent)
|
||||
}
|
||||
}
|
||||
@@ -16,7 +16,7 @@ const contexts: Context[] = []
|
||||
interface ToolHarness {
|
||||
readonly ctx: Context
|
||||
readonly agent: Agent
|
||||
readonly flushes: { count: number; outcomes: Array<'resolve' | 'reject'> }
|
||||
readonly flushes: { count: number; outcomes: Array<'resolve' | 'reject' | Promise<'resolve' | 'reject'>> }
|
||||
readonly changes: { count: number }
|
||||
readonly disposeTools: () => void
|
||||
}
|
||||
@@ -50,11 +50,12 @@ async function harness(withPersistence = true): Promise<ToolHarness> {
|
||||
await ctx.plugin(ToolRegistry)
|
||||
const agent = stubAgent(ctx, `schedule-tools-${Math.random()}`)
|
||||
ctx.agents.register(agent)
|
||||
const flushes = { count: 0, outcomes: [] as Array<'resolve' | 'reject'> }
|
||||
const flushes = { count: 0, outcomes: [] as Array<'resolve' | 'reject' | Promise<'resolve' | 'reject'>> }
|
||||
if (withPersistence) {
|
||||
ctx.on('session/flush', async () => {
|
||||
flushes.count += 1
|
||||
if (flushes.outcomes.shift() === 'reject') return Promise.reject(new Error('disk unavailable'))
|
||||
const outcome = await (flushes.outcomes.shift() ?? 'resolve')
|
||||
if (outcome === 'reject') return Promise.reject(new Error('disk unavailable'))
|
||||
return true as const
|
||||
})
|
||||
}
|
||||
@@ -278,6 +279,31 @@ describe('Schedule persistence failure boundaries', () => {
|
||||
expect(test.changes.count).toBe(2)
|
||||
})
|
||||
|
||||
it('serializes concurrent management transactions across both persistence barriers', async () => {
|
||||
const test = await harness()
|
||||
let releaseCreatePreflight: (() => void) | undefined
|
||||
const createPreflight = new Promise<'resolve'>((resolve) => {
|
||||
releaseCreatePreflight = () => { resolve('resolve') }
|
||||
})
|
||||
test.flushes.outcomes.push(createPreflight, 'reject', 'resolve')
|
||||
|
||||
const creating = execute(test, 'schedule_create', { prompt: 'persist me', after_seconds: 10 })
|
||||
await vi.waitFor(() => { expect(test.flushes.count).toBe(1) })
|
||||
const listing = execute(test, 'schedule_list', {})
|
||||
await Promise.resolve()
|
||||
expect(test.flushes.count).toBe(1)
|
||||
|
||||
if (releaseCreatePreflight === undefined) throw new Error('missing create preflight release')
|
||||
releaseCreatePreflight()
|
||||
expect(value(await creating)).toMatchObject({
|
||||
code: 'persistence_uncertain', operation: 'create', id: 'schedule-1',
|
||||
})
|
||||
expect(value(await listing)).toEqual([
|
||||
expect.objectContaining({ id: 'schedule-1', prompt: 'persist me' }),
|
||||
])
|
||||
expect(test.flushes.count).toBe(3)
|
||||
})
|
||||
|
||||
it('returns uncertainty before create or delete reads when their preflight rejects', async () => {
|
||||
const createTest = await harness()
|
||||
createTest.flushes.outcomes.push('reject')
|
||||
|
||||
@@ -186,8 +186,6 @@ interface SessionState {
|
||||
|
||||
/** One live session's initialization and bounded write-behind controller. */
|
||||
interface LiveSessionState {
|
||||
/** Exclusive end of the immutable Session prefix present when this lifecycle was first seen. */
|
||||
seedEnd: number
|
||||
/** Initialization settlement; retained after success and cleared only after rejection. */
|
||||
init: Promise<void> | undefined
|
||||
writes: SessionWriteBehind
|
||||
@@ -1129,18 +1127,17 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
private createLiveState(session: Session): LiveSessionState {
|
||||
let live: LiveSessionState
|
||||
live = {
|
||||
seedEnd: session.events.length,
|
||||
init: undefined,
|
||||
writes: this.createWriteBehind(session, () => this.ensureInitialized(session, live)),
|
||||
}
|
||||
return live
|
||||
}
|
||||
|
||||
/** Start or join one initialization attempt, rebuilding the immutable seed prefix on retry. */
|
||||
/** Start or join one initialization attempt; a retry borrows current Session events and reconciles the durable cursor. */
|
||||
private ensureInitialized(session: Session, live: LiveSessionState): Promise<void> {
|
||||
if (live.init !== undefined) return live.init
|
||||
const init = this.serialize(session.header.id, async () => {
|
||||
const seed = session.events.slice(0, live.seedEnd)
|
||||
const seed = session.events
|
||||
await this.onCreated(session, seed)
|
||||
}).catch((error: unknown) => {
|
||||
live.init = undefined
|
||||
|
||||
@@ -55,7 +55,6 @@ interface MemoryConfig { store?: MemoryStore }
|
||||
interface CoordinatorInternals {
|
||||
states: Map<unknown, unknown>
|
||||
live: Map<unknown, {
|
||||
seedEnd: number
|
||||
init: Promise<void> | undefined
|
||||
writes: { pending: unknown[]; active: Promise<void> | undefined; hasWork: boolean }
|
||||
}>
|
||||
@@ -400,6 +399,8 @@ describe('PersistenceCoordinator retryable live initialization', () => {
|
||||
const session = ctx.sessions.create(SessionId('retry-new-empty'))
|
||||
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) })
|
||||
const first = ctx.sessions.flush(session)
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
loadGate.resolve(undefined)
|
||||
await expect(first).rejects.toThrow('transient init read failure')
|
||||
const retries = [ctx.sessions.flush(session), ctx.sessions.flush(session)]
|
||||
@@ -411,17 +412,14 @@ describe('PersistenceCoordinator retryable live initialization', () => {
|
||||
// more reads.
|
||||
expect(backend.loadAttempts).toBe(3)
|
||||
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.sessions.flush(session)
|
||||
expect(backend.store.get(session.id)?.events.map(event => event.seq)).toEqual([0, 1])
|
||||
|
||||
const live = [...(coordinator as unknown as CoordinatorInternals).live.values()][0]
|
||||
if (live === undefined) throw new Error('live controller was not retained')
|
||||
expect(live.seedEnd).toBe(0)
|
||||
expect(live.init).toBeInstanceOf(Promise)
|
||||
expect(live).not.toHaveProperty('initialized')
|
||||
expect(live).not.toHaveProperty('seed')
|
||||
expect(live).not.toHaveProperty('seedEnd')
|
||||
} finally {
|
||||
loadGate.resolve(undefined)
|
||||
retryGate.resolve(undefined)
|
||||
|
||||
Reference in New Issue
Block a user