diff --git a/.github/release-state.json b/.github/release-state.json index 02ba434..a97cbe7 100644 --- a/.github/release-state.json +++ b/.github/release-state.json @@ -1,5 +1,5 @@ { - "pullRequest": 7, - "sourceSHA": "0607dc761f2bff2bafca5c820f1427109cc13fde", - "version": "0.6.3" + "pullRequest": 9, + "sourceSHA": "a5804b907dc37d653bb9b339300dc3865159ec84", + "version": "0.6.4" } diff --git a/CHANGELOG.md b/CHANGELOG.md index ef7de51..ad90690 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,19 @@ include the full version, for example `## 0.7.0-beta.1`. ## Unreleased +## 0.6.4 + +### Fixed + +- Prevent hourly owner eviction using durable activity on one empty maintenance + session. The previous plugin-list heartbeat did not refresh OpenCode's session + inactivity timer. Keepalive never prompts a model or modifies a work session. +- Bound SDK waits independently of adapters that ignore AbortSignal, allowing + shutdown to release locks without interrupting a healthy worktree session. + A replacement dispatcher reconciles the saved session before publication. +- Wait briefly for a retiring owner's locks, bound RPC disposal, and reject + queued checkpoint writes after shutdown to prevent failed plugin reloads. + ## 0.6.3 ### Fixed diff --git a/README.md b/README.md index 5d9c8cf..36701ef 100644 --- a/README.md +++ b/README.md @@ -26,12 +26,12 @@ Nothing needs to be published to npm. `$HOME` expands to your home directory. Run this command on the machine running OpenCode 2: -Latest stable release: **[v0.6.3](https://github.com/d3cker/opencode2-github-automation/releases/tag/v0.6.3)**. +Latest stable release: **[v0.6.4](https://github.com/d3cker/opencode2-github-automation/releases/tag/v0.6.4)**. -[Download the .tgz package](https://github.com/d3cker/opencode2-github-automation/releases/download/v0.6.3/opencode2-automation-0.6.3.tgz) · [SHA-256 checksum](https://github.com/d3cker/opencode2-github-automation/releases/download/v0.6.3/opencode2-automation-0.6.3.tgz.sha256) +[Download the .tgz package](https://github.com/d3cker/opencode2-github-automation/releases/download/v0.6.4/opencode2-automation-0.6.4.tgz) · [SHA-256 checksum](https://github.com/d3cker/opencode2-github-automation/releases/download/v0.6.4/opencode2-automation-0.6.4.tgz.sha256) ```bash -npm install --global --prefix "$HOME/.local" "https://github.com/d3cker/opencode2-github-automation/releases/download/v0.6.3/opencode2-automation-0.6.3.tgz" +npm install --global --prefix "$HOME/.local" "https://github.com/d3cker/opencode2-github-automation/releases/download/v0.6.4/opencode2-automation-0.6.4.tgz" ``` diff --git a/docs/advanced.md b/docs/advanced.md index d64e3c9..286176f 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -101,19 +101,23 @@ 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. +The shared service evicts owner locations after roughly an hour without durable +session activity, even if plugin RPC or HTTP requests continue. Components use one +deterministically identified, empty maintenance session per owner directory and +rename it at startup and every ten minutes. Creation is idempotent, metadata and +location are checked before renaming, and no model is prompted. Requests have a +15-second deadline, never overlap, and require a matching service PID. Standalone +servers without a matching service registration skip this mechanism; use the +shared service for unattended automation. + +SDK adapters may ignore AbortSignal. The plugin therefore bounds its own SDK waits, +preserves healthy worker execution on owner disposal, and settles local state writes +before releasing ownership. A replacement waits up to 15 seconds for the retiring +owner's locks. RPC disposal has a five-second deadline per component; cleanup still +attempts every remaining step. No live lock is forcibly removed. A held-lock startup +error should be investigated via plugin details and server logs. Back up the queue, +worktree, and session database before recovery. Reconcile an already-published PR +and saved session instead of restarting implementation or deleting the worktree. Only one issue executes at a time. Checks must succeed before publication. Push uses the exact verified commit without force. Worktrees remain available for diff --git a/docs/architecture.md b/docs/architecture.md index 4912a70..aee64b2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -42,12 +42,14 @@ 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. +In the shared background service, each scheduler/dispatcher component renews one +empty maintenance session in the owner location at startup and every ten minutes. +OpenCode counts durable session events as activity; polling plugin APIs alone +does not prevent its hourly inactivity eviction. No model is prompted by keepalive. +The service PID must match the plugin process before any maintenance session is +created. Shutdown releases local SDK waits independently of adapter cancellation, +settles state writes, bounds RPC disposal, and releases locks. A healthy worktree +session continues and the replacement owner reconciles its saved identity. 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 cf64421..d20810e 100644 --- a/docs/bot-workflow.md +++ b/docs/bot-workflow.md @@ -35,20 +35,22 @@ 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. +- Each component renews one empty owner maintenance session at startup and every + ten minutes. Durable session events refresh OpenCode's inactivity timer; listing + plugins does not. No model is prompted. A PID check prevents touching a different + service. Requests do not overlap and have a 15-second deadline. Pausing issue + scans does not pause keepalive. Standalone servers without a matching registered + service skip it. - 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. 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. +- Shutdown clears timers and aborts local SDK waits even if the adapter ignores + cancellation. It settles local writes, bounds RPC disposal to five seconds per + component, and attempts all cleanup steps before releasing ownership. A replacement + waits up to 15 seconds for locks without removing them. Healthy worktree sessions + continue; the next owner reconciles the saved session before verifying and publishing. + 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 0806555..7f5bfe0 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -46,11 +46,18 @@ 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. +shared background service, keepalive runs at startup and every ten minutes, +independently of issue polling and its pause setting. It reuses one empty session +named `Automation owner keepalive` in the primary checkout. The session has no +messages, consumes no model tokens, and is not an issue task. Its durable rename +event refreshes OpenCode's activity timer; requests to list plugins do not. +Keepalive only targets a service whose PID matches the plugin process. Standalone +servers without a matching registered service do not receive this protection. + +On owner reload, the dispatcher stops waiting locally and releases its lock without +interrupting a healthy worker. The next owner reconnects to the saved session ID, +checks its outcome, verifies the changes, and resumes publication. Explicit task +cancellation, unanswered questions, and session deadlines still interrupt workers. 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 diff --git a/package-lock.json b/package-lock.json index 7e8d435..70b740c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "opencode2-automation", - "version": "0.6.3", + "version": "0.6.4", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "opencode2-automation", - "version": "0.6.3", + "version": "0.6.4", "hasInstallScript": true, "dependencies": { "@opencode/client": "0.0.0-beta-19398", diff --git a/package.json b/package.json index addaa96..c078a43 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode2-automation", - "version": "0.6.3", + "version": "0.6.4", "description": "Issue-to-PR automation for OpenCode 2 with a scheduler and GitHub dispatcher", "main": "./dist/index.js", "files": [ diff --git a/src/dispatcher.ts b/src/dispatcher.ts index e969b2a..6a7f395 100644 --- a/src/dispatcher.ts +++ b/src/dispatcher.ts @@ -77,6 +77,7 @@ export class Dispatcher { this.signal.throwIfAborted(); let announce = false; await this.serial.run(async () => { + this.signal.throwIfAborted(); announce = Boolean(patch.sessionReady && !task.sessionReady) || Boolean(patch.status && patch.status !== task.status && ["done", "blocked", "failed", "waiting"].includes(patch.status)); announce ||= patch.pr?.state === "closed" && task.pr?.state !== "closed"; if (patch.sessionID) task.sessionIDs = [...new Set([...task.sessionIDs ?? [], ...[task.previousSessionID, task.sessionID, patch.sessionID].filter((id): id is string => Boolean(id))])]; diff --git a/src/executor.ts b/src/executor.ts index d43f237..f264e04 100644 --- a/src/executor.ts +++ b/src/executor.ts @@ -9,6 +9,7 @@ import { analysisDecision } from "./analysis.js"; import { baseChoice, type BranchInput } from "./branch.js"; import { installWorkerPlugin } from "./worker.js"; import { Blocked, WaitingForAnswer, type Executor, type Task } from "./dispatcher.js"; +import { cancellable } from "./lifecycle.js"; export type CommandRunner = (cwd: string, argv: string[]) => Promise; export function commandRunner(signal: AbortSignal, timeout: number, secretEnv: string): CommandRunner { @@ -104,6 +105,7 @@ export class GitWorkspace { export class OpenCodeExecutor implements Executor { private git: GitWorkspace; constructor(private ctx: Plugin.Context, private options: GithubOptions, private signal: AbortSignal, private runtimeInstaller = installWorkerPlugin) { + this.ctx = { ...ctx, ...(ctx.session ? { session: cancellable(ctx.session) } : {}), ...(ctx.generate ? { generate: cancellable(ctx.generate) } : {}) }; this.git = new GitWorkspace(options.stateDirectory, commandRunner(signal, options.commandTimeoutSeconds * 1000, options.tokenEnv)); } async title(task: Task) { @@ -156,11 +158,9 @@ export class OpenCodeExecutor implements Executor { return this.runtimeInstaller(directory, this.options, commandRunner(this.signal, this.options.commandTimeoutSeconds * 1000, this.options.tokenEnv)); } async run(task: Task, checkpoint: (patch: Partial) => Promise) { - try { await this.runSession(task, checkpoint); } - catch (error) { - if (this.signal.aborted) await this.cancel(task); - throw error; - } + // Owner disposal must not interrupt a healthy worker in another location. + // The replacement owner resumes waiting on the saved session ID. + await this.runSession(task, checkpoint); } private async runSession(task: Task, checkpoint: (patch: Partial) => Promise) { if (!task.worktree || !task.route) throw new Blocked("Missing execution configuration"); diff --git a/src/lifecycle.ts b/src/lifecycle.ts index b5ae498..e433079 100644 --- a/src/lifecycle.ts +++ b/src/lifecycle.ts @@ -1,5 +1,6 @@ import { OpenCode } from "@opencode/client"; import { Service } from "@opencode/client/service"; +import { createHash } from "node:crypto"; // 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. @@ -13,7 +14,10 @@ export async function cleanup(...steps: (() => void | Promise)[]) { export interface OwnerClient { health: { get(options: { signal: AbortSignal }): Promise<{ pid: number }> }; - plugin: { list(input: { location: { directory: string } }, options: { signal: AbortSignal }): Promise }; + session: { + create(input: { id: string; title: string; location: { directory: string }; metadata: Record }, options: { signal: AbortSignal }): Promise<{ id: string; location: { directory: string }; metadata?: Record }>; + rename(input: { sessionID: string; title: string }, options: { signal: AbortSignal }): Promise; + }; } export async function touchOwner(directory: string, signal: AbortSignal, connect: () => Promise = async () => { @@ -25,22 +29,57 @@ export async function touchOwner(directory: string, signal: AbortSignal, connect // 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 }); + // OpenCode's inactivity sweep observes durable session events, NOT HTTP + // requests or plugin RPC. Reuse one empty maintenance session; never prompt + // a model, touch a user's session, or create a new session on every tick. + const id = `ses_${createHash("sha256").update(`opencode2-automation-owner:${directory}`).digest("hex").slice(0, 32)}`; + const title = "Automation owner keepalive"; + const session = await client.session.create({ id, title, location: { directory }, metadata: { automation: "owner-keepalive" } }, { signal }); + if (session.location.directory !== directory || session.metadata?.automation !== "owner-keepalive") throw new Error("Automation keepalive session identity mismatch"); + // Rename emits Session.Renamed even when the title is unchanged. + await client.session.rename({ sessionID: session.id, title }, { signal }); return true; } -export function heartbeat(touch: (signal: AbortSignal) => Promise, report: (error: unknown) => void, everyMs = 30_000) { +export function heartbeat(touch: (signal: AbortSignal) => Promise, report: (error: unknown) => void, everyMs = 10 * 60_000) { const controller = new AbortController(); let active: Promise | undefined; - const timer = setInterval(() => { + const tick = () => { 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); + }; + const timer = setInterval(tick, everyMs); + tick(); return async () => { clearInterval(timer); controller.abort(); await active; }; } + +// Some OpenCode Promise plugin adapters ignore the request's AbortSignal. +// Bound our wait independently. Late SDK results are observed but cannot resume +// the retired dispatcher; durable session identity allows the next owner to +// reconcile work that continues in the server. +export function abortable(action: () => Promise, signal?: AbortSignal): Promise { + if (!signal) return action(); + return new Promise((resolve, reject) => { + if (signal.aborted) { reject(signal.reason); return; } + const abort = () => reject(signal.reason); + signal.addEventListener("abort", abort, { once: true }); + Promise.resolve().then(() => { signal.throwIfAborted(); return action(); }) + .then(resolve, reject).finally(() => signal.removeEventListener("abort", abort)); + }); +} + +export function cancellable(api: T): T { + return new Proxy(api, { + get(target, property, receiver) { + const method: unknown = Reflect.get(target, property, receiver); + if (typeof method !== "function") return method; + return (input: unknown, request?: { signal?: AbortSignal }) => abortable(() => Reflect.apply(method, target, [input, request]), request?.signal); + }, + }); +} diff --git a/src/plugins/github.ts b/src/plugins/github.ts index 1d1f5cc..9b15449 100644 --- a/src/plugins/github.ts +++ b/src/plugins/github.ts @@ -10,7 +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"; +import { abortable, cleanup, heartbeat, touchOwner } from "../lifecycle.js"; export default Plugin.define({ id: "automation.github", @@ -19,7 +19,7 @@ export default Plugin.define({ if (await realpath(ctx.location.directory) !== await realpath(options.ownerDirectory)) return; const token = await githubToken(options.tokenEnv); const controller = new AbortController(); - const release = await acquire(options.stateDirectory, "github", error => controller.abort(error)); + const release = await acquire(options.stateDirectory, "github", error => controller.abort(error), true); const executor = new OpenCodeExecutor(ctx, options, controller.signal); 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)); @@ -31,7 +31,7 @@ export default Plugin.define({ () => { clearInterval(timer); controller.abort(); }, () => stopHeartbeat?.(), () => dispatcher.settle(), - () => registration?.dispose(), + () => abortable(async () => { await registration?.dispose(); }, AbortSignal.timeout(5_000)), () => releaseBridge?.(), release, ); diff --git a/src/plugins/scheduler.ts b/src/plugins/scheduler.ts index e6b43f0..2862116 100644 --- a/src/plugins/scheduler.ts +++ b/src/plugins/scheduler.ts @@ -5,7 +5,7 @@ import { SchedulerOptions } from "../config.js"; import { Scheduler, SchedulerState } from "../scheduler.js"; import { SchedulerRpc, handlerRpc } from "../rpc.js"; import { acquire, JsonStore, redact } from "../state.js"; -import { cleanup, heartbeat, touchOwner } from "../lifecycle.js"; +import { abortable, cleanup, heartbeat, touchOwner } from "../lifecycle.js"; export default Plugin.define({ id: "automation.scheduler", @@ -13,11 +13,12 @@ export default Plugin.define({ const options = SchedulerOptions.parse(ctx.options); if (await realpath(ctx.location.directory) !== await realpath(options.ownerDirectory)) return; const controller = new AbortController(); - const release = await acquire(options.stateDirectory, "scheduler", error => controller.abort(error)); + const release = await acquire(options.stateDirectory, "scheduler", error => controller.abort(error), true); const scheduler = new Scheduler(options.jobs, new JsonStore(join(options.stateDirectory, "scheduler.json"), SchedulerState, () => []), async job => { controller.signal.throwIfAborted(); const method = ctx.rpc(handlerRpc(job.rpcID, job.method))[job.method]!; - return method(job.input, { signal: AbortSignal.any([controller.signal, AbortSignal.timeout(120_000)]) }); + const signal = AbortSignal.any([controller.signal, AbortSignal.timeout(120_000)]); + return abortable(() => method(job.input, { signal }), signal); }); let registration: { dispose(): Promise } | undefined; let stopHeartbeat: (() => Promise) | undefined; @@ -26,7 +27,7 @@ export default Plugin.define({ () => { clearInterval(timer); controller.abort(); }, () => stopHeartbeat?.(), () => scheduler.settle(), - () => registration?.dispose(), + () => abortable(async () => { await registration?.dispose(); }, AbortSignal.timeout(5_000)), release, ); try { diff --git a/src/state.ts b/src/state.ts index 3a3533d..9fc65c6 100644 --- a/src/state.ts +++ b/src/state.ts @@ -28,10 +28,14 @@ export class JsonStore implements Store { } } -export async function acquire(directory: string, name: string, compromised: (error: Error) => void) { +export async function acquire(directory: string, name: string, compromised: (error: Error) => void, waitForPrevious = false) { await mkdir(directory, { recursive: true, mode: 0o700 }); return lockfile.lock(join(directory, name), { - realpath: false, stale: 30_000, update: 10_000, retries: 0, onCompromised: compromised, + realpath: false, stale: 30_000, update: 10_000, + // Location invalidation can start a replacement before the old plugin's + // finalizers finish. Wait for release, never remove a live owner's lock. + retries: waitForPrevious ? { retries: 30, factor: 1, minTimeout: 500, maxTimeout: 500 } : 0, + onCompromised: compromised, }); } diff --git a/test/core.test.ts b/test/core.test.ts index 174c8be..f82f815 100644 --- a/test/core.test.ts +++ b/test/core.test.ts @@ -9,6 +9,8 @@ import { Scheduler } from "../src/scheduler.js"; import { Dispatcher, Blocked, Queue, type Executor, type GithubPort } from "../src/dispatcher.js"; import { JsonStore, acquire, type Store } from "../src/state.js"; import { Github, GithubError, type Issue, type Comment } from "../src/github.js"; +import type { Plugin } from "@opencode/plugin"; +import { OpenCodeExecutor } from "../src/executor.js"; const route = { agent: "build", model: { providerID: "deepseek", id: "test-model" } }; const options = GithubOptions.parse({ ownerDirectory: "/repo", stateDirectory: "/state", repositories: [{ repo: "owner/repo", directory: "/repo", baseBranch: "main", allowedAuthors: ["alice"], checks: [["npm", "test"]] }], routes: { "@deepseek": route } }); @@ -44,6 +46,39 @@ function fixture() { return { events, store, github, executor, make, advance: () => { time += 4_000_000; } }; } +test("an evicted dispatcher resumes the completed saved session and publishes exactly once", async () => { + const f = fixture(), controller = new AbortController(); + let prompts = 0, interrupts = 0, finished = false; + let entered!: () => void, complete!: () => void; + const waiting = new Promise(resolve => { entered = resolve; }); + const sdkWait = new Promise(resolve => { complete = resolve; }); + const ctx = { session: { + get: async () => ({ location: { directory: "/worktree" }, outcome: finished ? "succeeded" : undefined }), + prompt: async () => { prompts++; }, + // Deliberately ignore request options, as the affected OpenCode adapter does. + wait: async () => { entered(); if (!finished) await sdkWait; }, + context: async () => [{ type: "user", text: "opencode2-task:owner/repo#1" }, { type: "assistant", finish: "stop" }], + interrupt: async () => { interrupts++; }, + } } as unknown as Plugin.Context; + const original = new OpenCodeExecutor(ctx, options, controller.signal, async () => {}); + f.executor.run = original.run.bind(original); + const old = new Dispatcher(options, f.store, f.github, f.executor, controller.signal, [], () => 1000); + await old.init(); await old.scan(); + const running = old.tick(); await waiting; + const sessionID = f.store.data.tasks[0]!.sessionID; + controller.abort(); await running; await old.settle(); + assert.equal(f.store.data.tasks[0]!.phase, "running"); + assert.equal(interrupts, 0); + finished = true; complete(); + const resumed = new OpenCodeExecutor(ctx, options, new AbortController().signal, async () => {}); + f.executor.run = resumed.run.bind(resumed); + const next = f.make(); await next.init(); await next.tick(); await next.tick(); + assert.equal(next.status()[0]!.status, "done"); + assert.equal(next.status()[0]!.sessionID, sessionID); + assert.equal(prompts, 1); + assert.deepEqual(f.events, ["analyze", "comment", "prepare", "verify", "push", "pr"]); +}); + function proposalFixture(botLogin = "alice") { const f = fixture(); const comments: Comment[] = []; diff --git a/test/executor.test.ts b/test/executor.test.ts index 62bf6eb..f0bb269 100644 --- a/test/executor.test.ts +++ b/test/executor.test.ts @@ -68,16 +68,16 @@ test("uncertain prompt and failed session outcomes block verification", async () messages = [{ type: "user", text: "opencode2-task:owner/repo#1" }, { type: "assistant", finish: "error" }]; await assert.rejects(executor.run(t, async () => {}), /did not complete successfully/); }); -test("abort of a running wait interrupts the server session", async () => { +test("owner disposal releases a wait that ignores cancellation and preserves the worker for reconciliation", async () => { let interrupted = false; const controller = new AbortController(); const ctx = { session: { get: async () => ({ location: { directory: "/worktree" } }), - wait: async () => { controller.abort(); throw new Error("aborted"); }, + wait: async () => { controller.abort(); return new Promise(() => {}); }, interrupt: async () => { interrupted = true; }, } } as unknown as Plugin.Context; await assert.rejects(new OpenCodeExecutor(ctx, options, controller.signal, async () => {}).run({ ...task(), sessionID: "ses_test", promptAttempted: true }, async () => {})); - assert.equal(interrupted, true); + assert.equal(interrupted, false); }); test("real git worktree isolates a fix, verifies, commits and pushes to a local bare remote", async () => { diff --git a/test/lifecycle.test.ts b/test/lifecycle.test.ts index f16d5bb..39317f2 100644 --- a/test/lifecycle.test.ts +++ b/test/lifecycle.test.ts @@ -5,7 +5,7 @@ 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 { abortable, 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"; @@ -39,7 +39,7 @@ for (const kind of ["github", "scheduler"] as const) { } : { 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); + const stop = await plugin.setup({ location: { directory }, options, rpc, session: {} } as unknown as Plugin.Context); assert.ok(stop); await assert.rejects(async () => stop(), /Automation cleanup failed/); const release = await acquire(directory, kind, () => {}); @@ -57,7 +57,10 @@ test("owner heartbeat only touches the matching service process and owner direct let pid = process.pid + 1; const client: OwnerClient = { health: { get: async () => ({ pid }) }, - plugin: { list: async ({ location }) => { calls.push(location.directory); } }, + session: { + create: async input => { calls.push(input.location.directory); return input; }, + rename: async () => { calls.push("activity"); }, + }, }; const signal = new AbortController().signal; assert.equal(await touchOwner("/owner", signal, async () => undefined), false); @@ -65,7 +68,29 @@ test("owner heartbeat only touches the matching service process and owner direct assert.deepEqual(calls, []); pid = process.pid; assert.equal(await touchOwner("/owner", signal, async () => client), true); - assert.deepEqual(calls, ["/owner"]); + assert.deepEqual(calls, ["/owner", "activity"]); +}); + +test("keepalive emits durable owner activity, reuses one session, and never invokes a model", async () => { + let session: Awaited> | undefined; + let created = 0, renamed = 0, clock = 0, expiresAt = 60; + const client: OwnerClient = { + health: { get: async () => ({ pid: process.pid }) }, + session: { + create: async input => { if (!session) { session = input; created++; } assert.equal(input.id, session.id); return session; }, + // OpenCode LocationActivity refreshes only on durable SessionEvent events. + rename: async ({ sessionID }) => { assert.equal(sessionID, session!.id); expiresAt = clock + 60; renamed++; }, + }, + }; + for (clock = 0; clock <= 180; clock += 10) { + assert.ok(clock < expiresAt); + await touchOwner("/owner", new AbortController().signal, async () => client); + } + assert.equal(created, 1); + assert.equal(renamed, 19); + session!.location.directory = "/unrelated"; + await assert.rejects(touchOwner("/owner", new AbortController().signal, async () => client), /identity mismatch/); + assert.equal(renamed, 19); }); test("heartbeat does not overlap requests and aborts pending work on cleanup", async t => { @@ -77,7 +102,7 @@ test("heartbeat does not overlap requests and aborts pending work on cleanup", a calls++; pendingSignal = signal; return new Promise((_, reject) => signal.addEventListener("abort", () => reject(signal.reason), { once: true })); - }, error => errors.push(error)); + }, error => errors.push(error), 30_000); t.mock.timers.tick(30_000); await setImmediate(); t.mock.timers.tick(60_000); @@ -95,9 +120,9 @@ 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)); + const stop = heartbeat(async () => { if (++calls === 1) throw new Error("connection lost"); }, error => errors.push(error), 30_000); try { - t.mock.timers.tick(30_000); + // The initial touch runs immediately, before the first interval. await setImmediate(); t.mock.timers.tick(30_000); await setImmediate(); @@ -105,3 +130,28 @@ test("heartbeat retries after a transient failure", async t => { assert.equal(errors.length, 1); } finally { await stop(); } }); + +test("abortable exits even when an SDK ignores its signal and observes late rejection", async () => { + const controller = new AbortController(); + let reject!: (reason: Error) => void; + const pending = abortable(() => new Promise((_, fail) => { reject = fail; }), controller.signal); + await setImmediate(); + controller.abort(new Error("owner disposed")); + await assert.rejects(pending, /owner disposed/); + reject(new Error("late SDK failure")); + await setImmediate(); +}); + +test("replacement waits for the real ownership lock without stealing it", async () => { + const directory = await realpath(await mkdtemp(join(tmpdir(), "oc2-handoff-"))); + try { + const release = await acquire(directory, "github", () => {}); + let acquired = false; + const replacement = acquire(directory, "github", () => {}, true).then(stop => { acquired = true; return stop; }); + await setImmediate(); + assert.equal(acquired, false); + await release(); + await (await replacement)(); + assert.equal(acquired, true); + } finally { await rm(directory, { recursive: true, force: true }); } +});