fix(schedule): preserve resumed ancestor receipts

This commit is contained in:
pku-xht
2026-08-06 07:03:49 +08:00
committed by Tianyi Cui
parent 2e187ccf14
commit b669ae46bc
8 changed files with 117 additions and 34 deletions
@@ -49,7 +49,7 @@ Agent or plugin disposal cancels timers, stops new work, unwinds the three tool
### Commit-aware Web receipt ### Commit-aware Web receipt
The Schedule package owns `scheduleReminderPresentation()`, which derives `{ scheduleId, prompt, occurrenceAt, deliveryMode }` from create plus dispatch. A dispatch inside an inherited fork prefix folds from its nearest preceding `session/end-seed` boundary, preserving nested-generation id reuse; a child-owned dispatch folds only the child suffix. Presentation therefore never changes live ownership. The Schedule package owns `scheduleReminderPresentation()`, which derives `{ scheduleId, prompt, occurrenceAt, deliveryMode }` from create plus dispatch. The current fork's `seedLength` is a hard boundary for child-owned dispatches. An inherited dispatch instead pairs with its nearest preceding same-id create because `session/end-seed` also marks replay or resume construction, not only fork ownership. This keeps resumed ancestor receipts renderable, preserves nested-generation id reuse, and never changes live ownership.
The Host continues to send every raw event on append. It keeps one monotonic watermark per exact live `Session` in a `WeakMap`; only `session/flushed` advancement makes it redeliver newly covered dispatch events with the generic `{ for: 'event', view }` sidecar. The durable `schedule/change` type selects the client renderer. Taking the maximum contains reversed concurrent flush completion, and exact object identity prevents a reused Session id from inheriting another lifecycle's cursor. The Host continues to send every raw event on append. It keeps one monotonic watermark per exact live `Session` in a `WeakMap`; only `session/flushed` advancement makes it redeliver newly covered dispatch events with the generic `{ for: 'event', view }` sidecar. The durable `schedule/change` type selects the client renderer. Taking the maximum contains reversed concurrent flush completion, and exact object identity prevents a reused Session id from inheriting another lifecycle's cursor.
@@ -49,7 +49,7 @@ Agent 或插件 dispose 会取消 timer、停止新工作、撤销三个工具
### Commit-aware Web 回执 ### Commit-aware Web 回执
Schedule package 拥有 `scheduleReminderPresentation()`,从 create 加 dispatch 派生 `{ scheduleId, prompt, occurrenceAt, deliveryMode }`。位于继承 fork 前缀中的 dispatch 会从其最近的前置 `session/end-seed` 边界开始折叠,保留嵌套 generation 的 id 复用;child 自有 dispatch 只折叠 child 后缀。因此 presentation 永远不会改变 live ownership。 Schedule package 拥有 `scheduleReminderPresentation()`,从 create 加 dispatch 派生 `{ scheduleId, prompt, occurrenceAt, deliveryMode }`。当前 fork 的 `seedLength` 是 child 自有 dispatch 的硬边界。继承的 dispatch 则会与它之前最近的同 id create 配对,因为 `session/end-seed` 也会标记回放或恢复构造,而不仅标记 fork 所有权。这使恢复后的祖先回执仍可渲染,保留嵌套 generation 的 id 复用,并且绝不会改变 live ownership。
Host 在 append 时继续发送所有 raw event。它在 `WeakMap` 中按 exact live `Session` 保存一个单调 watermark;只有 `session/flushed` 前进时,才会用通用 `{ for: 'event', view }` sidecar 重投新覆盖的 dispatch event。持久 `schedule/change` 类型用于选择 client renderer。取最大值可以收容反序完成的并发 flush,按对象身份键控则阻止复用的 Session id 继承另一个生命周期的 cursor。 Host 在 append 时继续发送所有 raw event。它在 `WeakMap` 中按 exact live `Session` 保存一个单调 watermark;只有 `session/flushed` 前进时,才会用通用 `{ for: 'event', view }` sidecar 重投新覆盖的 dispatch event。持久 `schedule/change` 类型用于选择 client renderer。取最大值可以收容反序完成的并发 flush,按对象身份键控则阻止复用的 Session id 继承另一个生命周期的 cursor。
+2 -2
View File
@@ -470,8 +470,8 @@ function viewFor(ctx: Context, event: SessionEvent, argsFor: (callId: string) =>
/** /**
* Derive one Schedule-owned event sidecar without allowing corrupt domain data * Derive one Schedule-owned event sidecar without allowing corrupt domain data
* to break raw event delivery. `seedLength` selects the parent-prefix or * to break raw event delivery. `seedLength` keeps a child-owned dispatch inside
* child-suffix ownership segment inside the package helper. * its own suffix while the package helper pairs inherited receipts by id.
*/ */
function scheduleViewFor( function scheduleViewFor(
ctx: Context, ctx: Context,
@@ -22,6 +22,20 @@ interface FlushControl {
handler: () => true | Promise<true> handler: () => true | Promise<true>
} }
function reminderCreateData(id: string, prompt: string) {
return {
version: 1 as const,
operation: 'create' as const,
schedule: {
id: ScheduleId(id),
kind: 'after' as const,
prompt,
afterSeconds: 1,
scheduledAt: '2026-08-05T12:00:01.000Z',
},
}
}
async function harness(control?: FlushControl): Promise<Context> { async function harness(control?: FlushControl): Promise<Context> {
const ctx = new Context() const ctx = new Context()
await ctx.plugin(SessionStore) await ctx.plugin(SessionStore)
@@ -39,17 +53,7 @@ function appendReminder(
prompt: string, prompt: string,
): { create: SessionEvent; dispatch: SessionEvent } { ): { create: SessionEvent; dispatch: SessionEvent } {
const scheduleId = ScheduleId(id) const scheduleId = ScheduleId(id)
const create = session.append('schedule/change', { const create = session.append('schedule/change', reminderCreateData(id, prompt))
version: 1,
operation: 'create',
schedule: {
id: scheduleId,
kind: 'after',
prompt,
afterSeconds: 1,
scheduledAt: '2026-08-05T12:00:01.000Z',
},
})
const dispatch = session.append('schedule/change', { const dispatch = session.append('schedule/change', {
version: 1, version: 1,
operation: 'dispatch', operation: 'dispatch',
@@ -148,6 +152,45 @@ describe('commit-aware Schedule live views', () => {
}) })
describe('Schedule history views', () => { describe('Schedule history views', () => {
it('presents a resumed ancestor dispatch copied into a fork seed', async () => {
const ctx = await harness()
const scheduleId = ScheduleId('resumed-reminder')
const resumed = ctx.sessions.create(SessionId('schedule-resumed'), {
seed: [{
type: 'schedule/change',
seq: 0,
time: 1,
data: reminderCreateData('resumed-reminder', 'after restart'),
}],
meta: { cwd: '/tmp' },
})
const dispatch = resumed.append('schedule/change', {
version: 1,
operation: 'dispatch',
id: scheduleId,
})
const child = ctx.sessions.fork(resumed, undefined, SessionId('schedule-fork'))
ctx.provide('sessionPersistence', {
inspect: () => Promise.resolve({ meta: child.header, events: [...child.events] }),
} as never)
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const response = await api.sessions.history({
rpcId: RpcId('schedule-resumed-fork'), payload: { sessionId: child.id },
})
if (!response.result.ok) throw new Error(response.result.error.message)
expect(response.result.value.events.find(entry => entry.event.seq === dispatch.seq)?.view).toEqual({
for: 'event',
view: {
scheduleId,
prompt: 'after restart',
occurrenceAt: '2026-08-05T12:00:01.000Z',
deliveryMode: 'session-local',
},
})
await ctx.fiber.dispose()
})
it('uses only the attached identity-matching stored prefix and fails soft to raw history', async () => { it('uses only the attached identity-matching stored prefix and fails soft to raw history', async () => {
const ctx = await harness() const ctx = await harness()
const parent = ctx.sessions.create(SessionId('schedule-parent'), { meta: { cwd: '/tmp' } }) const parent = ctx.sessions.create(SessionId('schedule-parent'), { meta: { cwd: '/tmp' } })
+1 -1
View File
@@ -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. 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 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. `scheduleReminderPresentation(events, dispatchSeq, seedLength)` is the pure Host-facing receipt projection. It returns `scheduleId`, prompt, occurrence, and `session-local` mode from the dispatch's nearest preceding same-id create. The current fork's `seedLength` is a hard boundary for child-owned dispatches, while inherited dispatches search their persisted prefix; resumed ancestors therefore remain renderable, nested generations may reuse session-local ids, and presentation never changes live ownership.
## Management tools ## Management tools
+1 -1
View File
@@ -16,7 +16,7 @@
回放会拒绝未知版本、额外字段、重复使用的 id,以及针对非活动记录的 delete 或 dispatch 转换。普通会话折叠完整日志。fork 只折叠 `session.events.slice(session.header.seedLength ?? 0)`,因此不会继承父会话的提醒。此包的 `./invariant` 配套项会对现有日志和候选事件应用相同策略。 回放会拒绝未知版本、额外字段、重复使用的 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 会从最近的前置 `session/end-seed` 边界开始折叠,因此嵌套 generation 可以复用会话本地 id,而不会隐藏祖先回执;child 自有 dispatch 只折叠 child 后缀,因此 presentation 绝不会改变 live ownership。 `scheduleReminderPresentation(events, dispatchSeq, seedLength)` 是供 Host 使用的纯回执投影。它从 dispatch 之前最近的同 id create 返回 `scheduleId`、prompt、occurrence 和 `session-local` 模式。当前 fork 的 `seedLength` 是 child 自有 dispatch 的硬边界,而继承的 dispatch 则会搜索其已持久前缀;因此恢复后的祖先仍可渲染,嵌套 generation 可以复用会话本地 id,presentation 绝不会改变 live ownership。
## 管理工具 ## 管理工具
+30 -17
View File
@@ -289,10 +289,9 @@ export function scheduleView(record: AfterScheduleRecord, now: number): Schedule
/** /**
* Derive the Web receipt for one dispatch from its owning stream segment. * Derive the Web receipt for one dispatch from its owning stream segment.
* A dispatch inside an inherited fork prefix folds from its nearest preceding * A child-owned dispatch cannot cross the current fork's `seedLength`.
* `session/end-seed` boundary; a child-owned dispatch folds only the child * An inherited dispatch pairs with its nearest preceding same-id create, so
* suffix. Nested forks can therefore reuse session-local ids without hiding a * resumed ancestors remain renderable and nested forks may reuse local ids.
* persisted ancestor receipt in descendant history.
* @param events - Complete contiguous Session log. * @param events - Complete contiguous Session log.
* @param dispatchSeq - Exact event seq to present. * @param dispatchSeq - Exact event seq to present.
* @param seedLength - Inherited fork prefix length. * @param seedLength - Inherited fork prefix length.
@@ -317,20 +316,34 @@ export function scheduleReminderPresentation(
const dispatch = decodeScheduleChange(event.data) const dispatch = decodeScheduleChange(event.data)
if (dispatch.operation !== 'dispatch') return undefined if (dispatch.operation !== 'dispatch') return undefined
const segmentStart = dispatchSeq < seedLength const segmentStart = dispatchSeq < seedLength ? 0 : seedLength
? events.slice(0, dispatchSeq).findLastIndex(candidate => candidate.type === 'session/end-seed') + 1 for (let index = dispatchSeq - 1; index >= segmentStart; index -= 1) {
: seedLength const candidate = events[index]
const before = foldScheduleEvents(events.slice(segmentStart, dispatchSeq)) if (candidate?.type !== 'schedule/change') continue
const record = before.active.find(candidate => candidate.id === dispatch.id) const change = decodeScheduleChange(candidate.data)
if (record === undefined) { switch (change.operation) {
throw new ScheduleLogError(`schedule dispatch targets inactive id ${JSON.stringify(dispatch.id)}`) case 'create':
if (change.schedule.id !== dispatch.id) break
return Object.freeze({
scheduleId: change.schedule.id,
prompt: change.schedule.prompt,
occurrenceAt: change.schedule.scheduledAt,
deliveryMode: 'session-local',
})
case 'delete':
case 'dispatch':
if (change.id === dispatch.id) {
throw new ScheduleLogError(`schedule dispatch targets inactive id ${JSON.stringify(dispatch.id)}`)
}
break
/* v8 ignore next 3 -- decodeScheduleChange returns a closed operation union. */
default: {
const unreachable: never = change
throw new ScheduleLogError(`unknown decoded schedule change ${String(unreachable)}`)
}
}
} }
return Object.freeze({ throw new ScheduleLogError(`schedule dispatch targets inactive id ${JSON.stringify(dispatch.id)}`)
scheduleId: record.id,
prompt: record.prompt,
occurrenceAt: record.scheduledAt,
deliveryMode: 'session-local',
})
} }
/** /**
@@ -122,6 +122,33 @@ describe('version-1 Schedule decoding and folding', () => {
occurrenceAt: '2026-08-05T12:00:00.000Z', occurrenceAt: '2026-08-05T12:00:00.000Z',
deliveryMode: 'session-local', deliveryMode: 'session-local',
}) })
const resumedThenForked = [
scheduleEvent(createData('resumed-id', 'resumed prompt'), 0),
{ type: 'session/end-seed', seq: 1, time: 1, data: {} } as SessionEvent,
scheduleEvent({ version: 1, operation: 'dispatch', id: 'resumed-id' }, 2),
]
expect(scheduleReminderPresentation(resumedThenForked, 2, 3)).toEqual({
scheduleId: 'resumed-id',
prompt: 'resumed prompt',
occurrenceAt: '2026-08-05T12:00:00.000Z',
deliveryMode: 'session-local',
})
expect(() => scheduleReminderPresentation([
scheduleEvent(createData('parent-only'), 0),
{ type: 'session/end-seed', seq: 1, time: 1, data: {} },
scheduleEvent({ version: 1, operation: 'dispatch', id: 'parent-only' }, 2),
], 2, 2)).toThrow(/inactive id/)
expect(scheduleReminderPresentation([
scheduleEvent(createData('target'), 0),
scheduleEvent(createData('other'), 1),
scheduleEvent({ version: 1, operation: 'delete', id: 'other' }, 2),
scheduleEvent({ version: 1, operation: 'dispatch', id: 'target' }, 3),
], 3)).toMatchObject({ scheduleId: 'target' })
expect(() => scheduleReminderPresentation([
scheduleEvent(createData('ended'), 0),
scheduleEvent({ version: 1, operation: 'delete', id: 'ended' }, 1),
scheduleEvent({ version: 1, operation: 'dispatch', id: 'ended' }, 2),
], 2)).toThrow(/inactive id/)
expect(scheduleReminderPresentation(events, 2, 2)).toBeUndefined() expect(scheduleReminderPresentation(events, 2, 2)).toBeUndefined()
expect(scheduleReminderPresentation([ expect(scheduleReminderPresentation([
{ type: 'session/end-seed', seq: 0, time: 1, data: {} }, { type: 'session/end-seed', seq: 0, time: 1, data: {} },