Skip to content
Open
6 changes: 6 additions & 0 deletions bunfig.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
[test]
# A bare `bun test` from the repo root (or any dir above packages/daemon)
# would otherwise skip the daemon's DB-isolation preload and let daemon tests
# inherit the caller's env — how the production engine DB got wiped
# (ranxianglei/ework#7). Keep in sync with packages/daemon/bunfig.toml.
preload = ["./packages/daemon/tests/setup.ts"]
2 changes: 2 additions & 0 deletions packages/daemon/src/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ export const configSchema = z.object({
infraRetryMax: z.coerce.number().int().nonnegative().default(3),
infraRetryBaseMs: z.coerce.number().int().positive().default(15_000),
recoveryReport: z.boolean().default(true),
reconcileScopes: z.array(z.string()).default([]),
}),
db: z.object({
driver: z.enum(["sqlite", "mysql"]).default("sqlite"),
Expand Down Expand Up @@ -129,6 +130,7 @@ function readWorkSection() {
: 3,
infraRetryBaseMs: process.env.WORK_INFRA_RETRY_BASE_MS ? Math.max(1, Number(process.env.WORK_INFRA_RETRY_BASE_MS)) : 15_000,
recoveryReport: !(process.env.WORK_RECOVERY_REPORT === "false" || process.env.WORK_RECOVERY_REPORT === "0"),
reconcileScopes: (process.env.WORK_RECONCILE_SCOPES ?? "").split(",").map((s) => s.trim()).filter(Boolean),
};
}

Expand Down
5 changes: 5 additions & 0 deletions packages/daemon/src/db.ts
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,11 @@ export function applyPrefix(sql: string): string {
}

// ---- driver selection (env-only; read once at module load) ----
// Exported so tests can assert which file initDB() will open (the ranxianglei/
// ework#7 regression suite compares this against the isolation-pinned path).
export const RESOLVED_DB_PATH: string | undefined =
(process.env.WORK_DB_DRIVER ?? "sqlite").trim().toLowerCase() === "sqlite" ? DB_PATH : undefined;

const DB_DRIVER = (process.env.WORK_DB_DRIVER ?? "sqlite").trim().toLowerCase();
const DB_SKIP_CREATE =
process.env.WORK_DB_SKIP_CREATE === "1" || process.env.WORK_DB_SKIP_CREATE === "true";
Expand Down
7 changes: 5 additions & 2 deletions packages/daemon/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,9 +60,12 @@ async function boot() {

// Multi-machine coordination boot:
// 1. Release any stale owners (dead-daemon cleanup) so we can adopt orphans.
// 2. Register this daemon (adopts an orphan slot if available).
// 3. First-boot migration: claim all pre-existing ownerless issues.
// 2. Release owners whose daemons row vanished entirely (DB wipe recovery).
// 3. Register this daemon (adopts an orphan slot if available).
// 4. First-boot migration: claim all pre-existing ownerless issues.
await store.releaseDeadOwners(config.work.leaseTtlMs);
const dangling = await store.releaseDanglingOwners();
if (dangling > 0) log.info(` recovered ${dangling} issue(s) owned by vanished daemon rows`);
const displayName = hostname();
const internalEndpoint = config.daemon.endpoint || `${config.daemon.host}:${config.daemon.port}`;
const daemonId = await store.registerDaemon(displayName, internalEndpoint, config.work.capacity, config.work.leaseTtlMs);
Expand Down
21 changes: 21 additions & 0 deletions packages/daemon/src/op.ts
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,14 @@ export function sessionToTrackerRef(session: OpSession, issue: Issue): TrackerRe
};
}

/** True for SQLite/MySQL foreign-key violations (message- or errno-based). */
export function isForeignKeyError(err: unknown): boolean {
if (!err || typeof err !== "object") return false;
const e = err as { message?: string; errno?: number };
if (typeof e.message === "string" && /FOREIGN KEY constraint failed/i.test(e.message)) return true;
return e.errno === 1452; // MySQL ER_NO_REFERENCED_ROW_2
}

// ─── Store (async DAO over the global AsyncDatabase from db.ts) ───

