diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..0f021be --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,22 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + node-version: [18, 20, 22] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: ${{ matrix.node-version }} + cache: npm + - run: npm ci + - run: npm test diff --git a/README.md b/README.md index 40b00d9..3fd4c24 100644 --- a/README.md +++ b/README.md @@ -2,24 +2,31 @@ [![npm version](https://img.shields.io/npm/v/opencode-plugin-loop.svg)](https://www.npmjs.com/package/opencode-plugin-loop) [![npm downloads](https://img.shields.io/npm/dm/opencode-plugin-loop.svg)](https://www.npmjs.com/package/opencode-plugin-loop) +[![CI](https://github.com/jkrandom-sudo/opencode-plugin-loop/actions/workflows/ci.yml/badge.svg)](https://github.com/jkrandom-sudo/opencode-plugin-loop/actions/workflows/ci.yml) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/jkrandom-sudo/opencode-plugin-loop/blob/main/LICENSE) A drop-in `/loop` command for [opencode](https://opencode.ai), modeled after Claude Code's `/loop`. Each `/loop` task is bound to the session that created it — never leaks to other sessions. +> **Upgrading to 0.4.0?** Two behavior changes to know about: (1) since 0.3.0, tasks die with the opencode process by default (`ephemeralTasks: false` restores persistence) — the upgrade drops the pre-0.3.0 `tasks.json` once; (2) scheduling-like input that used to silently create an Adaptive task (cron syntax, bare intervals like `/loop 5m`, unknown flags) now returns an explicit error pointing at `/loop help`. + ## Features - **`/loop 5m `** — fixed interval (s/m/h/d supported) - **`/loop `** — runs immediately in Adaptive mode, then keeps the random fallback, reschedules from the result, or converts a clear recurring cadence to Fixed -- **`/loop`** — bare: read `.opencode/loop.md` or run built-in maintenance +- **`/loop`** — bare: read `.opencode/loop.md` or run built-in maintenance, immediately +- **`/loop 30s --once `** — one-shot: fires once, then auto-cancels +- **`/loop help`** — full usage, flags, and examples in the terminal +- **Claude Code-style flags** — `--cancel/--list/--status/--pause/--resume/--stop/--stop-all` map to the matching subcommand - **Per-session scoping** — tasks are bound to a `sessionID`; other sessions never see or fire them - **Subcommands** — `list | status | cancel | pause | resume | stop-all` (session-scoped; add `--all` to cross sessions) - **Internal ticker** — 5s loop drives task firing (no longer depends on `session.idle` events) +- **Single-leader instance lock** — when several plugin instances share one `tasks.json` (case-variant plugin paths, per-command `opencode run` instances), only the leader fires; merge-writes prevent task loss - **Inflight guard** — double-set at ticker and `fireTask` level prevents double-firing even if opencode hot-reloads the plugin -- **Persistent tasks** — survive session restarts; auto-migrated (tasks without `sessionID` are dropped on load) +- **Wall-clock scheduling** — fixed tasks anchor to fire start; model-turn duration never inflates the interval - **Ephemeral lifecycle (default)** — tasks die with the OpenCode process and are dropped on the next start, matching Claude Code's `/loop`. Set `ephemeralTasks: false` to persist tasks across process restarts - **Auto-cleanup on `session.deleted`** — all tasks for that session are cancelled automatically - **Configurable Jitter** — deterministic Fixed-task offset, controllable per command, tool call, or programmatic default -- **Auto-expire** — tasks older than 7 days are removed on load +- **Auto-expire** — tasks idle for more than 7 days are removed on load (active tasks never expire) - **Max 50 concurrent tasks** - **LLM-callable tools** — `loop_schedule`, `loop_status` (session-bound by default) - **Interactive Loop results** — `/loop` results open in a dedicated native dialog instead of writing over the prompt @@ -59,6 +66,14 @@ opencode plugin opencode-plugin-loop --global --force The `--force` flag replaces the installed plugin version and refreshes both global config entries without requiring a version-number change. Restart OpenCode after the command completes. +**Upgrade self-check.** opencode keeps its own plugin package cache at `~/.cache/opencode/packages`, and `--force` does not always refresh it. If an upgrade reports success but behavior does not change (e.g. `npm view opencode-plugin-loop version` disagrees with what you see), clear the cache and restart: + +```bash +rm -rf ~/.cache/opencode/packages/opencode-plugin-loop* +``` + +Then verify with `/loop help` — new flags and subcommands show up there immediately. + ### Option 2: Manual configuration Add the same package name to the `plugin` array in both configuration files. @@ -70,7 +85,7 @@ Server config (`~/.config/opencode/opencode.json`): "plugin": ["opencode-plugin-loop"], "command": { "loop": { - "description": "定时重复执行 prompt。可选间隔: s/m/h/d。子命令: list | status | cancel | pause | resume | stop-all（加 --all 跨 session）", + "description": "Run prompts on a schedule. Intervals: s/m/h/d. Subcommands: help | list | status | cancel | pause | resume | stop-all (add --all to cross sessions)", "template": "$ARGUMENTS", "agent": "build" } @@ -107,6 +122,7 @@ Re-run `npm run build` after editing `src/`, then restart OpenCode to load the r /loop 30s ping the health endpoint /loop 2h look for failing CI runs /loop 2m --jitter=false check the latest package version +/loop 30s --once remind me to stretch # one-shot: fires once, then auto-cancels ``` Fixed tasks use deterministic Jitter by default for backward compatibility. Add @@ -143,19 +159,34 @@ address each one. If everything is green, say so in one line. All subcommands are **session-scoped by default**. Add `--all` to operate across all sessions. ``` +/loop help # full usage, flags, and examples /loop list # show tasks in current session /loop list --all # show all sessions (with [s:xxxx] tags) /loop status # alias for list /loop cancel # cancel one task in current session /loop cancel --all # override scope /loop pause # pause one -/loop resume # resume one (re-arms fixed interval) +/loop resume # resume one (re-arms per mode) /loop stop-all # cancel all tasks in current session /loop stop-all --all # cancel ALL tasks across sessions ``` If you try `cancel ` for a task owned by another session, you'll get a refusal with a hint to add `--all`. The same strict scoping applies to `loop_schedule` and `loop_status` tools. +### Migrating from Claude Code + +| Claude Code `/loop` | opencode-plugin-loop | +|---|---| +| `/loop 5m ` | identical | +| `/loop ` (self-paced) | Adaptive: runs now, model picks the next check (fallback 1m–1h) | +| cancel/list via cron tools | `/loop cancel `, `/loop list` | +| `--cancel`, `--list`, `--stop` | accepted — mapped to `cancel`, `list`, `stop` | +| one-off reminder ("in 30m tell me X") | `/loop 30s --once ` | +| jobs die when the session ends | same default since 0.3.0 (`ephemeralTasks: false` opts out) | +| cron expressions (`*/5 * * * *`) | not supported — use `5m` form (explicit error) | + +Two behavioral differences worth knowing: tasks only fire for the **currently active session** (switch sessions and the others wait; switch back and they catch up once), and fixed tasks fire on a 5-second ticker rather than exact wall-clock cron times (up to one ticker period late). + ### Interactive result dialog Every `/loop` command result opens in a separate native OpenCode dialog. It keeps task output away from the prompt and provides: diff --git a/commands/loop.md b/commands/loop.md index 0da2e56..e63d31a 100644 --- a/commands/loop.md +++ b/commands/loop.md @@ -1,6 +1,6 @@ --- -description: 定时重复执行 prompt。自然语言 Adaptive 请求会立即执行并判断后续调度；显式间隔支持 --jitter=true|false。子命令加 --all 可跨 session。 -argument-hint: "[5m] [--jitter=true|false] [prompt text... | list | cancel | pause | resume | stop-all] [--all]" +description: Run prompts on a schedule. Natural-language Adaptive requests run immediately and the model decides the next check; explicit intervals support --jitter=true|false and --once. Subcommands add --all to cross sessions. See /loop help. +argument-hint: "[5m] [--jitter=true|false] [--once] [prompt text... | help | list | cancel | pause | resume | stop-all] [--all]" agent: build --- diff --git a/examples/loop.md.example b/examples/loop.md.example index fc0abeb..7e6e843 100644 --- a/examples/loop.md.example +++ b/examples/loop.md.example @@ -10,5 +10,5 @@ quiet, say so in one line. After completing the work, call: loop_schedule({ action: "cancel", taskId: "" }) to end the loop, OR - loop_schedule({ action: "reschedule", taskId: "", nextDueAtMs: Date.now() + 5*60*1000 }) + loop_schedule({ action: "reschedule", taskId: "", delayMs: 5*60*1000 }) to continue checking. \ No newline at end of file diff --git a/package.json b/package.json index b5f3293..9e5816e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode-plugin-loop", - "version": "0.3.0", + "version": "0.4.0", "description": "/loop command for opencode — run prompts on a schedule (fixed, adaptive, or maintenance), modeled after Claude Code's /loop", "type": "module", "main": "./dist/index.js", diff --git a/src/index.ts b/src/index.ts index 0fa6af9..19012f3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -25,6 +25,7 @@ import type { Plugin, Hooks, PluginModule } from "@opencode-ai/plugin" import { LoopStore } from "./store.js" +import { InstanceLock } from "./instance-lock.js" import { Scheduler } from "./scheduler.js" import { CronParser } from "./cron-parser.js" import { Jitter } from "./jitter.js" @@ -47,12 +48,13 @@ const DEFAULT_CONFIG: Required = { tickerIntervalMs: 5_000, defaultJitterEnabled: true, ephemeralTasks: true, + instanceLock: true, } function commandAction(args: string): string { const head = args.trim().split(/\s+/, 1)[0]?.toLowerCase() if (!head) return "maintenance" - if (["list", "status", "cancel", "stop", "pause", "resume", "stop-all"].includes(head)) { + if (["list", "status", "cancel", "stop", "pause", "resume", "stop-all", "help"].includes(head)) { return head } return "schedule" @@ -107,9 +109,15 @@ export const LoopPlugin: Plugin = async (ctx) => { // Internal ticker: every 5s, fire any due tasks whose sessionID matches the active session. // This replaces the old session.idle-event-driven firing and runs even when no user input. + // Only the lock leader fires: other instances sharing this tasks.json (B1) + // keep their tickers idle but may take over if the leader goes stale. + const lock = InstanceLock({ storageDir, logger }) + const lockEnabled = config.instanceLock + if (lockEnabled) lock.start() const inflight = new Set() const ticker = setInterval(async () => { try { + if (lockEnabled && !lock.isLeader()) return if (!activeSessionID) return const due = await scheduler.getDueTasksForSession(activeSessionID) if (due.length === 0) return @@ -188,6 +196,7 @@ export const LoopPlugin: Plugin = async (ctx) => { ;(hooks as any)._ticker = ticker hooks.dispose = async () => { clearInterval(ticker) + lock.stop() } return hooks @@ -204,6 +213,7 @@ export default plugin // ---- Public API exports (for users who want to compose) ---- export { LoopStore } from "./store.js" +export { InstanceLock } from "./instance-lock.js" export { Scheduler } from "./scheduler.js" export { CronParser } from "./cron-parser.js" export { Jitter } from "./jitter.js" diff --git a/src/instance-lock.ts b/src/instance-lock.ts new file mode 100644 index 0000000..c5172e8 --- /dev/null +++ b/src/instance-lock.ts @@ -0,0 +1,191 @@ +/** + * InstanceLock: single-leader election between plugin instances that share one + * tasks.json. opencode can load the same plugin through two case-variant paths + * (macOS case-insensitive FS), and every `opencode run --attach` spawns another + * in-process instance — each with its own ticker. Without coordination every + * instance fires the same due task (B1: duplicate fires + lost writes). + * + * Design: + * - The lock is a DIRECTORY ({storageDir}/loop.lock/) so acquisition is an + * atomic mkdirSync. Inside it, lock.json records the owner. + * - Same-process instances share a pid, so ownership is keyed by a random + * instanceId, not by pid. + * - The leader heartbeats by touching lock.json every heartbeatMs. A follower + * takes over only when the lock is stale (no heartbeat for staleMs) and it + * wins an atomic rename race. + * - Followers keep their ticker running but skip firing; commands and tool + * calls still work because every store write goes through merge-write. + * + * Implementation note: factory pattern (no `this` reliance) so opencode's + * plugin loader can call us with or without `new`. + */ + +import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, statSync, utimesSync, writeFileSync } from "node:fs" +import { hostname } from "node:os" +import { join } from "node:path" +import { randomUUID } from "node:crypto" +import { errorMessage, type LoopLogger } from "./runtime-feedback.js" + +export interface InstanceLockOptions { + storageDir: string + /** Injectable for tests; defaults to a random UUID. */ + instanceId?: string + /** Lock without a heartbeat for this long is considered abandoned (default 15_000). */ + staleMs?: number + /** Heartbeat / takeover-probe interval (default 2_500). */ + heartbeatMs?: number + logger?: LoopLogger + /** Injectable clock for tests. */ + now?: () => number +} + +export interface InstanceLockInstance { + instanceId: string + isLeader(): boolean + /** Begin heartbeat / takeover probing. Safe to call once. */ + start(): void + /** Stop probing; release the lock if leader. */ + stop(): void +} + +interface LockFile { + instanceId: string + pid: number + hostname: string + startedAt: number +} + +export function InstanceLock(this: unknown, options: InstanceLockOptions): InstanceLockInstance { + void this + const logger: LoopLogger = options.logger ?? (async () => {}) + const now = options.now ?? Date.now + const instanceId = options.instanceId ?? randomUUID() + const staleMs = options.staleMs ?? 15_000 + const heartbeatMs = options.heartbeatMs ?? 2_500 + const lockDir = join(options.storageDir, "loop.lock") + const lockFile = join(lockDir, "lock.json") + + let leading = false + let timer: ReturnType | null = null + + const writeLockFile = () => { + const body: LockFile = { + instanceId, + pid: process.pid, + hostname: hostname(), + startedAt: now(), + } + writeFileSync(lockFile, JSON.stringify(body, null, 2), "utf-8") + } + + const readLockMtime = (): number | null => { + try { + return statSync(lockFile).mtimeMs + } catch { + return null + } + } + + const acquire = (): boolean => { + try { + mkdirSync(lockDir) + writeLockFile() + return true + } catch { + return false + } + } + const tryTakeover = async (): Promise => { + if (acquire()) { + await logger("info", "loop instance lock acquired", { instanceId }) + return true + } + // Lock held: am I the owner? (e.g. after a same-process reload) + try { + const owner = JSON.parse(readFileSync(lockFile, "utf-8")) as LockFile + if (owner.instanceId === instanceId) return true + } catch { + // Unreadable lock file: fall through to staleness check + } + const mtime = readLockMtime() + const stale = mtime === null || now() - mtime > staleMs + if (!stale) return false + // Abandoned lock: win an atomic rename race before deleting it, so two + // followers cannot both take over. + const graveyard = `${lockDir}.stale.${instanceId}` + try { + renameSync(lockDir, graveyard) + } catch { + return false + } + try { + rmSync(graveyard, { recursive: true, force: true }) + } catch { + // Non-fatal: a stale graveyard directory does not block acquisition. + } + const won = acquire() + if (won) { + await logger("info", "loop instance lock taken over from stale owner", { instanceId }) + } + return won + } + + const tick = async () => { + try { + if (leading) { + // Still mine? A same-process follower may have taken over after + // deciding our heartbeat stopped (e.g. event-loop stall). + try { + const owner = JSON.parse(readFileSync(lockFile, "utf-8")) as LockFile + if (owner.instanceId !== instanceId) { + leading = false + await logger("warn", "loop instance lock lost", { instanceId }) + return + } + } catch { + leading = false + await logger("warn", "loop instance lock lost (unreadable)", { instanceId }) + return + } + try { + const at = new Date(now()) + utimesSync(lockFile, at, at) + } catch (err) { + await logger("warn", "loop lock heartbeat failed", { error: errorMessage(err) }) + } + return + } + leading = await tryTakeover() + } catch (err) { + await logger("warn", "loop instance lock tick failed", { error: errorMessage(err) }) + } + } + + const inst: InstanceLockInstance = { + instanceId, + isLeader: () => leading, + start: () => { + if (timer) return + void tick() + timer = setInterval(() => void tick(), heartbeatMs) + // Never keep the process alive just for the lock. + if (typeof timer.unref === "function") timer.unref() + }, + stop: () => { + if (timer) { + clearInterval(timer) + timer = null + } + if (leading) { + try { + const owner = JSON.parse(readFileSync(lockFile, "utf-8")) as LockFile + if (owner.instanceId === instanceId) rmSync(lockDir, { recursive: true, force: true }) + } catch { + // Lock already gone or unreadable — nothing to release. + } + leading = false + } + }, + } + return inst +} diff --git a/src/scheduler.ts b/src/scheduler.ts index b47b61e..707931b 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -56,7 +56,7 @@ interface SchedulerInstance { getDueTasksForSession(sessionID: string, now?: number): Promise nextDueAt(task: LoopTask, now?: number): Promise executeTask(task: LoopTask, ctx: any, now?: number): Promise - fireTask(task: LoopTask, ctx: any): Promise + fireTask(task: LoopTask, ctx: any): Promise rearmFixed(task: LoopTask, now?: number): Promise rearmAdaptive(task: LoopTask, now?: number): Promise adaptiveNextDueAt(task: LoopTask, now?: number): number @@ -79,6 +79,68 @@ function extractJitterFlag(text: string): { prompt: string; jitterEnabled?: bool } } +/** Strip one layer of matching surrounding quotes (B10). */ +function stripOuterQuotes(text: string): string { + const t = text.trim() + if (t.length >= 2) { + const first = t[0] + const last = t[t.length - 1] + if ((first === '"' && last === '"') || (first === "'" && last === "'")) { + return t.slice(1, -1).trim() + } + } + return t +} + +/** Claude Code-style flags accepted as subcommand aliases (P-1). */ +const CC_FLAG_MAP: Record = { + "--cancel": "cancel", + "--stop": "stop", + "--list": "list", + "--status": "status", + "--pause": "pause", + "--resume": "resume", + "--stop-all": "stop-all", +} + +/** Flags that are meaningful in command position (not errors when leading). */ +const LEADING_OK = new Set(["--all", "--jitter=true", "--jitter=false", "--once"]) + +/** crude cron-expression detector (five-field crontab syntax) (B9). */ +function looksLikeCron(tokens: string[]): boolean { + if (tokens.length < 5) return false + return tokens.slice(0, 5).every((t) => /^[\d*,/\-]+$/.test(t) && /[*,/\-]|\d/.test(t)) +} + +export const LOOP_HELP = `/loop — run prompts on a schedule + +Usage: + /loop Adaptive: runs now, the model picks the next check (fallback 1m–1h) + /loop Fixed interval: 30s, 5m, 2h, 1d (min 1s) + /loop Maintenance mode (uses .opencode/loop.md when present) + /loop help Show this help + +Subcommands (session-scoped; add --all to cross sessions): + list | status [--all] Show loop tasks + cancel [--all] Cancel one task + pause [--all] Pause one task + resume [--all] Resume one task + stop-all [--all] Cancel all tasks + +Flags: + --all Operate across all sessions + --jitter=true|false Force Jitter on/off for a fixed task + --once Fire once, then auto-cancel (fixed tasks only) + +Claude Code-style flags are accepted too: --cancel, --list, --status, +--pause, --resume, --stop, --stop-all map to the matching subcommand. + +Examples: + /loop 5m check the deploy status + /loop 30s --once remind me to stretch + /loop every two minutes check CI + /loop cancel a1b2c3d4` + export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInstance { void this const logger: LoopLogger = opts.logger ?? (async () => {}) @@ -94,14 +156,33 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta async handleUserCommand(args, directory, sessionID) { if (sessionID !== undefined) inst.currentSessionID = sessionID - const trimmed = args.trim() - const tokens = trimmed.split(/\s+/) + const trimmed = stripOuterQuotes(args.trim()) + const tokens = trimmed === "" ? [] : trimmed.split(/\s+/) const allFlag = tokens.includes("--all") - const head = tokens[0]?.toLowerCase() + const onceFlag = tokens.includes("--once") + let head = tokens[0]?.toLowerCase() + + // Claude Code-style leading flags map to subcommands (P-1). + if (head && CC_FLAG_MAP[head]) { + head = CC_FLAG_MAP[head] + tokens[0] = head + } else if (head === "help" || head === "--help" || head === "-h") { + return { message: LOOP_HELP } + } else if (head?.startsWith("--") && !LEADING_OK.has(head)) { + return { + message: `❌ Unknown flag "${tokens[0]}". Run \`/loop help\` to see usage.`, + } + } + + // Leading --all is sugar: `/loop --all list` == `/loop list --all`. + if (head === "--all") { + tokens.shift() + head = tokens[0]?.toLowerCase() + } if (head === "cancel" || head === "stop") { const id = tokens[1] - if (!id) return { message: "❌ 用法: /loop cancel [--all]" } + if (!id) return { message: "❌ Usage: /loop cancel [--all]" } return inst.handleCancel(id, allFlag) } if (head === "list" || head === "status") { @@ -112,12 +193,12 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta } if (head === "pause") { const id = tokens[1] - if (!id) return { message: "❌ 用法: /loop pause [--all]" } + if (!id) return { message: "❌ Usage: /loop pause [--all]" } return inst.handlePause(id, allFlag) } if (head === "resume") { const id = tokens[1] - if (!id) return { message: "❌ 用法: /loop resume [--all]" } + if (!id) return { message: "❌ Usage: /loop resume [--all]" } return inst.handleResume(id, allFlag) } if (head === "stop-all") { @@ -143,34 +224,73 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta source: "default", sessionID, }) + // Run the maintenance prompt immediately in this turn (matching + // Adaptive's run-now behavior), then re-arm on the slow cycle. + await inst.opts.store.markFired(task.id, Date.now() + inst.opts.adaptiveMaxMs) return { task, - message: `🔁 Loop started (maintenance mode): task ${task.id} (session ${sessionID.slice(0, 8)}). Auto re-arms every ${inst.opts.adaptiveMaxMs / 1000}s. Use \`/loop cancel ${task.id}\` to stop.`, + modelPrompt: prompt, + message: `🔁 Loop started (maintenance mode): task ${task.id} (session ${sessionID.slice(0, 8)}). Running now; then re-arms every ${inst.opts.adaptiveMaxMs / 1000}s. Use \`/loop cancel ${task.id}\` to stop.`, } } const { interval, rest } = inst.opts.cron.extractInterval(trimmed) - if (interval && rest.trim()) { + if (interval) { const fixed = extractJitterFlag(rest) - if (!fixed.prompt) return { message: "❌ Empty loop command" } + // Strip command flags that leaked into the prompt area (B2). + fixed.prompt = fixed.prompt + .split(/\s+/) + .filter((t) => t !== "--all" && t !== "--once") + .join(" ") + .trim() + if (!fixed.prompt) { + return { + message: `❌ Missing prompt after interval "${tokens[0]}". Usage: /loop — see \`/loop help\`.`, + } + } const task = await inst.opts.store.create({ prompt: fixed.prompt, mode: "fixed", intervalMs: interval.ms, jitterEnabled: fixed.jitterEnabled ?? inst.opts.defaultJitterEnabled ?? true, + once: onceFlag || undefined, directory, source: "user", sessionID, }) return { task, - message: `🔁 Loop started: every ${interval.display}, prompt "${fixed.prompt.slice(0, 50)}${fixed.prompt.length > 50 ? "..." : ""}" [id=${task.id}] [s=${sessionID.slice(0, 8)}]. Cancel: \`/loop cancel ${task.id}\``, + message: `🔁 Loop started: every ${interval.display}, prompt "${fixed.prompt.slice(0, 50)}${fixed.prompt.length > 50 ? "..." : ""}" [id=${task.id}] [s=${sessionID.slice(0, 8)}]${task.once ? " (runs once)" : ""}. Cancel: \`/loop cancel ${task.id}\``, + } + } + + // Reject inputs that look like scheduling syntax but are not supported, + // instead of silently creating an Adaptive task out of them. + if (looksLikeCron(tokens)) { + return { + message: `❌ Cron expressions are not supported. Use an interval like \`5m\` instead, e.g. \`/loop 5m ${tokens.slice(5).join(" ") || "check the build"}\`.`, + } + } + if (/^\d/.test(tokens[0] ?? "")) { + return { + message: `❌ Invalid interval "${tokens[0]}". Use s/m/h/d (min 1s), e.g. 30s, 5m, 2h, 1d — see \`/loop help\`.`, } } if (trimmed) { + // Command flags are scheduling metadata, not prompt text (B2). + const prompt = tokens + .filter((t) => t !== "--all" && t !== "--jitter=true" && t !== "--jitter=false" && t !== "--once") + .join(" ") + .trim() + if (!prompt) { + return { message: "❌ Empty loop command — see `/loop help`." } + } + if (onceFlag) { + return { message: "❌ --once is only supported for fixed-interval tasks, e.g. `/loop 30s --once `." } + } const task = await inst.opts.store.create({ - prompt: trimmed, + prompt, mode: "adaptive", adaptiveMinMs: inst.opts.adaptiveMinMs, adaptiveMaxMs: inst.opts.adaptiveMaxMs, @@ -186,11 +306,11 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta minMs: inst.opts.adaptiveMinMs, maxMs: inst.opts.adaptiveMaxMs, }), - message: `🔁 Loop started (adaptive ${inst.opts.adaptiveMinMs / 1000}s–${inst.opts.adaptiveMaxMs / 1000}s): "${trimmed.slice(0, 50)}${trimmed.length > 50 ? "..." : ""}" [id=${task.id}] [s=${sessionID.slice(0, 8)}]. Cancel: \`/loop cancel ${task.id}\``, + message: `🔁 Loop started (adaptive ${inst.opts.adaptiveMinMs / 1000}s–${inst.opts.adaptiveMaxMs / 1000}s): "${prompt.slice(0, 50)}${prompt.length > 50 ? "..." : ""}" [id=${task.id}] [s=${sessionID.slice(0, 8)}]. Cancel: \`/loop cancel ${task.id}\``, } } - return { message: "❌ Empty loop command" } + return { message: "❌ Empty loop command — see `/loop help`." } }, handleCancel(id, allFlag) { @@ -228,8 +348,16 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta } } const r = await inst.opts.store.setPaused(id, false) - if (r && r.mode === "fixed" && r.intervalMs) { - await inst.rearmFixed(r) + if (r) { + // Re-arm per mode (B6): without this, adaptive/maintenance tasks keep + // a stale nextDueAt and catch-up fire immediately on resume. + if (r.mode === "fixed" && r.intervalMs) { + await inst.rearmFixed(r) + } else if (r.mode === "adaptive") { + await inst.rearmAdaptive(r) + } else if (r.mode === "maintenance" && r.adaptiveMaxMs) { + await inst.opts.store.reschedule(r.id, Date.now() + r.adaptiveMaxMs) + } } return { message: r ? `▶ Resumed ${id}` : `❌ No task ${id}` } }, @@ -246,10 +374,11 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta ? `adaptive ${(t.adaptiveMinMs ?? 0) / 1000}s–${(t.adaptiveMaxMs ?? 0) / 1000}s` : `maintenance ${(t.adaptiveMaxMs ?? 0) / 1000}s` const status = t.paused ? "⏸ paused" : "▶ active" + const onceTag = t.once ? " • once" : "" const sessionTag = showSession && t.sessionID ? ` [s:${t.sessionID.slice(0, 8)}]` : "" const preview = t.prompt.length > 60 ? t.prompt.slice(0, 60) + "..." : t.prompt - lines.push(` [${t.id}]${sessionTag} ${status} • ${interval} • ${preview}`) + lines.push(` [${t.id}]${sessionTag} ${status} • ${interval}${onceTag} • ${preview}`) } lines.push( `Manage: \`/loop cancel|pause|resume \` (add \`--all\` to cross sessions) or \`/loop stop-all\`` @@ -313,13 +442,21 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta return } - await inst.fireTask(task, ctx) - const next = await inst.nextDueAt(task, now ?? Date.now()) + // Wall-clock scheduling: the next cycle is anchored to when this fire + // STARTED, so long-running model turns do not inflate the interval. + const fireStartedAt = now ?? Date.now() + const fired = await inst.fireTask(task, ctx) + // One-shot tasks end after their first successful fire (P-4). + if (task.once && fired) { + await inst.opts.store.cancel(task.id) + return + } + const next = await inst.nextDueAt(task, fireStartedAt) await inst.opts.store.markFired(task.id, next) }, async fireTask(task, ctx) { - if (inst.inflight.has(task.id)) return + if (inst.inflight.has(task.id)) return false inst.inflight.add(task.id) try { const sessionID = task.sessionID @@ -336,12 +473,12 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta if (!sessionID) { await logger("warn", "task has no sessionID; skipping", { taskId: task.id }) await inst.opts.store.logFire(task, false) - return + return false } if (!client?.session?.prompt) { await logger("warn", "client.session.prompt not available", { taskId: task.id }) await inst.opts.store.logFire(task, false) - return + return false } try { await client.session.prompt({ @@ -359,12 +496,14 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta query: { directory }, }) await inst.opts.store.logFire(task, true) + return true } catch (err) { await inst.opts.store.logFire(task, false) await logger("error", "failed to fire task", { taskId: task.id, error: errorMessage(err), }) + return false } } finally { inst.inflight.delete(task.id) @@ -416,4 +555,4 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta */ export const DEFAULT_MAINTENANCE_PROMPT = `Continue any unfinished work from this conversation. Tend to the current branch's pull request: review comments, failed CI runs, merge conflicts. Run cleanup passes such as bug hunts or simplification when nothing else is pending. -Do not start new initiatives outside the above scope. Irreversible actions such as pushing or deleting only proceed when they continue something the transcript already authorized. After completing the work, call loop_schedule(action="cancel", taskId="") to end the loop, or call loop_schedule(action="reschedule", taskId="", nextDueAtMs=) to continue.` +Do not start new initiatives outside the above scope. Irreversible actions such as pushing or deleting only proceed when they continue something the transcript already authorized. After completing the work, call loop_schedule(action="cancel", taskId="") to end the loop, or call loop_schedule(action="reschedule", taskId="", delayMs=) to continue.` diff --git a/src/store.ts b/src/store.ts index 6f555fd..479758d 100644 --- a/src/store.ts +++ b/src/store.ts @@ -13,7 +13,7 @@ * does NOT depend on `this`: every method reads from a closed-over `inst`. */ -import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs" +import { appendFileSync, existsSync, mkdirSync, readFileSync, renameSync, statSync, writeFileSync } from "node:fs" import { join, dirname } from "node:path" import type { LoopTask, CreateTaskInput } from "./types.js" import { errorMessage, type LoopLogger } from "./runtime-feedback.js" @@ -99,6 +99,24 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI state.pid === identity.pid && state.startedAt !== undefined && Math.abs(state.startedAt - identity.startedAt) <= PID_START_TOLERANCE_MS + // Merge-write bookkeeping. Multiple plugin instances can share one + // tasks.json (case-variant plugin paths, per-command `opencode run` + // instances). `tombstones` are ids this instance cancelled — they must never + // be resurrected from another instance's stale write. `dirtyIds` are ids + // this instance touched — its version wins over the disk copy on merge. + const tombstones = new Set() + const dirtyIds = new Set() + const readDisk = (): PersistedState | null => { + try { + if (!existsSync(inst.filePath)) return null + const parsed = JSON.parse(readFileSync(inst.filePath, "utf-8")) as PersistedState + if (parsed?.version !== 1 || !Array.isArray(parsed.tasks)) return null + return parsed + } catch { + return null + } + } + const HISTORY_MAX_BYTES = 1_048_576 const inst: LoopStoreInstance = { state: { version: 1, tasks: [] }, filePath: "", @@ -117,6 +135,8 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI } if (ephemeralTasks && !isSameProcess(parsed)) { const dropped = parsed.tasks.length + // Tombstone every id so merge-write cannot resurrect them. + for (const t of parsed.tasks) tombstones.add(t.id) inst.state = { version: 1, tasks: [] } await inst.persist() if (dropped > 0) { @@ -128,8 +148,16 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI return } const cutoff = Date.now() - inst.taskTtlMs - const filtered = parsed.tasks.filter((t) => t.createdAt > cutoff && !!t.sessionID) + // B4: expire by last ACTIVITY, not creation — a task that keeps + // firing must not be dropped just because it was created 7 days ago. + const filtered = parsed.tasks.filter((t) => Math.max(t.createdAt, t.lastFiredAt ?? 0) > cutoff && !!t.sessionID && !tombstones.has(t.id)) + // Tombstone load-time deletions (expired/orphan) so merge-write + // cannot resurrect them on the persist below. + for (const t of parsed.tasks) { + if (!filtered.includes(t)) tombstones.add(t.id) + } inst.state = { version: 1, tasks: filtered } + dirtyIds.clear() if (inst.state.tasks.length !== parsed.tasks.length) { await inst.persist() const dropped = parsed.tasks.length - inst.state.tasks.length @@ -152,12 +180,33 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI } }, persist: async () => { + // Merge with the on-disk state instead of blindly overwriting it, so + // concurrent instances do not lose each other's tasks (B1). + const disk = readDisk() + if (disk) { + const diskIds = new Set(disk.tasks.map((t) => t.id)) + const byId = new Map() + for (const dt of disk.tasks) { + if (tombstones.has(dt.id)) continue + byId.set(dt.id, dt) + } + for (const t of inst.state.tasks) { + if (!dirtyIds.has(t.id) && !diskIds.has(t.id)) { + // Vanished from disk and untouched by us: another instance + // cancelled it — accept the deletion. + continue + } + byId.set(t.id, t) + } + inst.state.tasks = Array.from(byId.values()) + } inst.state.pid = identity.pid inst.state.startedAt = identity.startedAt const tmp = `${inst.filePath}.tmp` mkdirSync(dirname(tmp), { recursive: true }) writeFileSync(tmp, JSON.stringify(inst.state, null, 2), "utf-8") renameSync(tmp, inst.filePath) + dirtyIds.clear() }, generateId: () => { const alphabet = "abcdefghijklmnopqrstuvwxyz0123456789" @@ -193,8 +242,12 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI directory: input.directory, sessionID: input.sessionID, paused: false, + // Only present on one-shot tasks — keeps persisted JSON stable for + // tasks that never use the field. + ...(input.once ? { once: true as const } : {}), } inst.state.tasks.push(task) + dirtyIds.add(task.id) await inst.persist() return task }, @@ -202,6 +255,8 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI const idx = inst.state.tasks.findIndex((t) => t.id === id) if (idx < 0) return null const [removed] = inst.state.tasks.splice(idx, 1) + tombstones.add(id) + dirtyIds.delete(id) await inst.persist() return removed }, @@ -211,18 +266,28 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI return inst.cancel(task.id) }, cancelAll: async () => { + for (const t of inst.state.tasks) tombstones.add(t.id) + // Also tombstone ids only known to the disk copy (created by other + // instances) so stop-all --all really empties the shared file. + const disk = readDisk() + if (disk) for (const t of disk.tasks) tombstones.add(t.id) const n = inst.state.tasks.length inst.state.tasks = [] + dirtyIds.clear() await inst.persist() return n }, cancelBySession: async (sessionID) => { if (!sessionID) return 0 - const before = inst.state.tasks.length + const removed = inst.state.tasks.filter((t) => t.sessionID === sessionID) + if (removed.length === 0) return 0 + for (const t of removed) { + tombstones.add(t.id) + dirtyIds.delete(t.id) + } inst.state.tasks = inst.state.tasks.filter((t) => t.sessionID !== sessionID) - const removed = before - inst.state.tasks.length - if (removed > 0) await inst.persist() - return removed + await inst.persist() + return removed.length }, list: () => [...inst.state.tasks], listBySession: (sessionID) => { @@ -251,6 +316,7 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI } else if (task.mode === "maintenance" && task.adaptiveMaxMs) { task.nextDueAt = task.lastFiredAt + task.adaptiveMaxMs } + dirtyIds.add(id) await inst.persist() return task }, @@ -258,6 +324,7 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI const task = inst.get(id) if (!task) return null task.nextDueAt = nextDueAt + dirtyIds.add(id) await inst.persist() return task }, @@ -271,6 +338,7 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI delete task.adaptiveMaxMs task.lastFiredAt = now task.nextDueAt = now + intervalMs + dirtyIds.add(id) await inst.persist() return task }, @@ -278,23 +346,29 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI const task = inst.get(id) if (!task) return null task.paused = paused + dirtyIds.add(id) await inst.persist() return task }, logFire: async (task, success) => { const logFile = join(dirname(inst.filePath), "history.log") - const line = JSON.stringify({ - ts: Date.now(), - taskId: task.id, - mode: task.mode, - sessionID: task.sessionID, - prompt: task.prompt.slice(0, 200), - success, - }) + const line = + JSON.stringify({ + ts: Date.now(), + taskId: task.id, + mode: task.mode, + sessionID: task.sessionID, + prompt: task.prompt.slice(0, 200), + success, + }) + "\n" try { mkdirSync(dirname(logFile), { recursive: true }) - const existing = existsSync(logFile) ? readFileSync(logFile, "utf-8") : "" - writeFileSync(logFile, existing + line + "\n", "utf-8") + // O(1) append instead of read-rewrite (B3). Rotate BEFORE appending + // when the log is oversize, keeping a single backup. + if (existsSync(logFile) && statSync(logFile).size > HISTORY_MAX_BYTES) { + renameSync(logFile, join(dirname(inst.filePath), "history.1.log")) + } + appendFileSync(logFile, line, "utf-8") } catch (err) { await logger("warn", "failed to write fire history", { error: errorMessage(err), diff --git a/src/tools/loop-tools.ts b/src/tools/loop-tools.ts index ef0cde6..7259dae 100644 --- a/src/tools/loop-tools.ts +++ b/src/tools/loop-tools.ts @@ -32,6 +32,7 @@ export async function buildLoopTools( delayMs: z.number().finite().optional().describe("Relative delay in milliseconds (preferred for Adaptive reschedule)"), nextDueAtMs: z.number().finite().optional().describe("Absolute epoch ms for reschedule; cannot be combined with delayMs"), jitterEnabled: z.boolean().optional().describe("Fixed-task Jitter policy for create or set_fixed"), + once: z.boolean().optional().describe("One-shot task: auto-cancel after the first successful fire (fixed mode only)"), mode: z .enum(["fixed", "adaptive", "maintenance"]) .optional() @@ -121,8 +122,15 @@ export async function buildLoopTools( }) } const r = await store.setPaused(args.taskId, false) - if (r && r.mode === "fixed" && r.intervalMs) { - await scheduler.rearmFixed(r) + if (r) { + // Re-arm per mode (B6), same as /loop resume. + if (r.mode === "fixed" && r.intervalMs) { + await scheduler.rearmFixed(r) + } else if (r.mode === "adaptive") { + await scheduler.rearmAdaptive(r) + } else if (r.mode === "maintenance" && r.adaptiveMaxMs) { + await store.reschedule(r.id, Date.now() + r.adaptiveMaxMs) + } } return JSON.stringify({ ok: !!r, task: r }, null, 2) } @@ -134,9 +142,13 @@ export async function buildLoopTools( if (!sid) return JSON.stringify({ ok: false, error: "No sessionID in context" }) const mode = args.mode ?? (args.intervalMs ? "fixed" : "adaptive") + if (args.once && mode !== "fixed") { + return JSON.stringify({ ok: false, error: "once is supported only for fixed tasks" }) + } const input: any = { prompt: args.prompt, mode, + once: args.once || undefined, directory, source: "user", sessionID: sid, @@ -147,8 +159,9 @@ export async function buildLoopTools( args.jitterEnabled ?? scheduler.opts.defaultJitterEnabled ?? true } if (mode === "adaptive") { - input.adaptiveMinMs = 60_000 - input.adaptiveMaxMs = 3_600_000 + // B5: honor the configured adaptive bounds instead of hardcoding. + input.adaptiveMinMs = scheduler.opts.adaptiveMinMs + input.adaptiveMaxMs = scheduler.opts.adaptiveMaxMs } const task = await store.create(input) if (task.mode === "adaptive") await scheduler.rearmAdaptive(task) @@ -246,11 +259,16 @@ export async function buildLoopTools( error: "set_fixed requires an Adaptive task", }) } + // B7: apply the configured jitter policy to the FIRST cycle too, + // so conversion and later re-arms behave identically. + const conversionTime = Date.now() const r = await store.setFixed( args.taskId, args.intervalMs as number, - args.jitterEnabled ?? false + args.jitterEnabled ?? false, + conversionTime ) + if (r) await scheduler.rearmFixed(r, conversionTime) return JSON.stringify( { ok: !!r, diff --git a/src/types.ts b/src/types.ts index 86173d0..aa465e9 100644 --- a/src/types.ts +++ b/src/types.ts @@ -32,6 +32,8 @@ export interface LoopTask { sessionID: string /** Disabled? */ paused: boolean + /** One-shot task: auto-cancelled after the first successful fire (fixed mode only). */ + once?: boolean } export interface LoopConfig { @@ -57,6 +59,13 @@ export interface LoopConfig { * keep the legacy behavior of persisting tasks across process restarts. */ ephemeralTasks?: boolean + /** + * Single-leader instance lock (default true): when several plugin instances + * share one tasks.json (case-variant plugin paths, per-command `opencode + * run` instances), only the leader's ticker fires tasks. Set to false to + * disable coordination (not recommended). + */ + instanceLock?: boolean } export interface CreateTaskInput { @@ -70,6 +79,8 @@ export interface CreateTaskInput { directory: string /** Required: the session this task is bound to */ sessionID: string + /** One-shot task (fixed mode only): auto-cancel after the first successful fire. */ + once?: boolean } export interface FireResult { diff --git a/tests/instance-lock.test.mjs b/tests/instance-lock.test.mjs new file mode 100644 index 0000000..8f458db --- /dev/null +++ b/tests/instance-lock.test.mjs @@ -0,0 +1,160 @@ +import { test } from "node:test" +import assert from "node:assert/strict" +import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from "node:fs" +import { join } from "node:path" +import { tmpdir } from "node:os" +import { InstanceLock } from "../dist/instance-lock.js" + +const tick = (ms = 10) => new Promise((r) => setTimeout(r, ms)) + +function makeLock(dir, instanceId, extra = {}) { + return new InstanceLock({ + storageDir: dir, + instanceId, + staleMs: 60, + heartbeatMs: 20, + ...extra, + }) +} + +test("first instance becomes the only leader", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-lock-")) + try { + const a = makeLock(dir, "aaa") + const b = makeLock(dir, "bbb") + a.start() + b.start() + await tick(60) + assert.equal(a.isLeader(), true) + assert.equal(b.isLeader(), false) + a.stop() + b.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test("follower takes over after the leader stops heartbeating", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-lock-")) + try { + const a = makeLock(dir, "aaa") + a.start() + await tick(40) + assert.equal(a.isLeader(), true) + // Simulate a crashed leader: stop the heartbeat but leave the lock behind. + a.stop() + mkdirSync(join(dir, "loop.lock"), { recursive: true }) + writeFileSync( + join(dir, "loop.lock", "lock.json"), + JSON.stringify({ instanceId: "dead", pid: 1, hostname: "x", startedAt: Date.now() - 10_000 }), + "utf-8" + ) + // Backdate the heartbeat so the lock is stale. + const past = new Date(Date.now() - 10_000) + const { utimesSync } = await import("node:fs") + utimesSync(join(dir, "loop.lock", "lock.json"), past, past) + + const b = makeLock(dir, "bbb") + b.start() + await tick(80) + assert.equal(b.isLeader(), true) + b.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test("fresh lock is not taken over before the stale window", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-lock-")) + try { + const a = makeLock(dir, "aaa") + a.start() + await tick(40) + const b = makeLock(dir, "bbb") + b.start() + await tick(50) + assert.equal(a.isLeader(), true, "leader keeps heartbeating") + assert.equal(b.isLeader(), false) + a.stop() + b.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test("concurrent takeover race: exactly one follower wins", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-lock-")) + try { + // Plant a stale lock from a dead owner. + mkdirSync(join(dir, "loop.lock"), { recursive: true }) + writeFileSync( + join(dir, "loop.lock", "lock.json"), + JSON.stringify({ instanceId: "dead", pid: 1, hostname: "x", startedAt: Date.now() - 10_000 }), + "utf-8" + ) + const past = new Date(Date.now() - 10_000) + const { utimesSync } = await import("node:fs") + utimesSync(join(dir, "loop.lock", "lock.json"), past, past) + + const b = makeLock(dir, "bbb") + const c = makeLock(dir, "ccc") + b.start() + c.start() + await tick(120) + const leaders = [b.isLeader(), c.isLeader()].filter(Boolean) + assert.equal(leaders.length, 1, `exactly one leader, got ${leaders.length}`) + b.stop() + c.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test("stop releases the lock so the next instance acquires immediately", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-lock-")) + try { + const a = makeLock(dir, "aaa") + a.start() + await tick(40) + assert.equal(a.isLeader(), true) + a.stop() + const b = makeLock(dir, "bbb") + b.start() + await tick(40) + assert.equal(b.isLeader(), true) + b.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test("crashed leader (no stop) leaves a lock that is eventually taken over", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-lock-")) + try { + const a = makeLock(dir, "aaa") + a.start() + await tick(40) + assert.equal(a.isLeader(), true) + // Crash: do NOT call a.stop(). The lock file stays, heartbeats from a keep + // it fresh though — so stop a's timer without releasing (real crash). + a.stop() + // Replant an abandoned lock older than staleMs. + mkdirSync(join(dir, "loop.lock"), { recursive: true }) + writeFileSync( + join(dir, "loop.lock", "lock.json"), + JSON.stringify({ instanceId: "ghost", pid: 99999, hostname: "x", startedAt: Date.now() - 60_000 }), + "utf-8" + ) + const past = new Date(Date.now() - 60_000) + const { utimesSync } = await import("node:fs") + utimesSync(join(dir, "loop.lock", "lock.json"), past, past) + + const b = makeLock(dir, "bbb") + b.start() + await tick(100) + assert.equal(b.isLeader(), true) + b.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) diff --git a/tests/integration.test.mjs b/tests/integration.test.mjs index 5b0ec36..779dad5 100644 --- a/tests/integration.test.mjs +++ b/tests/integration.test.mjs @@ -355,11 +355,14 @@ test("TUI-safe /loop list uses toast and consumes the model-facing command", asy assert.match(toastCalls[0].body.message, /loop task/i) assert.doesNotMatch(output.parts[0].text, /^list$/) assert.match(output.parts[0].text, /already handled/i) - assert.equal(logCalls.length, 1) - assert.equal(logCalls[0].throwOnError, true) - assert.equal(logCalls[0].body.extra.action, "list") - assert.equal(logCalls[0].body.extra.argumentLength, 4) - assert.equal(logCalls[0].body.extra.command, undefined) + // The instance lock also logs ("loop instance lock acquired") — select + // the command log entry. + const commandLogs = logCalls.filter((c) => c.body?.extra?.action !== undefined) + assert.equal(commandLogs.length, 1) + assert.equal(commandLogs[0].throwOnError, true) + assert.equal(commandLogs[0].body.extra.action, "list") + assert.equal(commandLogs[0].body.extra.argumentLength, 4) + assert.equal(commandLogs[0].body.extra.command, undefined) } finally { if (hooks) await hooks.dispose() console.log = originalConsole.log @@ -520,10 +523,11 @@ test("toast transport failure is recorded in structured logs", async () => { { parts: [] } ) - assert.equal(logCalls.length, 2) - assert.equal(logCalls[1].body.level, "warn") - assert.equal(logCalls[1].body.message, "failed to show loop command result") - assert.equal(logCalls[1].body.extra.error, "toast unavailable") + // Lock acquisition adds a log entry; find the toast-failure warn. + const warn = logCalls.find((c) => c.body?.message === "failed to show loop command result") + assert.ok(warn, "toast failure recorded") + assert.equal(warn.body.level, "warn") + assert.equal(warn.body.extra.error, "toast unavailable") } finally { if (hooks) await hooks.dispose() rmSync(dir, { recursive: true }) diff --git a/tests/package-exports.test.mjs b/tests/package-exports.test.mjs index 19f3832..522457d 100644 --- a/tests/package-exports.test.mjs +++ b/tests/package-exports.test.mjs @@ -13,8 +13,8 @@ const builtDialogView = await readFile( "utf8", ) -test("publishes the ephemeral lifecycle release", () => { - assert.equal(packageJson.version, "0.3.0") +test("publishes the 0.4.0 fixes release", () => { + assert.equal(packageJson.version, "0.4.0") }) test("publishes explicit server and TUI plugin entrypoints", () => { diff --git a/tests/scheduler.test.mjs b/tests/scheduler.test.mjs index 797554f..10d17e8 100644 --- a/tests/scheduler.test.mjs +++ b/tests/scheduler.test.mjs @@ -392,3 +392,222 @@ test("fireTask failure uses structured logger without console output", async () rmSync(dir, { recursive: true }) } }) + +// --- c2: parsing hygiene + help --- + +test("/loop help shows usage with all flags", async () => { + const { sched, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("help", "/tmp", "s1") + assert.equal(r.task, undefined) + assert.match(r.message, /--all/) + assert.match(r.message, /--jitter=true\|false/) + assert.match(r.message, /--once/) + assert.match(r.message, /--cancel, --list/) + assert.match(r.message, /cancel /) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("/loop 5m (interval, no prompt) errors instead of creating a task", async () => { + const { sched, store, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("5m", "/tmp", "s1") + assert.equal(r.task, undefined) + assert.match(r.message, /Missing prompt/) + assert.equal(store.list().length, 0) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("/loop 0s x and /loop 999x x are rejected as invalid intervals", async () => { + const { sched, store, dir } = makeScheduler() + try { + for (const args of ["0s test", "999x test", "5 test"]) { + const r = await sched.handleUserCommand(args, "/tmp", "s1") + assert.equal(r.task, undefined, args) + assert.match(r.message, /Invalid interval/, args) + } + assert.equal(store.list().length, 0) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("cron-shaped input is rejected with guidance", async () => { + const { sched, store, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("*/5 * * * * check build", "/tmp", "s1") + assert.equal(r.task, undefined) + assert.match(r.message, /Cron expressions are not supported/) + assert.equal(store.list().length, 0) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("--all is stripped from fixed and adaptive prompts (B2)", async () => { + const { sched, dir } = makeScheduler() + try { + const r1 = await sched.handleUserCommand("5m --all check deploy", "/tmp", "s1") + assert.equal(r1.task.prompt, "check deploy") + const r2 = await sched.handleUserCommand("--all check the weather", "/tmp", "s1") + assert.equal(r2.task.prompt, "check the weather") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("surrounding quotes are stripped before parsing (B10)", async () => { + const { sched, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand('"5m check deploy"', "/tmp", "s1") + assert.equal(r.task.mode, "fixed") + assert.equal(r.task.prompt, "check deploy") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("Claude Code-style flags map to subcommands (P-1)", async () => { + const { sched, store, dir } = makeScheduler() + try { + const created = await sched.handleUserCommand("5m check deploy", "/tmp", "s1") + const id = created.task.id + + const listed = await sched.handleUserCommand("--list", "/tmp", "s1") + assert.match(listed.message, /loop task/) + + const cancelled = await sched.handleUserCommand(`--cancel ${id}`, "/tmp", "s1") + assert.match(cancelled.message, /Cancelled/) + assert.equal(store.get(id), null) + + const bad = await sched.handleUserCommand("--bogus do something", "/tmp", "s1") + assert.equal(bad.task, undefined) + assert.match(bad.message, /Unknown flag/) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("usage errors are in English", async () => { + const { sched, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("cancel", "/tmp", "s1") + assert.match(r.message, /Usage: \/loop cancel /) + assert.doesNotMatch(r.message, /用法/) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +// --- c3: scheduling semantics --- + +test("bare /loop maintenance runs immediately and re-arms on the slow cycle", async () => { + const { sched, store, dir } = makeScheduler() + try { + const before = Date.now() + const r = await sched.handleUserCommand("", "/tmp", "s1") + assert.equal(r.task.mode, "maintenance") + assert.ok(r.modelPrompt, "maintenance prompt runs in the current turn") + assert.match(r.message, /Running now/) + const t = store.get(r.task.id) + assert.ok(t.lastFiredAt >= before, "marked as fired") + assert.ok(t.nextDueAt >= before + 3_500_000, "next run ~1h out") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("resume re-arms adaptive and maintenance tasks instead of catch-up firing (B6)", async () => { + const { sched, store, dir } = makeScheduler() + try { + const a = await sched.handleUserCommand("check things", "/tmp", "s1") + await store.reschedule(a.task.id, Date.now() - 60_000) + await sched.handleUserCommand(`pause ${a.task.id}`, "/tmp", "s1") + const rr = await sched.handleUserCommand(`resume ${a.task.id}`, "/tmp", "s1") + assert.match(rr.message, /Resumed/) + const after = store.get(a.task.id) + assert.ok(after.nextDueAt > Date.now(), "adaptive re-armed into the future") + + const m = await sched.handleUserCommand("", "/tmp", "s1") + await store.reschedule(m.task.id, Date.now() - 60_000) + await sched.handleUserCommand(`pause ${m.task.id}`, "/tmp", "s1") + await sched.handleUserCommand(`resume ${m.task.id}`, "/tmp", "s1") + const mAfter = store.get(m.task.id) + assert.ok(mAfter.nextDueAt > Date.now() + 3_500_000, "maintenance re-armed ~1h out") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("fixed tasks re-arm from fire start, not fire completion (no drift)", async () => { + const { sched, store, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("60s --jitter=false say hi", "/tmp", "s1") + const t0 = 1_800_000_000_000 + const mockCtx = { client: { session: { async prompt() { return { info: {}, parts: [] } } } } } + await sched.executeTask(store.get(r.task.id), mockCtx, t0) + const after = store.get(r.task.id) + assert.equal(after.nextDueAt, t0 + 60_000, "next fire anchored to fire start") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +// --- c4: --once one-shot tasks --- + +test("/loop 30s --once fires exactly once then auto-cancels", async () => { + const { sched, store, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("30s --once say hi", "/tmp", "s1") + assert.equal(r.task.mode, "fixed") + assert.equal(r.task.once, true) + assert.equal(r.task.prompt, "say hi") + assert.match(r.message, /runs once/) + + const t0 = 1_800_000_000_000 + const mockCtx = { client: { session: { async prompt() { return { info: {}, parts: [] } } } } } + await sched.executeTask(store.get(r.task.id), mockCtx, t0) + assert.equal(store.get(r.task.id), null, "task auto-cancelled after first fire") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("--once task survives a failed fire (retry next cycle)", async () => { + const { sched, store, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("30s --once say hi", "/tmp", "s1") + const failingCtx = { client: { session: { async prompt() { throw new Error("boom") } } } } + await sched.executeTask(store.get(r.task.id), failingCtx, 1_800_000_000_000) + assert.ok(store.get(r.task.id), "task kept for retry") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("--once without an interval is rejected", async () => { + const { sched, store, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("--once check things", "/tmp", "s1") + assert.equal(r.task, undefined) + assert.match(r.message, /--once is only supported for fixed/) + assert.equal(store.list().length, 0) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("list marks one-shot tasks", async () => { + const { sched, dir } = makeScheduler() + try { + await sched.handleUserCommand("30s --once say hi", "/tmp", "s1") + const r = await sched.handleUserCommand("list", "/tmp", "s1") + assert.match(r.message, /once/) + } finally { + rmSync(dir, { recursive: true }) + } +}) diff --git a/tests/store.test.mjs b/tests/store.test.mjs index 4b7a23d..35af7d7 100644 --- a/tests/store.test.mjs +++ b/tests/store.test.mjs @@ -1,6 +1,6 @@ import { test } from "node:test" import assert from "node:assert/strict" -import { mkdtempSync, rmSync, writeFileSync } from "node:fs" +import { mkdtempSync, rmSync, writeFileSync, readFileSync } from "node:fs" import { join } from "node:path" import { tmpdir } from "node:os" import { LoopStore } from "../dist/store.js" @@ -620,3 +620,162 @@ test("ephemeral: empty legacy state adopts identity without a cleanup log", asyn rmSync(dir, { recursive: true }) } }) + +// --- merge-write (B1: concurrent instances sharing one tasks.json) --- + +test("merge-write: concurrent creates from two instances keep both tasks", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const id = { pid: 1, startedAt: Date.now() } + const s1 = new LoopStore({ storageDir: dir, processIdentity: id }) + const s2 = new LoopStore({ storageDir: dir, processIdentity: id }) + await s1.load() + await s2.load() + await s1.create({ prompt: "from s1", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + // s2 has stale in-memory state (does not know about s1's task) + await s2.create({ prompt: "from s2", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sB" }) + + const s3 = new LoopStore({ storageDir: dir, processIdentity: id }) + await s3.load() + const prompts = s3.list().map((t) => t.prompt).sort() + assert.deepEqual(prompts, ["from s1", "from s2"], "no task lost to last-writer-wins") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("merge-write: my cancel is not resurrected by a stale disk write", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const id = { pid: 1, startedAt: Date.now() } + const s1 = new LoopStore({ storageDir: dir, processIdentity: id }) + await s1.load() + const t = await s1.create({ prompt: "doomed", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + await s1.cancel(t.id) + + // A stale peer rewrites the cancelled task back to disk. + const disk = JSON.parse(readFileSync(join(dir, "tasks.json"), "utf-8")) + disk.tasks.push(t) + writeFileSync(join(dir, "tasks.json"), JSON.stringify(disk), "utf-8") + + // s1's next write must keep the tombstone: the task stays gone. + await s1.create({ prompt: "new", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + const s2 = new LoopStore({ storageDir: dir, processIdentity: id }) + await s2.load() + assert.deepEqual(s2.list().map((x) => x.prompt), ["new"]) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("merge-write: tasks cancelled by another instance are accepted on next persist", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const id = { pid: 1, startedAt: Date.now() } + const s1 = new LoopStore({ storageDir: dir, processIdentity: id }) + const s2 = new LoopStore({ storageDir: dir, processIdentity: id }) + await s1.load() + const t = await s1.create({ prompt: "shared", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + await s2.load() + await s2.cancel(t.id) + // s1 still holds the task in memory and touches an unrelated field via create. + await s1.create({ prompt: "other", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + assert.deepEqual(s1.list().map((x) => x.prompt), ["other"], "peer deletion accepted after merge") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("cancelAll also tombstones ids only present on disk", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const id = { pid: 1, startedAt: Date.now() } + const s1 = new LoopStore({ storageDir: dir, processIdentity: id }) + const s2 = new LoopStore({ storageDir: dir, processIdentity: id }) + await s1.load() + await s1.create({ prompt: "mine", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + await s2.load() + await s2.create({ prompt: "peer", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sB" }) + // s1's memory only knows "mine"; stop-all must also kill s2's "peer". + await s1.cancelAll() + const s3 = new LoopStore({ storageDir: dir, processIdentity: id }) + await s3.load() + assert.equal(s3.list().length, 0) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +// --- history.log append + rotation (B3) --- + +test("logFire appends without rewriting and rotates at 1MB", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const id = { pid: 1, startedAt: Date.now() } + const store = new LoopStore({ storageDir: dir, processIdentity: id }) + await store.load() + const task = await store.create({ prompt: "x", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "s1" }) + await store.logFire(task, true) + await store.logFire(task, false) + const lines = readFileSync(join(dir, "history.log"), "utf-8").trim().split("\n") + assert.equal(lines.length, 2) + assert.equal(JSON.parse(lines[0]).success, true) + assert.equal(JSON.parse(lines[1]).success, false) + + // Force rotation: pre-fill beyond 1MB, then one more fire. + writeFileSync(join(dir, "history.log"), "x".repeat(1_048_577), "utf-8") + await store.logFire(task, true) + const { existsSync } = await import("node:fs") + assert.equal(existsSync(join(dir, "history.1.log")), true, "rotated backup exists") + const fresh = readFileSync(join(dir, "history.log"), "utf-8").trim().split("\n") + assert.equal(fresh.length, 1, "fresh log starts over") + assert.equal(JSON.parse(fresh[0]).success, true) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("TTL uses last activity: old-but-active tasks survive (B4)", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const now = Date.now() + const data = { + version: 1, + ...currentProcess(), + tasks: [ + { + id: "active-old", + prompt: "created 8d ago, fired recently", + mode: "fixed", + intervalMs: 60_000, + createdAt: now - 8 * 86_400_000, + lastFiredAt: now - 60_000, + nextDueAt: now + 60_000, + source: "user", + directory: "/tmp", + sessionID: "s1", + paused: false, + }, + { + id: "dead-old", + prompt: "created 8d ago, never fired", + mode: "fixed", + intervalMs: 60_000, + createdAt: now - 8 * 86_400_000, + lastFiredAt: 0, + nextDueAt: now + 60_000, + source: "user", + directory: "/tmp", + sessionID: "s1", + paused: false, + }, + ], + } + writeFileSync(join(dir, "tasks.json"), JSON.stringify(data), "utf-8") + const store = new LoopStore({ storageDir: dir, taskTtlMs: 7 * 86_400_000 }) + await store.load() + assert.deepEqual(store.list().map((t) => t.id), ["active-old"]) + } finally { + rmSync(dir, { recursive: true }) + } +})