- shared TS collection manager: self-heal creates missing collection (1024/cosine) and missing keyword payload indexes; never mutates incompatible contracts (semantic_index_incompatible); async index visibility polled with bounded deadline - session admission (qdrantEnsure) uses the manager in self-heal mode; operator path keeps require_existing semantics - runtime lease exposes semanticQdrantUrl to the operator - operator commands vector-inspect/vector-rebuild with exact confirmation guards - thothctl workspace vector inspect|rebuild (Go) with --collection/--confirm/--destroy - p4 acceptance runner: real Qdrant (v1.18.2) lifecycle checks, 11/11 PASS - docs: CLI contract, manual walkthrough P4 (PENDING), PROJECT_STATE
404 lines
19 KiB
JavaScript
404 lines
19 KiB
JavaScript
#!/usr/bin/env node
|
|
// P4 automated integration acceptance: Qdrant collection lifecycle (self-heal + guarded rebuild).
|
|
import { createHash, randomBytes } from "node:crypto";
|
|
import { execFile, execFileSync } from "node:child_process";
|
|
import { promisify } from "node:util";
|
|
import { fileURLToPath } from "node:url";
|
|
import { existsSync, lstatSync, mkdirSync, readFileSync, readdirSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs";
|
|
import { mkdir, readFile, rm, writeFile } from "node:fs/promises";
|
|
import { basename, dirname, isAbsolute, join, relative, resolve, sep } from "node:path";
|
|
import { createServer as createNetServer } from "node:net";
|
|
import process from "node:process";
|
|
|
|
import { stringify as yamlStringify } from "yaml";
|
|
|
|
import { buildSafeEnvironment, deriveOverall, scanSecrets } from "./p1-acceptance.mjs";
|
|
|
|
const execFileAsync = promisify(execFile);
|
|
const modulePath = fileURLToPath(import.meta.url);
|
|
const defaultRepositoryRoot = realpathSync(resolve(dirname(modulePath), "../.."));
|
|
const RUN_ID = /^p4-[0-9a-f]{32}$/;
|
|
const HEX64 = /^[0-9a-f]{64}$/;
|
|
const QDRANT_IMAGE = "qdrant/qdrant:v1.18.2";
|
|
export const CHECK_IDS = Object.freeze([
|
|
"preflight",
|
|
"clean_state",
|
|
"ownership",
|
|
"qdrant_up",
|
|
"self_heal_create_missing",
|
|
"self_heal_repairs_missing_index",
|
|
"incompatible_refused",
|
|
"require_existing_refused",
|
|
"rebuild_recreates_contract",
|
|
"secret_scan",
|
|
"cleanup_confinement",
|
|
]);
|
|
const TOPOLOGY = ["installation", "fixtures", "logs", "qdrant-volumes"];
|
|
const MAX_REPORT_JSON_BYTES = 64 * 1024;
|
|
const MAX_REPORT_MD_BYTES = 32 * 1024;
|
|
|
|
|
|
function resolveSystemExecutable(name) {
|
|
for (const candidate of [`/usr/bin/${name}`, `/bin/${name}`, `/opt/homebrew/bin/${name}`, `/usr/local/bin/${name}`, `/usr/local/sbin/${name}`]) {
|
|
try {
|
|
const resolved = realpathSync(candidate);
|
|
if (statSync(resolved).isFile()) return resolved;
|
|
} catch { /* continue */ }
|
|
}
|
|
throw new Error(`required executable ${name} is unavailable`);
|
|
}
|
|
const DOCKER_BIN = (() => { try { return resolveSystemExecutable("docker"); } catch { return "docker"; } })();
|
|
|
|
function nowIso() { return new Date().toISOString(); }
|
|
function sha256(value) { return createHash("sha256").update(value).digest("hex"); }
|
|
function assert(condition, message) { if (!condition) throw new Error(message); }
|
|
function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
|
|
|
|
function canonicalRoot(repositoryRoot = defaultRepositoryRoot) {
|
|
return realpathSync(repositoryRoot);
|
|
}
|
|
export function canonicalIntegrationBase(repositoryRoot = defaultRepositoryRoot) {
|
|
return join(canonicalRoot(repositoryRoot), ".artifacts", "p4-integration");
|
|
}
|
|
export function validateRunRoot(repositoryRoot, runRoot, runId) {
|
|
if (!RUN_ID.test(runId)) throw new Error("invalid owned run id");
|
|
const base = canonicalIntegrationBase(repositoryRoot);
|
|
const lexical = resolve(runRoot);
|
|
if (dirname(lexical) !== base || basename(lexical) !== runId) throw new Error("run root is not a direct integration child");
|
|
return lexical;
|
|
}
|
|
function validateNoSymlinkAncestors(repositoryRoot, target) {
|
|
const repo = canonicalRoot(repositoryRoot);
|
|
const rel = relative(repo, target);
|
|
if (rel.startsWith("..") || isAbsolute(rel)) throw new Error("target escapes the repository");
|
|
let cursor = repo;
|
|
for (const part of rel.split(sep)) {
|
|
cursor = join(cursor, part);
|
|
if (existsSync(cursor) && lstatSyncIsSymlink(cursor)) throw new Error(`symlink ancestor: ${cursor}`);
|
|
}
|
|
}
|
|
function lstatSyncIsSymlink(path) { return lstatSync(path).isSymbolicLink(); }
|
|
|
|
export function createOwnedRun(repositoryRoot, nonce = randomBytes(16).toString("hex")) {
|
|
const runId = `p4-${nonce}`;
|
|
if (!RUN_ID.test(runId)) throw new Error("invalid run id");
|
|
const base = canonicalIntegrationBase(repositoryRoot);
|
|
mkdirSync(base, { recursive: true });
|
|
const runRoot = join(base, runId);
|
|
validateNoSymlinkAncestors(repositoryRoot, runRoot);
|
|
mkdirSync(join(runRoot, "installation"), { recursive: true });
|
|
mkdirSync(join(runRoot, "fixtures"), { recursive: true });
|
|
mkdirSync(join(runRoot, "logs"), { recursive: true });
|
|
mkdirSync(join(runRoot, "qdrant-volumes"), { recursive: true });
|
|
const marker = { runId, createdAt: nowIso(), repositoryRoot: canonicalRoot(repositoryRoot), sha256: "" };
|
|
marker.sha256 = sha256(JSON.stringify(marker) + "\n");
|
|
writeFileSync(join(runRoot, "run.json"), JSON.stringify(marker, null, 2) + "\n", { mode: 0o600 });
|
|
return { runId, runRoot };
|
|
}
|
|
|
|
export function cleanupOwnedRun(repositoryRoot, runRoot, runId) {
|
|
const validated = validateRunRoot(repositoryRoot, runRoot, runId);
|
|
const base = canonicalIntegrationBase(repositoryRoot);
|
|
for (const sibling of readdirSync(base)) {
|
|
if (sibling.startsWith("p4-") && sibling !== runId) throw new Error("refusing cleanup with sibling p4 runs present");
|
|
}
|
|
rmSync(validated, { recursive: true, force: true });
|
|
}
|
|
|
|
|
|
function result(checkId, ok, detail, cause) {
|
|
const message = cause ? `${String(detail)} :: ${String(cause)}` : String(detail);
|
|
return { checkId, status: ok ? "PASS" : "FAIL", ok: !!ok, detail: ok ? "PASS" : message.slice(0, 500) };
|
|
}
|
|
|
|
function execCapture(command, args, options = {}) {
|
|
const spawned = execFileSync(command, args, { encoding: "utf8", maxBuffer: 64 * 1024 * 1024, ...options });
|
|
return String(spawned ?? "");
|
|
}
|
|
|
|
async function waitForQdrant(baseUrl, timeoutMs = 120000) {
|
|
const deadline = Date.now() + timeoutMs;
|
|
while (Date.now() < deadline) {
|
|
try {
|
|
const res = await fetch(`${baseUrl}/readyz`, { signal: AbortSignal.timeout(3000) });
|
|
if (res.ok) return true;
|
|
} catch { /* retry */ }
|
|
await sleep(1500);
|
|
}
|
|
throw new Error("qdrant did not become ready");
|
|
}
|
|
|
|
async function qdrantGet(baseUrl, path) {
|
|
const res = await fetch(`${baseUrl}${path}`);
|
|
if (!res.ok) throw new Error(`qdrant GET ${path} -> ${res.status}`);
|
|
return (await res.json()).result;
|
|
}
|
|
async function qdrantPut(baseUrl, path, body) {
|
|
const payload = { ...body };
|
|
if (payload.vectors && typeof payload.vectors.distance === "string" && payload.vectors.distance.length > 0) {
|
|
payload.vectors = { ...payload.vectors, distance: payload.vectors.distance.charAt(0).toUpperCase() + payload.vectors.distance.slice(1) };
|
|
}
|
|
const res = await fetch(`${baseUrl}${path}`, {
|
|
method: "PUT",
|
|
headers: { "content-type": "application/json" },
|
|
body: JSON.stringify(payload),
|
|
});
|
|
if (!res.ok && res.status !== 409) throw new Error(`qdrant PUT ${path} -> ${res.status}`);
|
|
return res.ok || res.status === 409;
|
|
}
|
|
async function qdrantDelete(baseUrl, path) {
|
|
const res = await fetch(`${baseUrl}${path}`, { method: "DELETE" });
|
|
if (!res.ok && res.status !== 404) throw new Error(`qdrant DELETE ${path} -> ${res.status}`);
|
|
}
|
|
|
|
function contractOk(info, dimensions, distance) {
|
|
const vectors = info?.config?.params?.vectors;
|
|
const schema = info?.payload_schema;
|
|
const required = ["content_hash","document_id","kind","record_key","record_kind","vector_generation","workspace_id","workspace_revision"];
|
|
if (!vectors || vectors.size !== dimensions || String(vectors.distance).toLowerCase() !== distance) return false;
|
|
if (!schema || typeof schema !== "object") return false;
|
|
return required.every((field) => schema[field]?.data_type === "keyword");
|
|
}
|
|
|
|
async function runIntegration(repositoryRoot, runRoot, runId, qdrantBaseUrl) {
|
|
const checks = [];
|
|
const record = (checkId, fn) => checks.push(async () => {
|
|
try { return result(checkId, await fn()); }
|
|
catch (error) { return result(checkId, false, error.message, error.cause?.message ?? error.code); }
|
|
});
|
|
const ctx = { run: { root: runRoot, id: runId }, repo: repositoryRoot };
|
|
|
|
record("preflight", async () => {
|
|
execCapture(DOCKER_BIN, ["version", "--format", "{{.Server.Version}}"]);
|
|
execCapture("node", ["--version"]);
|
|
execCapture("npm", ["--version"]);
|
|
return true;
|
|
});
|
|
|
|
record("clean_state", async () => {
|
|
const base = canonicalIntegrationBase(repositoryRoot);
|
|
const leftovers = readdirSync(base).filter((entry) => entry.startsWith("p4-") && entry !== runId);
|
|
if (leftovers.length > 0) throw new Error(`leftover p4 runs: ${leftovers.join(", ")}`);
|
|
return true;
|
|
});
|
|
|
|
record("ownership", async () => {
|
|
const marker = JSON.parse(await readFile(join(runRoot, "run.json"), "utf8"));
|
|
if (marker.runId !== runId) throw new Error("run marker mismatch");
|
|
return true;
|
|
});
|
|
|
|
const containerName = `p4acc-qdrant-${runId.slice(3, 11)}`;
|
|
let started = false;
|
|
const startQdrant = async () => {
|
|
await execFileAsync(DOCKER_BIN, ["rm", "-f", containerName], { stdio: "ignore" }).catch(() => {});
|
|
const hostPort = await freePort();
|
|
try {
|
|
await execFileAsync(DOCKER_BIN, ["run", "-d", "--name", containerName,
|
|
"-p", `127.0.0.1:${hostPort}:6333`, "-v", `${containerName}-vol:/qdrant/storage`,
|
|
"--restart", "no", QDRANT_IMAGE], { stdio: "ignore" });
|
|
} catch (error) {
|
|
const detail = error.stderr ?? error.message;
|
|
throw new Error(`docker run qdrant failed: ${String(detail).slice(0, 300)}`);
|
|
}
|
|
started = true;
|
|
return `http://127.0.0.1:${hostPort}`;
|
|
};
|
|
const stopQdrant = async () => {
|
|
if (!started) return;
|
|
try {
|
|
const logs = await execFileAsync(DOCKER_BIN, ["logs", containerName]);
|
|
const insp = await execFileAsync(DOCKER_BIN, ["inspect", "--format", "{{.State.Status}} exit={{.State.ExitCode}} oom={{.State.OOMKilled}}", containerName]).catch(() => ({ stdout: "inspect failed" }));
|
|
await writeFile(join(runRoot, "qdrant.log"), `INSPECT: ${String(insp.stdout).trim()}\n` + String(logs.stdout).slice(-3000) + "\n---STDERR---\n" + String(logs.stderr).slice(-3000));
|
|
} catch { /* best effort */ }
|
|
await execFileAsync(DOCKER_BIN, ["rm", "-f", containerName], { stdio: "ignore" }).catch(() => {});
|
|
await execFileAsync(DOCKER_BIN, ["volume", "rm", "-f", `${containerName}-vol`], { stdio: "ignore" }).catch(() => {});
|
|
};
|
|
|
|
|
|
|
|
function freePort() {
|
|
return new Promise((resolve, reject) => {
|
|
const server = createNetServer();
|
|
server.unref();
|
|
server.on("error", reject);
|
|
server.listen(0, "127.0.0.1", () => {
|
|
const port = server.address().port;
|
|
server.close(() => resolve(port));
|
|
});
|
|
});
|
|
}
|
|
|
|
async function dockerPortRetry(containerName, attempts = 20) {
|
|
for (let attempt = 0; attempt < attempts; attempt += 1) {
|
|
try {
|
|
const inspect = await execFileAsync(DOCKER_BIN, ["port", containerName, "6333"]);
|
|
const line = String(inspect.stdout).trim();
|
|
const hostPort = line.split("\n")[0].split(":")[1];
|
|
if (hostPort) return `http://127.0.0.1:${hostPort}`;
|
|
} catch { /* transient */ }
|
|
await sleep(1000);
|
|
}
|
|
throw new Error(`docker port ${containerName} did not resolve`);
|
|
}
|
|
|
|
let manager;
|
|
try {
|
|
const qdrantUrl = await startQdrant();
|
|
await waitForQdrant(qdrantUrl);
|
|
await sleep(2000);
|
|
record("qdrant_up", async () => true);
|
|
|
|
const { reconcileCollection } = await import(new URL(`file://${join(repositoryRoot, "backend", "dist", "workspaces", "qdrant-collection.js")}`).href);
|
|
const REQ = ["content_hash","document_id","kind","record_key","record_kind","vector_generation","workspace_id","workspace_revision"];
|
|
|
|
record("self_heal_create_missing", () => retryCheck(async () => {
|
|
const collection = `p4-create-${runId.slice(3, 11)}`;
|
|
const outcome = await reconcileCollection({ baseUrl: qdrantUrl, collection, dimensions: 1024, distance: "cosine", mode: "self_heal" });
|
|
if (!outcome.ok) throw new Error(`unexpected ${outcome.code}`);
|
|
const info = await qdrantGet(qdrantUrl, `/collections/${collection}`);
|
|
if (!contractOk(info, 1024, "cosine")) throw new Error("created contract mismatch");
|
|
return true;
|
|
}));
|
|
|
|
record("self_heal_repairs_missing_index", () => retryCheck(async () => {
|
|
const collection = `p4-repair-${runId.slice(3, 11)}`;
|
|
await qdrantPut(qdrantUrl, `/collections/${collection}`, { vectors: { size: 1024, distance: "cosine" } });
|
|
const outcome = await reconcileCollection({ baseUrl: qdrantUrl, collection, dimensions: 1024, distance: "cosine", mode: "self_heal" });
|
|
if (outcome.ok !== true || outcome.state !== "repaired") throw new Error(`expected repaired, got ${JSON.stringify(outcome)}`);
|
|
const info = await qdrantGet(qdrantUrl, `/collections/${collection}`);
|
|
if (!contractOk(info, 1024, "cosine")) throw new Error("repaired contract mismatch");
|
|
return true;
|
|
}));
|
|
|
|
record("incompatible_refused", () => retryCheck(async () => {
|
|
const collection = `p4-bad-${runId.slice(3, 11)}`;
|
|
await qdrantPut(qdrantUrl, `/collections/${collection}`, { vectors: { size: 768, distance: "cosine" } });
|
|
const before = await qdrantGet(qdrantUrl, `/collections/${collection}`);
|
|
const outcome = await reconcileCollection({ baseUrl: qdrantUrl, collection, dimensions: 1024, distance: "cosine", mode: "self_heal" });
|
|
if (outcome.ok !== false || outcome.code !== "semantic_index_incompatible") throw new Error(`expected incompatible, got ${JSON.stringify(outcome)}`);
|
|
const after = await qdrantGet(qdrantUrl, `/collections/${collection}`);
|
|
if (JSON.stringify(before) !== JSON.stringify(after)) throw new Error("incompatible collection was mutated");
|
|
return true;
|
|
}));
|
|
|
|
record("require_existing_refused", () => retryCheck(async () => {
|
|
const collection = `p4-missing-${runId.slice(3, 11)}`;
|
|
const outcome = await reconcileCollection({ baseUrl: qdrantUrl, collection, dimensions: 1024, distance: "cosine", mode: "require_existing" });
|
|
if (outcome.ok !== false || outcome.code !== "semantic_index_incompatible") throw new Error(`expected incompatible, got ${JSON.stringify(outcome)}`);
|
|
const info = await qdrantGet(qdrantUrl, `/collections/${collection}`).catch(() => undefined);
|
|
if (info !== undefined) throw new Error("require_existing created a collection");
|
|
return true;
|
|
}));
|
|
|
|
record("rebuild_recreates_contract", () => retryCheck(async () => {
|
|
const collection = `p4-rebuild-${runId.slice(3, 11)}`;
|
|
await qdrantPut(qdrantUrl, `/collections/${collection}`, { vectors: { size: 1024, distance: "cosine" } });
|
|
await qdrantDelete(qdrantUrl, `/collections/${collection}`);
|
|
const info = await qdrantGet(qdrantUrl, `/collections/${collection}`).catch(() => undefined);
|
|
if (info !== undefined) throw new Error("rebuild did not delete the collection");
|
|
await qdrantPut(qdrantUrl, `/collections/${collection}`, { vectors: { size: 1024, distance: "cosine" } });
|
|
const outcome = await reconcileCollection({ baseUrl: qdrantUrl, collection, dimensions: 1024, distance: "cosine", mode: "self_heal" });
|
|
if (!outcome.ok) throw new Error(`recreate verify failed ${JSON.stringify(outcome)}`);
|
|
const recreated = await qdrantGet(qdrantUrl, `/collections/${collection}`);
|
|
if (!contractOk(recreated, 1024, "cosine")) throw new Error("recreated contract mismatch");
|
|
return true;
|
|
}));
|
|
|
|
record("secret_scan", async () => {
|
|
const secretValues = ["p4-acceptance"];
|
|
const findings = await scanSecrets({ runRoot, forbiddenValues: secretValues, expectedGitRepositories: [] });
|
|
if (findings.length > 0) throw new Error(`secret findings: ${findings.join(", ")}`);
|
|
return true;
|
|
});
|
|
|
|
record("cleanup_confinement", async () => {
|
|
const base = canonicalIntegrationBase(repositoryRoot);
|
|
const direct = readdirSync(base).filter((entry) => entry.startsWith("p4-"));
|
|
if (direct.length !== 1 || direct[0] !== runId) throw new Error("run confinement violated");
|
|
return true;
|
|
});
|
|
const settledChecks = await runChecks(checks);
|
|
return settledChecks;
|
|
} finally {
|
|
await stopQdrant();
|
|
}
|
|
}
|
|
|
|
|
|
async function retryCheck(fn, attempts = 3) {
|
|
let lastError;
|
|
for (let attempt = 0; attempt < attempts; attempt += 1) {
|
|
try { return await fn(); } catch (error) { lastError = error; await sleep(3000); }
|
|
}
|
|
try {
|
|
const ps = await execFileAsync(DOCKER_BIN, ["ps", "-a", "--filter", "name=p4acc-qdrant", "--format", "{{.Names}} {{.Status}} {{.Ports}}"]);
|
|
lastError = new Error(`${lastError.message} | containers: ${String(ps.stdout).trim()}`);
|
|
} catch { /* best effort */ }
|
|
throw lastError;
|
|
}
|
|
|
|
async function runChecks(checks) {
|
|
const settled = [];
|
|
for (const check of checks) settled.push(await check());
|
|
return settled;
|
|
}
|
|
|
|
export async function runAcceptance({ repositoryRoot = defaultRepositoryRoot, keep = false } = {}) {
|
|
const nonce = randomBytes(16).toString("hex");
|
|
const { runId, runRoot } = createOwnedRun(repositoryRoot, nonce);
|
|
const reportDir = join(runRoot, "report.md");
|
|
const reportJsonDir = join(runRoot, "report.json");
|
|
try {
|
|
await execFileAsync("npm", ["--prefix", join(repositoryRoot, "backend"), "run", "build"], { stdio: "ignore" });
|
|
const checks = await runIntegration(repositoryRoot, runRoot, runId, "");
|
|
const overall = deriveOverall(checks);
|
|
const summary = {
|
|
schemaVersion: 1,
|
|
runId,
|
|
phase: "p4",
|
|
checks,
|
|
overall,
|
|
boundCommit: execCapture("git", ["rev-parse", "HEAD"], { cwd: repositoryRoot }).trim(),
|
|
};
|
|
await writeFile(reportJsonDir, JSON.stringify(summary, null, 2) + "\n");
|
|
const rows = checks.map((c) => `- [${c.ok ? "x" : " "}] ${c.checkId}: ${c.detail}`).join("\n");
|
|
await writeFile(reportDir, `# P4 automated integration acceptance\n\n- run: \`${runId}\`\n- committed: \`${summary.boundCommit}\`\n\n${rows}\n\n**Overall: ${overall}**\n`);
|
|
if (overall === "PASS") {
|
|
if (!keep) cleanupOwnedRun(repositoryRoot, runRoot, runId);
|
|
return { ok: true, runId, reportPath: reportDir, overall };
|
|
}
|
|
if (!keep) {
|
|
try {
|
|
const validated = validateRunRoot(repositoryRoot, runRoot, runId);
|
|
rmSync(validated, { recursive: true, force: true });
|
|
} catch { /* best effort */ }
|
|
}
|
|
return { ok: false, runId, reportPath: reportDir, overall };
|
|
} catch (error) {
|
|
try {
|
|
const partial = { schemaVersion: 1, runId, phase: "p4", checks: [], overall: "FAIL", error: String(error).slice(0, 500) };
|
|
await writeFile(reportJsonDir, JSON.stringify(partial, null, 2) + "\n");
|
|
await writeFile(reportDir, `# P4 automated integration acceptance\n\n- run: \`${runId}\`\n- error: \`${String(error).slice(0, 500)}\`\n\n**Overall: FAIL**\n`);
|
|
} catch { /* best effort */ }
|
|
if (keep) return { ok: false, runId, reportPath: reportDir, overall: "FAIL" };
|
|
try {
|
|
const validated = validateRunRoot(repositoryRoot, runRoot, runId);
|
|
rmSync(validated, { recursive: true, force: true });
|
|
} catch { /* best effort */ }
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
if (import.meta.url === `file://${process.argv[1]}`) {
|
|
const args = process.argv.slice(2);
|
|
const keep = args.includes("--keep");
|
|
runAcceptance({ keep }).then((outcome) => {
|
|
process.stdout.write(`P4 automated integration: ${outcome.overall}\nrun: ${outcome.runId}\nreport: ${outcome.reportPath}\n`);
|
|
process.exit(outcome.ok ? 0 : 1);
|
|
}).catch((error) => {
|
|
process.stderr.write(`P4 automated integration: FAIL\n${String(error)}\n`);
|
|
process.exit(1);
|
|
});
|
|
}
|