export class Store {
Expand Down Expand Up @@ -566,6 +574,19 @@ export class Store {
return res.changes;
}

/**
* Clear owner refs pointing at daemons rows that no longer exist (wiped or
* deleted). Such refs are permanent deadlocks: claimIssue only accepts
* owner IS NULL, and releaseDeadOwners only matches existing-but-stale
* rows. (ranxianglei/ework#7: an emptied daemons table stranded every thread.)
*/
async releaseDanglingOwners(): Promise<number> {
const res = await getDB().run(
"UPDATE {{issues}} SET owner_daemon_id = NULL WHERE owner_daemon_id IS NOT NULL AND owner_daemon_id NOT IN (SELECT id FROM {{daemons}})"
);
return res.changes;
}

/**
* Atomic claim: affected_rows decides the winner. Returns true iff this
* daemon won the race. Re-claiming an issue you already own also returns
Expand Down
117 changes: 106 additions & 11 deletions packages/daemon/src/opencode.ts
Original file line number Diff line number Diff line change
@@ -1,17 +1,18 @@
import { Database } from "bun:sqlite";
import { mkdirSync, writeFileSync, unlinkSync, rmdirSync, readdirSync, existsSync, readFileSync } from "fs";
import { join, dirname, resolve, isAbsolute } from "path";
import { homedir } from "os";
import { homedir, hostname } from "os";
import { log } from "./logger";
import { freeSpaceMb, listBusyOpencodeWorkdirs, purgeStaleNodeModules, purgeStaleWorkdirs, shouldBlockSpawn } from "./workdir-gc";
import type { Config } from "./config";
import type { Store } from "./op";
import { isForeignKeyError, type Store } from "./op";
import type { IssueTracker, TrackerRef, TrackerEvent, TrackerComment, Issue, OpSession, Message } from "./trackers/types";
import { formatKey, parseKey } from "./trackers/types";
import type { RuntimeBackend, RuntimeHandle } from "./runtime/types";
import { OpencodeBackend } from "./runtime/opencode-backend";
import { PiBackend } from "./runtime/pi-backend";
import { downloadIssueAttachments, attachmentNote } from "./attachments";
import { reconcileWebIssues } from "./sync/web-reconcile";

// ─── Types ───

Expand Down Expand Up @@ -544,7 +545,9 @@ export class Engine {
private cfg: Config;
private store: Store;
private trackers: TrackerRegistry;
private readonly daemonId: number;
// Not readonly: healDaemonRow() re-registers the daemon after its identity
// row was lost (DB wipe) and reassigns this to the new row's id.
private daemonId: number;
private readonly takeover: TakeoverStrategy;
private readonly backend: RuntimeBackend;
private gateChecker: (issue: Issue) => Promise<{ allowed: boolean; reason: string; resetMs?: number; concurrency?: number | null; unreachable?: boolean }>;
Expand Down Expand Up @@ -585,6 +588,7 @@ export class Engine {

private observedIssues = new Set<string>();
private lastWorkdirGcAt = 0;
private lastWebReconcileAt = 0;
private badgeWrites = new Map<string, string>();
private observerTimer?: ReturnType<typeof setInterval>;

Expand Down Expand Up @@ -615,6 +619,7 @@ export class Engine {
private static MAX_STUCK_NUDGE_ROUNDS = 1;
private static MAX_RUNTIME_MS = 3 * 60 * 60 * 1000;
private static OBSERVER_INTERVAL_MS = 5 * 60 * 1000;
private static WEB_RECONCILE_INTERVAL_MS = 60 * 60_000;
private static STUCK_THRESHOLD_MS = 30 * 60 * 1000;
private static MAX_REPLY_BURST = 8;
private static REPLY_BURST_WINDOW_MS = 5 * 60 * 1000;
Expand Down Expand Up @@ -877,16 +882,80 @@ export class Engine {
/**
* Ensure this engine owns the issue before doing work on it. Returns true
* if we own it (either already, or just claimed). Returns false if another
* daemon won the claim — caller must skip.
* daemon won the claim — caller must skip. Public so tests can drive the
* self-heal paths directly (see sweepStrandedInterrupted precedent).
*
* Self-heals (ranxianglei/ework#7): a missing issues row is re-created
* first (claiming a nonexistent row matches zero rows and would read as
* "lost the race"), and an FK failure on claim means our daemons row
* vanished (e.g. DB wipe) — re-register and retry the claim once.
*/
private async ensureOwned(issue: Issue): Promise<boolean> {
if (issue.ownerDaemonId === this.daemonId) return true;
const won = await this.store.claimIssue(issue.id, this.daemonId);
if (!won) {
log.info(`engine: lost claim on issue ${issue.id} to another daemon (owner=${issue.ownerDaemonId})`);
return false;
async ensureOwned(issue: Issue): Promise<boolean> {
let target: Issue | undefined = await this.store.getIssue(issue.id);
if (!target) {
// Row was lost (DB wipe etc.): re-create it. Note findOrCreateIssue
// mints a fresh local uid, so all subsequent steps must use the
// returned row — claiming the stale uid would match zero rows. A
// freshly created row has no owner, so fall through to the claim
// instead of trusting the (stale) in-memory ownership.
target = await this.store.findOrCreateIssue(
{ trackerType: issue.trackerType, scope: issue.trackerScope, issueId: issue.trackerIssueId },
issue.trackerScopeKey,
issue.title,
);
} else if (issue.ownerDaemonId === this.daemonId) {
return true;
}

try {
const won = await this.store.claimIssue(target.id, this.daemonId);
if (!won) {
const current = await this.store.getIssue(target.id);
log.info(`engine: lost claim on issue ${target.id} to another daemon (owner=${current?.ownerDaemonId ?? "unknown"})`);
return false;
}
return true;
} catch (err) {
if (!isForeignKeyError(err)) throw err;
log.warn(`engine: claim on issue ${target.id} hit FK violation — daemon row missing, re-registering`);
const healed = await this.healDaemonRow();
if (healed === null) return false;
try {
const won = await this.store.claimIssue(target.id, this.daemonId);
if (!won) {
const current = await this.store.getIssue(target.id);
log.warn(`engine: retry claim on issue ${target.id} after heal still lost (owner=${current?.ownerDaemonId ?? "unknown"})`);
return false;
}
log.info(`engine: recovered ownership of issue ${target.id} after daemon row heal`);
return true;
} catch (retryErr) {
log.error(`engine: retry claim on issue ${target.id} after heal failed:`, (retryErr as Error).message);
return false;
}
}
}

/**
* Re-register this daemon after its identity row was lost (DB wipe etc.),
* absorbing same-host duplicate rows. Returns the new daemon id, or null
* when re-registration itself fails.
*/
private async healDaemonRow(): Promise<number | null> {
try {
const displayName = hostname();
const endpoint = this.cfg.daemon.endpoint || `${this.cfg.daemon.host}:${this.cfg.daemon.port}`;
const id = await this.store.registerDaemon(displayName, endpoint, this.cfg.work.capacity, this.cfg.work.leaseTtlMs);
await this.store.absorbSameHostDaemons(id, displayName, endpoint);
if (id !== this.daemonId) {
log.warn(`engine: daemon identity changed ${this.daemonId} → ${id} (previous row was lost)`);
this.daemonId = id;
}
return id;
} catch (err) {
log.error("engine: daemon re-registration after FK failure failed:", (err as Error).message);
return null;
}
return true;
}

private get stuckThresholdMs(): number {
Expand Down Expand Up @@ -2523,6 +2592,8 @@ export class Engine {
private async runObserverCycle() {
try {
await this.store.releaseDeadOwners(this.cfg.work.leaseTtlMs);
const dangling = await this.store.releaseDanglingOwners();
if (dangling > 0) log.info(`engine: released ${dangling} issue(s) owned by vanished daemon rows`);
} catch (err) {
log.error("engine: releaseDeadOwners failed:", (err as Error).message);
}
Expand Down Expand Up @@ -2593,6 +2664,28 @@ export class Engine {
} catch { /* transient store error — next cycle retries */ }
}

// Web→daemon row reconciliation: backfill open issue rows lost from the
// engine DB (the 2026-09-17 test-wipe incident made that permanent).
// Runs when reachability was confirmed through an owned issue OR when
// there are no owned issues at all — the latter is exactly the state a
// wipe leaves behind, where ownership can never confirm reachability.
if ((webReachable || allOwned.length === 0) && Date.now() - this.lastWebReconcileAt >= Engine.WEB_RECONCILE_INTERVAL_MS) {
this.lastWebReconcileAt = Date.now();
try {
const active = await this.store.listActiveIssues();
const scopes = [...new Set([...active.map((i) => i.trackerScopeKey), ...this.cfg.work.reconcileScopes])];
const result = await reconcileWebIssues({
webUrl: this.cfg.gitea.url,
token: this.cfg.gitea.token,
scopes,
store: this.store,
});
if (result.restored > 0) log.info(`engine: web reconcile restored ${result.restored} missing open issue row(s) (${result.matched}/${result.checked} in scope)`);
} catch (err) {
log.warn("engine: web reconcile failed:", (err as Error).message);
}
}

// Badge reconcile: the badge is written by many racing paths (enqueue,
// drain, finish, expire). Converge it to live truth each cycle:
// processing ⇐ running proc, queued ⇐ pending msgs, else clear.
Expand Down Expand Up @@ -2802,6 +2895,8 @@ export class Engine {
// reclaimable).
try {
await this.store.releaseDeadOwners(this.cfg.work.leaseTtlMs);
const dangling = await this.store.releaseDanglingOwners();
if (dangling > 0) log.info(`engine: released ${dangling} issue(s) owned by vanished daemon rows at boot`);
} catch (err) {
log.error("engine: releaseDeadOwners at boot failed:", (err as Error).message);
}
Expand Down
38 changes: 38 additions & 0 deletions packages/daemon/src/runtime/env-deny.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
/**
* Environment variables that must NEVER reach a spawned agent subprocess,
* no matter what the caller passes in RuntimeSpawnOpts.env.
*
* Two classes:
* - OpenCode process-identity vars: a child must not impersonate its parent
* runtime (would break session/output introspection).
* - Daemon engine DB pointers + mode selector: if an agent-spawned process
* (including a test harness it runs) inherits these, it will open — and can
* wipe — the PRODUCTION engine database. This exact leak emptied the live
* daemon DB via `bun test` in a workdir (ranxianglei/ework#7, incident
* bc#853). Stripping them here is the root-cause fix; tests/setup.ts is
* defense in depth.
*/
export const ENV_DENY_ALWAYS = [
"OPENCODE",
"OPENCODE_PID",
"OPENCODE_RUN_ID",
"OPENCODE_PROCESS_ROLE",
"DAEMON_DB_PATH",
"WORK_DB_PATH",
"OPENCODE_DB_PATH",
"DAEMON_ENV",
] as const;

/**
* Return a copy of `env` with all always-denied keys removed, plus any
* per-deployment `extraDeny` keys. Never mutates the input.
*/
export function stripDeniedEnv(
env: Record<string, string | undefined>,
extraDeny: readonly string[] = [],
): Record<string, string | undefined> {
const out: Record<string, string | undefined> = { ...env };
for (const key of ENV_DENY_ALWAYS) delete out[key];
for (const key of extraDeny) delete out[key];
return out;
}
7 changes: 2 additions & 5 deletions packages/daemon/src/runtime/opencode-backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,7 @@ import type {
SessionOutputResult,
LastModelResult,
} from "./types";

const ENV_DENY_ALWAYS = ["OPENCODE", "OPENCODE_PID", "OPENCODE_RUN_ID", "OPENCODE_PROCESS_ROLE"] as const;
import { stripDeniedEnv } from "./env-deny";

export class OpencodeBackend implements RuntimeBackend {
readonly name = "opencode";
Expand All @@ -32,10 +31,8 @@ export class OpencodeBackend implements RuntimeBackend {
}
args.push(opts.prompt);

const childEnv = { ...opts.env };
const childEnv = stripDeniedEnv(opts.env, this.childEnvDeny);
if (!opts.model) delete childEnv.OPENCODE_MODEL;
for (const key of ENV_DENY_ALWAYS) delete childEnv[key];
for (const key of this.childEnvDeny) delete childEnv[key];

const proc = spawn({
cmd: args,
Expand Down
4 changes: 2 additions & 2 deletions packages/daemon/src/runtime/pi-backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import type {
SessionOutputResult,
LastModelResult,
} from "./types";
import { stripDeniedEnv } from "./env-deny";

export class PiBackend implements RuntimeBackend {
readonly name = "pi";
Expand Down Expand Up @@ -44,8 +45,7 @@ export class PiBackend implements RuntimeBackend {

args.push(opts.prompt);

const childEnv = { ...opts.env };
for (const key of this.childEnvDeny) delete childEnv[key];
const childEnv = stripDeniedEnv(opts.env, this.childEnvDeny);

const proc = spawn({
cmd: args,
Expand Down
Loading
Loading