mirror of
https://github.com/garrytan/gstack.git
synced 2026-10-31 14:06:50 +01:00
A4: import transcripts into per-repository, non-federated gbrain sources
`gbrain import` ran without --source-id, so transcripts from every project
landed in the shared (often federated) default source.
- Each transcript partition imports with --source-id into its originating
repository's own source (gstack-transcripts-<repo>, built with
constrainSourceId, now exported from lib/gbrain-sources.ts). The source
is registered machine-local and --no-federated before its first import,
and the registration is cached in state. Partitions import one after
another (PGLite is a single writer); a failing source no longer blocks
the others, except a timeout, whose checkpoint must survive.
- If gbrain cannot register the source, that partition stays local and
unstamped, and the run says why (even under --quiet).
- --include-unattributed sends repo-less transcripts only to the
machine-local gstack-transcripts-unattributed source, keeps them on the
machine when the brain is remote (Postgres, Supabase or remote-HTTP),
and never writes them to the publishable ~/.gstack/transcripts staging.
- A page's recorded source is sticky: pages stamped before this release
("default") keep importing there, so no source gains a duplicate.
- The staging dir name carries -src-<id> so a resumed import keeps its
source.
Test fakes that recorded only the last import now record every import of
one run, and answer `gbrain sources`; remote-http scan tests use an
attributed transcript, since unattributed ones no longer stage there.
Co-authored-by: Sinabina <sinabina@Sinabinas-MacBook-Pro-4.local>
This commit is contained in:
1 parent
97d4e58dcc
commit
d02c8effb7
7 files changed
+400
-62
No files matched your search
@@ -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-<hash>", not "${prefix}-kill-270c0001-<hash>".
|
||||
*/
|
||||
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 {
|
||||
|
||||
+122
-9
@@ -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<BulkResult
|
||||
skipped_policy_readonly: prep.skippedPolicyReadonly,
|
||||
skipped_policy_deny: prep.skippedPolicyDeny,
|
||||
partial_pages: prep.partialPages,
|
||||
kept_local: 0,
|
||||
};
|
||||
|
||||
if (remoteHttpMode) return stageForRemoteBrain(args, prep.prepared, state, base, t0);
|
||||
// A4: unattributed transcripts stay on this machine when the brain is
|
||||
// remote, and never enter the publishable remote-http staging.
|
||||
const remoteBrain = brainIsRemote(remoteHttpMode);
|
||||
const keptLocal = remoteBrain ? prep.prepared.filter(isUnattributed) : [];
|
||||
if (keptLocal.length > 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<BulkResult
|
||||
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)`,
|
||||
payloadClass: `transcript-pages count=${resumed ? resumed.staged.size : toImport.length} (sent by gbrain subprocess)`,
|
||||
bytes: 0,
|
||||
sha256: null,
|
||||
consent: "gbrain setup consent (/setup-gbrain)",
|
||||
@@ -2211,9 +2226,23 @@ async function ingestPass(args: CliArgs, state: IngestState): Promise<BulkResult
|
||||
// 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)) {
|
||||
// A4: one partition per gbrain source; a transcript source is registered
|
||||
// (non-federated) before its first import, or its pages stay local.
|
||||
const listed: { ids: Set<string> | 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<string>;
|
||||
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<string | null, PreparedPage[]>();
|
||||
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<string> | 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<string, PreparedPage>;
|
||||
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}`;
|
||||
|
||||
@@ -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 <repo> on this machine: could not register gbrain source <id> (...)`
|
||||
|
||||
**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:
|
||||
|
||||
@@ -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-<hash>", not "${prefix}-kill-270c0001-<hash>".
|
||||
*/
|
||||
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}`;
|
||||
}
|
||||
@@ -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 ;;
|
||||
|
||||
@@ -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
|
||||
`;
|
||||
|
||||
@@ -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<string, string> = {};
|
||||
let brain: ReturnType<typeof installFakeBrain>;
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user