332 lines
18 KiB
TypeScript
332 lines
18 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import fs from "node:fs";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import { spawnSync } from "node:child_process";
|
|
import { Agent } from "@earendil-works/pi-agent-core";
|
|
import { after, afterEach, test } from "node:test";
|
|
import { createFauxCore, fauxAssistantMessage } from "@earendil-works/pi-ai";
|
|
import { toLlmMessages, type LhMessage } from "../src/core/native-messages.ts";
|
|
import { buildSubagentSteer, type JobKind, type SteerMessage } from "../src/core/contracts.ts";
|
|
|
|
// Install isolation before importing any storage/configuration composition.
|
|
const directory = fs.mkdtempSync(path.join(os.tmpdir(), "lh-job-delivery-"));
|
|
process.env.LH_DATA_DIR = path.join(directory, "data");
|
|
process.env.LH_AUTH_FILE = path.join(directory, "auth.json");
|
|
fs.writeFileSync(process.env.LH_AUTH_FILE, "{}");
|
|
const { stores } = await import("../src/server/stores.ts");
|
|
const { jobTools } = await import("../src/agent/tools/jobs.ts");
|
|
const { createJobRunner, researchModelOverride } = await import("../src/agent/subagents/runner.ts");
|
|
const { isJobLive } = await import("../src/agent/subagents/aborts.ts");
|
|
const { beginTurn, endTurn } = await import("../src/server/turn-liveness.ts");
|
|
const {
|
|
deliverJobResult, discardChatDeliveries, noteSteersDurable, recoverStaleJobs,
|
|
registerLiveAgent, unregisterLiveAgent, settleSteers, setTurnRunner,
|
|
} = await import("../src/server/jobs.ts");
|
|
const { activityFeed } = await import("../src/server/observability.ts");
|
|
const { deleteChatSession } = await import("../src/server/chats.ts");
|
|
const chats: string[] = [];
|
|
const result = { brief: "Fixture research result", sources: [] };
|
|
const question = {
|
|
id: "question-message", role: "assistant", content: [{ type: "toolCall", id: "question-call", name: "quiz", arguments: {} }],
|
|
api: "openai-completions", provider: "faux", model: "faux", timestamp: 1,
|
|
} as unknown as LhMessage;
|
|
const delay = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
|
|
async function until(predicate: () => boolean): Promise<void> {
|
|
const end = Date.now() + 8000;
|
|
while (!predicate()) { assert(Date.now() < end, "fixture condition timed out"); await delay(10); }
|
|
}
|
|
function sweep(): void {
|
|
const global = globalThis as { __lhWakePipeline?: { lastSweepAt: number } };
|
|
if (global.__lhWakePipeline) global.__lhWakePipeline.lastSweepAt = 0;
|
|
recoverStaleJobs();
|
|
}
|
|
function chat(blocked = false) {
|
|
const value = stores.chatStore.createChat("Isolated delivery fixture");
|
|
chats.push(value.id);
|
|
if (blocked) stores.chatStore.saveMessages(value.id, [question]);
|
|
return value;
|
|
}
|
|
function pending(chatId: string) {
|
|
const value = stores.jobStore.createJob({ kind: "research", chatId, toolCallId: "fixture-call" });
|
|
const job = stores.jobStore.updateJob(value.id, {
|
|
status: "done", result, activity: "finished", delivery: { kind: "result", status: "pending" },
|
|
})!;
|
|
return { job, steer: buildSubagentSteer(job, result) };
|
|
}
|
|
function agent() {
|
|
const handed: SteerMessage[] = [];
|
|
return { handed, value: { steer: (marker: SteerMessage) => handed.push(marker) } as unknown as Parameters<typeof registerLiveAgent>[1] };
|
|
}
|
|
function persist(chatId: string, steers: readonly LhMessage[]): void {
|
|
stores.chatStore.saveMessages(chatId, [...stores.chatStore.getChat(chatId)!.messages, ...steers]);
|
|
noteSteersDurable(chatId, steers);
|
|
}
|
|
afterEach(() => {
|
|
researchModelOverride.current = undefined;
|
|
for (const id of chats.splice(0)) {
|
|
unregisterLiveAgent(id); endTurn(id); discardChatDeliveries(id);
|
|
stores.jobStore.deleteJobsForChat(id); stores.chatStore.deleteChat(id);
|
|
}
|
|
setTurnRunner(async () => 500);
|
|
});
|
|
after(() => fs.rmSync(directory, { recursive: true, force: true }));
|
|
|
|
for (const [kind, input] of Object.entries({
|
|
research: { question: "Saved research question", context: "Saved context" },
|
|
demo: { title: "Fixture", brief: "Saved demo brief" },
|
|
visual: { brief: "Saved figure brief" },
|
|
pdf: { path: "fixture.pdf", question: "Saved PDF question" },
|
|
})) test(`${kind}: dispatch saves input before spawn and does not overwrite it on repeated dispatch`, async () => {
|
|
const owner = chat();
|
|
let spawns = 0;
|
|
const tools = jobTools({ chatId: owner.id, jobStore: stores.jobStore, spawn: (job) => {
|
|
spawns++;
|
|
assert.deepEqual(stores.jobStore.getJob(job.id)!.input, input);
|
|
} });
|
|
await tools[kind]!.execute!("fixture-dispatch", input);
|
|
await tools[kind]!.execute!("fixture-dispatch", input);
|
|
assert.equal(spawns, 1);
|
|
assert.equal(stores.jobStore.listJobs(owner.id)[0]!.kind, kind as JobKind);
|
|
});
|
|
|
|
test("saved question survives absent transcript; runner ends while delivery is still pending", async () => {
|
|
const owner = chat(true);
|
|
const input = { question: "The dispatch turn was never saved", context: "Fixture context" };
|
|
const job = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "lost-turn", input });
|
|
const faux = createFauxCore({});
|
|
let seen = "";
|
|
faux.setResponses([(context) => { seen = JSON.stringify(context.messages); return fauxAssistantMessage("Fixture brief"); }]);
|
|
researchModelOverride.current = { model: faux.getModel(), streamFn: faux.streamSimple };
|
|
let delivery: Promise<void> | undefined;
|
|
const runner = createJobRunner({ ...stores, deliver: (steer) => { delivery = deliverJobResult(steer); return delivery; } });
|
|
await runner.runJob(job.id);
|
|
assert(seen.includes(input.question)); assert(seen.includes(input.context));
|
|
assert.equal(isJobLive(job.id), false, "worker registration no longer spans delivery");
|
|
assert.equal(stores.jobStore.getJob(job.id)!.status, "done");
|
|
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
|
|
assert(!activityFeed().jobs.some((item) => item.id === job.id), "no running background-agent row");
|
|
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
|
|
assert.equal(next.handed.length, 1);
|
|
persist(owner.id, next.handed); unregisterLiveAgent(owner.id); settleSteers(owner.id, next.handed); endTurn(owner.id);
|
|
await delivery;
|
|
});
|
|
|
|
test("missing or blank research question records an honest failure without invoking a model", async () => {
|
|
for (const input of [undefined, { question: " " }]) {
|
|
const owner = chat();
|
|
const job = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "missing-question", input });
|
|
const faux = createFauxCore({}); let calls = 0;
|
|
faux.setResponses([() => { calls++; return fauxAssistantMessage("must not run"); }]);
|
|
researchModelOverride.current = { model: faux.getModel(), streamFn: faux.streamSimple };
|
|
const steers: SteerMessage[] = [];
|
|
await createJobRunner({ ...stores, deliver: async (steer) => { steers.push(steer); } }).runJob(job.id);
|
|
assert.equal(calls, 0);
|
|
assert.equal(stores.jobStore.getJob(job.id)!.error, "research question unavailable");
|
|
assert.equal(stores.jobStore.getJob(job.id)!.status, "error");
|
|
assert.match(steers[0]!.content, /^Research failed: the research question is unavailable\./);
|
|
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
|
|
}
|
|
});
|
|
|
|
test("repeated recovery behind a question yields one handed and one durable marker", async () => {
|
|
const owner = chat(true); const { job, steer } = pending(owner.id);
|
|
let wakes = 0; setTurnRunner(async () => { wakes++; return 500; });
|
|
sweep(); sweep(); sweep();
|
|
const first = deliverJobResult(steer);
|
|
assert.strictEqual(deliverJobResult(steer), first, "duplicate callers share one logical delivery");
|
|
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
|
|
assert.deepEqual(next.handed.map((message) => message.id), [steer.id]);
|
|
persist(owner.id, next.handed); unregisterLiveAgent(owner.id); settleSteers(owner.id, next.handed); endTurn(owner.id);
|
|
await first;
|
|
assert.equal(wakes, 0);
|
|
assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1);
|
|
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "delivered");
|
|
});
|
|
|
|
for (const refused of [409, 403]) test(`duplicate during a ${refused} wake retains one logical owner`, async () => {
|
|
const owner = chat(); const { steer } = pending(owner.id);
|
|
let release!: () => void; const gate = new Promise<void>((resolve) => { release = resolve; });
|
|
let attempts = 0;
|
|
setTurnRunner(async (id, input) => {
|
|
assert.equal(input.type, "steers"); if (input.type !== "steers") return 500;
|
|
assert.equal(input.steers.length, 1);
|
|
attempts++;
|
|
if (attempts === 1) { await gate; return refused; }
|
|
persist(id, input.steers); return 200;
|
|
});
|
|
const first = deliverJobResult(steer);
|
|
await until(() => attempts === 1);
|
|
assert.strictEqual(deliverJobResult(steer), first, "wake cargo is still in the registry");
|
|
release();
|
|
if (refused === 403) {
|
|
await delay(30);
|
|
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
|
|
assert.equal(next.handed.length, 1);
|
|
persist(owner.id, next.handed); unregisterLiveAgent(owner.id); settleSteers(owner.id, next.handed); endTurn(owner.id);
|
|
}
|
|
await first;
|
|
assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1);
|
|
});
|
|
|
|
test("failed start stays quiet across sweeps, then joins a learner run", async () => {
|
|
const owner = chat(); const { job, steer } = pending(owner.id);
|
|
let attempts = 0; setTurnRunner(async () => { attempts++; return 503; });
|
|
const delivery = deliverJobResult(steer); await until(() => attempts === 1); await delay(30);
|
|
sweep(); sweep(); await delay(900);
|
|
assert.equal(attempts, 1);
|
|
assert.equal(stores.jobStore.getJob(job.id)!.status, "done");
|
|
assert.equal(stores.jobStore.getJob(job.id)!.error, undefined);
|
|
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
|
|
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
|
|
assert.equal(next.handed.length, 1);
|
|
persist(owner.id, next.handed); unregisterLiveAgent(owner.id); settleSteers(owner.id, next.handed); endTurn(owner.id);
|
|
await delivery;
|
|
});
|
|
|
|
test("HTTP acceptance without a saved marker is not delivery completion or a retry trigger", async () => {
|
|
const owner = chat(); const { job, steer } = pending(owner.id);
|
|
let attempts = 0;
|
|
const accepted = agent();
|
|
setTurnRunner(async () => {
|
|
attempts++; assert(beginTurn(owner.id));
|
|
registerLiveAgent(owner.id, accepted.value, false, new Set([steer.id]));
|
|
return 200;
|
|
});
|
|
const delivery = deliverJobResult(steer); await until(() => attempts === 1); await delay(30);
|
|
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
|
|
unregisterLiveAgent(owner.id); endTurn(owner.id); settleSteers(owner.id, [], true, accepted.value);
|
|
sweep(); await delay(900); assert.equal(attempts, 1);
|
|
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
|
|
assert.equal(next.handed.length, 1);
|
|
persist(owner.id, next.handed); unregisterLiveAgent(owner.id); endTurn(owner.id); await delivery;
|
|
});
|
|
|
|
test("a racing foreground settlement cannot release an outstanding staged wake", async () => {
|
|
const owner = chat(); const { steer } = pending(owner.id);
|
|
let started = false;
|
|
let release!: () => void;
|
|
const startup = new Promise<void>((resolve) => { release = resolve; });
|
|
let modelCopies = 0;
|
|
let nativeCopies = 0;
|
|
setTurnRunner(async (id, input) => {
|
|
assert.equal(input.type, "steers"); if (input.type !== "steers") return 500;
|
|
started = true;
|
|
// Subscription wake startup can await catalog refresh before claiming
|
|
// the session. Its cargo is not yet a durable input prefix.
|
|
await startup;
|
|
const faux = createFauxCore({});
|
|
faux.setResponses([(context) => {
|
|
modelCopies = JSON.stringify(context.messages).split(result.brief).length - 1;
|
|
return fauxAssistantMessage("Fixture wake response");
|
|
}]);
|
|
const wake = new Agent({ initialState: { model: faux.getModel(), messages: input.steers }, convertToLlm: toLlmMessages, streamFn: faux.streamSimple });
|
|
assert(beginTurn(id));
|
|
registerLiveAgent(id, wake, false, new Set(input.steers.map((message) => message.id)));
|
|
await wake.continue();
|
|
nativeCopies = wake.state.messages.filter((message) => (message as LhMessage).id === steer.id).length;
|
|
persist(id, wake.state.messages.filter((message) => message.role !== "system") as LhMessage[]);
|
|
unregisterLiveAgent(id); endTurn(id); settleSteers(id, stores.chatStore.getChat(id)!.messages, false, wake);
|
|
return 200;
|
|
});
|
|
const delivery = deliverJobResult(steer);
|
|
await until(() => started);
|
|
const foreground = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, foreground.value);
|
|
assert.equal(foreground.handed.length, 0, "outstanding wake owns the marker");
|
|
unregisterLiveAgent(owner.id); endTurn(owner.id); settleSteers(owner.id, [], false, foreground.value);
|
|
release(); await delivery;
|
|
assert.equal(modelCopies, 1, "one result in the model's initial context");
|
|
assert.equal(nativeCopies, 1, "no self-steering of staged wake cargo");
|
|
assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1);
|
|
});
|
|
|
|
test("canonical save wins even if delivery bookkeeping fails; no marker replay", async () => {
|
|
const owner = chat(true); const { job, steer } = pending(owner.id);
|
|
const delivery = deliverJobResult(steer);
|
|
const update = stores.jobStore.updateJob;
|
|
stores.jobStore.updateJob = (id, patch) => {
|
|
if (id === job.id && patch.delivery?.status === "delivered") throw new Error("fixture acknowledgement write failure");
|
|
return update(id, patch);
|
|
};
|
|
try { persist(owner.id, [steer]); await delivery; }
|
|
finally { stores.jobStore.updateJob = update; }
|
|
let wakes = 0; setTurnRunner(async () => { wakes++; return 500; });
|
|
sweep();
|
|
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "delivered");
|
|
assert.equal(wakes, 0);
|
|
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
|
|
assert.equal(next.handed.length, 0); unregisterLiveAgent(owner.id); endTurn(owner.id);
|
|
});
|
|
|
|
test("old running results become finished pending work; old terminal records are untouched", () => {
|
|
const owner = chat(true);
|
|
const old = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "legacy-pending" });
|
|
stores.jobStore.updateJob(old.id, { result });
|
|
const terminal = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "legacy-terminal" });
|
|
stores.jobStore.updateJob(terminal.id, { status: "done", result });
|
|
sweep();
|
|
assert.equal(stores.jobStore.getJob(old.id)!.status, "done");
|
|
assert.equal(stores.jobStore.getJob(old.id)!.delivery?.status, "pending");
|
|
assert.equal(stores.jobStore.getJob(terminal.id)!.delivery, undefined);
|
|
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
|
|
assert.deepEqual(next.handed.map((message) => message.id), [`steer_${old.id}`]);
|
|
unregisterLiveAgent(owner.id); endTurn(owner.id);
|
|
});
|
|
|
|
test("fresh process restores a pending delivery on registration without a jobs poll", () => {
|
|
const owner = chat(true); const { job, steer } = pending(owner.id);
|
|
const script = `
|
|
const { stores } = await import('./src/server/stores.ts');
|
|
const { registerLiveAgent, noteSteersDurable, unregisterLiveAgent } = await import('./src/server/jobs.ts');
|
|
const { beginTurn, endTurn } = await import('./src/server/turn-liveness.ts');
|
|
const handed = []; beginTurn(${JSON.stringify(owner.id)});
|
|
registerLiveAgent(${JSON.stringify(owner.id)}, { steer: marker => handed.push(marker) });
|
|
const record = stores.chatStore.getChat(${JSON.stringify(owner.id)});
|
|
stores.chatStore.saveMessages(record.id, [...record.messages, ...handed]);
|
|
noteSteersDurable(record.id, handed); unregisterLiveAgent(record.id); endTurn(record.id);
|
|
console.log(JSON.stringify(handed.map(marker => marker.id)));
|
|
`;
|
|
const child = spawnSync(process.execPath, ["--import", "tsx", "--input-type=module", "-e", script], {
|
|
cwd: process.cwd(), encoding: "utf8", timeout: 15000,
|
|
env: { PATH: process.env.PATH, HOME: path.join(directory, "home"), NODE_ENV: "test", LH_DATA_DIR: process.env.LH_DATA_DIR, LH_AUTH_FILE: process.env.LH_AUTH_FILE },
|
|
});
|
|
assert.equal(child.status, 0, child.stderr);
|
|
assert.deepEqual(JSON.parse(child.stdout.trim()), [steer.id]);
|
|
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "delivered");
|
|
assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1);
|
|
});
|
|
|
|
test("deleting legacy completed work needs no preceding recovery poll", async () => {
|
|
const owner = chat(true);
|
|
const old = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "legacy-delete" });
|
|
stores.jobStore.updateJob(old.id, { result });
|
|
const outcome = await deleteChatSession(owner.id, "preserve-graph");
|
|
assert(outcome.ok);
|
|
assert.equal(stores.jobStore.getJob(old.id), null);
|
|
assert.equal(stores.chatStore.getChat(owner.id), null);
|
|
});
|
|
|
|
test("deletion still refuses a computing worker or a live turn", async () => {
|
|
const owner = chat();
|
|
const computing = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "computing-delete" });
|
|
const blocked = await deleteChatSession(owner.id, "preserve-graph");
|
|
assert(!blocked.ok && blocked.code === "jobs_running");
|
|
stores.jobStore.updateJob(computing.id, { status: "done" });
|
|
assert(beginTurn(owner.id));
|
|
const live = await deleteChatSession(owner.id, "preserve-graph");
|
|
assert(!live.ok && live.code === "turn_in_flight");
|
|
endTurn(owner.id);
|
|
});
|
|
|
|
test("deleting a session discards its pending deliveries without starting a turn", async () => {
|
|
const owner = chat(true); const { job, steer } = pending(owner.id);
|
|
let wakes = 0; setTurnRunner(async () => { wakes++; return 500; });
|
|
const delivery = deliverJobResult(steer);
|
|
const outcome = await deleteChatSession(owner.id, "preserve-graph");
|
|
assert(outcome.ok);
|
|
await delivery; await delay(450);
|
|
assert.equal(stores.jobStore.getJob(job.id), null);
|
|
assert.equal(stores.chatStore.getChat(owner.id), null);
|
|
assert.equal(wakes, 0);
|
|
});
|