feat(linkedin-studio): RE-R2a — item→store capture bridge + publishedAt persistence (schema v1→v2, lossless migrate) [skip-docs]

Closes the research-engine capture loop RE-R1 deferred:
- itemToInput(item, capturedAt): pure envelope→TrendInput bridge in item.ts —
  injects capturedAt, carries publishedAt verbatim; no id, no re-validate
- publishedAt persisted: TrendRecord/TrendInput gain it; addTrend conditional-spread,
  first-sight kept on re-capture (no back-fill). SCHEMA_VERSION 1→2 with a lossless
  forward migrate-on-load: Math.max(onDisk, current) + numeric-typeof coercion
  (string/NaN/absent → current; non-array trends coercion preserved verbatim)
- `capture` CLI: stdin raw item|batch → normalize → bridge → addTrend → saveStore once;
  tally {added,duplicates,merged,errors} from AddResult; content-invalid → errors[],
  exit 2 only on bad stdin; --json summary
- wiring: trend-spotter.md Step 4.5 N×`add` → one normalizing `capture` batch; README
  add/capture framing corrected; test-runner Section 16h (capture wiring, unconditional)
  + floors bumped (trends 62→79, ASSERT 87→90)

TDD: 17 new tests (12 genuinely-RED logic-RED + 5 regression guards), tsc clean,
gate 105/0/0. No version bump (additive, v0.5.2 dev).

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

scripts/trends/src/cli.ts

@@ -9,15 +9,20 @@
  *   node --import tsx src/cli.ts status [--store <path>] [--json]
  *   echo '<raw item|batch>'  | node --import tsx src/cli.ts normalize
  *   echo '<scored candidates>' | node --import tsx src/cli.ts score [--mode kortform|long-form] [--threshold N]
+ *   echo '<raw item|batch>'  | node --import tsx src/cli.ts capture [--store <path>] [--json]
  *
- * The capture agent (research-engine) calls `add` to fold a freshly-polled trend
- * into the store, and `query`/`list` to reason over accumulated history. The
+ * The capture agent (research-engine) folds freshly-polled trends into the store via
+ * `capture` (the normalizing batch path: stdin → normalizeItem(s) → itemToInput →
+ * addTrend), and reasons over accumulated history via `query`/`list`. `add` is the
+ * MANUAL single-trend path (raw flags, no normalization, publish-date-free). The
  * polling + relevance-scoring itself lives upstream; this is the deterministic store.
  *
- * `normalize` + `score` (RE-R1) are the deterministic research-engine seam: both read
- * their JSON PAYLOAD FROM STDIN (so they do not overload `--json`, which stays an
- * output toggle) and print JSON to stdout. `normalize` validates raw items into the
- * canonical envelope; `score` triages scored candidates (composite/band/threshold).
+ * `normalize` + `score` (RE-R1) and `capture` (RE-R2a) are the deterministic
+ * research-engine seam: all read their JSON PAYLOAD FROM STDIN (so they do not overload
+ * `--json`, which stays an output toggle). `normalize` validates raw items into the
+ * canonical envelope; `score` triages scored candidates (composite/band/threshold);
+ * `capture` normalizes + folds each valid item into the store (persisting `publishedAt`),
+ * reporting content-invalid items in the summary `errors[]`, never via the exit code.
  *
  * Exit code: 0 on success, 2 on usage error (incl. unparseable stdin / bad flag).
  */
@@ -33,7 +38,7 @@ import {
   queryByTopic,
   saveStore,
 } from "./store.js";
-import { normalizeItem, normalizeItems } from "./item.js";
+import { normalizeItem, normalizeItems, itemToInput } from "./item.js";
 import { triage } from "./score.js";
 import type { ScoreMode } from "./score.js";
 
@@ -72,7 +77,8 @@ function usage(msg: string): never {
       "  list [--since <YYYY-MM-DD>] [--limit <n>] [--store <path>] [--json]\n" +
       "  status [--store <path>] [--json]\n" +
       "  normalize   < raw-item-or-batch.json\n" +
-      "  score [--mode kortform|long-form] [--threshold N]   < scored-candidates.json",
+      "  score [--mode kortform|long-form] [--threshold N]   < scored-candidates.json\n" +
+      "  capture [--store <path>] [--json]   < raw-item-or-batch.json",
   );
   process.exit(2);
 }
@@ -228,6 +234,34 @@ function main(): void {
     return;
   }
 
+  if (command === "capture") {
+    const payload = readStdinJson(); // exits 2 on empty/unparseable stdin
+    const raw = Array.isArray(payload) ? payload : [payload];
+    const { items, errors } = normalizeItems(raw);
+    const store = loadStore(storePath);
+    // Tally derived from AddResult {added, merged} (no `duplicates` field): a fold is
+    // `added` (new), else `merged` (existing gained topics), else a plain `duplicate`.
+    let added = 0;
+    let merged = 0;
+    let duplicates = 0;
+    for (const item of items) {
+      const res = addTrend(store, itemToInput(item, today()));
+      if (res.added) added++;
+      else if (res.merged) merged++;
+      else duplicates++;
+    }
+    saveStore(storePath, store);
+    if (asJson) {
+      console.log(JSON.stringify({ added, duplicates, merged, errors }, null, 2));
+      return;
+    }
+    console.log(
+      `Captured into ${storePath}: ${added} added, ${merged} merged, ` +
+        `${duplicates} duplicate, ${errors.length} invalid (${store.trends.length} total)`,
+    );
+    return;
+  }
+
   usage(command ? `unknown command: ${command}` : "no command given");
 }
 

