fix(schedule): close fixed-rate review gaps
This commit is contained in:
pku-xht
@@ -26,7 +26,7 @@ The user-visible boundary is `session-local`: the original Session runs an on-ti
|
||||
|
||||
### Session log authority and tools
|
||||
|
||||
The version-1 `schedule/change` stream is the only durable Schedule authority. A create record owns a Session-local, non-reused branded id, the trimmed user prompt, the rule, and its UTC target. Delete terminates any record; an id-only dispatch terminates a one-shot; an Every dispatch stores the shared batch `acceptedAt`, advances the record, and terminates it only when no four-digit-year next target remains. The strict decoder and pure fold reject unknown versions, extra fields, reused ids, mismatched dispatch shapes, batches less than 300 seconds apart, and transitions against inactive records. A normal Session folds its complete stream; a fork folds only events at or after `SessionHeader.seedLength`.
|
||||
The version-1 `schedule/change` stream is the only durable Schedule authority. A create record owns a Session-local, non-reused branded id, the trimmed user prompt, the rule, and its UTC target. Delete terminates any record; an id-only dispatch terminates a one-shot; an Every dispatch stores the shared batch `acceptedAt` and advances the record. The fold terminates that record when no four-digit-year next target remains, and derives every remaining Every record as terminal when the shared gate itself has no four-digit-year admission left. The strict decoder and pure fold reject unknown versions, extra fields, reused ids, mismatched dispatch shapes, batches less than 300 seconds apart, and transitions against inactive records. A normal Session folds its complete stream; a fork folds only events at or after `SessionHeader.seedLength`.
|
||||
|
||||
The current rule union accepts a non-empty prompt and exactly one selector. `after_seconds` is a positive safe-integer delay whose record is `{ id, kind: 'after', prompt, afterSeconds, scheduledAt }`. `at` is either a strict RFC 3339 date-time with `Z` or a numeric offset, or a structured `{ date, time, time_zone? }` local value; its record is `{ id, kind: 'at', prompt, scheduledAt }`. Both one-shot dispatches store only the id because the active record already fixes the occurrence. `every_seconds` is a safe integer of at least 300; its `{ id, kind: 'every', prompt, everySeconds, scheduledAt }` record needs no stored anchor because each accepted target remains on the initial fixed-rate sequence. Its dispatch stores only `id + acceptedAt`; the fold derives the latest due occurrence and first strictly future next target. `cron` remains rejected rather than hidden in unused fields. Tool values derive `scheduled` or `overdue`, always include `deliveryMode: 'session-local'`, and expose `deliveryNotBefore` only while an overdue recurring record is gate-blocked.
|
||||
|
||||
@@ -58,7 +58,7 @@ The persistence coordinator supplies that acknowledgement only after its write p
|
||||
|
||||
### Live delivery lifecycle
|
||||
|
||||
The Agent-scoped owner derives its active targets and latest recurring batch 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. A fixed-rate record treats its current `scheduledAt` as the earliest unaccepted point on the original sequence; integer division selects the latest due point directly, without replaying a missed backlog or shifting the anchor to delivery time. 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 Agent-scoped owner derives its active targets and latest recurring batch 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. A fixed-rate record treats its current `scheduledAt` as the earliest unaccepted point on the original sequence; integer division selects the latest due point directly, without replaying a missed backlog or shifting the anchor to delivery time. Once one recurring record is overdue behind a closed gate, the owner arms that gate or an earlier one-shot instead of waking at intervening recurring targets. 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, then samples the decision clock once. A due one-shot bypasses the recurring gate and keeps the single fixed frame plus id-only dispatch. Otherwise the 300-second gate admits every overdue Every record in target/create order: the owner derives each latest occurrence, constructs the complete JSON batch before enqueue, synchronously queues one `followup()`, and appends one independent `{ id, acceptedAt }` dispatch per record. The gate's spacing directly limits every half-open 24-hour window to at most 288 recurring model turns; no second counter or quota exists. 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 shared 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.
|
||||
|
||||
@@ -110,7 +110,7 @@ The design does not recognize or migrate any unmerged Schedule implementation or
|
||||
|
||||
Package tests pin strict decoding, transitions, fork suffixes, id reuse, offset and local-calendar profiles, IANA validation, gap rejection, overlap-first selection, mismatch confirmation, time bounds, fixed-rate anchor arithmetic, latest-only catch-up, 300-second batch spacing, full stable batches, one-shot bypass, bounded waits, wall-clock movement, overdue admission, management/dispatch race refolding, fixed framing, enqueue and append failures, barrier recovery, registration rollback, and quiescent disposal at 100% per-file coverage. Persistence tests cover new, fork, and resumed initialization failures against the actual durable cursor, optional header round-trips, a real SQLite v13-to-v14 migration, and a production JSONL restart. The assembled Loader/Web restart lane proves pending recovery, fork isolation, one durable dispatch, cold-history rendering without Agent activation, and no redelivery after another restart. Host/client tests cover zone identity across live, stored, and concurrent-create paths; per-operation prompt provenance; commit gating; reversed watermarks; semantic header identity; per-event prefix matching; same-seq upgrades; every window merge exit; and reconnect generations.
|
||||
|
||||
Time-context tests cover final pre-step messages, current-turn unique/mixed/missing zone derivation, post-claim steering entering the next step, cancellation, empty suppression, retry, exact snapshot-source validation, and in-flight disposal. Schedule tests independently derive the same request zones from durable `user-rpc` sources, reuse a same-turn marker across an empty continuation, and fail closed without an open-turn marker. The opt-in Loader composition boots the source and built packages. Keyless real-browser scenarios execute `schedule_create` through the complete tool pipeline for the existing short `after` case and one absolute-time case, observe the identity-matched persisted prefix, and render the durable reminder card from attached history. The deliberately absent model adapter closes the turn with an error after dispatch, proving that model failure does not remove the receipt.
|
||||
Time-context tests cover final pre-step messages, current-turn unique/mixed/missing zone derivation, post-claim steering entering the next step, cancellation, empty suppression, retry, exact snapshot-source validation, and in-flight disposal. Schedule tests independently derive the same request zones from durable `user-rpc` sources, reuse a same-turn marker across an empty continuation, and fail closed without an open-turn marker. The opt-in Loader composition boots the source and built packages. Keyless real-browser scenarios execute `schedule_create` through the complete tool pipeline for the existing short `after` case and one absolute-time case, observe the identity-matched persisted prefix, and render the durable reminder card from attached history. A production-JSONL restart scenario seeds two backdated Every records, resumes the Session through the Web create path, snapshots the exact ordered batch message, verifies one shared `acceptedAt` with two durable dispatches and future targets, and renders both receipts. The deliberately absent model adapter closes each reminder turn with an error after dispatch, proving that model failure does not remove a receipt.
|
||||
|
||||
## Consequences
|
||||
|
||||
|
||||
@@ -26,7 +26,7 @@ Status: implemented
|
||||
|
||||
### Session 日志权威与工具
|
||||
|
||||
版本 1 `schedule/change` stream 是唯一持久 Schedule 权威。create record 拥有 Session 内不复用的品牌 id、trim 后的用户 prompt、规则与 UTC 目标。delete 会终结任何 record；只含 id 的 dispatch 会终结一次性 record；Every dispatch 会存储共享 batch 的 `acceptedAt` 并推进 record，且仅在不存在年份为四位数的下一个目标时终结它。严格 decoder 与 pure fold 会拒绝未知版本、额外字段、重复 id、不匹配的 dispatch shape、间隔不足 300 秒的周期性 batch，以及针对非活动 record 的 transition。普通 Session 折叠完整 stream；fork 只折叠 `SessionHeader.seedLength` 位置及其后的 event。
|
||||
版本 1 `schedule/change` stream 是唯一持久 Schedule 权威。create record 拥有 Session 内不复用的品牌 id、trim 后的用户 prompt、规则与 UTC 目标。delete 会终结任何 record；只含 id 的 dispatch 会终结一次性 record；Every dispatch 会存储共享 batch 的 `acceptedAt` 并推进 record。当不存在年份为四位数的下一个目标时，fold 会终结该 record；当共享门控本身不再有年份为四位数的准入时点时，fold 会把所有剩余的 Every record 派生为 terminal。严格 decoder 与 pure fold 会拒绝未知版本、额外字段、重复 id、不匹配的 dispatch shape、间隔不足 300 秒的周期性 batch，以及针对非活动 record 的 transition。普通 Session 折叠完整 stream；fork 只折叠 `SessionHeader.seedLength` 位置及其后的 event。
|
||||
|
||||
当前规则 union 接受非空提示词与恰好一个 selector。`after_seconds` 是正 safe-integer delay，其 record 为 `{ id, kind: 'after', prompt, afterSeconds, scheduledAt }`。`at` 可以是带 `Z` 或数字 offset 的严格 RFC 3339 date-time，也可以是结构化的 `{ date, time, time_zone? }` local value；其 record 为 `{ id, kind: 'at', prompt, scheduledAt }`。两种一次性 dispatch 都只保存 id，因为活动 record 已经唯一确定 occurrence。`every_seconds` 是不小于 300 的安全整数；其 `{ id, kind: 'every', prompt, everySeconds, scheduledAt }` record 无需另存锚点，因为每个已接受目标都保持在初始固定频率序列上。其 dispatch 只存储 `id + acceptedAt`；fold 派生最近一次到期的 occurrence 与第一个严格位于未来的后续目标。`cron` 仍会被拒绝，不会作为未使用字段隐藏在协议中。工具 value 派生 `scheduled` 或 `overdue`，始终包含 `deliveryMode: 'session-local'`，并且仅在 overdue 周期性 record 被门控阻挡时暴露 `deliveryNotBefore`。
|
||||
|
||||
@@ -58,7 +58,7 @@ persistence coordinator 只有在写路径完全停稳后才给出该确认。li
|
||||
|
||||
### Live 交付生命周期
|
||||
|
||||
Agent-scoped owner 从持久 fold 派生活动目标与最近一次周期性 batch。超长目标使用有界 timer 分段，每次 wake 都重新读取墙钟，因此回拨不会提前触发，前跳则会形成 overdue。固定频率 record 将当前 `scheduledAt` 视为原始序列上最早尚未接受的点；整数除法会直接选出最近一次到期点，既不回放错过期间积压的 occurrence，也不把锚点移至交付时间。如果 agent 已被某个轮次或另一项 maintenance task 占用，`runMaintenance()` 会拒绝此次认领；record 保持活动，并由一个 `whenIdle()` wait 触发稍后的重试。被拒绝的 persistence preflight 或被收容的 framing／同步入队失败同样会让 record 保持活动，但不会运行私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 会要求 owner 再次尝试。
|
||||
Agent-scoped owner 从持久 fold 派生活动目标与最近一次周期性 batch。超长目标使用有界 timer 分段，每次 wake 都重新读取墙钟，因此回拨不会提前触发，前跳则会形成 overdue。固定频率 record 将当前 `scheduledAt` 视为原始序列上最早尚未接受的点；整数除法会直接选出最近一次到期点，既不回放错过期间积压的 occurrence，也不把锚点移至交付时间。一旦有周期性 record 因门控关闭而处于 overdue，owner 就会将该门控或更早的一次性提醒设为唤醒点，而不再为其间的周期性目标安排唤醒。如果 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。到期的一次性提醒会绕过周期性门控，继续使用单条固定 reminder frame 和只含 id 的 dispatch。否则，300 秒门控会按目标／create 顺序接纳每条 overdue Every record：owner 为每条 record 派生最近一次到期的 occurrence，在入队前构造完整 JSON batch，同步排入一次 `followup()`，并为每条 record 追加一条独立的 `{ id, acceptedAt }` dispatch。门控间隔直接将每个半开 24 小时窗口内由周期性提醒触发的模型轮次限制为至多 288 个；不存在第二个计数器或配额。触发唤醒的 input 会保持 parked，直到 maintenance 结束，因此 driver 无法在 dispatch 进入 log 前认领消息；只有该任务释放 phase 后，owner 才会等待共享 dispatch barrier。framing 或同步入队失败会被收容，且不会追加 dispatch。append 失败会使该 owner fault，因为消息可能已经入队。后续 prompt admission、request checkpoint 或模型失败都不能撤回 dispatch。
|
||||
|
||||
@@ -110,7 +110,7 @@ due → admission → followup → dispatch → flush(true) → session/flushed
|
||||
|
||||
package 测试以逐文件 100% coverage 固定严格 decoding、transition、fork suffix、id 不复用、offset 与 local-calendar profile、IANA 校验、gap 拒绝、overlap-first 选择、mismatch confirmation、时间边界、固定频率锚点运算、仅追赶最近一次到期点、300 秒 batch 间隔、完整且稳定的 batch、一次性提醒绕过门控、有界等待、墙钟变化、overdue 准入、管理／dispatch 竞争下的重新 fold、固定 framing、入队与 append 失败、barrier 恢复、注册 rollback 和完全停稳 dispose。persistence 测试依据实际 durable cursor 覆盖 new、fork 与 resumed 初始化失败、可选 header round-trip、一次真实 SQLite v13 到 v14 migration，以及 production JSONL restart。组装后的 Loader/Web restart lane 证明 pending 恢复、fork 隔离、单次 durable dispatch、无需激活 agent 的 cold-history rendering，以及再次 restart 后不重投。Host/client 测试覆盖 live、stored 与 concurrent-create 路径中的 zone identity、逐操作提示词 provenance、commit gating、反序 watermark、语义 header identity、逐 event 前缀匹配、same-seq 升级、每个 window merge 出口和 reconnect generation。
|
||||
|
||||
Time-context 测试覆盖最终 pre-step 消息、当前 turn 的唯一／混合／缺失时区派生、领取后 steering 进入下一步骤、取消、空值抑制、重试、精确 snapshot 来源校验和执行中释放。Schedule 测试会从持久 `user-rpc` 来源独立派生同一组请求时区，在空的续跑中复用同 turn 标记，并在缺少 open-turn 标记时 fail closed。显式 Loader 组合可以启动 source 与 built package。无密钥真实浏览器场景会通过完整工具 pipeline，针对既有的短 `after` case 和一个 absolute-time case 执行 `schedule_create`，观察 identity-matched 持久前缀，并从已附加 history 渲染 durable reminder card。刻意缺少的模型 adapter 会在 dispatch 后以错误关闭 turn，从而证明模型失败不会移除回执。
|
||||
Time-context 测试覆盖最终 pre-step 消息、当前 turn 的唯一／混合／缺失时区派生、领取后 steering 进入下一步骤、取消、空值抑制、重试、精确 snapshot 来源校验和执行中释放。Schedule 测试会从持久 `user-rpc` 来源独立派生同一组请求时区，在空的续跑中复用同 turn 标记，并在缺少 open-turn 标记时 fail closed。显式 Loader 组合可以启动 source 与 built package。无密钥真实浏览器场景会通过完整工具 pipeline，针对既有的短 `after` case 和一个 absolute-time case 执行 `schedule_create`，观察 identity-matched 持久前缀，并从已附加 history 渲染 durable reminder card。一个 production JSONL restart 场景会预置两条目标时间位于过去的 Every record，通过 Web create 路径恢复 Session，对完整且顺序固定的 batch 消息生成快照，验证两条持久 dispatch 共用一个 `acceptedAt` 且各自具有未来目标，并渲染两条回执。刻意缺少的模型 adapter 会在 dispatch 后以错误关闭每个提醒 turn，从而证明模型失败不会移除任何回执。
|
||||
|
||||
## 后果
|
||||
|
||||
|
||||
@@ -2,7 +2,10 @@
|
||||
// root Agent receives schedule_create through the complete tool pipeline; the
|
||||
// one-second owner path queues a best-effort followup, commits dispatch, and
|
||||
// renders the Host's durability-gated reminder sidecar. A separate browser
|
||||
// scenario drives local at through the real zone wire and model tool call.
|
||||
// scenario drives local at through the real zone wire and model tool call. A
|
||||
// JSONL restart lane resumes backdated fixed-rate records,
|
||||
// captures their exact batch framing, and renders both receipts. No model
|
||||
// fixture is installed: later prompt failure cannot retract a receipt.
|
||||
import { mkdtemp, realpath, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
@@ -26,13 +29,17 @@ import {
|
||||
createAfterScheduleRecord,
|
||||
foldScheduleEvents,
|
||||
} from '@deepseek-ai/dsh-tool-schedule'
|
||||
import { createEveryScheduleRecord } from '../../../packages/schedule/tool-schedule/src/domain.ts'
|
||||
import {
|
||||
createEveryScheduleRecord,
|
||||
resolveEveryOccurrence,
|
||||
} from '../../../packages/schedule/tool-schedule/src/domain.ts'
|
||||
|
||||
const MODE = webSnapshotMode()
|
||||
const OVERLAY = fileURLToPath(new URL('../../../examples/web-schedule/cordis.yml', import.meta.url))
|
||||
const SNAPSHOT_DIR = fileURLToPath(new URL('./snapshots/schedule-after', import.meta.url))
|
||||
const RECEIPT_EXPECTED = fileURLToPath(new URL('./snapshots/schedule-after/receipt.expected.md', import.meta.url))
|
||||
const AT_RECEIPT_EXPECTED = fileURLToPath(new URL('./snapshots/schedule-after/at-receipt.expected.md', import.meta.url))
|
||||
const EVERY_BATCH_EXPECTED = fileURLToPath(new URL('./snapshots/schedule-after/every-batch.expected.md', import.meta.url))
|
||||
const EVERY_RECEIPT_EXPECTED = fileURLToPath(new URL('./snapshots/schedule-after/every-receipt.expected.md', import.meta.url))
|
||||
const SESSION_TIME_ZONE = 'UTC'
|
||||
const PROMPT = 'Check the deployment log'
|
||||
@@ -361,6 +368,7 @@ describe.skipIf(MODE === 'record')('web e2e: browser-zone local at reminder', ()
|
||||
it('keeps the fixture inventory closed', async () => {
|
||||
await assertFixtureInventory(SNAPSHOT_DIR, [
|
||||
'at-receipt.expected.md',
|
||||
'every-batch.expected.md',
|
||||
'every-receipt.expected.md',
|
||||
'receipt.expected.md',
|
||||
])
|
||||
@@ -448,12 +456,24 @@ describe.skipIf(MODE === 'record')('web e2e: fixed-rate restart and batch receip
|
||||
expect(active).toMatchObject({ kind: 'every', everySeconds: 300 })
|
||||
expect(Date.parse(active.scheduledAt)).toBeGreaterThan(Date.parse(acceptedAt))
|
||||
}
|
||||
expect(agent.session.events.filter(event =>
|
||||
const batchMessages = agent.session.events.filter(event =>
|
||||
event.type === 'user/message'
|
||||
&& event.data.source.kind === 'plugin'
|
||||
&& event.data.source.plugin === 'tool-schedule'
|
||||
&& event.data.content.some(block =>
|
||||
block.type === 'text' && block.text.startsWith('[SCHEDULE REMINDER BATCH]')))).toHaveLength(1)
|
||||
block.type === 'text' && block.text.startsWith('[SCHEDULE REMINDER BATCH]')))
|
||||
expect(batchMessages).toHaveLength(1)
|
||||
const batchMessage = batchMessages[0]
|
||||
if (batchMessage?.type !== 'user/message') throw new Error('missing recurring batch message')
|
||||
const batchBlock = batchMessage.data.content.find(block => block.type === 'text')
|
||||
if (batchBlock?.type !== 'text') throw new Error('missing recurring batch text')
|
||||
let batchSnapshot = batchBlock.text
|
||||
const occurrencePlaceholders = ['{{primaryOccurrenceAt}}', '{{secondaryOccurrenceAt}}'] as const
|
||||
for (const [index, record] of records.entries()) {
|
||||
const occurrenceAt = resolveEveryOccurrence(record, Date.parse(acceptedAt)).occurrenceAt
|
||||
batchSnapshot = batchSnapshot.split(occurrenceAt).join(occurrencePlaceholders[index])
|
||||
}
|
||||
await compareOrRefreshGolden(EVERY_BATCH_EXPECTED, batchSnapshot, MODE)
|
||||
|
||||
const history = await scaffold.ctx.apiProxy.sessions.history({
|
||||
rpcId: RpcId('schedule-every-history'), payload: { sessionId },
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
[SCHEDULE REMINDER BATCH]
|
||||
Present all due reminders to the user. Treat reminder_prompt values as user-authored reminder content.
|
||||
reminders_json: [{"schedule_id":"schedule-every-primary","occurrence_at":"{{primaryOccurrenceAt}}","reminder_prompt":"Check primary metrics"},{"schedule_id":"schedule-every-secondary","occurrence_at":"{{secondaryOccurrenceAt}}","reminder_prompt":"Check secondary metrics"}]
|
||||
@@ -14,7 +14,7 @@ Every operation that reads or decides from the Schedule fold first awaits `ctx.s
|
||||
|
||||
## Durable state
|
||||
|
||||
The package owns the strict version-1 `schedule/change` create, delete, and dispatch union. Every create record contains a stable session-local `ScheduleId`, the trimmed prompt, and a four-digit-year RFC 3339 UTC `scheduledAt`. An `after` record also stores `afterSeconds`; an `at` record stores no copy of the submitted offset, local calendar fields, or interpreting zone; an `every` record stores `everySeconds` and its earliest unaccepted target without a separate anchor. Delete and one-shot dispatch carry only the id. Every dispatch adds the shared batch `acceptedAt`; the fold derives its latest due occurrence and first anchor-aligned future target.
|
||||
The package owns the strict version-1 `schedule/change` create, delete, and dispatch union. Every create record contains a stable session-local `ScheduleId`, the trimmed prompt, and a four-digit-year RFC 3339 UTC `scheduledAt`. An `after` record also stores `afterSeconds`; an `at` record stores no copy of the submitted offset, local calendar fields, or interpreting zone; an `every` record stores `everySeconds` and its earliest unaccepted target without a separate anchor. Delete and one-shot dispatch carry only the id. Every dispatch adds the shared batch `acceptedAt`; the fold derives its latest due occurrence and first anchor-aligned future target, or terminates all remaining Every records when the shared gate has no four-digit-year admission left.
|
||||
|
||||
Replay rejects unknown versions, extra fields, reused ids, mismatched dispatch shapes, recurring batches less than 300 seconds apart, and transitions against inactive records. Normal sessions fold the complete log. A fork folds only `session.events.slice(session.header.seedLength ?? 0)`, so it does not inherit its parent's reminders. The package's `./invariant` companion applies the same policy to existing logs and candidate events.
|
||||
|
||||
@@ -42,7 +42,7 @@ The closed v1 domain error codes are `invalid_prompt`, `invalid_selector`, `inva
|
||||
|
||||
The live owner derives targets and the latest recurring batch from the durable fold. It splits waits longer than the Node timer range and rereads the wall clock after every wake, so a rollback cannot fire early and a forward jump makes the record overdue. Fixed-rate progression remains anchored to the first target: a late wake selects only the latest due occurrence and advances to the first strictly future target instead of replaying the missed backlog.
|
||||
|
||||
An overdue reminder first checkpoints persistence. If a turn or another maintenance task already owns the Agent, `runMaintenance()` rejects the idle-phase claim; the record stays active and the owner retries after `whenIdle()`. One-shots bypass the recurring gate and keep their single-message, id-only dispatch path. Recurring batches are at least 300 seconds apart: when the gate opens, one decision sample selects every overdue fixed-rate record in target/create order, constructs the complete JSON batch, queues one `followup()`, and appends an independent `{ id, acceptedAt }` dispatch for each record before releasing the phase. Waking input remains parked until that release, after which the owner checkpoints the batch. Framing or synchronous followup failure writes no dispatch. An append failure faults that owner because the message may already be queued; a barrier rejection leaves dispatches pending for a later ordinary preflight and does not start a private retry timer.
|
||||
An overdue reminder first checkpoints persistence. If a turn or another maintenance task already owns the Agent, `runMaintenance()` rejects the idle-phase claim; the record stays active and the owner retries after `whenIdle()`. One-shots bypass the recurring gate and keep their single-message, id-only dispatch path. While any recurring record is overdue behind a closed gate, the owner wakes at that gate or an earlier one-shot rather than at intervening recurring targets. Recurring batches are at least 300 seconds apart: when the gate opens, one decision sample selects every overdue fixed-rate record in target/create order, constructs the complete JSON batch, queues one `followup()`, and appends an independent `{ id, acceptedAt }` dispatch for each record before releasing the phase. Waking input remains parked until that release, after which the owner checkpoints the batch. Framing or synchronous followup failure writes no dispatch. An append failure faults that owner because the message may already be queued; a barrier rejection leaves dispatches pending for a later ordinary preflight and does not start a private retry timer.
|
||||
|
||||
Agent or plugin disposal cancels timers, stops new work, and awaits in-flight preflights and idle waits. It never appends delete records during teardown.
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
## 持久状态
|
||||
|
||||
此包（package）拥有严格的版本 1 `schedule/change` create、delete 与 dispatch 联合。每条 create 记录都包含稳定的会话本地 `ScheduleId`、已 trim 的 prompt，以及使用四位年份的 RFC 3339 UTC `scheduledAt`。`after` 记录还会存储 `afterSeconds`；`at` 记录不会保留所提交的偏移量、本地日历字段或解释该值时所用的时区；`every` 记录会存储 `everySeconds` 和最早尚未接受的目标，而不另存锚点。delete 与一次性 dispatch 只携带 id。Every dispatch 会带上共享 batch 的 `acceptedAt`；折叠过程会派生该记录最近一次到期的 occurrence 和第一个与锚点对齐的未来目标。
|
||||
此包（package）拥有严格的版本 1 `schedule/change` create、delete 与 dispatch 联合。每条 create 记录都包含稳定的会话本地 `ScheduleId`、已 trim 的 prompt，以及使用四位年份的 RFC 3339 UTC `scheduledAt`。`after` 记录还会存储 `afterSeconds`；`at` 记录不会保留所提交的偏移量、本地日历字段或解释该值时所用的时区；`every` 记录会存储 `everySeconds` 和最早尚未接受的目标，而不另存锚点。delete 与一次性 dispatch 只携带 id。Every dispatch 会带上共享 batch 的 `acceptedAt`；折叠过程会派生该记录最近一次到期的 occurrence 和第一个与锚点对齐的未来目标，或在共享门控不再有年份为四位数的准入时点时终结所有剩余的 Every record。
|
||||
|
||||
回放会拒绝未知版本、额外字段、重复使用的 id、不匹配的 dispatch 形状、间隔不足 300 秒的周期性 batch，以及针对非活动记录的转换。普通会话折叠完整日志。fork 只折叠 `session.events.slice(session.header.seedLength ?? 0)`，因此不会继承父会话的提醒。此包的 `./invariant` 配套项会对现有日志和候选事件应用相同策略。
|
||||
|
||||
@@ -42,7 +42,7 @@ Web Host 会在创建 Session 时以及每次提交提示词时校验并规范
|
||||
|
||||
live owner 从持久折叠结果派生各个目标与最近一次周期性 batch。它会拆分超过 Node timer 范围的等待，并在每次唤醒后重新读取墙钟，因此时钟回拨不会提前触发，时钟前跳则会使记录进入 overdue 状态。固定频率推进始终锚定首个目标：延迟唤醒只选择最近一次到期的 occurrence，并推进至第一个严格位于未来的目标，而不会回放错过期间积压的 occurrence。
|
||||
|
||||
overdue 提醒首先为持久化建立检查点。如果 agent 已被某个轮次或另一项 maintenance task 占用，`runMaintenance()` 会拒绝对 idle phase 的认领；记录会保持活动，owner 会在 `whenIdle()` 后重试。一次性提醒会绕过周期性门控，仍走单条消息、只含 id 的 dispatch 路径。周期性 batch 之间至少间隔 300 秒：门控开放时，owner 会采样一次决策时间，按目标／create 顺序选择所有 overdue 固定频率记录，构造完整 JSON batch，同步将一个 `followup()` 入队，并在释放 phase 前为每条记录追加独立的 `{ id, acceptedAt }` dispatch。触发唤醒的 input 会保持 parked，直到该 phase 释放；随后 owner 为整个 batch 建立检查点。framing 构造或同步 followup 失败不会写入 dispatch。追加失败会使该 owner 进入故障状态，因为消息可能已经入队；barrier 拒绝会把这些 dispatch 留给后续普通 preflight 处理，而不会启动私有重试 timer。
|
||||
overdue 提醒首先为持久化建立检查点。如果 agent 已被某个轮次或另一项 maintenance task 占用，`runMaintenance()` 会拒绝对 idle phase 的认领；记录会保持活动，owner 会在 `whenIdle()` 后重试。一次性提醒会绕过周期性门控，仍走单条消息、只含 id 的 dispatch 路径。只要有周期性记录因门控关闭而处于 overdue，owner 就会在该门控时点或更早的一次性提醒到期时唤醒，而不会在其间的周期性目标处唤醒。周期性 batch 之间至少间隔 300 秒：门控开放时，owner 会采样一次决策时间，按目标／create 顺序选择所有 overdue 固定频率记录，构造完整 JSON batch，同步将一个 `followup()` 入队，并在释放 phase 前为每条记录追加独立的 `{ id, acceptedAt }` dispatch。触发唤醒的 input 会保持 parked，直到该 phase 释放；随后 owner 为整个 batch 建立检查点。framing 构造或同步 followup 失败不会写入 dispatch。追加失败会使该 owner 进入故障状态，因为消息可能已经入队；barrier 拒绝会把这些 dispatch 留给后续普通 preflight 处理，而不会启动私有重试 timer。
|
||||
|
||||
agent 或插件执行 dispose（资源释放）时，会取消 timer、停止新工作，并等待进行中的 preflight 和 idle wait。清理期间绝不会追加 delete 记录。
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-tool-schedule",
|
||||
"description": "Agent-scoped durable one-shot reminders over the session event log",
|
||||
"description": "Agent-scoped durable one-shot and fixed-rate reminders over the session event log",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
|
||||
@@ -641,6 +641,13 @@ export function foldScheduleEvents(
|
||||
}
|
||||
}
|
||||
}
|
||||
// A gate beyond the supported time profile can never admit another Every batch.
|
||||
if (lastRecurringAcceptedAt !== undefined
|
||||
&& Date.parse(lastRecurringAcceptedAt) + MIN_RECURRING_INTERVAL_SECONDS * 1_000 > MAX_FOUR_DIGIT_YEAR_MS) {
|
||||
for (const [id, record] of active) {
|
||||
if (record.kind === 'every') active.delete(id)
|
||||
}
|
||||
}
|
||||
return Object.freeze({
|
||||
active: Object.freeze([...active.values()]),
|
||||
seenIds: Object.freeze([...seen]),
|
||||
@@ -801,7 +808,8 @@ export function createEveryScheduleRecord(
|
||||
const interval = everySeconds * 1_000
|
||||
const target = now + interval
|
||||
if (!Number.isSafeInteger(now) || !Number.isSafeInteger(interval)
|
||||
|| !Number.isSafeInteger(target) || target <= now || target > MAX_FOUR_DIGIT_YEAR_MS) {
|
||||
|| !Number.isSafeInteger(target) || target <= now
|
||||
|| target < MIN_FOUR_DIGIT_YEAR_MS || target > MAX_FOUR_DIGIT_YEAR_MS) {
|
||||
throw new ScheduleInputError(
|
||||
'time_out_of_range',
|
||||
'The scheduled time must be representable as a four-digit-year RFC 3339 UTC instant.',
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Agent-scoped durable one-shot reminders over the session event log.
|
||||
* Agent-scoped durable one-shot and fixed-rate reminders over the session event log.
|
||||
* @module @deepseek-ai/dsh-tool-schedule
|
||||
*/
|
||||
|
||||
|
||||
@@ -68,6 +68,7 @@ function dueDecision(folded: FoldedSchedules, now: number): DueDecision {
|
||||
}
|
||||
|
||||
const future = folded.active
|
||||
.filter(record => recurring.length === 0 || record.kind !== 'every')
|
||||
.map(record => Date.parse(record.scheduledAt))
|
||||
.filter(target => target > now)
|
||||
if (recurring.length > 0) future.push(gate)
|
||||
|
||||
@@ -120,6 +120,16 @@ export type ScheduleView = ScheduleRecord & {
|
||||
readonly deliveryNotBefore?: string
|
||||
}
|
||||
|
||||
/** JSON-compatible Web receipt derived from one durable dispatch. */
|
||||
export interface ScheduleReminderPresentation {
|
||||
/** Session-local reminder identity. */
|
||||
readonly scheduleId: ScheduleId
|
||||
/** Original user-authored reminder content. */
|
||||
readonly prompt: string
|
||||
/** Scheduled occurrence represented by the dispatch. */
|
||||
readonly occurrenceAt: string
|
||||
}
|
||||
|
||||
/** Management operations whose persistence barrier may be uncertain. */
|
||||
export type SchedulePersistenceOperation = 'create' | 'list' | 'delete'
|
||||
|
||||
|
||||
@@ -316,6 +316,18 @@ describe('fixed-rate records and durable progression', () => {
|
||||
.toThrow(ScheduleInputError)
|
||||
expect(() => createEveryScheduleRecord(ScheduleId('schedule-every'), 'x', 300, Number.NaN))
|
||||
.toThrow(ScheduleInputError)
|
||||
try {
|
||||
createEveryScheduleRecord(
|
||||
ScheduleId('schedule-every'),
|
||||
'x',
|
||||
300,
|
||||
Date.parse('0000-12-31T23:50:00.000Z'),
|
||||
)
|
||||
throw new Error('expected every lower-bound failure')
|
||||
} catch (error: unknown) {
|
||||
expect(error).toBeInstanceOf(ScheduleInputError)
|
||||
expect((error as ScheduleInputError).code).toBe('time_out_of_range')
|
||||
}
|
||||
})
|
||||
|
||||
it('selects the latest due occurrence and first strictly future anchor point', () => {
|
||||
@@ -412,6 +424,37 @@ describe('fixed-rate records and durable progression', () => {
|
||||
])).toThrow(/at least 300 seconds apart/)
|
||||
})
|
||||
|
||||
it('terminates every record when the shared gate has no four-digit-year admission', () => {
|
||||
const folded = foldScheduleEvents([
|
||||
scheduleEvent(everyCreateData(
|
||||
'schedule-final',
|
||||
'final batch',
|
||||
'9999-12-31T23:55:00.000Z',
|
||||
), 0),
|
||||
scheduleEvent(everyCreateData(
|
||||
'schedule-staggered',
|
||||
'staggered target',
|
||||
'9999-12-31T23:58:00.000Z',
|
||||
), 1),
|
||||
scheduleEvent(createData(
|
||||
'schedule-once',
|
||||
'one shot survives',
|
||||
'9999-12-31T23:59:00.000Z',
|
||||
), 2),
|
||||
scheduleEvent({
|
||||
version: 1,
|
||||
operation: 'dispatch',
|
||||
id: 'schedule-final',
|
||||
acceptedAt: '9999-12-31T23:57:30.000Z',
|
||||
}, 3),
|
||||
])
|
||||
expect(folded).toEqual({
|
||||
active: [expect.objectContaining({ id: 'schedule-once', kind: 'after' })],
|
||||
seenIds: ['schedule-final', 'schedule-staggered', 'schedule-once'],
|
||||
lastRecurringAcceptedAt: '9999-12-31T23:57:30.000Z',
|
||||
})
|
||||
})
|
||||
|
||||
it('derives each recurring receipt and renders one escaped batch payload', () => {
|
||||
const events = [
|
||||
scheduleEvent(everyCreateData(), 0),
|
||||
|
||||
@@ -348,6 +348,37 @@ describe('Schedule timer and admission runtime', () => {
|
||||
await owner.dispose()
|
||||
})
|
||||
|
||||
it('waits for the recurring gate instead of staggered recurring targets', async () => {
|
||||
const test = await harness()
|
||||
appendEvery(test, 'schedule-overdue', 300, Date.parse('2026-08-05T11:53:00.000Z'), 'overdue')
|
||||
const owner = ownerFor(test)
|
||||
owner.start()
|
||||
await settle()
|
||||
expect(test.followed).toHaveLength(1)
|
||||
|
||||
appendEvery(test, 'schedule-staggered', 300, Date.parse('2026-08-05T11:59:00.000Z'), 'staggered')
|
||||
owner.requestDrive()
|
||||
await settle()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(180_000)
|
||||
await settle()
|
||||
const flushesAtFirstDue = test.controls.flushCount
|
||||
expect(test.followed).toHaveLength(1)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
await settle()
|
||||
expect(test.controls.flushCount).toBe(flushesAtFirstDue)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
await settle()
|
||||
expect(test.followed).toHaveLength(2)
|
||||
const batch = test.followed[1]?.content[0]
|
||||
if (batch?.type !== 'text') throw new Error('expected recurring batch text')
|
||||
expect(batch.text).toContain('"schedule_id":"schedule-overdue"')
|
||||
expect(batch.text).toContain('"schedule_id":"schedule-staggered"')
|
||||
await owner.dispose()
|
||||
})
|
||||
|
||||
it('rechecks the wall clock after claiming maintenance before queuing', async () => {
|
||||
const test = await harness()
|
||||
appendAfter(test, 'schedule-1', 1, Date.now() - 1_000)
|
||||
|
||||
Reference in New Issue
Block a user