feat(plugins): snapshot GC worker + od plugin run shorthand

Plan §3.A5 (snapshot GC) + §3.B3 (od plugin run shorthand).

- New apps/daemon/src/plugins/gc.ts wraps pruneExpiredSnapshots() in a
  periodic worker driven by OD_SNAPSHOT_GC_INTERVAL_MS. Daemon boot
  starts the worker and runs an immediate sweep so a fresh daemon does
  not wait the full interval before reaping pre-existing expired rows.
  Setting the interval to 0 disables the worker entirely.
- New HTTP endpoints:
  * GET  /api/applied-plugins                  → list all snapshots
  * GET  /api/projects/:projectId/applied-plugins → list snapshots for
    one project
  * POST /api/applied-plugins/prune { before? } → operator escape hatch
    that calls pruneExpiredSnapshots() with the optional cutoff so a
    hosted operator can force-delete unreferenced rows older than ts.
- New CLI verbs:
  * od plugin snapshots list  [--project <id>]
  * od plugin snapshots prune [--before <unix-ms>]
  * od plugin run <id> --project <projectId> [--inputs <json>]
    [--grant-caps a,b] (apply + run start shorthand). Capability gate
    failures map to exit 66; missing-input to exit 67.
- ConnectorApiErrorCode gains CONNECTOR_NOT_GRANTED so the §3.A3 gate
  rejection can use the canonical sender.

Daemon tests: 1425 → 1429 (added plugins-snapshot-gc with 4 cases).

