mirror of
https://github.com/garrytan/gstack.git
synced 2026-11-03 07:18:03 +01:00
A1: verify gbrain landing before stamping transcripts, plus --reconcile
gbrain import exits 0 while skipping pages (returned per-file failures, malformed filenames, managed-import Pending receipts), and its failure ledger exists only for git import dirs, so a skipped page was stamped ingested and never retried (field reports: 37 and 69 lost entries). Write, verify, then stamp (lib/memory-ingest-landing.ts, a new ratcheted module): - State schema 2 with a transition table (staged -> imported_unverified -> ingested, plus refused, quarantined, unrecoverable). Schema 1 files migrate in place: entries become ingested in the legacy "default" source and a reconcile pass is marked pending. - Named failures come from --json `failures`, stderr "Skipped <rel>:" lines and the ledger; a failure gbrain does not attribute refuses the whole batch. After 3 refused batches the batch is bisected and the page that poisons it is quarantined (printed even under --quiet). - Landing check: one `gbrain list --source-id` per source per run and at most 50 `gbrain get --json` content checks against a gstack_content_sha256 frontmatter field gstack now writes. Lookup errors and pglite_busy mean "not yet checked", never absent; pages over the cap stay imported_unverified and are not re-imported. - Each entry records its gbrain source_id. Remote-http mode records "staged", never "ingested". - The source fingerprint comes from the exact bytes parsed, so a transcript that grows during import is not stamped as its newer self (the old unscanned contract test is migrated; it pinned that bug). - --reconcile mode (bounded by --limit, default 200; resumable; --dry-run; backs up the state file) prints "reconcile: checked N, present M, re-queued K, not yet checked L". --request-reconcile records a pending pass without calling gbrain; the v1.91.17.0 migration (placeholder name) uses it, and the next ingest run does one bounded pass. - One lock for every state writer; a state-save failure fails the run. The #1802 static pins in regression-1611 move to the new structure (remote-http returns before importBatch; the finally honors preserveStaging). Co-authored-by: afshaker <andrew@phone2.io>
This commit is contained in:
1 parent
96f473abda
commit
97d4e58dcc
11 files changed
+1939
-612
No files matched your search
+2
-1
@@ -31,7 +31,8 @@
|
||||
{ "file": "test/helpers/pty/runners/floor.ts", "movedFrom": "test/helpers/claude-pty-runner.ts" },
|
||||
{ "file": "test/helpers/pty/runners/observation.ts", "movedFrom": "test/helpers/claude-pty-runner.ts" },
|
||||
{ "file": "test/helpers/pty/screen.ts", "movedFrom": "test/helpers/pty-screen.ts" },
|
||||
{ "file": "test/helpers/pty/session.ts", "movedFrom": "test/helpers/claude-pty-runner.ts" }
|
||||
{ "file": "test/helpers/pty/session.ts", "movedFrom": "test/helpers/claude-pty-runner.ts" },
|
||||
{ "file": "lib/memory-ingest-landing.ts", "movedFrom": "(new: A1 memory ingest landing check, state transitions and reconcile)" }
|
||||
],
|
||||
"residualFiles": [
|
||||
{ "file": "scripts/test-free-shards.ts", "maxLines": 2361, "reason": "free lane policy after the shard engine took spawn, sandbox, logs, seeds and flags" },
|
||||
|
||||
@@ -482,11 +482,18 @@ if (process.env.LIMIT_STAGE_WRITES === '1') {
|
||||
expect(readFileSync(join(home, "imports"), "utf8").trim().split("\n")).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("keeps the no-scan import stamping contract unchanged", () => {
|
||||
// A1 changes the old no-scan contract on purpose: without --scan-secrets
|
||||
// the stamp used to describe the file as it was AFTER the import (its
|
||||
// newer self), so the appended record was never imported.
|
||||
it("does not stamp an append during an unscanned import, and the next run imports it (A1)", () => {
|
||||
const path = source();
|
||||
env.APPEND_DURING_IMPORT = path;
|
||||
env.APPEND_RECORD = appendRecord();
|
||||
expect(run().status).toBe(0);
|
||||
expect(sessions()[path]).toBeUndefined();
|
||||
delete env.APPEND_DURING_IMPORT;
|
||||
expect(run().status).toBe(0);
|
||||
expect(imported()[0].body).toContain("late ordinary update");
|
||||
expect(sessions()[path]).toMatchObject({ sha256: createHash("sha256").update(readFileSync(path)).digest("hex") });
|
||||
});
|
||||
});
|
||||
@@ -958,7 +965,7 @@ describe("gstack-memory-ingest CLI", () => {
|
||||
// ── State file behavior ────────────────────────────────────────────────────
|
||||
|
||||
describe("gstack-memory-ingest state file", () => {
|
||||
it("--incremental on empty home creates state file with schema_version: 1", () => {
|
||||
it("--incremental on empty home creates state file with schema_version: 2", () => {
|
||||
const home = makeTestHome();
|
||||
const gstackHome = join(home, ".gstack");
|
||||
mkdirSync(gstackHome, { recursive: true });
|
||||
@@ -967,7 +974,7 @@ describe("gstack-memory-ingest state file", () => {
|
||||
const statePath = join(gstackHome, ".transcript-ingest-state.json");
|
||||
expect(existsSync(statePath)).toBe(true);
|
||||
const state = JSON.parse(readFileSync(statePath, "utf-8"));
|
||||
expect(state.schema_version).toBe(1);
|
||||
expect(state.schema_version).toBe(2);
|
||||
expect(state.last_writer).toBe("gstack-memory-ingest");
|
||||
rmSync(home, { recursive: true, force: true });
|
||||
});
|
||||
@@ -984,7 +991,7 @@ describe("gstack-memory-ingest state file", () => {
|
||||
expect(existsSync(statePath + ".bak")).toBe(true);
|
||||
|
||||
const fresh = JSON.parse(readFileSync(statePath, "utf-8"));
|
||||
expect(fresh.schema_version).toBe(1);
|
||||
expect(fresh.schema_version).toBe(2);
|
||||
rmSync(home, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
/**
|
||||
* v1.91.17.0 migration (placeholder name; the release queue may rename it):
|
||||
* step 1 records a pending memory-ingest reconcile (A1) without calling gbrain.
|
||||
*/
|
||||
import { describe, test, expect, beforeEach, afterEach } from "bun:test";
|
||||
import * as fs from "fs";
|
||||
import * as os from "os";
|
||||
import * as path from "path";
|
||||
import { spawnSync } from "child_process";
|
||||
|
||||
const ROOT = path.resolve(import.meta.dir, "..");
|
||||
const MIGRATION = path.join(ROOT, "gstack-upgrade", "migrations", "v1.91.17.0.sh");
|
||||
|
||||
let home: string;
|
||||
let gstackHome: string;
|
||||
let bin: string;
|
||||
|
||||
beforeEach(() => {
|
||||
home = fs.mkdtempSync(path.join(os.tmpdir(), "mig-v1.91.17-"));
|
||||
gstackHome = path.join(home, ".gstack");
|
||||
bin = path.join(home, "bin");
|
||||
fs.mkdirSync(bin, { recursive: true });
|
||||
fs.writeFileSync(path.join(bin, "gbrain"), `#!/bin/sh\necho "$@" >> "${home}/gbrain-calls.log"\nexit 1\n`, { mode: 0o755 });
|
||||
});
|
||||
afterEach(() => fs.rmSync(home, { recursive: true, force: true }));
|
||||
|
||||
function run() {
|
||||
const r = spawnSync("bash", [MIGRATION], {
|
||||
env: { PATH: `${bin}:${process.env.PATH}`, HOME: home, GSTACK_HOME: gstackHome },
|
||||
encoding: "utf-8",
|
||||
cwd: home,
|
||||
timeout: 30_000,
|
||||
});
|
||||
return { code: r.status, stdout: r.stdout || "", stderr: r.stderr || "" };
|
||||
}
|
||||
|
||||
describe("v1.91.17.0 migration: memory reconcile pending (A1)", () => {
|
||||
test("marks a reconcile pending on an existing ingest state and never calls gbrain", () => {
|
||||
fs.mkdirSync(gstackHome, { recursive: true });
|
||||
const statePath = path.join(gstackHome, ".transcript-ingest-state.json");
|
||||
fs.writeFileSync(statePath, JSON.stringify({
|
||||
schema_version: 1, last_writer: "gstack-memory-ingest",
|
||||
sessions: { "/t/a.jsonl": { mtime_ns: 1, sha256: "h", ingested_at: "2026-09-01T00:00:00Z", page_slug: "transcripts/x/a" } },
|
||||
}));
|
||||
const first = run();
|
||||
expect(first.code).toBe(0);
|
||||
expect(first.stdout).toContain("reconcile pending");
|
||||
const state = JSON.parse(fs.readFileSync(statePath, "utf-8"));
|
||||
expect(state.schema_version).toBe(2);
|
||||
expect(state.reconcile.pending).toBe(true);
|
||||
expect(state.sessions["/t/a.jsonl"]).toMatchObject({ status: "ingested", source_id: "default" });
|
||||
expect(run().code).toBe(0);
|
||||
expect(fs.existsSync(path.join(home, "gbrain-calls.log"))).toBe(false);
|
||||
});
|
||||
|
||||
test("is a no-op without an ingest state", () => {
|
||||
expect(run().code).toBe(0);
|
||||
expect(fs.existsSync(path.join(gstackHome, ".transcript-ingest-state.json"))).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,117 @@
|
||||
/**
|
||||
* A fake `gbrain` CLI backed by a JSON file, for memory-ingest landing tests
|
||||
* (A1/A4). It speaks the gbrain HEAD (0.60) shapes gstack relies on:
|
||||
* `import <dir> --json [--source-id]`, `list --source-id --limit`,
|
||||
* `get <slug> --source-id --json`, `sources list --json`, `sources add`.
|
||||
*
|
||||
* Behavior switches (env, read by the fake at run time):
|
||||
* FAKE_FAIL=<substr> pages whose path contains it fail by name (HEAD `failures`)
|
||||
* FAKE_PENDING=<substr> accepted as a managed-import Pending receipt: counted, never stored
|
||||
* FAKE_POISON=<substr> any batch containing it fails without naming a page
|
||||
* FAKE_OLD_JSON=1 pre-0.48: no failures/unchanged/source_id keys, and a
|
||||
* returned failure counts only as skipped (exit 0)
|
||||
* FAKE_LIST_UNSCOPED=1 `list` ignores --source-id (pre-0.46.25)
|
||||
* FAKE_BUSY=1 list/get answer {"error":"pglite_busy"} with exit 1
|
||||
* FAKE_NO_SOURCES=1 `sources add` is unsupported
|
||||
* Every call is appended to $HOME/gbrain-calls.log.
|
||||
*/
|
||||
import { chmodSync, mkdirSync, writeFileSync } from "fs";
|
||||
import { join } from "path";
|
||||
|
||||
export function installFakeBrain(home: string): { binDir: string; brainFile: string; callLog: string } {
|
||||
const binDir = join(home, "fake-brain-bin");
|
||||
mkdirSync(binDir, { recursive: true });
|
||||
const brainFile = join(home, "fake-brain.json");
|
||||
const callLog = join(home, "gbrain-calls.log");
|
||||
writeFileSync(join(binDir, "gbrain"), `#!${process.execPath}
|
||||
const fs = require("fs");
|
||||
const path = require("path");
|
||||
const args = process.argv.slice(2);
|
||||
const env = process.env;
|
||||
fs.appendFileSync(${JSON.stringify(callLog)}, JSON.stringify(args) + "\\n");
|
||||
const brainFile = ${JSON.stringify(brainFile)};
|
||||
const brain = fs.existsSync(brainFile) ? JSON.parse(fs.readFileSync(brainFile, "utf8")) : { sources: { default: { federated: true } }, pages: {} };
|
||||
const save = () => fs.writeFileSync(brainFile, JSON.stringify(brain, null, 2));
|
||||
const flag = (name) => { const i = args.indexOf(name); return i === -1 ? undefined : args[i + 1]; };
|
||||
const busy = () => { if (env.FAKE_BUSY === "1") { console.log(JSON.stringify({ error: "pglite_busy" })); process.exit(1); } };
|
||||
const cmd = args[0];
|
||||
if (cmd === "--help") { console.log("Commands:\\n import <dir> Import"); process.exit(0); }
|
||||
if (cmd === "import") {
|
||||
const dir = args[1];
|
||||
const source = flag("--source-id") ?? "default";
|
||||
if (!brain.sources[source]) { console.error("Source not found: " + source); process.exit(1); }
|
||||
const files = [];
|
||||
(function walk(d) {
|
||||
for (const e of fs.readdirSync(d, { withFileTypes: true })) {
|
||||
if (e.name.startsWith(".")) continue;
|
||||
const p = path.join(d, e.name);
|
||||
if (e.isDirectory()) walk(p); else if (e.name.endsWith(".md")) files.push(path.relative(dir, p));
|
||||
}
|
||||
})(dir);
|
||||
if (env.FAKE_POISON && files.some((f) => f.includes(env.FAKE_POISON))) {
|
||||
console.error(" (suppressing further errors)");
|
||||
console.log(JSON.stringify({ status: "partial", imported: 0, skipped: files.length, errors: 1, unchanged: files.length - 1, failures: [], source_id: source, total_files: files.length }));
|
||||
process.exit(1);
|
||||
}
|
||||
let imported = 0, skipped = 0, errors = 0;
|
||||
const failures = [];
|
||||
for (const rel of files) {
|
||||
if (env.FAKE_FAIL && rel.includes(env.FAKE_FAIL)) {
|
||||
// Before 0.48.5.0 a returned per-file failure counted only as skipped (exit 0).
|
||||
skipped++; if (env.FAKE_OLD_JSON !== "1") errors++;
|
||||
failures.push({ path: rel, error: "Invalid YAML frontmatter" });
|
||||
console.error(" Skipped " + rel + ": Invalid YAML frontmatter");
|
||||
continue;
|
||||
}
|
||||
if (env.FAKE_PENDING && rel.includes(env.FAKE_PENDING)) {
|
||||
skipped++;
|
||||
console.error(" Pending: " + rel + " was accepted and is still publishing; rerun to confirm it.");
|
||||
continue;
|
||||
}
|
||||
const body = fs.readFileSync(path.join(dir, rel), "utf8");
|
||||
const sha = (body.match(/^gstack_content_sha256: (\\S+)$/m) || [])[1] ?? null;
|
||||
const slug = rel.replace(/\\.md$/, "").toLowerCase();
|
||||
brain.pages[source] ??= {};
|
||||
if (brain.pages[source][slug]?.sha === sha) skipped++; else imported++;
|
||||
brain.pages[source][slug] = { sha };
|
||||
}
|
||||
save();
|
||||
const report = env.FAKE_OLD_JSON === "1"
|
||||
? { status: "success", imported, skipped, errors, total_files: files.length }
|
||||
: { status: errors ? "partial" : "success", imported, skipped, errors, unchanged: skipped - failures.length, malformed_skipped: 0, failures, source_id: source, total_files: files.length };
|
||||
console.log(JSON.stringify(report));
|
||||
process.exit(errors ? 1 : 0);
|
||||
}
|
||||
if (cmd === "list") {
|
||||
busy();
|
||||
const source = env.FAKE_LIST_UNSCOPED === "1" ? undefined : flag("--source-id");
|
||||
if (source !== undefined && !brain.sources[source]) { console.error("Source not found: " + source); process.exit(1); }
|
||||
const slugs = Object.entries(brain.pages).filter(([s]) => source === undefined || s === source).flatMap(([, p]) => Object.keys(p));
|
||||
console.log(slugs.length ? slugs.map((s) => s + "\\ttranscript\\t2026-10-03\\t" + s).join("\\n") : "No pages found.");
|
||||
process.exit(0);
|
||||
}
|
||||
if (cmd === "get") {
|
||||
busy();
|
||||
const source = flag("--source-id") ?? "default";
|
||||
const page = brain.pages[source]?.[args[1]];
|
||||
if (!page) { console.error("Page not found: " + args[1]); process.exit(1); }
|
||||
console.log(JSON.stringify({ slug: args[1], source_id: source, frontmatter: page.sha ? { gstack_content_sha256: page.sha } : {} }));
|
||||
process.exit(0);
|
||||
}
|
||||
if (cmd === "sources" && args[1] === "list") {
|
||||
console.log(JSON.stringify({ sources: Object.entries(brain.sources).map(([id, s]) => ({ id, federated: s.federated, page_count: Object.keys(brain.pages[id] ?? {}).length })) }));
|
||||
process.exit(0);
|
||||
}
|
||||
if (cmd === "sources" && args[1] === "add") {
|
||||
if (env.FAKE_NO_SOURCES === "1") { console.error("Unknown command: sources add"); process.exit(2); }
|
||||
brain.sources[args[2]] = { federated: args.includes("--federated") };
|
||||
save();
|
||||
console.log("Source added: " + args[2]);
|
||||
process.exit(0);
|
||||
}
|
||||
console.error("Unknown command: " + cmd);
|
||||
process.exit(2);
|
||||
`);
|
||||
chmodSync(join(binDir, "gbrain"), 0o755);
|
||||
return { binDir, brainFile, callLog };
|
||||
}
|
||||
@@ -0,0 +1,355 @@
|
||||
/**
|
||||
* A1 (#2778): memory ingest stamps a page only after confirming it landed.
|
||||
*
|
||||
* Unit tests drive lib/memory-ingest-landing.ts directly; CLI tests run
|
||||
* bin/gstack-memory-ingest.ts against a fake gbrain that keeps a JSON brain
|
||||
* and speaks gbrain HEAD's import/list/get shapes (test/helpers/fake-gbrain-brain.ts).
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from "bun:test";
|
||||
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync, readdirSync } from "fs";
|
||||
import { tmpdir } from "os";
|
||||
import { join } from "path";
|
||||
import { spawn, spawnSync } from "child_process";
|
||||
import {
|
||||
acquireStateLock,
|
||||
checkLanding,
|
||||
classifyImport,
|
||||
formatReconcileSummary,
|
||||
gbrainLookups,
|
||||
migrateState,
|
||||
reconcile,
|
||||
saveStateFile,
|
||||
setEntry,
|
||||
emptyState,
|
||||
type IngestState,
|
||||
type LandingLookups,
|
||||
type StateEntry,
|
||||
} from "../lib/memory-ingest-landing";
|
||||
import { installFakeBrain } from "./helpers/fake-gbrain-brain";
|
||||
|
||||
const SCRIPT = join(import.meta.dir, "..", "bin", "gstack-memory-ingest.ts");
|
||||
|
||||
const entry = (over: Partial<StateEntry> = {}): StateEntry => ({
|
||||
mtime_ns: 1, sha256: "s", page_slug: "ceo-plans/demo/2026-10-03-a", status: "ingested", source_id: "default", ...over,
|
||||
});
|
||||
|
||||
function lookupsFrom(pages: Record<string, Record<string, string | null>> | null, opts: { busyGet?: boolean } = {}): LandingLookups & { gets: number } {
|
||||
const l = {
|
||||
gets: 0,
|
||||
list: (source: string) => (pages ? new Set(Object.keys(pages[source] ?? {})) : null),
|
||||
contentSha: (source: string, slug: string) => {
|
||||
l.gets++;
|
||||
if (opts.busyGet) return undefined;
|
||||
const v = pages?.[source]?.[slug];
|
||||
return v === undefined ? undefined : v;
|
||||
},
|
||||
};
|
||||
return l;
|
||||
}
|
||||
|
||||
describe("landing state (unit)", () => {
|
||||
it("migrates a schema-1 file: entries become ingested in the legacy default source, reconcile pending", () => {
|
||||
const m = migrateState({ schema_version: 1, last_writer: "x", sessions: { "/a.jsonl": { mtime_ns: 5, sha256: "h", ingested_at: "t", page_slug: "p" } } });
|
||||
expect(m?.migrated).toBe(true);
|
||||
expect(m?.state.schema_version).toBe(2);
|
||||
expect(m?.state.sessions["/a.jsonl"]).toMatchObject({ status: "ingested", source_id: "default", page_slug: "p", ingested_at: "t" });
|
||||
expect(m?.state.reconcile?.pending).toBe(true);
|
||||
expect(migrateState({ schema_version: 999, sessions: {} })).toBeNull();
|
||||
});
|
||||
|
||||
it("rejects transitions outside the table", () => {
|
||||
const state = emptyState();
|
||||
setEntry(state, "/a", entry({ status: "ingested" }));
|
||||
setEntry(state, "/a", entry({ status: "unrecoverable" }));
|
||||
expect(() => setEntry(state, "/a", entry({ status: "quarantined" }))).toThrow(/illegal transition unrecoverable -> quarantined/);
|
||||
});
|
||||
|
||||
it("names HEAD --json failures and treats Pending receipts as counted, not failed", () => {
|
||||
const staged = new Set(["a.md", "b.md", "c.md"]);
|
||||
const v = classifyImport(
|
||||
{ imported: 1, skipped: 2, errors: 1, unchanged: 1, failures: [{ path: "b.md", error: "Invalid YAML" }], source_id: "default" },
|
||||
" Skipped b.md: Invalid YAML\n Pending: c.md was accepted and is still publishing; rerun to confirm it.\n",
|
||||
[],
|
||||
staged,
|
||||
);
|
||||
expect([...v.named]).toEqual([["b.md", "Invalid YAML"]]);
|
||||
expect(v.refuseAll).toBeUndefined();
|
||||
});
|
||||
|
||||
it("reads pre-0.48 stderr failures and refuses the batch when a thrown failure is unnamed", () => {
|
||||
const staged = new Set(["a.md", "b.md"]);
|
||||
const named = classifyImport({ imported: 1, skipped: 1, errors: 0 }, " Skipped b.md: too large\n", [], staged);
|
||||
expect([...named.named.keys()]).toEqual(["b.md"]);
|
||||
expect(named.refuseAll).toBeUndefined();
|
||||
const hidden = classifyImport({ imported: 1, skipped: 1, errors: 1 }, "", [], staged);
|
||||
expect(hidden.refuseAll).toMatch(/1 failure\(s\) it did not attribute/);
|
||||
const short = classifyImport({ imported: 0, skipped: 0, errors: 0, total_files: 0 }, "", [], staged);
|
||||
expect(short.refuseAll).toMatch(/accounted for 0 of 2 staged page\(s\)/);
|
||||
});
|
||||
|
||||
it("checks presence per recorded source and content within the get budget", () => {
|
||||
const l = lookupsFrom({ default: { "ceo-plans/demo/2026-10-03-a": "abc" } });
|
||||
const budget = { gets: 1 };
|
||||
expect(checkLanding(entry({ content_sha256: "abc", status: "imported_unverified" }), l, budget)).toBe("ingested");
|
||||
expect(checkLanding(entry({ content_sha256: "abc", status: "imported_unverified" }), l, budget)).toBe("unchecked");
|
||||
expect(checkLanding(entry({ content_sha256: "zzz" }), l, { gets: 1 })).toBe("mismatch");
|
||||
expect(checkLanding(entry({ source_id: "gstack-tx-x" }), l, { gets: 1 })).toBe("absent");
|
||||
expect(checkLanding(entry(), lookupsFrom(null), { gets: 1 })).toBe("unchecked");
|
||||
expect(checkLanding(entry({ content_sha256: "abc" }), lookupsFrom({ default: { "ceo-plans/demo/2026-10-03-a": "abc" } }, { busyGet: true }), { gets: 1 })).toBe("unchecked");
|
||||
});
|
||||
|
||||
it("treats pglite_busy and a list that ignores --source-id as not yet checked", () => {
|
||||
const busy = gbrainLookups(() => ({ status: 1, stdout: '{"error":"pglite_busy"}', stderr: "" }));
|
||||
expect(busy.list("default")).toBeNull();
|
||||
const unscoped = gbrainLookups(() => ({ status: 0, stdout: "a\ttranscript\t2026-10-03\ta\n", stderr: "" }));
|
||||
expect(unscoped.list("default")).toBeNull();
|
||||
const calls: string[][] = [];
|
||||
const scoped = gbrainLookups((a) => {
|
||||
calls.push(a);
|
||||
return a.includes("gstack-capability-probe-0") ? { status: 1, stdout: "", stderr: "Source not found" } : { status: 0, stdout: "x/y\ttranscript\t2026-10-03\tt\n", stderr: "" };
|
||||
});
|
||||
expect([...scoped.list("default")!]).toEqual(["x/y"]);
|
||||
scoped.list("default");
|
||||
expect(calls).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("reconcile (unit)", () => {
|
||||
function state(): IngestState {
|
||||
const s = emptyState();
|
||||
for (const name of ["a", "b", "c", "d"]) s.sessions[`/t/${name}.jsonl`] = entry({ page_slug: `transcripts/x/${name}` });
|
||||
return s;
|
||||
}
|
||||
const brain = { default: { "transcripts/x/a": null, "transcripts/x/b": null } };
|
||||
|
||||
it("finds legacy default-source entries present and re-queues only missing pages whose file still exists", () => {
|
||||
const s = state();
|
||||
const sum = reconcile(s, lookupsFrom(brain), { limit: 10, dryRun: false, fileExists: (p) => p !== "/t/d.jsonl" });
|
||||
expect(sum).toMatchObject({ checked: 4, present: 2, requeued: 1, unrecoverable: 1, notChecked: 0, complete: true });
|
||||
expect(s.sessions["/t/c.jsonl"].status).toBe("refused");
|
||||
expect(s.sessions["/t/d.jsonl"].status).toBe("unrecoverable");
|
||||
expect(s.reconcile?.pending).toBe(false);
|
||||
expect(formatReconcileSummary(sum, false)).toBe("reconcile: checked 4, present 2, re-queued 1, not yet checked 0, unrecoverable 1 (source file gone)");
|
||||
});
|
||||
|
||||
it("--dry-run changes nothing; --limit bounds the pass and the next pass resumes; a third pass is a no-op", () => {
|
||||
const s = state();
|
||||
const before = JSON.stringify(s);
|
||||
reconcile(s, lookupsFrom(brain), { limit: 10, dryRun: true, fileExists: () => true });
|
||||
expect(JSON.stringify(s)).toBe(before);
|
||||
const first = reconcile(s, lookupsFrom(brain), { limit: 2, dryRun: false, fileExists: () => true });
|
||||
expect(first).toMatchObject({ checked: 2, complete: false });
|
||||
expect(formatReconcileSummary(first, false)).toMatch(/\(run again to continue\)$/);
|
||||
const second = reconcile(s, lookupsFrom(brain), { limit: 2, dryRun: false, fileExists: () => true });
|
||||
expect(second).toMatchObject({ checked: 2, present: 0, requeued: 2, complete: true });
|
||||
const third = reconcile(s, lookupsFrom(brain), { limit: 2, dryRun: false, fileExists: () => true });
|
||||
expect(third).toMatchObject({ checked: 2, present: 2, requeued: 0 });
|
||||
});
|
||||
|
||||
it("a failed lookup is not yet checked, never absent, and keeps the reconcile pending", () => {
|
||||
const s = state();
|
||||
const sum = reconcile(s, lookupsFrom(null), { limit: 10, dryRun: false, fileExists: () => true });
|
||||
expect(sum).toMatchObject({ checked: 0, notChecked: 4, requeued: 0, complete: false });
|
||||
expect(Object.values(s.sessions).every((e) => e.status === "ingested")).toBe(true);
|
||||
expect(s.reconcile?.pending).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("state writers (unit)", () => {
|
||||
let dir = "";
|
||||
beforeEach(() => { dir = mkdtempSync(join(tmpdir(), "gstack-landing-lock-")); });
|
||||
afterEach(() => rmSync(dir, { recursive: true, force: true }));
|
||||
|
||||
it("one lock: a live holder blocks, a dead holder's lock is taken over", async () => {
|
||||
const lockPath = join(dir, "state.json.lock");
|
||||
const sleeper = spawn("sleep", ["30"]);
|
||||
try {
|
||||
writeFileSync(lockPath, `${sleeper.pid}\n`);
|
||||
expect(acquireStateLock(lockPath)).toEqual({ ok: false, holder: sleeper.pid! });
|
||||
} finally {
|
||||
sleeper.kill("SIGKILL");
|
||||
}
|
||||
await new Promise((r) => sleeper.on("exit", r));
|
||||
const lock = acquireStateLock(lockPath);
|
||||
expect(lock.ok).toBe(true);
|
||||
if (lock.ok) lock.release();
|
||||
expect(existsSync(lockPath)).toBe(false);
|
||||
});
|
||||
|
||||
it("a state save failure throws instead of being swallowed", () => {
|
||||
const target = join(dir, "state.json");
|
||||
mkdirSync(target);
|
||||
expect(() => saveStateFile(target, emptyState())).toThrow(/could not save ingest state/);
|
||||
});
|
||||
});
|
||||
|
||||
// ── CLI against the fake brain ─────────────────────────────────────────────
|
||||
|
||||
describe("gstack-memory-ingest landing check (CLI, fake gbrain)", () => {
|
||||
let home = "";
|
||||
let gstackHome = "";
|
||||
let env: Record<string, string> = {};
|
||||
let brain: ReturnType<typeof installFakeBrain>;
|
||||
|
||||
beforeEach(() => {
|
||||
home = mkdtempSync(join(tmpdir(), "gstack-landing-"));
|
||||
gstackHome = join(home, ".gstack");
|
||||
mkdirSync(join(gstackHome, "projects", "demo", "ceo-plans"), { recursive: true });
|
||||
brain = installFakeBrain(home);
|
||||
env = { HOME: home, GSTACK_HOME: gstackHome, PATH: `${brain.binDir}:/usr/bin:/bin`, TMPDIR: home };
|
||||
});
|
||||
afterEach(() => rmSync(home, { recursive: true, force: true }));
|
||||
|
||||
function plans(...names: string[]): string[] {
|
||||
return names.map((n) => {
|
||||
const p = join(gstackHome, "projects", "demo", "ceo-plans", `${n}.md`);
|
||||
writeFileSync(p, `# Plan ${n}\n\nbody of ${n}\n`);
|
||||
return p;
|
||||
});
|
||||
}
|
||||
function run(args: string[] = [], extra: Record<string, string> = {}) {
|
||||
const r = spawnSync(process.execPath, [SCRIPT, "--sources", "ceo-plan", ...args], {
|
||||
env: { ...env, ...extra }, cwd: home, encoding: "utf-8", timeout: 60_000,
|
||||
});
|
||||
return { status: r.status, stdout: r.stdout || "", stderr: r.stderr || "" };
|
||||
}
|
||||
const statePath = () => join(gstackHome, ".transcript-ingest-state.json");
|
||||
const sessions = (): Record<string, StateEntry> => existsSync(statePath()) ? JSON.parse(readFileSync(statePath(), "utf8")).sessions : {};
|
||||
const calls = (): string[][] => existsSync(brain.callLog) ? readFileSync(brain.callLog, "utf8").trim().split("\n").filter(Boolean).map((l) => JSON.parse(l)) : [];
|
||||
|
||||
it("stamps verified pages ingested with their recorded source and content hash", () => {
|
||||
const [a] = plans("alpha");
|
||||
const r = run(["--bulk", "--quiet"]);
|
||||
expect(r.status).toBe(0);
|
||||
expect(sessions()[a]).toMatchObject({ status: "ingested", source_id: "default" });
|
||||
expect(sessions()[a].content_sha256).toMatch(/^[0-9a-f]{64}$/);
|
||||
expect(calls().some((c) => c[0] === "get" && c.includes("--json"))).toBe(true);
|
||||
});
|
||||
|
||||
it("a page gbrain names as failed (exit 1, HEAD failures[]) is left un-stamped and printed under --quiet", () => {
|
||||
const [a, b] = plans("alpha", "broken");
|
||||
const r = run(["--bulk", "--quiet"], { FAKE_FAIL: "broken" });
|
||||
expect(r.stderr).toMatch(/\[memory-ingest\] FAILED ceo-plans\/demo\/\S+broken\.md: Invalid YAML frontmatter/);
|
||||
expect(sessions()[a]?.status).toBe("ingested");
|
||||
expect(sessions()[b]).toBeUndefined();
|
||||
expect(r.status).toBe(0);
|
||||
});
|
||||
|
||||
it("pre-0.48 gbrain: a page skipped with exit 0 and only a stderr 'Skipped' line is never stamped (#2778 field case)", () => {
|
||||
const [a, b] = plans("alpha", "broken");
|
||||
const r = run(["--bulk", "--quiet"], { FAKE_FAIL: "broken", FAKE_OLD_JSON: "1" });
|
||||
expect(r.status).toBe(0);
|
||||
expect(r.stderr).toMatch(/FAILED ceo-plans\/demo\/\S+broken\.md/);
|
||||
expect(sessions()[b]).toBeUndefined();
|
||||
expect(sessions()[a].source_id).toBe("default");
|
||||
});
|
||||
|
||||
it("a Pending receipt (counted unchanged, never stored) is re-queued after the landing check, not stamped ingested", () => {
|
||||
const [a, p] = plans("alpha", "pending");
|
||||
const r = run(["--bulk", "--quiet"], { FAKE_PENDING: "pending" });
|
||||
expect(r.stderr).toMatch(/re-queued ceo-plans\/demo\/\S+pending: not found in gbrain source default after import/);
|
||||
expect(sessions()[a].status).toBe("ingested");
|
||||
expect(sessions()[p].status).toBe("refused");
|
||||
const again = run(["--incremental", "--quiet"]);
|
||||
expect(again.status).toBe(0);
|
||||
expect(sessions()[p].status).toBe("ingested");
|
||||
});
|
||||
|
||||
it("without a usable lookup (list ignores --source-id, or PGLite busy) pages stay imported_unverified and are not re-imported", () => {
|
||||
const [a] = plans("alpha");
|
||||
expect(run(["--bulk", "--quiet"], { FAKE_LIST_UNSCOPED: "1" }).status).toBe(0);
|
||||
expect(sessions()[a].status).toBe("imported_unverified");
|
||||
const imports = () => calls().filter((c) => c[0] === "import").length;
|
||||
const before = imports();
|
||||
expect(run(["--incremental", "--quiet"], { FAKE_BUSY: "1" }).status).toBe(0);
|
||||
expect(imports()).toBe(before);
|
||||
expect(sessions()[a].status).toBe("imported_unverified");
|
||||
expect(run(["--incremental", "--quiet"]).status).toBe(0);
|
||||
expect(sessions()[a].status).toBe("ingested");
|
||||
});
|
||||
|
||||
it("an unnamed failure refuses the batch; after 3 refusals the batch is bisected and the poison page quarantined", () => {
|
||||
const pages = plans("p1", "p2", "poison", "p4");
|
||||
for (let i = 1; i <= 2; i++) {
|
||||
const r = run(["--bulk", "--quiet"], { FAKE_POISON: "poison" });
|
||||
expect(r.status).toBe(1);
|
||||
expect(r.stderr).toMatch(/did not attribute to a staged page/);
|
||||
expect(Object.keys(sessions())).toHaveLength(0);
|
||||
}
|
||||
const r = run(["--bulk", "--quiet"], { FAKE_POISON: "poison" });
|
||||
expect(r.status).toBe(0);
|
||||
expect(r.stderr).toMatch(/quarantined ceo-plans\/demo\/\S+poison .*gbrain refuses every batch that contains it/);
|
||||
const s = sessions();
|
||||
expect(s[pages[2]].status).toBe("quarantined");
|
||||
for (const p of [pages[0], pages[1], pages[3]]) expect(s[p].status).toBe("ingested");
|
||||
const next = run(["--bulk", "--quiet"], { FAKE_POISON: "poison" });
|
||||
expect(next.status).toBe(0);
|
||||
expect(sessions()[pages[2]].status).toBe("quarantined");
|
||||
});
|
||||
|
||||
it("remote-http mode records pages as staged, never ingested", () => {
|
||||
const [a] = plans("alpha");
|
||||
writeFileSync(join(home, ".claude.json"), JSON.stringify({ mcpServers: { gbrain: { type: "http", url: "http://fixture.invalid/mcp" } } }));
|
||||
expect(run(["--bulk", "--quiet"]).status).toBe(0);
|
||||
expect(sessions()[a]).toMatchObject({ status: "staged" });
|
||||
expect(calls().filter((c) => c[0] === "import")).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("--reconcile re-queues a stamped page missing from its recorded source, backs up state, and supports --dry-run and --limit", () => {
|
||||
const [a, b] = plans("alpha", "beta");
|
||||
writeFileSync(statePath(), JSON.stringify({
|
||||
schema_version: 1, last_writer: "gstack-memory-ingest",
|
||||
sessions: Object.fromEntries([a, b].map((p) => [p, { mtime_ns: 1, sha256: "old", ingested_at: "2026-09-01T00:00:00Z", page_slug: `ceo-plans/demo/${p.endsWith("alpha.md") ? "x-alpha" : "x-beta"}` }])),
|
||||
}));
|
||||
writeFileSync(brain.brainFile, JSON.stringify({ sources: { default: { federated: true } }, pages: { default: { "ceo-plans/demo/x-alpha": { sha: null } } } }));
|
||||
const dry = run(["--reconcile", "--dry-run"]);
|
||||
expect(dry.stderr).toContain("reconcile (dry run): checked 2, present 1, re-queued 1, not yet checked 0");
|
||||
expect(JSON.parse(readFileSync(statePath(), "utf8")).schema_version).toBe(1);
|
||||
const limited = run(["--reconcile", "--limit", "1"]);
|
||||
expect(limited.stderr).toContain("reconcile: checked 1, present 1, re-queued 0, not yet checked 0 (run again to continue)");
|
||||
expect(existsSync(`${statePath()}.pre-reconcile.bak`)).toBe(true);
|
||||
const rest = run(["--reconcile"]);
|
||||
expect(rest.stderr).toContain("reconcile: checked 1, present 0, re-queued 1, not yet checked 0");
|
||||
expect(sessions()[b].status).toBe("refused");
|
||||
expect(sessions()[a]).toMatchObject({ status: "ingested", source_id: "default" });
|
||||
});
|
||||
|
||||
it("--request-reconcile records a pending pass without calling gbrain; the next ingest runs one bounded pass and prints the summary under --quiet", () => {
|
||||
const [a] = plans("alpha");
|
||||
expect(run(["--bulk", "--quiet"]).status).toBe(0);
|
||||
rmSync(brain.callLog, { force: true });
|
||||
const req = run(["--request-reconcile", "--quiet"]);
|
||||
expect(req.status).toBe(0);
|
||||
expect(calls()).toEqual([]);
|
||||
expect(JSON.parse(readFileSync(statePath(), "utf8")).reconcile.pending).toBe(true);
|
||||
const next = run(["--incremental", "--quiet"]);
|
||||
expect(next.stderr).toContain("[memory-ingest] reconcile: checked 1, present 1, re-queued 0, not yet checked 0");
|
||||
expect(JSON.parse(readFileSync(statePath(), "utf8")).reconcile.pending).toBe(false);
|
||||
expect(sessions()[a].status).toBe("ingested");
|
||||
});
|
||||
|
||||
it("a state file that cannot be saved fails the run", () => {
|
||||
plans("alpha");
|
||||
mkdirSync(statePath());
|
||||
const r = run(["--bulk", "--quiet"]);
|
||||
expect(r.status).toBe(1);
|
||||
expect(r.stderr).toMatch(/could not save ingest state/);
|
||||
});
|
||||
|
||||
it("a live writer's lock stops a second run before it touches state", () => {
|
||||
plans("alpha");
|
||||
writeFileSync(`${statePath()}.lock`, `${process.pid}\n`);
|
||||
const r = run(["--bulk", "--quiet"]);
|
||||
expect(r.status).toBe(1);
|
||||
expect(r.stderr).toContain(`another memory ingest (pid ${process.pid})`);
|
||||
expect(existsSync(statePath())).toBe(false);
|
||||
expect(calls().filter((c) => c[0] === "import")).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("--help lists --reconcile as a mode", () => {
|
||||
const r = spawnSync(process.execPath, [SCRIPT, "--help"], { env, encoding: "utf-8", timeout: 20_000 });
|
||||
expect(r.stderr).toMatch(/\[--probe\|--incremental\|--bulk\|--reconcile\]/);
|
||||
expect(r.stderr).toMatch(/--dry-run/);
|
||||
expect(readdirSync(gstackHome).filter((f) => f.startsWith(".staging-ingest-"))).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -164,7 +164,7 @@ describe("V1 memory ingest pipeline E2E", () => {
|
||||
rmSync(home, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("--incremental writes a state file with schema_version: 1 + last_writer", () => {
|
||||
it("--incremental writes a state file with schema_version: 2 + last_writer", () => {
|
||||
const home = makeFixtureHome();
|
||||
const { gstackHome } = setupFixture(home);
|
||||
const env = fakeGbrainEnv(home, gstackHome);
|
||||
@@ -174,7 +174,7 @@ describe("V1 memory ingest pipeline E2E", () => {
|
||||
const statePath = join(gstackHome, ".transcript-ingest-state.json");
|
||||
expect(existsSync(statePath)).toBe(true);
|
||||
const state = JSON.parse(readFileSync(statePath, "utf-8"));
|
||||
expect(state.schema_version).toBe(1);
|
||||
expect(state.schema_version).toBe(2);
|
||||
expect(state.last_writer).toBe("gstack-memory-ingest");
|
||||
expect(typeof state.last_full_walk).toBe("string");
|
||||
|
||||
|
||||
@@ -373,10 +373,15 @@ describe("#1802 D1 — remote-http finally gate (static invariant)", () => {
|
||||
"utf-8",
|
||||
);
|
||||
|
||||
test("finally gates cleanupStagingDir on !remoteHttpMode", () => {
|
||||
// Tolerates additional guards (e.g. C3's !preserveStaging) in the same
|
||||
// condition — the load-bearing invariant is that remote-http never deletes.
|
||||
expect(ingest).toMatch(/if \(!remoteHttpMode[^)]*\) cleanupStagingDir\(stagingDir\)/);
|
||||
test("the remote-http branch returns before any staging cleanup can run", () => {
|
||||
// A1 moved remote-http staging into stageForRemoteBrain, which ingestPass
|
||||
// returns before importBatch (the only owner of the cleanup finally) runs.
|
||||
// The load-bearing invariant is unchanged: remote-http never deletes.
|
||||
const at = ingest.indexOf("function stageForRemoteBrain(");
|
||||
expect(at).toBeGreaterThan(-1);
|
||||
const body = ingest.slice(at, ingest.indexOf("\n}\n", at));
|
||||
expect(body).not.toMatch(/cleanupStagingDir/);
|
||||
expect(ingest).toMatch(/if \(remoteHttpMode\) return stageForRemoteBrain\(/);
|
||||
});
|
||||
|
||||
test("the only finally-scoped cleanup call is the gated one", () => {
|
||||
@@ -411,9 +416,7 @@ describe("#1802 C3 — import-timeout preserve (static invariant)", () => {
|
||||
});
|
||||
|
||||
test("finally honors preserveStaging", () => {
|
||||
expect(ingest).toMatch(
|
||||
/if \(!remoteHttpMode && !preserveStaging\) cleanupStagingDir\(stagingDir\)/,
|
||||
);
|
||||
expect(ingest).toMatch(/if \(!preserveStaging\) cleanupStagingDir\(stagingDir\)/);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
Reference in new issue
Block a user