mirror of
https://github.com/vectorize-io/hindsight.git
synced 2026-09-14 19:31:49 +08:00
fix(coding-agents): admit one codebase survey per bank, held by a heartbeat lease (#4303)
* fix(coding-agents): admit one codebase survey per bank * fix(coding-agents): hold the survey lease with a heartbeat supervisor The PID-owned survey lock could wedge a bank forever: a recycled PID answers kill(pid, 0) as alive, and the lock had no expiry. Replace it with a lease the holder refreshes every 5s, reclaimable after 30s of silence. The survey agent cannot heartbeat, so it now runs under a tiny detached Node supervisor (dist/survey-supervisor.js) that holds the lease for exactly the agent's lifetime, releases on exit, and kills the agent if the lease was taken over. Only rename/utimes/stat are used, so it behaves the same on macOS, Linux and Windows; no flock, O_EXCL or PID probes. * fix(coding-agents): hand the survey spec to its supervisor via env, not argv Reproducing #4255 end to end (6 concurrent SessionStart hooks on a cold bank) showed the supervisor SIGKILLed within milliseconds of starting, so no survey ran at all. A harmless stand-in launched from the same point with the spec in argv was killed 5/5; the same launch with the spec in the environment survived 5/5. The spec carries the whole agent command line (prompt, inline MCP config, tool deny-list), which endpoint security (SentinelOne here) kills on sight when a hook-launched node process carries it. The supervisor now reads HINDSIGHT_SURVEY_SPEC and deletes it before starting the agent, so the agent and its MCP server never inherit it. --------- Co-authored-by: r266-tech <r266-tech@users.noreply.github.com>
This commit is contained in:
Nicolò Boschi
@@ -14,7 +14,7 @@ describe("buildSessionStartContext", () => {
|
||||
it("cold git repo + autoSeed on -> seeds + surveys, note in systemMessage (user-visible) + roster in additionalContext (model)", async () => {
|
||||
const client = { listDocumentIds: async () => new Set<string>(), listPages: listPagesOk };
|
||||
const startSeed = vi.fn();
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const out = await buildSessionStartContext({
|
||||
cwd: "/repo/dir",
|
||||
bankId: "bank-1",
|
||||
@@ -58,7 +58,7 @@ describe("buildSessionStartContext", () => {
|
||||
harness: "codex",
|
||||
hasGit: () => true,
|
||||
startSeed,
|
||||
startSurvey: vi.fn(),
|
||||
startSurvey: vi.fn().mockResolvedValue(true),
|
||||
});
|
||||
expect(startSeed).toHaveBeenCalledWith("/repo/dir", { limit: 300, harness: "codex" });
|
||||
});
|
||||
@@ -66,7 +66,7 @@ describe("buildSessionStartContext", () => {
|
||||
it("cold git repo + codebaseSurvey:false -> starts the seed but NOT the survey", async () => {
|
||||
const client = { listDocumentIds: async () => new Set<string>(), listPages: listPagesOk };
|
||||
const startSeed = vi.fn();
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const out = await buildSessionStartContext({
|
||||
cwd: "/repo/dir",
|
||||
bankId: "bank-1",
|
||||
@@ -152,7 +152,7 @@ describe("buildSessionStartContext", () => {
|
||||
|
||||
it("warm bank (non-empty doc set) -> deepen engine fires, but no survey/note", async () => {
|
||||
const startSeed = vi.fn();
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const client = { listDocumentIds: async () => new Set(["git:abc"]), listPages: listPagesOk };
|
||||
const out = await buildSessionStartContext({
|
||||
cwd: "/repo/dir",
|
||||
@@ -325,7 +325,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
];
|
||||
|
||||
it(">= threshold since the latest reachable baseline -> re-surveys + records a new baseline", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const retain = vi.fn();
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
@@ -358,7 +358,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
client: warmClient(["survey-baseline:oldsha"], retain),
|
||||
hasGit: () => true,
|
||||
startSeed: vi.fn(),
|
||||
startSurvey: vi.fn(),
|
||||
startSurvey: vi.fn().mockResolvedValue(true),
|
||||
headSha: () => "newsha",
|
||||
commitsSince: () => 25,
|
||||
});
|
||||
@@ -374,7 +374,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
});
|
||||
|
||||
it("< threshold -> no re-survey, no new baseline", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const retain = vi.fn();
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
@@ -392,7 +392,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
});
|
||||
|
||||
it("no baseline yet (upgrade from a pre-feature bank) -> records HEAD as baseline, does NOT survey", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const retain = vi.fn();
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
@@ -410,7 +410,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
});
|
||||
|
||||
it("all markers unreachable (rebase/gc) -> re-baselines to HEAD, does NOT survey", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const retain = vi.fn();
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
@@ -428,7 +428,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
});
|
||||
|
||||
it("takes the MIN reachable count (newest survey), ignoring older + dead-branch markers", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const counts: Record<string, number | null> = { old1: 50, old2: 10, dead: null };
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
@@ -445,7 +445,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
});
|
||||
|
||||
it("surveyRefreshCommits=0 disables re-survey even far past threshold", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
bankId: "bank-1",
|
||||
@@ -461,7 +461,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
});
|
||||
|
||||
it("cold seed records the survey baseline", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const retain = vi.fn();
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
@@ -481,7 +481,7 @@ describe("buildSessionStartContext — periodic re-survey (bank-stored commit co
|
||||
|
||||
describe("buildSessionStartContext — crashed-survey retry (baseline without findings)", () => {
|
||||
it("re-fires the survey when a baseline exists but NO findings docs ever arrived", async () => {
|
||||
const startSurvey = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(true);
|
||||
const retain = vi.fn();
|
||||
const client = {
|
||||
listDocumentIds: async (tag: string) =>
|
||||
@@ -508,3 +508,33 @@ describe("buildSessionStartContext — crashed-survey retry (baseline without fi
|
||||
expect(out).toBeTruthy();
|
||||
});
|
||||
});
|
||||
|
||||
describe("survey launch admission", () => {
|
||||
it.each([false, true])(
|
||||
"does not advance a %s warm/cold baseline when launch is skipped or fails",
|
||||
async (warm) => {
|
||||
const retain = vi.fn();
|
||||
const startSurvey = vi.fn().mockResolvedValue(false);
|
||||
await buildSessionStartContext({
|
||||
cwd: "/repo",
|
||||
bankId: "bank-1",
|
||||
cfg: resolveConfig({ surveyRefreshCommits: 20 }),
|
||||
client: {
|
||||
listDocumentIds: async (tag) =>
|
||||
tag === "source:survey-baseline"
|
||||
? new Set(["survey-baseline:oldsha"])
|
||||
: new Set(warm ? ["git:old"] : []),
|
||||
listPages: listPagesOk,
|
||||
retain,
|
||||
},
|
||||
hasGit: () => true,
|
||||
startSeed: vi.fn(),
|
||||
startSurvey,
|
||||
headSha: () => "newsha",
|
||||
commitsSince: () => 25,
|
||||
});
|
||||
expect(startSurvey).toHaveBeenCalledOnce();
|
||||
expect(retain).not.toHaveBeenCalled();
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
@@ -147,10 +147,7 @@ export async function buildSessionStartContext(args: {
|
||||
stateDir?: string;
|
||||
hasGit?: (dir: string) => boolean;
|
||||
startSeed?: (repoDir: string, opts?: { limit?: number; harness?: string }) => void;
|
||||
startSurvey?: (
|
||||
repoDir: string,
|
||||
opts?: { harness?: SurveyHarness; model?: string; budgetUsd?: number }
|
||||
) => void;
|
||||
startSurvey?: typeof startCodebaseSurvey;
|
||||
headSha?: (dir: string) => string | null;
|
||||
commitsSince?: (dir: string, sinceSha: string) => number | null;
|
||||
}): Promise<SessionStartOutput> {
|
||||
@@ -232,13 +229,13 @@ export async function buildSessionStartContext(args: {
|
||||
if (docIds.size === 0) {
|
||||
if (cfg.codebaseSurvey !== false) {
|
||||
// Run the survey under the current harness's own CLI (falls back to any available agent).
|
||||
startSurvey(cwd, {
|
||||
const started = await startSurvey(cwd, {
|
||||
harness: harness as SurveyHarness,
|
||||
model: cfg.surveyModel,
|
||||
budgetUsd: cfg.surveyBudgetUsd,
|
||||
});
|
||||
const sha = resolveHeadSha(cwd);
|
||||
if (sha) recordSurveyBaseline(sha); // baseline for the commit-count re-survey below
|
||||
if (started && sha) recordSurveyBaseline(sha);
|
||||
}
|
||||
diag(harness, "seed_started", { bank: bankId });
|
||||
} else if (cfg.codebaseSurvey !== false && cfg.surveyRefreshCommits > 0) {
|
||||
@@ -268,17 +265,19 @@ export async function buildSessionStartContext(args: {
|
||||
const findingsAbsent =
|
||||
counts.length > 0 && !SURVEY_DOC_IDS.some((id) => uploads.has(id));
|
||||
if ((sinceLast !== null && sinceLast >= cfg.surveyRefreshCommits) || findingsAbsent) {
|
||||
startSurvey(cwd, {
|
||||
const started = await startSurvey(cwd, {
|
||||
harness: harness as SurveyHarness,
|
||||
model: cfg.surveyModel,
|
||||
budgetUsd: cfg.surveyBudgetUsd,
|
||||
});
|
||||
recordSurveyBaseline(sha);
|
||||
diag(harness, "survey_refresh", {
|
||||
bank: bankId,
|
||||
commits: sinceLast,
|
||||
retry: findingsAbsent,
|
||||
});
|
||||
if (started) {
|
||||
recordSurveyBaseline(sha);
|
||||
diag(harness, "survey_refresh", {
|
||||
bank: bankId,
|
||||
commits: sinceLast,
|
||||
retry: findingsAbsent,
|
||||
});
|
||||
}
|
||||
} else if (sinceLast === null) {
|
||||
recordSurveyBaseline(sha); // first baseline, or reset after a rebase — no survey
|
||||
}
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import { mkdirSync, mkdtempSync, readdirSync, rmSync, utimesSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
acquireLease,
|
||||
heartbeatLease,
|
||||
LEASE_STALE_MS,
|
||||
releaseLease,
|
||||
superviseSurvey,
|
||||
type SurveyLease,
|
||||
} from "./survey-lease";
|
||||
|
||||
let root: string;
|
||||
beforeEach(() => {
|
||||
root = mkdtempSync(join(tmpdir(), "hindsight-lease-test-"));
|
||||
});
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
/** Backdate the lease's heartbeat, as if its holder died `ms` ago. */
|
||||
function age(lease: SurveyLease, ms: number) {
|
||||
const then = new Date(Date.now() - ms);
|
||||
utimesSync(join(lease.directory, lease.owner), then, then);
|
||||
}
|
||||
|
||||
describe("acquireLease", () => {
|
||||
it("admits exactly one holder per key, and leaves no staging behind", () => {
|
||||
const first = acquireLease(root, "k");
|
||||
expect(first).toBeDefined();
|
||||
expect(acquireLease(root, "k")).toBeUndefined();
|
||||
expect(acquireLease(root, "other")).toBeDefined();
|
||||
expect(readdirSync(root).sort()).toEqual(["survey-k.lock", "survey-other.lock"]);
|
||||
expect(readdirSync(first!.directory)).toEqual([first!.owner]);
|
||||
});
|
||||
|
||||
it("is free again once released", () => {
|
||||
releaseLease(acquireLease(root, "k")!);
|
||||
expect(readdirSync(root)).toEqual([]);
|
||||
expect(acquireLease(root, "k")).toBeDefined();
|
||||
});
|
||||
|
||||
it("keeps a fresh lease however old the process that took it is — liveness is the heartbeat", () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
age(lease, LEASE_STALE_MS - 5_000);
|
||||
expect(acquireLease(root, "k")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("reclaims a lease whose heartbeat stopped, exactly once", () => {
|
||||
const dead = acquireLease(root, "k")!;
|
||||
age(dead, LEASE_STALE_MS + 5_000);
|
||||
const next = acquireLease(root, "k");
|
||||
expect(next).toBeDefined();
|
||||
expect(next!.owner).not.toBe(dead.owner);
|
||||
expect(acquireLease(root, "k")).toBeUndefined(); // the new holder is fresh
|
||||
});
|
||||
|
||||
it("treats a heartbeat from the future (clock stepped back) as stale", () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
age(lease, -(LEASE_STALE_MS + 60_000));
|
||||
expect(acquireLease(root, "k")).toBeDefined();
|
||||
});
|
||||
|
||||
it("honors a custom stale window", () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
age(lease, 3_000);
|
||||
expect(acquireLease(root, "k", 10_000)).toBeUndefined();
|
||||
expect(acquireLease(root, "k", 1_000)).toBeDefined();
|
||||
});
|
||||
|
||||
it("reclaims an unowned directory left by an interrupted release", () => {
|
||||
mkdirSync(join(root, "survey-k.lock"));
|
||||
expect(acquireLease(root, "k")).toBeDefined();
|
||||
});
|
||||
|
||||
it("never touches a lock directory it does not recognise", () => {
|
||||
const directory = join(root, "survey-k.lock");
|
||||
mkdirSync(join(directory, "a"), { recursive: true });
|
||||
mkdirSync(join(directory, "b"));
|
||||
expect(acquireLease(root, "k")).toBeUndefined();
|
||||
expect(readdirSync(directory).sort()).toEqual(["a", "b"]);
|
||||
});
|
||||
|
||||
it("fails closed when the root cannot be created", () => {
|
||||
expect(acquireLease(join(root, "\0bad"), "k")).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("releaseLease / heartbeatLease", () => {
|
||||
it("an old generation can neither release nor refresh its successor", () => {
|
||||
const old = acquireLease(root, "k")!;
|
||||
age(old, LEASE_STALE_MS + 5_000);
|
||||
const current = acquireLease(root, "k")!;
|
||||
releaseLease(old);
|
||||
expect(heartbeatLease(old)).toBe(false);
|
||||
expect(readdirSync(current.directory)).toEqual([current.owner]);
|
||||
expect(heartbeatLease(current)).toBe(true);
|
||||
});
|
||||
|
||||
it("a heartbeat keeps an aged lease from being reclaimed", () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
age(lease, LEASE_STALE_MS + 5_000);
|
||||
expect(heartbeatLease(lease)).toBe(true);
|
||||
expect(acquireLease(root, "k")).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("superviseSurvey", () => {
|
||||
function fakeAgent() {
|
||||
const child = Object.assign(new EventEmitter(), { kill: vi.fn() });
|
||||
const spawn = vi.fn().mockReturnValue(child);
|
||||
return { child, spawn };
|
||||
}
|
||||
|
||||
it("runs the agent and releases the lease when it exits", async () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
const { child, spawn } = fakeAgent();
|
||||
const run = superviseSurvey({ lease, bin: "/bin/agent", args: ["-p", "x"] }, spawn);
|
||||
expect(spawn).toHaveBeenCalledWith("/bin/agent", ["-p", "x"], {
|
||||
stdio: "ignore",
|
||||
windowsHide: true,
|
||||
});
|
||||
child.emit("exit", 0);
|
||||
await run.done;
|
||||
expect(readdirSync(root)).toEqual([]);
|
||||
});
|
||||
|
||||
it("heartbeats while the agent runs", async () => {
|
||||
vi.useFakeTimers();
|
||||
const lease = acquireLease(root, "k")!;
|
||||
const { child, spawn } = fakeAgent();
|
||||
const run = superviseSurvey({ lease, bin: "a", args: [], heartbeatMs: 1_000 }, spawn);
|
||||
age(lease, LEASE_STALE_MS + 5_000);
|
||||
vi.advanceTimersByTime(1_000);
|
||||
expect(acquireLease(root, "k")).toBeUndefined();
|
||||
child.emit("exit", 0);
|
||||
await run.done;
|
||||
});
|
||||
|
||||
it("kills the agent, and keeps its hands off the lease, once the lease is taken over", async () => {
|
||||
vi.useFakeTimers();
|
||||
const lease = acquireLease(root, "k")!;
|
||||
const { child, spawn } = fakeAgent();
|
||||
const run = superviseSurvey({ lease, bin: "a", args: [], heartbeatMs: 1_000 }, spawn);
|
||||
age(lease, LEASE_STALE_MS + 5_000); // the machine slept past the stale window
|
||||
const successor = acquireLease(root, "k")!;
|
||||
vi.advanceTimersByTime(1_000);
|
||||
await run.done;
|
||||
expect(child.kill).toHaveBeenCalled();
|
||||
child.emit("exit", null);
|
||||
expect(readdirSync(successor.directory)).toEqual([successor.owner]);
|
||||
});
|
||||
|
||||
it("does not start the agent if the lease was lost before it could", async () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
releaseLease(lease);
|
||||
const { spawn } = fakeAgent();
|
||||
await superviseSurvey({ lease, bin: "a", args: [] }, spawn).done;
|
||||
expect(spawn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("releases the lease when the agent cannot be spawned", async () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
const { child, spawn } = fakeAgent();
|
||||
const run = superviseSurvey({ lease, bin: "missing", args: [] }, spawn);
|
||||
child.emit("error", new Error("spawn missing ENOENT"));
|
||||
await run.done;
|
||||
expect(readdirSync(root)).toEqual([]);
|
||||
|
||||
const throwing = vi.fn(() => {
|
||||
throw new Error("EMFILE");
|
||||
});
|
||||
await superviseSurvey({ lease: acquireLease(root, "k")!, bin: "a", args: [] }, throwing).done;
|
||||
expect(readdirSync(root)).toEqual([]);
|
||||
});
|
||||
|
||||
it("stop() kills the agent and releases", async () => {
|
||||
const lease = acquireLease(root, "k")!;
|
||||
const { child, spawn } = fakeAgent();
|
||||
const run = superviseSurvey({ lease, bin: "a", args: [] }, spawn);
|
||||
run.stop();
|
||||
await run.done;
|
||||
expect(child.kill).toHaveBeenCalled();
|
||||
expect(readdirSync(root)).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,200 @@
|
||||
/**
|
||||
* Single-flight admission for the codebase survey (#4255): at most one survey per destination
|
||||
* (API + credential + bank) runs at a time, however many SessionStart hooks race the cold check.
|
||||
*
|
||||
* The lease is a lock DIRECTORY holding exactly one owner file named by a random token, and it is
|
||||
* kept alive by a HEARTBEAT: the holder rewrites the owner file's mtime every few seconds, and a
|
||||
* lease whose mtime is older than the stale threshold belongs to a dead holder and may be taken.
|
||||
* Only filesystem primitives that behave the same on macOS, Linux and Windows are used — no
|
||||
* `flock`/`fcntl` (advisory, fd-bound, unreliable over NFS), no `O_EXCL` (broken on NFS), and no
|
||||
* PID liveness probes (PIDs are recycled, `kill(pid, 0)` answers for whatever process now has the
|
||||
* number, and it cannot be probed across users at all).
|
||||
*
|
||||
* A heartbeat needs a holder that runs OUR code, and the survey agent (`claude -p`, `codex exec`,
|
||||
* …) does not — while the hook that won the lease exits immediately. So the survey runs under a
|
||||
* tiny detached supervisor (`survey-supervisor.ts` → `superviseSurvey`) that holds the lease for
|
||||
* exactly the agent's lifetime: it heartbeats, releases when the agent exits, and kills the agent
|
||||
* if it finds the lease was taken from it (a machine that slept past the stale window), so a
|
||||
* reclaimed lease never leaves two surveys running.
|
||||
*/
|
||||
import { spawn as realSpawn, type ChildProcess } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import {
|
||||
mkdirSync,
|
||||
mkdtempSync,
|
||||
readdirSync,
|
||||
renameSync,
|
||||
rmdirSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
unlinkSync,
|
||||
utimesSync,
|
||||
writeFileSync,
|
||||
} from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
/** How often the supervisor refreshes its lease. */
|
||||
export const LEASE_HEARTBEAT_MS = 5_000;
|
||||
/** A lease not refreshed for this long belongs to a dead holder. Six missed heartbeats: generous
|
||||
* enough for a loaded machine, short enough that a crashed survey is retried on the next session. */
|
||||
export const LEASE_STALE_MS = 30_000;
|
||||
|
||||
export interface SurveyLease {
|
||||
/** The lock directory: `<root>/survey-<key>.lock`. */
|
||||
directory: string;
|
||||
/** The owner file inside it — a random token, so a generation can only ever touch its own. */
|
||||
owner: string;
|
||||
}
|
||||
|
||||
/** Environment variable carrying the JSON `SurveySupervisorSpec` from the hook to the supervisor.
|
||||
*
|
||||
* NOT argv: the spec holds the whole agent command line (survey prompt, inline MCP config,
|
||||
* `--disallowedTools Bash Write …`), and a node process launched by the hook with that on its own
|
||||
* command line was SIGKILLed by endpoint security within milliseconds of starting — 5/5 runs on a
|
||||
* SentinelOne-managed Mac, vs 0/5 for the identical launch with the spec in the environment (#4255
|
||||
* repro). The supervisor deletes it before starting the agent, so it goes no further. */
|
||||
export const SURVEY_SPEC_ENV = "HINDSIGHT_SURVEY_SPEC";
|
||||
|
||||
/** What the hook hands the detached supervisor (via `SURVEY_SPEC_ENV`). */
|
||||
export interface SurveySupervisorSpec {
|
||||
lease: SurveyLease;
|
||||
bin: string;
|
||||
args: string[];
|
||||
heartbeatMs?: number;
|
||||
}
|
||||
|
||||
function tryRename(from: string, to: string): boolean {
|
||||
try {
|
||||
renameSync(from, to);
|
||||
return true;
|
||||
} catch {
|
||||
return false; // ENOTEMPTY / EEXIST / EPERM: someone else holds it
|
||||
}
|
||||
}
|
||||
|
||||
/** Remove only this generation's owner file. A replacement lease is nonempty, so the rmdir that
|
||||
* follows cannot take a newer owner with it even if one moved in between the two calls. */
|
||||
export function releaseLease(lease: SurveyLease): void {
|
||||
try {
|
||||
unlinkSync(join(lease.directory, lease.owner));
|
||||
} catch {
|
||||
return; // already released or taken over: never touch an unknown generation
|
||||
}
|
||||
try {
|
||||
rmdirSync(lease.directory);
|
||||
} catch {
|
||||
/* a new owner is already there */
|
||||
}
|
||||
}
|
||||
|
||||
/** Refresh the lease. `false` only when the owner file is gone — the lease was reclaimed from us.
|
||||
* Any other failure (a transient EBUSY/EPERM on Windows) is retried on the next beat; if it
|
||||
* persists the lease goes stale and the ENOENT arrives then. */
|
||||
export function heartbeatLease(lease: SurveyLease): boolean {
|
||||
try {
|
||||
const now = new Date();
|
||||
utimesSync(join(lease.directory, lease.owner), now, now);
|
||||
return true;
|
||||
} catch (error) {
|
||||
return (error as NodeJS.ErrnoException).code !== "ENOENT";
|
||||
}
|
||||
}
|
||||
|
||||
/** Clear the way for a new owner when the current lease is abandoned. Returns whether it may be
|
||||
* retried; the retry's own rename is what decides between racing reclaimers. */
|
||||
function reclaimIfStale(directory: string, staleMs: number): boolean {
|
||||
const owners = readdirSync(directory);
|
||||
if (owners.length === 0) {
|
||||
// A release interrupted between its unlink and its rmdir: unowned, and on Windows a rename
|
||||
// cannot replace even an empty directory.
|
||||
try {
|
||||
rmdirSync(directory);
|
||||
} catch {
|
||||
/* a new owner moved in; the retried rename will lose */
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (owners.length !== 1) return false; // not a layout we wrote: leave it alone
|
||||
const age = Date.now() - statSync(join(directory, owners[0])).mtimeMs;
|
||||
// A future mtime (the clock stepped back) is no proof of a live holder either: a live one
|
||||
// rewrites it with the current clock within one heartbeat.
|
||||
if (Math.abs(age) <= staleMs) return false;
|
||||
releaseLease({ directory, owner: owners[0] });
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Take the lease for `key` under `root`, or `undefined` when a live holder has it. Never throws:
|
||||
* anything that prevents proving admission counts as "held" — skipping a survey is cheap,
|
||||
* launching a duplicate one is not.
|
||||
*
|
||||
* The owner file is written into a private staging directory which is then RENAMED into place: a
|
||||
* rename onto a nonempty directory fails on every platform, so exactly one contender wins, and
|
||||
* unlike `mkdir` there is no instant where the lock exists without its owner.
|
||||
*/
|
||||
export function acquireLease(
|
||||
root: string,
|
||||
key: string,
|
||||
staleMs: number = LEASE_STALE_MS
|
||||
): SurveyLease | undefined {
|
||||
let staging: string | undefined;
|
||||
try {
|
||||
mkdirSync(root, { recursive: true, mode: 0o700 });
|
||||
staging = mkdtempSync(join(root, "claim-"));
|
||||
const owner = randomUUID();
|
||||
writeFileSync(join(staging, owner), "", { flag: "wx", mode: 0o600 });
|
||||
const directory = join(root, `survey-${key}.lock`);
|
||||
if (!tryRename(staging, directory)) {
|
||||
if (!reclaimIfStale(directory, staleMs)) return undefined;
|
||||
if (!tryRename(staging, directory)) return undefined; // another reclaimer won
|
||||
}
|
||||
return { directory, owner };
|
||||
} catch {
|
||||
return undefined;
|
||||
} finally {
|
||||
if (staging) rmSync(staging, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the survey agent while holding `spec.lease` (the body of the detached supervisor). Resolves
|
||||
* once the agent is gone; the lease is released when the agent exits and left alone when it was
|
||||
* lost. `stop()` ends the survey early (the supervisor's signal handlers).
|
||||
*/
|
||||
export function superviseSurvey(
|
||||
spec: SurveySupervisorSpec,
|
||||
spawnFn: typeof realSpawn = realSpawn
|
||||
): { done: Promise<void>; stop(): void } {
|
||||
let stop = () => {};
|
||||
const done = new Promise<void>((resolve) => {
|
||||
// The hook took the lease moments ago; if it is already gone, someone else is surveying.
|
||||
if (!heartbeatLease(spec.lease)) return resolve();
|
||||
let child: ChildProcess;
|
||||
try {
|
||||
child = spawnFn(spec.bin, spec.args, { stdio: "ignore", windowsHide: true });
|
||||
} catch {
|
||||
releaseLease(spec.lease);
|
||||
return resolve();
|
||||
}
|
||||
let finished = false;
|
||||
const finish = (release: boolean) => {
|
||||
if (finished) return;
|
||||
finished = true;
|
||||
clearInterval(timer);
|
||||
if (release) releaseLease(spec.lease);
|
||||
resolve();
|
||||
};
|
||||
const timer = setInterval(() => {
|
||||
if (heartbeatLease(spec.lease)) return;
|
||||
child.kill(); // the lease now belongs to another survey
|
||||
finish(false);
|
||||
}, spec.heartbeatMs ?? LEASE_HEARTBEAT_MS);
|
||||
child.on("error", () => finish(true));
|
||||
child.once("exit", () => finish(true));
|
||||
stop = () => {
|
||||
child.kill();
|
||||
finish(true);
|
||||
};
|
||||
});
|
||||
return { done, stop: () => stop() };
|
||||
}
|
||||
@@ -1,11 +1,44 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
resolveClaudeBin,
|
||||
startCodebaseSurvey,
|
||||
startCodebaseSurvey as startSurvey,
|
||||
SURVEY_AGENT,
|
||||
SURVEY_AGENT_CONFIG,
|
||||
SURVEY_PROMPT,
|
||||
} from "./survey";
|
||||
import { releaseLease, SURVEY_SPEC_ENV, type SurveySupervisorSpec } from "./survey-lease";
|
||||
|
||||
import { EventEmitter } from "node:events";
|
||||
import { mkdtempSync, rmSync, readdirSync, writeFileSync } from "node:fs";
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { pathToFileURL, fileURLToPath } from "node:url";
|
||||
import { buildSync } from "esbuild";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { resolveHostConfig } from "./host-client";
|
||||
import { resolveConfig } from "./config";
|
||||
|
||||
vi.mock("./host-client", () => ({ resolveHostConfig: vi.fn() }));
|
||||
let lockDir: string;
|
||||
beforeEach(() => {
|
||||
lockDir = mkdtempSync(join(tmpdir(), "hindsight-survey-test-"));
|
||||
vi.mocked(resolveHostConfig).mockReturnValue({
|
||||
cfg: resolveConfig({ apiUrl: "https://api.example.test", apiToken: "test-token" }),
|
||||
bankId: "bank-1",
|
||||
});
|
||||
});
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
rmSync(lockDir, { recursive: true, force: true });
|
||||
});
|
||||
function startCodebaseSurvey(repoDir: string, opts: Parameters<typeof startSurvey>[1] = {}) {
|
||||
return startSurvey(repoDir, {
|
||||
lease: { dir: lockDir },
|
||||
supervisorPath: "/x/survey-supervisor.js",
|
||||
...opts,
|
||||
});
|
||||
}
|
||||
|
||||
describe("resolveClaudeBin", () => {
|
||||
const ORIGINAL_ENV = process.env.HINDSIGHT_CLAUDE_BIN;
|
||||
@@ -15,17 +48,17 @@ describe("resolveClaudeBin", () => {
|
||||
else process.env.HINDSIGHT_CLAUDE_BIN = ORIGINAL_ENV;
|
||||
});
|
||||
|
||||
it("an explicit argument wins over everything", () => {
|
||||
it("an explicit argument wins over everything", async () => {
|
||||
process.env.HINDSIGHT_CLAUDE_BIN = "/env/claude";
|
||||
expect(resolveClaudeBin("/explicit/claude")).toBe("/explicit/claude");
|
||||
});
|
||||
|
||||
it("HINDSIGHT_CLAUDE_BIN env var wins when no explicit arg is given", () => {
|
||||
it("HINDSIGHT_CLAUDE_BIN env var wins when no explicit arg is given", async () => {
|
||||
process.env.HINDSIGHT_CLAUDE_BIN = "/env/claude";
|
||||
expect(resolveClaudeBin()).toBe("/env/claude");
|
||||
});
|
||||
|
||||
it("falls back to the bare 'claude' PATH lookup when nothing else resolves", () => {
|
||||
it("falls back to the bare 'claude' PATH lookup when nothing else resolves", async () => {
|
||||
delete process.env.HINDSIGHT_CLAUDE_BIN;
|
||||
const bin = resolveClaudeBin();
|
||||
expect(typeof bin).toBe("string");
|
||||
@@ -35,14 +68,33 @@ describe("resolveClaudeBin", () => {
|
||||
|
||||
describe("startCodebaseSurvey", () => {
|
||||
function fakeSpawn() {
|
||||
return vi.fn().mockReturnValue({ on: vi.fn(), unref: vi.fn() });
|
||||
return vi.fn().mockImplementation(() => {
|
||||
const child = Object.assign(new EventEmitter(), {
|
||||
pid: process.pid,
|
||||
unref: vi.fn(),
|
||||
kill: vi.fn(),
|
||||
});
|
||||
vi.spyOn(child, "on");
|
||||
queueMicrotask(() => child.emit("spawn"));
|
||||
return child;
|
||||
});
|
||||
}
|
||||
const yes = () => true;
|
||||
/** The survey agent is launched through the lease supervisor: decode the agent's bin + argv
|
||||
* from the supervisor's spec; the options are the real spawn's (the agent inherits them). */
|
||||
function launched(spawn: ReturnType<typeof fakeSpawn>, i = 0) {
|
||||
const [node, argv, options] = spawn.mock.calls[i];
|
||||
expect(node).toBe("node");
|
||||
// Only the script path on the command line: the payload rides in the environment (#4255).
|
||||
expect(argv).toEqual(["/x/survey-supervisor.js"]);
|
||||
const spec = JSON.parse(options.env[SURVEY_SPEC_ENV]) as SurveySupervisorSpec;
|
||||
return [spec.bin, spec.args, options, spec] as const;
|
||||
}
|
||||
|
||||
// ── claude recipe (the default / self-contained inline-MCP one) ────────────────────────────────
|
||||
it("claude: spawns the resolved binary with the expected argv, sandbox, and options", () => {
|
||||
it("claude: spawns the resolved binary with the expected argv, sandbox, and options", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
startCodebaseSurvey("/repo", {
|
||||
await startCodebaseSurvey("/repo", {
|
||||
model: "sonnet",
|
||||
mcpServerPath: "/x/mcp-server.js",
|
||||
claudeBin: "/bin/claude",
|
||||
@@ -51,7 +103,7 @@ describe("startCodebaseSurvey", () => {
|
||||
});
|
||||
|
||||
expect(spawn).toHaveBeenCalledTimes(1);
|
||||
const [bin, argv, options] = spawn.mock.calls[0];
|
||||
const [bin, argv, options] = launched(spawn);
|
||||
expect(bin).toBe("/bin/claude");
|
||||
|
||||
expect(argv).toContain("-p");
|
||||
@@ -91,24 +143,24 @@ describe("startCodebaseSurvey", () => {
|
||||
expect(child.unref).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("claude: defaults model to 'haiku' and --max-budget-usd to 2", () => {
|
||||
it("claude: defaults model to 'haiku' and --max-budget-usd to 2", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
startCodebaseSurvey("/repo", { claudeBin: "/bin/claude", spawn, exists: yes });
|
||||
const argv = spawn.mock.calls[0][1];
|
||||
await startCodebaseSurvey("/repo", { claudeBin: "/bin/claude", spawn, exists: yes });
|
||||
const argv = launched(spawn)[1];
|
||||
expect(argv[argv.indexOf("--model") + 1]).toBe("haiku");
|
||||
expect(argv[argv.indexOf("--max-budget-usd") + 1]).toBe("2");
|
||||
});
|
||||
|
||||
// ── codex recipe (read-only sandbox + inline -c MCP) ───────────────────────────────────────────
|
||||
it("codex: spawns `codex exec --sandbox read-only` with inline MCP overrides + the prompt", () => {
|
||||
it("codex: spawns `codex exec --sandbox read-only` with inline MCP overrides + the prompt", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
startCodebaseSurvey("/repo", {
|
||||
await startCodebaseSurvey("/repo", {
|
||||
harness: "codex",
|
||||
mcpServerPath: "/x/mcp-server.js",
|
||||
spawn,
|
||||
exists: (b) => b === "codex",
|
||||
});
|
||||
const [bin, argv, options] = spawn.mock.calls[0];
|
||||
const [bin, argv, options] = launched(spawn);
|
||||
expect(bin).toBe("codex");
|
||||
expect(argv.slice(0, 3)).toEqual(["exec", "--sandbox", "read-only"]);
|
||||
expect(argv).toContain(SURVEY_PROMPT);
|
||||
@@ -125,28 +177,28 @@ describe("startCodebaseSurvey", () => {
|
||||
});
|
||||
|
||||
// ── Antigravity recipe (plan read-only mode + global MCP config) ───────────────────────────────
|
||||
it("antigravity: spawns `agy -p` in plan mode", () => {
|
||||
it("antigravity: spawns `agy -p` in plan mode", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
startCodebaseSurvey("/repo", {
|
||||
await startCodebaseSurvey("/repo", {
|
||||
harness: "antigravity-cli",
|
||||
spawn,
|
||||
exists: (b) => b === "agy",
|
||||
});
|
||||
const [bin, argv, options] = spawn.mock.calls[0];
|
||||
const [bin, argv, options] = launched(spawn);
|
||||
expect(bin).toBe("agy");
|
||||
expect(argv).toEqual(["-p", SURVEY_PROMPT, "--mode=plan"]);
|
||||
expect(options.env.HINDSIGHT_DISABLE_HOOKS).toBe("1");
|
||||
});
|
||||
|
||||
// ── opencode recipe (our own read-only agent; tools from the loaded plugin) ────────────────────
|
||||
it("opencode: spawns `opencode run` under OUR survey agent, never the built-in plan agent", () => {
|
||||
it("opencode: spawns `opencode run` under OUR survey agent, never the built-in plan agent", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
startCodebaseSurvey("/repo", {
|
||||
await startCodebaseSurvey("/repo", {
|
||||
harness: "opencode",
|
||||
spawn,
|
||||
exists: (b) => b === "opencode",
|
||||
});
|
||||
const [bin, argv, options] = spawn.mock.calls[0];
|
||||
const [bin, argv, options] = launched(spawn);
|
||||
expect(bin).toBe("opencode");
|
||||
expect(argv).toEqual(["run", "--agent", SURVEY_AGENT, SURVEY_PROMPT]);
|
||||
// `plan` appends a read-only system-reminder that talks models out of the ingest call the
|
||||
@@ -157,7 +209,7 @@ describe("startCodebaseSurvey", () => {
|
||||
|
||||
// The recipe above is only safe because the agent it names is read-only. opencode drops denied
|
||||
// tools from the model's tool list entirely, so this ruleset IS the sandbox.
|
||||
it("the survey agent denies everything except reading and the one ingest tool", () => {
|
||||
it("the survey agent denies everything except reading and the one ingest tool", async () => {
|
||||
expect(SURVEY_AGENT_CONFIG.permission["*"]).toBe("deny");
|
||||
expect(SURVEY_AGENT_CONFIG.permission.hindsight_ingest_document).toBe("allow");
|
||||
const allowed = Object.entries(SURVEY_AGENT_CONFIG.permission)
|
||||
@@ -171,45 +223,49 @@ describe("startCodebaseSurvey", () => {
|
||||
});
|
||||
|
||||
// ── agent selection + fallback ─────────────────────────────────────────────────────────────────
|
||||
it("honors the HINDSIGHT_CODEX_BIN override for the codex binary", () => {
|
||||
it("honors the HINDSIGHT_CODEX_BIN override for the codex binary", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
process.env.HINDSIGHT_CODEX_BIN = "/opt/codex";
|
||||
try {
|
||||
startCodebaseSurvey("/repo", { harness: "codex", spawn, exists: (b) => b === "/opt/codex" });
|
||||
await startCodebaseSurvey("/repo", {
|
||||
harness: "codex",
|
||||
spawn,
|
||||
exists: (b) => b === "/opt/codex",
|
||||
});
|
||||
} finally {
|
||||
delete process.env.HINDSIGHT_CODEX_BIN;
|
||||
}
|
||||
expect(spawn.mock.calls[0][0]).toBe("/opt/codex");
|
||||
expect(launched(spawn)[0]).toBe("/opt/codex");
|
||||
});
|
||||
|
||||
it("falls back to another available agent when the preferred harness's CLI is missing", () => {
|
||||
it("falls back to another available agent when the preferred harness's CLI is missing", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
// Prefer Antigravity, but only codex is installed → survey runs under codex.
|
||||
startCodebaseSurvey("/repo", {
|
||||
await startCodebaseSurvey("/repo", {
|
||||
harness: "antigravity-cli",
|
||||
mcpServerPath: "/x/mcp-server.js",
|
||||
spawn,
|
||||
exists: (b) => b === "codex",
|
||||
});
|
||||
const [bin, argv] = spawn.mock.calls[0];
|
||||
const [bin, argv] = launched(spawn);
|
||||
expect(bin).toBe("codex");
|
||||
expect(argv[0]).toBe("exec");
|
||||
});
|
||||
|
||||
it("no capable agent found → no spawn (fail open; the git-log seed still ran)", () => {
|
||||
it("no capable agent found → no spawn (fail open; the git-log seed still ran)", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
startCodebaseSurvey("/repo", { harness: "antigravity-cli", spawn, exists: () => false });
|
||||
await startCodebaseSurvey("/repo", { harness: "antigravity-cli", spawn, exists: () => false });
|
||||
expect(spawn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── fail-safe ──────────────────────────────────────────────────────────────────────────────────
|
||||
it("fail-safe: a spawn that throws synchronously does not throw out of startCodebaseSurvey", () => {
|
||||
it("fail-safe: a spawn that throws synchronously does not throw out of startCodebaseSurvey", async () => {
|
||||
const spawn = vi.fn().mockImplementation(() => {
|
||||
throw new Error("spawn EMFILE");
|
||||
});
|
||||
expect(() =>
|
||||
await expect(
|
||||
startCodebaseSurvey("/repo", { claudeBin: "/bin/claude", spawn, exists: yes })
|
||||
).not.toThrow();
|
||||
).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("fail-safe: an async 'error' event on the child does not crash the caller", async () => {
|
||||
@@ -217,7 +273,186 @@ describe("startCodebaseSurvey", () => {
|
||||
const child = new EventEmitter() as InstanceType<typeof EventEmitter> & { unref: () => void };
|
||||
child.unref = vi.fn();
|
||||
const spawn = vi.fn().mockReturnValue(child);
|
||||
startCodebaseSurvey("/repo", { claudeBin: "/bin/claude", spawn, exists: yes });
|
||||
const launched = startCodebaseSurvey("/repo", { claudeBin: "/bin/claude", spawn, exists: yes });
|
||||
expect(() => child.emit("error", new Error("ENOENT"))).not.toThrow();
|
||||
await expect(launched).resolves.toBe(false);
|
||||
await expect(startCodebaseSurvey("/repo", { spawn: fakeSpawn(), exists: yes })).resolves.toBe(
|
||||
true
|
||||
);
|
||||
});
|
||||
|
||||
it("launches the agent under the detached lease supervisor, which owns the lease", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
await expect(
|
||||
startCodebaseSurvey("/repo", { claudeBin: "/bin/claude", spawn, exists: yes })
|
||||
).resolves.toBe(true);
|
||||
const [bin, , options, spec] = launched(spawn);
|
||||
expect(bin).toBe("/bin/claude");
|
||||
expect(options).toMatchObject({ cwd: "/repo", detached: true, stdio: "ignore" });
|
||||
expect(readdirSync(spec.lease.directory)).toEqual([spec.lease.owner]);
|
||||
expect(spawn.mock.results[0].value.unref).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("admits one concurrent survey and permits retry once the lease is released", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
const results = await Promise.all(
|
||||
Array.from({ length: 6 }, () => startCodebaseSurvey("/repo", { spawn, exists: yes }))
|
||||
);
|
||||
expect(spawn).toHaveBeenCalledTimes(1);
|
||||
expect(results.filter(Boolean)).toHaveLength(1);
|
||||
releaseLease(launched(spawn)[3].lease); // what the supervisor does when the agent exits
|
||||
await expect(startCodebaseSurvey("/repo", { spawn, exists: yes })).resolves.toBe(true);
|
||||
expect(spawn).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("releases the lease when the supervisor cannot be spawned", async () => {
|
||||
const child = Object.assign(new EventEmitter(), { unref: vi.fn() });
|
||||
const failing = vi.fn().mockReturnValue(child);
|
||||
const launch = startCodebaseSurvey("/repo", { spawn: failing, exists: yes });
|
||||
child.emit("error", Object.assign(new Error("spawn node ENOENT"), { code: "ENOENT" }));
|
||||
await expect(launch).resolves.toBe(false);
|
||||
await expect(startCodebaseSurvey("/repo", { spawn: fakeSpawn(), exists: yes })).resolves.toBe(
|
||||
true
|
||||
);
|
||||
});
|
||||
|
||||
it("keys admission by the resolved API, credential and bank, not repository or asking harness", async () => {
|
||||
const spawn = fakeSpawn();
|
||||
await expect(startCodebaseSurvey("/repo-a", { spawn, exists: yes })).resolves.toBe(true);
|
||||
// A second directory / fallback harness can write the same destination.
|
||||
await expect(
|
||||
startCodebaseSurvey("/repo-b", { harness: "codex", spawn, exists: yes })
|
||||
).resolves.toBe(false);
|
||||
for (const [apiUrl, apiToken, bankId] of [
|
||||
["https://other.example.test", "test-token", "bank-1"],
|
||||
["https://api.example.test", "other-token", "bank-1"],
|
||||
["https://api.example.test", "test-token", "bank-2"],
|
||||
]) {
|
||||
vi.mocked(resolveHostConfig).mockReturnValue({
|
||||
cfg: resolveConfig({ apiUrl, apiToken }),
|
||||
bankId,
|
||||
});
|
||||
await expect(startCodebaseSurvey("/repo-a", { spawn, exists: yes })).resolves.toBe(true);
|
||||
}
|
||||
expect(spawn).toHaveBeenCalledTimes(4);
|
||||
});
|
||||
|
||||
it("a disabled plugin launches nothing", async () => {
|
||||
vi.mocked(resolveHostConfig).mockReturnValue({
|
||||
cfg: { ...resolveConfig({}), disabled: true },
|
||||
bankId: "",
|
||||
});
|
||||
const spawn = fakeSpawn();
|
||||
await expect(startCodebaseSurvey("/repo", { spawn, exists: yes })).resolves.toBe(false);
|
||||
expect(spawn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// Real processes end to end: bundled hook + bundled supervisor + a stand-in agent.
|
||||
it("single-flights independent hooks, survives their exit, and recovers a dead supervisor", async () => {
|
||||
const build = (entry: string, outfile: string) =>
|
||||
buildSync({
|
||||
entryPoints: [fileURLToPath(new URL(entry, import.meta.url))],
|
||||
bundle: true,
|
||||
platform: "node",
|
||||
format: "esm",
|
||||
outfile,
|
||||
banner: {
|
||||
js: 'import { createRequire } from "node:module"; const require = createRequire(import.meta.url);',
|
||||
},
|
||||
});
|
||||
const bundle = join(lockDir, "survey.mjs");
|
||||
const supervisor = join(lockDir, "survey-supervisor.mjs");
|
||||
build("./survey.ts", bundle);
|
||||
build("../survey-supervisor.ts", supervisor);
|
||||
const config = join(lockDir, "config.json");
|
||||
writeFileSync(
|
||||
config,
|
||||
JSON.stringify({ bankId: "bank-1", apiUrl: "https://api.example.test", apiToken: "t" })
|
||||
);
|
||||
// The stand-in agent records its pid and its supervisor's, then idles like a long survey.
|
||||
const agent = `const fs = require('node:fs'); if (process.env.${SURVEY_SPEC_ENV}) fs.writeFileSync(${JSON.stringify(lockDir)} + '/leaked-spec', ''); fs.writeFileSync(${JSON.stringify(lockDir)} + '/pid-' + process.pid + '-' + process.ppid, ''); setInterval(() => {}, 1000);`;
|
||||
const staleMs = 2_000;
|
||||
const hook = `
|
||||
import { startCodebaseSurvey } from ${JSON.stringify(pathToFileURL(bundle).href)};
|
||||
import { spawn } from 'node:child_process';
|
||||
const started = await startCodebaseSurvey(${JSON.stringify(lockDir)}, {
|
||||
exists: () => true,
|
||||
mcpServerPath: '/unused-mcp-server.js',
|
||||
supervisorPath: ${JSON.stringify(supervisor)},
|
||||
lease: { dir: ${JSON.stringify(join(lockDir, "locks"))}, staleMs: ${staleMs}, heartbeatMs: 200 },
|
||||
// Swap the real agent for the stand-in; the supervisor itself runs for real.
|
||||
spawn: (_node, argv, options) => {
|
||||
const spec = { ...JSON.parse(options.env.${SURVEY_SPEC_ENV}), bin: process.execPath, args: ['-e', ${JSON.stringify(agent)}] };
|
||||
return spawn(process.execPath, argv, { ...options, env: { ...options.env, ${SURVEY_SPEC_ENV}: JSON.stringify(spec) } });
|
||||
},
|
||||
});
|
||||
console.log(JSON.stringify(started));
|
||||
`;
|
||||
const runHook = async () => {
|
||||
const { stdout } = await promisify(execFile)(
|
||||
process.execPath,
|
||||
["--input-type=module", "-e", hook],
|
||||
{
|
||||
env: {
|
||||
PATH: process.env.PATH,
|
||||
HOME: lockDir,
|
||||
HINDSIGHT_CONFIG: config,
|
||||
SystemRoot: process.env.SystemRoot,
|
||||
},
|
||||
}
|
||||
);
|
||||
return JSON.parse(stdout.trim()) as boolean;
|
||||
};
|
||||
const race = async () =>
|
||||
(await Promise.all(Array.from({ length: 6 }, runHook))).filter(Boolean).length;
|
||||
const agents = () =>
|
||||
readdirSync(lockDir)
|
||||
.filter((name) => name.startsWith("pid-"))
|
||||
.map((name) => {
|
||||
const [, pid, ppid] = name.split("-").map(Number);
|
||||
return { pid, ppid };
|
||||
});
|
||||
const alive = (pid: number) => {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
};
|
||||
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
|
||||
try {
|
||||
expect(await race()).toBe(1);
|
||||
await vi.waitFor(() => expect(agents()).toHaveLength(1));
|
||||
// The supervisor strips the spec before starting the agent.
|
||||
expect(readdirSync(lockDir)).not.toContain("leaked-spec");
|
||||
// Every hook has exited; past the stale window the heartbeat still holds the lease.
|
||||
await sleep(staleMs * 2);
|
||||
expect(await runHook()).toBe(false);
|
||||
|
||||
// The agent finishes: the supervisor releases, and exactly one new survey is admitted.
|
||||
const first = agents()[0];
|
||||
process.kill(first.pid, "SIGTERM");
|
||||
await vi.waitFor(() => expect(alive(first.ppid)).toBe(false));
|
||||
expect(await race()).toBe(1);
|
||||
await vi.waitFor(() => expect(agents().filter((a) => alive(a.pid))).toHaveLength(1));
|
||||
|
||||
// The supervisor dies without releasing: its lease goes stale, and one reclaimer wins.
|
||||
const second = agents().find((a) => alive(a.pid))!;
|
||||
process.kill(second.ppid, "SIGKILL");
|
||||
expect(await runHook()).toBe(false); // not stale yet
|
||||
await sleep(staleMs * 2);
|
||||
expect(await race()).toBe(1);
|
||||
} finally {
|
||||
for (const { pid, ppid } of agents()) {
|
||||
for (const p of [pid, ppid]) {
|
||||
try {
|
||||
process.kill(p, "SIGKILL");
|
||||
} catch {
|
||||
/* already exited */
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}, 30_000);
|
||||
});
|
||||
|
||||
@@ -37,11 +37,20 @@
|
||||
* missing binary or a spawn failure must silently no-op, never crash the caller.
|
||||
*/
|
||||
import { spawn as realSpawn } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import { existsSync } from "node:fs";
|
||||
import { homedir } from "node:os";
|
||||
import { homedir, tmpdir } from "node:os";
|
||||
import { dirname, join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { binOnPath } from "./util";
|
||||
import { resolveHostConfig } from "./host-client";
|
||||
import {
|
||||
acquireLease,
|
||||
LEASE_STALE_MS,
|
||||
releaseLease,
|
||||
SURVEY_SPEC_ENV,
|
||||
type SurveySupervisorSpec,
|
||||
} from "./survey-lease";
|
||||
|
||||
/** Deterministic doc ids of the survey's findings (its fixed titles slugified by
|
||||
* hindsight_ingest_document). Their presence in the bank = the survey actually FINISHED —
|
||||
@@ -282,9 +291,12 @@ function buildSurveyPlan(
|
||||
* Spawn a DETACHED headless agent to survey `repoDir` and ingest structural findings via the
|
||||
* `hindsight_ingest_document` tool. Runs the survey under the current harness's own CLI when
|
||||
* available, else falls back to any available agent (claude → codex → antigravity → opencode — claude and
|
||||
* codex first because their inline-MCP recipes are self-contained). Fire-and-forget; never throws.
|
||||
* codex first because their inline-MCP recipes are self-contained). The agent runs under the
|
||||
* detached lease supervisor (survey-supervisor.ts), so only one survey per destination runs at a
|
||||
* time (#4255). Resolves once launched, not when the survey finishes; false means no launch.
|
||||
* Never throws.
|
||||
*/
|
||||
export function startCodebaseSurvey(
|
||||
export async function startCodebaseSurvey(
|
||||
repoDir: string,
|
||||
opts: {
|
||||
harness?: SurveyHarness;
|
||||
@@ -294,8 +306,10 @@ export function startCodebaseSurvey(
|
||||
claudeBin?: string;
|
||||
spawn?: typeof realSpawn;
|
||||
exists?: (bin: string) => boolean; // seam for tests
|
||||
supervisorPath?: string;
|
||||
lease?: { dir?: string; staleMs?: number; heartbeatMs?: number }; // seam for tests
|
||||
} = {}
|
||||
): void {
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const spawnFn = opts.spawn ?? realSpawn;
|
||||
const exists = opts.exists ?? binOnPath;
|
||||
@@ -324,22 +338,62 @@ export function startCodebaseSurvey(
|
||||
budgetUsd: opts.budgetUsd,
|
||||
mcpServerPath,
|
||||
});
|
||||
const child = spawnFn(plan.bin, plan.args, {
|
||||
cwd: repoDir,
|
||||
detached: true,
|
||||
stdio: "ignore",
|
||||
windowsHide: true,
|
||||
env: plan.env,
|
||||
// Use the SAME resolver as the selected agent's MCP/plugin, including fallback harnesses
|
||||
// and bank overrides. Hash credentials too: equal bank ids on one API can belong to
|
||||
// different tenants, but neither tokens nor bank names should appear in scratch paths.
|
||||
const { cfg, bankId } = resolveHostConfig(harness, repoDir);
|
||||
if (cfg.disabled) return false;
|
||||
const key = createHash("sha256")
|
||||
.update(JSON.stringify([cfg.apiUrl.replace(/\/+$/, ""), cfg.apiToken ?? "", bankId]))
|
||||
.digest("hex");
|
||||
const lease = acquireLease(
|
||||
opts.lease?.dir ?? join(tmpdir(), "hindsight-coding-agent", "surveys"),
|
||||
key,
|
||||
opts.lease?.staleMs ?? LEASE_STALE_MS
|
||||
);
|
||||
if (!lease) return false;
|
||||
const spec: SurveySupervisorSpec = {
|
||||
lease,
|
||||
bin: plan.bin,
|
||||
args: plan.args,
|
||||
...(opts.lease?.heartbeatMs ? { heartbeatMs: opts.lease.heartbeatMs } : {}),
|
||||
};
|
||||
const supervisorPath =
|
||||
opts.supervisorPath ??
|
||||
join(dirname(fileURLToPath(import.meta.url)), "survey-supervisor.js");
|
||||
return await new Promise<boolean>((resolve) => {
|
||||
try {
|
||||
// The agent inherits the supervisor's cwd and env. The spec rides in the environment,
|
||||
// never argv — see SURVEY_SPEC_ENV for the endpoint-security kill that argv triggers.
|
||||
const child = spawnFn("node", [supervisorPath], {
|
||||
cwd: repoDir,
|
||||
detached: true,
|
||||
stdio: "ignore",
|
||||
windowsHide: true,
|
||||
env: { ...plan.env, [SURVEY_SPEC_ENV]: JSON.stringify(spec) },
|
||||
});
|
||||
let spawned = false;
|
||||
// spawn() failures (node not found, EACCES, sandboxes) arrive as an async 'error' event;
|
||||
// unhandled, it would crash the caller.
|
||||
child.on("error", () => {
|
||||
if (spawned) return; // the supervisor owns the lease now
|
||||
releaseLease(lease);
|
||||
resolve(false);
|
||||
});
|
||||
child.once("spawn", () => {
|
||||
spawned = true;
|
||||
child.unref();
|
||||
resolve(true);
|
||||
});
|
||||
} catch {
|
||||
releaseLease(lease);
|
||||
resolve(false);
|
||||
}
|
||||
});
|
||||
// spawn() failures (binary not found, EACCES, sandboxed environments) often arrive
|
||||
// ASYNCHRONOUSLY as an 'error' event on the child, not as a synchronous throw — an unhandled
|
||||
// 'error' event would crash the caller. Swallow it: fire-and-forget best-effort.
|
||||
child.on("error", () => {});
|
||||
child.unref();
|
||||
return; // one survey agent is enough
|
||||
}
|
||||
// No capable agent found — fail open (the git-log seed already ran; the survey is a bonus).
|
||||
} catch {
|
||||
/* best-effort: a failed spawn must not break the caller */
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* Hold the codebase-survey lease for the lifetime of the survey agent. Spawned DETACHED by
|
||||
* core/survey.ts, which hands over the `SurveySupervisorSpec` in the `SURVEY_SPEC_ENV` variable
|
||||
* (not argv — see there), because the hook that won the lease exits immediately and the agent
|
||||
* itself cannot heartbeat. See core/survey-lease.ts.
|
||||
*/
|
||||
import { SURVEY_SPEC_ENV, superviseSurvey, type SurveySupervisorSpec } from "./core/survey-lease";
|
||||
|
||||
const spec = JSON.parse(process.env[SURVEY_SPEC_ENV] ?? "") as SurveySupervisorSpec;
|
||||
delete process.env[SURVEY_SPEC_ENV]; // the agent (and its MCP server) inherit this environment
|
||||
const run = superviseSurvey(spec);
|
||||
for (const signal of ["SIGTERM", "SIGINT", "SIGHUP"] as const) process.once(signal, run.stop);
|
||||
void run.done; // the agent and the heartbeat keep this process alive until it resolves
|
||||
@@ -57,6 +57,8 @@ export default defineConfig({
|
||||
"droid-stop-hook": "src/droid-stop-hook.ts",
|
||||
// Spawned DETACHED to start the local daemon — a cold start outlives every hook timeout.
|
||||
"daemon-start": "src/daemon-start.ts",
|
||||
// Spawned DETACHED to hold the codebase-survey lease for the survey agent's lifetime.
|
||||
"survey-supervisor": "src/survey-supervisor.ts",
|
||||
"mcp-server": "src/mcp-server.ts",
|
||||
"hindsight-seed": "src/hindsight-seed.ts",
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user