diff --git a/index.ts b/index.ts index 9f07d0a..1eef8e5 100644 --- a/index.ts +++ b/index.ts @@ -434,6 +434,35 @@ const STATUS_ZOMBIE_MS = 60 * 60_000 const STATUS_PATH = process.env.TAAS_AFFINITY_RUNS_STATUS_PATH || path.join(os.homedir(), ".openclaw", "alien-studio", "runs-status.json") +// ── Zombie auto-abort configuration ─────────────────────────────────────────── +// ⚠️ PLUGIN BLOCKED: No plugin-side dispatch for chat.abort exists in the OpenClaw +// plugin SDK. The default abortRun function logs a warning and is a no-op. +// When the SDK adds api.runtime.chat.abort() or dispatchGatewayMethod(), +// replace the default abortRun in register() below. +// Set TAAS_AFFINITY_AUTO_ABORT_ZOMBIES=true to enable (currently opt-in). +const AUTO_ABORT_ENABLED = process.env.TAAS_AFFINITY_AUTO_ABORT_ZOMBIES === "true" +const AUTO_ABORT_DRY_RUN = process.env.TAAS_AFFINITY_AUTO_ABORT_DRY_RUN === "true" +const AUTO_ABORT_THRESHOLD_MS = Number(process.env.TAAS_AFFINITY_AUTO_ABORT_THRESHOLD_MS) || STATUS_ZOMBIE_MS +const AUTO_ABORT_CHECK_INTERVAL_MS = Number(process.env.TAAS_AFFINITY_AUTO_ABORT_CHECK_INTERVAL_MS) || 60_000 +const AUTO_ABORT_ABORTED_SET_MAX = 1000 + +/** In-memory set of sessionKeys already aborted in this process (LRU-bounded). */ +const abortedInThisProcess: string[] = [] + +/** + * Injected abort function. Default logs a warning because the plugin SDK does not + * expose a way to dispatch chat.abort from inside a plugin. Replace via + * setAbortRunFn() in tests or when the SDK adds the capability. + */ +let abortRun: (sessionKey: string) => Promise = async (sessionKey: string) => { + console.warn("[taas-affinity] auto-abort: chat.abort not available via plugin SDK; sessionKey=" + sessionKey) +} + +/** Replace the abort function (for testing or future SDK integration). */ +function setAbortRunFn(fn: (sessionKey: string) => Promise): void { + abortRun = fn +} + function stopRequesterBridgePoller(leaseId?: string): void { if (!leaseId) return const state = activePollers.get(leaseId) @@ -894,6 +923,94 @@ function writeRunStatus(agentsDir?: string, statusPath?: string): void { } } +// ── Zombie auto-abort ────────────────────────────────────────────────────────── +let abortCheckInProgress = false + +function runAbortCheck(agentsDir?: string): void { + if (abortCheckInProgress) return + abortCheckInProgress = true + try { + const base = agentsDir || path.join(os.homedir(), ".openclaw", "agents") + let agents: string[] + try { agents = fs.readdirSync(base) } catch { agents = [] } + + const zombies: Array<{ sessionKey: string; idleMs: number }> = [] + + for (const agentId of agents) { + const sessionsDir = path.join(base, agentId, "sessions") + let entries: string[] + try { entries = fs.readdirSync(sessionsDir) } catch { continue } + + let mainKey: string | null = null + try { + const idJson = fs.readFileSync(path.join(base, agentId, "agent", "identity.json"), "utf8") + const parsed = JSON.parse(idJson) + if (typeof parsed.mainKey === "string") mainKey = parsed.mainKey + } catch { /* no identity file */ } + + let mainUuid: string | null = null + if (mainKey) { + const parts = mainKey.split(":") + const lastPart = parts[parts.length - 1] + if (lastPart !== "main" && lastPart.length >= 8) { + mainUuid = lastPart + } + } + + for (const name of entries) { + if (!name.endsWith(".jsonl.lock")) continue + const sessionUuid = name.replace(/\.jsonl\.lock$/, "") + const fp = path.join(sessionsDir, name) + let st: fs.Stats + try { st = fs.statSync(fp) } catch { continue } + + const idleMs = Date.now() - st.mtimeMs + if (idleMs < AUTO_ABORT_THRESHOLD_MS) continue + + let sessionKey: string + if (mainKey && sessionUuid === mainUuid) { + sessionKey = "agent:" + agentId + ":main" + } else { + sessionKey = "agent:" + agentId + ":session:" + sessionUuid + } + + zombies.push({ sessionKey, idleMs }) + } + } + + // When auto-abort is disabled (default), just log candidates + if (!AUTO_ABORT_ENABLED) { + if (zombies.length > 0) { + console.info("[taas-affinity] zombie candidates (auto-abort disabled): " + zombies.length + " runs: " + zombies.map(z => z.sessionKey).join(", ")) + } + return + } + + // Auto-abort is enabled — abort each zombie not already aborted in this process + for (const z of zombies) { + if (abortedInThisProcess.includes(z.sessionKey)) continue + // Cap the set at AUTO_ABORT_ABORTED_SET_MAX (FIFO eviction) + if (abortedInThisProcess.length >= AUTO_ABORT_ABORTED_SET_MAX) { + abortedInThisProcess.shift() + } + abortedInThisProcess.push(z.sessionKey) + + if (AUTO_ABORT_DRY_RUN) { + console.info("[taas-affinity] auto-abort DRY RUN: would abort sessionKey=" + z.sessionKey + " idleMs=" + z.idleMs) + } else { + console.warn("[taas-affinity] auto-aborting zombie run sessionKey=" + z.sessionKey + " idleMs=" + z.idleMs) + abortRun(z.sessionKey).catch((e: unknown) => { + console.warn("[taas-affinity] auto-abort failed sessionKey=" + z.sessionKey, e) + }) + } + } + } catch (e) { + console.warn("[taas-affinity] abort-check error", e) + } finally { + abortCheckInProgress = false + } +} + // ── Background task scheduler ──────────────────────────────────────────────── const backgroundTimers: (NodeJS.Timeout)[] = [] @@ -912,6 +1029,13 @@ function startBackgroundTasks(): void { backgroundTimers.push(setInterval(() => writeRunStatus(), STATUS_INTERVAL_MS)) }, 5_000) backgroundTimers.push(statusInit) + + // Zombie auto-abort — 10s initial delay, then every AUTO_ABORT_CHECK_INTERVAL_MS + const abortInit = setTimeout(() => { + runAbortCheck() + backgroundTimers.push(setInterval(() => runAbortCheck(), AUTO_ABORT_CHECK_INTERVAL_MS)) + }, 10_000) + backgroundTimers.push(abortInit) } const LAST_ROUTE_LIMIT = 256 @@ -1161,7 +1285,12 @@ export default { _testExports: { runTrashSweep, writeRunStatus, + runAbortCheck, + setAbortRunFn, resetSweepInProgress: () => { sweepInProgress = false }, resetStatusInProgress: () => { statusInProgress = false }, + resetAbortCheckInProgress: () => { abortCheckInProgress = false }, + clearAbortedInThisProcess: () => { abortedInThisProcess.length = 0 }, + getAbortedInThisProcess: () => [...abortedInThisProcess], }, } diff --git a/test/auto-abort.test.ts b/test/auto-abort.test.ts new file mode 100644 index 0000000..c04af0d --- /dev/null +++ b/test/auto-abort.test.ts @@ -0,0 +1,229 @@ +import assert from "node:assert/strict" +import fs from "node:fs" +import os from "node:os" +import path from "node:path" +import { test } from "node:test" + +// Helper: dynamic import with cache bust +async function loadPlugin(env: Record = {}) { + const oldEnv: Record = {} + for (const [key, value] of Object.entries(env)) { + oldEnv[key] = process.env[key] + if (value === undefined) delete process.env[key] + else process.env[key] = value + } + const mod = await import(`../index.ts?bust=${Date.now()}-${Math.random()}`) + for (const [key, value] of Object.entries(oldEnv)) { + if (value === undefined) delete process.env[key] + else process.env[key] = value + } + return mod +} + +// Helper: create a temp agents dir with zombie lock files +function createTempAgentsDir(structure: Array<{ + agentId: string + sessions: Array<{ uuid: string; mtimeAgeMs: number; pid?: number }> + identityJson?: Record +}>) { + const base = fs.mkdtempSync(path.join(os.tmpdir(), "taas-abort-test-")) + for (const { agentId, sessions = [], identityJson } of structure) { + const sessDir = path.join(base, agentId, "sessions") + fs.mkdirSync(sessDir, { recursive: true }) + for (const s of sessions) { + const lockPath = path.join(sessDir, s.uuid + ".jsonl.lock") + fs.writeFileSync(lockPath, s.pid != null ? String(s.pid) : "12345") + const newMtime = new Date(Date.now() - s.mtimeAgeMs) + fs.utimesSync(lockPath, newMtime, newMtime) + } + if (identityJson) { + const agentDir = path.join(base, agentId, "agent") + fs.mkdirSync(agentDir, { recursive: true }) + fs.writeFileSync(path.join(agentDir, "identity.json"), JSON.stringify(identityJson)) + } + } + return base +} + +// Collect console output +function collectConsole() { + const infoLogs: string[] = [] + const warnLogs: string[] = [] + const origInfo = console.info + const origWarn = console.warn + console.info = (...args: unknown[]) => { infoLogs.push(args.map(String).join(" ")) } + console.warn = (...args: unknown[]) => { warnLogs.push(args.map(String).join(" ")) } + return { + infoLogs, + warnLogs, + restore() { + console.info = origInfo + console.warn = origWarn + }, + } +} + +// AC-AUTO-ABORT.1: With auto-abort enabled, classify one zombie session, mock the abort callback, assert it was called with the right sessionKey. +test("AC-AUTO-ABORT.1: auto-abort enabled calls abortRun for zombie session", async () => { + const mod = await loadPlugin({ TAAS_AFFINITY_AUTO_ABORT_ZOMBIES: "true" }) + const { runAbortCheck, setAbortRunFn, resetAbortCheckInProgress, clearAbortedInThisProcess } = mod.default._testExports + + clearAbortedInThisProcess() + resetAbortCheckInProgress() + + const aborted: Array<{ sessionKey: string }> = [] + setAbortRunFn(async (sessionKey: string) => { + aborted.push({ sessionKey }) + }) + + // Create a zombie: lock file older than 60 min + const agentsDir = createTempAgentsDir([{ + agentId: "test-agent", + sessions: [{ uuid: "abc12345-6789-def0-1234-567890abcdef", mtimeAgeMs: 65 * 60 * 1000 }], + }]) + + const logs = collectConsole() + try { + runAbortCheck(agentsDir) + } finally { + logs.restore() + } + + assert.ok(aborted.length === 1, `Expected 1 abort, got ${aborted.length}`) + assert.match(aborted[0].sessionKey, /^agent:test-agent:session:abc12345/) +}) + +// AC-AUTO-ABORT.2: With auto-abort disabled (default), classify zombies, assert the abort callback was NOT called and one log line listing candidates. +test("AC-AUTO-ABORT.2: auto-abort disabled logs candidates without aborting", async () => { + const mod = await loadPlugin({ TAAS_AFFINITY_AUTO_ABORT_ZOMBIES: undefined }) + const { runAbortCheck, setAbortRunFn, resetAbortCheckInProgress, clearAbortedInThisProcess } = mod.default._testExports + + clearAbortedInThisProcess() + resetAbortCheckInProgress() + + const aborted: string[] = [] + setAbortRunFn(async (sessionKey: string) => { + aborted.push(sessionKey) + }) + + // Create two zombie lock files + const agentsDir = createTempAgentsDir([{ + agentId: "main", + sessions: [ + { uuid: "deadbeef-0000-0000-0000-000000000001", mtimeAgeMs: 70 * 60 * 1000 }, + { uuid: "deadbeef-0000-0000-0000-000000000002", mtimeAgeMs: 65 * 60 * 1000 }, + ], + }]) + + const logs = collectConsole() + try { + runAbortCheck(agentsDir) + } finally { + logs.restore() + } + + assert.equal(aborted.length, 0, "abortRun should NOT be called when auto-abort is disabled") + // Check that we logged the candidates + const candidateLog = logs.infoLogs.find(l => l.includes("zombie candidates (auto-abort disabled)")) + assert.ok(candidateLog, `Expected candidate log, got: ${JSON.stringify(logs.infoLogs)}`) + assert.ok(candidateLog!.includes("2 runs"), `Expected '2 runs' in log, got: ${candidateLog}`) +}) + +// AC-AUTO-ABORT.3: Same zombie classified twice in successive ticks → abort callback called ONLY ONCE (idempotence via the in-memory set). +test("AC-AUTO-ABORT.3: idempotent — same zombie only aborted once across two ticks", async () => { + const mod = await loadPlugin({ TAAS_AFFINITY_AUTO_ABORT_ZOMBIES: "true" }) + const { runAbortCheck, setAbortRunFn, resetAbortCheckInProgress, clearAbortedInThisProcess } = mod.default._testExports + + clearAbortedInThisProcess() + resetAbortCheckInProgress() + + const aborted: string[] = [] + setAbortRunFn(async (sessionKey: string) => { + aborted.push(sessionKey) + }) + + const agentsDir = createTempAgentsDir([{ + agentId: "test-agent", + sessions: [{ uuid: "idempotent-0000-0000-000000000001", mtimeAgeMs: 66 * 60 * 1000 }], + }]) + + // First tick + resetAbortCheckInProgress() + runAbortCheck(agentsDir) + assert.equal(aborted.length, 1, "First tick should abort once") + + // Second tick — same zombie is still there + resetAbortCheckInProgress() + runAbortCheck(agentsDir) + assert.equal(aborted.length, 1, "Second tick should NOT abort again — idempotent") +}) + +// AC-AUTO-ABORT.4: 1001 distinct zombies all aborted → the in-memory set caps at 1000 entries. +test("AC-AUTO-ABORT.4: aborted set caps at 1000 entries (LRU eviction)", async () => { + const mod = await loadPlugin({ TAAS_AFFINITY_AUTO_ABORT_ZOMBIES: "true" }) + const { runAbortCheck, setAbortRunFn, resetAbortCheckInProgress, clearAbortedInThisProcess, getAbortedInThisProcess } = mod.default._testExports + + clearAbortedInThisProcess() + resetAbortCheckInProgress() + + const aborted: string[] = [] + setAbortRunFn(async (sessionKey: string) => { + aborted.push(sessionKey) + }) + + // Create 1100 zombie lock files (more than the 1000 cap) + const sessions = [] + for (let i = 0; i < 1100; i++) { + sessions.push({ + uuid: `zombie-${String(i).padStart(8, "0")}-0000-000000000000`, + mtimeAgeMs: (61 + (i % 10)) * 60 * 1000, // 61-70 min idle + }) + } + const agentsDir = createTempAgentsDir([{ + agentId: "load-agent", + sessions, + }]) + + resetAbortCheckInProgress() + runAbortCheck(agentsDir) + + // All 1100 should have been passed to abortRun (they're new each time) + assert.equal(aborted.length, 1100, `Expected 1100 aborted calls, got ${aborted.length}`) + + // The in-memory set should be capped at 1000 + const set = getAbortedInThisProcess() + assert.ok(set.length <= 1000 && set.length > 0, `Expected set <= 1000, got ${set.length}`) +}) + +// Cleanup temp dirs (node:test handles most, but ensure) +test("AC-AUTO-ABORT.5: dry-run mode logs but does not call abortRun", async () => { + const mod = await loadPlugin({ + TAAS_AFFINITY_AUTO_ABORT_ZOMBIES: "true", + TAAS_AFFINITY_AUTO_ABORT_DRY_RUN: "true", + }) + const { runAbortCheck, setAbortRunFn, resetAbortCheckInProgress, clearAbortedInThisProcess } = mod.default._testExports + + clearAbortedInThisProcess() + resetAbortCheckInProgress() + + const aborted: string[] = [] + setAbortRunFn(async (sessionKey: string) => { + aborted.push(sessionKey) + }) + + const agentsDir = createTempAgentsDir([{ + agentId: "dry-agent", + sessions: [{ uuid: "dryrun-test-0000-0000-000000000001", mtimeAgeMs: 65 * 60 * 1000 }], + }]) + + const logs = collectConsole() + try { + runAbortCheck(agentsDir) + } finally { + logs.restore() + } + + assert.equal(aborted.length, 0, "dry-run should NOT call abortRun") + const dryLog = logs.infoLogs.find(l => l.includes("auto-abort DRY RUN")) + assert.ok(dryLog, `Expected DRY RUN log, got: ${JSON.stringify(logs.infoLogs)}`) +})