diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b690f0b6..0309f459 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -96,3 +96,22 @@ jobs: - name: Exercise packaged offline chat with verified provenance run: xvfb-run -a nub run test:e2e:release:packaged-core + + windows_provider_acceptance: + runs-on: windows-latest + timeout-minutes: 15 + steps: + - name: Check out repository + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + persist-credentials: false + - name: Set up Nub and Node + uses: nubjs/setup-nub@47e5e7393d50f4e9544ddaf756aa277972afba2d # v0 + with: + nub-version: 0.9.7 + node-version-file: .node-version + cache-dependency-path: nub.lock + - name: Install dependencies + run: nub install --frozen-lockfile --ignore-scripts + - name: Verify provider process trees and capture budgets on Windows + run: nub run test:scope tests/boundary/provider-command.test.ts diff --git a/apps/desktop/src/main/bot-automation-broker.ts b/apps/desktop/src/main/bot-automation-broker.ts index 0cef3f2d..670880f3 100644 --- a/apps/desktop/src/main/bot-automation-broker.ts +++ b/apps/desktop/src/main/bot-automation-broker.ts @@ -484,12 +484,12 @@ export class BotAutomationBroker { ), ); const job = await this.job(fire.jobId); - this.check(job, this.ledger.approval(fire.approvalId)); + const fireApproval = this.ledger.approval(fire.approvalId); + if (!fireApproval) + throw new Error("The original automation approval is unavailable."); + this.check(job, fireApproval); await this.bots.validateAutomationConnection(fire.targetBotId); - this.check( - await this.job(fire.jobId), - this.ledger.approval(fire.approvalId), - ); + this.check(await this.job(fire.jobId), fireApproval); signal?.throwIfAborted(); if ( ["cancelled", "error"].includes( diff --git a/apps/desktop/src/main/bot-automation-history.ts b/apps/desktop/src/main/bot-automation-history.ts new file mode 100644 index 00000000..9e64c676 --- /dev/null +++ b/apps/desktop/src/main/bot-automation-history.ts @@ -0,0 +1,248 @@ +import { + closeSync, + constants, + existsSync, + fchmodSync, + fstatSync, + lstatSync, + mkdirSync, + openSync, +} from "node:fs"; +import { dirname } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { isDeepStrictEqual } from "node:util"; +import type { + AutomationTargetApproval, + AutomationTargetFire, +} from "./bot-automation-ledger"; + +/** Cold receipts remain indexed on disk for permanent duplicate prevention. */ +export class BotAutomationHistory { + private required: boolean; + private allowInitializeEmpty = false; + constructor(private readonly path: string) { + this.required = existsSync(path); + } + + setRecoveryRequirement(required: boolean): void { + this.required ||= required; + this.allowInitializeEmpty = !required; + } + + exists(): boolean { + const exists = existsSync(this.path); + if (this.required && !exists) + throw new Error("Automation history is missing. Recovery is required."); + return exists; + } + + private open(): DatabaseSync { + mkdirSync(dirname(this.path), { recursive: true, mode: 0o700 }); + let fd: number; + const flags = + constants.O_RDWR | + (constants.O_NOFOLLOW ?? 0) | + (constants.O_NONBLOCK ?? 0); + try { + fd = openSync(this.path, "wx+", 0o600); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + fd = openSync(this.path, flags); + } + try { + if ( + !fstatSync(fd).isFile() || + (!constants.O_NOFOLLOW && lstatSync(this.path).isSymbolicLink()) + ) { + throw new Error("Automation history is not a regular file."); + } + fchmodSync(fd, 0o600); + } finally { + closeSync(fd); + } + const db = new DatabaseSync(this.path); + try { + const tables = db + .prepare( + "SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%'", + ) + .all(); + if (!tables.length && this.allowInitializeEmpty) { + db.exec("BEGIN IMMEDIATE"); + try { + db.exec(`CREATE TABLE fires ( + fire_id TEXT PRIMARY KEY, job_id TEXT NOT NULL, digest TEXT NOT NULL, + run_id TEXT NOT NULL UNIQUE, record TEXT NOT NULL, + UNIQUE(job_id, digest) + ); CREATE TABLE approvals (id TEXT PRIMARY KEY, record TEXT NOT NULL)`); + db.exec("COMMIT"); + } catch (error) { + db.exec("ROLLBACK"); + throw error; + } + } + db.prepare( + "SELECT fire_id, job_id, digest, run_id, record FROM fires LIMIT 0", + ).all(); + db.prepare("SELECT id, record FROM approvals LIMIT 0").all(); + this.required = true; + this.allowInitializeEmpty = false; + return db; + } catch (error) { + db.close(); + throw error; + } + } + + private decode( + row: Record | undefined, + ): AutomationTargetFire | undefined { + if (!row) return undefined; + const fire = JSON.parse(String(row.record)) as AutomationTargetFire; + if ( + fire.fireId !== row.fire_id || + fire.jobId !== row.job_id || + fire.fireKeyDigest !== row.digest || + fire.runId !== row.run_id || + !["complete", "cancelled", "error"].includes(fire.status) + ) { + throw new Error("Automation history receipt is invalid."); + } + return fire; + } + + find( + field: "fire_id" | "run_id" | "job_id", + id: string, + digest?: string, + ): AutomationTargetFire | undefined { + if (!this.exists()) return undefined; + const db = this.open(); + try { + return this.decode( + field === "job_id" + ? db + .prepare("SELECT * FROM fires WHERE job_id = ? AND digest = ?") + .get(id, digest ?? "") + : db.prepare(`SELECT * FROM fires WHERE ${field} = ?`).get(id), + ); + } finally { + db.close(); + } + } + + approval(id: string): AutomationTargetApproval | undefined { + if (!this.exists()) return undefined; + const db = this.open(); + try { + const row = db + .prepare("SELECT record FROM approvals WHERE id = ?") + .get(id); + if (!row) return undefined; + const approval = JSON.parse( + String(row.record), + ) as AutomationTargetApproval; + if (approval.id !== id) + throw new Error("Automation approval history is invalid."); + return approval; + } finally { + db.close(); + } + } + + /** An archive commit may precede an interrupted hot-ledger publication. */ + reconcile( + fires: AutomationTargetFire[], + approvals: AutomationTargetApproval[], + ): AutomationTargetFire[] { + if (!this.exists()) return fires; + const db = this.open(); + try { + const lookupApproval = db.prepare( + "SELECT record FROM approvals WHERE id = ?", + ); + for (const approval of approvals) { + const archived = lookupApproval.get(approval.id); + if ( + archived && + !isDeepStrictEqual(JSON.parse(String(archived.record)), approval) + ) + throw new Error("Automation approval histories disagree."); + } + const lookup = db.prepare( + "SELECT * FROM fires WHERE fire_id = ? OR (job_id = ? AND digest = ?)", + ); + return fires.filter((fire) => { + const matches = lookup.all(fire.fireId, fire.jobId, fire.fireKeyDigest); + for (const row of matches) { + const archived = this.decode(row); + if (!archived) + throw new Error("Automation history receipt is unavailable."); + const { status: _oldStatus, updatedAt: _oldTime, ...identity } = fire; + const { + status: _status, + updatedAt: _time, + ...archivedIdentity + } = archived; + if (!isDeepStrictEqual(identity, archivedIdentity)) + throw new Error("Automation history identities disagree."); + if ( + ["complete", "cancelled", "error"].includes(fire.status) && + fire.status !== archived.status + ) + throw new Error("Automation history terminal statuses disagree."); + } + return matches.length === 0; + }); + } finally { + db.close(); + } + } + + retire( + fires: AutomationTargetFire[], + approvals: AutomationTargetApproval[], + ): void { + if (!fires.length && !approvals.length) return; + const db = this.open(); + try { + db.exec("BEGIN IMMEDIATE"); + const lookup = db.prepare("SELECT * FROM fires WHERE fire_id = ?"); + const insert = db.prepare("INSERT INTO fires VALUES (?, ?, ?, ?, ?)"); + for (const fire of fires) { + const previous = this.decode(lookup.get(fire.fireId)); + if (previous) { + if (!isDeepStrictEqual(previous, fire)) + throw new Error("Automation history receipts disagree."); + continue; + } + if (!["complete", "cancelled", "error"].includes(fire.status)) + throw new Error("Active automation cannot be archived."); + insert.run( + fire.fireId, + fire.jobId, + fire.fireKeyDigest, + fire.runId, + JSON.stringify(fire), + ); + } + const lookupApproval = db.prepare( + "SELECT record FROM approvals WHERE id = ?", + ); + const insertApproval = db.prepare("INSERT INTO approvals VALUES (?, ?)"); + for (const approval of approvals) { + const previous = lookupApproval.get(approval.id); + if (previous) { + if (!isDeepStrictEqual(JSON.parse(String(previous.record)), approval)) + throw new Error("Automation approval histories disagree."); + } else insertApproval.run(approval.id, JSON.stringify(approval)); + } + db.exec("COMMIT"); + } catch (error) { + db.exec("ROLLBACK"); + throw error; + } finally { + db.close(); + } + } +} diff --git a/apps/desktop/src/main/bot-automation-ledger.ts b/apps/desktop/src/main/bot-automation-ledger.ts index 00a7b32f..19397738 100644 --- a/apps/desktop/src/main/bot-automation-ledger.ts +++ b/apps/desktop/src/main/bot-automation-ledger.ts @@ -2,6 +2,7 @@ import { chmodSync, existsSync, mkdirSync } from "node:fs"; import { isAbsolute, resolve } from "node:path"; import type { BotPermissions } from "@doolittle/contracts/bots"; import { writeJsonAtomicSync } from "@elizaos/agent/utils/atomic-json"; +import { BotAutomationHistory } from "./bot-automation-history"; import { readBoundedRegularFile } from "./regular-file"; export interface AutomationTargetApproval { @@ -29,6 +30,7 @@ export interface AutomationTargetFire { } interface Stored { version: 1; + historyRequired?: true; revision: number; approvals: AutomationTargetApproval[]; fires: AutomationTargetFire[]; @@ -47,6 +49,7 @@ function parse(path: string): Stored | null { const value = JSON.parse(raw) as Stored; if ( value?.version !== 1 || + (value.historyRequired !== undefined && value.historyRequired !== true) || !Number.isSafeInteger(value.revision) || value.revision < 0 || !Array.isArray(value.approvals) || @@ -120,7 +123,11 @@ export class BotAutomationLedger { private readonly backup: string; private stored: Stored; private failed = false; + private readonly history: BotAutomationHistory; constructor(dataDir: string) { + this.history = new BotAutomationHistory( + resolve(dataDir, "bots", "automation-history.sqlite"), + ); this.path = resolve(dataDir, "bots", "automation-dispatches.json"); this.backup = resolve(dataDir, "bots", "automation-dispatches.backup.json"); let primary: Stored | null = null; @@ -152,11 +159,47 @@ export class BotAutomationLedger { primary && (!secondary || primary.revision >= secondary.revision) ? primary : (secondary ?? { version: 1, revision: 0, approvals: [], fires: [] }); + this.history.setRecoveryRequirement(this.stored.historyRequired === true); + if (this.stored.historyRequired && !this.history.exists()) + throw new Error("Automation history is missing. Recovery is required."); + this.stored.fires = this.history.reconcile( + this.stored.fires, + this.stored.approvals, + ); } private save(next: Stored): void { if (this.failed) throw new Error("Automation dispatch ledger requires recovery."); - const value = { ...next, revision: this.stored.revision + 1 }; + const terminal = next.fires + .filter((fire) => + ["complete", "cancelled", "error"].includes(fire.status), + ) + .sort((a, b) => Date.parse(b.updatedAt) - Date.parse(a.updatedAt)); + const retired = terminal.slice(256); + const retiredIds = new Set(retired.map((fire) => fire.fireId)); + const fires = next.fires.filter((fire) => !retiredIds.has(fire.fireId)); + const referenced = new Set(fires.map((fire) => fire.approvalId)); + const approvals = [...referenced].map((id) => { + const approval = + next.approvals.find((row) => row.id === id) ?? + this.history.approval(id); + if (!approval) + throw new Error("The original automation approval is unavailable."); + return approval; + }); + const retainedApprovals = new Set(approvals.map((approval) => approval.id)); + const archivedApprovals = next.approvals.filter( + (approval) => !retainedApprovals.has(approval.id), + ); + const value: Stored = { + ...next, + revision: this.stored.revision + 1, + fires, + ...(this.history.exists() || retired.length || archivedApprovals.length + ? { historyRequired: true } + : {}), + approvals, + }; if ( Buffer.byteLength(JSON.stringify(value, null, 2)) + 1 > 8_000_000 || value.approvals.length > 2_000 || @@ -166,6 +209,8 @@ export class BotAutomationLedger { "Automation dispatch ledger is full. No run was submitted.", ); try { + // Commit immutable replay receipts before removing them from either hot copy. + this.history.retire(retired, archivedApprovals); mkdirSync(resolve(this.path, ".."), { recursive: true, mode: 0o700 }); writeJsonAtomicSync(this.path, value, { trailingNewline: true }); chmodSync(this.path, 0o600); @@ -180,7 +225,10 @@ export class BotAutomationLedger { approval(id: string): AutomationTargetApproval | undefined { if (this.failed) throw new Error("Automation dispatch ledger requires recovery."); - return structuredClone(this.stored.approvals.find((row) => row.id === id)); + return structuredClone( + this.stored.approvals.find((row) => row.id === id) ?? + this.history.approval(id), + ); } approve(row: AutomationTargetApproval): void { this.save({ @@ -191,15 +239,27 @@ export class BotAutomationLedger { get(id: string): AutomationTargetFire | undefined { if (this.failed) throw new Error("Automation dispatch ledger requires recovery."); - return structuredClone(this.stored.fires.find((row) => row.fireId === id)); + return structuredClone( + this.stored.fires.find((row) => row.fireId === id) ?? + this.history.find("fire_id", id), + ); } forKey(jobId: string, digest: string): AutomationTargetFire | undefined { - return this.list().find( - (row) => row.jobId === jobId && row.fireKeyDigest === digest, + if (this.failed) + throw new Error("Automation dispatch ledger requires recovery."); + return structuredClone( + this.stored.fires.find( + (row) => row.jobId === jobId && row.fireKeyDigest === digest, + ) ?? this.history.find("job_id", jobId, digest), ); } byRun(runId: string): AutomationTargetFire | undefined { - return this.list().find((row) => row.runId === runId); + if (this.failed) + throw new Error("Automation dispatch ledger requires recovery."); + return structuredClone( + this.stored.fires.find((row) => row.runId === runId) ?? + this.history.find("run_id", runId), + ); } list(): AutomationTargetFire[] { if (this.failed) diff --git a/docs/engineering/automation-history.md b/docs/engineering/automation-history.md new file mode 100644 index 00000000..26becf97 --- /dev/null +++ b/docs/engineering/automation-history.md @@ -0,0 +1,36 @@ +# Desktop automation history + +ElizaOS owns scheduling. The desktop records target-bot dispatch identities before +submitting runs, so retrying a scheduled fire cannot submit it twice. + +The hot `bots/automation-dispatches.json` ledger and its backup keep active work +and the 256 most recently updated terminal receipts. Active work is never retired. +The existing limits of 5,000 hot fires, 2,000 hot approvals and 8 MB still refuse +excess simultaneous work rather than dropping it. + +Completed, cancelled and failed receipts move into the private +`bots/automation-history.sqlite` archive. Indexed lookups by fire, run, and job/fire +identity retain duplicate protection across restarts without reading or rewriting +the entire history. Only approvals referenced by hot fires stay in the hot ledger; all other +approvals remain available in the archive, including current unused approvals. Result +retrieval checks the original approval, never substitutes the current approval. + +Archive inserts commit in one SQLite transaction before either hot JSON copy is +replaced. Restart reconciles overlapping records left by an interrupted write. +Both archive tables are created in one transaction. An interrupted, empty first +archive can be initialized only before the ledger requires history; an empty +required archive is never reset. Identity conflicts fail closed. A ledger that requires an archive refuses new +submissions if the archive is missing; a corrupt or symlinked archive also fails +closed. Existing version-one ledgers migrate on their next successful write. + +Cold history is retained on disk indefinitely. It has no fixed row-count expiry; +this preserves the existing replay contract for arbitrary SDK fire identities. +Disk usage grows with actual dispatch history. This change bounds the frequently +rewritten state and keeps historical lookups indexed; it does not promise +unlimited storage or remove limits on concurrent active work. + +For a consistent manual backup, stop the desktop/runtime and copy the complete +runtime `bots` state directory, including both JSON copies and the SQLite archive. +Do not delete or reset the archive to resolve a startup error: doing so discards +the evidence needed to prevent duplicate work. Restore a consistent backup or +investigate the storage failure. diff --git a/docs/engineering/codebase-review-2026-10-25.md b/docs/engineering/codebase-review-2026-10-25.md index 949ae26b..fe9bbee2 100644 --- a/docs/engineering/codebase-review-2026-10-25.md +++ b/docs/engineering/codebase-review-2026-10-25.md @@ -2,6 +2,8 @@ **What this repo does:** Doolittle supplies desktop and terminal clients for an ElizaOS agent, including coding tools, conversations, background work, account connections, and wallet/trading tools. Both clients share the engine and application services. This review assumes one local operator, several bots and concurrent sessions, and recurring jobs left running for weeks. +> Follow-up: the four findings below are addressed in the [October 26 reliability changes](review-findings-follow-up-2026-10-26.md). The assessment below records the original review, including its verification limits. + ## Assessment This is a substantial, functioning alpha with real monorepo boundaries. It is beyond a prototype, but its current evidence does not establish a dependable replacement for a mature coding assistant across long sessions and every supported platform. The next work should strengthen recovery and long-running operation before adding more surface area. diff --git a/docs/engineering/review-findings-follow-up-2026-10-26.md b/docs/engineering/review-findings-follow-up-2026-10-26.md new file mode 100644 index 00000000..f138e931 --- /dev/null +++ b/docs/engineering/review-findings-follow-up-2026-10-26.md @@ -0,0 +1,72 @@ +# Reliability review follow-up — October 26, 2026 + +This follows the [October 25 codebase review](codebase-review-2026-10-25.md). +All four findings have implementation or regression-test coverage in this change. +ElizaOS remains the engine for desktop and CLI; wallet/trading tools remain installed. + +## 1. Durable automation retention + +Terminal receipts move into indexed SQLite history, leaving active work and the +256 newest terminal receipts in the hot ledger. Historical fire/run lookup and +duplicate prevention survive restart. Unreferenced approvals are archived too, avoiding a lifetime job-approval cap; +retrieving an old result checks its original authorization. + +Archive commits precede hot-ledger publication. Restart reconciles overlaps from +interrupted writes. Schema creation is transactional; empty first-creation +recovery is allowed only before persisted history is required. Missing, corrupt or conflicting required history refuses +new work instead of forgetting a fire identity. Tests exercise the former +5,000-record capacity, restart/replay, interrupted publication, original approval +lookup, and unavailable history. See [automation history](automation-history.md) +for storage limits and backup requirements. Cold history grows on disk; this is +not an unlimited-storage claim or a multi-day soak result. + +## 2. Conversation recovery + +The real compression command and native rollback helpers now have durable +failure-injection tests. The native-memory adapter is a controlled SQLite-backed +fixture; the projection uses the production SQLite message store and schema. + +Seven journeys cover summary rejection before and after commit, partial native +deletion followed by rejection, a real SQLite trigger aborting projection insert, +failed compensation, successful compression, and cleanup of a failed replay +without deleting an unrelated message. The tests reopen both stores and inspect +persisted state. Failed compensation remains an explicit error, not a claimed +successful recovery. These are focused contract tests, not a complete Eliza SQL +adapter certification. + +## 3. Bounded provider output + +The shared provider subprocess runner caps combined stdout/stderr at 8 MiB by +default. Overflow stops the owned process and returns a nonzero `output_limit` +result. Truncated responses cannot become successful model answers. Tests cover +each stream separately, their combined budget, UTF-8, and invalid budgets. + +## 4. Windows process trees and cancellation + +Windows cancellation, deadlines and output overflow use the system `taskkill.exe` +with `/T /F`; POSIX uses the owned process group. Cleanup has a bounded Windows +termination attempt, and retained pipes cannot hold the request open forever. +Failed termination is reported explicitly and has a failure-injection test. +The executable comes from a fixed local-drive Windows system-directory allowlist; +nonstandard Windows system directories fail explicitly. Natural exit racing cleanup is handled +separately. Devin now forwards the SDK cancellation signal to its subprocess. + +The process tests start a real launcher and a worker that holds inherited pipes, +then verify both stop on cancellation and deadline. A dedicated Windows PR job +runs these tests natively; Linux and macOS run the POSIX path. This is process +lifecycle control, not a security sandbox for hostile programs that deliberately +escape their process ancestry. If Windows tree termination fails, the request +reports that failure rather than claiming descendants stopped. + +## Verification boundaries + +The fast boundary suite, all workspace types, lint, architecture checks and +zero-production-advisory policy passed locally. Native source/desktop acceptance, +real linked-provider coding in each client, Windows process acceptance and +clean-HEAD packaged acceptance are distinct checks in the delivery workflow. +Their run results belong in the PR and completion report. A passing small coding +fixture does not establish long-task quality or reliability over weeks. + +Seven dependency patches and three SDK compatibility workspaces remain. This +change addresses the four review findings; it does not remove those upstream +compatibility requirements. diff --git a/packages/plugins/plugin-devin/src/cli.ts b/packages/plugins/plugin-devin/src/cli.ts index b5df8cef..4f1fde28 100644 --- a/packages/plugins/plugin-devin/src/cli.ts +++ b/packages/plugins/plugin-devin/src/cli.ts @@ -32,6 +32,7 @@ async function runDevinProcess( NO_COLOR: "1", }, timeoutMs, + signal: params.signal, }); const cleanStdout = stripAnsi(result.stdout).trim(); const cleanStderr = stripAnsi(result.stderr).trim(); diff --git a/packages/plugins/plugin-devin/src/generate.ts b/packages/plugins/plugin-devin/src/generate.ts index 0da8f79d..5943be39 100644 --- a/packages/plugins/plugin-devin/src/generate.ts +++ b/packages/plugins/plugin-devin/src/generate.ts @@ -55,6 +55,7 @@ export async function runDevinTextGeneration( cwd: options.getCwd?.() ?? options.cwd, timeoutMs: options.timeoutMs ?? DEFAULT_DEVIN_TIMEOUT_MS, permissionMode: "auto", + signal: params.signal, }); runtime.logger?.info( { diff --git a/packages/plugins/plugin-devin/src/types.ts b/packages/plugins/plugin-devin/src/types.ts index c26fe523..5d4a1106 100644 --- a/packages/plugins/plugin-devin/src/types.ts +++ b/packages/plugins/plugin-devin/src/types.ts @@ -17,6 +17,7 @@ export interface DevinCliPrintParams { command?: string; cwd?: string; timeoutMs?: number; + signal?: AbortSignal; permissionMode?: "auto" | "dangerous"; } diff --git a/packages/plugins/provider-transport/README.md b/packages/plugins/provider-transport/README.md index 0f6492ed..1f21a69b 100644 --- a/packages/plugins/provider-transport/README.md +++ b/packages/plugins/provider-transport/README.md @@ -35,5 +35,17 @@ const result = await runProviderCommand({ ``` The runner does not invoke a shell. It preserves stdout and stderr separately, -applies Eliza core's environment policy, and terminates the process group on a -deadline or caller cancellation. +applies Eliza core's environment policy, and terminates the POSIX process group +or Windows process tree on a deadline, caller cancellation, or excessive output. +The combined stdout/stderr budget defaults to 8 MiB (`maxOutputBytes` can select +1 byte through 64 MiB). Overflow returns exit code 125 and `output_limit`, never +successful truncated output. The bounded diagnostic can add up to 256 bytes to +the captured output budget. + +Windows termination uses the system `taskkill.exe` with `/T /F` and a five-second +termination deadline. Its executable is selected from fixed `Windows/System32` +locations on local drives, rather than executing an environment-supplied path. +Nonstandard Windows system directories fail explicitly. Failed tree termination is reported as `termination_error` +(exit code 126), or rejects cancellation explicitly. Retained descendant pipes +cannot keep the request pending indefinitely; failure is not reported as a +successful cancellation. diff --git a/packages/plugins/provider-transport/src/command.ts b/packages/plugins/provider-transport/src/command.ts index 1bd609e2..41bb3592 100644 --- a/packages/plugins/provider-transport/src/command.ts +++ b/packages/plugins/provider-transport/src/command.ts @@ -1,9 +1,15 @@ -import { type ChildProcess, spawn } from "node:child_process"; +import { type ChildProcess, execFile, spawn } from "node:child_process"; +import { win32 } from "node:path"; import process from "node:process"; import { StringDecoder } from "node:string_decoder"; import { sanitizeSpawnEnv } from "@elizaos/core"; -export type ProviderCommandTermination = "exit" | "spawn_error" | "timeout"; +export type ProviderCommandTermination = + | "exit" + | "spawn_error" + | "timeout" + | "output_limit" + | "termination_error"; export interface ProviderCommandRequest { command: string; @@ -12,6 +18,8 @@ export interface ProviderCommandRequest { env?: Record; timeoutMs?: number; signal?: AbortSignal; + /** Combined stdout/stderr capture budget; defaults to 8 MiB. */ + maxOutputBytes?: number; } export interface ProviderCommandResult { @@ -43,19 +51,48 @@ function sanitizedEnvironment( }; } -function killProcessTree(child: ChildProcess): void { - try { - if (process.platform !== "win32" && child.pid !== undefined) { - process.kill(-child.pid, "SIGKILL"); - return; - } - } catch { - // The process may have exited between the close check and the kill. +async function killProcessTree(child: ChildProcess): Promise { + if (child.pid === undefined) return; + if (process.platform === "win32") { + const configuredRoot = + process.env.SystemRoot?.trim() || process.env.WINDIR?.trim(); + // Select a fixed system-tool location, never execute an environment-supplied path. + const root = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + .split("") + .map((drive) => `${drive}:\\Windows`) + .find( + (candidate) => + candidate.toLowerCase() === + win32 + .normalize(configuredRoot ?? "") + .replace(/\\$/u, "") + .toLowerCase(), + ); + if (!root) + throw new Error( + "Windows SystemRoot must use the standard Windows directory on a local drive.", + ); + await new Promise((resolve, reject) => { + execFile( + win32.join(root, "System32", "taskkill.exe"), + ["/PID", String(child.pid), "/T", "/F"], + { timeout: 5000, maxBuffer: 16 * 1024, windowsHide: true }, + (error) => + error + ? reject( + new Error("Windows provider process-tree termination failed.", { + cause: error, + }), + ) + : resolve(), + ); + }); + return; } try { - child.kill("SIGKILL"); - } catch { - // The process is already gone. + process.kill(-child.pid, "SIGKILL"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } } @@ -84,6 +121,16 @@ export async function runProviderCommand( throw commandAbortReason(request.signal); } + const maxOutputBytes = request.maxOutputBytes ?? 8 * 1024 * 1024; + if ( + !Number.isSafeInteger(maxOutputBytes) || + maxOutputBytes <= 0 || + maxOutputBytes > 64 * 1024 * 1024 + ) { + throw new RangeError( + "Provider command maxOutputBytes must be an integer between 1 and 64 MiB.", + ); + } const startedAt = Date.now(); return await new Promise((resolve, reject) => { let child: ChildProcess; @@ -111,8 +158,13 @@ export async function runProviderCommand( let stdout = ""; let stderr = ""; let settled = false; + let closed = false; let timedOut = false; let cancelled = false; + let outputExceeded = false; + let outputBytes = 0; + let stopping: Promise | undefined; + let terminationFailure = false; const cleanup = () => { clearTimeout(timer); @@ -122,22 +174,55 @@ export async function runProviderCommand( stdout += stdoutDecoder.end(); stderr += stderrDecoder.end(); }; + const stop = () => { + if (stopping || settled) return; + stopping = killProcessTree(child) + .catch(() => { + // A short command may close naturally while Windows taskkill starts. + // Closed stdio needs no further drain; do not mislabel output overflow. + if (closed) return; + terminationFailure = true; + // Best effort fallback; report that descendant termination was not verified. + try { + child.kill("SIGKILL"); + } catch { + /* termination failure is reported below */ + } + }) + .finally(() => { + // A descendant retaining a pipe must not keep this request pending forever. + child.stdout?.destroy(); + child.stderr?.destroy(); + finish(null); + }); + }; const onAbort = () => { cancelled = true; - killProcessTree(child); + stop(); }; const timer = setTimeout(() => { timedOut = true; - killProcessTree(child); + stop(); }, timeoutMs); timer.unref?.(); request.signal?.addEventListener("abort", onAbort, { once: true }); + const capture = (chunk: Buffer, decoder: StringDecoder): string => { + if (outputExceeded) return ""; + const remaining = maxOutputBytes - outputBytes; + const accepted = chunk.subarray(0, remaining); + outputBytes += accepted.byteLength; + if (chunk.byteLength > remaining) { + outputExceeded = true; + stop(); + } + return decoder.write(accepted); + }; child.stdout?.on("data", (chunk: Buffer) => { - stdout += stdoutDecoder.write(chunk); + stdout += capture(chunk, stdoutDecoder); }); child.stderr?.on("data", (chunk: Buffer) => { - stderr += stderrDecoder.write(chunk); + stderr += capture(chunk, stderrDecoder); }); child.once("error", (error) => { if (settled) return; @@ -152,15 +237,43 @@ export async function runProviderCommand( termination: "spawn_error", }); }); - child.once("close", (code) => { + const finish = (code: number | null) => { if (settled) return; settled = true; cleanup(); flushOutput(); + if (terminationFailure) { + if (cancelled) { + reject( + new Error( + "Provider cancellation could not terminate its process tree.", + ), + ); + return; + } + resolve({ + exitCode: 126, + stdout, + stderr: `${stderr}\n[provider-transport] process-tree termination failed`, + durationMs: Date.now() - startedAt, + termination: "termination_error", + }); + return; + } if (cancelled && request.signal) { reject(commandAbortReason(request.signal)); return; } + if (outputExceeded) { + resolve({ + exitCode: 125, + stdout, + stderr: `${stderr}\n[provider-transport] combined output exceeded ${maxOutputBytes} bytes`, + durationMs: Date.now() - startedAt, + termination: "output_limit", + }); + return; + } if (timedOut) { const marker = `[provider-transport] command timed out after ${timeoutMs}ms`; resolve({ @@ -179,6 +292,11 @@ export async function runProviderCommand( durationMs: Date.now() - startedAt, termination: "exit", }); + }; + child.once("close", (code) => { + closed = true; + if (!stopping) finish(code); }); + if (request.signal?.aborted) onAbort(); }); } diff --git a/scripts/vitest-scope.ts b/scripts/vitest-scope.ts index f0f8106e..226ad88d 100644 --- a/scripts/vitest-scope.ts +++ b/scripts/vitest-scope.ts @@ -1,4 +1,6 @@ import { spawnSync } from "node:child_process"; +import { createRequire } from "node:module"; +import { dirname, resolve } from "node:path"; // Nub forwards a literal `--` to Vitest, where it can hide every file filter. // Guard only argument forwarding; selection and execution remain Vitest's job. @@ -16,8 +18,16 @@ if ( } const result = spawnSync( - "nubx", - ["vitest", mode, ...(mode === "related" ? ["--run"] : []), ...filters], + process.execPath, + [ + resolve( + dirname(createRequire(import.meta.url).resolve("vitest/package.json")), + "vitest.mjs", + ), + mode, + ...(mode === "related" ? ["--run"] : []), + ...filters, + ], { stdio: "inherit" }, ); if (result.error) console.error(result.error.message); diff --git a/tests/boundary/automation-retention.test.ts b/tests/boundary/automation-retention.test.ts new file mode 100644 index 00000000..fb11fd88 --- /dev/null +++ b/tests/boundary/automation-retention.test.ts @@ -0,0 +1,254 @@ +import { randomUUID } from "node:crypto"; +import { + mkdirSync, + mkdtempSync, + readFileSync, + renameSync, + rmSync, + symlinkSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { + type AutomationTargetFire, + BotAutomationLedger, +} from "../../apps/desktop/src/main/bot-automation-ledger"; + +const dirs: string[] = []; +afterEach(() => { + for (const dir of dirs.splice(0)) + rmSync(dir, { recursive: true, force: true }); +}); +function fixture() { + const dir = mkdtempSync(resolve(tmpdir(), "doolittle-automation-retention-")); + dirs.push(dir); + mkdirSync(resolve(dir, "bots")); + const jobId = randomUUID(), + targetBotId = randomUUID(), + approvalId = randomUUID(); + const stamp = "2026-10-25T00:00:00.000Z"; + const approval = { + id: approvalId, + jobId, + targetBotId, + targetRevision: stamp, + definitionDigest: "a".repeat(64), + workspacePath: resolve(dir), + createdAt: stamp, + permissions: { + connectionIds: [], + workspacePaths: [], + toolIds: [], + allowMutation: false, + allowDelegation: false, + }, + }; + function fire( + index: number, + status: AutomationTargetFire["status"] = "complete", + ): AutomationTargetFire { + const fireId = randomUUID(); + return { + fireId, + fireKeyDigest: index.toString(16).padStart(64, "0"), + jobId, + targetBotId, + approvalId, + sessionId: `automation:${jobId}:${fireId}`, + runId: `automation:${fireId}`, + status, + createdAt: stamp, + updatedAt: stamp, + }; + } + const fires = Array.from({ length: 5000 }, (_, index) => fire(index)); + fires[0] = fire(0, "accepted"); + const raw = JSON.stringify({ + version: 1, + revision: 0, + approvals: [approval], + fires, + }); + const path = resolve(dir, "bots/automation-dispatches.json"); + const backup = resolve(dir, "bots/automation-dispatches.backup.json"); + writeFileSync(path, raw); + writeFileSync(backup, raw); + return { dir, path, backup, fires, fire, jobId, approval }; +} +it("runs beyond the old capacity and retains active work and permanent replay identities across restart", () => { + const f = fixture(); + let ledger = new BotAutomationLedger(f.dir); + for (let index = 5000; index < 5020; index++) { + const next = f.fire(index, "prepared"); + ledger.prepare(next); + ledger.update(next.fireId, "complete"); + } + ledger = new BotAutomationLedger(f.dir); + expect(ledger.list().length).toBeLessThanOrEqual(257); + expect(ledger.get(f.fires[0].fireId)?.status).toBe("accepted"); + const archived = f.fires[4999]; + expect(ledger.forKey(f.jobId, archived.fireKeyDigest)).toEqual(archived); + expect(ledger.byRun(archived.runId)).toEqual(archived); + expect(() => ledger.prepare(archived)).toThrow(/already exists/); + expect(readFileSync(f.path).byteLength).toBeLessThan(200_000); +}); +it("recovers an archive committed before an interrupted hot-ledger publication", () => { + const f = fixture(); + const ledger = new BotAutomationLedger(f.dir); + renameSync(f.path, `${f.path}.old`); + mkdirSync(f.path); + expect(() => ledger.prepare(f.fire(6000, "prepared"))).toThrow(); + expect(() => ledger.list()).toThrow(/recovery/); + const restored = new BotAutomationLedger(f.dir); + expect(restored.get(f.fires[0].fireId)).toEqual(f.fires[0]); + expect(restored.forKey(f.jobId, f.fires[4999].fireKeyDigest)).toEqual( + f.fires[4999], + ); + expect(() => restored.prepare(f.fires[4999])).toThrow(/already exists/); +}); +it("refuses a missing archive instead of forgetting old replay identities", () => { + const f = fixture(); + const ledger = new BotAutomationLedger(f.dir); + ledger.prepare(f.fire(6000, "prepared")); + rmSync(resolve(f.dir, "bots/automation-history.sqlite")); + expect(() => ledger.forKey(f.jobId, f.fires[4999].fireKeyDigest)).toThrow( + /history is missing/, + ); + expect(() => new BotAutomationLedger(f.dir)).toThrow(/history is missing/); +}); + +it("archives superseded approvals without substituting the current authorization", () => { + const f = fixture(); + const ledger = new BotAutomationLedger(f.dir); + ledger.prepare(f.fire(6000, "prepared")); + const next = { + ...f.approval, + id: randomUUID(), + definitionDigest: "b".repeat(64), + }; + ledger.approve(next); + // The old active run still pins its original approval in the hot ledger. + expect(ledger.approval(f.approval.id)).toEqual(f.approval); + ledger.update(f.fires[0].fireId, "complete"); + // A fresh job without fires can retire its previous approval immediately. + const other = { ...f.approval, id: randomUUID(), jobId: randomUUID() }; + ledger.approve(other); + ledger.approve({ + ...other, + id: randomUUID(), + definitionDigest: "c".repeat(64), + }); + const restored = new BotAutomationLedger(f.dir); + expect(restored.approval(other.id)).toEqual(other); + expect( + JSON.parse(readFileSync(f.path, "utf8")).approvals.some( + (row: { id: string }) => row.id === other.id, + ), + ).toBe(false); +}); + +it("refuses corrupt cold history instead of treating archived fires as unsubmitted", () => { + const f = fixture(); + const ledger = new BotAutomationLedger(f.dir); + ledger.prepare(f.fire(6000, "prepared")); + writeFileSync( + resolve(f.dir, "bots/automation-history.sqlite"), + "not a database", + ); + expect(() => ledger.forKey(f.jobId, f.fires[4999].fireKeyDigest)).toThrow(); + expect(() => new BotAutomationLedger(f.dir)).toThrow(); +}); + +it("rejects a symlinked archive without touching the target", () => { + const f = fixture(); + const ledger = new BotAutomationLedger(f.dir); + ledger.prepare(f.fire(6000, "prepared")); + const archive = resolve(f.dir, "bots/automation-history.sqlite"); + const outside = resolve(f.dir, "outside"); + writeFileSync(outside, "preserved"); + rmSync(archive); + symlinkSync(outside, archive); + expect(() => ledger.forKey(f.jobId, f.fires[4999].fireKeyDigest)).toThrow(); + expect(readFileSync(outside, "utf8")).toBe("preserved"); +}); + +for (const kind of ["zero-byte", "empty-sqlite"]) { + it(`recovers an interrupted first archive creation (${kind}) before history is required`, async () => { + const f = fixture(); + const archive = resolve(f.dir, "bots/automation-history.sqlite"); + if (kind === "zero-byte") writeFileSync(archive, ""); + else { + const { DatabaseSync } = await import("node:sqlite"); + const db = new DatabaseSync(archive); + db.exec("PRAGMA user_version = 0"); + db.close(); + } + const ledger = new BotAutomationLedger(f.dir); + ledger.prepare(f.fire(6000, "prepared")); + expect( + new BotAutomationLedger(f.dir).forKey( + f.jobId, + f.fires[4999].fireKeyDigest, + ), + ).toEqual(f.fires[4999]); + }); +} +it("refuses to initialize an empty archive once persisted history is required", () => { + const f = fixture(); + const ledger = new BotAutomationLedger(f.dir); + ledger.prepare(f.fire(6000, "prepared")); + const archive = resolve(f.dir, "bots/automation-history.sqlite"); + writeFileSync(archive, ""); + expect(() => ledger.forKey(f.jobId, f.fires[4999].fireKeyDigest)).toThrow(); + expect(() => new BotAutomationLedger(f.dir)).toThrow(); + expect(readFileSync(archive).byteLength).toBe(0); +}); + +it("refuses a hot receipt that conflicts with an archived job/fire identity", () => { + const f = fixture(); + new BotAutomationLedger(f.dir).prepare(f.fire(6000, "prepared")); + const hot = JSON.parse(readFileSync(f.path, "utf8")); + const fireId = randomUUID(); + hot.fires.push({ + ...f.fires[4999], + fireId, + sessionId: `automation:${f.jobId}:${fireId}`, + runId: `automation:${fireId}`, + }); + for (const path of [f.path, f.backup]) + writeFileSync(path, JSON.stringify(hot)); + expect(() => new BotAutomationLedger(f.dir)).toThrow(/identities disagree/); +}); + +it("admits a new job beyond the old lifetime approval cap without losing previous consent", () => { + const f = fixture(); + const unused = Array.from({ length: 1999 }, () => ({ + ...f.approval, + id: randomUUID(), + jobId: randomUUID(), + })); + const stored = JSON.parse(readFileSync(f.path, "utf8")); + stored.approvals.push(...unused); + for (const path of [f.path, f.backup]) + writeFileSync(path, JSON.stringify(stored)); + const ledger = new BotAutomationLedger(f.dir); + const added = { ...f.approval, id: randomUUID(), jobId: randomUUID() }; + ledger.approve(added); + const restored = new BotAutomationLedger(f.dir); + expect(restored.approval(added.id)).toEqual(added); + expect(restored.approval(unused[1998].id)).toEqual(unused[1998]); + expect(restored.approval(f.approval.id)).toEqual(f.approval); + expect(JSON.parse(readFileSync(f.path, "utf8")).approvals).toHaveLength(1); + const started = { + ...f.fire(7000, "prepared"), + jobId: added.jobId, + approvalId: added.id, + }; + started.sessionId = `automation:${added.jobId}:${started.fireId}`; + restored.prepare(started); + const active = new BotAutomationLedger(f.dir); + expect(active.get(started.fireId)).toEqual(started); + expect(active.approval(added.id)).toEqual(added); +}); diff --git a/tests/boundary/conversation-recovery.test.ts b/tests/boundary/conversation-recovery.test.ts new file mode 100644 index 00000000..3be55035 --- /dev/null +++ b/tests/boundary/conversation-recovery.test.ts @@ -0,0 +1,259 @@ +import { randomUUID } from "node:crypto"; +import { EventEmitter } from "node:events"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import type { Memory, UUID } from "@elizaos/core"; +import { afterEach, expect, it } from "vitest"; +import type { AgentExecutionContext } from "../../packages/agent/src/runtime/chat"; +import { rollbackNativeConversationReplay } from "../../packages/agent/src/runtime/chat-turn/conversation-persistence"; +import { handleSessionCommand } from "../../packages/agent/src/runtime/commands/session-router"; +import { SessionMessageStore } from "../../packages/agent/src/services/session/messages"; +import { createSessionDatabase } from "../../packages/agent/src/services/session/service/support"; +import type { StoredMessage } from "../../packages/agent/src/types"; + +const cleanup: Array<() => void> = []; +afterEach(() => { + for (const close of cleanup.splice(0).reverse()) close(); +}); +type Failure = + | "none" + | "summary" + | "summary-after-commit" + | "partial-delete" + | "projection" + | "rollback"; +function fixture(failure: Failure) { + const dir = mkdtempSync( + resolve(tmpdir(), "doolittle-conversation-recovery-"), + ); + cleanup.push(() => rmSync(dir, { recursive: true, force: true })); + let native = new DatabaseSync(resolve(dir, "native.sqlite")); + let projection = createSessionDatabase(dir); + cleanup.push(() => { + native.close(); + projection.close(); + }); + native.exec( + "CREATE TABLE memories (id TEXT PRIMARY KEY, room_id TEXT NOT NULL, payload TEXT NOT NULL)", + ); + let store = new SessionMessageStore(projection, new EventEmitter()); + const roomId = randomUUID(), + agentId = randomUUID(), + entityId = randomUUID(); + const sessionId = "cli:recovery"; + const messages: StoredMessage[] = Array.from({ length: 7 }, (_, index) => ({ + id: randomUUID(), + sessionId, + roomId, + entityId, + role: index % 2 ? "assistant" : "user", + text: `Original message ${index}`, + createdAt: new Date(Date.UTC(2026, 9, 25, 0, 0, index)).toISOString(), + })); + for (const message of messages) { + store.storeMessage(message); + const memory = { + id: message.id, + roomId, + entityId, + agentId, + content: { text: message.text }, + metadata: { doolittle: { role: message.role } }, + createdAt: Date.parse(message.createdAt), + }; + native + .prepare("INSERT INTO memories VALUES (?, ?, ?)") + .run(message.id, roomId, JSON.stringify(memory)); + } + if (failure === "projection" || failure === "rollback") + projection.exec(` + CREATE TRIGGER fail_summary BEFORE INSERT ON messages + WHEN NEW.text LIKE '[CONTEXT SUMMARY%' + BEGIN SELECT RAISE(ABORT, 'forced projection failure'); END; + `); + let summaryFailed = false, + deletionFailed = false, + deleted = 0; + const runtime = { + agentId, + async getMemories(input: { + roomId: string; + limit: number; + offset: number; + }) { + return native + .prepare( + "SELECT payload FROM memories WHERE room_id = ? ORDER BY json_extract(payload, '$.createdAt') DESC LIMIT ? OFFSET ?", + ) + .all(input.roomId, input.limit, input.offset) + .map((row) => JSON.parse(String(row.payload)) as Memory); + }, + async createMemory(memory: Memory) { + const summary = memory.content.text?.startsWith("[CONTEXT SUMMARY"); + if (summary && failure === "summary" && !summaryFailed) { + summaryFailed = true; + throw new Error("forced summary failure"); + } + native + .prepare( + "INSERT INTO memories VALUES (?, ?, ?) ON CONFLICT(id) DO NOTHING", + ) + .run(String(memory.id), String(memory.roomId), JSON.stringify(memory)); + if (summary && failure === "summary-after-commit" && !summaryFailed) { + summaryFailed = true; + throw new Error("forced summary acknowledgement failure"); + } + }, + async deleteMemory(id: string) { + const row = native + .prepare("SELECT payload FROM memories WHERE id = ?") + .get(id); + const memory = row + ? (JSON.parse(String(row.payload)) as Memory) + : undefined; + if ( + failure === "rollback" && + memory?.content.text?.startsWith("[CONTEXT SUMMARY") + ) + throw new Error("forced compensation failure"); + native.prepare("DELETE FROM memories WHERE id = ?").run(id); + deleted++; + if (failure === "partial-delete" && deleted === 2 && !deletionFailed) { + deletionFailed = true; + throw new Error("forced deletion acknowledgement failure"); + } + }, + }; + const context = { + runtime, + services: { + sessions: { + messagesBySession: store.messagesBySession.bind(store), + replaceSessionMessages: store.replaceSessionMessages.bind(store), + continuityKey: () => sessionId, + }, + contextCompression: { + measure: (rows: unknown[]) => ({ estimatedTokens: rows.length * 100 }), + }, + trajectoryEvaluation: { recordEvent: () => undefined }, + }, + } as unknown as AgentExecutionContext; + const nativeRows = () => + native + .prepare("SELECT payload FROM memories ORDER BY id") + .all() + .map((row) => JSON.parse(String(row.payload)) as Memory); + const originals = nativeRows(); + return { + messages, + originals, + nativeRows, + async compress() { + return handleSessionCommand( + { message: "/compress", source: "cli", userId: entityId }, + "/compress", + sessionId, + context, + { + runAnalysis: async () => "Durable summary", + executeDelegationTask: async () => { + throw new Error("Delegation is outside this compression fixture."); + }, + }, + ); + }, + async rollbackReplay() { + const replayId = randomUUID() as UUID; + const seed = originals[0], + second = originals[1]; + if (!seed || !second) throw new Error("Missing replay seed"); + const concurrent: Memory = { + ...seed, + id: randomUUID() as UUID, + content: { text: "Unrelated concurrent message" }, + }; + native + .prepare("DELETE FROM memories WHERE id IN (?, ?)") + .run(String(seed.id), String(second.id)); + await runtime.createMemory({ + ...seed, + id: replayId, + content: { text: "Failed replay" }, + }); + await runtime.createMemory({ + ...seed, + id: randomUUID() as UUID, + content: { text: "Partial replay answer", inReplyTo: replayId }, + }); + await runtime.createMemory(concurrent); + await rollbackNativeConversationReplay({ + context, + roomId: roomId as UUID, + replayMessageId: replayId, + originalSnapshots: originals.slice(0, 2), + }); + return concurrent; + }, + restart() { + native.close(); + projection.close(); + native = new DatabaseSync(resolve(dir, "native.sqlite")); + projection = createSessionDatabase(dir); + store = new SessionMessageStore(projection, new EventEmitter()); + return { + native: nativeRows(), + projection: store.messagesBySession(sessionId, 100), + }; + }, + }; +} +for (const failure of [ + "summary", + "summary-after-commit", + "partial-delete", + "projection", +] as const) { + it(`preserves both durable transcripts after ${failure} failure and restart`, async () => { + const f = fixture(failure); + await expect(f.compress()).rejects.toThrow(); + const restored = f.restart(); + expect(restored.native).toEqual(f.originals); + expect(restored.projection).toEqual(f.messages); + }); +} +it("reports failed compensation instead of claiming compression succeeded", async () => { + const f = fixture("rollback"); + await expect(f.compress()).rejects.toThrow(/native context rollback failed/); + const restored = f.restart(); + expect(restored.projection).toEqual(f.messages); + expect( + restored.native.some((row) => + row.content.text?.includes("CONTEXT SUMMARY"), + ), + ).toBe(true); +}); +it("publishes the same successful compressed history to both stores", async () => { + const f = fixture("none"); + await expect(f.compress()).resolves.toContain("Context compressed"); + const restored = f.restart(); + expect(restored.projection).toHaveLength(5); + expect(restored.native).toHaveLength(5); + expect(restored.native.map((row) => row.id).sort()).toEqual( + restored.projection.map((row) => row.id).sort(), + ); + expect(restored.projection[1].text).toContain("Durable summary"); +}); + +it("removes only a failed replay and restores its original exchange across restart", async () => { + const f = fixture("none"); + const concurrent = await f.rollbackReplay(); + const restored = f.restart(); + expect(restored.native).toEqual( + [...f.originals, concurrent].sort((a, b) => + String(a.id).localeCompare(String(b.id)), + ), + ); + expect(restored.projection).toEqual(f.messages); +}); diff --git a/tests/boundary/provider-command.test.ts b/tests/boundary/provider-command.test.ts new file mode 100644 index 00000000..a3451b6a --- /dev/null +++ b/tests/boundary/provider-command.test.ts @@ -0,0 +1,193 @@ +import { expect, it, vi } from "vitest"; +import { runProviderCommand } from "../../packages/plugins/provider-transport/src/command"; + +it("preserves separate UTF-8 streams below the capture budget", async () => { + const result = await runProviderCommand({ + command: process.execPath, + args: [ + "-e", + "process.stdout.write('héllo');process.stderr.write('diagnostic')", + ], + maxOutputBytes: 4096, + }); + expect(result).toMatchObject({ + exitCode: 0, + termination: "exit", + stdout: "héllo", + stderr: "diagnostic", + }); +}); +for (const stream of ["stdout", "stderr"] as const) { + it(`terminates excessive ${stream} without retaining an unbounded response`, async () => { + const result = await runProviderCommand({ + command: process.execPath, + args: [ + "-e", + `setInterval(() => process.${stream}.write('é'.repeat(16_384)), 1)`, + ], + maxOutputBytes: 4096, + timeoutMs: 5000, + }); + expect(result.termination).toBe("output_limit"); + expect(result.exitCode).not.toBe(0); + expect(Buffer.byteLength(result.stdout + result.stderr)).toBeLessThan(4352); + expect(result.stderr).toContain("combined output exceeded"); + }); +} +it("counts both output streams against one shared budget", async () => { + const result = await runProviderCommand({ + command: process.execPath, + args: [ + "-e", + "process.stdout.write('a'.repeat(3000));process.stderr.write('b'.repeat(3000))", + ], + maxOutputBytes: 4096, + }); + expect(result.termination).toBe("output_limit"); + expect(Buffer.byteLength(result.stdout + result.stderr)).toBeLessThan(4352); +}); +it("rejects invalid capture budgets before starting a command", async () => { + await expect( + runProviderCommand({ + command: process.execPath, + args: [], + maxOutputBytes: Infinity, + }), + ).rejects.toThrow(/maxOutputBytes/); +}); + +for (const mode of ["cancel", "timeout"] as const) { + it(`stops the provider launcher and its pipe-holding worker on ${mode}`, async () => { + const { mkdtempSync, readFileSync, existsSync, rmSync } = await import( + "node:fs" + ); + const { tmpdir } = await import("node:os"); + const { resolve } = await import("node:path"); + const { spawnSync } = await import("node:child_process"); + const root = mkdtempSync(resolve(tmpdir(), "doolittle-provider-tree-")); + const pidFile = resolve(root, "worker.pid"); + const controller = new AbortController(); + const worker = + "require('node:fs').writeFileSync(process.argv[1], String(process.pid));setInterval(() => process.stdout.write('.'), 25)"; + const launcher = + "const {spawn}=require('node:child_process');require('node:fs').writeFileSync(process.argv[1]+'.root',String(process.pid));spawn(process.execPath,['-e',process.argv[2],process.argv[1]],{stdio:['ignore','inherit','inherit']});setInterval(()=>{},1000)"; + const result = runProviderCommand({ + command: process.execPath, + args: ["-e", launcher, pidFile, worker], + signal: controller.signal, + timeoutMs: mode === "timeout" ? 3000 : 10_000, + }).then( + (value) => ({ value, error: undefined }), + (error: Error) => ({ value: undefined, error }), + ); + const pids: number[] = []; + const running = (pid: number) => { + try { + process.kill(pid, 0); + } catch { + return false; + } + if (process.platform === "win32") return true; + // Orphan zombies cannot execute or hold pipes, but may await the CI init reaper. + const status = spawnSync("ps", ["-o", "stat=", "-p", String(pid)], { + encoding: "utf8", + }).stdout.trim(); + return Boolean(status) && !status.startsWith("Z"); + }; + try { + await expect + .poll(() => existsSync(pidFile), { timeout: 8000 }) + .toBe(true); + pids.push( + Number(readFileSync(`${pidFile}.root`, "utf8")), + Number(readFileSync(pidFile, "utf8")), + ); + if (mode === "cancel") controller.abort(); + const ended = await result; + if (mode === "cancel") expect(ended.error?.name).toBe("AbortError"); + else expect(ended.value?.termination).toBe("timeout"); + for (const pid of pids) + await expect.poll(() => running(pid), { timeout: 3000 }).toBe(false); + } finally { + controller.abort(); + await result; + for (const path of [`${pidFile}.root`, pidFile]) { + if (!existsSync(path)) continue; + const pid = Number(readFileSync(path, "utf8")); + if (Number.isSafeInteger(pid) && pid > 0 && running(pid)) { + try { + process.kill(pid, "SIGKILL"); + } catch { + /* already exited */ + } + } + } + rmSync(root, { recursive: true, force: true }); + } + }); +} + +it("cancels a Devin model request through its linked provider transport", async () => { + const { runDevinTextGeneration } = await import( + "../../packages/plugins/plugin-devin/src/generate" + ); + const controller = new AbortController(); + const request = runDevinTextGeneration( + { + getSetting: () => JSON.stringify({ model: { provider: "devin" } }), + } as unknown as import("@elizaos/core").IAgentRuntime, + { prompt: "Cancellation fixture", signal: controller.signal }, + { + getStatus: () => ({ + provider: "devin", + available: true, + reusable: true, + detail: "fixture", + }), + invokeCliPrint: async (params) => { + const result = await runProviderCommand({ + command: process.execPath, + args: ["-e", "setInterval(()=>{}, 1000)"], + signal: params.signal, + timeoutMs: 3000, + }); + return result.stdout; + }, + }, + ); + const assertion = expect(request).rejects.toMatchObject({ + code: "cancelled", + }); + controller.abort(); + await assertion; +}); + +it("reports an unsuccessful tree termination instead of claiming cancellation succeeded", async () => { + const controller = new AbortController(); + const previousRoot = process.env.SystemRoot; + const kill = + process.platform === "win32" + ? undefined + : vi.spyOn(process, "kill").mockImplementation(() => { + throw Object.assign(new Error("forced termination failure"), { + code: "EPERM", + }); + }); + try { + const request = runProviderCommand({ + command: process.execPath, + args: ["-e", "setInterval(()=>{},1000)"], + signal: controller.signal, + }); + const rejected = expect(request).rejects.toThrow( + "could not terminate its process tree", + ); + if (process.platform === "win32") process.env.SystemRoot = "invalid"; + controller.abort(); + await rejected; + } finally { + kill?.mockRestore(); + if (previousRoot === undefined) delete process.env.SystemRoot; + else process.env.SystemRoot = previousRoot; + } +});