/** * Ingest — the published-post gold signal (SB-S1). * * Captures the user's ACTUAL published posts into `ingest/published/` tagged * `provenance=published`, so the voice/profile-learning surface can learn from * human-published content ONLY (the model-collapse guard — never from ai-draft). * * This module has two halves: * - PURE: `PublishedRecord` + `parsePublishedRecord` / `serializePublishedRecord` * (a file-per-post constrained-header grammar; NO YAML dep, mirroring the brain * profile.md line-grammar idiom). `parse ∘ serialize = identity` (SC2). * - IO: `writePublished` / `ingestText` / `scanInbox` / `listPublished` — idempotent, * collision-safe, create-on-demand under the brain `dataRoot` (SC1/SC3/SC4). * * The record id is `mintContentId(VERBATIM body)` (id.ts) — byte-identity, so two * structurally-different posts never collide and the write path never silently * drops a differing body. */ import { existsSync, mkdirSync, readFileSync, readdirSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { dataRoot } from "./dataRoot.js"; import { mintContentId, normalizeProvenance } from "./id.js"; /** One ingested published post. `provenance` is always `published` in SB-S1. */ export interface PublishedRecord { /** mintContentId(body) — 12 hex, the dedupe key + filename stem. */ id: string; /** Always `published` — the type pins it; parse rejects any other value. */ provenance: "published"; /** YYYY-MM-DD — when the post was published (operator-supplied or = captured_at). */ published_date: string; /** YYYY-MM-DD — when this record was ingested. */ captured_at: string; /** `manual` (default) | a future connector token. */ source: string; /** The verbatim post body — byte-exact, may contain anything (`]`/`|`/newlines/`---`). */ body: string; } /** The one separator string, used identically by serialize and parse. */ const SENTINEL = "\n---\n"; /** * Serialize a record to the file grammar: a fixed 5-line header, the `---` * sentinel, then the verbatim body. No title, no blank-line padding (the body is * captured byte-exact, so any padding would break the round-trip identity). */ export function serializePublishedRecord(rec: PublishedRecord): string { const header = [ `id: ${rec.id}`, `provenance: ${rec.provenance}`, `published_date: ${rec.published_date}`, `captured_at: ${rec.captured_at}`, `source: ${rec.source}`, ].join("\n"); return header + SENTINEL + rec.body; } function headerScalar(header: string, key: string): string { const m = header.match(new RegExp(`^${key}:\\s*(.*?)\\s*$`, "m")); if (!m) throw new Error(`malformed published record: missing "${key}:" header`); return m[1]; } /** * Parse a record back from the file grammar. Splits on the FIRST `---` sentinel * (the header lines are constrained `key: value` pairs that never contain one), so * a body that itself contains `\n---\n` is preserved verbatim. The header scalars * are read only from the header slice — a header-shaped body line cannot leak in. * `provenance` is asserted to be `published`: a `published/` record carrying any * other provenance is a corruption signal and throws (never silently accepted). */ export function parsePublishedRecord(text: string): PublishedRecord { const idx = text.indexOf(SENTINEL); if (idx === -1) throw new Error("malformed published record: no `---` separator"); const header = text.slice(0, idx); const body = text.slice(idx + SENTINEL.length); const id = headerScalar(header, "id"); if (!/^[0-9a-f]{12}$/.test(id)) { throw new Error(`malformed published record: bad id ${JSON.stringify(id)}`); } const provenance = normalizeProvenance(headerScalar(header, "provenance")); if (provenance !== "published") { throw new Error( `corrupt published record: provenance is "${provenance}", expected "published"`, ); } return { id, provenance, published_date: headerScalar(header, "published_date"), captured_at: headerScalar(header, "captured_at"), source: headerScalar(header, "source"), body, }; } // ── IO ─────────────────────────────────────────────────────────────────────── // // All IO resolves through the brain package's own `dataRoot` (LINKEDIN_STUDIO_DATA // overrides the root for tests + installs). Create-on-demand: `ingest/published/` // is mkdir'd as needed, so ingest does NOT depend on a prior `brain init`. const PUBLISHED_SUB = join("ingest", "published"); const INBOX_SUB = join("ingest", "inbox"); /** The body stored at `path`, or `null` if it can't be read/parsed (treat as differing). */ function storedBody(path: string): string | null { try { return parsePublishedRecord(readFileSync(path, "utf8")).body; } catch { return null; } } export interface WriteResult { /** true iff a file was written (false = identical record already present). */ written: boolean; /** The path written or matched. */ path: string; /** true iff an id-collision with a DIFFERING body forced a disambiguated name. */ collision?: boolean; } /** * Write a record idempotently + collision-safely. If `.md` already holds the * SAME body → no-op (`written:false`). If it holds a DIFFERENT body (a 48-bit hash * collision or an unreadable file) → write the next free `-N.md` rather than * clobber or silently skip — so a differing gold record is never lost (B2). */ export function writePublished(rec: PublishedRecord): WriteResult { const dir = dataRoot(PUBLISHED_SUB); mkdirSync(dir, { recursive: true }); const text = serializePublishedRecord(rec); const primary = join(dir, `${rec.id}.md`); if (!existsSync(primary)) { writeFileSync(primary, text, "utf8"); return { written: true, path: primary }; } if (storedBody(primary) === rec.body) return { written: false, path: primary }; for (let n = 2; ; n++) { const cand = join(dir, `${rec.id}-${n}.md`); if (!existsSync(cand)) { writeFileSync(cand, text, "utf8"); return { written: true, path: cand, collision: true }; } if (storedBody(cand) === rec.body) return { written: false, path: cand, collision: true }; } } export interface IngestResult extends WriteResult { record: PublishedRecord; } /** * Mint a record from a verbatim post body and write it. `provenance` is always * `published`; `published_date` defaults to `captured_at` when not supplied. Pure * inputs (the caller supplies `captured_at`), idempotent + collision-safe write. */ export function ingestText(opts: { body: string; captured_at: string; source?: string; published_date?: string; }): IngestResult { const record: PublishedRecord = { id: mintContentId(opts.body), provenance: "published", published_date: opts.published_date ?? opts.captured_at, captured_at: opts.captured_at, source: opts.source ?? "manual", body: opts.body, }; return { record, ...writePublished(record) }; } export interface ScanResult { /** ids of records written this scan. */ processed: string[]; /** ids of records already present (skipped as duplicates). */ skipped: string[]; } /** * Process the inbox drop-zone into `published/`. Reads TOP-LEVEL `*.md` files only, * skipping dotfiles (`.DS_Store` etc.) and non-markdown — no recursion. An empty or * absent inbox is a clean no-op. Non-destructive in SB-S1: processed inbox files are * left in place (a re-scan re-skips via the published dedup). */ export function scanInbox(opts: { captured_at: string; source?: string }): ScanResult { const dir = dataRoot(INBOX_SUB); const processed: string[] = []; const skipped: string[] = []; if (!existsSync(dir)) return { processed, skipped }; for (const entry of readdirSync(dir, { withFileTypes: true })) { if (!entry.isFile() || entry.name.startsWith(".") || !entry.name.endsWith(".md")) continue; const body = readFileSync(join(dir, entry.name), "utf8"); const res = ingestText({ body, captured_at: opts.captured_at, source: opts.source }); (res.written ? processed : skipped).push(res.record.id); } return { processed, skipped }; } export interface PublishedSummary { id: string; provenance: string; published_date: string; firstLine: string; } /** * List the published gold corpus. Each file is parsed inside a try/catch — a * malformed/hand-dropped file is counted as `skipped`, never crashing the command. * Surfaces `provenance` so the operator can eyeball that nothing `ai-draft` leaked. */ export function listPublished(): { records: PublishedSummary[]; skipped: number } { const dir = dataRoot(PUBLISHED_SUB); const records: PublishedSummary[] = []; let skipped = 0; if (!existsSync(dir)) return { records, skipped }; for (const name of readdirSync(dir)) { if (name.startsWith(".") || !name.endsWith(".md")) continue; try { const rec = parsePublishedRecord(readFileSync(join(dir, name), "utf8")); records.push({ id: rec.id, provenance: rec.provenance, published_date: rec.published_date, firstLine: rec.body.split("\n", 1)[0], }); } catch { skipped++; } } records.sort((a, b) => a.published_date.localeCompare(b.published_date)); return { records, skipped }; }