scripts/trends/src/item.ts

@@ -17,6 +17,7 @@
  */
 
 import { normalizeField } from "./store.js";
+import type { TrendInput } from "./store.js";
 
 export interface TrendItem {
   /** Capture origin: a research-MCP name ("tavily"), "websearch", or "manual". Stored VERBATIM. */
@@ -117,6 +118,26 @@ export function normalizeItem(raw: unknown): NormalizeResult {
   return { ok: true, item };
 }
 
+/**
+ * Map a validated envelope to a store input (the item→store bridge RE-R1 deferred).
+ * Pure: injects `capturedAt` (the store's "when WE saw it", supplied by the caller —
+ * never derived here) and carries the rest verbatim. Does NOT re-validate (the item is
+ * already validated by normalizeItem) and does NOT derive an `id` (the store owns id via
+ * addTrend→trendId). `publishedAt`/`summary` are carried only when present (key omitted
+ * otherwise), mirroring the store's conditional-spread idiom.
+ */
+export function itemToInput(item: TrendItem, capturedAt: string): TrendInput {
+  return {
+    source: item.source,
+    title: item.title,
+    url: item.url,
+    capturedAt,
+    topics: [...item.topics],
+    ...(item.publishedAt !== undefined ? { publishedAt: item.publishedAt } : {}),
+    ...(item.summary !== undefined ? { summary: item.summary } : {}),
+  };
+}
+
 /** Partition a raw batch into normalized items + per-index errors (never throws). */
 export function normalizeItems(raw: unknown[]): { items: TrendItem[]; errors: ItemError[] } {
   const items: TrendItem[] = [];

scripts/trends/src/store.ts

@@ -28,6 +28,8 @@ export interface TrendInput {
   url: string;
   source: string;
   capturedAt: string;
+  /** The source's own publish date (ISO-8601), if known. Distinct from capturedAt; persisted first-sight, never back-filled. */
+  publishedAt?: string;
   topics: string[];
   summary?: string;
 }
@@ -74,8 +76,15 @@ export function emptyStore(): TrendStore {
 export function loadStore(path: string): TrendStore {
   if (!existsSync(path)) return emptyStore();
   const parsed = JSON.parse(readFileSync(path, "utf8")) as Partial<TrendStore>;
+  // Forward migrate-on-load: stamp to the current version, never downgrade. v1→v2 is
+  // purely additive-optional (an old record is already a valid v2 record that simply
+  // lacks the optional publishedAt), so the migration is the version stamp alone —
+  // records pass through untouched (lossless + idempotent for any well-formed store).
+  // A string / NaN / absent version coerces to the current version (never crashes); the
+  // non-array `trends` coercion below is unchanged and out of the losslessness claim.
+  const onDisk = typeof parsed.schemaVersion === "number" ? parsed.schemaVersion : SCHEMA_VERSION;
   return {
-    schemaVersion: parsed.schemaVersion ?? SCHEMA_VERSION,
+    schemaVersion: Math.max(onDisk, SCHEMA_VERSION),
     trends: Array.isArray(parsed.trends) ? parsed.trends : [],
   };
 }
@@ -122,6 +131,7 @@ export function addTrend(store: TrendStore, input: TrendInput): AddResult {
     url: input.url,
     source: input.source,
     capturedAt: input.capturedAt,
+    ...(input.publishedAt !== undefined ? { publishedAt: input.publishedAt } : {}),
     topics: [...input.topics],
     ...(input.summary !== undefined ? { summary: input.summary } : {}),
   };

scripts/trends/src/types.ts

@@ -34,6 +34,13 @@ export interface TrendRecord {
   source: string;
   /** ISO-8601 date the trend was captured. Supplied by the caller (CLI edge). */
   capturedAt: string;
+  /**
+   * The SOURCE's own publish date (ISO-8601), if the captured item carried one —
+   * distinct from `capturedAt` (when WE saw it): this is when the source published.
+   * First-sight provenance: kept, never overwritten on re-capture. Forward-compat
+   * for B4 freshness ranking. Absent when the source gave no date (key omitted).
+   */
+  publishedAt?: string;
   /** Topic tags for query-by-topic. Unioned across re-captures of the same trend. */
   topics: string[];
   /** Optional short summary of the trend, stored VERBATIM. */
@@ -52,4 +59,4 @@ export interface TrendQueryHit {
   topicOverlap: number;
 }
 
-export const SCHEMA_VERSION = 1;
+export const SCHEMA_VERSION = 2;