feat(linkedin-studio): SB-S3c — cross-silo id-threading + post→analytics assembler [skip-docs]

Hub-side design: the published record now carries the specifics/trends ids
it was built from (additive, omit-empty → byte-backward-compatible), and a
new pure assembler (scripts/brain/src/assemble.ts + `brain assemble`) joins
post↔analytics by normalized title-prefix + date with honest confidence
tiers (high/low/none). Answers the arc's north-star query: which raw
material actually performs? (specific → post → measured analytics).

All four tributaries untouched (analytics READ-only via inlined raw-JSON,
no package import); profile.md grammar untouched (the fact→post link stays
OUT — C-1). The repeatable --specific/--trend ingest flags collect via a
new collectRepeated helper, leaving parseFlags untouched.

TDD: 19 new brain tests (ingest 4 + publish 3 + assemble 8 + cli 4), all
SC1–SC12. brain 113/113, gate 95/0/0, BRAIN_TESTS_FLOOR 94→113,
ASSERT_BASELINE_FLOOR unchanged at 80. Light-Voyage hardened
(brief-review 5 FIX · plan-critic 1 BLOCK+4 MAJOR+4 MINOR · scope-guardian ALIGNED).

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

scripts/brain/src/assemble.ts

@@ -0,0 +1,156 @@
+/**
+ * SB-S3c — the cross-silo assembler (the payoff).
+ *
+ * Answers the arc's north-star query (`architecture.md:17`): *which raw material
+ * actually performs?* — `specific → post → measured analytics`. The post record
+ * (`ingest/published/<contentId>.md`) already carries the `specifics`/`trends` ids
+ * it was built from (SB-S3c hub-side threading); this module joins each post to
+ * its measured analytics row and surfaces the whole graph.
+ *
+ * PURE core: `assemblePostGraph({records, analytics})` takes already-loaded inputs
+ * and returns the graph — no FS/clock/network. The analytics↔post join is an honest
+ * HEURISTIC, never a guaranteed key: analytics carries no body and no URN (only a
+ * title-prefix + date), so `matchRow` joins by normalized title-prefix + date with
+ * explicit confidence tiers (`high`/`low`/`none`) — a real-CSV `none` is a
+ * normalization-tightening signal, not a proof of no match.
+ *
+ * DECOUPLED: this module treats tributary ids as opaque 12-hex strings and takes a
+ * minimal `AnalyticsRowInput` shape — it never imports the analytics/trends/
+ * specifics packages. The thin read-only `loadAnalyticsRows` IO inlines a raw-JSON
+ * read of the shared data-root (NOT the analytics package's `loadAllPosts`).
+ */
+
+import { existsSync, readFileSync, readdirSync } from "node:fs";
+import { join } from "node:path";
+
+import { dataRoot } from "./dataRoot.js";
+import type { PublishedRecord } from "./ingest.js";
+
+/**
+ * The minimal analytics-row shape the resolver needs, extracted from the raw
+ * `AnalyticsBatch.posts[]` JSON (`analytics/src/models/types.ts`). Note the field
+ * is `publishedDate` (analytics) vs `published_date` (the brain record).
+ */
+export interface AnalyticsRowInput {
+  title: string;
+  publishedDate: string; // YYYY-MM-DD
+  metrics?: { engagementRate?: number } & Record<string, unknown>;
+}
+
+export type MatchConfidence = "high" | "low" | "none";
+
+/** A post's matched analytics: the WHOLE row reference (FIX 4), or none. */
+export interface PostMatch {
+  confidence: MatchConfidence;
+  row?: AnalyticsRowInput;
+}
+
+export interface PostGraphNode {
+  contentId: string;
+  published_date: string;
+  specifics: string[];
+  trends: string[];
+  match: PostMatch;
+}
+
+/**
+ * Minimum normalized-title length to attempt a prefix match. The hook quality-rule
+ * floor is 110 chars; 24 normalized chars (~3–5 words) is the shortest opener
+ * specific enough that a prefix-match is not coincidental, while staying well under
+ * any real hook. Below floor → `none` (an operator can still eyeball).
+ */
+const PREFIX_FLOOR = 24;
+
+/** Brain-local copy of the specifics-bank `normalizeContent` idiom (NOT imported). */
+export function normalize(s: string): string {
+  return s.trim().toLowerCase().replace(/\s+/g, " ");
+}
+
+/** Strip a trailing LinkedIn truncation marker (`…`/`...`) so a `…`-suffixed export title still prefix-matches. */
+function stripTrailingEllipsis(s: string): string {
+  return s.replace(/(?:…|\.{3})\s*$/, "").trimEnd();
+}
+
+/**
+ * Match one analytics row to one published record. Returns the tiered match, or
+ * `null` when the row does not qualify (no prefix / below floor) — STUB until S3c
+ * Step 3.
+ */
+export function matchRow(record: PublishedRecord, row: AnalyticsRowInput): PostMatch | null {
+  const nt = stripTrailingEllipsis(normalize(row.title));
+  if (nt.length < PREFIX_FLOOR) return null; // too short to discriminate → none
+  if (!normalize(record.body).startsWith(nt)) return null; // no prefix → none
+  const confidence: MatchConfidence = record.published_date === row.publishedDate ? "high" : "low";
+  return { confidence, row };
+}
+
+/**
+ * Assemble the post → raw-material → performance graph. Pure (no FS/clock/network).
+ * For each record, the BEST qualifying analytics row: `high` (same date) beats `low`
+ * (different date); within a tier, the longest matched title wins. The analytics rows
+ * are sorted once (publishedDate desc, title asc) so an exact-length tie is stable —
+ * never `readdirSync`-order-dependent.
+ */
+export function assemblePostGraph(args: {
+  records: PublishedRecord[];
+  analytics: AnalyticsRowInput[];
+}): PostGraphNode[] {
+  const analytics = [...args.analytics].sort(
+    (a, b) => b.publishedDate.localeCompare(a.publishedDate) || a.title.localeCompare(b.title),
+  );
+  return args.records.map((record) => {
+    let best: PostMatch | null = null;
+    let bestLen = -1;
+    for (const row of analytics) {
+      const m = matchRow(record, row);
+      if (!m) continue;
+      const len = stripTrailingEllipsis(normalize(row.title)).length;
+      const better =
+        best === null ||
+        (m.confidence === "high" && best.confidence === "low") ||
+        (m.confidence === best.confidence && len > bestLen);
+      if (better) {
+        best = m;
+        bestLen = len;
+      }
+    }
+    return {
+      contentId: record.id,
+      published_date: record.published_date,
+      specifics: record.specifics,
+      trends: record.trends,
+      match: best ?? { confidence: "none" },
+    };
+  });
+}
+
+/**
+ * Read-only loader: inline a raw-JSON read of the analytics batches under the shared
+ * data-root and extract the minimal row shape. STUB until S3c Step 3.
+ *
+ * NOTE (root-skew caveat): resolves via the brain `dataRoot` (`${LINKEDIN_STUDIO_DATA}/
+ * analytics/posts`); the analytics package additionally honours the deprecated
+ * `ANALYTICS_ROOT` override, which this path does NOT — if set to a non-default
+ * path, the join degrades to every-post-`none` (accepted cost of the no-import
+ * decoupling; the M0 default leaves `ANALYTICS_ROOT` unset).
+ */
+export function loadAnalyticsRows(): AnalyticsRowInput[] {
+  const dir = dataRoot(join("analytics", "posts"));
+  if (!existsSync(dir)) return []; // fresh-clone / no imports yet → no rows
+  const rows: AnalyticsRowInput[] = [];
+  for (const name of readdirSync(dir)) {
+    if (!name.endsWith(".json") || name.startsWith(".")) continue;
+    try {
+      const batch = JSON.parse(readFileSync(join(dir, name), "utf8")) as { posts?: unknown[] };
+      for (const p of batch?.posts ?? []) {
+        const row = p as Partial<AnalyticsRowInput>;
+        if (typeof row?.title === "string" && typeof row?.publishedDate === "string") {
+          rows.push({ title: row.title, publishedDate: row.publishedDate, metrics: row.metrics });
+        }
+      }
+    } catch {
+      // skip a malformed/unreadable batch file — never crash (mirrors listPublished)
+    }
+  }
+  return rows;
+}

