From 729465a88ccae17456f4f5aa5392db320c334387 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Mon, 14 Sep 2026 16:16:27 -0400 Subject: [PATCH 1/3] fix: harden memory compose lease recovery --- bridges/kimaki/plugins/dm-agent-sync.ts | 149 ++++++++++++++++++------ tests/dm-agent-sync.mjs | 126 +++++++++++++++++++- 2 files changed, 237 insertions(+), 38 deletions(-) diff --git a/bridges/kimaki/plugins/dm-agent-sync.ts b/bridges/kimaki/plugins/dm-agent-sync.ts index 343b1a5..7506dc2 100644 --- a/bridges/kimaki/plugins/dm-agent-sync.ts +++ b/bridges/kimaki/plugins/dm-agent-sync.ts @@ -7,7 +7,7 @@ import { spawn } from "node:child_process"; import { createHash, randomUUID } from "node:crypto"; -import { mkdir, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { Plugin } from "@opencode-ai/plugin"; @@ -18,7 +18,9 @@ const DEFAULT_COMPOSE_TIMEOUT_MS = 10_000; const OUTPUT_LIMIT = 16 * 1024; type ComposeResult = { exitCode: number; output: string; timedOut: boolean }; -type ComposeReceipt = { completedAt: number; result: "refreshed" | "stale_fallback" }; +type ComposeReceipt = { completedAt: number; operationId: string; result: "refreshed" | "stale_fallback" }; +type ComposeOwner = { deadlineAt: number; operationId: string; token: string }; +type ComposeLease = { acquired: boolean; deadlineAt: number; operationId: string; path: string; token: string }; const dmAgentSync: Plugin = async () => { let sessionConfig: { wpCli: WpCli; sitePath: string; agentSlug: string } | undefined; @@ -64,7 +66,7 @@ async function composeMemory(wpCli: WpCli, sitePath: string, agentSlug: string): const lease = await acquireComposeLease(scope, timeoutMs); if (!lease.acquired) { - const receipt = await waitForComposeReceipt(scope, startedAt, timeoutMs); + const receipt = await waitForComposeReceipt(scope, lease.operationId, lease.deadlineAt); const durationMs = Date.now() - startedAt; if (receipt?.result === "refreshed") { // eslint-disable-next-line no-console -- intentional operational log to the OpenCode session console @@ -84,65 +86,114 @@ async function composeMemory(wpCli: WpCli, sitePath: string, agentSlug: string): // prevent OpenCode from accepting a chat message. // eslint-disable-next-line no-console -- intentional operational log to the OpenCode session console console.warn(`[dm-agent-sync] memory compose timed out after ${durationMs}ms; using existing memory files`); - await finishComposeLease(scope, lease.token, "stale_fallback"); + await finishComposeLease(scope, lease, "stale_fallback"); return; } if (result.exitCode !== 0) { // eslint-disable-next-line no-console -- intentional operational log to the OpenCode session console console.warn(`[dm-agent-sync] memory compose failed (exit ${result.exitCode}) after ${durationMs}ms: ${result.output}`); - await finishComposeLease(scope, lease.token, "stale_fallback"); + await finishComposeLease(scope, lease, "stale_fallback"); return; } // eslint-disable-next-line no-console -- intentional operational log to the OpenCode session console console.warn(`[dm-agent-sync] refreshed Data Machine memory in ${durationMs}ms`); - await finishComposeLease(scope, lease.token, "refreshed"); + await finishComposeLease(scope, lease, "refreshed"); } function composeScope(wpCli: WpCli, sitePath: string, agentSlug: string): string { - return createHash("sha256").update(JSON.stringify({ wpCli, sitePath, agentSlug })).digest("hex"); + // The command name alone is not enough: relative executables and WP-CLI + // configuration resolve from the runtime's working directory and identity. + return createHash("sha256").update(JSON.stringify({ + agentSlug, + cwd: process.cwd(), + home: process.env.HOME || "", + path: process.env.PATH || "", + sitePath, + user: process.env.USER || process.env.LOGNAME || "", + wpCli, + wpCliCache: process.env.WP_CLI_CACHE_DIR || "", + wpCliConfig: process.env.WP_CLI_CONFIG_PATH || "", + })).digest("hex"); } function composeStatePath(scope: string): string { - return join(tmpdir(), "wp-coding-agents", "dm-compose", scope); + return join(composeStateDirectory(), scope); } function composeReceiptPath(scope: string): string { return `${composeStatePath(scope)}.receipt`; } -async function acquireComposeLease(scope: string, timeoutMs: number): Promise<{ acquired: boolean; token: string }> { +function composeStateDirectory(): string { + return process.env.DATAMACHINE_COMPOSE_STATE_DIR || join(tmpdir(), "wp-coding-agents", "dm-compose"); +} + +async function acquireComposeLease(scope: string, timeoutMs: number): Promise { const path = composeStatePath(scope); - const token = randomUUID(); + const directory = composeStateDirectory(); try { - await mkdir(join(tmpdir(), "wp-coding-agents", "dm-compose"), { recursive: true, mode: 0o700 }); - await mkdir(path, { recursive: false, mode: 0o700 }); - await writeFile(join(path, "owner"), token, { mode: 0o600 }); - return { acquired: true, token }; - } catch (error: unknown) { - if (!isAlreadyExists(error)) { - return { acquired: false, token: "" }; - } + await mkdir(directory, { recursive: true, mode: 0o700 }); + } catch { + return unavailableLease(); + } + + const created = await createComposeLease(path, timeoutMs); + if (created) { + return created; } - // A killed runtime can leave a lease behind. It cannot block the next chat - // longer than the same bounded compose interval. + const owner = await waitForComposeOwner(path, timeoutMs); + if (!owner) { + // An unreadable or partially written lease is never deleted by a waiter. + // It falls back within one local timeout and a later clean invocation can + // acquire normally once the path disappears. + return unavailableLease(timeoutMs); + } + if (owner.deadlineAt > Date.now()) { + return waitingLease(path, owner); + } + + // Expired owners are never removed or renamed. A separate recovery lease + // makes recovery safe even when the old process wakes after its deadline. + const recoveryPath = `${path}.recovery.${owner.operationId}`; + const recovery = await createComposeLease(recoveryPath, timeoutMs); + if (recovery) return recovery; + const recoveryOwner = await waitForComposeOwner(recoveryPath, timeoutMs); + return recoveryOwner ? waitingLease(recoveryPath, recoveryOwner) : unavailableLease(timeoutMs); +} + +async function createComposeLease(path: string, timeoutMs: number): Promise { + const owner: ComposeOwner = { + deadlineAt: Date.now() + timeoutMs, + operationId: randomUUID(), + token: randomUUID(), + }; + let created = false; try { - if (Date.now() - (await stat(path)).mtimeMs > timeoutMs) { + await mkdir(path, { recursive: false, mode: 0o700 }); + created = true; + await writeFile(join(path, "owner"), JSON.stringify(owner), { mode: 0o600 }); + return { acquired: true, path, ...owner }; + } catch (error: unknown) { + if (created) { await rm(path, { recursive: true, force: true }); - return acquireComposeLease(scope, timeoutMs); } - } catch { - return acquireComposeLease(scope, timeoutMs); + return undefined; } - return { acquired: false, token: "" }; } -async function waitForComposeReceipt(scope: string, startedAt: number, timeoutMs: number): Promise { - const path = composeStatePath(scope); - const deadline = startedAt + timeoutMs; - while (Date.now() < deadline) { +function unavailableLease(timeoutMs = getComposeTimeoutMs()): ComposeLease { + return { acquired: false, deadlineAt: Date.now() + timeoutMs, operationId: "", path: "", token: "" }; +} + +function waitingLease(path: string, owner: ComposeOwner): ComposeLease { + return { acquired: false, path, deadlineAt: owner.deadlineAt, operationId: owner.operationId, token: "" }; +} + +async function waitForComposeReceipt(scope: string, operationId: string, deadlineAt: number): Promise { + while (operationId && Date.now() < deadlineAt) { const receipt = await readComposeReceipt(composeReceiptPath(scope)); - if (receipt && receipt.completedAt >= startedAt) { + if (receipt?.operationId === operationId) { return receipt; } await new Promise((resolve) => setTimeout(resolve, 25)); @@ -150,21 +201,49 @@ async function waitForComposeReceipt(scope: string, startedAt: number, timeoutMs return undefined; } -async function finishComposeLease(scope: string, token: string, result: ComposeReceipt["result"]): Promise { - const path = composeStatePath(scope); +async function finishComposeLease(scope: string, lease: ComposeLease, result: ComposeReceipt["result"]): Promise { + const path = lease.path; try { - if ((await readFile(join(path, "owner"), "utf8")) !== token) { + const owner = await readComposeOwner(path); + if (!owner || owner.token !== lease.token) { return; } // Keep the receipt beside the lock: releasing the lock must not erase the // successful result before the other processes that joined it can read it. - await writeFile(composeReceiptPath(scope), JSON.stringify({ completedAt: Date.now(), result }), { mode: 0o600 }); + await writeFile(composeReceiptPath(scope), JSON.stringify({ completedAt: Date.now(), operationId: owner.operationId, result }), { mode: 0o600 }); await rm(path, { recursive: true, force: true }); } catch { // A best-effort lease failure must not prevent a chat message. } } +async function readComposeOwner(path: string): Promise { + try { + const owner: unknown = JSON.parse(await readFile(join(path, "owner"), "utf8")); + if ( + typeof owner === "object" && owner !== null && + typeof (owner as ComposeOwner).deadlineAt === "number" && + typeof (owner as ComposeOwner).operationId === "string" && + typeof (owner as ComposeOwner).token === "string" + ) { + return owner as ComposeOwner; + } + } catch { + // The owner may still be writing or the state directory may be unavailable. + } + return undefined; +} + +async function waitForComposeOwner(path: string, timeoutMs: number): Promise { + const deadline = Date.now() + Math.min(timeoutMs, 100); + while (Date.now() < deadline) { + const owner = await readComposeOwner(path); + if (owner) return owner; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + return readComposeOwner(path); +} + async function readComposeReceipt(path: string): Promise { try { const receipt: unknown = JSON.parse(await readFile(path, "utf8")); @@ -273,7 +352,7 @@ function getComposeTimeoutMs(): number { } function getSitePath(): string { - return process.env.DATAMACHINE_SITE_PATH || process.env.SITE_PATH || process.env.PWD || ""; + return process.env.DATAMACHINE_SITE_PATH || process.env.SITE_PATH || process.env.PWD || process.cwd(); } function getAgentSlug(input: { instructions?: string[] }): string { diff --git a/tests/dm-agent-sync.mjs b/tests/dm-agent-sync.mjs index 0082572..ade0c7d 100644 --- a/tests/dm-agent-sync.mjs +++ b/tests/dm-agent-sync.mjs @@ -2,7 +2,8 @@ import assert from "node:assert/strict" import { spawn } from "node:child_process" -import { access, chmod, mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises" +import { createHash } from "node:crypto" +import { access, chmod, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" import dmAgentSync from "../bridges/kimaki/plugins/dm-agent-sync.ts" @@ -79,29 +80,50 @@ await withEnv({ const recorder = join(directory, "compose") const count = join(directory, "count") const worker = join(directory, "worker.mjs") + const stateDirectory = join(directory, "state") await writeFile(recorder, `#!/bin/sh printf x >> "$DM_COMPOSE_COUNT" sleep 0.15 `) await chmod(recorder, 0o755) + await writeFile(count, "") + await mkdir(stateDirectory) await writeFile(worker, ` const { default: dmAgentSync } = await import(process.env.DM_AGENT_SYNC_MODULE) const plugin = await dmAgentSync({}) await plugin.config({ instructions: ["/tmp/datamachine-site/agents/intelligence-chubes4/SOUL.md"] }) +if (process.env.DM_READY_FILE) { + const { writeFile } = await import("node:fs/promises") + await writeFile(process.env.DM_READY_FILE, "ready") +} +if (process.env.DM_START_FILE) { + const { access } = await import("node:fs/promises") + while (true) { + try { + await access(process.env.DM_START_FILE) + break + } catch { + await new Promise((resolve) => setTimeout(resolve, 5)) + } + } +} await plugin["chat.message"]({ sessionID: process.env.DM_SESSION_ID }, {}) `) - const runWorker = (sessionID, executable = recorder, composeCount = count) => new Promise((resolve, reject) => { + const runWorker = (sessionID, executable = recorder, composeCount = count, timeout = "1000", stateDirectoryOverride = stateDirectory, startFile = "", readyFile = "") => new Promise((resolve, reject) => { const child = spawn(process.execPath, [worker], { env: { ...process.env, DATAMACHINE_SITE_PATH: sitePath, DATAMACHINE_WP_TRANSPORT_JSON: JSON.stringify([executable]), DATAMACHINE_AGENT_SLUG: "intelligence-chubes4", - DATAMACHINE_COMPOSE_TIMEOUT_MS: "1000", + DATAMACHINE_COMPOSE_TIMEOUT_MS: timeout, + DATAMACHINE_COMPOSE_STATE_DIR: stateDirectoryOverride, DM_COMPOSE_COUNT: composeCount, DM_AGENT_SYNC_MODULE: new URL("../bridges/kimaki/plugins/dm-agent-sync.ts", import.meta.url).href, DM_SESSION_ID: sessionID, + DM_READY_FILE: readyFile, + DM_START_FILE: startFile, EXTERNAL_WORDPRESS: "", }, }) @@ -131,6 +153,104 @@ exit 1 assert.equal((await readFile(failureCount, "utf8")).length, 1) assert.equal(failureOutputs.filter((output) => output.includes("memory compose failed")).length, 1) assert.equal(failureOutputs.filter((output) => output.includes("memory compose stale fallback")).length, 1) + + const scopeFor = (executable) => createHash("sha256").update(JSON.stringify({ + agentSlug: "intelligence-chubes4", + cwd: process.cwd(), + home: process.env.HOME || "", + path: process.env.PATH || "", + sitePath, + user: process.env.USER || process.env.LOGNAME || "", + wpCli: [executable], + wpCliCache: process.env.WP_CLI_CACHE_DIR || "", + wpCliConfig: process.env.WP_CLI_CONFIG_PATH || "", + })).digest("hex") + const writeLease = async (stateDirectory, executable, owner, receipt) => { + const scope = scopeFor(executable) + const leasePath = join(stateDirectory, scope) + await mkdir(leasePath, { recursive: true }) + await writeFile(join(leasePath, "owner"), JSON.stringify(owner)) + if (receipt) await writeFile(`${leasePath}.receipt`, JSON.stringify(receipt)) + return leasePath + } + const waitForFiles = async (files) => { + const deadline = Date.now() + 5000 + while (Date.now() < deadline) { + if (await Promise.all(files.map((file) => access(file).then(() => true, () => false))).then((ready) => ready.every(Boolean))) return + await new Promise((resolve) => setTimeout(resolve, 5)) + } + throw new Error(`workers did not become ready: ${files.join(", ")}`) + } + + // Three independent processes race to reclaim the same expired lease. The + // stale receipt must not be reused by the replacement operation. + const staleStateDirectory = join(directory, "stale-state") + await mkdir(staleStateDirectory) + const staleOperation = "expired-operation" + const staleLeasePath = await writeLease(staleStateDirectory, failingRecorder, { + deadlineAt: Date.now() - 1, + operationId: staleOperation, + token: "expired-token", + }, { + completedAt: Date.now(), + operationId: staleOperation, + result: "refreshed", + }) + const staleFailureCount = join(directory, "stale-failure-count") + const staleOutputs = await Promise.all([ + runWorker("stale-one", failingRecorder, staleFailureCount, "1000", staleStateDirectory), + runWorker("stale-two", failingRecorder, staleFailureCount, "1000", staleStateDirectory), + runWorker("stale-three", failingRecorder, staleFailureCount, "1000", staleStateDirectory), + ]) + assert.equal((await readFile(staleFailureCount, "utf8")).length, 1) + assert.equal(staleOutputs.filter((output) => output.includes("memory compose failed")).length, 1) + assert.equal(staleOutputs.filter((output) => output.includes("reused fresh Data Machine memory")).length, 0) + assert.equal(staleOutputs.filter((output) => output.includes("memory compose stale fallback")).length, 2) + await access(staleLeasePath) + + // Waiters use the owner's recorded deadline, not their shorter local timeout. + const activeStateDirectory = join(directory, "active-state") + await mkdir(activeStateDirectory) + const activeLeasePath = await writeLease(activeStateDirectory, recorder, { + deadlineAt: Date.now() + 1500, + operationId: "active-operation", + token: "active-token", + }) + const activeStartedAt = Date.now() + const activeCount = join(directory, "active-count") + const activeStartFile = join(directory, "active-start") + const activeReadyFiles = [join(directory, "active-one-ready"), join(directory, "active-two-ready")] + const activeWorkers = [ + runWorker("active-one", recorder, activeCount, "10", activeStateDirectory, activeStartFile, activeReadyFiles[0]), + runWorker("active-two", recorder, activeCount, "500", activeStateDirectory, activeStartFile, activeReadyFiles[1]), + ] + await waitForFiles(activeReadyFiles) + await writeFile(activeStartFile, "start") + const activeOutputs = await Promise.all(activeWorkers) + assert.ok(Date.now() - activeStartedAt >= 1200) + assert.equal(activeOutputs.filter((output) => output.includes("memory compose stale fallback")).length, 2) + const recoveryStartFile = join(directory, "recovery-start") + const recoveryReadyFiles = [join(directory, "recovery-one-ready"), join(directory, "recovery-two-ready")] + const recoveryWorkers = [ + runWorker("recovery-one", recorder, activeCount, "500", activeStateDirectory, recoveryStartFile, recoveryReadyFiles[0]), + runWorker("recovery-two", recorder, activeCount, "500", activeStateDirectory, recoveryStartFile, recoveryReadyFiles[1]), + ] + await waitForFiles(recoveryReadyFiles) + await writeFile(recoveryStartFile, "start") + const recoveryOutputs = await Promise.all(recoveryWorkers) + assert.equal((await readFile(activeCount, "utf8")).length, 1) + assert.equal(recoveryOutputs.filter((output) => output.includes("refreshed Data Machine memory")).length, 1) + assert.equal(recoveryOutputs.filter((output) => output.includes("reused fresh Data Machine memory")).length, 1, recoveryOutputs.join("\n")) + await rm(activeLeasePath, { recursive: true, force: true }) + + // A malformed state path is bounded fallback, never recursive reacquisition. + const errorStateDirectory = join(directory, "error-state") + await mkdir(errorStateDirectory) + await writeFile(join(errorStateDirectory, scopeFor(recorder)), "not a lease directory") + const errorStartedAt = Date.now() + const errorOutput = await runWorker("error", recorder, join(directory, "unexpected-error-count"), "25", errorStateDirectory) + assert.ok(Date.now() - errorStartedAt < 2000) + assert.ok(errorOutput.includes("memory compose stale fallback")) } await withEnv({ From f066aaa235b013b8a53de9e7b5b5ac16d46c47cf Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Mon, 14 Sep 2026 16:26:33 -0400 Subject: [PATCH 2/3] fix: coalesce explicit site across worktrees --- bridges/kimaki/plugins/dm-agent-sync.ts | 5 ++--- tests/dm-agent-sync.mjs | 14 +++++++++++--- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/bridges/kimaki/plugins/dm-agent-sync.ts b/bridges/kimaki/plugins/dm-agent-sync.ts index 7506dc2..fdfe5d0 100644 --- a/bridges/kimaki/plugins/dm-agent-sync.ts +++ b/bridges/kimaki/plugins/dm-agent-sync.ts @@ -101,11 +101,10 @@ async function composeMemory(wpCli: WpCli, sitePath: string, agentSlug: string): } function composeScope(wpCli: WpCli, sitePath: string, agentSlug: string): string { - // The command name alone is not enough: relative executables and WP-CLI - // configuration resolve from the runtime's working directory and identity. + // An explicit site identifies the target across worktrees. Keep transport + // identity separate so different WP-CLI configurations never share a lease. return createHash("sha256").update(JSON.stringify({ agentSlug, - cwd: process.cwd(), home: process.env.HOME || "", path: process.env.PATH || "", sitePath, diff --git a/tests/dm-agent-sync.mjs b/tests/dm-agent-sync.mjs index ade0c7d..9333dd2 100644 --- a/tests/dm-agent-sync.mjs +++ b/tests/dm-agent-sync.mjs @@ -110,8 +110,9 @@ if (process.env.DM_START_FILE) { await plugin["chat.message"]({ sessionID: process.env.DM_SESSION_ID }, {}) `) - const runWorker = (sessionID, executable = recorder, composeCount = count, timeout = "1000", stateDirectoryOverride = stateDirectory, startFile = "", readyFile = "") => new Promise((resolve, reject) => { + const runWorker = (sessionID, executable = recorder, composeCount = count, timeout = "1000", stateDirectoryOverride = stateDirectory, startFile = "", readyFile = "", workingDirectory) => new Promise((resolve, reject) => { const child = spawn(process.execPath, [worker], { + cwd: workingDirectory, env: { ...process.env, DATAMACHINE_SITE_PATH: sitePath, @@ -133,7 +134,15 @@ await plugin["chat.message"]({ sessionID: process.env.DM_SESSION_ID }, {}) child.on("close", (code) => code === 0 ? resolve(output) : reject(new Error(`worker exited ${code}: ${output}`))) }) - const outputs = await Promise.all([runWorker("one"), runWorker("two"), runWorker("three")]) + const worktreeOne = join(directory, "worktree-one") + const worktreeTwo = join(directory, "worktree-two") + await mkdir(worktreeOne) + await mkdir(worktreeTwo) + const outputs = await Promise.all([ + runWorker("one", recorder, count, "1000", stateDirectory, "", "", worktreeOne), + runWorker("two", recorder, count, "1000", stateDirectory, "", "", worktreeTwo), + runWorker("three", recorder, count, "1000", stateDirectory, "", "", worktreeOne), + ]) assert.equal((await readFile(count, "utf8")).length, 1) assert.equal(outputs.filter((output) => output.includes("refreshed Data Machine memory")).length, 1) assert.equal(outputs.filter((output) => output.includes("reused fresh Data Machine memory")).length, 2) @@ -156,7 +165,6 @@ exit 1 const scopeFor = (executable) => createHash("sha256").update(JSON.stringify({ agentSlug: "intelligence-chubes4", - cwd: process.cwd(), home: process.env.HOME || "", path: process.env.PATH || "", sitePath, From 08e128b387664d5c7c64b45f255019bc3468ed80 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Mon, 14 Sep 2026 16:34:22 -0400 Subject: [PATCH 3/3] fix: recover nested abandoned compose leases --- bridges/kimaki/plugins/dm-agent-sync.ts | 43 ++++++++++--------------- tests/dm-agent-sync.mjs | 11 ++++++- 2 files changed, 27 insertions(+), 27 deletions(-) diff --git a/bridges/kimaki/plugins/dm-agent-sync.ts b/bridges/kimaki/plugins/dm-agent-sync.ts index fdfe5d0..351f931 100644 --- a/bridges/kimaki/plugins/dm-agent-sync.ts +++ b/bridges/kimaki/plugins/dm-agent-sync.ts @@ -16,6 +16,7 @@ type WpCli = string[]; const DEFAULT_COMPOSE_TIMEOUT_MS = 10_000; const OUTPUT_LIMIT = 16 * 1024; +const MAX_RECOVERY_GENERATIONS = 3; type ComposeResult = { exitCode: number; output: string; timedOut: boolean }; type ComposeReceipt = { completedAt: number; operationId: string; result: "refreshed" | "stale_fallback" }; @@ -119,8 +120,8 @@ function composeStatePath(scope: string): string { return join(composeStateDirectory(), scope); } -function composeReceiptPath(scope: string): string { - return `${composeStatePath(scope)}.receipt`; +function composeReceiptPath(scope: string, operationId: string): string { + return `${composeStatePath(scope)}.receipt.${operationId}`; } function composeStateDirectory(): string { @@ -128,7 +129,7 @@ function composeStateDirectory(): string { } async function acquireComposeLease(scope: string, timeoutMs: number): Promise { - const path = composeStatePath(scope); + let path = composeStatePath(scope); const directory = composeStateDirectory(); try { await mkdir(directory, { recursive: true, mode: 0o700 }); @@ -136,29 +137,19 @@ async function acquireComposeLease(scope: string, timeoutMs: number): Promise Date.now()) { - return waitingLease(path, owner); - } + const owner = await waitForComposeOwner(path, timeoutMs); + if (!owner) return unavailableLease(timeoutMs); + if (owner.deadlineAt > Date.now()) return waitingLease(path, owner); - // Expired owners are never removed or renamed. A separate recovery lease - // makes recovery safe even when the old process wakes after its deadline. - const recoveryPath = `${path}.recovery.${owner.operationId}`; - const recovery = await createComposeLease(recoveryPath, timeoutMs); - if (recovery) return recovery; - const recoveryOwner = await waitForComposeOwner(recoveryPath, timeoutMs); - return recoveryOwner ? waitingLease(recoveryPath, recoveryOwner) : unavailableLease(timeoutMs); + // Expired owners are immutable. Each bounded generation is a distinct + // lease, so a late owner cannot release or replace a newer operation. + path = `${path}.recovery.${owner.operationId}`; + } + return unavailableLease(timeoutMs); } async function createComposeLease(path: string, timeoutMs: number): Promise { @@ -191,7 +182,7 @@ function waitingLease(path: string, owner: ComposeOwner): ComposeLease { async function waitForComposeReceipt(scope: string, operationId: string, deadlineAt: number): Promise { while (operationId && Date.now() < deadlineAt) { - const receipt = await readComposeReceipt(composeReceiptPath(scope)); + const receipt = await readComposeReceipt(composeReceiptPath(scope, operationId)); if (receipt?.operationId === operationId) { return receipt; } @@ -209,7 +200,7 @@ async function finishComposeLease(scope: string, lease: ComposeLease, result: Co } // Keep the receipt beside the lock: releasing the lock must not erase the // successful result before the other processes that joined it can read it. - await writeFile(composeReceiptPath(scope), JSON.stringify({ completedAt: Date.now(), operationId: owner.operationId, result }), { mode: 0o600 }); + await writeFile(composeReceiptPath(scope, owner.operationId), JSON.stringify({ completedAt: Date.now(), operationId: owner.operationId, result }), { mode: 0o600 }); await rm(path, { recursive: true, force: true }); } catch { // A best-effort lease failure must not prevent a chat message. diff --git a/tests/dm-agent-sync.mjs b/tests/dm-agent-sync.mjs index 9333dd2..3d2aa97 100644 --- a/tests/dm-agent-sync.mjs +++ b/tests/dm-agent-sync.mjs @@ -178,7 +178,7 @@ exit 1 const leasePath = join(stateDirectory, scope) await mkdir(leasePath, { recursive: true }) await writeFile(join(leasePath, "owner"), JSON.stringify(owner)) - if (receipt) await writeFile(`${leasePath}.receipt`, JSON.stringify(receipt)) + if (receipt) await writeFile(`${leasePath}.receipt.${receipt.operationId}`, JSON.stringify(receipt)) return leasePath } const waitForFiles = async (files) => { @@ -204,6 +204,13 @@ exit 1 operationId: staleOperation, result: "refreshed", }) + const abandonedRecoveryPath = `${staleLeasePath}.recovery.${staleOperation}` + await mkdir(abandonedRecoveryPath) + await writeFile(join(abandonedRecoveryPath, "owner"), JSON.stringify({ + deadlineAt: Date.now() - 1, + operationId: "abandoned-recovery-operation", + token: "abandoned-recovery-token", + })) const staleFailureCount = join(directory, "stale-failure-count") const staleOutputs = await Promise.all([ runWorker("stale-one", failingRecorder, staleFailureCount, "1000", staleStateDirectory), @@ -215,6 +222,8 @@ exit 1 assert.equal(staleOutputs.filter((output) => output.includes("reused fresh Data Machine memory")).length, 0) assert.equal(staleOutputs.filter((output) => output.includes("memory compose stale fallback")).length, 2) await access(staleLeasePath) + await access(abandonedRecoveryPath) + assert.equal(JSON.parse(await readFile(`${staleLeasePath}.receipt.${staleOperation}`, "utf8")).result, "refreshed") // Waiters use the owner's recorded deadline, not their shorter local timeout. const activeStateDirectory = join(directory, "active-state")