fix(schedule): close review gaps
This commit is contained in:
@@ -16,7 +16,7 @@ The package owns the strict version-1 `schedule/change` create, delete, and disp
|
||||
|
||||
Replay rejects unknown versions, extra fields, reused ids, and delete or dispatch 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.
|
||||
|
||||
`scheduleReminderPresentation(events, dispatchSeq, seedLength)` is the pure Host-facing receipt projection. It pairs a dispatch with the active create in the same ownership segment and returns `scheduleId`, prompt, occurrence, and `session-local` mode. A dispatch inside a persisted fork prefix folds that parent prefix for history display; a child-owned dispatch folds only the child suffix, so presentation never changes live ownership.
|
||||
`scheduleReminderPresentation(events, dispatchSeq, seedLength)` is the pure Host-facing receipt projection. It pairs a dispatch with the active create in the same ownership segment and returns `scheduleId`, prompt, occurrence, and `session-local` mode. A dispatch inside a persisted fork prefix folds from its nearest preceding `session/end-seed` boundary, so nested generations may reuse session-local ids without hiding ancestor receipts; a child-owned dispatch folds only the child suffix, so presentation never changes live ownership.
|
||||
|
||||
## Management tools
|
||||
|
||||
@@ -79,6 +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.
|
||||
- **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.
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
回放会拒绝未知版本、额外字段、重复使用的 id,以及针对非活动记录的 delete 或 dispatch 转换。普通会话折叠完整日志。fork 只折叠 `session.events.slice(session.header.seedLength ?? 0)`,因此不会继承父会话的提醒。此包的 `./invariant` 配套项会对现有日志和候选事件应用相同策略。
|
||||
|
||||
`scheduleReminderPresentation(events, dispatchSeq, seedLength)` 是供 Host 使用的纯回执投影。它把 dispatch 与同一 ownership segment 中的活动 create 配对,并返回 `scheduleId`、prompt、occurrence 和 `session-local` 模式。位于已持久 fork 前缀中的 dispatch 会折叠对应 parent 前缀用于 history 显示;child 自有 dispatch 只折叠 child 后缀,因此 presentation 绝不会改变 live ownership。
|
||||
`scheduleReminderPresentation(events, dispatchSeq, seedLength)` 是供 Host 使用的纯回执投影。它把 dispatch 与同一 ownership segment 中的活动 create 配对,并返回 `scheduleId`、prompt、occurrence 和 `session-local` 模式。位于已持久 fork 前缀中的 dispatch 会从最近的前置 `session/end-seed` 边界开始折叠,因此嵌套 generation 可以复用会话本地 id,而不会隐藏祖先回执;child 自有 dispatch 只折叠 child 后缀,因此 presentation 绝不会改变 live ownership。
|
||||
|
||||
## 管理工具
|
||||
|
||||
@@ -79,6 +79,7 @@ reminder_prompt_json: <JSON.stringify(prompt)>
|
||||
## 已知限制与暂缓事项
|
||||
|
||||
- **仅限会话本地交付**:提醒只有在原会话 live 时才能准时运行;cold 会话不会收到外部通知,只有恢复后才会处理 overdue 记录。
|
||||
- **活动驱动的持久化重试**:到期 preflight 被拒绝后,overdue 记录仍保持活动,但不会启动私有重试 timer;后续 agent 活动进入 idle,或成功的 Schedule 管理 preflight 要求 owner 重新计算后,owner 会重试。
|
||||
- **仅支持 after 协议**:版本 1 拒绝 `at`、`every_seconds`、`cron` 和 `time_zone`;这些规则需要后续协议变体,而不是隐藏的兼容字段。
|
||||
- **存在狭窄的崩溃重复窗口**:同步 `followup` 获得准入后、dispatch 检查点完成前发生崩溃,可能使提醒在恢复后重复;此包不承诺模型完成、用户确认或外部副作用恰好一次。
|
||||
- **加载顺序边界**:插件不会扫描或接管加载时已经 live 的 agent。
|
||||
|
||||
@@ -289,10 +289,10 @@ export function scheduleView(record: AfterScheduleRecord, now: number): Schedule
|
||||
|
||||
/**
|
||||
* Derive the Web receipt for one dispatch from its owning stream segment.
|
||||
* A dispatch inside an inherited fork prefix folds that original prefix; a
|
||||
* child-owned dispatch folds only the child suffix, preserving the same
|
||||
* `seedLength` ownership rule as the live runtime while still allowing a
|
||||
* persisted parent receipt to render in child history.
|
||||
* A dispatch inside an inherited fork prefix folds from its nearest preceding
|
||||
* `session/end-seed` boundary; a child-owned dispatch folds only the child
|
||||
* suffix. Nested forks can therefore reuse session-local ids without hiding a
|
||||
* persisted ancestor receipt in descendant history.
|
||||
* @param events - Complete contiguous Session log.
|
||||
* @param dispatchSeq - Exact event seq to present.
|
||||
* @param seedLength - Inherited fork prefix length.
|
||||
@@ -317,7 +317,9 @@ export function scheduleReminderPresentation(
|
||||
const dispatch = decodeScheduleChange(event.data)
|
||||
if (dispatch.operation !== 'dispatch') return undefined
|
||||
|
||||
const segmentStart = dispatchSeq < seedLength ? 0 : seedLength
|
||||
const segmentStart = dispatchSeq < seedLength
|
||||
? events.slice(0, dispatchSeq).findLastIndex(candidate => candidate.type === 'session/end-seed') + 1
|
||||
: seedLength
|
||||
const before = foldScheduleEvents(events.slice(segmentStart, dispatchSeq))
|
||||
const record = before.active.find(candidate => candidate.id === dispatch.id)
|
||||
if (record === undefined) {
|
||||
|
||||
@@ -156,6 +156,22 @@ export class ScheduleOwner {
|
||||
)
|
||||
}
|
||||
|
||||
/** Fold the current exact owner suffix and contain a corrupt durable stream. */
|
||||
private readEarliest(): AfterScheduleRecord | undefined {
|
||||
try {
|
||||
const folded = foldScheduleEvents(
|
||||
this.agent.session.events,
|
||||
this.agent.session.header.seedLength ?? 0,
|
||||
)
|
||||
return earliest(folded.active)
|
||||
} catch (error: unknown) {
|
||||
this.faulted = true
|
||||
const detail = error instanceof ScheduleLogError ? error.message : renderThrown(error)
|
||||
this.ctx.logger.warn(`tool-schedule: corrupt schedule log for agent "${this.agent.id}": ${detail}`)
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/** Preflight, fold, arm, or dispatch the next active one-shot reminder. */
|
||||
private async driveOnce(): Promise<void> {
|
||||
this.clearTimer()
|
||||
@@ -171,19 +187,7 @@ export class ScheduleOwner {
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- disposal or replacement can win while persistence is awaited.
|
||||
if (this.stopping || !this.isLive()) return
|
||||
|
||||
let record: AfterScheduleRecord | undefined
|
||||
try {
|
||||
const folded = foldScheduleEvents(
|
||||
this.agent.session.events,
|
||||
this.agent.session.header.seedLength ?? 0,
|
||||
)
|
||||
record = earliest(folded.active)
|
||||
} catch (error: unknown) {
|
||||
this.faulted = true
|
||||
const detail = error instanceof ScheduleLogError ? error.message : renderThrown(error)
|
||||
this.ctx.logger.warn(`tool-schedule: corrupt schedule log for agent "${this.agent.id}": ${detail}`)
|
||||
return
|
||||
}
|
||||
const record = this.readEarliest()
|
||||
if (record === undefined) return
|
||||
|
||||
const target = Date.parse(record.scheduledAt)
|
||||
@@ -193,47 +197,50 @@ export class ScheduleOwner {
|
||||
return
|
||||
}
|
||||
|
||||
const release = this.agent.reserveTurnAdmission()
|
||||
if (release === undefined) {
|
||||
this.waitForIdle()
|
||||
let maintenance: Promise<boolean>
|
||||
try {
|
||||
maintenance = this.agent.runMaintenance(() => {
|
||||
if (this.stopping || !this.isLive()) return Promise.resolve(false)
|
||||
const claimedRecord = this.readEarliest()
|
||||
if (claimedRecord === undefined) return Promise.resolve(false)
|
||||
const claimedTarget = Date.parse(claimedRecord.scheduledAt)
|
||||
const decisionNow = Date.now()
|
||||
if (decisionNow < claimedTarget) {
|
||||
this.arm(claimedTarget, decisionNow)
|
||||
return Promise.resolve(false)
|
||||
}
|
||||
try {
|
||||
const message = createUserMessage({
|
||||
content: [{ type: 'text', text: renderReminderFraming(claimedRecord) }],
|
||||
source: { kind: 'plugin', plugin: 'tool-schedule' },
|
||||
})
|
||||
this.agent.followup(message)
|
||||
} catch (error: unknown) {
|
||||
if (this.isLive()) {
|
||||
this.ctx.logger.warn(`tool-schedule: framing or followup failed for agent "${this.agent.id}": ${renderThrown(error)}`)
|
||||
}
|
||||
return Promise.resolve(false)
|
||||
}
|
||||
try {
|
||||
this.agent.session.append('schedule/change', {
|
||||
version: 1,
|
||||
operation: 'dispatch',
|
||||
id: claimedRecord.id,
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
this.faulted = true
|
||||
this.clearTimer()
|
||||
this.ctx.logger.warn(`tool-schedule: dispatch append failed for agent "${this.agent.id}": ${renderThrown(error)}`)
|
||||
return Promise.resolve(false)
|
||||
}
|
||||
return Promise.resolve(true)
|
||||
})
|
||||
} catch (_busy: unknown) {
|
||||
// `runMaintenance` rejects synchronously only while another agent activity owns the idle phase.
|
||||
if (this.isLive()) this.waitForIdle()
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- reservation can invalidate the owner.
|
||||
if (this.stopping || !this.isLive()) return
|
||||
const decisionNow = Date.now()
|
||||
if (decisionNow < target) {
|
||||
this.arm(target, decisionNow)
|
||||
return
|
||||
}
|
||||
const message = createUserMessage({
|
||||
content: [{ type: 'text', text: renderReminderFraming(record) }],
|
||||
source: { kind: 'plugin', plugin: 'tool-schedule' },
|
||||
})
|
||||
try {
|
||||
this.agent.followup(message)
|
||||
} catch (error: unknown) {
|
||||
if (this.isLive()) {
|
||||
this.ctx.logger.warn(`tool-schedule: followup failed for agent "${this.agent.id}": ${renderThrown(error)}`)
|
||||
}
|
||||
return
|
||||
}
|
||||
try {
|
||||
this.agent.session.append('schedule/change', {
|
||||
version: 1,
|
||||
operation: 'dispatch',
|
||||
id: record.id,
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
this.faulted = true
|
||||
this.clearTimer()
|
||||
this.ctx.logger.warn(`tool-schedule: dispatch append failed for agent "${this.agent.id}": ${renderThrown(error)}`)
|
||||
return
|
||||
}
|
||||
} finally {
|
||||
release()
|
||||
}
|
||||
if (!await maintenance) return
|
||||
|
||||
try {
|
||||
await flushSchedulePersistence(this.ctx, this.agent.session)
|
||||
|
||||
@@ -109,6 +109,19 @@ describe('version-1 Schedule decoding and folding', () => {
|
||||
occurrenceAt: '2026-08-05T12:00:00.000Z',
|
||||
deliveryMode: 'session-local',
|
||||
})
|
||||
const nested = [
|
||||
scheduleEvent(createData('same-id', 'grandparent prompt'), 0),
|
||||
scheduleEvent({ version: 1, operation: 'dispatch', id: 'same-id' }, 1),
|
||||
{ type: 'session/end-seed', seq: 2, time: 1, data: {} } as SessionEvent,
|
||||
scheduleEvent(createData('same-id', 'parent prompt'), 3),
|
||||
scheduleEvent({ version: 1, operation: 'dispatch', id: 'same-id' }, 4),
|
||||
]
|
||||
expect(scheduleReminderPresentation(nested, 4, 5)).toEqual({
|
||||
scheduleId: 'same-id',
|
||||
prompt: 'parent prompt',
|
||||
occurrenceAt: '2026-08-05T12:00:00.000Z',
|
||||
deliveryMode: 'session-local',
|
||||
})
|
||||
expect(scheduleReminderPresentation(events, 2, 2)).toBeUndefined()
|
||||
expect(scheduleReminderPresentation([
|
||||
{ type: 'session/end-seed', seq: 0, time: 1, data: {} },
|
||||
|
||||
@@ -277,6 +277,30 @@ describe('Schedule timer and admission runtime', () => {
|
||||
expect(test.followed).toHaveLength(1)
|
||||
await owner.dispose()
|
||||
})
|
||||
|
||||
it('rechecks the durable fold after claiming maintenance', async () => {
|
||||
const test = await harness()
|
||||
appendAfter(test, 'schedule-1', 1, Date.now() - 1_000)
|
||||
test.controls.onReserve = () => {
|
||||
test.controls.onReserve = undefined
|
||||
test.agent.session.append('schedule/change', {
|
||||
version: 1,
|
||||
operation: 'delete',
|
||||
id: ScheduleId('schedule-1'),
|
||||
})
|
||||
}
|
||||
const owner = ownerFor(test)
|
||||
owner.start()
|
||||
await settle()
|
||||
|
||||
expect(test.controls.releaseCount).toBe(1)
|
||||
expect(test.followed).toEqual([])
|
||||
expect(test.agent.session.events.at(-1)?.data).toMatchObject({ operation: 'delete' })
|
||||
owner.requestDrive()
|
||||
await settle()
|
||||
expect(test.followed).toEqual([])
|
||||
await owner.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
describe('Schedule runtime failure and teardown boundaries', () => {
|
||||
@@ -459,7 +483,7 @@ describe('Schedule runtime failure and teardown boundaries', () => {
|
||||
expect(unreadable.followed).toEqual([])
|
||||
})
|
||||
|
||||
it('contains owner startup and run failures', async () => {
|
||||
it('contains owner startup, maintenance, and framing failures', async () => {
|
||||
const startup = await harness()
|
||||
const startSpy = vi.spyOn(startup.ctx.agents, 'withoutInitiator')
|
||||
.mockImplementation(() => { throw new Error('initiator closing') })
|
||||
@@ -477,6 +501,29 @@ describe('Schedule runtime failure and teardown boundaries', () => {
|
||||
expect(departedStartup.controls.flushCount).toBe(0)
|
||||
departedStartSpy.mockRestore()
|
||||
|
||||
const maintenanceFailure = await harness()
|
||||
appendAfter(maintenanceFailure, 'schedule-1', 1, Date.now() - 1_000)
|
||||
const maintenanceSpy = vi.spyOn(maintenanceFailure.agent, 'runMaintenance')
|
||||
.mockImplementation(() => Promise.reject(new Error('maintenance failed')))
|
||||
const maintenanceOwner = ownerFor(maintenanceFailure)
|
||||
maintenanceOwner.start()
|
||||
await settle()
|
||||
expect(maintenanceFailure.followed).toEqual([])
|
||||
maintenanceOwner.requestDrive()
|
||||
await settle()
|
||||
expect(maintenanceSpy).toHaveBeenCalledOnce()
|
||||
|
||||
const departedMaintenance = await harness()
|
||||
appendAfter(departedMaintenance, 'schedule-1', 1, Date.now() - 1_000)
|
||||
vi.spyOn(departedMaintenance.agent, 'runMaintenance').mockImplementation(() => {
|
||||
departedMaintenance.disposeAgent()
|
||||
return Promise.reject(new Error('maintenance failed after detach'))
|
||||
})
|
||||
const departedMaintenanceOwner = ownerFor(departedMaintenance)
|
||||
departedMaintenanceOwner.start()
|
||||
await settle()
|
||||
expect(departedMaintenance.followed).toEqual([])
|
||||
|
||||
const runFailure = await harness()
|
||||
appendAfter(runFailure, 'schedule-1', 1, Date.now() - 1_000)
|
||||
const uuidSpy = vi.spyOn(globalThis.crypto, 'randomUUID').mockImplementation(() => { throw 'message failed' })
|
||||
@@ -486,7 +533,7 @@ describe('Schedule runtime failure and teardown boundaries', () => {
|
||||
uuidSpy.mockRestore()
|
||||
failingOwner.requestDrive()
|
||||
await settle()
|
||||
expect(runFailure.followed).toEqual([])
|
||||
expect(runFailure.followed).toHaveLength(1)
|
||||
|
||||
const departedRun = await harness()
|
||||
appendAfter(departedRun, 'schedule-1', 1, Date.now() - 1_000)
|
||||
|
||||
Reference in New Issue
Block a user