feat(linkedin-studio): SB-S1 ingest IO — writePublished/ingestText/scanInbox/listPublished [skip-docs]

Idempotent, collision-safe, create-on-demand IO under the brain dataRoot:
- writePublished: identical body → no-op; differing body at same id → disambiguated
  <id>-N.md (never clobber, never silently drop a gold record, B2 defence-in-depth).
- ingestText: provenance always published; published_date defaults to captured_at.
- scanInbox: top-level *.md only (skips .DS_Store/dotfiles/non-md), non-destructive,
  empty/absent inbox = clean no-op, re-scan re-skips via dedup.
- listPublished: parses each file in try/catch (malformed → skipped count, never
  crashes), surfaces provenance so ai-draft leakage is eyeball-visible.
9 temp-dir IO tests. brain 48→57 tests, tsc clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RigJBiRFNtFZKCz21qNbQ4

scripts/brain/src/ingest.ts

@@ -17,6 +17,10 @@
  * 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. */
@@ -93,3 +97,148 @@ export function parsePublishedRecord(text: string): PublishedRecord {
     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 `<id>.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 `<id>-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 };
+}