scripts/brain/src/cli.ts

@@ -25,6 +25,7 @@ import {
   type Candidate,
   type ProfileDiff,
 } from "./consolidate.js";
+import { assemblePostGraph, loadAnalyticsRows } from "./assemble.js";
 import { dataRoot } from "./dataRoot.js";
 import { ingestText, listPublished, parsePublishedRecord, scanInbox } from "./ingest.js";
 import { parseProfile, serializeProfile } from "./profile.js";
@@ -59,6 +60,25 @@ function parseFlags(args: string[]): Record<string, string> {
   return out;
 }
 
+/**
+ * SB-S3c: collect EVERY value of a repeatable `--key <value>` flag into an array
+ * (the single-value `parseFlags` keeps only the last). Scans the raw args directly,
+ * leaving `parseFlags` untouched — so single-value flag behaviour is unchanged.
+ */
+function collectRepeated(args: string[], key: string): string[] {
+  const out: string[] = [];
+  for (let i = 0; i < args.length; i++) {
+    if (args[i] === `--${key}`) {
+      const next = args[i + 1];
+      if (next !== undefined && !next.startsWith("--")) {
+        out.push(next);
+        i++;
+      }
+    }
+  }
+  return out;
+}
+
 function today(): string {
   return new Date().toISOString().slice(0, 10);
 }
