mirror of
https://github.com/openclaw/openclaw.git
synced 2026-10-03 09:39:25 +00:00
fix(test): Telegram skill follow-up control and binding checkpoint tests fail on loaded hosts (#163326)
Both cases raced wall-clock deadlines that were not the behavior under test. The follow-up control test bounded its child with a 5 s spawnSync timeout and a 2 s polling budget. The binding checkpoint test ran the product checkpoint on the real clock, so its 90 s readiness and 60 s verification budgets, which bound live runs, failed routing tests with VERIFICATION_DEADLINE when the host stalled. Sampling stuck children showed them blocked in kernel rename() for up to 194 s. Children now run through an async helper bounded by a 90 s node:test timeout, with one after hook per test that kills each process group, joins every child, and only then removes the temp root. The follow-up child waits for the product's status sequence and names the step it is waiting for. The checkpoint child freezes performance.now() through a data: preload, and its three cases run concurrently so the file stays inside the harness's 120 s budget. Assertions are unchanged.
This commit is contained in:
Peter Steinberger
GitHub
parent
be5b7292d8
commit
b26dde75dc
2 changed files with 215 additions and 73 deletions
|
|
@ -1,14 +1,77 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import { spawn } from "node:child_process";
|
||||
import path from "node:path";
|
||||
import test from "node:test";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../../test/helpers/temp-dir.ts";
|
||||
|
||||
test("holds one real callback invocation and releases it once", () => {
|
||||
const root = fs.mkdtempSync(path.join(os.tmpdir(), "followup-control-test-"));
|
||||
try {
|
||||
// Leave time inside the harness's 120 s budget for after hooks to join children.
|
||||
const TEST_TIMEOUT_MS = 90_000;
|
||||
|
||||
function run(context, children, command, args, options) {
|
||||
context.signal.throwIfAborted();
|
||||
const child = spawn(command, args, {
|
||||
...options,
|
||||
detached: true,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
child.stdout.setEncoding("utf8").on("data", (chunk) => {
|
||||
stdout += chunk;
|
||||
});
|
||||
child.stderr.setEncoding("utf8").on("data", (chunk) => {
|
||||
stderr += chunk;
|
||||
});
|
||||
const closed = new Promise((resolve) => {
|
||||
child.once("close", (status, signal) => resolve({ status, signal, stdout, stderr }));
|
||||
});
|
||||
children.add({ child, closed });
|
||||
return new Promise((resolve, reject) => {
|
||||
const abort = () => {
|
||||
const error = new Error(
|
||||
`Child did not finish before the test ended:\n${stderr.slice(-4000)}`,
|
||||
{
|
||||
cause: context.signal.reason,
|
||||
},
|
||||
);
|
||||
context.diagnostic(error.message);
|
||||
reject(error);
|
||||
};
|
||||
context.signal.addEventListener("abort", abort, { once: true });
|
||||
child.once("error", reject);
|
||||
closed.then(resolve).finally(() => context.signal.removeEventListener("abort", abort));
|
||||
});
|
||||
}
|
||||
|
||||
// A descendant can hold the pipes after its parent exits, so always kill the group.
|
||||
async function stopChildren(children) {
|
||||
for (const { child } of children) {
|
||||
if (child.pid) {
|
||||
try {
|
||||
process.kill(-child.pid, "SIGKILL");
|
||||
} catch (error) {
|
||||
if (error.code !== "ESRCH" && error.code !== "EPERM") {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
await Promise.all([...children].map(({ closed }) => closed));
|
||||
}
|
||||
|
||||
test(
|
||||
"holds one real callback invocation and releases it once",
|
||||
{ timeout: TEST_TIMEOUT_MS },
|
||||
async (context) => {
|
||||
const children = new Set();
|
||||
const temporary = useAutoCleanupTempDirTracker((cleanup) =>
|
||||
context.after(async () => {
|
||||
await stopChildren(children);
|
||||
cleanup();
|
||||
}),
|
||||
);
|
||||
const root = temporary.make("followup-control-test-");
|
||||
const commandPath = path.join(root, "command.json");
|
||||
const statusPath = path.join(root, "status.json");
|
||||
const preload = fileURLToPath(new URL("./followup-drain-control-preload.mjs", import.meta.url));
|
||||
|
|
@ -18,14 +81,13 @@ test("holds one real callback invocation and releases it once", () => {
|
|||
const statusPath = process.env.TELEGRAM_E2E_FOLLOWUP_CONTROL_STATUS;
|
||||
const write = (value) => fs.writeFileSync(commandPath, JSON.stringify(value));
|
||||
const wait = async (seq) => {
|
||||
for (let i = 0; i < 200; i += 1) {
|
||||
for (;;) {
|
||||
if (fs.existsSync(statusPath)) {
|
||||
const value = JSON.parse(fs.readFileSync(statusPath, "utf8"));
|
||||
if (value.seq === seq) return value;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
throw new Error("control status timeout");
|
||||
};
|
||||
const key = "agent:main:main";
|
||||
let calls = 0;
|
||||
|
|
@ -36,17 +98,22 @@ test("holds one real callback invocation and releases it once", () => {
|
|||
draining: true, items: [run], inFlight: new Set([run]),
|
||||
}]]);
|
||||
write({ seq: 1, command: "arm", sessionKey: key });
|
||||
console.error("waiting for control seq 1 (arm)");
|
||||
await wait(1);
|
||||
const invocation = callbacks.get(key)(run);
|
||||
write({ seq: 2, command: "waitHeld" });
|
||||
console.error("waiting for control seq 2 (waitHeld)");
|
||||
const held = await wait(2);
|
||||
if (held.inFlight !== 1 || calls !== 0) throw new Error("callback was not held");
|
||||
write({ seq: 3, command: "release" });
|
||||
console.error("waiting for control seq 3 (release)");
|
||||
await wait(3);
|
||||
await invocation;
|
||||
if (calls !== 1) throw new Error("callback did not run exactly once");
|
||||
`;
|
||||
const result = spawnSync(
|
||||
const result = await run(
|
||||
context,
|
||||
children,
|
||||
process.execPath,
|
||||
[`--import=${preload}`, "--input-type=module", "--eval", script],
|
||||
{
|
||||
|
|
@ -55,12 +122,8 @@ test("holds one real callback invocation and releases it once", () => {
|
|||
TELEGRAM_E2E_FOLLOWUP_CONTROL_COMMAND: commandPath,
|
||||
TELEGRAM_E2E_FOLLOWUP_CONTROL_STATUS: statusPath,
|
||||
},
|
||||
encoding: "utf8",
|
||||
timeout: 5_000,
|
||||
},
|
||||
);
|
||||
assert.equal(result.status, 0, result.stderr);
|
||||
} finally {
|
||||
fs.rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
},
|
||||
);
|
||||
|
|
|
|||
|
|
@ -1,13 +1,69 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { spawn } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import test, { afterEach } from "node:test";
|
||||
import { describe, test } from "node:test";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../../test/helpers/temp-dir.ts";
|
||||
|
||||
const temporary = useAutoCleanupTempDirTracker(afterEach);
|
||||
// Leave time inside the harness's 120 s budget for after hooks to join children.
|
||||
const TEST_TIMEOUT_MS = 90_000;
|
||||
|
||||
function run(context, children, command, args, options) {
|
||||
context.signal.throwIfAborted();
|
||||
const child = spawn(command, args, {
|
||||
...options,
|
||||
detached: true,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
child.stdout.setEncoding("utf8").on("data", (chunk) => {
|
||||
stdout += chunk;
|
||||
});
|
||||
child.stderr.setEncoding("utf8").on("data", (chunk) => {
|
||||
stderr += chunk;
|
||||
});
|
||||
const closed = new Promise((resolve) => {
|
||||
child.once("close", (status, signal) => resolve({ status, signal, stdout, stderr }));
|
||||
});
|
||||
children.add({ child, closed });
|
||||
return new Promise((resolve, reject) => {
|
||||
const abort = () => {
|
||||
const error = new Error(
|
||||
`Child did not finish before the test ended:\n${stderr.slice(-4000)}`,
|
||||
{
|
||||
cause: context.signal.reason,
|
||||
},
|
||||
);
|
||||
context.diagnostic(error.message);
|
||||
reject(error);
|
||||
};
|
||||
context.signal.addEventListener("abort", abort, { once: true });
|
||||
child.once("error", reject);
|
||||
closed.then(resolve).finally(() => context.signal.removeEventListener("abort", abort));
|
||||
});
|
||||
}
|
||||
|
||||
// A descendant can hold the pipes after its parent exits, so always kill the group.
|
||||
async function stopChildren(children) {
|
||||
for (const { child } of children) {
|
||||
if (child.pid) {
|
||||
try {
|
||||
process.kill(-child.pid, "SIGKILL");
|
||||
} catch (error) {
|
||||
if (error.code !== "ESRCH" && error.code !== "EPERM") {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
await Promise.all([...children].map(({ closed }) => closed));
|
||||
}
|
||||
|
||||
// The 90 s / 60 s budgets bound live runs and are not under test here.
|
||||
const frozenClock = "data:text/javascript,const t=performance.now();performance.now=()=>t;";
|
||||
const checkpointPath = fileURLToPath(new URL("./telegram-binding-checkpoint.mjs", import.meta.url));
|
||||
const sourceRoot = fs.realpathSync(fileURLToPath(new URL("../../../../", import.meta.url)));
|
||||
const runId = "upgrade-proof";
|
||||
|
|
@ -87,7 +143,14 @@ function spawnTranscripts() {
|
|||
};
|
||||
}
|
||||
|
||||
function createRun() {
|
||||
function createRun(context) {
|
||||
const children = new Set();
|
||||
const temporary = useAutoCleanupTempDirTracker((cleanup) =>
|
||||
context.after(async () => {
|
||||
await stopChildren(children);
|
||||
cleanup();
|
||||
}),
|
||||
);
|
||||
const owned = temporary.make("openclaw-tg-user-mock-sut-");
|
||||
const proof = path.join(owned, "proof");
|
||||
const packageRoot = path.join(owned, "package");
|
||||
|
|
@ -148,7 +211,7 @@ function createRun() {
|
|||
}
|
||||
fs.writeFileSync(eventsPath, events.map((event) => JSON.stringify(event) + "\n").join(""));
|
||||
},
|
||||
checkpoint(phase, transcripts) {
|
||||
async checkpoint(phase, transcripts) {
|
||||
fs.writeFileSync(transcriptsPath, JSON.stringify(transcripts));
|
||||
fs.writeFileSync(
|
||||
path.join(proof, "installed-runtime.json"),
|
||||
|
|
@ -159,9 +222,12 @@ function createRun() {
|
|||
);
|
||||
const output = path.join(proof, `routing-${phase}.json`);
|
||||
const previous = { before: "spawn", after: "before" }[phase];
|
||||
const child = spawnSync(
|
||||
const child = await run(
|
||||
context,
|
||||
children,
|
||||
process.execPath,
|
||||
[
|
||||
`--import=${frozenClock}`,
|
||||
checkpointPath,
|
||||
phase,
|
||||
parentKey,
|
||||
|
|
@ -172,7 +238,6 @@ function createRun() {
|
|||
],
|
||||
{
|
||||
cwd: sourceRoot,
|
||||
encoding: "utf8",
|
||||
env: {
|
||||
PATH: process.env.PATH,
|
||||
OPENCLAW_CONFIG_PATH: path.join(owned, "config.json"),
|
||||
|
|
@ -192,58 +257,72 @@ function createRun() {
|
|||
};
|
||||
}
|
||||
|
||||
test("candidate checkpoints require follow-ups in the parent topic session after a legacy takeover", () => {
|
||||
const run = createRun();
|
||||
const transcripts = spawnTranscripts();
|
||||
run.observe("SPAWN", ["PARENT", "CHILD"]);
|
||||
const spawned = run.checkpoint("spawn", transcripts);
|
||||
assert.equal(spawned.exitCode, 0, JSON.stringify(spawned.diagnostic));
|
||||
assert.equal(spawned.checkpoint.phases.CHILD.session, "child");
|
||||
describe("Telegram binding checkpoints", { concurrency: true }, () => {
|
||||
test(
|
||||
"candidate checkpoints require follow-ups in the parent topic session after a legacy takeover",
|
||||
{ timeout: TEST_TIMEOUT_MS },
|
||||
async (context) => {
|
||||
const fixture = createRun(context);
|
||||
const transcripts = spawnTranscripts();
|
||||
fixture.observe("SPAWN", ["PARENT", "CHILD"]);
|
||||
const spawned = await fixture.checkpoint("spawn", transcripts);
|
||||
assert.equal(spawned.exitCode, 0, JSON.stringify(spawned.diagnostic));
|
||||
assert.equal(spawned.checkpoint.phases.CHILD.session, "child");
|
||||
|
||||
run.observe("BEFORE", ["BEFORE"]);
|
||||
transcripts.parent.push(...turn("BEFORE"));
|
||||
const before = run.checkpoint("before", transcripts);
|
||||
assert.equal(before.exitCode, 0, JSON.stringify(before.diagnostic));
|
||||
assert.equal(before.checkpoint.childKey, childKey);
|
||||
assert.equal(before.checkpoint.phases.CHILD.session, "child");
|
||||
assert.equal(before.checkpoint.phases.BEFORE.session, "parent");
|
||||
fixture.observe("BEFORE", ["BEFORE"]);
|
||||
transcripts.parent.push(...turn("BEFORE"));
|
||||
const before = await fixture.checkpoint("before", transcripts);
|
||||
assert.equal(before.exitCode, 0, JSON.stringify(before.diagnostic));
|
||||
assert.equal(before.checkpoint.childKey, childKey);
|
||||
assert.equal(before.checkpoint.phases.CHILD.session, "child");
|
||||
assert.equal(before.checkpoint.phases.BEFORE.session, "parent");
|
||||
|
||||
run.observe("AFTER", ["AFTER"]);
|
||||
transcripts.parent.push(...turn("AFTER"));
|
||||
const after = run.checkpoint("after", transcripts);
|
||||
assert.equal(after.exitCode, 0, JSON.stringify(after.diagnostic));
|
||||
assert.deepEqual(
|
||||
Object.fromEntries(
|
||||
Object.entries(after.checkpoint.phases).map(([name, fact]) => [name, fact.session]),
|
||||
),
|
||||
{ CHILD: "child", BEFORE: "parent", AFTER: "parent" },
|
||||
fixture.observe("AFTER", ["AFTER"]);
|
||||
transcripts.parent.push(...turn("AFTER"));
|
||||
const after = await fixture.checkpoint("after", transcripts);
|
||||
assert.equal(after.exitCode, 0, JSON.stringify(after.diagnostic));
|
||||
assert.deepEqual(
|
||||
Object.fromEntries(
|
||||
Object.entries(after.checkpoint.phases).map(([name, fact]) => [name, fact.session]),
|
||||
),
|
||||
{ CHILD: "child", BEFORE: "parent", AFTER: "parent" },
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
"candidate checkpoints reject a follow-up still routed to the legacy spawned child",
|
||||
{ timeout: TEST_TIMEOUT_MS },
|
||||
async (context) => {
|
||||
const fixture = createRun(context);
|
||||
const transcripts = spawnTranscripts();
|
||||
fixture.observe("SPAWN", ["PARENT", "CHILD"]);
|
||||
assert.equal((await fixture.checkpoint("spawn", transcripts)).exitCode, 0);
|
||||
|
||||
fixture.observe("BEFORE", ["BEFORE"]);
|
||||
transcripts.child.push(...turn("BEFORE"));
|
||||
const before = await fixture.checkpoint("before", transcripts);
|
||||
assert.equal(before.exitCode, 1);
|
||||
assert.equal(before.diagnostic.code, "FOLLOWUP_ROUTED_TO_CHILD");
|
||||
assert.equal(before.diagnostic.earlyPhase, "BEFORE");
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
"candidate checkpoints reject a removed spawn-phase child transcript",
|
||||
{ timeout: TEST_TIMEOUT_MS },
|
||||
async (context) => {
|
||||
const fixture = createRun(context);
|
||||
const transcripts = spawnTranscripts();
|
||||
fixture.observe("SPAWN", ["PARENT", "CHILD"]);
|
||||
assert.equal((await fixture.checkpoint("spawn", transcripts)).exitCode, 0);
|
||||
|
||||
fixture.observe("BEFORE", ["BEFORE"]);
|
||||
transcripts.parent.push(...turn("BEFORE"));
|
||||
transcripts.child = [];
|
||||
const before = await fixture.checkpoint("before", transcripts);
|
||||
assert.equal(before.exitCode, 1);
|
||||
assert.equal(before.diagnostic.code, "PRIOR_CHILD_PHASE_DISAPPEARED");
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
test("candidate checkpoints reject a follow-up still routed to the legacy spawned child", () => {
|
||||
const run = createRun();
|
||||
const transcripts = spawnTranscripts();
|
||||
run.observe("SPAWN", ["PARENT", "CHILD"]);
|
||||
assert.equal(run.checkpoint("spawn", transcripts).exitCode, 0);
|
||||
|
||||
run.observe("BEFORE", ["BEFORE"]);
|
||||
transcripts.child.push(...turn("BEFORE"));
|
||||
const before = run.checkpoint("before", transcripts);
|
||||
assert.equal(before.exitCode, 1);
|
||||
assert.equal(before.diagnostic.code, "FOLLOWUP_ROUTED_TO_CHILD");
|
||||
assert.equal(before.diagnostic.earlyPhase, "BEFORE");
|
||||
});
|
||||
|
||||
test("candidate checkpoints reject a removed spawn-phase child transcript", () => {
|
||||
const run = createRun();
|
||||
const transcripts = spawnTranscripts();
|
||||
run.observe("SPAWN", ["PARENT", "CHILD"]);
|
||||
assert.equal(run.checkpoint("spawn", transcripts).exitCode, 0);
|
||||
|
||||
run.observe("BEFORE", ["BEFORE"]);
|
||||
transcripts.parent.push(...turn("BEFORE"));
|
||||
transcripts.child = [];
|
||||
const before = run.checkpoint("before", transcripts);
|
||||
assert.equal(before.exitCode, 1);
|
||||
assert.equal(before.diagnostic.code, "PRIOR_CHILD_PHASE_DISAPPEARED");
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue