diff --git a/bin/gstack-gbrain-sync.ts b/bin/gstack-gbrain-sync.ts index 5f9b65786..04cf99ef2 100644 --- a/bin/gstack-gbrain-sync.ts +++ b/bin/gstack-gbrain-sync.ts @@ -37,7 +37,7 @@ import { createHash } from "crypto"; import "../lib/conductor-env-shim"; import { detectEngineTier, withErrorContext, canonicalizeRemote } from "../lib/gstack-memory-helpers"; -import { ensureSourceRegistered, sourcePageCount, parseSourcesList, cycleCompleted, type CycleStatus } from "../lib/gbrain-sources"; +import { constrainSourceId, ensureSourceRegistered, sourcePageCount, parseSourcesList, cycleCompleted, type CycleStatus } from "../lib/gbrain-sources"; import { detectAutopilot, decideSourceRemove, decideCodeSync } from "../lib/gbrain-guards"; import { writeReceipt } from "../lib/egress-receipt"; import { dbUnreachableReason, localEngineStatus, localEngineStatusDetail, type LocalEngineStatus } from "../lib/gbrain-local-status"; @@ -814,49 +814,6 @@ export function removeOrphanedSource(oldId: string, env?: NodeJS.ProcessEnv): bo return safeSourcesRemove(oldId, env).removed; } -/** - * Build a gbrain-valid source id (1-32 lowercase alnum + interior hyphens). Sanitizes - * `raw`, prefixes with `prefix`, and falls back to a hashed-tail form when total length - * would exceed 32 chars. - * - * Truncation cuts on hyphen boundaries (whole-word units) from the right, never - * mid-word. Inputs like "drummerms-av-sow-wiz-skill-270c0001" produce - * "${prefix}-270c0001-", not "${prefix}-kill-270c0001-". - */ -function constrainSourceId(prefix: string, raw: string): string { - const MAX = 32; - const slug = raw.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-+|-+$/g, ""); - // Empty slug after sanitize (e.g. raw was all non-alnum like "___") would - // produce "${prefix}-" which fails gbrain's validator on the trailing - // hyphen. Fall back to a deterministic hash of the original input so the - // result is stable across runs of the same repo. - if (!slug) { - const hash = createHash("sha1").update(raw || "_empty").digest("hex").slice(0, 6); - return `${prefix}-${hash}`; - } - const full = `${prefix}-${slug}`; - if (full.length <= MAX) return full; - const hash = createHash("sha1").update(slug).digest("hex").slice(0, 6); - // Total budget: prefix + "-" + tail + "-" + hash - const tailBudget = MAX - prefix.length - 2 - hash.length; - if (tailBudget < 1) return `${prefix}-${hash}`; - // Cut on hyphen boundaries instead of mid-word. Walk tokens from the right, - // accumulating until adding the next token would exceed tailBudget. This - // preserves readable suffixes (pathhash, repo name) and avoids embarrassing - // mid-word artifacts like "skill" → "kill". - const tokens = slug.split("-").filter(Boolean); - const kept: string[] = []; - let len = 0; - for (let i = tokens.length - 1; i >= 0; i--) { - const add = kept.length === 0 ? tokens[i].length : tokens[i].length + 1; - if (len + add > tailBudget) break; - kept.unshift(tokens[i]); - len += add; - } - const tail = kept.join("-"); - return tail ? `${prefix}-${tail}-${hash}` : `${prefix}-${hash}`; -} - // ── Lock file (D1) ───────────────────────────────────────────────────────── interface LockInfo { diff --git a/bin/gstack-memory-ingest.ts b/bin/gstack-memory-ingest.ts index d8e731072..fa5feda7c 100644 --- a/bin/gstack-memory-ingest.ts +++ b/bin/gstack-memory-ingest.ts @@ -68,7 +68,8 @@ import { detectEngineTier, withErrorContext, } from "../lib/gstack-memory-helpers"; -import { execGbrainText, spawnGbrain, spawnGbrainAsync } from "../lib/gbrain-exec"; +import { execGbrainText, gbrainConfigDir, spawnGbrain, spawnGbrainAsync } from "../lib/gbrain-exec"; +import { constrainSourceId, parseSourcesList } from "../lib/gbrain-sources"; import { BISECT_AFTER_REFUSALS, DEFAULT_GET_CAP, @@ -2155,9 +2156,23 @@ async function ingestPass(args: CliArgs, state: IngestState): Promise 0) { + reportPage( + `kept ${keptLocal.length} unattributed transcript(s) on this machine: the brain is remote (Postgres or HTTP), ` + + `and unattributed pages go only to the machine-local source ${UNATTRIBUTED_SOURCE_ID}.`, + ); + } + const toImport = keptLocal.length > 0 ? prep.prepared.filter((p) => !isUnattributed(p)) : prep.prepared; + base.kept_local = keptLocal.length; + + if (remoteHttpMode) return stageForRemoteBrain(args, toImport, state, base, t0); const enforceResumePolicy = resuming && hasRepoPolicyStore(); let resumed: PreStaged | undefined; @@ -2186,7 +2201,7 @@ async function ingestPass(args: CliArgs, state: IngestState): Promise | null } = { ids: null }; + for (const batch of resumed ? [{ sourceId: resumed.sourceId, pages: [...resumed.staged.values()] }] : planBatches(toImport, state)) { + if (!resumed && batch.sourceId?.startsWith("gstack-transcripts")) { + const repo = isUnattributed(batch.pages[0]) ? "(no repository)" : batch.pages[0].git_remote!; + const why = ensureTranscriptSource(state, batch.sourceId, repo, listed); + if (why) { + base.kept_local += batch.pages.length; + reportPage(`kept ${batch.pages.length} transcript page(s) for ${repo} on this machine: ${why}. They import once the source can be registered (upgrade gbrain with gstack-gbrain-install).`); + continue; + } + } await importPartition(run, batch.pages, batch.sourceId, resumed); - if (run.systemError) break; + // Other sources still import after one source's import fails; only a + // timeout (whose checkpoint a later import would overwrite) stops the run. + if (run.stopImports) break; } // Landing check: confirm what this run (and earlier runs) imported is in @@ -2258,11 +2287,88 @@ interface RunContext { quarantined: number; importedThisRun: Set; systemError?: string; + /** A timed-out import may have left a gbrain checkpoint; no later import may overwrite it. */ + stopImports?: boolean; } -/** 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 }] : []; +/** + * The gbrain source a page imports into (A4); null = gbrain's own default + * routing, which curated artifacts keep. A page's recorded source is sticky: + * pages stamped before per-repo sources existed ("default") keep going where + * they went, so a source never gains a duplicate of an older page. New + * transcripts go to their originating repository's own transcript source; + * transcripts with no repository go to the machine-local unattributed source. + */ +function targetSource(p: PreparedPage, entry: StateEntry | undefined): string | null { + if (entry && entry.source_id !== "remote-http") return entry.source_id === LEGACY_SOURCE_ID ? null : entry.source_id; + if (p.type !== "transcript") return null; + return isUnattributed(p) ? UNATTRIBUTED_SOURCE_ID : transcriptSourceId(p.git_remote!); +} + +function isUnattributed(p: PreparedPage): boolean { + return p.type === "transcript" && (!p.git_remote || p.git_remote === "_unattributed"); +} + +/** Per-repository transcript source id: gbrain-valid, stable per canonical remote. */ +export function transcriptSourceId(remote: string): string { + return constrainSourceId("gstack-transcripts", remote); +} + +/** + * The brain is remote when gbrain's config names a database URL (Postgres or + * Supabase) or a remote MCP, or the agent host talks to a remote-HTTP brain. + * Unattributed transcripts never leave the machine (A4). + */ +function brainIsRemote(remoteHttpMode: boolean): boolean { + if (remoteHttpMode) return true; + try { + const cfg = JSON.parse(readFileSync(join(gbrainConfigDir(), "config.json"), "utf-8")); + return (typeof cfg?.database_url === "string" && cfg.database_url.trim() !== "" && cfg.engine !== "pglite") || !!cfg?.remote_mcp; + } catch { + return false; + } +} + +/** Pages grouped by the gbrain source they import into, in first-seen order. */ +function planBatches(pages: PreparedPage[], state: IngestState): Array<{ sourceId: string | null; pages: PreparedPage[] }> { + const groups = new Map(); + for (const p of pages) { + const id = targetSource(p, state.sessions[p.source_path]); + if (!groups.has(id)) groups.set(id, []); + groups.get(id)!.push(p); + } + return [...groups].map(([sourceId, group]) => ({ sourceId, pages: group })); +} + +/** + * Register a transcript source before its first import: machine-local and + * never federated. Registration is cached in state so it runs once per + * repository. Returns null when the source is usable, else why it is not. + */ +function ensureTranscriptSource( + state: IngestState, + id: string, + repo: string, + listed: { ids: Set | null }, +): string | null { + if (state.sources?.[id]) return null; + if (!listed.ids) { + const r = spawnGbrain(["sources", "list", "--json"], { timeout: 30_000 }); + let raw: unknown = null; + try { + raw = r.status === 0 ? JSON.parse(r.stdout || "null") : null; + } catch {} + listed.ids = new Set(parseSourcesList(raw).map((row) => row.id).filter((x): x is string => typeof x === "string")); + } + if (!listed.ids.has(id)) { + const add = spawnGbrain(["sources", "add", id, "--no-federated", "--name", `gstack transcripts: ${repo}`], { timeout: 30_000 }); + if (add.status !== 0) { + return `could not register gbrain source ${id} (${(add.stderr || add.stdout || `exit ${add.status}`).trim().split("\n")[0].slice(0, 200)})`; + } + listed.ids.add(id); + } + (state.sources ??= {})[id] = { repo, registered_at: new Date().toISOString() }; + return null; } /** Print even under --quiet: a page that did not land must never be silent (A1). */ @@ -2356,7 +2462,13 @@ async function importAndApply(run: RunContext, pages: PreparedPage[], sourceId: const r = await importBatch(args, pages, sourceId, pre); run.failed += r.stageErrors; if (r.systemError) { + // A source removed outside gstack: forget the cached registration so the + // next run registers it again. + if (sourceId && run.state.sources?.[sourceId] && /source/i.test(r.systemError) && /not found|unknown|does not exist/i.test(r.systemError)) { + delete run.state.sources[sourceId]; + } run.systemError ??= r.systemError; + if (r.timedOut) run.stopImports = true; run.failed += r.staged.size; return { kind: "error" }; } @@ -2409,6 +2521,7 @@ interface BatchRun { staged: Map; stagingDir: string; stageErrors: number; + timedOut?: boolean; report?: ImportReport; verdict?: BatchVerdict; systemError?: string; @@ -2468,7 +2581,7 @@ async function importBatch(args: CliArgs, pages: PreparedPage[], sourceId: strin : `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)`; console.error(`[memory-ingest] ${msg}`); - return { staged, stagingDir, stageErrors, systemError: msg }; + return { staged, stagingDir, stageErrors, systemError: msg, timedOut: true }; } const tail = (stderr.trim().split("\n").pop() || "").slice(0, 300); const msg = `gbrain import exited ${importResult.status}: ${tail}`; diff --git a/docs/gbrain-sync-errors.md b/docs/gbrain-sync-errors.md index a05535609..15206d83a 100644 --- a/docs/gbrain-sync-errors.md +++ b/docs/gbrain-sync-errors.md @@ -374,6 +374,40 @@ gbrain skips unchanged content. --- +## `[memory-ingest] kept N unattributed transcript(s) on this machine: the brain is remote (Postgres or HTTP), ...` + +**Problem.** `--include-unattributed` transcripts (sessions with no git +remote) were not imported. + +**Cause.** Transcripts that cannot be attributed to a repository go only to +the machine-local, never-federated source `gstack-transcripts-unattributed`. +Your brain is remote (a Postgres or Supabase `database_url`, or a remote-HTTP +MCP brain), so sending them would take them off this machine. They are not +marked ingested and are not written to the publishable `~/.gstack/transcripts/` +staging either. + +**Fix.** Nothing to do if that is what you want. To include such a session, +run it from a git repository with an `origin` remote so it is attributed. + +--- + +## `[memory-ingest] kept N transcript page(s) for on this machine: could not register gbrain source (...)` + +**Problem.** Transcripts from one repository were not imported. + +**Cause.** gstack imports each repository's transcripts into that +repository's own gbrain source (`gstack-transcripts-...`, machine-local and +not federated) and registers it before the first import. The installed gbrain +refused `gbrain sources add`, so the pages stay local and unstamped. + +**Fix.** +```bash +gstack-gbrain-install # upgrade gbrain +/sync-gbrain # the pages import on the next run +``` + +--- + ## Nothing is syncing but I expect it to **Not an error, but a common gotcha.** Check in order: diff --git a/lib/gbrain-sources.ts b/lib/gbrain-sources.ts index 831c1f8a6..c7730348e 100644 --- a/lib/gbrain-sources.ts +++ b/lib/gbrain-sources.ts @@ -10,6 +10,7 @@ */ import { execFileSync, spawnSync } from "child_process"; +import { createHash } from "crypto"; import { realpathSync } from "fs"; import { withErrorContext } from "./gstack-memory-helpers"; import { execGbrainJson, gbrainChildCwd, gbrainInvocation } from "./gbrain-exec"; @@ -349,3 +350,46 @@ export function cycleCompleted(sourceId: string, env?: NodeJS.ProcessEnv): Cycle } return "unknown"; } + +/** + * Build a gbrain-valid source id (1-32 lowercase alnum + interior hyphens). Sanitizes + * `raw`, prefixes with `prefix`, and falls back to a hashed-tail form when total length + * would exceed 32 chars. + * + * Truncation cuts on hyphen boundaries (whole-word units) from the right, never + * mid-word. Inputs like "drummerms-av-sow-wiz-skill-270c0001" produce + * "${prefix}-270c0001-", not "${prefix}-kill-270c0001-". + */ +export function constrainSourceId(prefix: string, raw: string): string { + const MAX = 32; + const slug = raw.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-+|-+$/g, ""); + // Empty slug after sanitize (e.g. raw was all non-alnum like "___") would + // produce "${prefix}-" which fails gbrain's validator on the trailing + // hyphen. Fall back to a deterministic hash of the original input so the + // result is stable across runs of the same repo. + if (!slug) { + const hash = createHash("sha1").update(raw || "_empty").digest("hex").slice(0, 6); + return `${prefix}-${hash}`; + } + const full = `${prefix}-${slug}`; + if (full.length <= MAX) return full; + const hash = createHash("sha1").update(slug).digest("hex").slice(0, 6); + // Total budget: prefix + "-" + tail + "-" + hash + const tailBudget = MAX - prefix.length - 2 - hash.length; + if (tailBudget < 1) return `${prefix}-${hash}`; + // Cut on hyphen boundaries instead of mid-word. Walk tokens from the right, + // accumulating until adding the next token would exceed tailBudget. This + // preserves readable suffixes (pathhash, repo name) and avoids embarrassing + // mid-word artifacts like "skill" → "kill". + const tokens = slug.split("-").filter(Boolean); + const kept: string[] = []; + let len = 0; + for (let i = tokens.length - 1; i >= 0; i--) { + const add = kept.length === 0 ? tokens[i].length : tokens[i].length + 1; + if (len + add > tailBudget) break; + kept.unshift(tokens[i]); + len += add; + } + const tail = kept.join("-"); + return tail ? `${prefix}-${tail}-${hash}` : `${prefix}-${hash}`; +} diff --git a/test/gstack-gbrain-sync-transcript-mode.test.ts b/test/gstack-gbrain-sync-transcript-mode.test.ts index dcc016f11..ac09c8142 100644 --- a/test/gstack-gbrain-sync-transcript-mode.test.ts +++ b/test/gstack-gbrain-sync-transcript-mode.test.ts @@ -164,8 +164,11 @@ case "$1" in fi echo '${sources}' ;; import) - ( cd "$2" && find . -name '*.md' -type f | sort ) > "${stagedList}" - n=$(wc -l < "${stagedList}" | tr -d ' ') + # One import per gbrain source (A4): pages of one ingest run accumulate; + # a new run (different parent pid) starts a fresh list. + if [ "$(cat "${stagedList}.ppid" 2>/dev/null)" != "$PPID" ]; then echo "$PPID" > "${stagedList}.ppid"; : > "${stagedList}"; fi + ( cd "$2" && find . -name '*.md' -type f | sort ) >> "${stagedList}" + n=$( cd "$2" && find . -name '*.md' -type f | wc -l | tr -d ' ') echo "{\\"status\\":\\"success\\",\\"imported\\":$n,\\"skipped\\":0,\\"errors\\":0,\\"total_files\\":$n}" ;; *) echo "unexpected gbrain $*" >&2; exit 1 ;; diff --git a/test/gstack-memory-ingest.test.ts b/test/gstack-memory-ingest.test.ts index 4728bf14d..dd9df4862 100644 --- a/test/gstack-memory-ingest.test.ts +++ b/test/gstack-memory-ingest.test.ts @@ -61,7 +61,14 @@ else if (args[0] === 'import') { } } walk(args[1]); - writeFileSync(join(process.env.HOME, 'imported.json'), JSON.stringify(files)); + // One import per gbrain source (A4): pages of one ingest run accumulate; + // a new run (different parent pid) starts a fresh list. + const log = join(process.env.HOME, 'imported.json'); + const owner = join(process.env.HOME, 'imported.ppid'); + let prior = []; + try { if (readFileSync(owner, 'utf8') === String(process.ppid)) prior = JSON.parse(readFileSync(log, 'utf8')); } catch {} + writeFileSync(owner, String(process.ppid)); + writeFileSync(log, JSON.stringify([...prior, ...files])); if (process.env.SNAPSHOT_STAGE) { if (!process.env.SNAPSHOT_STAGE.startsWith(process.env.GSTACK_HOME + '/.staging-ingest-')) process.exit(2); cpSync(args[1], process.env.SNAPSHOT_STAGE, { recursive: true }); @@ -192,6 +199,20 @@ if (process.env.LIMIT_STAGE_WRITES === '1') { return dir; } + // A4: unattributed transcripts never enter the publishable remote-http + // staging, so remote-http cases use a transcript from a repository. + function attributed(path: string): string { + const repo = join(home, "remote-repo"); + if (!existsSync(repo)) { + expect(spawnSync("git", ["init", "-q", repo], { env, cwd: home, timeout: 10000 }).status).toBe(0); + expect(spawnSync("git", ["-C", repo, "remote", "add", "origin", "https://example.com/remote.git"], { env, cwd: home, timeout: 10000 }).status).toBe(0); + } + const records = readFileSync(path, "utf8").trim().split("\n").map((line) => JSON.parse(line)); + records[0].payload.cwd = repo; + writeFileSync(path, records.map((entry) => JSON.stringify(entry)).join("\n") + "\n"); + return path; + } + function appendRecord(): string { return JSON.stringify({ type: "response_item", timestamp: new Date().toISOString(), payload: { type: "message", role: "user", content: [{ type: "input_text", text: "late ordinary update" }], @@ -291,6 +312,9 @@ if (process.env.LIMIT_STAGE_WRITES === '1') { writeFileSync(allowed, records.map((entry) => JSON.stringify(entry)).join("\n") + "\n"); const dir = interruptedStage(); expect(imported()).toHaveLength(2); + // A4: the two pages import into two sources, so the interrupted run + // made one (failed) import per source. + const importsBefore = readFileSync(join(home, "imports"), "utf8"); const policy = spawnSync(join(import.meta.dir, "..", "bin", "gstack-gbrain-repo-policy"), ["set", "_unattributed", tier], { env, cwd: home, encoding: "utf8", timeout: 10000, }); @@ -301,7 +325,7 @@ if (process.env.LIMIT_STAGE_WRITES === '1') { expect(result.stderr).toContain("[repo policy] staged page is not a current permitted source"); expect(result.stderr).toContain("resumed import refused"); expect(imported()).toEqual([]); - expect(readFileSync(join(home, "imports"), "utf8")).toBe("import\n"); + expect(readFileSync(join(home, "imports"), "utf8")).toBe(importsBefore); expect(sessions()).toEqual({}); expect(existsSync(dir)).toBe(true); delete env.GSTACK_INGEST_RESUME_DIR; @@ -401,7 +425,7 @@ if (process.env.LIMIT_STAGE_WRITES === '1') { it("does not stamp an append during the remote-http scan", () => { scanner("clean"); - const path = source(); + const path = attributed(source()); writeFileSync(join(home, ".claude.json"), JSON.stringify({ mcpServers: { gbrain: { type: "http", url: "http://fixture.invalid/mcp" } } })); env.APPEND_DURING_SCAN = path; env.APPEND_RECORD = appendRecord(); @@ -452,7 +476,8 @@ if (process.env.LIMIT_STAGE_WRITES === '1') { (process.platform === "linux" ? it : it.skip)(`keeps OS-limited partial writes out of outgoing pages (remote-http: ${remote})`, () => { scanner("clean"); if (remote) writeFileSync(join(home, ".claude.json"), JSON.stringify({ mcpServers: { gbrain: { type: "http", url: "http://fixture.invalid/mcp" } } })); - const path = source("ordinary conversation ".repeat(300)); + const created = source("ordinary conversation ".repeat(300)); + const path = remote ? attributed(created) : created; env.LIMIT_STAGE_WRITES = "1"; const result = run(["--scan-secrets"]); expect(result.stderr).toContain("EFBIG"); @@ -644,11 +669,27 @@ process.stdout.write(execFileSync(process.execPath, ['-e', 'process.stdout.write expect(sessions()[path]).toBeDefined(); }); + it("never stages unattributed transcripts for a remote-http brain (A4)", () => { + scanner("clean"); + writeFileSync(join(home, ".claude.json"), JSON.stringify({ mcpServers: { gbrain: { type: "http", url: "http://fixture.invalid/mcp" } } })); + const local = source("unattributed ordinary text"); + const shared = attributed(source("attributed ordinary text")); + const result = run([]); + expect(result.status).toBe(0); + expect(result.stderr).toContain("kept 1 unattributed transcript(s) on this machine: the brain is remote"); + expect(sessions()[local]).toBeUndefined(); + expect(sessions()[shared]).toMatchObject({ status: "staged" }); + const outgoing = join(env.GSTACK_HOME, "transcripts"); + const staged = readdirSync(outgoing, { recursive: true }).map(String).filter((f) => f.endsWith(".md")); + expect(staged).toHaveLength(1); + expect(staged[0]).not.toContain("_unattributed"); + }); + it("scans remote-http pages before persistent staging", () => { scanner("clean"); writeFileSync(join(home, ".claude.json"), JSON.stringify({ mcpServers: { gbrain: { type: "http", url: "http://fixture.invalid/mcp" } } })); - const bad = source('UNSAFE="synthetic"'); - const clean = source(); + const bad = attributed(source('UNSAFE="synthetic"')); + const clean = attributed(source()); expect(run(["--scan-secrets"]).status).toBe(0); expect(imported()).toEqual([]); expect(sessions()[bad]).toBeUndefined(); @@ -1220,6 +1261,14 @@ EOF fi exit 0 ;; + sources) + # A4: transcript sources are registered before their first import. + case "\${2:-}" in + list) echo '{"sources":[]}' ;; + add) echo "Source added: \${3:-}" ;; + esac + exit 0 + ;; put|put_page|put-page) # If new ingest code ever regresses to per-file puts, fail loudly so the # test signals a real architectural regression. @@ -1312,7 +1361,7 @@ describe("gstack-memory-ingest writer (gbrain v0.20+ batch `import` interface)", // Verify gbrain was called exactly ONCE with import, not per-file put. const calls = readFileSync(logFile, "utf-8").trim().split("\n").filter(Boolean); expect(calls.length).toBe(1); - expect(calls[0]).toMatch(/^import\s+\/.+\/\.staging-ingest-\d+-\d+$/); + expect(calls[0]).toMatch(/^import\s+\/.+\/\.staging-ingest-\d+-\d+(-src-[a-z0-9-]+)?$/); // Verify args: --no-embed and --json both present. const argDump = readFileSync(argsFile, "utf-8"); @@ -1436,6 +1485,7 @@ case "\${1:-}" in echo '{"status":"success","duration_s":0.1,"imported":1,"skipped":0,"errors":0,"chunks":1,"total_files":1}' fi exit 0 ;; + sources) [ "\${2:-}" = list ] && echo '{"sources":[]}'; exit 0 ;; *) echo "unknown"; exit 2 ;; esac `; @@ -1504,6 +1554,7 @@ case "\${1:-}" in echo '{"status":"success","duration_s":0.1,"imported":1,"skipped":0,"errors":0,"chunks":1,"total_files":1}' fi exit 0 ;; + sources) [ "\${2:-}" = list ] && echo '{"sources":[]}'; exit 0 ;; *) echo "unknown"; exit 2 ;; esac `; @@ -1589,6 +1640,7 @@ case "\${1:-}" in echo '{"status":"success","duration_s":0.1,"imported":1,"skipped":0,"errors":1,"chunks":1,"total_files":2}' fi exit 0 ;; + sources) [ "\${2:-}" = list ] && echo '{"sources":[]}'; exit 0 ;; *) echo "unknown"; exit 2 ;; esac `; diff --git a/test/memory-ingest-sources.test.ts b/test/memory-ingest-sources.test.ts new file mode 100644 index 000000000..c9867fd7a --- /dev/null +++ b/test/memory-ingest-sources.test.ts @@ -0,0 +1,135 @@ +/** + * A4 (#2140): transcripts import into their originating repository's own + * gbrain source, registered machine-local and non-federated before the first + * import; transcripts with no repository never reach the default or a + * federated source, and stay on the machine when the brain is remote. + */ +import { describe, it, expect, beforeEach, afterEach } from "bun:test"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "fs"; +import { tmpdir } from "os"; +import { join } from "path"; +import { spawnSync } from "child_process"; +import { installFakeBrain } from "./helpers/fake-gbrain-brain"; +import { transcriptSourceId } from "../bin/gstack-memory-ingest"; + +const SCRIPT = join(import.meta.dir, "..", "bin", "gstack-memory-ingest.ts"); + +describe("memory ingest: per-repository transcript sources (A4)", () => { + let home = ""; + let gstackHome = ""; + let env: Record = {}; + let brain: ReturnType; + + beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "gstack-tx-sources-")); + gstackHome = join(home, ".gstack"); + mkdirSync(gstackHome, { recursive: true }); + brain = installFakeBrain(home); + env = { + HOME: home, GSTACK_HOME: gstackHome, PATH: `${brain.binDir}:/usr/bin:/bin`, TMPDIR: home, + GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: join(home, ".gitconfig"), + }; + }); + afterEach(() => rmSync(home, { recursive: true, force: true })); + + function repo(name: string): string { + const dir = join(home, "repos", name); + mkdirSync(dir, { recursive: true }); + for (const args of [["init", "-q", dir], ["-C", dir, "remote", "add", "origin", `https://github.com/acme/${name}.git`]]) { + expect(spawnSync("git", args, { env, timeout: 10_000 }).status).toBe(0); + } + return dir; + } + function session(id: string, cwd: string): string { + const dir = join(home, ".claude", "projects", "p"); + mkdirSync(dir, { recursive: true }); + const path = join(dir, `${id}.jsonl`); + const ts = new Date().toISOString(); + writeFileSync(path, + JSON.stringify({ type: "user", message: { role: "user", content: `hello from ${id}` }, timestamp: ts, cwd }) + "\n" + + JSON.stringify({ type: "assistant", message: { role: "assistant", content: "ok" }, timestamp: ts }) + "\n"); + return path; + } + function run(args: string[]) { + const r = spawnSync(process.execPath, [SCRIPT, "--sources", "transcript", ...args], { env, cwd: home, encoding: "utf-8", timeout: 60_000 }); + return { status: r.status, stdout: r.stdout || "", stderr: r.stderr || "" }; + } + const sessions = () => JSON.parse(readFileSync(join(gstackHome, ".transcript-ingest-state.json"), "utf8")).sessions; + const calls = (): string[][] => existsSync(brain.callLog) ? readFileSync(brain.callLog, "utf8").trim().split("\n").map((l) => JSON.parse(l)) : []; + const fakeBrain = () => JSON.parse(readFileSync(brain.brainFile, "utf8")); + + it("a bulk run across three repositories registers three non-federated sources and imports each one sequentially", () => { + const paths = ["alpha", "beta", "gamma"].map((n) => session(`${n}00000000001`, repo(n))); + const r = run(["--bulk", "--quiet"]); + expect(r.status).toBe(0); + const ids = ["alpha", "beta", "gamma"].map((n) => transcriptSourceId(`github.com/acme/${n}`)); + expect(new Set(ids).size).toBe(3); + const log = calls(); + for (const id of ids) { + const add = log.findIndex((c) => c[0] === "sources" && c[1] === "add" && c[2] === id); + const imp = log.findIndex((c) => c[0] === "import" && c.includes("--source-id") && c[c.indexOf("--source-id") + 1] === id); + expect(add).toBeGreaterThan(-1); + expect(log[add]).toContain("--no-federated"); + expect(imp).toBeGreaterThan(add); + expect(fakeBrain().sources[id].federated).toBe(false); + } + expect(log.filter((c) => c[0] === "import")).toHaveLength(3); + expect(log.some((c) => c[0] === "import" && !c.includes("--source-id"))).toBe(false); + const s = sessions(); + paths.forEach((p, i) => expect(s[p]).toMatchObject({ status: "ingested", source_id: ids[i] })); + expect(Object.keys(fakeBrain().pages.default ?? {})).toEqual([]); + // Registration is cached: a second run with new pages adds no source. + session("alpha00000000002", join(home, "repos", "alpha")); + rmSync(brain.callLog); + expect(run(["--incremental", "--quiet"]).status).toBe(0); + expect(calls().some((c) => c[0] === "sources" && c[1] === "add")).toBe(false); + }); + + it("--include-unattributed sends repo-less transcripts only to the machine-local unattributed source", () => { + const loose = session("loose0000001", join(home, "not-a-repo")); + expect(run(["--bulk", "--quiet"]).status).toBe(0); + expect(calls().some((c) => c[0] === "import")).toBe(false); + expect(run(["--bulk", "--quiet", "--include-unattributed"]).status).toBe(0); + expect(sessions()[loose]).toMatchObject({ status: "ingested", source_id: "gstack-transcripts-unattributed" }); + expect(fakeBrain().sources["gstack-transcripts-unattributed"].federated).toBe(false); + expect(Object.keys(fakeBrain().pages.default ?? {})).toEqual([]); + }); + + it("refuses unattributed transcripts for a remote (Postgres) brain while attributed ones import", () => { + mkdirSync(join(home, ".gbrain"), { recursive: true }); + writeFileSync(join(home, ".gbrain", "config.json"), JSON.stringify({ engine: "postgres", database_url: "postgresql://u:p@db.example.com/brain" })); + const loose = session("loose0000001", join(home, "not-a-repo")); + const kept = session("alpha0000001", repo("alpha")); + const r = run(["--bulk", "--quiet", "--include-unattributed"]); + expect(r.status).toBe(0); + expect(r.stderr).toContain("kept 1 unattributed transcript(s) on this machine: the brain is remote"); + expect(sessions()[loose]).toBeUndefined(); + expect(sessions()[kept].status).toBe("ingested"); + expect(fakeBrain().sources["gstack-transcripts-unattributed"]).toBeUndefined(); + }); + + it("a transcript stamped before this release keeps its default source when it changes, so no duplicate appears", () => { + const path = session("legacy000001", repo("alpha")); + writeFileSync(join(gstackHome, ".transcript-ingest-state.json"), JSON.stringify({ + schema_version: 1, last_writer: "gstack-memory-ingest", + sessions: { [path]: { mtime_ns: 1, sha256: "old", ingested_at: "2026-09-01T00:00:00Z", page_slug: "transcripts/claude-code/acme-alpha/old" } }, + })); + expect(run(["--incremental", "--quiet"]).status).toBe(0); + const imports = calls().filter((c) => c[0] === "import"); + expect(imports).toHaveLength(1); + expect(imports[0]).not.toContain("--source-id"); + expect(sessions()[path]).toMatchObject({ status: "ingested", source_id: "default" }); + expect(calls().some((c) => c[0] === "sources" && c[1] === "add")).toBe(false); + }); + + it("when gbrain cannot register sources the partition stays local, unstamped, and the run says why", () => { + const path = session("alpha0000001", repo("alpha")); + const r = spawnSync(process.execPath, [SCRIPT, "--sources", "transcript", "--bulk", "--quiet"], { + env: { ...env, FAKE_NO_SOURCES: "1" }, cwd: home, encoding: "utf-8", timeout: 60_000, + }); + expect(r.status).toBe(0); + expect(r.stderr).toMatch(/kept 1 transcript page\(s\) for github\.com\/acme\/alpha on this machine: could not register gbrain source gstack-transcripts-\S+/); + expect(calls().some((c) => c[0] === "import")).toBe(false); + expect(sessions()[path]).toBeUndefined(); + }); +});