@@ -68,9 +88,10 @@ function usage(msg: string): never {
   console.error(
     "usage:\n" +
       "  init\n" +
-      "  ingest --file <path> [--source <s>] [--date <YYYY-MM-DD>]\n" +
+      "  ingest --file <path> [--source <s>] [--date <YYYY-MM-DD>] [--specific <id>]… [--trend <id>]…\n" +
       "  ingest --scan-inbox [--source <s>]\n" +
-      "  published list [--json]",
+      "  published list [--json]\n" +
+      "  assemble",
   );
   process.exit(2);
 }
@@ -89,7 +110,7 @@ function runInit(): void {
   if (created.length === 0) console.log("Already initialised — nothing to do.");
 }
 
-function runIngest(flags: Record<string, string>): void {
+function runIngest(rest: string[], flags: Record<string, string>): void {
   if (flags["scan-inbox"] === "true") {
     const res = scanInbox({ captured_at: today(), source: flags.source });
     console.log(
@@ -100,11 +121,14 @@ function runIngest(flags: Record<string, string>): void {
   const file = flags.file;
   if (!file || file === "true") usage("ingest needs --file <path> or --scan-inbox");
   const body = readFileSync(file, "utf8");
+  // SB-S3c: --specific / --trend are repeatable — read ONLY via collectRepeated.
   const res = ingestText({
     body,
     captured_at: today(),
     source: flags.source,
     published_date: flags.date === "true" ? undefined : flags.date,
+    specifics: collectRepeated(rest, "specific"),
+    trends: collectRepeated(rest, "trend"),
   });
   if (res.written && res.collision) {
     console.log(`Collision (different body, same id) → wrote ${res.path}`);
@@ -129,6 +153,53 @@ function runPublished(rest: string[], flags: Record<string, string>): void {
   }
 }
 
+/**
+ * SB-S3c: read-only cross-silo assembler view — post → raw-material → performance.
+ * Loads published records + analytics rows (inlined raw-JSON, no analytics import),
+ * prints the join newest-first. Writes NOTHING.
+ */
+function runAssemble(_flags: Record<string, string>): void {
+  const pubDir = dataRoot(join("ingest", "published"));
+  const records = existsSync(pubDir)
+    ? readdirSync(pubDir)
+        .filter((f) => f.endsWith(".md") && !f.startsWith("."))
+        .map((f) => {
+          try {
+            return parsePublishedRecord(readFileSync(join(pubDir, f), "utf8"));
+          } catch {
+            return null;
+          }
+        })
+        .filter((r): r is NonNullable<typeof r> => r !== null)
+    : [];
+  const analytics = loadAnalyticsRows();
+  const bodyById = new Map(records.map((r) => [r.id, r.body]));
+  const graph = assemblePostGraph({ records, analytics }).sort((a, b) =>
+    b.published_date.localeCompare(a.published_date),
+  );
+  if (graph.length === 0) {
+    console.log(
+      "No published records to assemble. Ingest posts with `brain ingest --file <p> [--specific <id>] [--trend <id>]`.",
+    );
+    return;
+  }
+  console.log(`Post graph — ${graph.length} record(s); ${analytics.length} analytics row(s):`);
+  for (const node of graph) {
+    const firstLine = (bodyById.get(node.contentId) ?? "").split("\n", 1)[0];
+    console.log(`\n· ${node.contentId} · ${node.published_date} · ${firstLine}`);
+    console.log(`  specifics: ${node.specifics.length ? node.specifics.join(", ") : "—"}`);
+    console.log(`  trends: ${node.trends.length ? node.trends.join(", ") : "—"}`);
+    if (node.match.confidence === "none") {
+      console.log("  analytics: none (no title-prefix+date match)");
+    } else {
+      const eng = node.match.row?.metrics?.engagementRate;
+      console.log(
+        `  analytics: ${node.match.confidence}${eng !== undefined ? ` [eng ${eng}%]` : ""}`,
+      );
+    }
+  }
+}
+
 function renderDiffMd(diff: ProfileDiff): string {
   const lines = ["# Pending profile diff", "", "> Operator-gated. Review, then `brain consolidate --apply --diff brain/pending-diff.json --confirm`.", ""];
   const section = (title: string, items: string[]) => {
@@ -235,9 +306,10 @@ function main(): void {
   const flags = parseFlags(rest);
 
   if (command === "init") return runInit();
-  if (command === "ingest") return runIngest(flags);
+  if (command === "ingest") return runIngest(rest, flags);
   if (command === "published") return runPublished(rest, flags);
   if (command === "consolidate") return runConsolidate(flags);
+  if (command === "assemble") return runAssemble(flags);
 
   usage(command ? `unknown command: ${command}` : "no command given");
 }

scripts/brain/src/id.ts

@@ -3,8 +3,10 @@
  *
  * Pure + deterministic — no filesystem, no clock, no network. The id is keyed on
  * a STABLE SLUG of the label, not the raw label, so editing a fact's *value* (or
- * the label's case/whitespace) never re-mints the id. SB-S3 will thread this id
- * through the tributaries; SB-S0 only establishes mint + shape.
+ * the label's case/whitespace) never re-mints the id. SB-S3c threads this id
+ * through the tributaries (the published record carries the `specifics`/`trends`
+ * ids it was built from; analytics is joined by resolver — see `assemble.ts`);
+ * SB-S0 established mint + shape.
  */
 
 import { createHash } from "node:crypto";

scripts/brain/src/ingest.ts

@@ -35,6 +35,14 @@ export interface PublishedRecord {
   captured_at: string;
   /** `manual` (default) | a future connector token. */
   source: string;
+  /**
+   * SB-S3c — the raw material this post was built from: specifics-bank + trends
+   * ids (12-hex), threaded hub-side so the post↔specific↔trend↔analytics graph is
+   * assemblable. Default `[]`; serialized only when non-empty (omit-empty keeps a
+   * pre-S3c record byte-identical). Validated to 12-hex on parse + at the producer.
+   */
+  specifics: string[];
+  trends: string[];
   /** The verbatim post body — byte-exact, may contain anything (`]`/`|`/newlines/`---`). */
   body: string;
 }
@@ -54,8 +62,36 @@ export function serializePublishedRecord(rec: PublishedRecord): string {
     `published_date: ${rec.published_date}`,
     `captured_at: ${rec.captured_at}`,
     `source: ${rec.source}`,
-  ].join("\n");
-  return header + SENTINEL + rec.body;
+  ];
+  // SB-S3c: the raw-material id lines are appended AFTER `source:` and emitted ONLY
+  // when non-empty — so a pre-S3c record (empty arrays) serializes to the unchanged
+  // 5-line header, byte-identical (SC2). Body-only `mintContentId` is unaffected.
+  if (rec.specifics.length > 0) header.push(`specifics: ${rec.specifics.join(",")}`);
+  if (rec.trends.length > 0) header.push(`trends: ${rec.trends.join(",")}`);
+  return header.join("\n") + SENTINEL + rec.body;
+}
+
+const ID_RE = /^[0-9a-f]{12}$/;
+
+/**
+ * SB-S3c: read an optional comma-separated 12-hex id list from the header slice.
+ * Absent key → `[]` (a NON-throwing reader — NOT `headerScalar`, which throws on a
+ * missing key and would break every pre-S3c record). A present-but-malformed id
+ * throws (never silently dropped).
+ */
+function headerIdList(header: string, key: string): string[] {
+  const m = header.match(new RegExp(`^${key}:\\s*(.*?)\\s*$`, "m"));
+  if (!m) return [];
+  return m[1]
+    .split(",")
+    .map((s) => s.trim())
+    .filter((s) => s !== "")
+    .map((id) => {
+      if (!ID_RE.test(id)) {
+        throw new Error(`malformed published record: bad ${key} id ${JSON.stringify(id)}`);
+      }
+      return id;
+    });
 }
 
 function headerScalar(header: string, key: string): string {
@@ -94,6 +130,8 @@ export function parsePublishedRecord(text: string): PublishedRecord {
     published_date: headerScalar(header, "published_date"),
     captured_at: headerScalar(header, "captured_at"),
     source: headerScalar(header, "source"),
+    specifics: headerIdList(header, "specifics"),
+    trends: headerIdList(header, "trends"),
     body,
   };
 }
@@ -167,13 +205,27 @@ export function ingestText(opts: {
   captured_at: string;
   source?: string;
   published_date?: string;
+  specifics?: string[];
+  trends?: string[];
 }): IngestResult {
+  // SB-S3c producer guard: a malformed raw-material id fails fast HERE (not only on
+  // a later re-parse). Empty/absent arrays skip validation, so every pre-S3c caller
+  // (incl. scanInbox) is unaffected.
+  const validateIds = (ids: string[] | undefined, kind: string): string[] => {
+    const list = ids ?? [];
+    for (const id of list) {
+      if (!ID_RE.test(id)) throw new Error(`ingest: bad ${kind} id ${JSON.stringify(id)}`);
+    }
+    return list;
+  };
   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",
+    specifics: validateIds(opts.specifics, "specifics"),
+    trends: validateIds(opts.trends, "trends"),
     body: opts.body,
   };
   return { record, ...writePublished(record) };