diff --git a/README.md b/README.md index cb1e2a3..26b3b0c 100644 --- a/README.md +++ b/README.md @@ -18,9 +18,10 @@ A drop-in `/loop` command for [opencode](https://opencode.ai), modeled after Cla - **`/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) +- **Subcommands** — `list | status | cancel | stop | pause | resume | stop-all` (session-scoped; add `--all` to cross sessions; bare `stop` cancels every task in scope) - **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 +- **Prompt fidelity** — flags (`--once`, `--all`, `--jitter=*`) are only recognized before the prompt begins; `--` forces the rest to be treated as prompt text verbatim, and whitespace/newlines are preserved +- **Per-process instance coordination** — plugin instances inside one process (case-variant plugin paths, per-command `opencode run` instances) elect a single leader so tasks never double-fire; a second OpenCode process in the same project fires its own tasks independently, and merge-writes prevent task loss - **Inflight guard** — double-set at ticker and `fireTask` level prevents double-firing even if opencode hot-reloads the plugin - **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 @@ -121,7 +122,16 @@ Re-run `npm run build` after editing `src/`, then restart OpenCode to load the r Fixed tasks use deterministic Jitter by default for backward compatibility. Add `--jitter=false` for an exact interval or `--jitter=true` to enable it explicitly. -The flag is scheduling metadata and is removed from the repeated prompt. +Flags are only recognized **before the prompt begins**: anything after the first +prompt word — including text that looks like `--once`, `--all`, or `--jitter=*` — +is part of the prompt and is preserved verbatim (whitespace and newlines +included). Use `--` to force everything after it to be treated as prompt text: + +``` +/loop 1m --once remind me to stretch # --once is a flag (prefix position) +/loop 1m explain what --once means # --once is prompt text, kept verbatim +/loop 1m -- --jitter=false is a flag too # -- forces the rest to be prompt +``` ### Adaptive interval (LLM decides next fire time) ``` @@ -159,6 +169,9 @@ All subcommands are **session-scoped by default**. Add `--all` to operate across /loop status # alias for list /loop cancel # cancel one task in current session /loop cancel --all # override scope +/loop stop # alias for cancel +/loop stop # cancel ALL tasks in current session +/loop stop --all # cancel ALL tasks across sessions /loop pause # pause one /loop resume # resume one (re-arms per mode) /loop stop-all # cancel all tasks in current session @@ -245,14 +258,17 @@ The built-in runtime defaults are: | Ephemeral tasks | enabled | **Ephemeral lifecycle.** With `ephemeralTasks` enabled (the default), every -`tasks.json` records the pid and start time of the process that wrote it. On -load, tasks written by any other process — e.g. after OpenCode exits and -restarts — are dropped, so loop tasks never outlive the process that created -them (the same lifecycle as Claude Code's `/loop`). Same-process plugin reloads -keep their tasks. Pass `{ ephemeralTasks: false }` in the plugin options to -restore the previous behavior of persisting tasks across process restarts. Note -that upgrading from a release without process-identity tracking drops the -existing `tasks.json` once, since it carries no trusted writer identity. +task records its owning process (`ownerPid` + start time) in `tasks.json`. On +load, tasks whose owner process is confirmed dead — e.g. after that OpenCode +process exits — are dropped, so loop tasks never outlive the process that +created them (the same lifecycle as Claude Code's `/loop`). Tasks owned by +other **live** OpenCode processes in the same project are kept: they remain +visible and manageable via `--all`, and each process fires only its own tasks. +Same-process plugin reloads keep their tasks. Pass `{ ephemeralTasks: false }` +in the plugin options to restore the previous behavior of persisting tasks +across process restarts. Note that upgrading from a release without per-task +owner tracking drops tasks that carry no owner identity once, since their +writer cannot be verified. Adaptive minimum and maximum delays are persisted on each task. The random fallback and any model-requested `reschedule` are both constrained by that task's bounds. Jitter @@ -276,12 +292,12 @@ Each `/loop` task carries a `sessionID` field: | User runs `/loop` in session B | Session B becomes active; A's task waits | | `session.deleted` for session A | All A's tasks cancelled automatically | | Plugin reload (`opencode` hot-reload) | Old tickers stop, new ticker starts; in-flight tasks guarded by `inflight` Set | -| Process restart (new pid) | With `ephemeralTasks` enabled (default), all tasks from the previous process are dropped on load; with it disabled, tasks resume as before | +| Process restart (new pid) | With `ephemeralTasks` enabled (default), tasks whose owner process is dead are dropped on load, while tasks owned by other live processes are kept; with it disabled, tasks resume as before | | Old `tasks.json` without `sessionID` | Dropped on load (with log message) | ## Storage -Tasks persist to `.opencode/cache/loop/tasks.json` (per project). Fire history is logged to `history.log` next to the store. The state file also records the writer's `pid` and `startedAt`, which the ephemeral lifecycle uses to detect process restarts. +Tasks persist to `.opencode/cache/loop/tasks.json` (per project). Fire history is logged to `history.log` next to the store. Each task records its owner's `ownerPid` and `ownerStartedAt`, which the ephemeral lifecycle uses to tell dead-process leftovers apart from tasks owned by other live OpenCode processes. ## Troubleshooting diff --git a/package.json b/package.json index 776d359..1ac446d 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode-plugin-loop", - "version": "0.7.1", + "version": "0.7.2", "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/cron-parser.ts b/src/cron-parser.ts index 5a52042..efe262e 100644 --- a/src/cron-parser.ts +++ b/src/cron-parser.ts @@ -25,6 +25,22 @@ const UNIT_TO_MS: Record = { } const MIN_INTERVAL_MS = 1_000 +/** + * Single validator shared by the slash parser, the LLM tool `create`, and + * `set_fixed` — all entries must accept/reject the same fixed intervals. + */ +export function validateFixedInterval( + ms: unknown +): { ok: true; ms: number } | { ok: false; error: string } { + if (typeof ms !== "number" || !Number.isFinite(ms)) { + return { ok: false, error: "intervalMs must be a finite number" } + } + if (ms < MIN_INTERVAL_MS) { + return { ok: false, error: `intervalMs must be at least ${MIN_INTERVAL_MS}ms` } + } + return { ok: true, ms } +} + export interface CronParserInstance { parse(input: string): ParsedInterval | null extractInterval(text: string): { interval: ParsedInterval | null; rest: string } @@ -62,17 +78,19 @@ export function CronParser(this: unknown): CronParserInstance { } }, - /** Try to extract an interval from a user command like "5m check deploy" */ + /** Try to extract an interval from a user command like "5m check deploy". + * `rest` is the ORIGINAL substring after the interval token (not a token + * re-join), so prompt whitespace and newlines are preserved verbatim. */ extractInterval(text: string) { - const tokens = text.trim().split(/\s+/) - if (tokens.length === 0) return { interval: null, rest: text } - - const first = tokens[0] + const trimmed = text.trim() + if (!trimmed) return { interval: null, rest: text } + const firstMatch = /^\S+/.exec(trimmed) + const first = firstMatch?.[0] ?? "" const parsed = inst.parse(first) if (parsed) { return { interval: parsed, - rest: tokens.slice(1).join(" "), + rest: trimmed.slice(first.length).replace(/^\s+/, ""), } } return { interval: null, rest: text } diff --git a/src/index.ts b/src/index.ts index db5bf22..68c88ff 100644 --- a/src/index.ts +++ b/src/index.ts @@ -110,20 +110,23 @@ 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. + // Firing requires lock.shouldFire(): the lock serializes same-process plugin + // instances only — when the lock is held by ANOTHER process, that process + // fires its own tasks and we fire ours (owner-pid filter below). 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 (lockEnabled && !lock.shouldFire()) return if (!activeSessionID) return const due = await scheduler.getDueTasksForSession(activeSessionID) if (due.length === 0) return for (const task of due) { if (task.sessionID !== activeSessionID) continue + // Tasks created by another live process are fired by that process. + if (task.ownerPid !== undefined && task.ownerPid !== process.pid) continue if (inflight.has(task.id)) continue inflight.add(task.id) try { diff --git a/src/instance-lock.ts b/src/instance-lock.ts index c5172e8..a20c946 100644 --- a/src/instance-lock.ts +++ b/src/instance-lock.ts @@ -42,6 +42,13 @@ export interface InstanceLockOptions { export interface InstanceLockInstance { instanceId: string isLeader(): boolean + /** + * Whether this instance's ticker may fire tasks. The lock only serializes + * instances inside ONE process (case-variant plugin paths); when the lock + * is held by a DIFFERENT process, that process fires its own tasks and we + * fire ours — so anything but "same-pid foreign leader" allows firing. + */ + shouldFire(): boolean /** Begin heartbeat / takeover probing. Safe to call once. */ start(): void /** Stop probing; release the lock if leader. */ @@ -66,6 +73,7 @@ export function InstanceLock(this: unknown, options: InstanceLockOptions): Insta const lockFile = join(lockDir, "lock.json") let leading = false + let ownerPid: number | null = null let timer: ReturnType | null = null const writeLockFile = () => { @@ -90,6 +98,7 @@ export function InstanceLock(this: unknown, options: InstanceLockOptions): Insta try { mkdirSync(lockDir) writeLockFile() + ownerPid = process.pid return true } catch { return false @@ -103,9 +112,11 @@ export function InstanceLock(this: unknown, options: InstanceLockOptions): Insta // Lock held: am I the owner? (e.g. after a same-process reload) try { const owner = JSON.parse(readFileSync(lockFile, "utf-8")) as LockFile + ownerPid = owner.pid if (owner.instanceId === instanceId) return true } catch { // Unreadable lock file: fall through to staleness check + ownerPid = null } const mtime = readLockMtime() const stale = mtime === null || now() - mtime > staleMs @@ -137,6 +148,7 @@ export function InstanceLock(this: unknown, options: InstanceLockOptions): Insta // deciding our heartbeat stopped (e.g. event-loop stall). try { const owner = JSON.parse(readFileSync(lockFile, "utf-8")) as LockFile + ownerPid = owner.pid if (owner.instanceId !== instanceId) { leading = false await logger("warn", "loop instance lock lost", { instanceId }) @@ -144,6 +156,7 @@ export function InstanceLock(this: unknown, options: InstanceLockOptions): Insta } } catch { leading = false + ownerPid = null await logger("warn", "loop instance lock lost (unreadable)", { instanceId }) return } @@ -164,6 +177,7 @@ export function InstanceLock(this: unknown, options: InstanceLockOptions): Insta const inst: InstanceLockInstance = { instanceId, isLeader: () => leading, + shouldFire: () => leading || ownerPid !== process.pid, start: () => { if (timer) return void tick() @@ -185,6 +199,7 @@ export function InstanceLock(this: unknown, options: InstanceLockOptions): Insta } leading = false } + ownerPid = null }, } return inst diff --git a/src/scheduler.ts b/src/scheduler.ts index 873a85a..2cd0079 100644 --- a/src/scheduler.ts +++ b/src/scheduler.ts @@ -66,16 +66,36 @@ interface SchedulerInstance { export type { SchedulerInstance } -function extractJitterFlag(text: string): { prompt: string; jitterEnabled?: boolean } { - const tokens = text.trim().split(/\s+/) - const index = tokens.findIndex( - (token) => token === "--jitter=true" || token === "--jitter=false" - ) - if (index < 0) return { prompt: text.trim() } - const [flag] = tokens.splice(index, 1) - return { - prompt: tokens.join(" ").trim(), - jitterEnabled: flag === "--jitter=true", +/** + * Parse scheduling flags ONLY in the leading option-prefix region of `text`. + * Scanning stops at the first non-flag token or at a `--` terminator; the + * remainder is returned as the original substring (whitespace and newlines + * preserved), so prompt text that merely looks like a flag (`--once`, + * `--all`, `--jitter=*`) is never stripped from the prompt body. + */ +const PREFIX_FLAGS = new Set(["--once", "--all", "--jitter=true", "--jitter=false"]) + +function parseFlagPrefix(text: string): { + once: boolean + jitterEnabled?: boolean + prompt: string +} { + let rest = text + let once = false + let jitterEnabled: boolean | undefined + for (;;) { + const m = /^\s*(\S+)/.exec(rest) + if (!m) return { once, jitterEnabled, prompt: "" } + const token = m[1] + if (token === "--") { + return { once, jitterEnabled, prompt: rest.slice(m[0].length).trim() } + } + if (!PREFIX_FLAGS.has(token)) { + return { once, jitterEnabled, prompt: rest.trim() } + } + if (token === "--once") once = true + else if (token !== "--all") jitterEnabled = token === "--jitter=true" + rest = rest.slice(m[0].length) } } @@ -104,7 +124,7 @@ const CC_FLAG_MAP: Record = { } /** Flags that are meaningful in command position (not errors when leading). */ -const LEADING_OK = new Set(["--all", "--jitter=true", "--jitter=false", "--once"]) +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 { @@ -123,11 +143,13 @@ Usage: Subcommands (session-scoped; add --all to cross sessions): list | status [--all] Show loop tasks cancel [--all] Cancel one task + stop [] [--all] Cancel one task, or every task in scope when no id is given pause [--all] Pause one task resume [--all] Resume one task stop-all [--all] Cancel all tasks -Flags: +Flags (recognized only before the prompt begins; use -- to force the rest +to be treated as prompt text verbatim): --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) @@ -159,7 +181,6 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta const trimmed = stripOuterQuotes(args.trim()) const tokens = trimmed === "" ? [] : trimmed.split(/\s+/) const allFlag = tokens.includes("--all") - const onceFlag = tokens.includes("--once") let head = tokens[0]?.toLowerCase() // Claude Code-style leading flags map to subcommands (P-1). @@ -181,8 +202,18 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta } if (head === "cancel" || head === "stop") { - const id = tokens[1] - if (!id) return { message: "❌ Usage: /loop cancel [--all]" } + const id = tokens[1] === "--all" ? tokens[2] : tokens[1] + if (!id) { + if (head === "cancel") return { message: "❌ Usage: /loop cancel [--all]" } + // Bare `/loop stop`: cancel every task in scope (session unless --all), + // matching Claude Code's stop semantics for the current context. + if (allFlag) { + const n = await inst.opts.store.cancelAll() + return { message: `🛑 Cancelled ${n} task(s) across all sessions` } + } + const removed = await inst.opts.store.cancelBySession(inst.currentSessionID ?? "") + return { message: `🛑 Cancelled ${removed} task(s) in current session` } + } return inst.handleCancel(id, allFlag) } if (head === "list" || head === "status") { @@ -236,24 +267,18 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta const { interval, rest } = inst.opts.cron.extractInterval(trimmed) if (interval) { - const fixed = extractJitterFlag(rest) - // 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) { + const flags = parseFlagPrefix(rest) + if (!flags.prompt) { return { message: `❌ Missing prompt after interval "${tokens[0]}". Usage: /loop — see \`/loop help\`.`, } } const task = await inst.opts.store.create({ - prompt: fixed.prompt, + prompt: flags.prompt, mode: "fixed", intervalMs: interval.ms, - jitterEnabled: fixed.jitterEnabled ?? inst.opts.defaultJitterEnabled ?? true, - once: onceFlag || undefined, + jitterEnabled: flags.jitterEnabled ?? inst.opts.defaultJitterEnabled ?? true, + once: flags.once || undefined, directory, source: "user", sessionID, @@ -261,12 +286,12 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta return { task, modelPrompt: buildLoopCreatedPrompt({ - prompt: fixed.prompt, + prompt: flags.prompt, schedule: `every ${interval.display}`, taskId: task.id, once: task.once, }), - 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}\``, + message: `🔁 Loop started: every ${interval.display}, prompt "${flags.prompt.slice(0, 50)}${flags.prompt.length > 50 ? "..." : ""}" [id=${task.id}] [s=${sessionID.slice(0, 8)}]${task.once ? " (runs once)" : ""}. Cancel: \`/loop cancel ${task.id}\``, } } @@ -284,19 +309,17 @@ export function Scheduler(this: unknown, opts: SchedulerOptions): SchedulerInsta } 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) { + // Flags are recognized only in the leading option prefix; everything + // after the first non-flag token (or `--`) is the prompt, verbatim. + const flags = parseFlagPrefix(trimmed) + if (!flags.prompt) { return { message: "❌ Empty loop command — see `/loop help`." } } - if (onceFlag) { + if (flags.once) { return { message: "❌ --once is only supported for fixed-interval tasks, e.g. `/loop 30s --once `." } } const task = await inst.opts.store.create({ - prompt, + prompt: flags.prompt, mode: "adaptive", adaptiveMinMs: inst.opts.adaptiveMinMs, adaptiveMaxMs: inst.opts.adaptiveMaxMs, @@ -312,7 +335,7 @@ 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): "${prompt.slice(0, 50)}${prompt.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): "${flags.prompt.slice(0, 50)}${flags.prompt.length > 50 ? "..." : ""}" [id=${task.id}] [s=${sessionID.slice(0, 8)}]. Cancel: \`/loop cancel ${task.id}\``, } } diff --git a/src/store.ts b/src/store.ts index 479758d..a4d80b3 100644 --- a/src/store.ts +++ b/src/store.ts @@ -24,13 +24,21 @@ export interface LoopStoreOptions { taskTtlMs?: number logger?: LoopLogger /** - * Ephemeral lifecycle (default true): tasks written by a different process are - * dropped on load. Process identity is tracked via pid + process start time so - * same-process plugin reloads keep their tasks. + * Ephemeral lifecycle (default true): tasks die with their owning process. + * Each task records ownerPid + ownerStartedAt; on load, tasks whose owner + * process is confirmed dead are dropped, while tasks owned by other LIVE + * processes are kept (visible/manageable via --all). Tasks without owner + * fields (written before owner leases existed) fall back to the file-writer + * identity rule (pid + start time). */ ephemeralTasks?: boolean /** Injectable process identity for tests; defaults to the current process. */ processIdentity?: ProcessIdentity + /** + * Liveness probe for task-owner processes (default: signal 0; EPERM counts + * as alive). Injectable so tests can simulate concurrent processes. + */ + processAlive?: (pid: number) => boolean } export interface ProcessIdentity { @@ -92,6 +100,24 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI pid: process.pid, startedAt: Date.now() - process.uptime() * 1000, } + let processAlive: (pid: number) => boolean = (pid) => { + try { + process.kill(pid, 0) + return true + } catch (err) { + // EPERM: the process exists but is owned by another user — still alive. + return (err as NodeJS.ErrnoException)?.code === "EPERM" + } + } + /** Probe wrapper: any probe failure keeps the task (never delete data we + * cannot verify). Only an explicit "not alive" verdict drops a task. */ + const ownerAlive = (pid: number): boolean => { + try { + return processAlive(pid) + } catch { + return true + } + } /** Same-process reload keeps tasks; a different pid — or a recycled pid whose * recorded start time diverges — means the writer was a previous process. */ const PID_START_TOLERANCE_MS = 30_000 @@ -99,6 +125,12 @@ 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 + /** Per-task owner lease: a task survives load when its owner is this process + * or a verifiably live foreign process; dead owners lose their tasks. */ + const isOwnTask = (t: LoopTask): boolean => + t.ownerPid === identity.pid && + t.ownerStartedAt !== undefined && + Math.abs(t.ownerStartedAt - 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 @@ -133,38 +165,53 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI if (parsed.version !== 1) { throw new Error(`Unsupported state version: ${parsed.version}`) } - 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() + let candidates = parsed.tasks + if (ephemeralTasks) { + candidates = parsed.tasks.filter((t) => { + if (t.ownerPid === undefined) { + // Legacy tasks written before owner leases existed: fall back to + // the file-writer rule (drop unless written by this process). + return isSameProcess(parsed) + } + return isOwnTask(t) || ownerAlive(t.ownerPid) + }) + const dropped = parsed.tasks.length - candidates.length if (dropped > 0) { - await logger("info", `ephemeral cleanup: dropped ${dropped} task(s) from previous process`, { + for (const t of parsed.tasks) { + if (!candidates.includes(t)) tombstones.add(t.id) + } + await logger("info", `ephemeral cleanup: dropped ${dropped} task(s) from dead process(es)`, { count: dropped, previousPid: parsed.pid, }) } - return } const cutoff = Date.now() - inst.taskTtlMs // 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)) + const filtered = candidates.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) { + for (const t of candidates) { if (!filtered.includes(t)) tombstones.add(t.id) } + const ttlDropped = candidates.length - filtered.length 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 - await logger("info", `cleaned ${dropped} task(s) on load (orphan/expired)`, { - count: dropped, + if (ttlDropped > 0) { + await logger("info", `cleaned ${ttlDropped} task(s) on load (orphan/expired)`, { + count: ttlDropped, }) } + // Adopt identity / persist deletions whenever the on-disk state + // diverges from what we keep (including the writer identity itself). + if ( + inst.state.tasks.length !== parsed.tasks.length || + parsed.pid !== identity.pid || + parsed.startedAt !== identity.startedAt + ) { + await inst.persist() + } } catch (err) { const backup = `${inst.filePath}.corrupted.${Date.now()}` try { @@ -245,6 +292,8 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI // Only present on one-shot tasks — keeps persisted JSON stable for // tasks that never use the field. ...(input.once ? { once: true as const } : {}), + ownerPid: identity.pid, + ownerStartedAt: identity.startedAt, } inst.state.tasks.push(task) dirtyIds.add(task.id) @@ -382,6 +431,7 @@ export function LoopStore(this: unknown, options?: LoopStoreOptions): LoopStoreI logger = options.logger ?? logger ephemeralTasks = options.ephemeralTasks ?? true identity = options.processIdentity ?? identity + processAlive = options.processAlive ?? processAlive inst.filePath = join(options.storageDir, "tasks.json") inst.maxTasks = options.maxTasks ?? 50 inst.taskTtlMs = options.taskTtlMs ?? 7 * 24 * 60 * 60 * 1000 diff --git a/src/tools/loop-tools.ts b/src/tools/loop-tools.ts index 7259dae..cad01b9 100644 --- a/src/tools/loop-tools.ts +++ b/src/tools/loop-tools.ts @@ -15,6 +15,7 @@ import { tool, type ToolDefinition } from "@opencode-ai/plugin/tool" import { z } from "zod" import type { LoopStoreInstance as LoopStore } from "../store.js" import type { SchedulerInstance as Scheduler } from "../scheduler.js" +import { validateFixedInterval } from "../cron-parser.js" export async function buildLoopTools( store: LoopStore, @@ -28,7 +29,7 @@ export async function buildLoopTools( action: z.enum(["create", "list", "cancel", "reschedule", "set_fixed", "pause", "resume"]), taskId: z.string().optional().describe("Required for cancel/reschedule/set_fixed/pause/resume"), prompt: z.string().optional().describe("Required for create; the prompt to re-inject each cycle"), - intervalMs: z.number().finite().optional().describe("Fixed interval in milliseconds (for create+fixed mode or set_fixed)"), + intervalMs: z.number().finite().min(1000).optional().describe("Fixed interval in milliseconds (min 1000; for create+fixed mode or set_fixed)"), 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"), @@ -141,7 +142,20 @@ export async function buildLoopTools( const sid = currentSID if (!sid) return JSON.stringify({ ok: false, error: "No sessionID in context" }) - const mode = args.mode ?? (args.intervalMs ? "fixed" : "adaptive") + const mode = args.mode ?? (args.intervalMs !== undefined ? "fixed" : "adaptive") + if (mode === "fixed") { + // Same rule as the slash parser and set_fixed: fixed requires a + // valid interval of at least 1000ms (rejects 0/undefined/NaN). + const v = validateFixedInterval(args.intervalMs) + if (!v.ok) { + return JSON.stringify({ ok: false, error: `fixed mode requires ${v.error}` }) + } + } else if (args.intervalMs !== undefined) { + return JSON.stringify({ + ok: false, + error: `intervalMs is only valid for fixed tasks (mode=${mode})`, + }) + } if (args.once && mode !== "fixed") { return JSON.stringify({ ok: false, error: "once is supported only for fixed tasks" }) } @@ -153,7 +167,7 @@ export async function buildLoopTools( source: "user", sessionID: sid, } - if (mode === "fixed" && args.intervalMs) { + if (mode === "fixed") { input.intervalMs = args.intervalMs input.jitterEnabled = args.jitterEnabled ?? scheduler.opts.defaultJitterEnabled ?? true @@ -239,11 +253,9 @@ export async function buildLoopTools( case "set_fixed": { if (!args.taskId) return JSON.stringify({ ok: false, error: "taskId required" }) - if (!Number.isFinite(args.intervalMs) || (args.intervalMs ?? 0) < 1_000) { - return JSON.stringify({ - ok: false, - error: "intervalMs must be a finite number of at least 1000ms", - }) + const v = validateFixedInterval(args.intervalMs) + if (!v.ok) { + return JSON.stringify({ ok: false, error: v.error }) } const t = store.get(args.taskId) if (!t) return JSON.stringify({ ok: false, error: `No task ${args.taskId}` }) diff --git a/src/types.ts b/src/types.ts index aa465e9..3aca625 100644 --- a/src/types.ts +++ b/src/types.ts @@ -34,6 +34,11 @@ export interface LoopTask { paused: boolean /** One-shot task: auto-cancelled after the first successful fire (fixed mode only). */ once?: boolean + /** Process that created (and fires) this task. Used to distinguish live + * foreign tasks from dead-process leftovers on load. */ + ownerPid?: number + /** Owner process start time (epoch ms), guards against pid reuse. */ + ownerStartedAt?: number } export interface LoopConfig { @@ -55,15 +60,18 @@ export interface LoopConfig { defaultJitterEnabled?: boolean /** * Ephemeral lifecycle (default true, matching Claude Code's /loop): tasks die - * with the opencode process and are dropped on the next load. Set to false to - * keep the legacy behavior of persisting tasks across process restarts. + * with the opencode process that created them. Each task records its owner + * process; on load, tasks whose owner is confirmed dead are dropped, while + * tasks owned by other LIVE processes are kept (visible via --all). Set to + * false to keep 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). + * Instance coordination (default true): serializes plugin instances inside + * one process (case-variant plugin paths, per-command `opencode run` + * instances) so a task never fires twice; a lock held by a different + * process does not block this process from firing its own tasks. Set to + * false to disable coordination (not recommended). */ instanceLock?: boolean } diff --git a/tests/instance-lock.test.mjs b/tests/instance-lock.test.mjs index 8f458db..59641f8 100644 --- a/tests/instance-lock.test.mjs +++ b/tests/instance-lock.test.mjs @@ -158,3 +158,43 @@ test("crashed leader (no stop) leaves a lock that is eventually taken over", asy rmSync(dir, { recursive: true, force: true }) } }) + +test("shouldFire: leader fires; same-process follower does not", 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(a.shouldFire(), true, "leader fires") + assert.equal(b.isLeader(), false) + assert.equal(b.shouldFire(), false, "same-process follower stays silent") + a.stop() + b.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test("shouldFire: a lock held by ANOTHER process does not block this one", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-lock-")) + try { + // Foreign process holds a fresh lock. + mkdirSync(join(dir, "loop.lock"), { recursive: true }) + writeFileSync( + join(dir, "loop.lock", "lock.json"), + JSON.stringify({ instanceId: "foreign", pid: 999_999, hostname: "x", startedAt: Date.now() }), + "utf-8" + ) + const b = makeLock(dir, "bbb") + b.start() + await tick(60) + assert.equal(b.isLeader(), false, "foreign lock is fresh — no takeover") + assert.equal(b.shouldFire(), true, "foreign process fires its own tasks; we fire ours") + b.stop() + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) diff --git a/tests/integration.test.mjs b/tests/integration.test.mjs index de242e3..b05e780 100644 --- a/tests/integration.test.mjs +++ b/tests/integration.test.mjs @@ -635,3 +635,66 @@ test("ephemeral lifecycle: foreign-pid tasks.json dropped by default, kept with rmSync(dir, { recursive: true }) } }) + +// --- LOOP-003: the LLM tool entry enforces the same fixed-interval rules as the slash parser --- + +test("loop_schedule create: fixed interval validation matches the slash parser", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-int-")) + try { + const mockClient = { tui: { appendPrompt: async () => true } } + const hooks = await pluginModule.LoopPlugin({ + client: mockClient, + project: { id: "test" }, + directory: dir, + worktree: dir, + $: {}, + serverUrl: new URL("http://localhost:3000"), + experimental_workspace: { register: () => {} }, + }) + const tool = hooks.tool.loop_schedule + const ctx = { + sessionID: "s1", + messageID: "m1", + agent: "build", + directory: dir, + worktree: dir, + abort: new AbortController().signal, + metadata: () => {}, + ask: async () => {}, + } + // Schema errors throw; business-layer errors return ok:false. Both are rejections. + const attempt = async (args) => { + try { + return JSON.parse(await tool.execute(args, ctx)) + } catch (e) { + return { ok: false, error: String(e) } + } + } + + for (const bad of [0, 999, -5, Number.NaN]) { + const r = await attempt({ action: "create", mode: "fixed", intervalMs: bad, prompt: "x" }) + assert.equal(r.ok, false, `intervalMs=${bad} rejected`) + } + const missing = await attempt({ action: "create", mode: "fixed", prompt: "x" }) + assert.equal(missing.ok, false, "fixed without intervalMs rejected") + + const adaptiveWithInterval = await attempt({ action: "create", mode: "adaptive", intervalMs: 60_000, prompt: "x" }) + assert.equal(adaptiveWithInterval.ok, false, "adaptive carrying intervalMs rejected") + + const good = await attempt({ action: "create", mode: "fixed", intervalMs: 1000, prompt: "x" }) + assert.equal(good.ok, true, "1000ms accepted") + assert.equal(good.task.intervalMs, 1000) + + // set_fixed uses the same validator. + const adaptive = JSON.parse(await tool.execute({ action: "create", prompt: "adapt me" }, ctx)) + assert.equal(adaptive.ok, true) + const badConvert = await attempt({ action: "set_fixed", taskId: adaptive.task.id, intervalMs: 0 }) + assert.equal(badConvert.ok, false, "set_fixed 0ms rejected") + const goodConvert = await attempt({ action: "set_fixed", taskId: adaptive.task.id, intervalMs: 5_000 }) + assert.equal(goodConvert.ok, true, "set_fixed 5000ms accepted") + + await hooks.dispose() + } finally { + rmSync(dir, { recursive: true }) + } +}) diff --git a/tests/package-exports.test.mjs b/tests/package-exports.test.mjs index 664ba59..7865e77 100644 --- a/tests/package-exports.test.mjs +++ b/tests/package-exports.test.mjs @@ -6,8 +6,8 @@ const packageJson = JSON.parse( await readFile(new URL("../package.json", import.meta.url), "utf8"), ) -test("publishes the 0.7.1 release", () => { - assert.equal(packageJson.version, "0.7.1") +test("publishes the 0.7.2 release", () => { + assert.equal(packageJson.version, "0.7.2") }) test("publishes explicit server and TUI plugin entrypoints", () => { diff --git a/tests/scheduler.test.mjs b/tests/scheduler.test.mjs index 2c46115..99ddfd7 100644 --- a/tests/scheduler.test.mjs +++ b/tests/scheduler.test.mjs @@ -181,7 +181,7 @@ test("explicit fixed commands use the programmatic jitter default", async () => } }) -test("fixed command jitter flags override defaults and are removed from the prompt", async () => { +test("fixed command jitter flags override defaults only in the option prefix", async () => { const { sched, dir } = makeScheduler(undefined, undefined, { defaultJitterEnabled: false, }) @@ -191,7 +191,8 @@ test("fixed command jitter flags override defaults and are removed from the prom "/tmp", "s1" ) - const disabled = await sched.handleUserCommand( + // A flag AFTER the prompt starts is prompt text, not an option (LOOP-001). + const verbatim = await sched.handleUserCommand( "2m check deploy --jitter=false", "/tmp", "s1" @@ -204,8 +205,8 @@ test("fixed command jitter flags override defaults and are removed from the prom assert.equal(enabled.task.jitterEnabled, true) assert.equal(enabled.task.prompt, "check version") - assert.equal(disabled.task.jitterEnabled, false) - assert.equal(disabled.task.prompt, "check deploy") + assert.equal(verbatim.task.jitterEnabled, false, "default applies; mid-prompt flag untouched") + assert.equal(verbatim.task.prompt, "check deploy --jitter=false") assert.equal(invalid.task.jitterEnabled, false) assert.equal(invalid.task.prompt, "--jitter=maybe keep this text") } finally { @@ -638,3 +639,116 @@ test("list marks one-shot tasks", async () => { rmSync(dir, { recursive: true }) } }) + +// --- LOOP-001: prompt fidelity (flags only parsed in the option prefix) --- + +test("LOOP-001: flag-like text inside the prompt is preserved verbatim", async () => { + const { sched, store, dir } = makeScheduler() + try { + const marker = '中文 空格 "双引号" \'单引号\' `反引号` --once --all $HOME' + const r = await sched.handleUserCommand(`1m --jitter=false 把 marker 原样写入：${marker}`, "/tmp", "s1") + assert.equal(r.task.prompt, `把 marker 原样写入：${marker}`) + assert.equal(store.get(r.task.id).prompt, `把 marker 原样写入：${marker}`, "persisted verbatim") + assert.equal(r.task.jitterEnabled, false, "prefix flag still parsed") + assert.equal(r.task.once, undefined, "--once in the body does not make the task one-shot") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("LOOP-001: prefix --once applies; body --once does not", async () => { + const { sched, dir } = makeScheduler() + try { + const oneShot = await sched.handleUserCommand("30s --once remind --once me", "/tmp", "s1") + assert.equal(oneShot.task.once, true) + assert.equal(oneShot.task.prompt, "remind --once me") + const recurring = await sched.handleUserCommand("30s remind --once me", "/tmp", "s1") + assert.equal(recurring.task.once, undefined) + assert.equal(recurring.task.prompt, "remind --once me") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("LOOP-001: -- terminator forces the rest to be prompt text", async () => { + const { sched, dir } = makeScheduler() + try { + const fixed = await sched.handleUserCommand("1m --once -- run tests --jitter=true --all", "/tmp", "s1") + assert.equal(fixed.task.once, true) + assert.equal(fixed.task.prompt, "run tests --jitter=true --all") + assert.equal(fixed.task.jitterEnabled, true, "default jitter; body flag not parsed") + const adaptive = await sched.handleUserCommand("-- check --once things", "/tmp", "s1") + assert.equal(adaptive.task.mode, "adaptive") + assert.equal(adaptive.task.prompt, "check --once things") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("LOOP-001: prompt whitespace and newlines are preserved", async () => { + const { sched, dir } = makeScheduler() + try { + const body = "line one\nline two spaced" + const r = await sched.handleUserCommand(`1m ${body}`, "/tmp", "s1") + assert.equal(r.task.prompt, body) + const a = await sched.handleUserCommand(`keep double spaces\nand newline`, "/tmp", "s1") + assert.equal(a.task.prompt, "keep double spaces\nand newline") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("LOOP-001: duplicate prefix flags are idempotent; last jitter wins", async () => { + const { sched, dir } = makeScheduler() + try { + const r = await sched.handleUserCommand("1m --once --once --jitter=true --jitter=false go", "/tmp", "s1") + assert.equal(r.task.once, true) + assert.equal(r.task.jitterEnabled, false) + assert.equal(r.task.prompt, "go") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +// --- /loop stop without an id cancels every task in scope --- + +test("/loop stop without an id cancels all tasks in the current session", async () => { + const { sched, store, dir } = makeScheduler() + try { + await sched.handleUserCommand("1m task A1", "/tmp", "sA") + await sched.handleUserCommand("1m task A2", "/tmp", "sA") + await sched.handleUserCommand("1m task B1", "/tmp", "sB") + const r = await sched.handleUserCommand("stop", "/tmp", "sA") + assert.match(r.message, /Cancelled 2 task\(s\) in current session/) + assert.equal(store.list().length, 1, "session B untouched") + assert.equal(store.list()[0].prompt, "task B1") + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("/loop stop still cancels a single task", async () => { + const { sched, store, dir } = makeScheduler() + try { + const a = await sched.handleUserCommand("1m task A1", "/tmp", "sA") + await sched.handleUserCommand("1m task A2", "/tmp", "sA") + const r = await sched.handleUserCommand(`stop ${a.task.id}`, "/tmp", "sA") + assert.match(r.message, new RegExp(`Cancelled ${a.task.id}`)) + assert.equal(store.list().length, 1) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("/loop stop --all without an id cancels across sessions", async () => { + const { sched, store, dir } = makeScheduler() + try { + await sched.handleUserCommand("1m task A1", "/tmp", "sA") + await sched.handleUserCommand("1m task B1", "/tmp", "sB") + const r = await sched.handleUserCommand("stop --all", "/tmp", "sA") + assert.match(r.message, /Cancelled 2 task\(s\) across all sessions/) + assert.equal(store.list().length, 0) + } finally { + rmSync(dir, { recursive: true }) + } +}) diff --git a/tests/store.test.mjs b/tests/store.test.mjs index 35af7d7..9eabe2a 100644 --- a/tests/store.test.mjs +++ b/tests/store.test.mjs @@ -509,6 +509,7 @@ test("ephemeral: tasks from a previous process are dropped on load", async () => const s2 = new LoopStore({ storageDir: dir, processIdentity: { pid: 2222, startedAt: Date.now() }, + processAlive: () => false, logger: async (level, message, extra) => logCalls.push({ level, message, extra }), }) await s2.load() @@ -596,6 +597,7 @@ test("ephemeral: recycled pid with divergent startedAt is treated as a new proce const s2 = new LoopStore({ storageDir: dir, processIdentity: { pid: 3333, startedAt: now }, + processAlive: () => false, }) await s2.load() assert.equal(s2.list().length, 0) @@ -779,3 +781,101 @@ test("TTL uses last activity: old-but-active tasks survive (B4)", async () => { rmSync(dir, { recursive: true }) } }) + +// --- LOOP-002: per-task owner leases replace whole-file writer identity --- + +test("ephemeral: tasks owned by a LIVE foreign process are kept on load", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const a = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 1111, startedAt: Date.now() - 10_000 }, + }) + await a.load() + const t = await a.create({ prompt: "A's task", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + assert.equal(t.ownerPid, 1111) + + // Process B loads while A is still alive: A's tasks must survive (LOOP-002). + const b = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 2222, startedAt: Date.now() }, + processAlive: (pid) => pid === 1111, + }) + await b.load() + assert.equal(b.list().length, 1, "live foreign task kept") + assert.equal(b.list()[0].id, t.id) + + // B can still manage A's task explicitly (cross-instance cancel). + const removed = await b.cancel(t.id) + assert.ok(removed) + assert.equal(b.list().length, 0) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("ephemeral: dead owner's tasks dropped, live foreign tasks kept, own tasks kept", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const dead = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 1111, startedAt: Date.now() - 20_000 }, + }) + await dead.load() + await dead.create({ prompt: "dead", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "s1" }) + + const live = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 2222, startedAt: Date.now() - 10_000 }, + processAlive: () => true, + }) + await live.load() + await live.create({ prompt: "live", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "s2" }) + + const me = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 3333, startedAt: Date.now() }, + processAlive: (pid) => pid === 2222, + }) + await me.load() + const prompts = me.list().map((t) => t.prompt).sort() + assert.deepEqual(prompts, ["live"], "only the dead owner's task is gone") + await me.create({ prompt: "mine", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "s3" }) + + // A fourth load still sees live-foreign + own tasks. + const again = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 3333, startedAt: Date.now() }, + processAlive: (pid) => pid === 2222, + }) + await again.load() + assert.equal(again.list().length, 2) + } finally { + rmSync(dir, { recursive: true }) + } +}) + +test("ephemeral: probe errors keep foreign tasks (conservative)", async () => { + const dir = mkdtempSync(join(tmpdir(), "loop-test-")) + try { + const a = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 1111, startedAt: Date.now() - 10_000 }, + }) + await a.load() + await a.create({ prompt: "foreign", mode: "fixed", intervalMs: 60_000, directory: "/tmp", sessionID: "sA" }) + + const b = new LoopStore({ + storageDir: dir, + processIdentity: { pid: 2222, startedAt: Date.now() }, + processAlive: () => { + throw new Error("probe unavailable") + }, + }) + // A failing probe must not crash load and must not delete data it cannot verify. + await b.load() + assert.equal(b.list().length, 1, "unverifiable foreign task kept") + } finally { + rmSync(dir, { recursive: true }) + } +})