From 97d4e58dccd9a459b379f5435d7aa5cf93037b72 Mon Sep 17 00:00:00 2001 From: garrytan Date: Sat, 3 Oct 2026 21:08:29 +0000 Subject: [PATCH] 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 :" 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 --- bin/gstack-memory-ingest.ts | 1321 +++++++++-------- docs/gbrain-sync-errors.md | 119 ++ gstack-upgrade/migrations/v1.91.17.0.sh | 28 + lib/memory-ingest-landing.ts | 512 +++++++ test/fixtures/module-size-ratchet.json | 3 +- test/gstack-memory-ingest.test.ts | 15 +- ...stack-upgrade-migration-v1_91_17_0.test.ts | 60 + test/helpers/fake-gbrain-brain.ts | 117 ++ test/memory-ingest-landing.test.ts | 355 +++++ test/memory-pipeline.test.ts | 4 +- ...regression-1611-gbrain-sync-resume.test.ts | 17 +- 11 files changed, 1939 insertions(+), 612 deletions(-) create mode 100755 gstack-upgrade/migrations/v1.91.17.0.sh create mode 100644 lib/memory-ingest-landing.ts create mode 100644 test/gstack-upgrade-migration-v1_91_17_0.test.ts create mode 100644 test/helpers/fake-gbrain-brain.ts create mode 100644 test/memory-ingest-landing.test.ts diff --git a/bin/gstack-memory-ingest.ts b/bin/gstack-memory-ingest.ts index 0e87de3b7..d8e731072 100644 --- a/bin/gstack-memory-ingest.ts +++ b/bin/gstack-memory-ingest.ts @@ -68,7 +68,30 @@ import { detectEngineTier, withErrorContext, } from "../lib/gstack-memory-helpers"; -import { execGbrainText, spawnGbrainAsync } from "../lib/gbrain-exec"; +import { execGbrainText, spawnGbrain, spawnGbrainAsync } from "../lib/gbrain-exec"; +import { + BISECT_AFTER_REFUSALS, + DEFAULT_GET_CAP, + DEFAULT_RECONCILE_LIMIT, + LEGACY_SOURCE_ID, + acquireStateLock, + backupStateFile, + checkLanding, + classifyImport, + formatReconcileSummary, + gbrainLookups, + isRetryable, + loadStateFile, + parseImportReport, + reconcile, + saveStateFile, + setEntry, + type BatchVerdict, + type ImportReport, + type IngestState, + type LandingLookups, + type StateEntry, +} from "../lib/memory-ingest-landing"; import { writeReceipt } from "../lib/egress-receipt"; import { checkOwnedStagingDir, STAGING_MARKER } from "../lib/staging-guard"; import { hasRepoPolicyStore, repoPolicyTierBatch } from "../lib/gbrain-repo-policy-client"; @@ -76,7 +99,7 @@ import { resolveStateRoot, readConfigKeyWithRoot, type StateRootEnv } from "../l // ── Types ────────────────────────────────────────────────────────────────── -type Mode = "probe" | "incremental" | "bulk"; +type Mode = "probe" | "incremental" | "bulk" | "reconcile"; interface CliArgs { mode: Mode; @@ -87,6 +110,10 @@ interface CliArgs { sources: Set; limit: number | null; noWrite: boolean; + /** --reconcile --dry-run: report without changing state. */ + dryRun: boolean; + /** --request-reconcile: record that a reconcile pass is pending (no gbrain calls). */ + requestReconcile: boolean; /** * Opt-in gitleaks scan of each rendered page during the prepare phase; * pages with findings, or that could not be scanned, are skipped. Off by @@ -124,22 +151,6 @@ interface PageRecord { content_sha256: string; } -interface IngestState { - schema_version: 1; - last_writer: string; - last_full_walk?: string; - sessions: Record< - string, - { - mtime_ns: number; - sha256: string; - ingested_at: string; - page_slug: string; - partial?: boolean; - } - >; -} - interface ProbeReport { total_files: number; total_bytes: number; @@ -174,6 +185,16 @@ interface BulkResult { failed: number; duration_ms: number; partial_pages: number; + /** A1: pages gbrain refused or that were not found after import (retried next run). */ + refused?: number; + /** A1: pages imported but not yet confirmed in the brain (checked on later runs). */ + unverified?: number; + /** A1: pages isolated as poisoning their batch. */ + quarantined?: number; + /** A4: transcript pages kept local (source not registered, or a remote brain for unattributed pages). */ + kept_local?: number; + /** The pass returned before changing anything; the state file is not rewritten. */ + state_untouched?: boolean; /** * D6: when set, indicates a process-level failure (gbrain CLI missing * or `gbrain import` crashed). Per-file errors (FILE_TOO_LARGE etc.) @@ -189,6 +210,9 @@ interface BulkResult { const HOME = homedir(); const GSTACK_HOME = resolveStateRoot(); const STATE_PATH = join(GSTACK_HOME, ".transcript-ingest-state.json"); +const LOCK_PATH = `${STATE_PATH}.lock`; +/** Machine-local, never-federated source for --include-unattributed transcripts (A4). */ +const UNATTRIBUTED_SOURCE_ID = "gstack-transcripts-unattributed"; const DEFAULT_INCREMENTAL_BUDGET_MS = 50; const ALL_TYPES: MemoryType[] = [ @@ -244,23 +268,35 @@ export function normalizeTranscriptConsent(env?: StateRootEnv): TranscriptConsen // ── CLI ──────────────────────────────────────────────────────────────────── function printUsage(): void { - console.error(`Usage: gstack-memory-ingest [--probe|--incremental|--bulk] [options] + console.error(`Usage: gstack-memory-ingest [--probe|--incremental|--bulk|--reconcile] [options] Modes: --probe Count what would ingest; no writes. Fastest. --incremental Default. mtime fast-path; only walks changed files. --bulk First-run; full walk; gates on permission elsewhere. + --reconcile Re-check pages already marked ingested against the + brain and re-queue any that are missing, so lost + transcripts import again. Bounded (--limit, default + ${DEFAULT_RECONCILE_LIMIT}) and resumable; backs up the state file first. + With --dry-run it only reports. Options: --quiet Suppress per-file output (still prints summary). --benchmark Time the run; report bytes-per-second + total. - --include-unattributed Ingest sessions with no resolvable git remote. + --include-unattributed Ingest sessions with no resolvable git remote into the + machine-local, never-federated gbrain source + ${UNATTRIBUTED_SOURCE_ID}. Refused when the brain is + remote (Postgres or HTTP); such pages stay local. --all-history Walk transcripts older than 90 days too. --sources Comma-separated subset: ${ALL_TYPES.join(",")} Default: every type, minus transcript unless transcript_ingest_mode is recent (90 days) or all (all history). A list naming transcript overrides it. - --limit Stop after N pages written (smoke testing). + --limit Stop after N pages written (smoke testing). With + --reconcile: check at most N entries this pass. + --dry-run With --reconcile: report what would change, change nothing. + --request-reconcile Record that a reconcile pass is pending (no gbrain + calls); the next ingest run does one bounded pass. --no-write Dry run: prepare pages and report counts, but import nothing and leave the state file untouched. --scan-secrets Opt-in gitleaks scan of outgoing rendered pages, including @@ -281,6 +317,8 @@ function parseArgs(): CliArgs { let sources: Set = new Set(ALL_TYPES); let sourcesExplicit = false; let noWrite = process.env.GSTACK_MEMORY_INGEST_NO_WRITE === "1"; + let dryRun = false; + let requestReconcile = false; let scanSecrets = process.env.GSTACK_MEMORY_INGEST_SCAN_SECRETS === "1"; for (let i = 0; i < args.length; i++) { @@ -289,6 +327,9 @@ function parseArgs(): CliArgs { case "--probe": mode = "probe"; break; case "--incremental": mode = "incremental"; break; case "--bulk": mode = "bulk"; break; + case "--reconcile": mode = "reconcile"; break; + case "--dry-run": dryRun = true; break; + case "--request-reconcile": requestReconcile = true; break; case "--quiet": quiet = true; break; case "--benchmark": benchmark = true; break; case "--include-unattributed": includeUnattributed = true; break; @@ -331,57 +372,31 @@ function parseArgs(): CliArgs { } } if (consent.window === "all") allHistory = true; + if (dryRun && mode !== "reconcile") { + console.error("--dry-run applies to --reconcile; use --no-write for an ingest dry run"); + process.exit(1); + } - return { mode, quiet, benchmark, includeUnattributed, allHistory, sources, limit, noWrite, scanSecrets }; + return { mode, quiet, benchmark, includeUnattributed, allHistory, sources, limit, noWrite, dryRun, requestReconcile, scanSecrets }; } // ── State file ───────────────────────────────────────────────────────────── +// Schema, transitions, migration and the lock live in lib/memory-ingest-landing.ts. function loadState(): IngestState { - if (!existsSync(STATE_PATH)) { - return { - schema_version: 1, - last_writer: "gstack-memory-ingest", - sessions: {}, - }; - } - try { - const raw = readFileSync(STATE_PATH, "utf-8"); - const parsed = JSON.parse(raw) as IngestState; - if (parsed.schema_version !== 1) { - console.error(`State file at ${STATE_PATH} has unknown schema_version ${parsed.schema_version}; backing up + resetting.`); - try { - writeFileSync(STATE_PATH + ".bak", raw, "utf-8"); - } catch { - // backup failure is non-fatal - } - return { schema_version: 1, last_writer: "gstack-memory-ingest", sessions: {} }; - } - return parsed; - } catch (err) { - console.error(`State file at ${STATE_PATH} corrupt; backing up + resetting.`); - try { - const raw = readFileSync(STATE_PATH, "utf-8"); - writeFileSync(STATE_PATH + ".bak", raw, "utf-8"); - } catch { - // best-effort - } - return { schema_version: 1, last_writer: "gstack-memory-ingest", sessions: {} }; + const loaded = loadStateFile(STATE_PATH); + if (loaded.kind === "unreadable") { + console.error( + `State file at ${STATE_PATH} is unreadable (${loaded.reason}); ` + + (loaded.backup ? `backed up to ${loaded.backup} and ` : "") + "starting fresh.", + ); } + return loaded.state; } +/** Throws when the state file cannot be written: the run then fails (A1). */ function saveState(state: IngestState): void { - // F6 (Codex finding 6): tmp+rename atomic write so a crash mid-write - // never leaves a truncated/corrupt state file. Matches the pattern - // in gstack-gbrain-sync.ts:saveSyncState. - try { - mkdirSync(dirname(STATE_PATH), { recursive: true }); - const tmp = `${STATE_PATH}.tmp.${process.pid}`; - writeFileSync(tmp, JSON.stringify(state, null, 2), "utf-8"); - renameSync(tmp, STATE_PATH); - } catch (err) { - console.error(`[state] write failed: ${(err as Error).message}`); - } + saveStateFile(STATE_PATH, state); } // ── File hash + change detection ─────────────────────────────────────────── @@ -402,7 +417,7 @@ function fileSha256(path: string): string { function fileChangedSinceState(path: string, state: IngestState, verifyHash = false): boolean { const entry = state.sessions[path]; - if (!entry) return true; + if (isRetryable(entry)) return true; try { const st = statSync(path); const mtimeNs = Math.floor(st.mtimeMs * 1e6); @@ -918,23 +933,23 @@ function buildArtifactPage(path: string, type: MemoryType, raw?: string): PageRe // ── Writer (batch via `gbrain import `) ─────────────────────────────── // -// Architecture (post plan-eng-review + Codex outside-voice): +// Architecture (write, verify, then stamp — A1): // // walkAllSources(ctx) -// → for each path: mtime-skip / parse / buildPage -// → renderPageBody injects title/type/tags; opt-in scan checks these bytes +// → for each path: mtime-skip / read bytes once (fingerprint) / parse / buildPage +// → renderPageBody injects title/type/tags + gstack_content_sha256 // → writeStaged: mkdir -p slug subdirs (D1), write ${slug}.md -// → snapshot ~/.gbrain/sync-failures.jsonl byte-offset (D7) -// → spawnSync `gbrain import --no-embed --json` (D6) -// → parseImportJson(stdout) → { imported, skipped, errors, ... } (D6 OK/ERR) -// → readNewFailures(preImportOffset, slugMap) → Set (D7) -// → state.sessions[path] = { ... } for prepared files NOT in failed set -// → saveStateAtomic (F6 tmp+rename) + cleanupStagingDir +// → one staging dir + `gbrain import --json [--source-id]` per source (D6) +// → classifyImport: named failures from --json `failures`, stderr +// "Skipped :" lines and the sync-failures ledger (written only for +// git import dirs, so usually empty here); an unnamed failure refuses +// the whole batch, and a batch refused 3 times is bisected +// → pages not named: imported_unverified, then the landing check +// (gbrain list per source, bounded gbrain get --json) stamps ingested +// → state saved atomically by main(); a save failure fails the run // -// We trust gbrain's content_hash idempotency (verified in -// ~/git/gbrain/src/core/import-file.ts:242-243, :478) — repeated imports -// of identical content are cheap. So we do NOT track per-file skip_reasons, -// do NOT keep a SIGTERM checkpoint, and do NOT advance a three-state verdict. +// gbrain's content_hash makes re-importing identical content cheap, so a +// page that did not land is simply retried on the next run. let _gbrainAvailability: boolean | null = null; function gbrainAvailable(): boolean { @@ -1014,7 +1029,10 @@ interface PreparedPage { source_path: string; /** Full markdown including frontmatter — ready to write. */ rendered_body: string; - source_fingerprint?: SourceFingerprint; + /** mtime + sha256 of the exact bytes parsed (A1): the stamp never describes a newer file. */ + source_fingerprint: SourceFingerprint; + /** gstack_content_sha256 written into the page's frontmatter; the landing check compares it. */ + content_sha256: string; /** Carry-through fields for state recording on success. */ page_slug: string; partial: boolean; @@ -1034,11 +1052,20 @@ function sourceFingerprintForStamp(page: PreparedPage): SourceFingerprint | null sha256: fileSha256(page.source_path), }; const prepared = page.source_fingerprint; - if (!prepared) return current; if (current.mtime_ns !== prepared.mtime_ns || current.sha256 !== prepared.sha256) return null; return prepared; } +/** + * Add the gstack_content_sha256 frontmatter field (a hash of the rendered + * page without it). gbrain stores its own content hash over the parsed page, + * which gstack cannot recompute, so the landing check compares this field. + */ +export function withContentHash(rendered: string): { body: string; sha: string } { + const sha = createHash("sha256").update(rendered).digest("hex"); + return { body: `---\ngstack_content_sha256: ${sha}\n${rendered.slice(4)}`, sha }; +} + interface StagingResult { staging_dir: string; written: number; @@ -1099,52 +1126,16 @@ function writeStaged(prepared: PreparedPage[], stagingDir: string, scanned = fal return { staging_dir: stagingDir, written, errors, stagedPathToSource }; } -interface ImportJsonResult { - status?: string; - duration_s?: number; - imported?: number; - skipped?: number; - errors?: number; - chunks?: number; - total_files?: number; -} - -/** - * Parse the `gbrain import --json` stdout payload (single JSON object on - * the last non-empty line per commands/import.ts:271-275). - * - * Returns parsed counts on success, or `null` to signal "unparseable" — the - * caller treats null as ERR (system_error) rather than silently passing - * through as zeros. Pre-2026-05-11 this returned zeros on parse failure, - * which silently masked gbrain crashes as "0 imported, 0 failed = OK". - */ -function parseImportJson(stdout: string): ImportJsonResult | null { - const lines = stdout.split("\n").map((s) => s.trim()).filter(Boolean); - for (let i = lines.length - 1; i >= 0; i--) { - const line = lines[i]; - if (line.startsWith("{") && line.endsWith("}")) { - try { - const parsed = JSON.parse(line); - if (typeof parsed === "object" && parsed && "imported" in parsed) { - return parsed as ImportJsonResult; - } - } catch { - // try next line up - } - } - } - return null; -} - /** * Read failures appended to ~/.gbrain/sync-failures.jsonl since the * snapshotted byte offset, and map them back to source paths. * - * D7: gbrain import writes per-file failures to sync-failures.jsonl - * (commands/import.ts:308-310) explicitly so "callers can gate state - * advances" (comment at :28). We snapshot the file size before import - * and read only the appended bytes after, so we never confuse new - * entries with prior-run leftovers. + * D7: gbrain import writes per-file failures to sync-failures.jsonl so + * "callers can gate state advances", but ONLY when the import dir is a git + * repo, and the staging dir never is. So for staged imports this is usually + * empty; classifyImport() also reads the --json `failures` list and stderr. + * We snapshot the file size before import and read only the appended bytes + * after, so we never confuse new entries with prior-run leftovers. * * Each line is `{ path, error, code, commit, ts }`. The `path` is the * staging-dir-relative filename gbrain saw (e.g. "transcripts/foo.md"). @@ -1184,9 +1175,7 @@ export function readNewFailures( closeSync(fd); } } catch { - // Best-effort. If we can't read failures, we conservatively assume - // none — caller will state-record all prepared files. Worst case: - // failed files get a retry-on-next-run shot anyway via content_hash. + // Best-effort: the landing check still refuses pages that did not land. } return failed; } @@ -1427,7 +1416,7 @@ async function probeMode(args: CliArgs): Promise { * self-heal on the next state write. */ export function disambiguateSlugs( - pages: PreparedPage[], + pages: Array>, state?: { sessions: Record }, ): void { // slug → owning source_path, from prior runs. First writer wins on legacy @@ -1529,21 +1518,23 @@ function preparePages( for (const { path, type } of walkAllSources(ctx)) { if (args.limit !== null && !policyStoreExists && prepared.length >= args.limit) break; - if (args.mode === "incremental" && !fileChangedSinceState(path, state, args.scanSecrets)) { + // Incremental skips unchanged sources; a quarantined page is retried + // only once its source changes, in every mode. + const known = state.sessions[path]; + if ((args.mode === "incremental" || known?.status === "quarantined") && !fileChangedSinceState(path, state, args.scanSecrets)) { skippedDedup++; continue; } let page: PageRecord; - let sourceFingerprint: SourceFingerprint | undefined; + let sourceFingerprint: SourceFingerprint; try { - let raw: string | undefined; - if (args.scanSecrets) { - const mtime_ns = Math.floor(statSync(path).mtimeMs * 1e6); - const bytes = readFileSync(path); - sourceFingerprint = { mtime_ns, sha256: createHash("sha256").update(bytes).digest("hex") }; - raw = bytes.toString("utf-8"); - } + // A1: fingerprint the exact bytes that are parsed, so a transcript that + // grows during the run is never stamped as its newer self. + const mtime_ns = Math.floor(statSync(path).mtimeMs * 1e6); + const bytes = readFileSync(path); + sourceFingerprint = { mtime_ns, sha256: createHash("sha256").update(bytes).digest("hex") }; + const raw = bytes.toString("utf-8"); if (type === "transcript") { const session = parseTranscriptJsonl(path, raw); if (!session) { @@ -1568,7 +1559,7 @@ function preparePages( continue; } - const renderedBody = renderPageBody(page); + const { body: renderedBody, sha: contentSha } = withContentHash(renderPageBody(page)); // Optional belt-and-suspenders: when --scan-secrets is set, gitleaks the // rendered page — the exact bytes writeStaged() hands to gbrain — and @@ -1602,6 +1593,7 @@ function preparePages( source_path: path, rendered_body: renderedBody, source_fingerprint: sourceFingerprint, + content_sha256: contentSha, page_slug: page.slug, partial: page.partial ?? false, type, @@ -1701,10 +1693,11 @@ function preparePages( * Make a per-run staging directory at ~/.gstack/.staging-ingest--/ * The pid+ts namespace avoids collisions when two ingest passes run * concurrently (the orchestrator's lock should prevent this, but - * defense-in-depth). + * defense-in-depth). A target gbrain source rides in the name as + * `-src-` so a resumed import (#1611) goes to the same source. */ -function makeStagingDir(): string { - const dir = join(GSTACK_HOME, `.staging-ingest-${process.pid}-${Date.now()}`); +function makeStagingDir(suffix = ""): string { + const dir = join(GSTACK_HOME, `.staging-ingest-${process.pid}-${Date.now()}${suffix}`); mkdirSync(dir, { recursive: true }); // Mint the ownership marker (#1802) so cleanupStagingDir() and decideResume() // can prove this dir was created by us before any recursive delete or resume. @@ -1918,8 +1911,9 @@ function failedOnUnknownIncludeGitignored(status: number | null, stderr: string) async function runGbrainImport( stagingDir: string, timeoutMs: number, + sourceId: string | null = null, ): Promise<{ status: number | null; stdout: string; stderr: string; timedOut: boolean }> { - const first = await runGbrainImportOnce(stagingDir, timeoutMs, true); + const first = await runGbrainImportOnce(stagingDir, timeoutMs, true, sourceId); if (failedOnUnknownIncludeGitignored(first.status, first.stderr)) { // Older gbrain: retry without the flag. If .gitignore then hides the // staged pages, the imported { installSignalForwarder(); return new Promise((resolve) => { @@ -1984,6 +1979,7 @@ function runGbrainImportOnce( stagingDir, "--no-embed", ...(includeGitignored ? ["--include-gitignored"] : []), + ...(sourceId ? ["--source-id", sourceId] : []), "--json", ], { baseEnv }, @@ -2029,9 +2025,8 @@ function runGbrainImportOnce( }); } -async function ingestPass(args: CliArgs): Promise { +async function ingestPass(args: CliArgs, state: IngestState): Promise { const t0 = Date.now(); - const state = loadState(); const ctx = makeWalkContext(args, state); const remoteHttpMode = isRemoteHttpMcpMode(); const resumeDir = process.env.GSTACK_INGEST_RESUME_DIR; @@ -2053,6 +2048,7 @@ async function ingestPass(args: CliArgs): Promise { if (prep.policyError) { console.error(`[memory-ingest] ERR: ${prep.policyError}`); return { + state_untouched: true, written: 0, skipped_secret: prep.skippedSecret, skipped_dedup: prep.skippedDedup, @@ -2077,6 +2073,7 @@ async function ingestPass(args: CliArgs): Promise { ); } return { + state_untouched: true, written: 0, skipped_secret: prep.skippedSecret, skipped_dedup: prep.skippedDedup, @@ -2090,11 +2087,14 @@ async function ingestPass(args: CliArgs): Promise { } if (prep.prepared.length === 0) { - // Nothing to import — still touch state.last_full_walk and exit. + // Nothing to import — still touch state.last_full_walk, and re-check pages + // an earlier run could not verify (A1: checked once a lookup works). state.last_full_walk = new Date().toISOString(); state.last_writer = "gstack-memory-ingest"; - saveState(state); + const hasUnverified = Object.values(state.sessions).some((e) => e.status === "imported_unverified"); + const check = hasUnverified && !remoteHttpMode && gbrainAvailable() ? verifyImported(state, new Set()) : null; return { + ...(check ? { unverified: check.unverified, refused: check.requeued } : {}), written: 0, skipped_secret: prep.skippedSecret, skipped_dedup: prep.skippedDedup, @@ -2112,6 +2112,7 @@ async function ingestPass(args: CliArgs): Promise { "gbrain CLI not in PATH or missing `import` subcommand. Run /setup-gbrain."; console.error(`[memory-ingest] ERR: ${msg}`); return { + state_untouched: true, written: 0, skipped_secret: prep.skippedSecret, skipped_dedup: prep.skippedDedup, @@ -2125,19 +2126,19 @@ async function ingestPass(args: CliArgs): Promise { }; } - // Phase 2: stage + (optionally) invoke gbrain import. + // Phase 2: stage + import, verify, then stamp (A1, INV-2). // // Split-engine branch per plan D11: in remote-http MCP mode, we stage to a // PERSISTENT dir under ~/.gstack/transcripts/ and SKIP `gbrain import` // entirely. gstack-brain-sync push will pick the dir up via its allowlist // and the brain admin's pull job will index transcripts into the remote - // brain. Local PGLite (if any) stays code-only. + // brain. Such pages are recorded as `staged`, never `ingested`: this + // machine cannot see whether the remote brain indexed them. // // Resume branch for #1611: when the orchestrator sets // GSTACK_INGEST_RESUME_DIR (because gbrain's import-checkpoint.json points // at an existing dir from a prior SIGTERM'd run), reuse that staging dir - // and skip the prepare/writeStaged phase entirely. gbrain's checkpoint - // tells it where to resume. + // and skip writeStaged. gbrain's checkpoint tells it where to resume. // #1802 second entry point: this binary is runnable directly, so it must not // trust GSTACK_INGEST_RESUME_DIR just because it exists — a stale/poisoned env // could make us `gbrain import` (and later clean up) an arbitrary directory. @@ -2147,136 +2148,294 @@ async function ingestPass(args: CliArgs): Promise { `[memory-ingest] ignoring GSTACK_INGEST_RESUME_DIR="${resumeDir}" — not a proven staging dir (#1802); staging fresh.`, ); } - const stagingDir = resuming - ? resumeDir! - : remoteHttpMode - ? makePersistentTranscriptDir() - : makeStagingDir(); + const base = { + skipped_secret: prep.skippedSecret, + skipped_dedup: prep.skippedDedup, + skipped_unattributed: prep.skippedUnattributed, + skipped_policy_readonly: prep.skippedPolicyReadonly, + skipped_policy_deny: prep.skippedPolicyDeny, + partial_pages: prep.partialPages, + }; + + if (remoteHttpMode) return stageForRemoteBrain(args, prep.prepared, state, base, t0); + + const enforceResumePolicy = resuming && hasRepoPolicyStore(); + let resumed: PreStaged | undefined; + if (resuming) { + const r = verifyResumeDir(args, prep, resumeDir!, enforceResumePolicy); + if ("error" in r) { + return { + state_untouched: true, + ...base, + written: 0, + skipped_secret: prep.skippedSecret + (r.scannerFailed ? 1 : 0), + failed: prep.parseFailed + prep.prepared.length, + duration_ms: Date.now() - t0, + system_error: r.error, + }; + } + resumed = r; + } + + // Egress receipt BEFORE the import (fail-closed): the gbrain DB may be a + // remote Postgres, so the ingest is a potential off-machine send. The + // gbrain subprocess owns the wire bytes (content-free receipt, sha256 + // null). The remote-http branch above stages locally only — its egress + // happens in gstack-brain-sync, which writes its own receipt at the push. + try { + writeReceipt({ + sink: "memory-ingest", + host: "gbrain-db (user-configured DATABASE_URL)", + payloadClass: `transcript-pages count=${resumed ? resumed.staged.size : prep.prepared.length} (sent by gbrain subprocess)`, + bytes: 0, + sha256: null, + consent: "gbrain setup consent (/setup-gbrain)", + }); + } catch (err) { + const msg = `EGRESS_RECEIPT_FAILED: ${(err as Error).message} — ingest refused`; + console.error(`[memory-ingest] ERR: ${msg}`); + if (resumed && !args.scanSecrets) cleanupStagingDir(resumed.stagingDir); + return { ...base, state_untouched: true, written: 0, failed: prep.parseFailed + prep.prepared.length, duration_ms: Date.now() - t0, system_error: msg }; + } + + const run: RunContext = { + args, + state, + nowIso: new Date().toISOString(), + written: 0, + failed: 0, + refused: 0, + quarantined: 0, + importedThisRun: new Set(), + }; + // D6: one batch import per gbrain source, sequentially (PGLite is a single + // writer). `--no-embed` matches the prior per-file behavior; `--json` gives + // structured counts and per-file failures. + for (const batch of resumed ? [{ sourceId: resumed.sourceId, pages: [...resumed.staged.values()] }] : planBatches(prep.prepared)) { + await importPartition(run, batch.pages, batch.sourceId, resumed); + if (run.systemError) break; + } + + // Landing check: confirm what this run (and earlier runs) imported is in + // its recorded source before stamping it ingested. Bounded per run. + const check = verifyImported(state, run.importedThisRun); + run.written -= check.requeuedThisRun; + run.refused += check.requeued; + run.failed += check.requeued; + + state.last_full_walk = new Date().toISOString(); + state.last_writer = "gstack-memory-ingest"; + return { + ...base, + written: run.written, + failed: run.failed + prep.parseFailed, + refused: run.refused, + unverified: check.unverified, + quarantined: run.quarantined, + duration_ms: Date.now() - t0, + ...(run.systemError ? { system_error: run.systemError } : {}), + }; +} + +interface PreStaged { + stagingDir: string; + sourceId: string | null; + /** Staged rel path → the prepared page whose rendered body it holds. */ + staged: Map; + /** Keep the dir when anything in it did not land (scanned or policy-checked resume). */ + preserveOnShortfall: boolean; + /** Pages the import count is checked against (saved pages on a strict resume). */ + expected?: number; +} + +interface RunContext { + args: CliArgs; + state: IngestState; + nowIso: string; + written: number; + failed: number; + refused: number; + quarantined: number; + importedThisRun: Set; + systemError?: string; +} + +/** Pages grouped by the gbrain source they import into; null = gbrain's own default routing. */ +function planBatches(pages: PreparedPage[]): Array<{ sourceId: string | null; pages: PreparedPage[] }> { + return pages.length > 0 ? [{ sourceId: null, pages }] : []; +} + +/** Print even under --quiet: a page that did not land must never be silent (A1). */ +function reportPage(msg: string): void { + console.error(`[memory-ingest] ${msg}`); +} + +function stampEntry(run: RunContext, p: PreparedPage, next: Pick & Partial): boolean { + let fingerprint: SourceFingerprint | null; + try { + fingerprint = sourceFingerprintForStamp(p); + } catch (err) { + console.error(`[state-record] ${p.source_path}: ${(err as Error).message}`); + return false; + } + // Changed since it was parsed: leave it for the next run, which sees the change. + if (!fingerprint) return false; + setEntry(run.state, p.source_path, { + ...fingerprint, + page_slug: p.page_slug, + ...(p.partial ? { partial: true } : {}), + content_sha256: p.content_sha256, + ...next, + }); + if (run.state.batch_refusals) delete run.state.batch_refusals[p.source_path]; + return true; +} + +/** + * Pages that did not land are never stamped: a new page keeps no entry and a + * changed page keeps its previous one, so the next run retries both. A + * whole-batch refusal is counted toward bisecting the batch. + */ +function markRefused(run: RunContext, pages: PreparedPage[], batchRefusal: boolean): void { + run.refused += pages.length; + if (!batchRefusal) return; + const counts = (run.state.batch_refusals ??= {}); + for (const p of pages) counts[p.source_path] = (counts[p.source_path] ?? 0) + 1; +} + +type BatchOutcome = { kind: "ok" } | { kind: "error" } | { kind: "refused"; pages: PreparedPage[]; reason: string; stagingDir: string }; + +/** + * Import one partition. A batch refused as a whole (a failure gbrain did not + * attribute to a page) is retried by later runs; after BISECT_AFTER_REFUSALS + * refusals it is split in halves to isolate the page that poisons it. + */ +async function importPartition(run: RunContext, pages: PreparedPage[], sourceId: string | null, pre?: PreStaged): Promise { + const outcome = await importAndApply(run, pages, sourceId, pre); + if (outcome.kind !== "refused") return; + const refusals = 1 + Math.max(0, ...outcome.pages.map((p) => run.state.batch_refusals?.[p.source_path] ?? 0)); + if (!pre && refusals >= BISECT_AFTER_REFUSALS) { + reportPage(`batch of ${outcome.pages.length} page(s) refused ${refusals} times (${outcome.reason}); importing it in halves to find the page gbrain cannot take`); + await isolatePoisonPage(run, outcome.pages, sourceId, outcome.reason); + return; + } + markRefused(run, outcome.pages, true); + run.systemError ??= + `${outcome.reason}. Refusing to advance state — the unaccounted pages would be marked ingested without ` + + `landing in the brain. If the count is 0, check whether ${outcome.stagingDir} is inside a git ` + + `repo that ignores it (gbrain import honours .gitignore).`; + console.error(`[memory-ingest] ERR: ${run.systemError}`); +} + +async function isolatePoisonPage(run: RunContext, group: PreparedPage[], sourceId: string | null, reason: string): Promise { + if (group.length === 1) { + const p = group[0]; + if (stampEntry(run, p, { status: "quarantined", source_id: sourceId ?? LEGACY_SOURCE_ID, reason: `refused alone: ${reason}` })) run.quarantined++; + reportPage(`quarantined ${p.page_slug} (${p.source_path}): gbrain refuses every batch that contains it (${reason}). It is retried when the source file changes.`); + return; + } + const mid = Math.ceil(group.length / 2); + const refusedHalves: Array<{ pages: PreparedPage[]; reason: string }> = []; + for (const half of [group.slice(0, mid), group.slice(mid)]) { + const outcome = await importAndApply(run, half, sourceId); + if (outcome.kind === "error") return; + if (outcome.kind === "refused") refusedHalves.push(outcome); + } + if (refusedHalves.length === 2) { + markRefused(run, group, true); + run.systemError ??= `${reason}; both halves of the batch were refused too, so no single page is to blame. Refusing to advance state.`; + console.error(`[memory-ingest] ERR: ${run.systemError}`); + return; + } + if (refusedHalves.length === 1) await isolatePoisonPage(run, refusedHalves[0].pages, sourceId, refusedHalves[0].reason); +} + +/** Stage, import and apply named failures and landings; a whole-batch refusal is returned to the caller. */ +async function importAndApply(run: RunContext, pages: PreparedPage[], sourceId: string | null, pre?: PreStaged): Promise { + const { args } = run; + const r = await importBatch(args, pages, sourceId, pre); + run.failed += r.stageErrors; + if (r.systemError) { + run.systemError ??= r.systemError; + run.failed += r.staged.size; + return { kind: "error" }; + } + const report = r.report!; + const verdict = r.verdict!; + const recordedSource = report.source_id ?? sourceId ?? LEGACY_SOURCE_ID; + if (verdict.refuseAll) return { kind: "refused", pages: [...r.staged.values()], reason: verdict.refuseAll, stagingDir: r.stagingDir }; + + for (const [rel, error] of verdict.named) { + const p = r.staged.get(rel)!; + run.failed++; + markRefused(run, [p], false); + reportPage(`FAILED ${rel}: ${error.slice(0, 300)} (left un-stamped; retried next run)`); + } + let stamped = 0; + for (const [rel, p] of r.staged) { + if (verdict.named.has(rel)) continue; + if (!stampEntry(run, p, { status: "imported_unverified", source_id: recordedSource, ingested_at: run.nowIso, reason: undefined })) continue; + run.written++; + stamped++; + run.importedThisRun.add(p.source_path); + if (!args.quiet) console.log(`[${run.written}] ${p.page_slug}${p.partial ? " [partial]" : ""}`); + } + // A scanned or policy-checked resume keeps its saved stage until every + // saved page landed and was stamped. + if (pre?.preserveOnShortfall && stamped >= (pre.expected ?? r.staged.size) && verdict.named.size === 0) { + cleanupStagingDir(pre.stagingDir); + } + if (!args.quiet) { + console.error( + `[memory-ingest] gbrain import: ${report.imported ?? 0} imported, ` + + `${report.unchanged ?? report.skipped ?? 0} unchanged, ${verdict.named.size} failed` + + (verdict.named.size > 0 ? " (named above; retried next run)" : ""), + ); + } + // Silent-zero pathology detector (#2144's other half): pages were staged + // but NOTHING imported or skipped-as-unchanged. That shape hid the dead + // ingest for months — it must be loud even under --quiet. + if (r.staged.size > 0 && (report.imported ?? 0) + (report.skipped ?? 0) === 0 && (report.errors ?? 0) === 0) { + console.error( + `[memory-ingest] WARNING: ${r.staged.size} page(s) staged but gbrain collected ZERO ` + + `(no imports, no unchanged-skips, no errors). This is the #2144 silent-zero shape — ` + + `check gbrain's import.collect_files log line and your gbrain version.`, + ); + } + return { kind: "ok" }; +} + +interface BatchRun { + staged: Map; + stagingDir: string; + stageErrors: number; + report?: ImportReport; + verdict?: BatchVerdict; + systemError?: string; +} + +async function importBatch(args: CliArgs, pages: PreparedPage[], sourceId: string | null, pre?: PreStaged): Promise { + const stagingDir = pre?.stagingDir ?? makeStagingDir(sourceId ? `-src-${sourceId}` : ""); // Register staging dir with the signal forwarder so SIGTERM/SIGINT can // either preserve (when gbrain checkpointed it) or synchronously clean up. - // The async finally block below does NOT run after a signal-handler exit. - // In remote-http mode we skip registration — the dir is meant to persist. - if (!remoteHttpMode) { - _activeStagingDir = stagingDir; - } + _activeStagingDir = stagingDir; // #1802 C3: set when the import-timeout branch leaves a resumable checkpoint - // pointing at this staging dir, so the finally preserves it for the next run - // instead of deleting it (the SIGTERM forwarder's preserve branch only runs - // when the PARENT is signalled, which an internal timeout never does). - let preserveStaging = resuming && args.scanSecrets; - const enforceResumePolicy = resuming && hasRepoPolicyStore(); + // pointing at this staging dir, so the finally preserves it for the next run. + let preserveStaging = !!pre?.preserveOnShortfall && args.scanSecrets; + let staged: Map; + let stageErrors = 0; try { - let staging: StagingResult; - if (resuming) { - const stagedPagePaths = new Set(); - const stagedPathToSource = new Map(); - if (args.scanSecrets || enforceResumePolicy) { - try { - if (enforceResumePolicy && !prep.policyStoreExists) { - throw new Error("[repo policy] policy store appeared after source preparation"); - } - const eligiblePages = enforceResumePolicy - ? new Map(prep.prepared.map((p) => [stagedRelPath(p.slug), p])) - : null; - const pending = [stagingDir]; - while (pending.length > 0) { - const dir = pending.pop()!; - for (const entry of readdirSync(dir, { withFileTypes: true })) { - const path = join(dir, entry.name); - if (entry.isDirectory()) pending.push(path); - else if (entry.isFile()) { - if (path === join(stagingDir, STAGING_MARKER)) continue; - if (args.scanSecrets) { - const scan = secretScanFile(path); - if (!scan.scanned || scan.scanner !== "gitleaks" || scan.findings.length > 0) { - const reason = scan.scanned ? "match" : scan.scanner; - throw new Error(`[secret-scan ${reason}] ${path}`); - } - } - if (entry.name.endsWith(".md")) { - const relPath = relative(stagingDir, path).split("\\").join("/"); - stagedPagePaths.add(relPath); - if (eligiblePages) { - const page = eligiblePages.get(relPath); - if (!page || readFileSync(path, "utf-8") !== page.rendered_body) { - throw new Error(`[repo policy] staged page is not a current permitted source: ${relPath}`); - } - stagedPathToSource.set(relPath, page.source_path); - } - } else if (enforceResumePolicy) { - throw new Error(`[repo policy] unrecognized staged file: ${path}`); - } - } else { - throw new Error(`[${args.scanSecrets ? "secret-scan error" : "repo policy"}] unsupported staging entry: ${path}`); - } - } - } - if (stagedPagePaths.size === 0) { - throw new Error(`[${args.scanSecrets ? "secret-scan error" : "repo policy"}] resumed staging contains no pages`); - } - if (!enforceResumePolicy) { - for (const p of prep.prepared) { - const path = stagedRelPath(p.slug); - if (stagedPagePaths.has(path) && readFileSync(join(stagingDir, path), "utf-8") === p.rendered_body) { - stagedPathToSource.set(path, p.source_path); - } - } - } - } catch (err) { - preserveStaging = true; - const cause = (err as Error).message; - const scannerFailed = cause.startsWith("[secret-scan"); - const msg = `${cause}; resumed import refused. Staging preserved; ` + - (scannerFailed - ? "repair gitleaks and retry, or rerun without resume to restage." - : "rerun without resume to restage under the current repo policy."); - console.error(`[memory-ingest] ERR: ${msg}`); - return { - written: 0, - skipped_secret: prep.skippedSecret + (scannerFailed ? 1 : 0), - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed: prep.parseFailed + prep.prepared.length, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, - system_error: msg, - }; - } - } - // Pages are already on disk from the previous run. Skip writeStaged. - // The "written" count for the verdict reflects what's on disk now; - // gbrain's import will skip already-completed entries via its own - // checkpoint (processedIndex+1). - if (!args.quiet) { - console.error( - `[memory-ingest] resuming previous staging dir ${stagingDir} (skipping prepare phase)`, - ); - } - // #1802 C4: reconstruct stagedPathToSource from the prepared pages so - // readNewFailures() can still map gbrain's per-file failures back to - // sources on resume. An empty map made every failed file fall through to - // state-recording — i.e. silently marked ingested despite failing. - if (!args.scanSecrets && !enforceResumePolicy) { - for (const p of prep.prepared) { - stagedPathToSource.set(stagedRelPath(p.slug), p.source_path); - } - } - staging = { - staging_dir: stagingDir, - written: args.scanSecrets || enforceResumePolicy ? stagedPagePaths.size : prep.prepared.length, - errors: [], - stagedPathToSource, - }; + if (pre) { + staged = pre.staged; + if (!args.quiet) console.error(`[memory-ingest] resuming previous staging dir ${stagingDir} (skipping prepare phase)`); } else { - staging = writeStaged(prep.prepared, stagingDir, args.scanSecrets); - } - failed += staging.errors.length; - if (!args.quiet && staging.errors.length > 0) { - for (const e of staging.errors.slice(0, 5)) { - console.error(`[stage-error] ${e.slug}: ${e.error}`); - } + const staging = writeStaged(pages, stagingDir, args.scanSecrets); + stageErrors = staging.errors.length; + if (!args.quiet) for (const e of staging.errors.slice(0, 5)) console.error(`[stage-error] ${e.slug}: ${e.error}`); + staged = new Map(pages + .filter((p) => staging.stagedPathToSource.get(stagedRelPath(p.slug)) === p.source_path) + .map((p) => [stagedRelPath(p.slug), p])); + if (!args.quiet) console.error(`[memory-ingest] staged ${staged.size} pages → ${stagingDir}; running gbrain import...`); } // D7: snapshot sync-failures.jsonl byte-offset before import so we @@ -2284,344 +2443,229 @@ async function ingestPass(args: CliArgs): Promise { const syncFailuresPath = join(homedir(), ".gbrain", "sync-failures.jsonl"); let preImportOffset = 0; try { - if (existsSync(syncFailuresPath)) { - preImportOffset = statSync(syncFailuresPath).size; - } + if (existsSync(syncFailuresPath)) preImportOffset = statSync(syncFailuresPath).size; } catch { - // best-effort; absent file → 0 offset, all future entries are "new" + // best-effort; absent file → 0 offset } - if (!args.quiet) { - const action = remoteHttpMode - ? "persisting to artifacts pipeline (skipping local gbrain import — remote-http mode)" - : "running gbrain import"; - console.error( - `[memory-ingest] staged ${staging.written} pages → ${stagingDir}; ${action}...`, - ); - } - - // Remote-http branch (split-engine D11): no local gbrain import. The - // staged markdown lives under ~/.gstack/transcripts// and the - // next gstack-brain-sync push will move it to the artifacts repo. From - // there the brain admin's pull job indexes into the remote brain. - // - // We treat ALL prepared pages as "written" since the import didn't run - // and we have no per-page failures from gbrain to filter on. The - // brain admin's pull pipeline is the authoritative gate; from this - // machine's perspective, the act of staging IS the write. - if (remoteHttpMode) { - const nowIso = new Date().toISOString(); - for (const p of prep.prepared) { - try { - if (args.scanSecrets && staging.stagedPathToSource.get(stagedRelPath(p.slug)) !== p.source_path) continue; - const fingerprint = sourceFingerprintForStamp(p); - if (!fingerprint) continue; - state.sessions[p.source_path] = { - ...fingerprint, - ingested_at: nowIso, - page_slug: p.page_slug, - partial: p.partial, - }; - written++; - } catch (err) { - console.error( - `[state-record] ${p.source_path}: ${(err as Error).message}`, - ); - } - } - state.last_full_walk = nowIso; - state.last_writer = "gstack-memory-ingest (remote-http mode)"; - saveState(state); - if (!args.quiet) { - console.error( - `[memory-ingest] persisted ${written} pages to ${stagingDir} (brain admin will index on next pull)`, - ); - } - // Skip the gbrain-import error handling + cleanupStagingDir paths - // below by short-circuiting the function. - return { - written, - skipped_secret: prep.skippedSecret, - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, - }; - } - - // D6: single batch import. `--no-embed` matches the prior per-file - // behavior (we never enabled embedding); embeddings happen on-demand - // via gbrain's own pipelines. `--json` gives us structured counts. - // - // Async spawn (not spawnSync) so the signal forwarder installed in - // runGbrainImport propagates SIGTERM/SIGINT to the child. With sync - // spawn, parent termination orphans the gbrain process (observed - // during 2026-05-10 cold-run testing — gbrain kept running 15 min - // after the orchestrator timed out). - // - // Egress receipt BEFORE the import (fail-closed): the gbrain DB may be a - // remote Postgres, so the ingest is a potential off-machine send. The - // gbrain subprocess owns the wire bytes (content-free receipt, sha256 - // null). The remote-http branch above stages locally only — its egress - // happens in gstack-brain-sync, which writes its own receipt at the push. - try { - writeReceipt({ - sink: "memory-ingest", - host: "gbrain-db (user-configured DATABASE_URL)", - payloadClass: `transcript-pages count=${staging.written} (sent by gbrain subprocess)`, - bytes: 0, - sha256: null, - consent: "gbrain setup consent (/setup-gbrain)", - }); - } catch (err) { - const msg = `EGRESS_RECEIPT_FAILED: ${(err as Error).message} — ingest refused`; - console.error(`[memory-ingest] ERR: ${msg}`); - failed += prep.prepared.length; - return { - written: 0, - skipped_secret: prep.skippedSecret, - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, - system_error: msg, - }; - } - const importResult = await runGbrainImport(stagingDir, resolveImportTimeoutMs()); - + const importResult = await runGbrainImport(stagingDir, resolveImportTimeoutMs(), sourceId); const stdout = importResult.stdout || ""; const stderr = importResult.stderr || ""; - const importJson = parseImportJson(stdout); - - if (importResult.status !== 0) { - // #1611/#1802 C3: on timeout, gbrain may have written - // import-checkpoint.json so the next /sync-gbrain can resume. But an - // INTERNAL timeout (runGbrainImport kills the child and returns here) - // never signals the parent, so the SIGTERM forwarder's preserve branch - // doesn't run — and the finally would otherwise delete the staging dir - // despite a "checkpoint preserved" message. Mirror the forwarder: preserve - // only when gbrain actually checkpointed against this dir; otherwise let - // the finally clean up (nothing to resume) and say so honestly. + const report = parseImportReport(stdout); + // A non-zero exit WITH a --json summary is a partial failure (gbrain + // exits 1 when any file throws): fall through to per-file accounting. + if (importResult.status !== 0 && (report === null || importResult.timedOut)) { + // #1611/#1802 C3: an INTERNAL timeout never signals the parent, so the + // SIGTERM forwarder's preserve branch doesn't run. Preserve only when + // gbrain actually checkpointed against this dir. if (importResult.timedOut) { const mins = Math.round(resolveImportTimeoutMs() / 60000); const checkpointed = stagingDirIsCheckpointed(stagingDir); + if (checkpointed) preserveStaging = true; const msg = checkpointed ? `gbrain import timed out after ${mins}min; checkpoint preserved — re-run ` + `/sync-gbrain to resume (raise GSTACK_INGEST_TIMEOUT_MS for big brains)` : `gbrain import timed out after ${mins}min before writing a checkpoint; ` + `re-run /sync-gbrain to restage (raise GSTACK_INGEST_TIMEOUT_MS for big brains)`; - if (checkpointed) preserveStaging = true; console.error(`[memory-ingest] ${msg}`); - return { - written: 0, - skipped_secret: prep.skippedSecret, - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, - system_error: msg, - }; + return { staged, stagingDir, stageErrors, systemError: msg }; } const tail = (stderr.trim().split("\n").pop() || "").slice(0, 300); const msg = `gbrain import exited ${importResult.status}: ${tail}`; console.error(`[memory-ingest] ERR: ${msg}`); - // We conservatively state-record nothing on a non-zero exit — per-run - // partial progress is invisible to us when the importer crashed. - // sync-failures.jsonl entries may still hold per-file detail. - failed += prep.prepared.length; - return { - written: 0, - skipped_secret: prep.skippedSecret, - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, - system_error: msg, - }; + return { staged, stagingDir, stageErrors, systemError: msg }; } - - if (!args.quiet) { - // Echo gbrain's own progress lines on stderr through so the user sees - // them when running interactively. Already on our stderr from the - // child via `stdio: pipe`, but we explicitly forward for clarity. - process.stderr.write(stderr); - } - - if (importJson === null) { - // gbrain exited 0 but didn't emit a parseable --json line. Treat as - // ERR rather than silently passing zeros through — silent zeros let - // a future gbrain-output regression mask data loss. - const msg = - "gbrain import exited 0 but emitted no parseable --json payload. " + - "Refusing to advance state."; + if (!args.quiet) process.stderr.write(stderr); + if (report === null) { + // Silent zeros would let a future gbrain-output regression mask data loss. + const msg = "gbrain import exited 0 but emitted no parseable --json payload. Refusing to advance state."; console.error(`[memory-ingest] ERR: ${msg}`); - failed += prep.prepared.length; - return { - written: 0, - skipped_secret: prep.skippedSecret, - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, - system_error: msg, - }; - } - - // D7: identify which staged files failed to import and exclude them - // from state recording. Source paths get a retry on the next run. - const failedSources = readNewFailures( - syncFailuresPath, - preImportOffset, - staging.stagedPathToSource, - ); - failed += failedSources.size; - - // Reconcile gbrain's own accounting against what we staged. Without this, - // a batch that gbrain never SAW is indistinguishable from a batch that - // succeeded: readNewFailures() only reports PER-FILE failures, so when - // `gbrain import` collects zero files it writes nothing to - // sync-failures.jsonl, failedSources is empty, and every prepared file - // gets state-recorded as ingested. The pass then reports "N written" - // while the brain gained nothing — and because state now says "done", - // no future run retries. Silent, permanent data loss. - // - // Observed cause: `gbrain import` honours .gitignore, and - // `gstack-artifacts-init` writes `.gitignore = "*"` into $GSTACK_HOME. - // makeStagingDir() stages under $GSTACK_HOME, so on any machine that has - // run artifacts-init, collect_files returns 0 for every batch. - // - // `skipped` counts content_hash no-ops, which ARE successful landings. - const expectedLandings = (args.scanSecrets || enforceResumePolicy ? staging.written : prep.prepared.length) - failedSources.size; - const accountedLandings = - (importJson.imported ?? 0) + (importJson.skipped ?? 0); - if (accountedLandings < expectedLandings) { - const collected = - importJson.total_files !== undefined - ? ` gbrain collected ${importJson.total_files} file(s) from the staging dir.` - : ""; - const msg = - `gbrain import accounted for ${accountedLandings} of ${expectedLandings} staged page(s) ` + - `(imported=${importJson.imported ?? 0}, unchanged=${importJson.skipped ?? 0}).${collected} ` + - `Refusing to advance state — the unaccounted pages would be marked ingested without ` + - `landing in the brain. If the count is 0, check whether ${stagingDir} is inside a git ` + - `repo that ignores it (gbrain import honours .gitignore).`; - console.error(`[memory-ingest] ERR: ${msg}`); - failed += prep.prepared.length; - return { - written: 0, - skipped_secret: prep.skippedSecret, - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, - system_error: msg, - }; - } - - // Phase 3: state recording. Only files that landed in gbrain get - // their mtime+sha256 stamped. Failed source paths are deliberately - // left un-state'd so the next run re-prepares them and gbrain's - // content_hash dedup short-circuits the import. - const nowIso = new Date().toISOString(); - for (const p of prep.prepared) { - if (failedSources.has(p.source_path)) continue; - try { - if ((args.scanSecrets || enforceResumePolicy) && staging.stagedPathToSource.get(stagedRelPath(p.slug)) !== p.source_path) continue; - if (resuming && (args.scanSecrets || enforceResumePolicy) && - readFileSync(join(stagingDir, stagedRelPath(p.slug)), "utf-8") !== p.rendered_body) continue; - const fingerprint = sourceFingerprintForStamp(p); - if (!fingerprint) continue; - state.sessions[p.source_path] = { - ...fingerprint, - ingested_at: nowIso, - page_slug: p.page_slug, - partial: p.partial, - }; - written++; - if (!args.quiet) { - const tag = p.partial ? " [partial]" : ""; - console.log(`[${written}] ${p.page_slug}${tag}`); - } - } catch (err) { - // statSync can fail if the source file was removed mid-run; skip - // recording but don't fail the whole pass. - console.error( - `[state-record] ${p.source_path}: ${(err as Error).message}`, - ); - } - } - - if (resuming && (args.scanSecrets || enforceResumePolicy)) preserveStaging = written < staging.written || failedSources.size > 0; - - if (!args.quiet) { - console.error( - `[memory-ingest] gbrain import: ${importJson.imported ?? 0} imported, ` + - `${importJson.skipped ?? 0} unchanged, ${importJson.errors ?? 0} failed` + - (failedSources.size > 0 - ? ` (see ~/.gbrain/sync-failures.jsonl for details)` - : ""), - ); - } - // Silent-zero pathology detector (#2144's other half): pages were staged - // but NOTHING imported or skipped-as-unchanged. That shape hid the dead - // ingest for months — it must be loud even under --quiet, because a run - // that indexes nothing is otherwise indistinguishable from a healthy one. - const importedCount = (importJson.imported ?? 0) + (importJson.skipped ?? 0); - if (prep.prepared.length > 0 && importedCount === 0 && (importJson.errors ?? 0) === 0) { - console.error( - `[memory-ingest] WARNING: ${prep.prepared.length} page(s) staged but gbrain collected ZERO ` + - `(no imports, no unchanged-skips, no errors). This is the #2144 silent-zero shape — ` + - `check gbrain's import.collect_files log line and your gbrain version.`, - ); + return { staged, stagingDir, stageErrors, systemError: msg }; } + const sourceToRel = new Map([...staged].map(([rel, p]) => [p.source_path, rel])); + const sourcePathByRel = new Map([...staged].map(([rel, p]) => [rel, p.source_path])); + const ledger = [...readNewFailures(syncFailuresPath, preImportOffset, sourcePathByRel)].map((src) => sourceToRel.get(src)!); + const verdict = classifyImport(report, stderr, ledger, new Set(staged.keys()), pre?.expected); + // The caller decides after stamping (importAndApply). + if (pre?.preserveOnShortfall && !verdict.refuseAll) preserveStaging = true; + return { staged, stagingDir, stageErrors, report, verdict }; } finally { - // #1802 D1: in remote-http mode `stagingDir` is the PERSISTENT transcript - // dir (makePersistentTranscriptDir, under ~/.gstack/transcripts/) that - // gstack-brain-sync push must pick up — it is NOT a `.staging-ingest-*` dir - // and must never be deleted here. The remote-http branch above already - // documents this intent ("Skip the ... cleanupStagingDir paths"), but a - // `finally` runs on its `return`, so the gate has to live here. Gating on - // mode (rather than widening the ownership guard) keeps checkOwnedStagingDir - // strict: it only ever sees `.staging-ingest-*` dirs. - if (!remoteHttpMode && !preserveStaging) cleanupStagingDir(stagingDir); + if (!preserveStaging) cleanupStagingDir(stagingDir); _activeStagingDir = null; } +} - state.last_full_walk = new Date().toISOString(); - state.last_writer = "gstack-memory-ingest"; - saveState(state); +/** + * Remote-http mode: write pages to the persistent transcript dir that + * gstack-brain-sync pushes and record them as `staged`. + */ +function stageForRemoteBrain( + args: CliArgs, + pages: PreparedPage[], + state: IngestState, + base: Omit, + t0: number, +): BulkResult { + const dir = makePersistentTranscriptDir(); + const staging = writeStaged(pages, dir, args.scanSecrets); + if (!args.quiet) { + for (const e of staging.errors.slice(0, 5)) console.error(`[stage-error] ${e.slug}: ${e.error}`); + console.error( + `[memory-ingest] staged ${staging.written} pages → ${dir}; persisting to artifacts pipeline (skipping local gbrain import — remote-http mode)...`, + ); + } + const nowIso = new Date().toISOString(); + let written = 0; + for (const p of pages) { + if (staging.stagedPathToSource.get(stagedRelPath(p.slug)) !== p.source_path) continue; + let fingerprint: SourceFingerprint | null; + try { + fingerprint = sourceFingerprintForStamp(p); + } catch (err) { + console.error(`[state-record] ${p.source_path}: ${(err as Error).message}`); + continue; + } + if (!fingerprint) continue; + setEntry(state, p.source_path, { + ...fingerprint, + page_slug: p.page_slug, + ...(p.partial ? { partial: true } : {}), + status: "staged", + source_id: "remote-http", + content_sha256: p.content_sha256, + ingested_at: nowIso, + }); + written++; + } + state.last_full_walk = nowIso; + state.last_writer = "gstack-memory-ingest (remote-http mode)"; + if (!args.quiet) { + console.error(`[memory-ingest] persisted ${written} pages to ${dir} (brain admin will index on next pull)`); + } + return { ...base, written, failed: staging.errors.length, duration_ms: Date.now() - t0 }; +} +/** + * Resume (#1611): validate the saved stage and map its pages back to their + * sources. With --scan-secrets every saved file is rescanned; with a repo + * policy store every saved page must be a current permitted source. + */ +function verifyResumeDir( + args: CliArgs, + prep: ReturnType, + stagingDir: string, + enforceResumePolicy: boolean, +): PreStaged | { error: string; scannerFailed: boolean } { + const staged = new Map(); + const stagedPagePaths = new Set(); + const strict = args.scanSecrets || enforceResumePolicy; + try { + if (enforceResumePolicy && !prep.policyStoreExists) { + throw new Error("[repo policy] policy store appeared after source preparation"); + } + const eligiblePages = new Map(prep.prepared.map((p) => [stagedRelPath(p.slug), p])); + const pending = [stagingDir]; + while (pending.length > 0) { + const dir = pending.pop()!; + for (const entry of readdirSync(dir, { withFileTypes: true })) { + const path = join(dir, entry.name); + if (entry.isDirectory()) pending.push(path); + else if (entry.isFile()) { + if (path === join(stagingDir, STAGING_MARKER)) continue; + if (args.scanSecrets) { + const scan = secretScanFile(path); + if (!scan.scanned || scan.scanner !== "gitleaks" || scan.findings.length > 0) { + const reason = scan.scanned ? "match" : scan.scanner; + throw new Error(`[secret-scan ${reason}] ${path}`); + } + } + if (entry.name.endsWith(".md")) { + const relPath = relative(stagingDir, path).split("\\").join("/"); + stagedPagePaths.add(relPath); + const page = eligiblePages.get(relPath); + const current = !!page && readFileSync(path, "utf-8") === page.rendered_body; + if (enforceResumePolicy && !current) { + throw new Error(`[repo policy] staged page is not a current permitted source: ${relPath}`); + } + if (current || (!strict && page)) staged.set(relPath, page!); + } else if (enforceResumePolicy) { + throw new Error(`[repo policy] unrecognized staged file: ${path}`); + } + } else if (strict) { + throw new Error(`[${args.scanSecrets ? "secret-scan error" : "repo policy"}] unsupported staging entry: ${path}`); + } + } + } + if (strict && stagedPagePaths.size === 0) { + throw new Error(`[${args.scanSecrets ? "secret-scan error" : "repo policy"}] resumed staging contains no pages`); + } + } catch (err) { + const cause = (err as Error).message; + const scannerFailed = cause.startsWith("[secret-scan"); + const msg = `${cause}; resumed import refused. Staging preserved; ` + + (scannerFailed + ? "repair gitleaks and retry, or rerun without resume to restage." + : "rerun without resume to restage under the current repo policy."); + console.error(`[memory-ingest] ERR: ${msg}`); + return { error: msg, scannerFailed }; + } + const m = basename(stagingDir).match(/-src-([a-z0-9-]+)$/); return { - written, - skipped_secret: prep.skippedSecret, - skipped_dedup: prep.skippedDedup, - skipped_unattributed: prep.skippedUnattributed, - skipped_policy_readonly: prep.skippedPolicyReadonly, - skipped_policy_deny: prep.skippedPolicyDeny, - failed: failed + prep.parseFailed, - duration_ms: Date.now() - t0, - partial_pages: prep.partialPages, + stagingDir, + sourceId: m ? m[1] : null, + staged, + preserveOnShortfall: strict, + // gbrain's count covers every saved page, not only the ones still mapped. + expected: strict ? stagedPagePaths.size : undefined, }; } +/** + * Landing check over imported_unverified entries, this run's first. Present + * with a matching gstack_content_sha256 → ingested. Missing or different → + * refused (re-queued), printed even under --quiet; a page whose source file + * is gone becomes unrecoverable. Lookup errors leave the entry unverified. + * Returns how many entries remain unverified and how many were re-queued. + */ +function verifyImported(state: IngestState, importedThisRun: Set): { unverified: number; requeued: number; requeuedThisRun: number } { + const out = { unverified: 0, requeued: 0, requeuedThisRun: 0 }; + const nowIso = new Date().toISOString(); + const paths = Object.keys(state.sessions) + .filter((p) => state.sessions[p].status === "imported_unverified") + .sort((a, b) => Number(importedThisRun.has(b)) - Number(importedThisRun.has(a))); + if (paths.length === 0) return out; + const lookups: LandingLookups = gbrainLookups((a) => { + const r = spawnGbrain(a, { timeout: 60_000 }); + return { status: r.status, stdout: r.stdout || "", stderr: r.stderr || "" }; + }); + const budget = { gets: DEFAULT_GET_CAP }; + for (const path of paths) { + const entry = state.sessions[path]; + const outcome = checkLanding(entry, lookups, budget); + if (outcome === "unchecked") { + out.unverified++; + continue; + } + if (outcome === "ingested") { + setEntry(state, path, { ...entry, status: "ingested", verified_at: nowIso }); + continue; + } + const why = outcome === "absent" + ? `not found in gbrain source ${entry.source_id} after import` + : "gbrain holds different content than the staged page"; + if (!existsSync(path)) { + setEntry(state, path, { ...entry, status: "unrecoverable", reason: `${why}; source file is gone` }); + reportPage(`unrecoverable ${entry.page_slug}: ${why}, and ${path} no longer exists`); + continue; + } + setEntry(state, path, { ...entry, status: "refused", reason: why }); + out.requeued++; + if (importedThisRun.has(path)) out.requeuedThisRun++; + reportPage(`re-queued ${entry.page_slug}: ${why} (retried next run)`); + } + return out; +} + // ── Output formatting ────────────────────────────────────────────────────── function formatBytes(n: number): string { @@ -2675,6 +2719,10 @@ function printBulkResult(r: BulkResult, args: CliArgs): void { console.log(` skipped (policy deny): ${r.skipped_policy_deny} (change with: gstack-gbrain-repo-policy set read-write)`); } console.log(` failed: ${r.failed}`); + if (r.refused) console.log(` refused (retry next run): ${r.refused}`); + if (r.unverified) console.log(` imported, not yet verified: ${r.unverified} (checked on later runs)`); + if (r.quarantined) console.log(` quarantined: ${r.quarantined} (retried when the source changes)`); + if (r.kept_local) console.log(` kept local: ${r.kept_local} (see messages above)`); console.log(` duration: ${(r.duration_ms / 1000).toFixed(1)}s`); if (args.benchmark) { const pps = r.duration_ms > 0 ? (r.written * 1000) / r.duration_ms : 0; @@ -2687,10 +2735,14 @@ function printBulkResult(r: BulkResult, args: CliArgs): void { async function main(): Promise { const args = parseArgs(); - // Engine tier detection — informational; routing happens in gbrain server-side. - const engine = detectEngineTier(); - if (!args.quiet) { - console.error(`[engine] ${engine.engine}${engine.engine === "supabase" ? ` (${engine.supabase_url || "configured"})` : ""}`); + // Recording a pending reconcile never calls gbrain (setup and upgrade + // migrations use it), so it runs before engine detection. + if (!args.requestReconcile) { + // Engine tier detection — informational; routing happens in gbrain server-side. + const engine = detectEngineTier(); + if (!args.quiet) { + console.error(`[engine] ${engine.engine}${engine.engine === "supabase" ? ` (${engine.supabase_url || "configured"})` : ""}`); + } } if (args.mode === "probe") { @@ -2699,23 +2751,96 @@ async function main(): Promise { return; } - if (args.mode === "incremental" && args.quiet) { - // Steady-state fast path: log nothing unless changes happen. - const t0 = Date.now(); - const result = await ingestPass(args); - const dt = Date.now() - t0; - if (result.written > 0 || result.failed > 0) { - console.error(`[memory-ingest] ${result.written} written, ${result.failed} failed in ${dt}ms`); + // One lock for every state writer: ingest, reconcile, the pending flag (A1). + const lock = acquireStateLock(LOCK_PATH); + if (!lock.ok) { + console.error( + `[memory-ingest] ERR: another memory ingest (pid ${lock.holder}) is writing ${STATE_PATH}; not run. ` + + `Fix: wait for it to finish, then re-run /sync-gbrain.`, + ); + process.exit(1); + } + let code: number; + try { + code = await runLocked(args); + } finally { + lock.release(); + } + if (code !== 0) process.exit(code); +} + +async function runLocked(args: CliArgs): Promise { + const state = loadState(); + const save = (): boolean => { + try { + saveState(state); + return true; + } catch (err) { + console.error(`[memory-ingest] ERR: ${(err as Error).message}. Nothing this run did is recorded; it will be redone next run.`); + return false; } - // D6: system_error → process-level failure; orchestrator sees ERR. - // Per-file errors do NOT exit non-zero. - if (result.system_error) process.exit(1); - return; + }; + + if (args.requestReconcile) { + state.reconcile = { ...state.reconcile, pending: true }; + if (!save()) return 1; + if (!args.quiet) console.error("[memory-ingest] reconcile pending: the next ingest run re-checks stamped pages (one bounded pass)."); + return 0; } - const result = await ingestPass(args); - printBulkResult(result, args); - if (result.system_error) process.exit(1); + if (args.mode === "reconcile") { + const status = reconcilePass(args, state, args.dryRun); + if (status !== 0 || args.dryRun) return status; + return save() ? 0 : 1; + } + + const t0 = Date.now(); + const result = await ingestPass(args, state); + if (!result.state_untouched) { + // A pending reconcile (set by the upgrade migration or a schema migration) + // gets one bounded pass per ingest run until it completes. + if (!result.system_error && state.reconcile?.pending && !isRemoteHttpMcpMode()) reconcilePass(args, state, false); + if (!save()) return 1; + } + + if (args.mode === "incremental" && args.quiet) { + // Steady-state fast path: log nothing unless changes happen. + if (result.written > 0 || result.failed > 0) { + console.error(`[memory-ingest] ${result.written} written, ${result.failed} failed in ${Date.now() - t0}ms`); + } + } else { + printBulkResult(result, args); + } + // D6: system_error → process-level failure; orchestrator sees ERR. + // Per-file failures do NOT exit non-zero; they are printed and retried. + return result.system_error ? 1 : 0; +} + +/** + * One bounded reconcile pass (A1). The summary prints even under --quiet. + * The state file is backed up before the first change. + */ +function reconcilePass(args: CliArgs, state: IngestState, dryRun: boolean): number { + if (isRemoteHttpMcpMode()) { + console.error("[memory-ingest] reconcile: not run (remote-http brain: pages are staged for the brain admin's pull, not imported here)."); + return 0; + } + if (!gbrainAvailable()) { + console.error("[memory-ingest] reconcile: not run (gbrain CLI not in PATH or missing `import`). Fix: run /setup-gbrain, then gstack-memory-ingest --reconcile."); + return 1; + } + if (!dryRun) backupStateFile(STATE_PATH, `${STATE_PATH}.pre-reconcile.bak`); + const lookups = gbrainLookups((a) => { + const r = spawnGbrain(a, { timeout: 60_000 }); + return { status: r.status, stdout: r.stdout || "", stderr: r.stderr || "" }; + }); + const summary = reconcile(state, lookups, { + limit: args.mode === "reconcile" && args.limit !== null ? args.limit : DEFAULT_RECONCILE_LIMIT, + dryRun, + fileExists: existsSync, + }); + console.error(`[memory-ingest] ${formatReconcileSummary(summary, dryRun)}`); + return 0; } // Guard so the module is import-safe for unit tests (e.g. resolveImportTimeoutMs). diff --git a/docs/gbrain-sync-errors.md b/docs/gbrain-sync-errors.md index 1633cf6a8..a05535609 100644 --- a/docs/gbrain-sync-errors.md +++ b/docs/gbrain-sync-errors.md @@ -255,6 +255,125 @@ and state recorded under the old name maps to the new one. Run --- +## `[memory-ingest] FAILED : (left un-stamped; retried next run)` + +**Problem.** gbrain refused one staged page during `/sync-gbrain`'s memory +stage (for example invalid frontmatter or an oversize page). + +**Cause.** `gbrain import` named the page as a per-file failure (in its +`--json` `failures` list or a `Skipped :` line). Before this release, +such pages were marked ingested and never retried. + +**Fix.** Nothing is lost: the page is not marked ingested and is retried on +every run. If the same page keeps failing, read the error, fix the source +file it came from, then run `/sync-gbrain`. + +--- + +## `[memory-ingest] re-queued : not found in gbrain source after import (retried next run)` + +**Problem.** gbrain said a page was imported (or unchanged), but the landing +check could not find it in the source it was imported into, or found +different content. + +**Cause.** Usually a managed import that was accepted but is still +publishing (`Pending:` in gbrain's output), or a page gbrain skipped without +naming it. + +**Fix.** Nothing to do; the next `/sync-gbrain` imports it again. If it +repeats on every run, upgrade gbrain (`gstack-gbrain-install`). + +--- + +## `[memory-ingest] ERR: gbrain reported N failure(s) it did not attribute to a staged page ... Refusing to advance state.` + +**Problem.** The memory stage failed and marked nothing ingested. + +**Cause.** gbrain counted failures it did not name (it stops printing them +after five of the same kind), so gstack cannot tell which pages landed. + +**Fix.** Re-run `/sync-gbrain`. After three refused runs the batch is split +in halves automatically to find the page gbrain cannot import; that page is +quarantined (next entry) and the rest import. + +--- + +## `[memory-ingest] quarantined (): gbrain refuses every batch that contains it (...)` + +**Problem.** One page kept making gbrain refuse the whole batch. + +**Cause.** After three refused batches, gstack imported the batch in halves +until it isolated the page. Every other page was imported. + +**Fix.** The page is retried automatically when its source file changes. +To retry now, fix or remove the source file named in the message, then run +`/sync-gbrain`. + +--- + +## `[memory-ingest] reconcile: checked N, present M, re-queued K, not yet checked L (run again to continue)` + +**Problem.** Not an error. This is the one-time catch-up after upgrading: +gstack re-checks transcripts it had marked ingested and re-queues the ones +missing from the brain, so transcripts lost by older versions import again. + +**Cause.** Older versions could mark a page ingested that gbrain had +skipped. The upgrade records that a reconcile is pending; each ingest run +then checks a bounded batch (200 entries) until done. + +**Fix.** Nothing to do; each `/sync-gbrain` continues where the last one +stopped. To run it now or check progress: +```bash +gstack-memory-ingest --reconcile --dry-run # report only +gstack-memory-ingest --reconcile # re-check the next 200 (--limit N) +``` +"not yet checked" counts pages whose lookup failed (gbrain busy or +unreachable); they are never treated as missing. "unrecoverable" counts pages +missing from the brain whose transcript file no longer exists. The state file +is backed up to `~/.gstack/.transcript-ingest-state.json.pre-reconcile.bak` +before each pass. + +--- + +## `[memory-ingest] reconcile: not run (...)` + +**Problem.** A reconcile pass was requested but could not run. + +**Cause.** gbrain is not installed or lacks `import`, or the brain is a +remote-HTTP brain (its pages are staged for the brain admin's pull, so there +is nothing to check locally). + +**Fix.** For a missing gbrain, run `/setup-gbrain`, then +`gstack-memory-ingest --reconcile`. For a remote-HTTP brain, nothing to do. + +--- + +## `[memory-ingest] ERR: another memory ingest (pid ) is writing ; not run.` + +**Problem.** The memory stage did nothing this run. + +**Cause.** Another ingest, reconcile or `--request-reconcile` holds the +state lock (`~/.gstack/.transcript-ingest-state.json.lock`). A lock left by a +process that no longer exists is taken over automatically. + +**Fix.** Wait for the other run to finish, then run `/sync-gbrain` again. + +--- + +## `[memory-ingest] ERR: could not save ingest state : .` + +**Problem.** The memory stage failed after importing. + +**Cause.** The state file could not be written (disk full, permissions, or a +directory in its place). The run fails instead of pretending it recorded +progress; pages it imported are re-checked next run, which is cheap because +gbrain skips unchanged content. + +**Fix.** Free disk space or fix permissions on `~/.gstack/`, then run +`/sync-gbrain`. + +--- + ## Nothing is syncing but I expect it to **Not an error, but a common gotcha.** Check in order: diff --git a/gstack-upgrade/migrations/v1.91.17.0.sh b/gstack-upgrade/migrations/v1.91.17.0.sh new file mode 100755 index 000000000..3cf57e85d --- /dev/null +++ b/gstack-upgrade/migrations/v1.91.17.0.sh @@ -0,0 +1,28 @@ +#!/usr/bin/env bash +# Migration: v1.91.17.0 — severe fix wave (memory ingest). +# Placeholder name: the release queue may rename this file at /ship. +# +# Each step is independent and guarded. Idempotent and non-fatal: every path +# exits 0, and no step calls gbrain or the network unless it has work to do. +set -u +_gstack_migration_dir="${BASH_SOURCE[0]//\\//}"; _gstack_migration_dir="${_gstack_migration_dir%/*}" +. "${_gstack_migration_dir}/../../bin/gstack-state-root.sh" 2>/dev/null || { echo "$0: cannot resolve the gstack state root: ${_gstack_migration_dir}/../../bin/gstack-state-root.sh is missing. fix: reinstall with ./setup or /gstack-upgrade (docs/state-root.md)" >&2; exit 1; } +gstack_state_root_select; GH="$_gstack_sr_root" + +# Step 1 (A1, #2778): older versions could mark a transcript ingested that +# gbrain had skipped. Record that a reconcile pass is pending; the next memory +# ingest re-checks stamped pages in bounded batches and re-queues the missing +# ones. This only writes a flag under the ingest's state lock; it never calls +# gbrain. +if [ -f "$GH/.transcript-ingest-state.json" ]; then + INGEST="${_gstack_migration_dir}/../../bin/gstack-memory-ingest.ts" + if command -v bun >/dev/null 2>&1 && [ -f "$INGEST" ]; then + if bun "$INGEST" --request-reconcile --quiet; then + echo "memory ingest: reconcile pending; the next /sync-gbrain re-checks transcripts already marked ingested." + else + echo "memory ingest: could not record a pending reconcile now; run: gstack-memory-ingest --reconcile" >&2 + fi + fi +fi + +exit 0 diff --git a/lib/memory-ingest-landing.ts b/lib/memory-ingest-landing.ts new file mode 100644 index 000000000..210f0e7f7 --- /dev/null +++ b/lib/memory-ingest-landing.ts @@ -0,0 +1,512 @@ +/** + * Memory ingest landing contract (A1, #2778): write, verify, then stamp. + * + * `gbrain import` exits 0 while skipping pages (returned per-file failures, + * malformed filenames, managed-import `Pending:` receipts), and its failure + * ledger is written only for git import dirs, so a staged page that never + * landed used to be stamped ingested and never retried. This module owns: + * + * - the versioned ingest state and its transition table, + * - parsing `gbrain import --json` plus its stderr into named failures and + * a whole-batch refusal when a failure cannot be named, + * - the landing check: one `gbrain list` per source per run for presence, + * and `gbrain get --json` (bounded per run) for the + * `gstack_content_sha256` frontmatter field gstack writes, + * - `--reconcile`: a bounded, resumable re-check of stamped entries, + * - the single writer lock and fail-loud state saves. + * + * Capability facts (gbrain is installed at latest HEAD, floor 0.20.0): + * `import --json` reports `failures`/`unchanged`/`malformed_skipped` from + * 0.48.5.0 and `source_id` from 0.50.1.0; `list`/`get` honor `--source-id` + * from 0.46.25. Capability is detected by behavior, never by version string. + * A lookup error or a busy PGLite means "not yet checked", never "absent". + */ + +import { closeSync, copyFileSync, existsSync, mkdirSync, openSync, readFileSync, renameSync, rmSync, writeFileSync, writeSync } from "fs"; +import { dirname } from "path"; + +// ── State schema ─────────────────────────────────────────────────────────── + +export const STATE_SCHEMA_VERSION = 2; +/** Source recorded for entries stamped before per-source imports existed. */ +export const LEGACY_SOURCE_ID = "default"; +export const DEFAULT_GET_CAP = 50; +export const DEFAULT_RECONCILE_LIMIT = 200; +export const BISECT_AFTER_REFUSALS = 3; +export const LIST_LIMIT = 1_000_000; + +export type EntryStatus = + | "staged" // remote-http: written to the publishable staging dir; the brain admin indexes it + | "imported_unverified" // import reported it landed; presence/content not yet confirmed + | "ingested" // landing check confirmed it in its recorded source + | "refused" // stamped earlier, then found absent or different in the brain; re-queued for the next run + | "quarantined" // isolated as the page that poisons its batch; retried only when the source changes + | "unrecoverable"; // absent from the brain and its source file is gone + +export interface StateEntry { + mtime_ns: number; + sha256: string; + page_slug: string; + partial?: boolean; + status: EntryStatus; + /** gbrain source the page was imported into ("default" for pre-source entries). */ + source_id: string; + /** Value of the gstack_content_sha256 frontmatter field written into the page. */ + content_sha256?: string; + /** When the page was last imported (or staged). */ + ingested_at?: string; + verified_at?: string; + reason?: string; +} + +export interface IngestState { + schema_version: typeof STATE_SCHEMA_VERSION; + last_writer: string; + last_full_walk?: string; + reconcile?: { pending?: boolean; cursor?: string | null; unchecked_in_sweep?: number }; + /** + * Whole-batch refusals per source path (the bisect trigger). A page that did + * not land is never stamped, so the count lives outside `sessions`. + */ + batch_refusals?: Record; + /** Transcript sources registered by gstack (A4), cached so registration runs once. */ + sources?: Record; + sessions: Record; +} + +/** + * Allowed transitions. "new" is the absence of an entry. Every write goes + * through setEntry(), which rejects anything not listed here. + */ +export const TRANSITIONS: Record = { + new: ["staged", "imported_unverified", "ingested", "refused", "quarantined"], + staged: ["staged", "imported_unverified", "ingested", "refused", "quarantined"], + imported_unverified: ["imported_unverified", "ingested", "refused", "quarantined", "unrecoverable", "staged"], + ingested: ["ingested", "imported_unverified", "refused", "quarantined", "unrecoverable", "staged"], + refused: ["staged", "imported_unverified", "ingested", "refused", "quarantined"], + quarantined: ["staged", "imported_unverified", "ingested", "refused", "quarantined"], + unrecoverable: ["staged", "imported_unverified", "ingested", "refused", "unrecoverable"], +}; + +export function setEntry(state: IngestState, path: string, next: StateEntry): void { + const from: EntryStatus | "new" = state.sessions[path]?.status ?? "new"; + if (!TRANSITIONS[from].includes(next.status)) { + throw new Error(`ingest state: illegal transition ${from} -> ${next.status} for ${path}`); + } + state.sessions[path] = next; +} + +/** Entries whose page is not considered landed are re-prepared every run. */ +export function isRetryable(entry: StateEntry | undefined): boolean { + return !entry || entry.status === "refused"; +} + +export function emptyState(): IngestState { + return { schema_version: STATE_SCHEMA_VERSION, last_writer: "gstack-memory-ingest", sessions: {} }; +} + +/** + * Migrate a parsed state file. Schema 1 entries were stamped without any + * landing check and without a source, so they become `ingested` in the + * legacy `default` source and a reconcile pass is marked pending. + */ +export function migrateState(raw: unknown): { state: IngestState; migrated: boolean } | null { + const r = raw as { schema_version?: unknown; sessions?: unknown } | null; + if (!r || typeof r !== "object" || typeof r.sessions !== "object" || r.sessions === null) return null; + if (r.schema_version === STATE_SCHEMA_VERSION) return { state: r as IngestState, migrated: false }; + if (r.schema_version !== 1) return null; + const v1 = r as { last_writer?: string; last_full_walk?: string; sessions: Record> }; + const state: IngestState = { + ...emptyState(), + last_full_walk: v1.last_full_walk, + reconcile: { pending: true, cursor: null, unchecked_in_sweep: 0 }, + }; + for (const [path, e] of Object.entries(v1.sessions)) { + if (!e || typeof e !== "object") continue; + state.sessions[path] = { + mtime_ns: Number(e.mtime_ns) || 0, + sha256: String(e.sha256 ?? ""), + page_slug: String(e.page_slug ?? ""), + ...(e.partial ? { partial: true } : {}), + status: "ingested", + source_id: LEGACY_SOURCE_ID, + ...(typeof e.ingested_at === "string" ? { ingested_at: e.ingested_at } : {}), + }; + } + return { state, migrated: true }; +} + +export type LoadedState = + | { kind: "ok"; state: IngestState; migrated: boolean } + | { kind: "absent"; state: IngestState } + | { kind: "unreadable"; state: IngestState; backup: string | null; reason: string }; + +/** Load the state file. An unreadable or unknown file is backed up, never silently reused. */ +export function loadStateFile(path: string): LoadedState { + if (!existsSync(path)) return { kind: "absent", state: emptyState() }; + let raw: string; + try { + raw = readFileSync(path, "utf-8"); + } catch (err) { + return { kind: "unreadable", state: emptyState(), backup: null, reason: (err as Error).message }; + } + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return { kind: "unreadable", state: emptyState(), backup: backupStateFile(path, `${path}.bak`), reason: "corrupt JSON" }; + } + const migrated = migrateState(parsed); + if (!migrated) { + const v = (parsed as { schema_version?: unknown } | null)?.schema_version; + return { kind: "unreadable", state: emptyState(), backup: backupStateFile(path, `${path}.bak`), reason: `unknown schema_version ${String(v)}` }; + } + return { kind: "ok", ...migrated }; +} + +/** Atomic save. Throws on failure: a run whose state could not be saved has failed. */ +export function saveStateFile(path: string, state: IngestState): void { + mkdirSync(dirname(path), { recursive: true }); + const tmp = `${path}.tmp.${process.pid}`; + try { + writeFileSync(tmp, JSON.stringify(state, null, 2), "utf-8"); + renameSync(tmp, path); + } catch (err) { + try { rmSync(tmp, { force: true }); } catch {} + throw new Error(`could not save ingest state ${path}: ${(err as Error).message}`); + } +} + +export function backupStateFile(path: string, dest: string): string | null { + if (!existsSync(path)) return null; + try { + copyFileSync(path, dest); + return dest; + } catch { + return null; + } +} + +// ── Single writer lock ───────────────────────────────────────────────────── + +function pidAlive(pid: number): boolean { + if (!Number.isInteger(pid) || pid <= 0) return false; + try { + process.kill(pid, 0); + return true; + } catch (err) { + return (err as NodeJS.ErrnoException).code === "EPERM"; + } +} + +/** + * One lock for every state writer (ingest, reconcile, the reconcile-pending + * flag). A lock left by a dead process is taken over. + */ +export function acquireStateLock(lockPath: string): { ok: true; release: () => void } | { ok: false; holder: number } { + mkdirSync(dirname(lockPath), { recursive: true }); + for (let attempt = 0; attempt < 2; attempt++) { + try { + const fd = openSync(lockPath, "wx"); + writeSync(fd, `${process.pid}\n`); + closeSync(fd); + return { + ok: true, + release: () => { + try { + if (readFileSync(lockPath, "utf-8").trim() === String(process.pid)) rmSync(lockPath, { force: true }); + } catch {} + }, + }; + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err; + const holder = Number.parseInt(readFileSync(lockPath, "utf-8").trim(), 10); + if (pidAlive(holder) && holder !== process.pid) return { ok: false, holder }; + rmSync(lockPath, { force: true }); + } + } + return { ok: false, holder: -1 }; +} + +// ── gbrain import result ─────────────────────────────────────────────────── + +export interface ImportReport { + status?: string; + imported?: number; + skipped?: number; + errors?: number; + unchanged?: number; + malformed_skipped?: number; + failures?: Array<{ path?: unknown; error?: unknown }>; + source_id?: string; + total_files?: number; +} + +/** The `--json` summary object on the last JSON line, or null when absent. */ +export function parseImportReport(stdout: string): ImportReport | null { + const lines = stdout.split("\n").map((s) => s.trim()).filter(Boolean); + for (let i = lines.length - 1; i >= 0; i--) { + const line = lines[i]; + if (!line.startsWith("{") || !line.endsWith("}")) continue; + try { + const parsed = JSON.parse(line); + if (parsed && typeof parsed === "object" && "imported" in parsed) return parsed as ImportReport; + } catch {} + } + return null; +} + +/** + * Per-file failures gbrain prints on stderr: ` Skipped : ` for a + * failure importFile returns (printed every time), ` Warning: skipped : + * ` for a thrown one (only the first 5 per error class), and + * ` Skipped (malformed filename — rename to import): `. + */ +export function parseStderrFailures(stderr: string): Array<{ path: string; error: string; thrown: boolean }> { + const out: Array<{ path: string; error: string; thrown: boolean }> = []; + for (const raw of stderr.split("\n")) { + const line = raw.replace(/\r$/, ""); + let m = line.match(/^\s+Warning: skipped (\S+): (.*)$/); + if (m) { out.push({ path: m[1], error: m[2], thrown: true }); continue; } + m = line.match(/^\s+Skipped \(malformed filename[^)]*\): (.*)$/); + if (m) { out.push({ path: m[1].trim(), error: "malformed filename", thrown: false }); continue; } + m = line.match(/^\s+Skipped (\S+): (.*)$/); + if (m) out.push({ path: m[1], error: m[2], thrown: false }); + } + return out; +} + +export interface BatchVerdict { + /** Staged rel path → gbrain's error, for every failure gbrain named. */ + named: Map; + /** Set when a failure cannot be attributed to a page: the whole batch is refused. */ + refuseAll?: string; +} + +/** + * Decide which staged pages failed. Named failures come from `--json` + * `failures`, stderr and the sync-failures ledger. Any failure that cannot + * be named (gbrain stops printing thrown failures after 5 per class, or a + * staged page was never collected) refuses the whole batch. + */ +export function classifyImport( + report: ImportReport, + stderr: string, + ledgerPaths: Iterable, + staged: ReadonlySet, + /** Files gbrain should account for when it differs from the mapped pages (a resumed stage). */ + stagedTotal: number = staged.size, +): BatchVerdict { + const named = new Map(); + const unmapped: string[] = []; + const note = (path: string, error: string) => { + if (staged.has(path)) { + if (!named.has(path)) named.set(path, error); + } else if (!unmapped.includes(path)) unmapped.push(path); + }; + const jsonFailures = Array.isArray(report.failures) ? report.failures : null; + for (const f of jsonFailures ?? []) { + if (typeof f?.path === "string") note(f.path, String(f.error ?? "unknown error")); + } + const stderrFailures = parseStderrFailures(stderr); + for (const f of stderrFailures) note(f.path, f.error); + // The ledger is shared by every gbrain process, so only entries naming a + // page of this batch count; it is complete for the run when present. + let ledgerCount = 0; + for (const p of ledgerPaths) { + if (!staged.has(p)) continue; + ledgerCount++; + if (!named.has(p)) named.set(p, "recorded in ~/.gbrain/sync-failures.jsonl"); + } + + const errors = report.errors ?? 0; + // With `failures` (gbrain >= 0.48.5.0) every error is listed there; before + // that, `errors` counted thrown failures, named by "Warning: skipped" lines + // (the first 5 per class) or by the ledger for git import dirs. + const namedErrors = jsonFailures + ? jsonFailures.length + : Math.max(stderrFailures.filter((f) => f.thrown).length, ledgerCount); + const hidden = Math.max(0, errors - namedErrors); + const accounted = (report.imported ?? 0) + (report.skipped ?? 0); + const expected = stagedTotal - named.size; + let refuseAll: string | undefined; + if (accounted < expected) { + refuseAll = + `gbrain import accounted for ${accounted} of ${expected} staged page(s) ` + + `(imported=${report.imported ?? 0}, skipped=${report.skipped ?? 0})` + + (report.total_files !== undefined ? `; gbrain collected ${report.total_files} file(s)` : ""); + } else if (hidden > 0 || unmapped.length > 0) { + refuseAll = + `gbrain reported ${hidden + unmapped.length} failure(s) it did not attribute to a staged page` + + (unmapped.length > 0 ? ` (${unmapped.slice(0, 5).join(", ")})` : ""); + } + return { named, ...(refuseAll ? { refuseAll } : {}) }; +} + +// ── Landing check ────────────────────────────────────────────────────────── + +export type GbrainRun = (args: string[]) => { status: number | null; stdout: string; stderr: string }; + +/** The slug gbrain stores for a staged path (gbrain core/sync.ts slugifyPath). */ +export function gbrainSlug(slug: string): string { + return slug + .replace(/\.mdx?$/i, "") + .split("/") + .map((seg) => + seg + .normalize("NFD") + .replace(/[\u0300-\u036f]/g, "") + .normalize("NFC") + .toLowerCase() + .replace(/[^\p{L}\p{N}\p{M}.\s_-]/gu, "") + .replace(/\s+/g, "-") + .replace(/-+/g, "-") + .replace(/^-|-$/g, ""), + ) + .filter(Boolean) + .join("/"); +} + +export interface LandingLookups { + /** Presence per source: slug set, or null when the lookup failed (not yet checked). */ + list(sourceId: string): Set | null; + /** The page's gstack_content_sha256, null when absent from the page, undefined when the lookup failed. */ + contentSha(sourceId: string, slug: string): string | null | undefined; +} + +/** + * gbrain-backed lookups with a per-run list cache. `list --source-id` is + * probed once with an id that cannot exist: a gbrain that ignores the flag + * answers it, and its unscoped listing cannot prove absence from a source. + */ +export function gbrainLookups(run: GbrainRun): LandingLookups { + const cache = new Map | null>(); + let scoped: boolean | null = null; + return { + list(sourceId) { + if (cache.has(sourceId)) return cache.get(sourceId)!; + if (scoped === null) scoped = run(["list", "--source-id", "gstack-capability-probe-0", "--limit", "1"]).status !== 0; + let slugs: Set | null = null; + if (scoped) { + const r = run(["list", "--source-id", sourceId, "--limit", String(LIST_LIMIT), "--sort", "slug"]); + if (r.status === 0 && !r.stdout.trimStart().startsWith("{")) { + slugs = new Set(); + for (const line of r.stdout.split("\n")) { + const slug = line.split("\t")[0]?.trim(); + if (slug && line.includes("\t")) slugs.add(slug); + } + } + } + cache.set(sourceId, slugs); + return slugs; + }, + contentSha(sourceId, slug) { + const r = run(["get", slug, "--source-id", sourceId, "--json"]); + if (r.status !== 0) return undefined; + try { + const page = JSON.parse(r.stdout) as { slug?: string; frontmatter?: Record }; + if (!page || typeof page !== "object" || typeof page.slug !== "string") return undefined; + const v = page.frontmatter?.gstack_content_sha256; + return typeof v === "string" ? v : null; + } catch { + return undefined; + } + }, + }; +} + +export type LandingOutcome = "ingested" | "absent" | "mismatch" | "unchecked"; + +/** + * Check one entry against its recorded source. Presence comes from the + * source's list; content is checked with `get` while `budget.gets` lasts. + * Entries without a content hash (stamped before this release) are + * confirmed by presence alone. + */ +export function checkLanding(entry: StateEntry, lookups: LandingLookups, budget: { gets: number }): LandingOutcome { + const slugs = lookups.list(entry.source_id); + if (!slugs) return "unchecked"; + const slug = gbrainSlug(entry.page_slug); + if (!slugs.has(slug)) return "absent"; + if (!entry.content_sha256) return "ingested"; + if (budget.gets <= 0) return "unchecked"; + budget.gets--; + const sha = lookups.contentSha(entry.source_id, slug); + if (sha === undefined) return "unchecked"; + return sha === entry.content_sha256 ? "ingested" : "mismatch"; +} + +// ── Reconcile ────────────────────────────────────────────────────────────── + +export interface ReconcileSummary { + checked: number; + present: number; + requeued: number; + notChecked: number; + unrecoverable: number; + /** True when this pass finished a full sweep with nothing left unchecked. */ + complete: boolean; +} + +/** + * Re-check up to `limit` stamped entries (resuming after the stored cursor) + * against their recorded source. A page missing from its source is + * re-queued when its transcript still exists, otherwise counted + * unrecoverable; pages are never moved between sources. A dry run computes + * the same summary without changing state. + */ +export function reconcile( + state: IngestState, + lookups: LandingLookups, + opts: { limit: number; dryRun: boolean; fileExists: (path: string) => boolean; getCap?: number }, +): ReconcileSummary { + const rec = state.reconcile ?? {}; + const cursor = rec.cursor ?? null; + const eligible = Object.keys(state.sessions) + .filter((p) => ["ingested", "imported_unverified"].includes(state.sessions[p].status)) + .sort(); + const pending = eligible.filter((p) => cursor === null || p > cursor); + const batch = pending.slice(0, opts.limit); + const budget = { gets: opts.getCap ?? DEFAULT_GET_CAP }; + const s: ReconcileSummary = { checked: 0, present: 0, requeued: 0, notChecked: 0, unrecoverable: 0, complete: false }; + const now = new Date().toISOString(); + for (const path of batch) { + const entry = state.sessions[path]; + const outcome = checkLanding(entry, lookups, budget); + if (outcome === "unchecked") { + s.notChecked++; + continue; + } + s.checked++; + if (outcome === "ingested") { + s.present++; + if (!opts.dryRun && entry.status !== "ingested") setEntry(state, path, { ...entry, status: "ingested", verified_at: now }); + continue; + } + const exists = opts.fileExists(path); + if (exists) s.requeued++; + else s.unrecoverable++; + if (opts.dryRun) continue; + setEntry(state, path, exists + ? { ...entry, status: "refused", reason: outcome === "absent" ? `absent from source ${entry.source_id} (reconcile)` : "content differs from the staged page (reconcile)" } + : { ...entry, status: "unrecoverable", reason: `absent from source ${entry.source_id} and the source file is gone` }); + } + const sweepDone = batch.length === pending.length; + const unchecked = (rec.unchecked_in_sweep ?? 0) + s.notChecked; + s.complete = sweepDone && unchecked === 0; + if (!opts.dryRun) { + state.reconcile = sweepDone + ? { pending: !s.complete, cursor: null, unchecked_in_sweep: 0 } + : { pending: true, cursor: batch[batch.length - 1] ?? cursor, unchecked_in_sweep: unchecked }; + } + return s; +} + +export function formatReconcileSummary(s: ReconcileSummary, dryRun: boolean): string { + return ( + `reconcile${dryRun ? " (dry run)" : ""}: checked ${s.checked}, present ${s.present}, ` + + `re-queued ${s.requeued}, not yet checked ${s.notChecked}` + + (s.unrecoverable > 0 ? `, unrecoverable ${s.unrecoverable} (source file gone)` : "") + + (s.complete ? "" : " (run again to continue)") + ); +} diff --git a/test/fixtures/module-size-ratchet.json b/test/fixtures/module-size-ratchet.json index 0a806e131..b2d5b66ae 100644 --- a/test/fixtures/module-size-ratchet.json +++ b/test/fixtures/module-size-ratchet.json @@ -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" }, diff --git a/test/gstack-memory-ingest.test.ts b/test/gstack-memory-ingest.test.ts index 81270cfc9..4728bf14d 100644 --- a/test/gstack-memory-ingest.test.ts +++ b/test/gstack-memory-ingest.test.ts @@ -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 }); }); diff --git a/test/gstack-upgrade-migration-v1_91_17_0.test.ts b/test/gstack-upgrade-migration-v1_91_17_0.test.ts new file mode 100644 index 000000000..1320c5c1c --- /dev/null +++ b/test/gstack-upgrade-migration-v1_91_17_0.test.ts @@ -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); + }); +}); diff --git a/test/helpers/fake-gbrain-brain.ts b/test/helpers/fake-gbrain-brain.ts new file mode 100644 index 000000000..8db931b95 --- /dev/null +++ b/test/helpers/fake-gbrain-brain.ts @@ -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 --json [--source-id]`, `list --source-id --limit`, + * `get --source-id --json`, `sources list --json`, `sources add`. + * + * Behavior switches (env, read by the fake at run time): + * FAKE_FAIL= pages whose path contains it fail by name (HEAD `failures`) + * FAKE_PENDING= accepted as a managed-import Pending receipt: counted, never stored + * FAKE_POISON= 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 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 }; +} diff --git a/test/memory-ingest-landing.test.ts b/test/memory-ingest-landing.test.ts new file mode 100644 index 000000000..a1ca42002 --- /dev/null +++ b/test/memory-ingest-landing.test.ts @@ -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 => ({ + mtime_ns: 1, sha256: "s", page_slug: "ceo-plans/demo/2026-10-03-a", status: "ingested", source_id: "default", ...over, +}); + +function lookupsFrom(pages: Record> | 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 = {}; + let brain: ReturnType; + + 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 = {}) { + 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 => 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([]); + }); +}); diff --git a/test/memory-pipeline.test.ts b/test/memory-pipeline.test.ts index eed2f2eec..4ddf7e422 100644 --- a/test/memory-pipeline.test.ts +++ b/test/memory-pipeline.test.ts @@ -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"); diff --git a/test/regression-1611-gbrain-sync-resume.test.ts b/test/regression-1611-gbrain-sync-resume.test.ts index 1f4a06a89..f65454597 100644 --- a/test/regression-1611-gbrain-sync-resume.test.ts +++ b/test/regression-1611-gbrain-sync-resume.test.ts @@ -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\)/); }); });