Co-authored-by: Tom Huang <1043269994@qq.com>
This commit is contained in:
Cursor Agent
2026-05-09 11:21:12 +00:00
co-authored by Tom Huang
parent 898e959ce8
commit 39820be28b
7 changed files with 404 additions and 2 deletions
+125
View File
@@ -767,6 +767,8 @@ async function runPlugin(args) {
case 'doctor': return runPluginDoctor(rest);
case 'replay': return runPluginReplay(rest);
case 'trust': return runPluginTrust(rest);
case 'snapshots': return runPluginSnapshots(rest);
case 'run': return runPluginRun(rest);
default:
console.error(`unknown subcommand: od plugin ${sub}`);
printPluginHelp();
@@ -774,6 +776,129 @@ async function runPlugin(args) {
}
}
// Plan §3.A5 / spec §16 Phase 5: operator escape hatch for snapshot GC.
// Two subcommands:
// - `od plugin snapshots list [--project <id>]` — list snapshots
// - `od plugin snapshots prune [--before <ts>]` — force-delete expired
// (and optionally older-than-cutoff unreferenced) rows.
async function runPluginSnapshots(args) {
const sub = args[0];
if (!sub || sub === 'help' || args.includes('--help') || args.includes('-h')) {
console.log(`Usage:
od plugin snapshots list [--project <id>] List applied plugin snapshots.
od plugin snapshots prune [--before <unix-ms>] Delete expired (or older-than-cutoff) snapshots.`);
process.exit(args.length === 0 ? 2 : 0);
}
const flags = parseFlags(args.slice(1), { string: PLUGIN_STRING_FLAGS, boolean: PLUGIN_BOOLEAN_FLAGS });
const base = pluginDaemonUrl(flags).replace(/\/$/, '');
if (sub === 'list') {
const url = flags.project
? `${base}/api/projects/${encodeURIComponent(flags.project)}/applied-plugins`
: `${base}/api/applied-plugins`;
const resp = await fetch(url);
if (!resp.ok) {
console.error(`GET ${url} failed: ${resp.status} ${await resp.text()}`);
process.exit(1);
}
const data = await resp.json();
process.stdout.write(JSON.stringify(data, null, 2) + '\n');
return;
}
if (sub === 'prune') {
const url = `${base}/api/applied-plugins/prune`;
const before = flags.before ? Number(flags.before) : undefined;
const resp = await fetch(url, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify(before ? { before } : {}),
});
if (!resp.ok) {
console.error(`POST ${url} failed: ${resp.status} ${await resp.text()}`);
process.exit(1);
}
const data = await resp.json();
if (flags.json) {
process.stdout.write(JSON.stringify(data, null, 2) + '\n');
return;
}
console.log(`[snapshots] pruned ${data.removed ?? 0} snapshot(s)`);
return;
}
console.error(`unknown subcommand: od plugin snapshots ${sub}`);
process.exit(2);
}
// Plan §3.B3: `od plugin run <id>` shorthand. Today this is a thin
// wrapper around `od plugin apply` + `POST /api/runs` so a code agent
// can drive the apply→start→follow loop without two hops. Phase 4
// adds full ND-JSON event streaming through `od run watch`.
async function runPluginRun(rest) {
const flags = parseFlags(rest, { string: PLUGIN_STRING_FLAGS, boolean: PLUGIN_BOOLEAN_FLAGS });
const id = rest.find((a) => !a.startsWith('-')
&& a !== flags['daemon-url']
&& a !== flags.source
&& a !== flags.inputs
&& a !== flags.project
&& a !== flags['snapshot-id']
&& a !== flags.capabilities
&& a !== flags['grant-caps']);
if (!id) {
console.error('Usage: od plugin run <id> --project <projectId> [--inputs <json>] [--grant-caps a,b]');
process.exit(2);
}
if (!flags.project) {
console.error('--project <projectId> is required (Phase 1.5 will add the auto-create wrapper)');
process.exit(2);
}
const inputs = flags.inputs ? safeParseJson(flags.inputs) ?? {} : {};
const grantCaps = typeof flags['grant-caps'] === 'string' && flags['grant-caps'].length > 0
? flags['grant-caps'].split(',').map((c) => c.trim()).filter(Boolean)
: [];
const base = pluginDaemonUrl(flags).replace(/\/$/, '');
// 1. Apply (returns ApplyResult + manifestSourceDigest).
const applyResp = await fetch(`${base}/api/plugins/${encodeURIComponent(id)}/apply`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ inputs, grantCaps, projectId: flags.project }),
});
const applyData = await applyResp.json().catch(() => ({}));
if (!applyResp.ok) {
console.error(`apply failed: ${applyResp.status} ${JSON.stringify(applyData)}`);
process.exit(applyResp.status === 422 ? 67 : 1);
}
// 2. Start the run with pluginId so the daemon resolver pins the
// snapshot to the run object.
const runResp = await fetch(`${base}/api/runs`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({
projectId: flags.project,
pluginId: id,
pluginInputs: inputs,
grantCaps,
}),
});
const runData = await runResp.json().catch(() => ({}));
if (!runResp.ok) {
if (runResp.status === 409 && runData?.error?.code === 'capabilities-required') {
const missing = (runData.error.data?.missing ?? []).join(',');
console.error(`[run] capabilities required: ${missing}`);
console.error(`[run] retry with --grant-caps ${missing} or run \`od plugin trust ${id} --capabilities ${missing}\``);
process.exit(66);
}
console.error(`run failed: ${runResp.status} ${JSON.stringify(runData)}`);
process.exit(1);
}
if (flags.json) {
process.stdout.write(JSON.stringify({ apply: applyData, run: runData }, null, 2) + '\n');
return;
}
console.log(`[run] started run ${runData.runId} (snapshot ${applyData?.appliedPlugin?.snapshotId ?? 'n/a'})`);
if (flags.follow) {
console.log(`[run] follow stream: GET ${base}/api/runs/${runData.runId}/events`);
}
}
function pluginDaemonUrl(flags) {
return (flags && flags['daemon-url']) || process.env.OD_DAEMON_URL || 'http://127.0.0.1:7456';
}
+1
View File
@@ -14,6 +14,7 @@ type ConnectorApiErrorCode =
| 'VALIDATION_FAILED'
| 'CONNECTOR_NOT_FOUND'
| 'CONNECTOR_NOT_CONNECTED'
| 'CONNECTOR_NOT_GRANTED'
| 'CONNECTOR_DISABLED'
| 'CONNECTOR_TOOL_NOT_FOUND'
| 'CONNECTOR_SAFETY_DENIED'
+86
View File
@@ -0,0 +1,86 @@
// Snapshot GC worker (plan §3.A5 / spec §16 Phase 5 / PB2).
//
// The unreferenced-snapshot TTL is enforced by `pruneExpiredSnapshots()`
// (defined in `snapshots.ts`); this module owns the periodic schedule
// and the audit log. The worker is started from `server.ts` boot and
// disabled when `OD_SNAPSHOT_GC_INTERVAL_MS` is set to `0`.
//
// Operator escape hatch: `od plugin snapshots prune --before <ts>` calls
// `pruneExpiredSnapshots(db, { before: cutoff })` synchronously without
// touching the periodic timer.
import type Database from 'better-sqlite3';
import { pruneExpiredSnapshots, type PruneExpiredResult } from './snapshots.js';
import { readPluginEnvKnobs } from '../app-config.js';
type SqliteDb = Database.Database;
export interface SnapshotGcOptions {
db: SqliteDb;
// Override the timer interval. When undefined we read
// `readPluginEnvKnobs().snapshotGcIntervalMs`. Setting `0` disables.
intervalMs?: number;
// Sink for audit messages. Defaults to console.log so packaged
// deployments capture it on stdout.
logger?: (msg: string, meta?: Record<string, unknown>) => void;
// Optional hook invoked after each tick so tests can synchronize on
// sweep completion.
onTick?: (result: PruneExpiredResult) => void;
}
export interface SnapshotGcHandle {
stop(): void;
// Force-runs a sweep synchronously. Useful for the CLI escape hatch
// and for tests that don't want to sleep through the timer.
sweep(now?: number): PruneExpiredResult;
}
const NOOP_HANDLE: SnapshotGcHandle = {
stop: () => undefined,
sweep: () => ({ removed: 0, ids: [] }),
};
export function startSnapshotGc(opts: SnapshotGcOptions): SnapshotGcHandle {
const knobs = readPluginEnvKnobs();
const intervalMs = opts.intervalMs ?? knobs.snapshotGcIntervalMs;
const log = opts.logger ?? defaultLogger;
if (!Number.isFinite(intervalMs) || intervalMs <= 0) {
log('[plugins] snapshot GC disabled (interval ms <= 0)');
return NOOP_HANDLE;
}
const tick = () => {
try {
const result = pruneExpiredSnapshots(opts.db);
if (result.removed > 0) {
log(`[plugins] snapshot GC removed ${result.removed} expired snapshot(s)`, {
ids: result.ids,
});
}
opts.onTick?.(result);
} catch (err) {
log(`[plugins] snapshot GC tick failed: ${(err as Error).message ?? String(err)}`);
}
};
const timer = setInterval(tick, intervalMs);
timer.unref?.();
return {
stop: () => {
clearInterval(timer);
},
sweep: (now?: number) => pruneExpiredSnapshots(opts.db, now ? { now } : {}),
};
}
function defaultLogger(msg: string, meta?: Record<string, unknown>): void {
if (meta) {
// eslint-disable-next-line no-console
console.log(msg, meta);
} else {
// eslint-disable-next-line no-console
console.log(msg);
}
}
+1
View File
@@ -9,6 +9,7 @@ export * from './installer.js';
export * from './persistence.js';
export * from './pipeline.js';
export * from './registry.js';
export * from './gc.js';
export * from './resolve-snapshot.js';
export * from './snapshots.js';
export * from './trust.js';
+63
View File
@@ -49,7 +49,9 @@ import {
listIterationsForRun,
MissingInputError,
pluginPromptBlock,
pruneExpiredSnapshots,
resolvePluginSnapshot,
startSnapshotGc,
uninstallPlugin,
} from './plugins/index.js';
import {
@@ -1950,6 +1952,23 @@ export async function startServer({
console.log('[od] Codex plugins disabled via OD_CODEX_DISABLE_PLUGINS=1');
}
// Plan §3.A5 / spec §16 Phase 5 / PB2: periodic snapshot GC. Disabled
// when OD_SNAPSHOT_GC_INTERVAL_MS is 0; otherwise one-time bootstrap
// sweep + interval. The function returns a NOOP_HANDLE when disabled
// so we don't have to branch on the result.
const snapshotGc = startSnapshotGc({ db });
// One immediate sweep so a daemon that just gained the ALTER doesn't
// wait the full interval before reaping pre-existing expired rows.
try {
const initialSweep = pruneExpiredSnapshots(db);
if (initialSweep.removed > 0) {
console.log(`[plugins] snapshot GC startup sweep removed ${initialSweep.removed} row(s)`);
}
} catch (err) {
console.warn(`[plugins] snapshot GC startup sweep failed: ${(err)?.message ?? err}`);
}
void snapshotGc; // keep handle alive for the daemon's lifetime
// Warm agent-capability probes (e.g. whether the installed Claude Code
// build advertises --include-partial-messages) so the first /api/chat
// hits a populated cache even if /api/agents hasn't been called yet.
@@ -3355,6 +3374,50 @@ export async function startServer({
}
});
// Plan §3.A5: list all applied snapshots; useful for `od plugin
// snapshots list` and the audit dashboard.
app.get('/api/applied-plugins', (_req, res) => {
try {
const rows = db
.prepare(`SELECT id FROM applied_plugin_snapshots ORDER BY applied_at DESC LIMIT 500`)
.all();
res.json({
snapshots: rows.map((r) => getSnapshot(db, (r).id)).filter((x) => x !== null),
});
} catch (err) {
res.status(500).json({ error: String(err) });
}
});
app.get('/api/projects/:projectId/applied-plugins', (req, res) => {
try {
const rows = db
.prepare(`SELECT id FROM applied_plugin_snapshots WHERE project_id = ? ORDER BY applied_at DESC`)
.all(req.params.projectId);
res.json({
snapshots: rows.map((r) => getSnapshot(db, (r).id)).filter((x) => x !== null),
});
} catch (err) {
res.status(500).json({ error: String(err) });
}
});
// Plan §3.A5 / spec §16 Phase 5: operator escape hatch for forced
// snapshot pruning. The periodic worker (`startSnapshotGc`) runs the
// unreferenced-TTL sweep automatically; this endpoint additionally
// accepts `{ before: <unix-ms> }` to force-delete unreferenced rows
// older than the cutoff. Referenced rows (run_id IS NOT NULL) stay
// pinned forever per PB2 reproducibility-first.
app.post('/api/applied-plugins/prune', (req, res) => {
try {
const body = req.body && typeof req.body === 'object' ? req.body : {};
const before = typeof body.before === 'number' ? body.before : undefined;
const result = pruneExpiredSnapshots(db, before ? { before } : {});
res.json({ ok: true, removed: result.removed, ids: result.ids });
} catch (err) {
res.status(500).json({ error: String(err) });
}
});
// Phase 2A: GenUI surface read/write + devloop iteration history + replay.
// Spec §10.3 for the surface lifecycle, §10.2 for devloop, §11.5 for the
// route shapes. The surface writers go through `apps/daemon/src/genui/store.ts`
@@ -0,0 +1,126 @@
// Plan §3.A5 / spec §16 Phase 5 / PB2 — snapshot GC behavior test.
//
// Locks the contract that:
// - Unreferenced snapshots stamp expires_at = appliedAt + TTL.
// - Referenced snapshots (linkSnapshotToRun) clear expires_at to NULL.
// - pruneExpiredSnapshots() removes only rows whose expires_at <= now.
// - The CLI escape hatch `before` cutoff additionally targets older
// unreferenced rows (no expires_at) when called explicitly.
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { mkdtemp, rm } from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import Database from 'better-sqlite3';
import { migratePlugins } from '../src/plugins/persistence.js';
import {
createSnapshot,
getSnapshot,
linkSnapshotToRun,
pruneExpiredSnapshots,
} from '../src/plugins/snapshots.js';
let db: Database.Database;
let tmpDir: string;
beforeEach(async () => {
tmpDir = await mkdtemp(path.join(os.tmpdir(), 'od-gc-'));
db = new Database(path.join(tmpDir, 'test.sqlite'));
db.exec(`
CREATE TABLE projects (id TEXT PRIMARY KEY, name TEXT);
CREATE TABLE conversations (id TEXT PRIMARY KEY, project_id TEXT, title TEXT);
`);
migratePlugins(db);
db.prepare('INSERT INTO projects (id, name) VALUES (?, ?)').run('project-1', 'Project 1');
});
afterEach(async () => {
db.close();
await rm(tmpDir, { recursive: true, force: true });
});
const baseInput = (extra = {}) => ({
projectId: 'project-1',
conversationId: null,
runId: null,
pluginId: 'sample-plugin',
pluginVersion: '1.0.0',
pluginTitle: 'Sample Plugin',
pluginDescription: 'Fixture',
manifestSourceDigest: 'digest-1',
taskKind: 'new-generation' as const,
inputs: { topic: 'design' },
resolvedContext: { items: [] },
pipeline: undefined,
genuiSurfaces: [],
capabilitiesGranted: ['prompt:inject'],
capabilitiesRequired: ['prompt:inject'],
assetsStaged: [],
connectorsRequired: [],
connectorsResolved: [],
mcpServers: [],
query: 'Make a deck',
...extra,
});
describe('snapshot GC', () => {
it('stamps expires_at on unreferenced snapshots and leaves referenced ones pinned', () => {
const unrefs = createSnapshot(db, baseInput());
const referenced = createSnapshot(db, baseInput());
linkSnapshotToRun(db, referenced.snapshotId, 'run-42');
const unrefRow = db
.prepare('SELECT expires_at, run_id FROM applied_plugin_snapshots WHERE id = ?')
.get(unrefs.snapshotId) as { expires_at: number | null; run_id: string | null };
const refRow = db
.prepare('SELECT expires_at, run_id FROM applied_plugin_snapshots WHERE id = ?')
.get(referenced.snapshotId) as { expires_at: number | null; run_id: string | null };
expect(unrefRow.expires_at).not.toBeNull();
expect(unrefRow.run_id).toBeNull();
expect(refRow.expires_at).toBeNull();
expect(refRow.run_id).toBe('run-42');
});
it('prunes expired snapshots and preserves still-valid ones + referenced pins', () => {
// Three snapshots, in three states:
// A) unreferenced + expires_at in the past → pruned.
// B) unreferenced + expires_at in the future → kept.
// C) referenced → kept (expires_at NULL, run_id set).
const a = createSnapshot(db, baseInput());
const b = createSnapshot(db, baseInput());
const c = createSnapshot(db, baseInput());
linkSnapshotToRun(db, c.snapshotId, 'run-c');
// Manually rewrite expires_at on A so we don't have to wait the TTL.
db.prepare('UPDATE applied_plugin_snapshots SET expires_at = ? WHERE id = ?')
.run(Date.now() - 1, a.snapshotId);
const result = pruneExpiredSnapshots(db);
expect(result.removed).toBe(1);
expect(result.ids).toEqual([a.snapshotId]);
expect(getSnapshot(db, a.snapshotId)).toBeNull();
expect(getSnapshot(db, b.snapshotId)).not.toBeNull();
expect(getSnapshot(db, c.snapshotId)).not.toBeNull();
});
it('honors the operator before-cutoff to force-prune unreferenced older rows even if not yet expired', () => {
const old = createSnapshot(db, baseInput());
const fresh = createSnapshot(db, baseInput());
// Backdate `old` so applied_at < cutoff. Don't touch `fresh`.
const ancient = Date.now() - 365 * 24 * 60 * 60 * 1000; // a year ago
db.prepare('UPDATE applied_plugin_snapshots SET applied_at = ?, expires_at = NULL WHERE id = ?')
.run(ancient, old.snapshotId);
const cutoff = Date.now() - 30 * 24 * 60 * 60 * 1000;
const result = pruneExpiredSnapshots(db, { before: cutoff });
expect(result.ids).toEqual([old.snapshotId]);
expect(getSnapshot(db, fresh.snapshotId)).not.toBeNull();
});
it('never prunes referenced rows even when before-cutoff is older than them', () => {
const referenced = createSnapshot(db, baseInput());
linkSnapshotToRun(db, referenced.snapshotId, 'run-r');
const ancient = Date.now() - 365 * 24 * 60 * 60 * 1000;
db.prepare('UPDATE applied_plugin_snapshots SET applied_at = ? WHERE id = ?')
.run(ancient, referenced.snapshotId);
const result = pruneExpiredSnapshots(db, { before: Date.now() });
expect(result.removed).toBe(0);
expect(getSnapshot(db, referenced.snapshotId)).not.toBeNull();
});
});
+2 -2
View File
@@ -433,7 +433,7 @@ describe('API proxy routes', () => {
}),
});
expect(res.status).toBe(409);
const body = await res.json();
const body = (await res.json()) as { error?: { code?: string } };
expect(body?.error?.code).toBe('PLUGIN_REQUIRES_DAEMON');
});
@@ -450,7 +450,7 @@ describe('API proxy routes', () => {
}),
});
expect(res.status).toBe(409);
const body = await res.json();
const body = (await res.json()) as { error?: { code?: string } };
expect(body?.error?.code).toBe('PLUGIN_REQUIRES_DAEMON');
});
}