From 5ce84dd003f2c556086ec33c6138b616c999a793 Mon Sep 17 00:00:00 2001 From: d3cker Date: Sun, 13 Sep 2026 12:29:40 +0200 Subject: [PATCH 1/2] Keep automation owners active and release locks after cleanup failures --- docs/advanced.md | 14 +++++ docs/architecture.md | 7 +++ docs/bot-workflow.md | 10 +++- docs/runtime.md | 12 +++++ src/lifecycle.ts | 46 +++++++++++++++++ src/plugins/github.ts | 27 ++++++++-- src/plugins/scheduler.ts | 25 +++++++-- test/lifecycle.test.ts | 107 +++++++++++++++++++++++++++++++++++++++ 8 files changed, 237 insertions(+), 11 deletions(-) create mode 100644 src/lifecycle.ts create mode 100644 test/lifecycle.test.ts diff --git a/docs/advanced.md b/docs/advanced.md index cc6bf2a..d64e3c9 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -101,6 +101,20 @@ uncertain network results. Uncertain initial prompt delivery is not automaticall resent. Issue replies and helper prompts use deterministic IDs for admission retries. Merge requests pin the verified head SHA and reconcile an already-merged PR. +The shared service can evict an idle owner location even while an agent works in +a different worktree. Scheduler and dispatcher components therefore refresh the +owner every 30 seconds through the public plugin-list API, after confirming the +service PID matches their own process. Transient heartbeat errors are logged and +retried; requests do not overlap. A standalone server without a matching service +registration skips this mechanism. Keep its owner location in use or use the +shared background service for unattended automation. + +Cleanup aborts and settles work before releasing ownership. All cleanup steps +are attempted even when RPC disposal fails; locks are not forcibly removed or +stolen from another owner. A held-lock startup error should be investigated via +plugin details and server logs. Back up the queue and worktree before recovery; +do not assume a failed session means its edits were lost. + Only one issue executes at a time. Checks must succeed before publication. Push uses the exact verified commit without force. Worktrees remain available for inspection; automatic cleanup is not implemented. diff --git a/docs/architecture.md b/docs/architecture.md index 8e7b3fb..4912a70 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -42,5 +42,12 @@ The primary checkout owns scheduling. Worker worktrees do not start additional schedulers. A shared Git-directory state folder holds the queue and locks; separate machines require separate test repositories to avoid duplicate execution. +In the shared background service, each scheduler/dispatcher component sends a +periodic request back to its owner location to prevent idle eviction while work +runs elsewhere. It checks the service PID before touching the location, so a +standalone instance cannot activate another owner in a different service. +Shutdown settles work and attempts all cleanup steps, including lock release, +even if an SDK registration fails to dispose after location eviction. + See [advanced configuration](advanced.md) for retry commands, limits, and RPC settings, and the [README](../README.md) for installation and user-facing behavior. diff --git a/docs/bot-workflow.md b/docs/bot-workflow.md index 772389b..cf64421 100644 --- a/docs/bot-workflow.md +++ b/docs/bot-workflow.md @@ -35,12 +35,20 @@ flowchart TD cancel accepted work or active sessions. - A scan cannot overlap another scan in the same dispatcher. Only one worker invocation runs at a time; one scheduler job cannot overlap itself. +- Each component refreshes the owner through the matching background service's + plugin-list API every 30 seconds. A PID check prevents activating a second + owner in another service. Requests do not overlap and have a 15-second deadline. + This heartbeat keeps the owner loaded while execution happens in worktrees; + pausing issue scans does not pause it. Standalone servers without a matching + registered service skip the refresh. - State is schema-validated and saved through a temporary file, file sync, and rename. A corrupt state file fails to load rather than resetting the queue. Local locks prevent duplicate owners sharing this state directory; independent machines do not share ownership. - Shutdown clears timers, aborts operations, waits for in-flight work, disposes - registrations, and releases locks. Executor shutdown interrupts its task session. + registrations, and releases locks. Cleanup attempts the remaining steps even + if an earlier step fails, so an evicted RPC registration cannot skip lock + release. Executor shutdown interrupts its task session. Sources: [index.ts](../src/index.ts), [easy.ts](../src/easy.ts), [GitHub plugin](../src/plugins/github.ts), diff --git a/docs/runtime.md b/docs/runtime.md index 5333d4d..0806555 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -45,6 +45,18 @@ The queue retains waiting questions and accepted replies across restarts. After restarting the service, load the owner project again to resume polling. No terminal UI is needed to answer in GitHub. +The automation owner must remain loaded while its worktree sessions run. In the +shared background service, the plugin refreshes the owner location every 30 +seconds, independently of issue polling and its pause setting. This prevents +idle owner eviction from disconnecting worker hooks and publication. The refresh +only targets a service whose PID matches the plugin process; standalone servers +without a matching registered service do not receive this heartbeat. + +If the plugin reports a held lock or a worker reports unavailable automation RPC, +inspect plugin details, logs, and queue state before retrying. Preserve the +worktree and queue; a failed session can still contain completed changes. Do not +remove an active lock or restart implementation just to recover publication. + ## Tabs after PR closure Each repository scan checks the state of tracked PRs independently of the diff --git a/src/lifecycle.ts b/src/lifecycle.ts new file mode 100644 index 0000000..b5ae498 --- /dev/null +++ b/src/lifecycle.ts @@ -0,0 +1,46 @@ +import { OpenCode } from "@opencode/client"; +import { Service } from "@opencode/client/service"; + +// Cleanup must reach the lock release even when an evicted SDK scope can no +// longer dispose its RPC registration. Keep the lock until work has settled. +export async function cleanup(...steps: (() => void | Promise)[]) { + const errors: unknown[] = []; + for (const step of steps) { + try { await step(); } catch (error) { errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, "Automation cleanup failed"); +} + +export interface OwnerClient { + health: { get(options: { signal: AbortSignal }): Promise<{ pid: number }> }; + plugin: { list(input: { location: { directory: string } }, options: { signal: AbortSignal }): Promise }; +} + +export async function touchOwner(directory: string, signal: AbortSignal, connect: () => Promise = async () => { + const endpoint = await Service.discover(); + return endpoint ? OpenCode.make({ baseUrl: endpoint.url, headers: Service.headers(endpoint) }) : undefined; +}) { + const client = await connect(); + if (!client) return false; + // A standalone server must never activate a second owner in a different + // background service. The public request must return to this exact process. + if ((await client.health.get({ signal })).pid !== process.pid) return false; + await client.plugin.list({ location: { directory } }, { signal }); + return true; +} + +export function heartbeat(touch: (signal: AbortSignal) => Promise, report: (error: unknown) => void, everyMs = 30_000) { + const controller = new AbortController(); + let active: Promise | undefined; + const timer = setInterval(() => { + if (active || controller.signal.aborted) return; + active = Promise.resolve().then(() => touch(AbortSignal.any([controller.signal, AbortSignal.timeout(15_000)]))) + .then(() => {}, error => { if (!controller.signal.aborted) report(error); }) + .finally(() => { active = undefined; }); + }, everyMs); + return async () => { + clearInterval(timer); + controller.abort(); + await active; + }; +} diff --git a/src/plugins/github.ts b/src/plugins/github.ts index 6449d8c..1d1f5cc 100644 --- a/src/plugins/github.ts +++ b/src/plugins/github.ts @@ -10,6 +10,7 @@ import { GithubRpc } from "../rpc.js"; import { acquire, JsonStore, redact } from "../state.js"; import { githubToken } from "../easy.js"; import type { Activity } from "../activity.js"; +import { cleanup, heartbeat, touchOwner } from "../lifecycle.js"; export default Plugin.define({ id: "automation.github", @@ -23,6 +24,17 @@ export default Plugin.define({ let publish: (activity: Activity) => Promise = async () => {}; const dispatcher = new Dispatcher(options, new JsonStore(join(options.stateDirectory, "queue.json"), Queue, () => ({ version: 1, tasks: [] })), new Github(token, controller.signal, fetch, options.signature), executor, controller.signal, [token], Date.now, activity => publish(activity)); let releaseBridge: (() => void) | undefined; + let registration: { dispose(): Promise } | undefined; + let stopHeartbeat: (() => Promise) | undefined; + let timer: ReturnType | undefined; + const stop = () => cleanup( + () => { clearInterval(timer); controller.abort(); }, + () => stopHeartbeat?.(), + () => dispatcher.settle(), + () => registration?.dispose(), + () => releaseBridge?.(), + release, + ); try { await dispatcher.init(); releaseBridge = registerRuntimeBridge(options.ownerDirectory, { @@ -30,7 +42,7 @@ export default Plugin.define({ question: async ({ sessionID, id, text, permission }) => dispatcher.question(sessionID, id, text, permission), helper: async ({ sessionID, callID, capability }) => dispatcher.helper(sessionID, callID, capability), }); - const registration = await ctx.rpc.register(GithubRpc, { + const rpc = await ctx.rpc.register(GithubRpc, { runtime: async ({ sessionID }) => JSON.parse(JSON.stringify(dispatcher.runtime(sessionID))), question: async ({ sessionID, id, text, permission }) => dispatcher.question(sessionID, id, text, permission), helper: async ({ sessionID, callID, capability }) => dispatcher.helper(sessionID, callID, capability), @@ -43,11 +55,16 @@ export default Plugin.define({ activity: async () => dispatcher.activity(), retry: async ({ key, restartSession }) => { controller.signal.throwIfAborted(); return { accepted: await dispatcher.retry(key, restartSession) }; }, }); - publish = activity => registration.events.emit("activity", activity); + registration = rpc; + publish = activity => rpc.events.emit("activity", activity); const tick = () => { if (!controller.signal.aborted) void dispatcher.tick().catch(error => { console.error("Dispatcher stopped", redact(error, [token])); controller.abort(error); }); }; - const timer = setInterval(tick, options.workerEverySeconds * 1000); + timer = setInterval(tick, options.workerEverySeconds * 1000); + stopHeartbeat = heartbeat(signal => touchOwner(options.ownerDirectory, signal), error => console.error("Automation owner heartbeat failed", redact(error, [token]))); tick(); - return async () => { clearInterval(timer); controller.abort(); await registration.dispose(); await dispatcher.settle(); releaseBridge?.(); await release(); }; - } catch (error) { controller.abort(); releaseBridge?.(); await release(); throw error; } + return stop; + } catch (error) { + await stop().catch(cause => console.error("Automation cleanup failed", redact(cause, [token]))); + throw error; + } }, }); diff --git a/src/plugins/scheduler.ts b/src/plugins/scheduler.ts index e42c953..e6b43f0 100644 --- a/src/plugins/scheduler.ts +++ b/src/plugins/scheduler.ts @@ -4,7 +4,8 @@ import { join } from "node:path"; import { SchedulerOptions } from "../config.js"; import { Scheduler, SchedulerState } from "../scheduler.js"; import { SchedulerRpc, handlerRpc } from "../rpc.js"; -import { acquire, JsonStore } from "../state.js"; +import { acquire, JsonStore, redact } from "../state.js"; +import { cleanup, heartbeat, touchOwner } from "../lifecycle.js"; export default Plugin.define({ id: "automation.scheduler", @@ -18,17 +19,31 @@ export default Plugin.define({ const method = ctx.rpc(handlerRpc(job.rpcID, job.method))[job.method]!; return method(job.input, { signal: AbortSignal.any([controller.signal, AbortSignal.timeout(120_000)]) }); }); + let registration: { dispose(): Promise } | undefined; + let stopHeartbeat: (() => Promise) | undefined; + let timer: ReturnType | undefined; + const stop = () => cleanup( + () => { clearInterval(timer); controller.abort(); }, + () => stopHeartbeat?.(), + () => scheduler.settle(), + () => registration?.dispose(), + release, + ); try { await scheduler.init(); - const registration = await ctx.rpc.register(SchedulerRpc, { + registration = await ctx.rpc.register(SchedulerRpc, { status: async () => JSON.parse(JSON.stringify(scheduler.status())), run: async ({ id }) => { controller.signal.throwIfAborted(); return { started: await scheduler.run(id) }; }, pause: async ({ id, paused }) => { controller.signal.throwIfAborted(); await scheduler.pause(id, paused); return { ok: true }; }, }); const tick = () => { if (!controller.signal.aborted) void scheduler.tick().catch(error => { console.error("Scheduler stopped", error); controller.abort(error); }); }; - const timer = setInterval(tick, 1000); + timer = setInterval(tick, 1000); + stopHeartbeat = heartbeat(signal => touchOwner(options.ownerDirectory, signal), error => console.error("Scheduler owner heartbeat failed", redact(error))); tick(); - return async () => { clearInterval(timer); controller.abort(); await registration.dispose(); await scheduler.settle(); await release(); }; - } catch (error) { controller.abort(); await release(); throw error; } + return stop; + } catch (error) { + await stop().catch(cause => console.error("Scheduler cleanup failed", redact(cause))); + throw error; + } }, }); diff --git a/test/lifecycle.test.ts b/test/lifecycle.test.ts new file mode 100644 index 0000000..f16d5bb --- /dev/null +++ b/test/lifecycle.test.ts @@ -0,0 +1,107 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { mkdtemp, realpath, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { setImmediate } from "node:timers/promises"; +import type { Plugin } from "@opencode/plugin"; +import { cleanup, heartbeat, touchOwner, type OwnerClient } from "../src/lifecycle.js"; +import github from "../src/plugins/github.js"; +import scheduler from "../src/plugins/scheduler.js"; +import { acquire } from "../src/state.js"; +import { runtimeBridge } from "../src/bridge.js"; + +test("cleanup settles work and releases ownership even after disposal failures", async () => { + const steps: string[] = []; + const failure = new Error("SDK scope was evicted"); + await assert.rejects(cleanup( + async () => { steps.push("settled"); }, + () => { steps.push("dispose"); throw failure; }, + () => { steps.push("bridge removed"); }, + () => { steps.push("lock released"); }, + ), (error: AggregateError) => error.errors[0] === failure); + assert.deepEqual(steps, ["settled", "dispose", "bridge removed", "lock released"]); +}); + +for (const kind of ["github", "scheduler"] as const) { + test(`${kind} plugin releases its actual lock when RPC disposal rejects`, async () => { + const directory = await realpath(await mkdtemp(join(tmpdir(), "oc2-lifecycle-"))); + const tokenName = "OC2_LIFECYCLE_TEST_TOKEN"; + process.env[tokenName] = "test-token"; + const rpc = Object.assign(() => ({ scan: async () => ({}) }), { + register: async () => ({ dispose: async () => { throw new Error("evicted registration"); }, events: { emit: async () => {} } }), + }); + const options = kind === "github" ? { + tokenEnv: tokenName, ownerDirectory: directory, stateDirectory: directory, + autoMerge: { enabled: false }, + repositories: [{ repo: "owner/repo", directory, baseBranch: "main", allowedAuthors: ["alice"], checks: [] }], + routes: { "@bot": { model: { providerID: "test", id: "model" } } }, + } : { ownerDirectory: directory, stateDirectory: directory, jobs: [{ id: "scan", everySeconds: 60 }] }; + try { + const plugin = kind === "github" ? github : scheduler; + const stop = await plugin.setup({ location: { directory }, options, rpc } as unknown as Plugin.Context); + assert.ok(stop); + await assert.rejects(async () => stop(), /Automation cleanup failed/); + const release = await acquire(directory, kind, () => {}); + await release(); + assert.equal(runtimeBridge(directory), undefined); + } finally { + delete process.env[tokenName]; + await rm(directory, { recursive: true, force: true }); + } + }); +} + +test("owner heartbeat only touches the matching service process and owner directory", async () => { + const calls: string[] = []; + let pid = process.pid + 1; + const client: OwnerClient = { + health: { get: async () => ({ pid }) }, + plugin: { list: async ({ location }) => { calls.push(location.directory); } }, + }; + const signal = new AbortController().signal; + assert.equal(await touchOwner("/owner", signal, async () => undefined), false); + assert.equal(await touchOwner("/owner", signal, async () => client), false); + assert.deepEqual(calls, []); + pid = process.pid; + assert.equal(await touchOwner("/owner", signal, async () => client), true); + assert.deepEqual(calls, ["/owner"]); +}); + +test("heartbeat does not overlap requests and aborts pending work on cleanup", async t => { + t.mock.timers.enable({ apis: ["setInterval"] }); + let calls = 0; + let pendingSignal: AbortSignal | undefined; + const errors: unknown[] = []; + const stop = heartbeat(signal => { + calls++; + pendingSignal = signal; + return new Promise((_, reject) => signal.addEventListener("abort", () => reject(signal.reason), { once: true })); + }, error => errors.push(error)); + t.mock.timers.tick(30_000); + await setImmediate(); + t.mock.timers.tick(60_000); + await setImmediate(); + assert.equal(calls, 1); + await stop(); + assert.equal(pendingSignal?.aborted, true); + assert.deepEqual(errors, []); + t.mock.timers.tick(60_000); + await setImmediate(); + assert.equal(calls, 1); +}); + +test("heartbeat retries after a transient failure", async t => { + t.mock.timers.enable({ apis: ["setInterval"] }); + let calls = 0; + const errors: unknown[] = []; + const stop = heartbeat(async () => { if (++calls === 1) throw new Error("connection lost"); }, error => errors.push(error)); + try { + t.mock.timers.tick(30_000); + await setImmediate(); + t.mock.timers.tick(30_000); + await setImmediate(); + assert.equal(calls, 2); + assert.equal(errors.length, 1); + } finally { await stop(); } +}); From dcbf7cc2952fc50b7cdec4ac12d86b2cf54535d0 Mon Sep 17 00:00:00 2001 From: d3cker Date: Sun, 13 Sep 2026 12:35:25 +0200 Subject: [PATCH 2/2] Release v0.6.2 --- package-lock.json | 4 ++-- package.json | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/package-lock.json b/package-lock.json index ff75a13..0b7d6c1 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "opencode2-automation", - "version": "0.6.1", + "version": "0.6.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "opencode2-automation", - "version": "0.6.1", + "version": "0.6.2", "hasInstallScript": true, "dependencies": { "@opencode/client": "0.0.0-beta-19398", diff --git a/package.json b/package.json index 32d03e5..0d91fc5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode2-automation", - "version": "0.6.1", + "version": "0.6.2", "description": "Issue-to-PR automation for OpenCode 2 with a scheduler and GitHub dispatcher", "main": "./dist/index.js", "files": [