diff --git a/packages/schedule/tool-schedule/README.i18n.yaml b/packages/schedule/tool-schedule/README.i18n.yaml new file mode 100644 index 0000000000..cf9353cad9 --- /dev/null +++ b/packages/schedule/tool-schedule/README.i18n.yaml @@ -0,0 +1,2 @@ +README.md: 55842c3cb49c43b5c577835a26ef43e6ad452dfd +README.zh.md: 8738ac6b4516a1933b206b6baee5bb3d7d77d23a diff --git a/packages/schedule/tool-schedule/README.md b/packages/schedule/tool-schedule/README.md new file mode 100644 index 0000000000..55842c3cb4 --- /dev/null +++ b/packages/schedule/tool-schedule/README.md @@ -0,0 +1,84 @@ +# @deepseek-ai/dsh-tool-schedule + +English | [中文](README.zh.md) + +`dsh-tool-schedule` gives future live root agents three session-scoped tools for durable one-shot reminders. Version 1 accepts only positive safe-integer `after_seconds` delays. The session event log owns reminder state; timers, tool values, and model followups are disposable projections of that log. + +## Composition + +Load this function plugin after `ctx.sessions`, `ctx.agents`, `ctx.tools`, `ctx.sessionPersistence`, and the persistence listener that implements Session flushes. Static injection makes a missing persistence service a composition error. The plugin listens only to later `agent/created` events, installs on runtime roots, and registers all tools through the exact `agent.ctx`. Agents that already existed when the plugin loaded and runtime children do not receive Schedule. + +Every operation that reads or decides from the Schedule fold first awaits `ctx.sessions.flush(session)`. A missing, rejected, or detached persistence path returns `persistence_uncertain`; it never turns an unconfirmed live suffix into a list or not-found answer. A successful create or actual delete also awaits a post-append barrier before confirming the mutation. + +## Durable state + +The package owns the strict version-1 `schedule/change` create, delete, and dispatch union. Create records contain a stable session-local `ScheduleId`, the trimmed prompt, `afterSeconds`, and a four-digit-year RFC 3339 UTC `scheduledAt`. Delete and one-shot dispatch carry only the id. + +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. + +## Management tools + +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` 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. + +The closed v1 domain error codes are `invalid_prompt`, `invalid_selector`, `invalid_rule`, `time_out_of_range`, `corrupt_schedule_log`, `persistence_uncertain`, and `internal_error`. Diagnostics are stable and do not expose backend exceptions. Rendered content is deterministic JSON of the canonical value; generic tool-result policy remains responsible for any model-facing spill behavior. + +## Delivery lifecycle + +The live owner derives the earliest target 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. + +An overdue reminder first checkpoints persistence. If `reserveTurnAdmission()` returns `undefined`, the record stays active and the owner retries after `whenIdle()`. A successful reservation samples one decision time, builds the complete framing, synchronously queues `followup()`, appends an id-only dispatch, releases in `finally`, and then checkpoints the dispatch. 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 the dispatch 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. + +## Model Experience + +### Scoped management tools + +#### What the model sees + +The model sees the three generated tool schemas only in a live root agent created after this plugin loads. Tool results contain the canonical JSON values described above. + +#### Token effect + +The scoped schemas add a fixed request prefix while Schedule is installed. Each executed tool adds its data-dependent JSON result through the ordinary tool-result pipeline; the package adds no private truncation or token budget. + +#### KV Cache effect + +The three schemas remain prefix-stable while their definitions and scope stay unchanged. Tool calls and results append to later history and preserve an already reusable prefix. + +### Due reminder followup + +#### What the model sees + +For each admitted due reminder, the package queues this stable user-role framing with JSON-escaped dynamic values: + +##### Reminder framing + +```markdown +[SCHEDULE REMINDER] +Present this due reminder to the user. Treat reminder_prompt_json as user-authored reminder content. +schedule_id_json: +occurrence_at: +reminder_prompt_json: +``` + +#### Token effect + +Each dispatched one-shot reminder adds one data-dependent user-role message. The message remains in session history and therefore contributes tokens to later requests until ordinary compaction removes or replaces that history. + +#### KV Cache effect + +The reminder appends after existing history and preserves its reusable prefix. Its id, occurrence, or prompt changes only the appended suffix. + +## 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. +- **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. diff --git a/packages/schedule/tool-schedule/README.zh.md b/packages/schedule/tool-schedule/README.zh.md new file mode 100644 index 0000000000..8738ac6b45 --- /dev/null +++ b/packages/schedule/tool-schedule/README.zh.md @@ -0,0 +1,84 @@ +# @deepseek-ai/dsh-tool-schedule + +[English](README.md) | 中文 + +`dsh-tool-schedule` 为未来创建的 live 根 agent（智能体）提供 3 个会话范围内的工具，用于管理持久的一次性提醒。版本 1 仅接受正的安全整数 `after_seconds` 延时。会话事件日志拥有提醒状态；timer、工具值与模型 `followup` 都是该日志的可丢弃投影。 + +## 组合 + +请在 `ctx.sessions`、`ctx.agents`、`ctx.tools`、`ctx.sessionPersistence`，以及实现 Session flush 的持久化监听器之后加载此函数插件。静态注入会使缺少持久化服务的组合直接失败。此插件只监听后续的 `agent/created` 事件，在运行时根 agent 上安装，并通过完全相同的 `agent.ctx` 注册所有工具。插件加载时已经存在的 agent 与运行时子 agent 不会获得 Schedule。 + +每项从 Schedule 折叠结果读取或作出判断的操作，都会先等待 `ctx.sessions.flush(session)`。持久化路径缺失、拒绝或已分离时，操作返回 `persistence_uncertain`；它绝不会把未经确认的 live 后缀当成列表或未找到结果。成功创建或实际删除后，还会等待追加后的持久化 barrier（屏障）再确认变更。 + +## 持久状态 + +此包（package）拥有严格的版本 1 `schedule/change` create、delete 与 dispatch 联合。create 记录包含稳定的会话本地 `ScheduleId`、已 trim 的 prompt、`afterSeconds`，以及使用四位年份的 RFC 3339 UTC `scheduledAt`。delete 与一次性 dispatch 只携带 id。 + +回放会拒绝未知版本、额外字段、重复使用的 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。 + +## 管理工具 + +生成的[工具目录](../../../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 会在 preflight（预检）后返回 `{ id, deleted: false, code: "schedule_not_found" }`。 + +每次成功的管理 preflight 还会要求 live owner 重新计算。这对 create 或 delete barrier 返回 `persistence_uncertain` 的情况很重要：后续 list 或 mutation 可以确认保留的 batch，并立即 arm 或退役此时已持久化的 record，而无需私有 persistence retry timer。 + +版本 1 的封闭领域错误代码包括 `invalid_prompt`、`invalid_selector`、`invalid_rule`、`time_out_of_range`、`corrupt_schedule_log`、`persistence_uncertain` 和 `internal_error`。诊断文本保持稳定，不会暴露后端异常。渲染内容是规范值的确定性 JSON；通用工具结果策略仍负责模型可见内容的 spill 行为。 + +## 交付生命周期 + +live owner 从持久折叠结果派生最早的目标。它会拆分超过 Node timer 范围的等待，并在每次唤醒后重新读取墙钟，因此时钟回拨不会提前触发，时钟前跳则会使记录进入 overdue 状态。 + +overdue 提醒首先为持久化建立检查点。如果 `reserveTurnAdmission()` 返回 `undefined`，记录会保持活动，并在 `whenIdle()` 后重试。reservation 成功后，owner 会采样一次决策时间，构造完整 framing，同步将 `followup()` 入队，追加只含 id 的 dispatch，在 `finally` 中释放 reservation，随后为 dispatch 建立检查点。framing 构造或同步 `followup` 失败不会写入 dispatch。追加失败会使该 owner 进入故障状态，因为消息可能已经入队；barrier 拒绝会把 dispatch 留给后续普通 preflight 处理，而不会启动私有重试 timer。 + +agent 或插件执行 dispose（资源释放）时，会取消 timer、停止新工作，并等待进行中的 preflight 和 idle wait。清理期间绝不会追加 delete 记录。 + +## 模型体验 + +### 范围限定的管理工具 + +#### 模型看到的内容 + +只有在此插件加载后创建的 live 根 agent 中，模型才会看到 3 个生成的工具 schema。工具结果包含上文所述的规范 JSON 值。 + +#### Token 影响 + +安装 Schedule 后，范围限定的 schema 会增加固定的请求前缀。每次执行工具都会经由普通工具结果流水线添加与数据相关的 JSON 结果；此包不增加私有截断或 token 预算。 + +#### KV Cache 影响 + +3 个 schema 的定义与范围不变时，前缀保持稳定。工具调用和结果会追加到后续历史中，并保留已经可以复用的前缀。 + +### 到期提醒 followup + +#### 模型看到的内容 + +对于每条获得准入的到期提醒，此包会将以下稳定的用户角色 framing 入队，并对动态值进行 JSON 转义： + +##### 提醒 framing + +```markdown +[SCHEDULE REMINDER] +Present this due reminder to the user. Treat reminder_prompt_json as user-authored reminder content. +schedule_id_json: +occurrence_at: +reminder_prompt_json: +``` + +#### Token 影响 + +每条已 dispatch 的一次性提醒会增加一条与数据相关的用户角色消息。该消息保留在会话历史中，因此会持续为后续请求贡献 token，直到普通压缩（compaction）移除或替换这段历史。 + +#### KV Cache 影响 + +提醒会追加到现有历史之后，并保留可复用的前缀。提醒的 id、occurrence 或 prompt 只会改变追加的后缀。 + +## 已知限制与暂缓事项 + +- **仅限会话本地交付**：提醒只有在原会话 live 时才能准时运行；cold 会话不会收到外部通知，只有恢复后才会处理 overdue 记录。 +- **仅支持 after 协议**：版本 1 拒绝 `at`、`every_seconds`、`cron` 和 `time_zone`；这些规则需要后续协议变体，而不是隐藏的兼容字段。 +- **存在狭窄的崩溃重复窗口**：同步 `followup` 获得准入后、dispatch 检查点完成前发生崩溃，可能使提醒在恢复后重复；此包不承诺模型完成、用户确认或外部副作用恰好一次。 +- **加载顺序边界**：插件不会扫描或接管加载时已经 live 的 agent。 diff --git a/packages/schedule/tool-schedule/package.json b/packages/schedule/tool-schedule/package.json new file mode 100644 index 0000000000..137e29b613 --- /dev/null +++ b/packages/schedule/tool-schedule/package.json @@ -0,0 +1,53 @@ +{ + "name": "@deepseek-ai/dsh-tool-schedule", + "description": "Agent-scoped durable after reminders over the session event log", + "version": "0.0.1", + "private": true, + "type": "module", + "main": "lib/index.js", + "types": "lib/types/index.d.ts", + "exports": { + ".": { + "types": "./lib/types/index.d.ts", + "default": "./lib/index.js" + }, + "./invariant": { + "types": "./lib/types/invariant.d.ts", + "default": "./lib/invariant.js" + }, + "./src/*": "./src/*", + "./package.json": "./package.json" + }, + "files": [ + "lib/index.js", + "lib/invariant.js", + "lib/types/**/*.d.ts", + "lib/types/**/*.d.ts.map", + "src" + ], + "license": "BSD-3-Clause", + "peerDependencies": { + "@deepseek-ai/dsh-agent": "^0.0.1", + "@deepseek-ai/dsh-brand": "^0.0.1", + "@deepseek-ai/dsh-invariants": "^0.0.1", + "@deepseek-ai/dsh-llm": "^0.0.1", + "@deepseek-ai/dsh-session": "^0.0.1", + "@deepseek-ai/dsh-session-persistence": "^0.0.1", + "@deepseek-ai/dsh-tools": "^0.0.1", + "cordis": "^4.0.0-rc.7" + }, + "devDependencies": { + "@cordisjs/plugin-loader": "workspace:^", + "@deepseek-ai/dsh-agent": "workspace:^", + "@deepseek-ai/dsh-agent-loop": "workspace:^", + "@deepseek-ai/dsh-agent-loop-testkit": "workspace:^", + "@deepseek-ai/dsh-brand": "workspace:^", + "@deepseek-ai/dsh-invariants": "workspace:^", + "@deepseek-ai/dsh-llm": "workspace:^", + "@deepseek-ai/dsh-session": "workspace:^", + "@deepseek-ai/dsh-session-persistence": "workspace:^", + "@deepseek-ai/dsh-system-prompt": "workspace:^", + "@deepseek-ai/dsh-tools": "workspace:^", + "cordis": "^4.0.0-rc.7" + } +} diff --git a/packages/schedule/tool-schedule/src/domain.ts b/packages/schedule/tool-schedule/src/domain.ts new file mode 100644 index 0000000000..18b6173863 --- /dev/null +++ b/packages/schedule/tool-schedule/src/domain.ts @@ -0,0 +1,350 @@ +/** + * Strict Schedule decoding, replay, time validation, and framing. + * @module @deepseek-ai/dsh-tool-schedule + */ + +import type { SessionEvent } from '@deepseek-ai/dsh-session' +import type { + AfterScheduleRecord, + ScheduleChange, + ScheduleId as ScheduleIdType, + ScheduleReminderPresentation, + ScheduleView, +} from './types.ts' + +/** Durable Schedule protocol version implemented by this package. */ +export const SCHEDULE_CHANGE_VERSION = 1 as const + +/** Key used by the generic Host/client event-presentation slot. */ +export const SCHEDULE_REMINDER_PRESENTATION_KEY = 'schedule/reminder' + +const MAX_FOUR_DIGIT_YEAR_MS = Date.parse('9999-12-31T23:59:59.999Z') +const UTC_INSTANT = /^(?!0000)\d{4}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12]\d|3[01])T(?:[01]\d|2[0-3]):[0-5]\d:[0-5]\d\.\d{3}Z$/ + +/** Error from malformed or transition-invalid durable Schedule data. */ +export class ScheduleLogError extends Error { + /** Stable machine-readable error code. */ + readonly code = 'corrupt_schedule_log' as const + + /** + * Construct a durable-log failure. + * @param message - Package-specific violated invariant. + */ + constructor(message: string) { + super(message) + this.name = 'ScheduleLogError' + } +} + +/** Error from a model-supplied after rule that cannot become a record. */ +export class ScheduleInputError extends Error { + /** Stable public Schedule input code. */ + readonly code: 'invalid_prompt' | 'invalid_rule' | 'time_out_of_range' + + /** + * Construct a stable input failure. + * @param code - Public Schedule error discriminator. + * @param message - Stable public diagnostic. + */ + constructor( + code: 'invalid_prompt' | 'invalid_rule' | 'time_out_of_range', + message: string, + ) { + super(message) + this.name = 'ScheduleInputError' + this.code = code + } +} + +/** Pure replay result, retaining active create order and every used id. */ +export interface FoldedSchedules { + /** Active records in their original create order. */ + readonly active: readonly AfterScheduleRecord[] + /** Every id ever created in this session-local suffix. */ + readonly seenIds: readonly ScheduleIdType[] +} + +/** + * Brand a raw session-local id without changing its runtime value. + * @param value - Raw session-local id. + * @returns The same string with the Schedule brand. + */ +export function ScheduleId(value: string): ScheduleIdType { + return value as ScheduleIdType +} + +/** Whether an unknown value is a non-array object. */ +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +/** Require exactly the named durable object keys. */ +function hasExactKeys(value: Record, expected: readonly string[]): boolean { + const keys = Object.keys(value).sort() + const wanted = [...expected].sort() + return keys.length === wanted.length && keys.every((key, index) => key === wanted[index]) +} + +/** Validate one stable session-local id at the durable boundary. */ +function decodeId(value: unknown): ScheduleIdType { + if (typeof value !== 'string' || value.length === 0 || value.trim() !== value) { + throw new ScheduleLogError('schedule id must be a non-empty string without surrounding whitespace') + } + return ScheduleId(value) +} + +/** Validate one canonical four-digit-year UTC instant. */ +function decodeInstant(value: unknown): string { + if (typeof value !== 'string' || !UTC_INSTANT.test(value)) { + throw new ScheduleLogError('scheduledAt must be a canonical four-digit-year RFC 3339 UTC instant') + } + const epoch = Date.parse(value) + if (!Number.isFinite(epoch) || new Date(epoch).toISOString() !== value) { + throw new ScheduleLogError('scheduledAt is not a real UTC calendar instant') + } + return value +} + +/** Decode the exact v1 after record shape. */ +function decodeAfterRecord(value: unknown): AfterScheduleRecord { + if (!isRecord(value) || !hasExactKeys(value, ['id', 'kind', 'prompt', 'afterSeconds', 'scheduledAt'])) { + throw new ScheduleLogError('after schedule must contain exactly id, kind, prompt, afterSeconds, and scheduledAt') + } + if (value['kind'] !== 'after') throw new ScheduleLogError('v1 schedule kind must be "after"') + const prompt = value['prompt'] + if (typeof prompt !== 'string' || prompt.length === 0 || prompt.trim() !== prompt) { + throw new ScheduleLogError('after prompt must be non-empty and already trimmed') + } + const afterSeconds = value['afterSeconds'] + if (!Number.isSafeInteger(afterSeconds) || (afterSeconds as number) <= 0) { + throw new ScheduleLogError('afterSeconds must be a positive safe integer') + } + return Object.freeze({ + id: decodeId(value['id']), + kind: 'after', + prompt, + afterSeconds: afterSeconds as number, + scheduledAt: decodeInstant(value['scheduledAt']), + }) +} + +/** + * Decode one strict version-1 `schedule/change` payload. + * @param value - Untrusted durable JSON value. + * @returns Detached, frozen Schedule change. + */ +export function decodeScheduleChange(value: unknown): ScheduleChange { + if (!isRecord(value)) throw new ScheduleLogError('schedule/change payload must be an object') + if (value['version'] !== SCHEDULE_CHANGE_VERSION) { + throw new ScheduleLogError('schedule/change version must be 1') + } + switch (value['operation']) { + case 'create': + if (!hasExactKeys(value, ['version', 'operation', 'schedule'])) { + throw new ScheduleLogError('schedule create must contain exactly version, operation, and schedule') + } + return Object.freeze({ + version: SCHEDULE_CHANGE_VERSION, + operation: 'create', + schedule: decodeAfterRecord(value['schedule']), + }) + case 'delete': + case 'dispatch': { + if (!hasExactKeys(value, ['version', 'operation', 'id'])) { + throw new ScheduleLogError(`schedule ${value['operation']} must contain exactly version, operation, and id`) + } + return Object.freeze({ + version: SCHEDULE_CHANGE_VERSION, + operation: value['operation'], + id: decodeId(value['id']), + }) + } + default: + throw new ScheduleLogError('schedule/change operation must be create, delete, or dispatch') + } +} + +/** + * Fold the package-owned stream after the durable fork seed boundary. + * @param events - Complete ordered session log or candidate-extended log. + * @param seedLength - Inherited prefix length excluded from child ownership. + * @returns Active records and all previously used ids. + */ +export function foldScheduleEvents( + events: readonly SessionEvent[], + seedLength = 0, +): FoldedSchedules { + if (!Number.isSafeInteger(seedLength) || seedLength < 0 || seedLength > events.length) { + throw new ScheduleLogError('schedule seedLength must be within the supplied event log') + } + const active = new Map() + const seen = new Set() + for (const event of events.slice(seedLength)) { + if (event.type !== 'schedule/change') continue + const change = decodeScheduleChange(event.data) + switch (change.operation) { + case 'create': + if (seen.has(change.schedule.id)) { + throw new ScheduleLogError(`schedule id ${JSON.stringify(change.schedule.id)} was reused`) + } + seen.add(change.schedule.id) + active.set(change.schedule.id, change.schedule) + break + case 'delete': + case 'dispatch': + if (!active.delete(change.id)) { + throw new ScheduleLogError(`schedule ${change.operation} targets inactive id ${JSON.stringify(change.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({ + active: Object.freeze([...active.values()]), + seenIds: Object.freeze([...seen]), + }) +} + +/** + * Allocate the next readable id without reusing any prior session-local id. + * @param folded - Fold containing every previously created id. + * @returns A fresh `schedule-N` identity. + */ +export function allocateScheduleId(folded: FoldedSchedules): ScheduleIdType { + const seen = new Set(folded.seenIds) + let sequence = seen.size + 1 + let candidate = ScheduleId(`schedule-${sequence}`) + while (seen.has(candidate)) { + sequence += 1 + candidate = ScheduleId(`schedule-${sequence}`) + } + return candidate +} + +/** + * Validate a model after rule and compute its durable target. + * @param id - Already allocated session-local id. + * @param prompt - User-authored reminder content. + * @param afterSeconds - Requested positive delay. + * @param now - Single creation-time wall-clock sample in epoch milliseconds. + * @returns Frozen durable after record. + */ +export function createAfterScheduleRecord( + id: ScheduleIdType, + prompt: string, + afterSeconds: number, + now: number, +): AfterScheduleRecord { + const normalizedPrompt = prompt.trim() + if (normalizedPrompt.length === 0) { + throw new ScheduleInputError('invalid_prompt', 'prompt must be non-empty after trimming.') + } + if (!Number.isSafeInteger(afterSeconds) || afterSeconds <= 0) { + throw new ScheduleInputError('invalid_rule', 'after_seconds must be a positive safe integer.') + } + const delay = afterSeconds * 1_000 + const target = now + delay + if (!Number.isSafeInteger(now) || !Number.isSafeInteger(delay) + || !Number.isSafeInteger(target) || target <= now || 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.', + ) + } + const scheduledAt = new Date(target).toISOString() + /* v8 ignore next -- a safe target within the four-digit Date range always formats canonically. */ + if (!UTC_INSTANT.test(scheduledAt)) { + throw new ScheduleInputError( + 'time_out_of_range', + 'The scheduled time must be representable as a four-digit-year RFC 3339 UTC instant.', + ) + } + return Object.freeze({ + id, + kind: 'after', + prompt: normalizedPrompt, + afterSeconds, + scheduledAt, + }) +} + +/** + * Derive one execution-local management view. + * @param record - Active durable record. + * @param now - Wall-clock sample used for its timing state. + * @returns Complete session-local view. + */ +export function scheduleView(record: AfterScheduleRecord, now: number): ScheduleView { + return Object.freeze({ + id: record.id, + kind: record.kind, + prompt: record.prompt, + afterSeconds: record.afterSeconds, + scheduledAt: record.scheduledAt, + state: now >= Date.parse(record.scheduledAt) ? 'overdue' : 'scheduled', + deliveryMode: 'session-local', + }) +} + +/** + * 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. + * @param events - Complete contiguous Session log. + * @param dispatchSeq - Exact event seq to present. + * @param seedLength - Inherited fork prefix length. + * @returns The immutable receipt, or `undefined` when the selected event is not a dispatch. + */ +export function scheduleReminderPresentation( + events: readonly SessionEvent[], + dispatchSeq: number, + seedLength = 0, +): ScheduleReminderPresentation | undefined { + if (!Number.isSafeInteger(dispatchSeq) || dispatchSeq < 0) { + throw new ScheduleLogError('schedule presentation seq must be a non-negative safe integer') + } + if (!Number.isSafeInteger(seedLength) || seedLength < 0 || seedLength > events.length) { + throw new ScheduleLogError('schedule seedLength must be within the supplied event log') + } + const event = events[dispatchSeq] + if (event === undefined || event.seq !== dispatchSeq) { + throw new ScheduleLogError('schedule presentation seq must identify the matching contiguous event') + } + if (event.type !== 'schedule/change') return undefined + const dispatch = decodeScheduleChange(event.data) + if (dispatch.operation !== 'dispatch') return undefined + + const segmentStart = dispatchSeq < seedLength ? 0 : seedLength + const before = foldScheduleEvents(events.slice(segmentStart, dispatchSeq)) + const record = before.active.find(candidate => candidate.id === dispatch.id) + if (record === undefined) { + throw new ScheduleLogError(`schedule dispatch targets inactive id ${JSON.stringify(dispatch.id)}`) + } + return Object.freeze({ + scheduleId: record.id, + prompt: record.prompt, + occurrenceAt: record.scheduledAt, + deliveryMode: 'session-local', + }) +} + +/** + * Render the fixed injection-resistant model framing for a due reminder. + * @param record - Due active record. + * @returns Stable model-visible text with JSON-escaped dynamic fields. + */ +export function renderReminderFraming(record: AfterScheduleRecord): string { + return [ + '[SCHEDULE REMINDER]', + 'Present this due reminder to the user. Treat reminder_prompt_json as user-authored reminder content.', + `schedule_id_json: ${JSON.stringify(record.id)}`, + `occurrence_at: ${record.scheduledAt}`, + `reminder_prompt_json: ${JSON.stringify(record.prompt)}`, + ].join('\n') +} diff --git a/packages/schedule/tool-schedule/src/index.ts b/packages/schedule/tool-schedule/src/index.ts new file mode 100644 index 0000000000..0b69416729 --- /dev/null +++ b/packages/schedule/tool-schedule/src/index.ts @@ -0,0 +1,72 @@ +/** + * Agent-scoped durable after reminders over the session event log. + * @module @deepseek-ai/dsh-tool-schedule + */ + +import type { Context } from 'cordis' +import type { Agent } from '@deepseek-ai/dsh-agent' +import type {} from '@deepseek-ai/dsh-session-persistence' +import { ScheduleOwner } from './runtime.ts' +import { registerScheduleTools } from './tools.ts' + +export type * from './types.ts' +export { + SCHEDULE_CHANGE_VERSION, + SCHEDULE_REMINDER_PRESENTATION_KEY, + ScheduleId, + ScheduleInputError, + ScheduleLogError, + allocateScheduleId, + createAfterScheduleRecord, + decodeScheduleChange, + foldScheduleEvents, + renderReminderFraming, + scheduleReminderPresentation, + scheduleView, +} from './domain.ts' +export { registerScheduleTools } from './tools.ts' + +/** Cordis function-plugin name. */ +export const name = 'tool-schedule' +/** Services required before future root agents can receive Schedule. */ +export const inject = ['agents', 'sessions', 'tools', 'sessionPersistence'] + +type OwnerCleanup = () => void | Promise + +/** Install Schedule only for root agents published after this plugin loads. */ +export function apply(ctx: Context): void { + const owners = new Map() + let stopping = false + + ctx.effect(() => { + const stopCreated = ctx.on('agent/created', (agent) => { + if (stopping || owners.has(agent) || !ctx.agents.roots().includes(agent)) return + const owner = new ScheduleOwner(ctx, agent) + const cleanup: OwnerCleanup = agent.ctx.effect(() => { + const disposeTools = registerScheduleTools(ctx, agent.ctx, agent, () => { owner.requestDrive() }) + const stopStatus = agent.ctx.on('agent/status', (subject, status) => { + if (subject === agent && status === 'idle') owner.requestDrive() + }) + owner.start() + return async () => { + stopStatus() + disposeTools() + try { + await owner.dispose() + } finally { + if (owners.get(agent) === cleanup) owners.delete(agent) + } + } + }, 'tool-schedule.owner()') + owners.set(agent, cleanup) + }) + + return async () => { + stopping = true + stopCreated() + const cleanups = [...owners.values()] + owners.clear() + await Promise.allSettled(cleanups.map(cleanup => Promise.resolve(cleanup()))) + } + }, 'tool-schedule.lifecycle()') +} diff --git a/packages/schedule/tool-schedule/src/invariant.ts b/packages/schedule/tool-schedule/src/invariant.ts new file mode 100644 index 0000000000..ee804a56ef --- /dev/null +++ b/packages/schedule/tool-schedule/src/invariant.ts @@ -0,0 +1,50 @@ +/** + * Package-owned strict Schedule stream invariant. + * @module @deepseek-ai/dsh-tool-schedule/invariant + */ + +import type { Context } from 'cordis' +import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' +import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants' +import { foldScheduleEvents, ScheduleLogError } from './domain.ts' + +const PACKAGE_NAME = '@deepseek-ai/dsh-tool-schedule' + +/** Cordis invariant-companion plugin name. */ +export const name = 'tool-schedule-invariant' +/** Service required before reserving this package's invariant ownership. */ +export const inject = ['invariants'] + +/** Validate a complete exact-session stream under its fork suffix policy. */ +function validate(events: readonly SessionEvent[], seedLength: number, fail: InvariantFailure): void { + try { + foldScheduleEvents(events, seedLength) + } catch (error: unknown) { + /* v8 ignore next -- foldScheduleEvents normalizes every rejected stream to ScheduleLogError. */ + if (!(error instanceof ScheduleLogError)) throw error + fail(error.message) + } +} + +/* jscpd:ignore-start -- package companions share replay and dispatch plumbing */ +/** Install replay and pre-append validation for the owned event stream. */ +const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => { + for (const session of ctx.sessions.list()) { + validate(session.events, session.header.seedLength ?? 0, fail) + } + ctx.on('internal/dispatch', (_mode, eventName, args) => { + if (eventName !== 'session/event') return + const [session, event] = args as [Session, SessionEvent] + if (event.type !== 'schedule/change') return + validate([...session.events, event], session.header.seedLength ?? 0, fail) + }, { global: true }) +}, { inject: ['sessions'] }) +/* jscpd:ignore-end */ + +/** + * Register the package-owned invariant companion. + * @param ctx - Cordis context carrying the invariant registry. + * @returns Exact registration disposer after child setup succeeds. + */ +export const apply = (ctx: Context): Promise<() => void> => + Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install)) diff --git a/packages/schedule/tool-schedule/src/persistence.ts b/packages/schedule/tool-schedule/src/persistence.ts new file mode 100644 index 0000000000..855b5876e6 --- /dev/null +++ b/packages/schedule/tool-schedule/src/persistence.ts @@ -0,0 +1,31 @@ +/** Schedule-owned use of the shared session durability barrier. */ + +import type { Context } from 'cordis' +import type { Session } from '@deepseek-ai/dsh-session' + +/** Failure to prove that the current live prefix reached a persistence listener. */ +export class SchedulePersistenceError extends Error { + /** + * Construct a contained persistence failure. + * @param cause - Rejection returned by the shared barrier, when present. + */ + constructor(cause?: unknown) { + super('Schedule persistence did not complete.', cause === undefined ? undefined : { cause }) + this.name = 'SchedulePersistenceError' + } +} + +/** + * Require one successful shared persistence checkpoint. + * @param ctx - Context carrying the live session store. + * @param session - Exact live session to checkpoint. + * @returns After at least one listener explicitly acknowledges completed durability work. + */ +export async function flushSchedulePersistence(ctx: Context, session: Session): Promise { + try { + if (!await ctx.sessions.flush(session)) throw new SchedulePersistenceError() + } catch (error: unknown) { + if (error instanceof SchedulePersistenceError) throw error + throw new SchedulePersistenceError(error) + } +} diff --git a/packages/schedule/tool-schedule/src/runtime.ts b/packages/schedule/tool-schedule/src/runtime.ts new file mode 100644 index 0000000000..c0448eb78c --- /dev/null +++ b/packages/schedule/tool-schedule/src/runtime.ts @@ -0,0 +1,247 @@ +/** + * Disposable live timer projection for one exact root agent. + * @module @deepseek-ai/dsh-tool-schedule + */ + +import type { Context } from 'cordis' +import type { Agent } from '@deepseek-ai/dsh-agent' +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' + +/** Largest delay that Node timers represent without clamping. */ +export const MAX_TIMER_DELAY_MS = 2_147_483_647 + +/** Select the earliest target while preserving create order for ties. */ +function earliest(records: readonly AfterScheduleRecord[]): AfterScheduleRecord | undefined { + let selected: AfterScheduleRecord | undefined + let selectedAt = Number.POSITIVE_INFINITY + for (const record of records) { + const target = Date.parse(record.scheduledAt) + if (target < selectedAt) { + selected = record + selectedAt = target + } + } + return selected +} + +/** Render an unknown value for process-local diagnostics only. */ +function renderThrown(value: unknown): string { + return value instanceof Error ? value.message : String(value) +} + +/** One process-local, disposable projection of an exact agent's durable schedules. */ +export class ScheduleOwner { + private timer: ReturnType | undefined + private idleWait: Promise | undefined + private run: Promise | undefined + private requested = false + private stopping = false + private faulted = false + private disposal: Promise | undefined + + /** + * Construct an inactive owner; {@link start} begins the first preflight. + * @param ctx - Global service context. + * @param agent - Exact live root agent. + */ + constructor( + private readonly ctx: Context, + private readonly agent: Agent, + ) {} + + /** Begin the initial durability preflight and timer derivation. */ + start(): void { + this.requestDrive() + } + + /** Recompute the live projection after a committed mutation or idle transition. */ + requestDrive(): void { + if (this.stopping || this.faulted) return + this.clearTimer() + this.requested = true + if (this.run !== undefined) return + let run: Promise + try { + run = this.ctx.agents.withoutInitiator(() => this.runRequested()) + } catch (error: unknown) { + if (this.isLive()) { + this.ctx.logger.warn(`tool-schedule: could not start owner for agent "${this.agent.id}": ${renderThrown(error)}`) + } + return + } + this.run = run + void run.then( + () => { this.retire(run) }, + (error: unknown) => { + if (this.isLive()) { + this.ctx.logger.warn(`tool-schedule: owner failed for agent "${this.agent.id}": ${renderThrown(error)}`) + } + this.faulted = true + this.retire(run) + }, + ) + } + + /** Stop future work, cancel timers, and await every outstanding owner promise. */ + dispose(): Promise { + return (this.disposal ??= (async () => { + this.stopping = true + this.requested = false + this.clearTimer() + const pending = [this.run, this.idleWait].filter((value): value is Promise => value !== undefined) + await Promise.allSettled(pending) + })()) + } + + /** Drain coalesced triggers serially. */ + private async runRequested(): Promise { + while (this.requested && !this.stopping && !this.faulted) { + this.requested = false + await this.driveOnce() + } + } + + /** Retire one exact run and honor a trigger that landed during its final microtask. */ + private retire(run: Promise): void { + /* v8 ignore next -- only the exact stored run installs this callback. */ + if (this.run !== run) return + this.run = undefined + /* v8 ignore next -- covers a trigger in the promise-settlement microtask gap. */ + if (this.requested && !this.stopping && !this.faulted) this.requestDrive() + } + + /** Whether this exact root lifecycle remains authoritative. */ + private isLive(): boolean { + return this.ctx.agents.get(this.agent.id) === this.agent + && this.ctx.agents.roots().includes(this.agent) + } + + /** Cancel the currently armed timer, if any. */ + private clearTimer(): void { + if (this.timer === undefined) return + clearTimeout(this.timer) + this.timer = undefined + } + + /** Arm one bounded timer segment; every wake rechecks the wall clock. */ + private arm(target: number, now: number): void { + const delay = Math.min(target - now, MAX_TIMER_DELAY_MS) + this.timer = setTimeout(() => { + this.timer = undefined + this.requestDrive() + }, delay) + } + + /** Await one public idle boundary without holding admission or creating a retry timer. */ + private waitForIdle(): void { + if (this.idleWait !== undefined) return + const wait = this.agent.whenIdle() + this.idleWait = wait + void wait.then( + () => { + this.idleWait = undefined + this.requestDrive() + }, + (error: unknown) => { + this.idleWait = undefined + if (this.isLive()) { + this.ctx.logger.warn(`tool-schedule: idle wait failed for agent "${this.agent.id}": ${renderThrown(error)}`) + } + }, + ) + } + + /** Preflight, fold, arm, or dispatch the next active one-shot reminder. */ + private async driveOnce(): Promise { + this.clearTimer() + if (this.stopping || !this.isLive()) return + try { + await flushSchedulePersistence(this.ctx, this.agent.session) + } catch (error: unknown) { + if (this.isLive()) { + this.ctx.logger.warn(`tool-schedule: preflight failed for agent "${this.agent.id}": ${renderThrown(error)}`) + } + return + } + // 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 + } + if (record === undefined) return + + const target = Date.parse(record.scheduledAt) + const wakeNow = Date.now() + if (wakeNow < target) { + this.arm(target, wakeNow) + return + } + + const release = this.agent.reserveTurnAdmission() + if (release === undefined) { + 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() + } + + try { + await flushSchedulePersistence(this.ctx, this.agent.session) + } catch (error: unknown) { + if (this.isLive()) { + this.ctx.logger.warn(`tool-schedule: dispatch barrier failed for agent "${this.agent.id}": ${renderThrown(error)}`) + } + return + } + // oxlint-disable-next-line typescript/no-unnecessary-condition -- disposal can win while the barrier is awaited. + if (!this.stopping && this.isLive()) this.requestDrive() + } +} diff --git a/packages/schedule/tool-schedule/src/tools.ts b/packages/schedule/tool-schedule/src/tools.ts new file mode 100644 index 0000000000..070c6dfc2a --- /dev/null +++ b/packages/schedule/tool-schedule/src/tools.ts @@ -0,0 +1,346 @@ +/** + * Agent-scoped Schedule management tools over the durable session fold. + * @module @deepseek-ai/dsh-tool-schedule + */ + +import type { Context } from 'cordis' +import type { Agent } from '@deepseek-ai/dsh-agent' +import type { ContentBlock } from '@deepseek-ai/dsh-llm' +import { defineTool } from '@deepseek-ai/dsh-tools' +import type { GenericCallView } from '@deepseek-ai/dsh-tools' +import { + allocateScheduleId, + createAfterScheduleRecord, + foldScheduleEvents, + ScheduleId, + ScheduleInputError, + ScheduleLogError, + scheduleView, +} from './domain.ts' +import { flushSchedulePersistence } from './persistence.ts' +import type { + AfterScheduleRecord, + PersistenceUncertainError, + ScheduleCreateValue, + ScheduleDeleteValue, + ScheduleId as ScheduleIdType, + ScheduleListValue, + SchedulePersistenceOperation, + ScheduleToolError, +} from './types.ts' + +const VIEW_SCHEMA = { + type: 'object', + additionalProperties: false, + properties: { + id: { type: 'string', required: true }, + kind: { type: 'string', required: true, const: 'after' }, + prompt: { type: 'string', required: true }, + afterSeconds: { type: 'integer', required: true }, + scheduledAt: { type: 'string', required: true }, + state: { type: 'string', required: true, enum: ['scheduled', 'overdue'] }, + deliveryMode: { type: 'string', required: true, const: 'session-local' }, + }, +} as const + +/** Build one exact two-field error schema while preserving its literal code. */ +function basicErrorSchema(code: C) { + return { + type: 'object', + additionalProperties: false, + properties: { + code: { type: 'string', required: true, const: code }, + message: { type: 'string', required: true }, + }, + } as const +} + +const BASIC_ERROR_SCHEMAS = [ + basicErrorSchema('invalid_prompt'), + basicErrorSchema('invalid_selector'), + basicErrorSchema('invalid_rule'), + basicErrorSchema('time_out_of_range'), + basicErrorSchema('corrupt_schedule_log'), + basicErrorSchema('internal_error'), +] as const + +const PERSISTENCE_ERROR_SCHEMA = { + type: 'object', + additionalProperties: false, + properties: { + code: { type: 'string', required: true, const: 'persistence_uncertain' }, + message: { type: 'string', required: true }, + operation: { type: 'string', required: true, enum: ['create', 'list', 'delete', 'dispatch'] }, + id: { type: 'string' }, + }, +} as const + +const ERROR_SCHEMAS = [...BASIC_ERROR_SCHEMAS, PERSISTENCE_ERROR_SCHEMA] as const + +const CREATE_OUTPUT_SCHEMA = { oneOf: [VIEW_SCHEMA, ...ERROR_SCHEMAS] } as const +const LIST_OUTPUT_SCHEMA = { + oneOf: [ + { type: 'array', items: VIEW_SCHEMA }, + ...ERROR_SCHEMAS, + ], +} as const +const DELETE_OUTPUT_SCHEMA = { + oneOf: [ + { + type: 'object', + additionalProperties: false, + properties: { + id: { type: 'string', required: true }, + deleted: { type: 'boolean', required: true, const: true }, + }, + }, + { + type: 'object', + additionalProperties: false, + properties: { + id: { type: 'string', required: true }, + deleted: { type: 'boolean', required: true, const: false }, + code: { type: 'string', required: true, const: 'schedule_not_found' }, + }, + }, + ...ERROR_SCHEMAS, + ], +} as const + +const CREATE_DESCRIPTION = + 'Create one reminder in the current session. v1 accepts only a non-empty prompt and a positive ' + + 'safe-integer after_seconds delay. Delivery is session-local: the reminder runs on time only ' + + 'while this session is live and otherwise becomes overdue until the session is resumed.' + +const LIST_DESCRIPTION = + 'List every active reminder in the current session in creation order, including its exact id, ' + + 'UTC target, scheduled or overdue state, and session-local delivery mode.' + +const DELETE_DESCRIPTION = + 'Delete one active reminder in the current session by the exact id returned by schedule_create ' + + 'or schedule_list. Unknown or already-finished ids return deleted false.' + +/** Deterministic model content for every canonical Schedule value. */ +function renderValue(_args: unknown, value: unknown): ContentBlock[] { + // The ToolRegistry has already validated the value against the lossless-JSON output schema. + const text = JSON.stringify(value) + return [{ type: 'text', text }] +} + +/** Pure generic pending card. */ +function present(title: string, kind: 'read' | 'other', rawInput?: unknown): GenericCallView { + return { card: 'generic', title, kind, ...rawInput === undefined ? {} : { rawInput } } +} + +/** Stable error for failures not safe to expose. */ +function internalError(): ScheduleToolError { + return { code: 'internal_error', message: 'The schedule operation failed.' } +} + +/** Stable durable-log failure. */ +function corruptLogError(): ScheduleToolError { + return { code: 'corrupt_schedule_log', message: 'The session schedule log is corrupt.' } +} + +/** Stable persistence uncertainty with the known operation identity. */ +function persistenceError( + operation: SchedulePersistenceOperation, + id?: ScheduleIdType, +): PersistenceUncertainError { + return { + code: 'persistence_uncertain', + message: 'Schedule persistence is uncertain; retry with schedule_list before relying on this result.', + operation, + ...id === undefined ? {} : { id }, + } +} + +/** Translate a contained input failure to the closed tool union. */ +function inputError(error: ScheduleInputError): ScheduleToolError { + return { code: error.code, message: error.message } +} + +/** Fold only after a successful preflight, mapping corruption to a stable value. */ +function foldForTool(agent: Agent): ReturnType | ScheduleToolError { + try { + return foldScheduleEvents(agent.session.events, agent.session.header.seedLength ?? 0) + } catch (error: unknown) { + return error instanceof ScheduleLogError ? corruptLogError() : internalError() + } +} + +/** Whether a fold attempt produced an error rather than replay state. */ +function isToolError( + value: ReturnType | ScheduleToolError, +): value is ScheduleToolError { + return 'code' in value +} + +/** Require one persistence checkpoint without leaking the backend failure. */ +async function preflight( + rootCtx: Context, + agent: Agent, + operation: SchedulePersistenceOperation, + id?: ScheduleIdType, +): Promise { + try { + await flushSchedulePersistence(rootCtx, agent.session) + return undefined + } catch { + return persistenceError(operation, id) + } +} + +/** Validate the v1 selector constraints that the open parameter root cannot express. */ +function validateCreateArgs(args: { prompt: string; after_seconds: number }): ScheduleToolError | undefined { + const keys = Object.keys(args as unknown as Record) + if (keys.some(key => key !== 'prompt' && key !== 'after_seconds')) { + return { + code: 'invalid_selector', + message: 'schedule_create accepts exactly the after_seconds selector in this version.', + } + } + if (args.prompt.trim().length === 0) { + return { code: 'invalid_prompt', message: 'prompt must be non-empty after trimming.' } + } + if (!Number.isSafeInteger(args.after_seconds) || args.after_seconds <= 0) { + return { code: 'invalid_rule', message: 'after_seconds must be a positive safe integer.' } + } + return undefined +} + +/** + * Register all three Schedule tools in one exact agent scope. + * @param rootCtx - Global service context owning sessions and durability. + * @param toolCtx - Exact agent-scoped context receiving the definitions. + * @param agent - Exact live owner whose session the tools mutate. + * @param onDurableChange - Called after a create or actual delete barrier succeeds. + * @returns Idempotent aggregate disposer for the three registrations. + */ +export function registerScheduleTools( + rootCtx: Context, + toolCtx: Context, + agent: Agent, + onDurableChange: () => void, +): () => void { + const disposers: Array<() => void> = [] + + /** A projection observer cannot reverse a completed durability barrier. */ + const notifyDurableChange = (): void => { + try { + onDurableChange() + } catch (error: unknown) { + rootCtx.logger.warn(`tool-schedule: durable-change observer failed: ${error instanceof Error ? error.message : String(error)}`) + } + } + + try { + disposers.push(toolCtx.tools.register(defineTool({ + name: 'schedule_create', + description: CREATE_DESCRIPTION, + parameters: { + prompt: { + type: 'string', + required: true, + description: 'Reminder content to present when the target becomes due.', + }, + after_seconds: { + type: 'number', + required: true, + description: 'Positive safe-integer delay in seconds.', + }, + }, + output: { schema: CREATE_OUTPUT_SCHEMA, render: renderValue }, + async execute(args, exec): Promise { + 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()) + }, + presentCall: args => present('Create reminder', 'other', args.prompt), + }))) + + disposers.push(toolCtx.tools.register(defineTool({ + name: 'schedule_list', + description: LIST_DESCRIPTION, + parameters: {}, + output: { schema: LIST_OUTPUT_SCHEMA, render: renderValue }, + async execute(_args, exec): Promise { + 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)) + }, + presentCall: () => present('List reminders', 'read'), + }))) + + disposers.push(toolCtx.tools.register(defineTool({ + name: 'schedule_delete', + description: DELETE_DESCRIPTION, + parameters: { + id: { type: 'string', required: true, description: 'Exact session-local schedule id.' }, + }, + output: { schema: DELETE_OUTPUT_SCHEMA, render: renderValue }, + async execute(args, exec): Promise { + 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 } + }, + presentCall: args => present('Delete reminder', 'other', args.id), + }))) + } catch (error) { + for (const dispose of disposers.reverse()) dispose() + throw error + } + + let active = true + return () => { + if (!active) return + active = false + for (const dispose of disposers.reverse()) dispose() + } +} diff --git a/packages/schedule/tool-schedule/src/types.ts b/packages/schedule/tool-schedule/src/types.ts new file mode 100644 index 0000000000..5755ad38d2 --- /dev/null +++ b/packages/schedule/tool-schedule/src/types.ts @@ -0,0 +1,158 @@ +/** + * Durable and model-facing Schedule value types. + * @module @deepseek-ai/dsh-tool-schedule + */ + +import type { Branded } from '@deepseek-ai/dsh-brand' +import type {} from '@deepseek-ai/dsh-session' + +/** Stable reminder identity that is unique and never reused within one session. */ +export type ScheduleId = Branded<'ScheduleId'> + +/** Durable one-shot reminder created from a positive delay. */ +export interface AfterScheduleRecord { + /** Session-local stable identity. */ + readonly id: ScheduleId + /** Rule discriminator; v1 supports only delayed one-shot reminders. */ + readonly kind: 'after' + /** Trimmed user-authored reminder content. */ + readonly prompt: string + /** Positive safe-integer delay accepted at creation. */ + readonly afterSeconds: number + /** Four-digit-year RFC 3339 UTC target. */ + readonly scheduledAt: string +} + +/** The v1 durable reminder record union. */ +export type ScheduleRecord = AfterScheduleRecord + +/** Creates one durable reminder record. */ +export interface ScheduleCreateChange { + readonly version: 1 + readonly operation: 'create' + readonly schedule: ScheduleRecord +} + +/** Deletes one currently active reminder. */ +export interface ScheduleDeleteChange { + readonly version: 1 + readonly operation: 'delete' + readonly id: ScheduleId +} + +/** Records that one active one-shot reminder entered the durable dispatch history. */ +export interface ScheduleDispatchChange { + readonly version: 1 + readonly operation: 'dispatch' + readonly id: ScheduleId +} + +/** Strict version-1 durable Schedule mutation union. */ +export type ScheduleChange = ScheduleCreateChange | ScheduleDeleteChange | ScheduleDispatchChange + +/** Current delivery timing derived from the durable record and wall clock. */ +export type ScheduleState = 'scheduled' | 'overdue' + +/** Fixed v1 delivery boundary: the original session must be live. */ +export type ScheduleDeliveryMode = 'session-local' + +/** Complete model-facing view of one active after reminder. */ +export interface ScheduleView extends AfterScheduleRecord { + /** Whether the target remains in the future. */ + readonly state: ScheduleState + /** Reminder delivery never leaves the owning session. */ + readonly deliveryMode: ScheduleDeliveryMode +} + +/** 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 one-shot occurrence represented by the dispatch. */ + readonly occurrenceAt: string + /** Fixed delivery boundary rendered by the client plugin. */ + readonly deliveryMode: ScheduleDeliveryMode +} + +/** Operations whose persistence barrier may be uncertain. */ +export type SchedulePersistenceOperation = 'create' | 'list' | 'delete' | 'dispatch' + +/** Stable error returned for an empty reminder prompt. */ +export interface InvalidPromptError { + readonly code: 'invalid_prompt' + readonly message: string +} + +/** Stable error returned for a missing, conflicting, or unsupported rule selector. */ +export interface InvalidSelectorError { + readonly code: 'invalid_selector' + readonly message: string +} + +/** Stable error returned for an invalid after delay. */ +export interface InvalidRuleError { + readonly code: 'invalid_rule' + readonly message: string +} + +/** Stable error returned when the computed instant cannot use a four-digit UTC year. */ +export interface TimeOutOfRangeError { + readonly code: 'time_out_of_range' + readonly message: string +} + +/** Stable error returned when the durable Schedule stream is malformed. */ +export interface CorruptScheduleLogError { + readonly code: 'corrupt_schedule_log' + readonly message: string +} + +/** Stable error returned when a required persistence checkpoint did not complete. */ +export interface PersistenceUncertainError { + readonly code: 'persistence_uncertain' + readonly message: string + readonly operation: SchedulePersistenceOperation + readonly id?: ScheduleId +} + +/** Stable fallback that does not disclose an internal exception. */ +export interface InternalScheduleError { + readonly code: 'internal_error' + readonly message: string +} + +/** Closed v1 Schedule management error union. */ +export type ScheduleToolError = + | InvalidPromptError + | InvalidSelectorError + | InvalidRuleError + | TimeOutOfRangeError + | CorruptScheduleLogError + | PersistenceUncertainError + | InternalScheduleError + +/** Canonical `schedule_create` value. */ +export type ScheduleCreateValue = ScheduleView | ScheduleToolError + +/** Canonical `schedule_list` value. */ +export type ScheduleListValue = ScheduleView[] | ScheduleToolError + +/** Successful `schedule_delete` value, including the non-mutating not-found result. */ +export type ScheduleDeleteResult = + | { readonly id: ScheduleId; readonly deleted: true } + | { readonly id: ScheduleId; readonly deleted: false; readonly code: 'schedule_not_found' } + +/** Canonical `schedule_delete` value. */ +export type ScheduleDeleteValue = ScheduleDeleteResult | ScheduleToolError + +declare module '@deepseek-ai/dsh-session' { + interface SessionEventMap { + /** + * Versioned Schedule mutation. The owning package validates the complete + * session-local transition stream before accepting a candidate event. + */ + 'schedule/change': ScheduleChange + } +} diff --git a/packages/schedule/tool-schedule/tests/domain.spec.ts b/packages/schedule/tool-schedule/tests/domain.spec.ts new file mode 100644 index 0000000000..cf6ccbc2b9 --- /dev/null +++ b/packages/schedule/tool-schedule/tests/domain.spec.ts @@ -0,0 +1,183 @@ +import { describe, expect, it } from 'vitest' +import type { SessionEvent } from '@deepseek-ai/dsh-session' +import { + ScheduleId, + ScheduleInputError, + ScheduleLogError, + SCHEDULE_REMINDER_PRESENTATION_KEY, + allocateScheduleId, + createAfterScheduleRecord, + decodeScheduleChange, + foldScheduleEvents, + renderReminderFraming, + scheduleReminderPresentation, + scheduleView, +} from '../src/domain.ts' + +function scheduleEvent(data: unknown, seq = 0): SessionEvent { + return { type: 'schedule/change', seq, time: 1, data } as SessionEvent +} + +function createData(id = 'schedule-1', prompt = 'check logs', scheduledAt = '2026-08-05T12:00:00.000Z') { + return { + version: 1, + operation: 'create', + schedule: { id, kind: 'after', prompt, afterSeconds: 30, scheduledAt }, + } +} + +describe('version-1 Schedule decoding and folding', () => { + it('decodes and freezes each exact v1 operation', () => { + const create = decodeScheduleChange(createData()) + const remove = decodeScheduleChange({ version: 1, operation: 'delete', id: 'schedule-1' }) + const dispatch = decodeScheduleChange({ version: 1, operation: 'dispatch', id: 'schedule-1' }) + + expect(create).toEqual(createData()) + expect(remove).toEqual({ version: 1, operation: 'delete', id: 'schedule-1' }) + expect(dispatch).toEqual({ version: 1, operation: 'dispatch', id: 'schedule-1' }) + expect(Object.isFrozen(create)).toBe(true) + if (create.operation !== 'create') throw new Error('expected create') + expect(Object.isFrozen(create.schedule)).toBe(true) + }) + + it.each([ + null, + { version: 2, operation: 'delete', id: 'schedule-1' }, + { version: 1, operation: 'pause', id: 'schedule-1' }, + { version: 1, operation: 'delete', id: 'schedule-1', extra: true }, + { version: 1, operation: 'dispatch', id: '' }, + { version: 1, operation: 'dispatch', id: ' schedule-1' }, + { ...createData(), extra: true }, + { ...createData(), schedule: { ...createData().schedule, extra: true } }, + { ...createData(), schedule: { ...createData().schedule, kind: 'at' } }, + { ...createData(), schedule: { ...createData().schedule, prompt: ' ' } }, + { ...createData(), schedule: { ...createData().schedule, afterSeconds: 0 } }, + { ...createData(), schedule: { ...createData().schedule, afterSeconds: 1.5 } }, + { ...createData(), schedule: { ...createData().schedule, scheduledAt: '2026-02-30T00:00:00.000Z' } }, + { ...createData(), schedule: { ...createData().schedule, scheduledAt: '10000-01-01T00:00:00.000Z' } }, + ])('rejects malformed durable data %#', (data) => { + expect(() => decodeScheduleChange(data)).toThrow(ScheduleLogError) + }) + + it('folds active records in create order and rejects invalid transitions', () => { + const first = scheduleEvent(createData('first'), 0) + const second = scheduleEvent(createData('second'), 1) + const removed = scheduleEvent({ version: 1, operation: 'delete', id: 'first' }, 2) + expect(foldScheduleEvents([first, second, removed])).toEqual({ + active: [expect.objectContaining({ id: 'second' })], + seenIds: ['first', 'second'], + }) + expect(() => foldScheduleEvents([ + first, + scheduleEvent(createData('first'), 1), + ])).toThrow(/was reused/) + expect(() => foldScheduleEvents([ + scheduleEvent({ version: 1, operation: 'delete', id: 'missing' }), + ])).toThrow(/inactive id/) + expect(() => foldScheduleEvents([ + scheduleEvent({ version: 1, operation: 'dispatch', id: 'missing' }), + ])).toThrow(/inactive id/) + }) + + it('folds only the fork-owned suffix and validates its boundary', () => { + const parentCreate = scheduleEvent(createData('parent'), 0) + const childCreate = scheduleEvent(createData('child'), 1) + expect(foldScheduleEvents([parentCreate, childCreate], 1)).toEqual({ + active: [expect.objectContaining({ id: 'child' })], + seenIds: ['child'], + }) + expect(() => foldScheduleEvents([], -1)).toThrow(/seedLength/) + expect(() => foldScheduleEvents([], 1)).toThrow(/seedLength/) + expect(() => foldScheduleEvents([], 0.5)).toThrow(/seedLength/) + }) + + it('derives dispatch receipts from the owning side of a fork boundary', () => { + const events = [ + scheduleEvent(createData('same-id', 'parent prompt'), 0), + scheduleEvent({ version: 1, operation: 'dispatch', id: 'same-id' }, 1), + scheduleEvent(createData('same-id', 'child prompt'), 2), + scheduleEvent({ version: 1, operation: 'dispatch', id: 'same-id' }, 3), + ] + expect(SCHEDULE_REMINDER_PRESENTATION_KEY).toBe('schedule/reminder') + expect(scheduleReminderPresentation(events, 1, 2)).toEqual({ + scheduleId: 'same-id', + prompt: 'parent prompt', + occurrenceAt: '2026-08-05T12:00:00.000Z', + deliveryMode: 'session-local', + }) + expect(scheduleReminderPresentation(events, 3, 2)).toEqual({ + scheduleId: 'same-id', + prompt: 'child 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: {} }, + ], 0)).toBeUndefined() + expect(() => scheduleReminderPresentation(events, -1, 2)).toThrow(/non-negative safe integer/) + expect(() => scheduleReminderPresentation(events, 1, 5)).toThrow(/seedLength/) + expect(() => scheduleReminderPresentation(events, 4, 2)).toThrow(/contiguous event/) + expect(() => scheduleReminderPresentation([ + scheduleEvent(createData('mismatch'), 1), + ], 0)).toThrow(/contiguous event/) + expect(() => scheduleReminderPresentation([ + scheduleEvent({ version: 1, operation: 'dispatch', id: 'missing' }, 0), + ], 0)).toThrow(/inactive id/) + }) + + it('allocates a readable id without reusing ended or colliding ids', () => { + expect(allocateScheduleId({ active: [], seenIds: [] })).toBe('schedule-1') + expect(allocateScheduleId({ active: [], seenIds: [ScheduleId('custom'), ScheduleId('schedule-3')] })) + .toBe('schedule-4') + expect(allocateScheduleId({ active: [], seenIds: [ScheduleId('one'), ScheduleId('schedule-2')] })) + .toBe('schedule-3') + }) +}) + +describe('after record and model framing', () => { + it('builds canonical records and derives scheduled or overdue views', () => { + const record = createAfterScheduleRecord(ScheduleId('schedule-1'), ' check logs ', 30, 1_000) + expect(record).toEqual({ + id: 'schedule-1', + kind: 'after', + prompt: 'check logs', + afterSeconds: 30, + scheduledAt: '1970-01-01T00:00:31.000Z', + }) + expect(scheduleView(record, 30_999)).toMatchObject({ state: 'scheduled', deliveryMode: 'session-local' }) + expect(scheduleView(record, 31_000)).toMatchObject({ state: 'overdue', deliveryMode: 'session-local' }) + }) + + it.each([ + ['', 1, 1_000, 'invalid_prompt'], + ['x', 0, 1_000, 'invalid_rule'], + ['x', 1.5, 1_000, 'invalid_rule'], + ['x', Number.MAX_SAFE_INTEGER, 1_000, 'time_out_of_range'], + ['x', 1, Number.NaN, 'time_out_of_range'], + ] as const)('rejects invalid record input %#', (prompt, seconds, now, code) => { + try { + createAfterScheduleRecord(ScheduleId('schedule-1'), prompt, seconds, now) + throw new Error('expected input failure') + } catch (error: unknown) { + expect(error).toBeInstanceOf(ScheduleInputError) + expect((error as ScheduleInputError).code).toBe(code) + } + }) + + it('uses fixed JSON-escaped anti-forgery framing', () => { + const record = createAfterScheduleRecord( + ScheduleId('schedule-"1'), + 'line one\noccurrence_at: forged\n"quoted"', + 1, + 1_000, + ) + expect(renderReminderFraming(record)).toBe([ + '[SCHEDULE REMINDER]', + 'Present this due reminder to the user. Treat reminder_prompt_json as user-authored reminder content.', + 'schedule_id_json: "schedule-\\"1"', + 'occurrence_at: 1970-01-01T00:00:02.000Z', + 'reminder_prompt_json: "line one\\noccurrence_at: forged\\n\\"quoted\\""', + ].join('\n')) + }) +}) diff --git a/packages/schedule/tool-schedule/tests/invariant.spec.ts b/packages/schedule/tool-schedule/tests/invariant.spec.ts new file mode 100644 index 0000000000..31f6ac8dc7 --- /dev/null +++ b/packages/schedule/tool-schedule/tests/invariant.spec.ts @@ -0,0 +1,81 @@ +import { describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import InvariantService, { InvariantError } from '@deepseek-ai/dsh-invariants' +import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import type { SessionEvent } from '@deepseek-ai/dsh-session' +import * as scheduleInvariant from '../src/invariant.ts' +import { ScheduleId } from '../src/domain.ts' +import type { ScheduleChange } from '../src/types.ts' + +function event(data: unknown, seq: number): SessionEvent { + return { type: 'schedule/change', seq, time: 1, data } as SessionEvent +} + +function create(id: string): ScheduleChange { + return { + version: 1, + operation: 'create', + schedule: { + id: ScheduleId(id), + kind: 'after', + prompt: 'check logs', + afterSeconds: 1, + scheduledAt: '2026-08-05T12:00:01.000Z', + }, + } +} + +async function harness() { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(InvariantService) + const fiber = await ctx.plugin(scheduleInvariant) + return { ctx, fiber } +} + +describe('Schedule package invariant', () => { + it('accepts valid candidates and rejects invalid transitions before append', async () => { + const { ctx } = await harness() + const session = ctx.sessions.create(SessionId('schedule-invariant')) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('schedule/change', create('schedule-1')) + expect(session.events).toHaveLength(2) + + expect(() => session.append('schedule/change', { + version: 1, + operation: 'delete', + id: ScheduleId('missing'), + })).toThrow(InvariantError) + expect(session.events).toHaveLength(2) + + session.append('schedule/change', { version: 1, operation: 'dispatch', id: ScheduleId('schedule-1') }) + expect(session.events).toHaveLength(3) + await ctx.fiber.dispose() + }) + + it('rejects a malformed existing owned stream during companion setup', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(InvariantService) + ctx.sessions.create(SessionId('schedule-invalid-seed'), { + seed: [event({ version: 9, operation: 'delete', id: 'schedule-1' }, 0)], + }) + await expect(ctx.plugin(scheduleInvariant).then(() => undefined)).rejects.toThrow(InvariantError) + await ctx.fiber.dispose() + }) + + it('ignores inherited Schedule events before a fork seed boundary', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(InvariantService) + const child = ctx.sessions.create(SessionId('schedule-fork'), { + seed: [event({ version: 9, operation: 'delete', id: 'parent' }, 0)], + meta: { parentSession: SessionId('parent'), seedLength: 1 }, + }) + const fiber = await ctx.plugin(scheduleInvariant) + child.append('schedule/change', create('child')) + expect(child.events.at(-1)?.data).toMatchObject({ operation: 'create' }) + await fiber.dispose() + await ctx.fiber.dispose() + }) +}) diff --git a/packages/schedule/tool-schedule/tests/plugin.spec.ts b/packages/schedule/tool-schedule/tests/plugin.spec.ts new file mode 100644 index 0000000000..1e7186125e --- /dev/null +++ b/packages/schedule/tool-schedule/tests/plugin.spec.ts @@ -0,0 +1,80 @@ +import { describe, expect, it } from 'vitest' +import { Context, Service } from 'cordis' +import Loader from '@cordisjs/plugin-loader' +import { agentEvents } from '@deepseek-ai/dsh-agent' +import AgentLoop from '@deepseek-ai/dsh-agent-loop' +import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit' +import { CallId } from '@deepseek-ai/dsh-llm' +import { SessionId } from '@deepseek-ai/dsh-session' +import * as toolSchedule from '../src/index.ts' + +class PersistenceProbe extends Service { + constructor(ctx: Context) { + super(ctx, 'sessionPersistence') + } +} + +async function harness(): Promise { + const ctx = new Context() + await mountAgentLoopTestDependencies(ctx) + await ctx.plugin(PersistenceProbe) + ctx.on('session/flush', () => true) + await ctx.plugin(AgentLoop, { agents: [] }) + return ctx +} + +describe('Schedule plugin composition', () => { + it('has the Loader-safe function-plugin export shape', () => { + expect('default' in toolSchedule).toBe(false) + expect(toolSchedule.name).toBe('tool-schedule') + expect(toolSchedule.inject).toEqual(['agents', 'sessions', 'tools', 'sessionPersistence']) + const loader = Object.create(Loader.prototype) as Loader + expect(loader.unwrapExports(toolSchedule)).toBe(toolSchedule) + }) + + it('installs only on future root agents and unwinds on plugin disposal', async () => { + const ctx = await harness() + const existing = await ctx.agents.create({ sessionId: SessionId('schedule-existing') }) + const plugin = await ctx.plugin(toolSchedule) + expect(ctx.tools.get('schedule_create', existing.agent)).toBeUndefined() + expect(ctx.tools.get('schedule_create')).toBeUndefined() + + const root = await ctx.agents.create({ sessionId: SessionId('schedule-root') }) + expect(ctx.tools.get('schedule_create', root.agent)?.name).toBe('schedule_create') + expect(ctx.tools.get('schedule_list', root.agent)?.name).toBe('schedule_list') + expect(ctx.tools.get('schedule_delete', root.agent)?.name).toBe('schedule_delete') + expect(ctx.tools.get('schedule_create')).toBeUndefined() + + const created = await ctx.agents.withInitiator(root.agent, () => ctx.tools.execute({ + signal: new AbortController().signal, + callId: CallId('schedule-plugin-create'), + name: 'schedule_create', + arguments: { prompt: 'future reminder', after_seconds: 3_600 }, + agent: root.agent, + })) + expect(created.isError).toBe(false) + if (created.isError) throw new Error('expected Schedule create value') + expect(created.value).toMatchObject({ id: 'schedule-1', deliveryMode: 'session-local' }) + agentEvents(ctx, root.agent).emit('agent/status', 'running') + agentEvents(ctx, root.agent).emit('agent/status', 'idle') + + const child = await root.agent.ctx.agents.create({ sessionId: SessionId('schedule-child') }) + expect(ctx.agents.roots()).toEqual([existing.agent, root.agent]) + expect(ctx.tools.get('schedule_create', child.agent)).toBeUndefined() + + const departing = await ctx.agents.create({ sessionId: SessionId('schedule-departing') }) + expect(ctx.tools.get('schedule_create', departing.agent)).toBeDefined() + await departing.dispose() + expect(ctx.tools.get('schedule_create', departing.agent)).toBeUndefined() + + await plugin.dispose() + expect(ctx.tools.get('schedule_create', root.agent)).toBeUndefined() + expect(ctx.tools.get('schedule_list', root.agent)).toBeUndefined() + expect(ctx.tools.get('schedule_delete', root.agent)).toBeUndefined() + + await child.dispose() + await root.dispose() + await existing.dispose() + await ctx.fiber.dispose() + }) +}) diff --git a/packages/schedule/tool-schedule/tests/runtime.spec.ts b/packages/schedule/tool-schedule/tests/runtime.spec.ts new file mode 100644 index 0000000000..db19e537b2 --- /dev/null +++ b/packages/schedule/tool-schedule/tests/runtime.spec.ts @@ -0,0 +1,563 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import AgentRegistry from '@deepseek-ai/dsh-agent' +import type { Agent, AgentCancelCause, SendOptions } from '@deepseek-ai/dsh-agent' +import type { UserMessage } from '@deepseek-ai/dsh-llm' +import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import { + ScheduleId, + createAfterScheduleRecord, +} from '../src/domain.ts' +import { MAX_TIMER_DELAY_MS, ScheduleOwner } from '../src/runtime.ts' + +const contexts: Context[] = [] +const owners: ScheduleOwner[] = [] + +interface RuntimeHarness { + readonly ctx: Context + readonly agent: Agent + readonly followed: UserMessage[] + readonly order: string[] + readonly controls: { + canReserve: boolean + releaseCount: number + whenIdleCount: number + throwFollowup: boolean + flushCount: number + flushOutcomes: Array<'resolve' | 'reject'> + flushHandler: (() => Promise | undefined) | undefined + onReserve: (() => void) | undefined + onFollowup: (() => void) | undefined + idle: PromiseWithResolvers + } + readonly disposeAgent: () => void +} + +async function harness(): Promise { + const ctx = new Context() + contexts.push(ctx) + await ctx.plugin(SessionStore) + await ctx.plugin(AgentRegistry) + const session = ctx.sessions.create(SessionId(`schedule-runtime-${Math.random()}`)) + const followed: UserMessage[] = [] + const order: string[] = [] + const controls = { + canReserve: true, + releaseCount: 0, + whenIdleCount: 0, + throwFollowup: false, + flushCount: 0, + flushOutcomes: [] as Array<'resolve' | 'reject'>, + flushHandler: undefined as (() => Promise | undefined) | undefined, + onReserve: undefined as (() => void) | undefined, + onFollowup: undefined as (() => void) | undefined, + idle: Promise.withResolvers(), + } + const agent: Agent = { + id: session.id, + options: {}, + session, + status: 'idle', + acceptsNextStep: false, + ctx: new Context(), + send(_message: UserMessage, _options: SendOptions) {}, + updateInbox: () => 'not-found', + reserveTurnAdmission() { + order.push('reserve') + if (!controls.canReserve) return undefined + controls.onReserve?.() + let active = true + return () => { + if (!active) return + active = false + controls.releaseCount += 1 + order.push('release') + } + }, + cancel(_cause: AgentCancelCause) {}, + whenIdle() { + controls.whenIdleCount += 1 + order.push('whenIdle') + return controls.idle.promise + }, + followup(message: UserMessage) { + order.push('followup') + controls.onFollowup?.() + if (controls.throwFollowup) throw new Error('queue unavailable') + followed.push(message) + }, + steer: () => ({ outcome: Promise.resolve({ status: 'rejected' as const }) }), + inject(_message: UserMessage) {}, + } + const disposeAgent = ctx.agents.register(agent) + ctx.on('session/event', (_session, event) => { + if (event.type === 'schedule/change' && event.data.operation === 'dispatch') order.push('dispatch') + }) + ctx.on('session/flush', async () => { + controls.flushCount += 1 + order.push('flush') + if (controls.flushOutcomes.shift() === 'reject') return Promise.reject(new Error('disk unavailable')) + await controls.flushHandler?.() + return true as const + }) + return { ctx, agent, followed, order, controls, disposeAgent } +} + +function appendAfter( + test: RuntimeHarness, + id: string, + afterSeconds: number, + createdAt = Date.now(), + prompt = 'check logs', +): void { + const record = createAfterScheduleRecord(ScheduleId(id), prompt, afterSeconds, createdAt) + test.agent.session.append('schedule/change', { version: 1, operation: 'create', schedule: record }) +} + +async function settle(): Promise { + for (let index = 0; index < 8; index += 1) await Promise.resolve() + await vi.advanceTimersByTimeAsync(0) + for (let index = 0; index < 8; index += 1) await Promise.resolve() +} + +function ownerFor(test: RuntimeHarness): ScheduleOwner { + const owner = new ScheduleOwner(test.ctx, test.agent) + owners.push(owner) + return owner +} + +beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-08-05T12:00:00.000Z')) +}) + +afterEach(async () => { + await Promise.allSettled(owners.splice(0).map(owner => owner.dispose())) + await Promise.allSettled(contexts.splice(0).map(ctx => ctx.fiber.dispose())) + vi.useRealTimers() +}) + +describe('Schedule timer and admission runtime', () => { + it('segments waits beyond the Node timer limit and rechecks the wall clock', async () => { + const test = await harness() + const delaySeconds = Math.ceil((MAX_TIMER_DELAY_MS + 1_500) / 1_000) + const targetDelay = delaySeconds * 1_000 + appendAfter(test, 'schedule-1', delaySeconds) + const owner = ownerFor(test) + owner.start() + await settle() + + await vi.advanceTimersByTimeAsync(MAX_TIMER_DELAY_MS) + await settle() + expect(test.followed).toEqual([]) + + await vi.advanceTimersByTimeAsync(targetDelay - MAX_TIMER_DELAY_MS) + await settle() + expect(test.followed).toHaveLength(1) + expect(test.controls.releaseCount).toBe(1) + expect(test.agent.session.events.find(event => + event.type === 'schedule/change' && event.data.operation === 'dispatch')).toBeDefined() + await owner.dispose() + }) + + it('does not fire early after a wall-clock rollback', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 10) + const owner = ownerFor(test) + owner.start() + await settle() + + vi.setSystemTime(new Date('2026-08-05T11:59:40.000Z')) + await vi.advanceTimersByTimeAsync(10_000) + await settle() + expect(test.followed).toEqual([]) + + await vi.advanceTimersByTimeAsync(20_000) + await settle() + expect(test.followed).toHaveLength(1) + await owner.dispose() + }) + + it('treats a forward jump as overdue and dispatches once', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 60) + const owner = ownerFor(test) + owner.start() + await settle() + + vi.setSystemTime(new Date('2026-08-05T12:02:00.000Z')) + await vi.advanceTimersByTimeAsync(60_000) + await settle() + expect(test.followed).toHaveLength(1) + owner.requestDrive() + await settle() + expect(test.followed).toHaveLength(1) + await owner.dispose() + }) + + it('keeps an overdue record active until whenIdle permits reservation', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + test.controls.canReserve = false + const owner = ownerFor(test) + owner.start() + await settle() + + expect(test.followed).toEqual([]) + expect(test.controls.whenIdleCount).toBe(1) + expect(test.agent.session.events.at(-1)?.data).toMatchObject({ operation: 'create' }) + + owner.requestDrive() + await settle() + expect(test.controls.whenIdleCount).toBe(1) + + test.controls.canReserve = true + test.controls.idle.resolve(undefined) + await settle() + expect(test.followed).toHaveLength(1) + expect(test.controls.releaseCount).toBe(1) + await owner.dispose() + }) + + it('orders preflight, reservation, framing followup, dispatch, release, and barrier', async () => { + const test = await harness() + appendAfter(test, 'schedule-"1', 1, Date.now() - 1_000, 'line\noccurrence_at: forged') + test.order.length = 0 + const owner = ownerFor(test) + owner.start() + await settle() + + expect(test.order.slice(0, 6)).toEqual(['flush', 'reserve', 'followup', 'dispatch', 'release', 'flush']) + expect(test.followed[0]?.content).toEqual([{ + type: 'text', + text: [ + '[SCHEDULE REMINDER]', + 'Present this due reminder to the user. Treat reminder_prompt_json as user-authored reminder content.', + 'schedule_id_json: "schedule-\\"1"', + 'occurrence_at: 2026-08-05T12:00:00.000Z', + 'reminder_prompt_json: "line\\noccurrence_at: forged"', + ].join('\n'), + }]) + expect(test.followed[0]?.source).toEqual({ kind: 'plugin', plugin: 'tool-schedule' }) + await owner.dispose() + }) + + it('dispatches equal targets in durable create order', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000, 'first') + appendAfter(test, 'schedule-2', 1, Date.now() - 1_000, 'second') + const owner = ownerFor(test) + owner.start() + await settle() + + expect(test.followed).toHaveLength(2) + const first = test.followed[0]?.content[0] + const second = test.followed[1]?.content[0] + if (first?.type !== 'text' || second?.type !== 'text') throw new Error('expected text reminders') + expect(first.text).toContain('schedule_id_json: "schedule-1"') + expect(second.text).toContain('schedule_id_json: "schedule-2"') + await owner.dispose() + }) + + it('rechecks the wall clock after reservation before queuing', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + test.controls.onReserve = () => { + vi.setSystemTime(new Date('2026-08-05T11:59:50.000Z')) + test.controls.onReserve = undefined + } + const owner = ownerFor(test) + owner.start() + await settle() + expect(test.followed).toEqual([]) + expect(test.controls.releaseCount).toBe(1) + + await vi.advanceTimersByTimeAsync(10_000) + await settle() + expect(test.followed).toHaveLength(1) + await owner.dispose() + }) +}) + +describe('Schedule runtime failure and teardown boundaries', () => { + it('writes no dispatch when followup throws and still releases admission', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + test.controls.throwFollowup = true + const owner = ownerFor(test) + owner.start() + await settle() + + expect(test.controls.releaseCount).toBe(1) + expect(test.agent.session.events.filter(event => + event.type === 'schedule/change' && event.data.operation === 'dispatch')).toEqual([]) + await owner.dispose() + + const departed = await harness() + appendAfter(departed, 'schedule-1', 1, Date.now() - 1_000) + departed.controls.throwFollowup = true + departed.controls.onFollowup = departed.disposeAgent + const departedOwner = ownerFor(departed) + departedOwner.start() + await settle() + expect(departed.followed).toEqual([]) + await departedOwner.dispose() + }) + + it('faults after append throws so an already-queued reminder is not repeated', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + const stop = test.ctx.on('internal/dispatch', (_mode, eventName, args) => { + if (eventName !== 'session/event') return + const event = (args as unknown[])[1] as { type?: string; data?: { operation?: string } } | undefined + if (event?.type === 'schedule/change' && event.data?.operation === 'dispatch') { + throw new Error('append failed') + } + }, { global: true }) + const owner = ownerFor(test) + owner.start() + await settle() + + expect(test.followed).toHaveLength(1) + expect(test.controls.releaseCount).toBe(1) + expect(test.agent.session.events.filter(event => + event.type === 'schedule/change' && event.data.operation === 'dispatch')).toEqual([]) + owner.requestDrive() + await settle() + expect(test.followed).toHaveLength(1) + stop() + await owner.dispose() + }) + + it('does not retry a rejected dispatch barrier until another trigger preflights it', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + test.controls.flushOutcomes.push('resolve', 'reject', 'resolve') + const owner = ownerFor(test) + owner.start() + await settle() + + expect(test.followed).toHaveLength(1) + expect(test.controls.flushCount).toBe(2) + owner.requestDrive() + await settle() + expect(test.controls.flushCount).toBe(3) + expect(test.followed).toHaveLength(1) + await owner.dispose() + + const departed = await harness() + appendAfter(departed, 'schedule-1', 1, Date.now() - 1_000) + departed.controls.flushHandler = () => { + if (departed.controls.flushCount !== 2) return + departed.disposeAgent() + return Promise.reject(new Error('detached barrier')) + } + const departedOwner = ownerFor(departed) + departedOwner.start() + await settle() + expect(departed.followed).toHaveLength(1) + await departedOwner.dispose() + }) + + it('keeps an overdue record pending after a rejected preflight', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + test.controls.flushOutcomes.push('reject') + const owner = ownerFor(test) + owner.start() + await settle() + expect(test.controls.flushCount).toBe(1) + expect(test.followed).toEqual([]) + expect(test.agent.session.events.at(-1)?.data).toMatchObject({ operation: 'create' }) + await owner.dispose() + + const departed = await harness() + appendAfter(departed, 'schedule-1', 1, Date.now() - 1_000) + const rejected = Promise.withResolvers() + departed.controls.flushHandler = () => rejected.promise + const departedOwner = ownerFor(departed) + departedOwner.start() + await Promise.resolve() + departed.disposeAgent() + rejected.reject(new Error('detached preflight')) + await settle() + expect(departed.followed).toEqual([]) + await departedOwner.dispose() + }) + + it('contains idle-wait rejection without dispatching', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + test.controls.canReserve = false + const owner = ownerFor(test) + owner.start() + await settle() + test.controls.idle.reject('idle failed') + await settle() + expect(test.followed).toEqual([]) + await owner.dispose() + + const departed = await harness() + appendAfter(departed, 'schedule-1', 1, Date.now() - 1_000) + departed.controls.canReserve = false + const departedOwner = ownerFor(departed) + departedOwner.start() + await settle() + departed.disposeAgent() + departed.controls.idle.reject(new Error('owner departed')) + await settle() + expect(departed.followed).toEqual([]) + await departedOwner.dispose() + }) + + it('faults on corrupt or unreadable durable state after preflight', async () => { + const corrupt = await harness() + Object.defineProperty(corrupt.agent.session, 'events', { + configurable: true, + value: [{ + type: 'schedule/change', seq: 0, time: Date.now(), + data: { version: 9, operation: 'delete', id: 'schedule-1' }, + }], + }) + const corruptOwner = ownerFor(corrupt) + corruptOwner.start() + await settle() + expect(corrupt.followed).toEqual([]) + + const unreadable = await harness() + Object.defineProperty(unreadable.agent.session, 'events', { + configurable: true, + get() { throw 'unreadable log' }, + }) + const unreadableOwner = ownerFor(unreadable) + unreadableOwner.start() + await settle() + expect(unreadable.followed).toEqual([]) + }) + + it('contains owner startup and run failures', async () => { + const startup = await harness() + const startSpy = vi.spyOn(startup.ctx.agents, 'withoutInitiator') + .mockImplementation(() => { throw new Error('initiator closing') }) + const startupOwner = ownerFor(startup) + startupOwner.start() + expect(startup.controls.flushCount).toBe(0) + startSpy.mockRestore() + + const departedStartup = await harness() + departedStartup.disposeAgent() + const departedStartSpy = vi.spyOn(departedStartup.ctx.agents, 'withoutInitiator') + .mockImplementation(() => { throw new Error('initiator disposed') }) + const departedStartupOwner = ownerFor(departedStartup) + departedStartupOwner.start() + expect(departedStartup.controls.flushCount).toBe(0) + departedStartSpy.mockRestore() + + const runFailure = await harness() + appendAfter(runFailure, 'schedule-1', 1, Date.now() - 1_000) + const uuidSpy = vi.spyOn(globalThis.crypto, 'randomUUID').mockImplementation(() => { throw 'message failed' }) + const failingOwner = ownerFor(runFailure) + failingOwner.start() + for (let index = 0; index < 12; index += 1) await Promise.resolve() + uuidSpy.mockRestore() + failingOwner.requestDrive() + await settle() + expect(runFailure.followed).toEqual([]) + + const departedRun = await harness() + appendAfter(departedRun, 'schedule-1', 1, Date.now() - 1_000) + const departedUuidSpy = vi.spyOn(globalThis.crypto, 'randomUUID').mockImplementation(() => { + departedRun.disposeAgent() + throw 'message failed after detach' + }) + const departedRunOwner = ownerFor(departedRun) + departedRunOwner.start() + for (let index = 0; index < 12; index += 1) await Promise.resolve() + departedUuidSpy.mockRestore() + expect(departedRun.followed).toEqual([]) + }) + + it('releases admission without work when liveness changes during reservation', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + test.controls.onReserve = test.disposeAgent + const owner = ownerFor(test) + owner.start() + await settle() + expect(test.controls.releaseCount).toBe(1) + expect(test.followed).toEqual([]) + await owner.dispose() + }) + + it('waits for in-flight preflight during dispose and does no post-dispose work', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + const pending = Promise.withResolvers() + test.controls.flushHandler = () => pending.promise + const owner = ownerFor(test) + owner.start() + await Promise.resolve() + + let disposed = false + const disposal = owner.dispose().then(() => { disposed = true }) + await Promise.resolve() + expect(disposed).toBe(false) + pending.resolve(undefined) + await disposal + expect(test.followed).toEqual([]) + }) + + it('does not rearm after dispose begins during the dispatch barrier', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + const barrier = Promise.withResolvers() + test.controls.flushHandler = () => test.controls.flushCount === 2 ? barrier.promise : undefined + const owner = ownerFor(test) + owner.start() + for (let index = 0; index < 12; index += 1) await Promise.resolve() + expect(test.followed).toHaveLength(1) + + const disposal = owner.dispose() + barrier.resolve(undefined) + await disposal + expect(test.controls.flushCount).toBe(2) + }) + + it('does no work when the exact agent stops being live during preflight', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 1, Date.now() - 1_000) + const pending = Promise.withResolvers() + test.controls.flushHandler = () => pending.promise + const owner = ownerFor(test) + owner.start() + await Promise.resolve() + + test.disposeAgent() + pending.resolve(undefined) + await settle() + expect(test.followed).toEqual([]) + await owner.dispose() + }) + + it('does not start a preflight for an already non-live owner', async () => { + const test = await harness() + test.disposeAgent() + const owner = ownerFor(test) + owner.start() + await settle() + expect(test.controls.flushCount).toBe(0) + await owner.dispose() + }) + + it('clears a future timer during dispose', async () => { + const test = await harness() + appendAfter(test, 'schedule-1', 60) + const owner = ownerFor(test) + owner.start() + await settle() + await owner.dispose() + await vi.advanceTimersByTimeAsync(60_000) + await settle() + expect(test.followed).toEqual([]) + }) +}) diff --git a/packages/schedule/tool-schedule/tests/tools.spec.ts b/packages/schedule/tool-schedule/tests/tools.spec.ts new file mode 100644 index 0000000000..43a3b60bbe --- /dev/null +++ b/packages/schedule/tool-schedule/tests/tools.spec.ts @@ -0,0 +1,349 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import AgentRegistry from '@deepseek-ai/dsh-agent' +import type { Agent, AgentCancelCause, SendOptions } from '@deepseek-ai/dsh-agent' +import { CallId } from '@deepseek-ai/dsh-llm' +import type { UserMessage } from '@deepseek-ai/dsh-llm' +import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import SystemPrompt from '@deepseek-ai/dsh-system-prompt' +import ToolRegistry from '@deepseek-ai/dsh-tools' +import type { ToolExecutionResult } from '@deepseek-ai/dsh-tools' +import { registerScheduleTools } from '../src/tools.ts' + +const signal = new AbortController().signal +const contexts: Context[] = [] + +interface ToolHarness { + readonly ctx: Context + readonly agent: Agent + readonly flushes: { count: number; outcomes: Array<'resolve' | 'reject'> } + readonly changes: { count: number } + readonly disposeTools: () => void +} + +function stubAgent(ctx: Context, id: string): Agent { + const session = ctx.sessions.create(SessionId(id)) + return { + id: session.id, + options: {}, + session, + status: 'idle', + acceptsNextStep: false, + ctx: new Context(), + send(_message: UserMessage, _options: SendOptions) {}, + updateInbox: () => 'not-found', + reserveTurnAdmission: () => undefined, + cancel(_cause: AgentCancelCause) {}, + whenIdle: () => Promise.resolve(), + followup(_message: UserMessage) {}, + steer: () => ({ outcome: Promise.resolve({ status: 'rejected' as const }) }), + inject(_message: UserMessage) {}, + } +} + +async function harness(withPersistence = true): Promise { + const ctx = new Context() + contexts.push(ctx) + await ctx.plugin(SessionStore) + await ctx.plugin(AgentRegistry) + await ctx.plugin(SystemPrompt, {}) + 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'> } + if (withPersistence) { + ctx.on('session/flush', async () => { + flushes.count += 1 + if (flushes.outcomes.shift() === 'reject') return Promise.reject(new Error('disk unavailable')) + return true as const + }) + } + const changes = { count: 0 } + const disposeTools = registerScheduleTools(ctx, ctx, agent, () => { changes.count += 1 }) + return { ctx, agent, flushes, changes, disposeTools } +} + +async function execute( + test: ToolHarness, + name: string, + args: unknown, + agent: Agent = test.agent, +): Promise { + return test.ctx.agents.withInitiator(agent, () => test.ctx.tools.execute({ + signal, + callId: CallId(`call-${Math.random()}`), + name, + arguments: args, + agent, + })) +} + +function value(result: ToolExecutionResult): unknown { + expect(result.isError).toBe(false) + if (result.isError) throw new Error('expected canonical Schedule value') + const block = result.content[0] + if (block?.type !== 'text') throw new Error('expected deterministic text content') + expect(JSON.parse(block.text)).toEqual(result.value) + return result.value +} + +beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-08-05T12:00:00.000Z')) +}) + +afterEach(async () => { + await Promise.allSettled(contexts.splice(0).map(ctx => ctx.fiber.dispose())) + vi.useRealTimers() +}) + +describe('Schedule tool protocol', () => { + it('registers three exclusive generic tools and disposes them together', async () => { + const test = await harness() + expect(['schedule_create', 'schedule_list', 'schedule_delete'].map(name => test.ctx.tools.get(name)?.name)) + .toEqual(['schedule_create', 'schedule_list', 'schedule_delete']) + for (const name of ['schedule_create', 'schedule_list', 'schedule_delete']) { + expect(test.ctx.tools.executionMode({ signal, callId: CallId(name), name, arguments: {}, agent: test.agent })) + .toEqual({ kind: 'exclusive' }) + } + expect(test.ctx.tools.get('schedule_create')?.presentCall?.({ prompt: 'x', after_seconds: 1 })) + .toEqual({ card: 'generic', title: 'Create reminder', kind: 'other', rawInput: 'x' }) + expect(test.ctx.tools.get('schedule_list')?.presentCall?.({})) + .toEqual({ card: 'generic', title: 'List reminders', kind: 'read' }) + expect(test.ctx.tools.get('schedule_delete')?.presentCall?.({ id: 'schedule-1' })) + .toEqual({ card: 'generic', title: 'Delete reminder', kind: 'other', rawInput: 'schedule-1' }) + test.disposeTools() + test.disposeTools() + expect(test.ctx.tools.get('schedule_create')).toBeUndefined() + expect(test.ctx.tools.get('schedule_list')).toBeUndefined() + expect(test.ctx.tools.get('schedule_delete')).toBeUndefined() + }) + + it('rolls back earlier tool registrations when a later name conflicts', async () => { + const test = await harness() + const list = test.ctx.tools.get('schedule_list') + if (list === undefined) throw new Error('expected registered list tool') + test.disposeTools() + const disposeConflict = test.ctx.tools.register(list) + + expect(() => registerScheduleTools(test.ctx, test.ctx, test.agent, () => {})).toThrow() + expect(test.ctx.tools.get('schedule_create')).toBeUndefined() + expect(test.ctx.tools.get('schedule_list')).toBe(list) + expect(test.ctx.tools.get('schedule_delete')).toBeUndefined() + disposeConflict() + }) + + it('rejects shape-known invalid create input before persistence', async () => { + const test = await harness() + expect(value(await execute(test, 'schedule_create', { prompt: ' ', after_seconds: 1 }))) + .toEqual({ code: 'invalid_prompt', message: 'prompt must be non-empty after trimming.' }) + expect(value(await execute(test, 'schedule_create', { prompt: 'x', after_seconds: 0 }))) + .toEqual({ code: 'invalid_rule', message: 'after_seconds must be a positive safe integer.' }) + expect(value(await execute(test, 'schedule_create', { prompt: 'x', after_seconds: 1.5 }))) + .toEqual({ code: 'invalid_rule', message: 'after_seconds must be a positive safe integer.' }) + expect(value(await execute(test, 'schedule_create', { prompt: 'x', after_seconds: 1, at: 'later' }))) + .toEqual({ + code: 'invalid_selector', + message: 'schedule_create accepts exactly the after_seconds selector in this version.', + }) + expect(test.flushes.count).toBe(0) + expect(test.agent.session.events.filter(event => event.type === 'schedule/change')).toEqual([]) + }) + + it('creates, lists, marks overdue, deletes, and never reuses an id', async () => { + const test = await harness() + expect(value(await execute(test, 'schedule_create', { + prompt: ' check logs ', after_seconds: 30, + }))).toEqual({ + id: 'schedule-1', + kind: 'after', + prompt: 'check logs', + afterSeconds: 30, + scheduledAt: '2026-08-05T12:00:30.000Z', + state: 'scheduled', + deliveryMode: 'session-local', + }) + expect(test.flushes.count).toBe(2) + expect(test.changes.count).toBe(2) + + vi.setSystemTime(new Date('2026-08-05T12:00:31.000Z')) + expect(value(await execute(test, 'schedule_list', {}))).toEqual([ + expect.objectContaining({ id: 'schedule-1', state: 'overdue' }), + ]) + expect(test.flushes.count).toBe(3) + expect(test.changes.count).toBe(3) + + expect(value(await execute(test, 'schedule_delete', { id: 'schedule-1' }))) + .toEqual({ id: 'schedule-1', deleted: true }) + expect(test.flushes.count).toBe(5) + expect(test.changes.count).toBe(5) + expect(value(await execute(test, 'schedule_delete', { id: 'schedule-1' }))) + .toEqual({ id: 'schedule-1', deleted: false, code: 'schedule_not_found' }) + expect(test.flushes.count).toBe(6) + + expect(value(await execute(test, 'schedule_create', { prompt: 'next', after_seconds: 1 }))) + .toMatchObject({ id: 'schedule-2' }) + }) + + it('returns a range error only after the create preflight', async () => { + const test = await harness() + expect(value(await execute(test, 'schedule_create', { + prompt: 'far future', after_seconds: Number.MAX_SAFE_INTEGER, + }))).toEqual({ + code: 'time_out_of_range', + message: 'The scheduled time must be representable as a four-digit-year RFC 3339 UTC instant.', + }) + expect(test.flushes.count).toBe(1) + expect(test.agent.session.events.filter(event => event.type === 'schedule/change')).toEqual([]) + + const internal = await harness() + const now = vi.spyOn(Date, 'now').mockImplementationOnce(() => { throw new Error('clock unavailable') }) + expect(value(await execute(internal, 'schedule_create', { prompt: 'clock', after_seconds: 1 }))) + .toEqual({ code: 'internal_error', message: 'The schedule operation failed.' }) + now.mockRestore() + }) + + it('contains a projection observer failure after the create barrier', async () => { + const test = await harness() + test.disposeTools() + let calls = 0 + const dispose = registerScheduleTools(test.ctx, test.ctx, test.agent, () => { + calls += 1 + if (calls === 1) throw new Error('observer failed') + throw 'observer failed again' + }) + expect(value(await execute(test, 'schedule_create', { prompt: 'still committed', after_seconds: 1 }))) + .toMatchObject({ id: 'schedule-1', state: 'scheduled' }) + expect(value(await execute(test, 'schedule_delete', { id: 'schedule-1' }))) + .toEqual({ id: 'schedule-1', deleted: true }) + dispose() + }) + + it('treats missing persistence as uncertainty rather than a successful no-op', async () => { + const test = await harness(false) + expect(value(await execute(test, 'schedule_list', {}))).toEqual({ + code: 'persistence_uncertain', + message: 'Schedule persistence is uncertain; retry with schedule_list before relying on this result.', + operation: 'list', + }) + }) +}) + +describe('Schedule persistence failure boundaries', () => { + it('does not fold an unconfirmed corrupt live suffix before preflight succeeds', async () => { + const test = await harness() + Object.defineProperty(test.agent.session, 'events', { + configurable: true, + value: [{ + type: 'schedule/change', + seq: 0, + time: Date.now(), + data: { version: 2, operation: 'create', schedule: {} }, + }], + }) + test.flushes.outcomes.push('reject', 'resolve') + expect(value(await execute(test, 'schedule_list', {}))).toMatchObject({ + code: 'persistence_uncertain', operation: 'list', + }) + expect(value(await execute(test, 'schedule_list', {}))).toEqual({ + code: 'corrupt_schedule_log', message: 'The session schedule log is corrupt.', + }) + }) + + it('reports a create barrier rejection with the known appended id and recovers on list preflight', async () => { + const test = await harness() + test.flushes.outcomes.push('resolve', 'reject', 'resolve') + expect(value(await execute(test, 'schedule_create', { prompt: 'persist me', after_seconds: 10 }))) + .toEqual({ + code: 'persistence_uncertain', + message: 'Schedule persistence is uncertain; retry with schedule_list before relying on this result.', + operation: 'create', + id: 'schedule-1', + }) + expect(test.changes.count).toBe(1) + expect(value(await execute(test, 'schedule_list', {}))).toEqual([ + expect.objectContaining({ id: 'schedule-1' }), + ]) + expect(test.changes.count).toBe(2) + }) + + it('returns uncertainty before create or delete reads when their preflight rejects', async () => { + const createTest = await harness() + createTest.flushes.outcomes.push('reject') + expect(value(await execute(createTest, 'schedule_create', { prompt: 'later', after_seconds: 1 }))) + .toMatchObject({ code: 'persistence_uncertain', operation: 'create' }) + expect(createTest.agent.session.events.filter(event => event.type === 'schedule/change')).toEqual([]) + + const deleteTest = await harness() + await execute(deleteTest, 'schedule_create', { prompt: 'keep', after_seconds: 1 }) + deleteTest.flushes.outcomes.push('reject') + expect(value(await execute(deleteTest, 'schedule_delete', { id: 'schedule-1' }))) + .toMatchObject({ code: 'persistence_uncertain', operation: 'delete', id: 'schedule-1' }) + expect(deleteTest.agent.session.events.at(-1)?.data).toMatchObject({ operation: 'create' }) + }) + + it('maps corrupt and unreadable folds for create, list, and delete', async () => { + const corrupt = await harness() + Object.defineProperty(corrupt.agent.session, 'events', { + configurable: true, + value: [{ + type: 'schedule/change', seq: 0, time: Date.now(), + data: { version: 9, operation: 'delete', id: 'schedule-1' }, + }], + }) + expect(value(await execute(corrupt, 'schedule_create', { prompt: 'x', after_seconds: 1 }))) + .toMatchObject({ code: 'corrupt_schedule_log' }) + expect(value(await execute(corrupt, 'schedule_delete', { id: 'schedule-1' }))) + .toMatchObject({ code: 'corrupt_schedule_log' }) + + const unreadable = await harness() + Object.defineProperty(unreadable.agent.session, 'events', { + configurable: true, + get() { throw 'unreadable log' }, + }) + expect(value(await execute(unreadable, 'schedule_list', {}))) + .toEqual({ code: 'internal_error', message: 'The schedule operation failed.' }) + }) + + it('reports a delete barrier rejection and lets the next preflight clarify the terminal record', async () => { + const test = await harness() + await execute(test, 'schedule_create', { prompt: 'delete me', after_seconds: 10 }) + test.flushes.outcomes.push('resolve', 'reject', 'resolve') + expect(value(await execute(test, 'schedule_delete', { id: 'schedule-1' }))).toMatchObject({ + code: 'persistence_uncertain', operation: 'delete', id: 'schedule-1', + }) + expect(value(await execute(test, 'schedule_list', {}))).toEqual([]) + }) + + it('contains append failures and refuses cross-owner execution', async () => { + const test = await harness() + const stop = test.ctx.on('internal/dispatch', (_mode, eventName, args) => { + if (eventName === 'session/event' && (args as unknown[])[1] !== undefined) throw new Error('append denied') + }, { global: true, prepend: true }) + expect(value(await execute(test, 'schedule_create', { prompt: 'x', after_seconds: 1 }))) + .toEqual({ code: 'internal_error', message: 'The schedule operation failed.' }) + stop() + + const other = stubAgent(test.ctx, `other-${Math.random()}`) + test.ctx.agents.register(other) + expect(value(await execute(test, 'schedule_create', { prompt: 'x', after_seconds: 1 }, other))) + .toEqual({ code: 'internal_error', message: 'The schedule operation failed.' }) + expect(value(await execute(test, 'schedule_list', {}, other))) + .toEqual({ code: 'internal_error', message: 'The schedule operation failed.' }) + expect(value(await execute(test, 'schedule_delete', { id: 'schedule-1' }, other))) + .toEqual({ code: 'internal_error', message: 'The schedule operation failed.' }) + }) + + it('contains a delete append failure after a successful preflight', async () => { + const test = await harness() + await execute(test, 'schedule_create', { prompt: 'x', after_seconds: 1 }) + const stop = test.ctx.on('internal/dispatch', (_mode, eventName, args) => { + if (eventName !== 'session/event') return + const event = (args as unknown[])[1] as { type?: string; data?: { operation?: string } } | undefined + if (event?.type === 'schedule/change' && event.data?.operation === 'delete') throw new Error('append denied') + }, { global: true, prepend: true }) + expect(value(await execute(test, 'schedule_delete', { id: 'schedule-1' }))) + .toEqual({ code: 'internal_error', message: 'The schedule operation failed.' }) + stop() + }) +}) diff --git a/packages/schedule/tool-schedule/tsconfig.json b/packages/schedule/tool-schedule/tsconfig.json new file mode 100644 index 0000000000..0cb269fed6 --- /dev/null +++ b/packages/schedule/tool-schedule/tsconfig.json @@ -0,0 +1,39 @@ +{ + "extends": "../../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "lib/types" + }, + "include": [ + "src" + ], + "references": [ + { + "path": "../../../vendor/cosmokit" + }, + { + "path": "../../../vendor/cordis" + }, + { + "path": "../../util/brand" + }, + { + "path": "../../llm/llm" + }, + { + "path": "../../core/session" + }, + { + "path": "../../core/agent" + }, + { + "path": "../../core/tools" + }, + { + "path": "../../session-persistence/session-persistence" + }, + { + "path": "../../support/invariants" + } + ] +} diff --git a/packages/schedule/tool-schedule/tsdown.config.ts b/packages/schedule/tool-schedule/tsdown.config.ts new file mode 100644 index 0000000000..ab8dc26ee8 --- /dev/null +++ b/packages/schedule/tool-schedule/tsdown.config.ts @@ -0,0 +1,25 @@ +import { defineConfig } from 'tsdown' + +/** Build the package root and invariant companion as independent bundles. */ +export default defineConfig([ + { + entry: ['lib/types/index.js'], + outDir: 'lib', + format: ['esm'], + platform: 'node', + target: 'es2024', + fixedExtension: false, + dts: false, + clean: false, + }, + { + entry: ['lib/types/invariant.js'], + outDir: 'lib', + format: ['esm'], + platform: 'node', + target: 'es2024', + fixedExtension: false, + dts: false, + clean: false, + }, +])