linkedin-studio/scripts/brain/src/consolidate.ts

/**
 * Consolidation engine (SB-S2) — the compounding mechanism.
 *
 * PURE core: `proposeDiff` reads candidate facts + the current two-layer profile
 * and returns a typed DIFF (never mutates, never writes); `applyDiff` produces the
 * next ProfileDoc. The operator-gated CLI is the only thing that writes profile.md.
 *
 * Invariants enforced IN CODE (not just docs):
 *  - provenance-gated: `ai-draft` candidates are rejected outright (model-collapse guard);
 *  - evidence-threshold promotion (dynamic→static at N observations);
 *  - contradiction → keep-both with DISTINCT ids (a genuine contradiction; both views coexist);
 *  - temporal-update → supersede (SB-S3b): an explicit `supersedes` signal retires the old fact
 *    (archival re-mint + status:superseded, replaced in place) and installs the winner at the key-id;
 *  - temporal decay flagging (dynamic ACTIVE facts only; static + superseded facts are decay-exempt).
 *
 * Id model (the no-duplicate-id guarantee): a concept's PRIMARY fact id is
 * `mintEntityId({kind:'observed', key})` (key-only); a conflict ALT fact id is
 * `mintContentId('observed-alt:'+key+'::'+value+'::'+date)` (byte-distinct). Folded
 * `profile-field` static seeds use a different kind, so consolidation never collides
 * with them (they stay immutable in S2). Matching always starts from the candidate's
 * key, so an existing fact's key never needs to be recovered.
 */

import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { dirname } from "node:path";

import { dataRoot } from "./dataRoot.js";
import { mintContentId, mintEntityId } from "./id.js";
import { SCHEMA_VERSION } from "./types.js";
import type { ProfileDoc, ProfileFact, Provenance } from "./types.js";

const OBSERVED_KIND = "observed";

/** A candidate fact the invoking session extracts from the gathered deltas. */
export interface Candidate {
  key: string; // the concept (keyed for the id; single-line)
  value: string; // single-line (no newline/CR — profile grammar)
  provenance: Provenance;
  source: string; // e.g. "published:<id>" | "manual"
  observed_date: string; // YYYY-MM-DD
  // SB-S3b: explicit temporal-update signal — the key of the active prior fact this
  // candidate retires. Absent for ordinary candidates (fully backward-compatible).
  supersedes?: string;
}

/**
 * SB-S3b temporal-update retirement: the active fact at `oldId` (whose value was
 * `oldValue` at propose time) is retired (re-minted to an archival id, status
 * `superseded`) and replaced by `winner` (filed at the canonical key-id). The full
 * winner fact is carried so `applyDiff` stays a pure projector (it has no clock).
 */
export interface SupersedeOp {
  oldId: string;
  oldValue: string;
  winner: ProfileFact;
}

export interface ProfileDiff {
  additions: ProfileFact[]; // new dynamic facts (primary adds + conflict alts)
  evidenceBumps: { id: string; newCount: number; last_seen: string }[];
  promotions: { id: string }[]; // dynamic→static (post-bump count ≥ N)
  conflicts: { primaryId: string; primaryValue: string; altId: string }[];
  staleFlags: { id: string; last_seen: string; daysStale: number }[];
  supersedes: SupersedeOp[]; // SB-S3b temporal-update retirements
}

export interface ConsolidateOpts {
  promoteThreshold?: number; // default 3
  decayDays?: number; // default 90
}

function daysBetween(from: string, to: string): number {
  return Math.floor((Date.parse(to) - Date.parse(from)) / 86400000);
}

function altId(c: Candidate): string {
  return mintContentId(`observed-alt:${c.key}::${c.value}::${c.observed_date}`);
}

/**
 * SB-S3b: archival id for a retired (superseded) fact. Seeded with the fact's
 * PRE-archival id (unique among active facts by the no-duplicate-id invariant), so
 * the re-mint is collision-free and never reuses the canonical key-id the winner takes.
 */
function archivalId(oldId: string, oldValue: string): string {
  return mintContentId(`superseded:${oldId}::${oldValue}`);
}

function newFact(id: string, c: Candidate, today: string): ProfileFact {
  return {
    id,
    value: c.value,
    first_seen: c.observed_date,
    last_seen: today,
    evidence_count: 1,
    provenance: c.provenance,
    status: "active",
  };
}

/**
 * Propose a diff over the current profile from a batch of candidates. Pure — never
 * mutates `current`/`candidates`, never touches the filesystem.
 */
export function proposeDiff(args: {
  current: ProfileDoc;
  candidates: Candidate[];
  today: string;
  opts?: ConsolidateOpts;
}): ProfileDiff {
  const { current, candidates, today, opts } = args;
  const N = opts?.promoteThreshold ?? 3;
  const DECAY = opts?.decayDays ?? 90;

  const byId = new Map<string, ProfileFact>([...current.static, ...current.dynamic].map((f) => [f.id, f]));
  const dynamicIds = new Set(current.dynamic.map((f) => f.id));

  const additions: ProfileFact[] = [];
  const evidenceBumps: ProfileDiff["evidenceBumps"] = [];
  const promotions: ProfileDiff["promotions"] = [];
  const conflicts: ProfileDiff["conflicts"] = [];

  // Track ids added/bumped/superseded this pass so a repeated candidate in one batch
  // doesn't double-add, and a key retired this pass ignores later candidates for it.
  const touched = new Set<string>();
  const supersededKeys = new Set<string>();
  const supersedes: SupersedeOp[] = [];
  const bump = (id: string, prevCount: number) => {
    const newCount = prevCount + 1;
    evidenceBumps.push({ id, newCount, last_seen: today });
    if (dynamicIds.has(id) && newCount >= N) promotions.push({ id });
  };

  for (const c of candidates) {
    if (c.provenance === "ai-draft") continue; // model-collapse guard (SC1b)
    const primaryId = mintEntityId({ kind: OBSERVED_KIND, key: c.key });

    // A key retired earlier in this same batch ignores later candidates for it (first-supersede-wins, SC8).
    if (supersededKeys.has(primaryId)) continue;

    // SB-S3b supersede fork (explicit signal): retire the active fact at the target key
    // and install the winner at the candidate's key-id. Routed ONLY when the target
    // exists, is active, and holds a DIFFERENT value — else a re-sent signal whose target
    // already holds the new value would self-supersede every run (value-guard, SC3a).
    if (c.supersedes) {
      const targetId = mintEntityId({ kind: OBSERVED_KIND, key: c.supersedes });
      const target = byId.get(targetId);
      if (target && target.status === "active" && target.value !== c.value && !touched.has(targetId) && !supersededKeys.has(targetId)) {
        supersedes.push({ oldId: targetId, oldValue: target.value, winner: newFact(primaryId, c, today) });
        supersededKeys.add(targetId);
        touched.add(targetId);
        touched.add(primaryId); // reserve the winner's id
        continue;
      }
      // else: no / inactive / value-equal target → fall through to ordinary add/bump/conflict (graceful, SC4)
    }

    const prev = byId.get(primaryId);
    if (!prev) {
      if (!touched.has(primaryId)) {
        additions.push(newFact(primaryId, c, today));
        touched.add(primaryId);
      }
      continue;
    }
    if (prev.value === c.value) {
      bump(primaryId, prev.evidence_count); // SC1c/SC1d
      continue;
    }
    // conflict — keep both with a distinct alt id (SC1e), old fact untouched
    const aId = altId(c);
    const altPrev = byId.get(aId);
    if (altPrev && altPrev.value === c.value) {
      bump(aId, altPrev.evidence_count); // idempotent re-conflict → bump the alt
    } else if (!touched.has(aId)) {
      additions.push(newFact(aId, c, today));
      conflicts.push({ primaryId, primaryValue: prev.value, altId: aId });
      touched.add(aId);
    }
  }

  // Decay flags exclude superseded facts (SB-S3b, SC11): a retained archival fact is
  // audit, not a live signal, so it never throws perpetual staleFlags.
  const staleFlags = current.dynamic
    .filter((f) => f.status !== "superseded" && daysBetween(f.last_seen, today) > DECAY)
    .map((f) => ({ id: f.id, last_seen: f.last_seen, daysStale: daysBetween(f.last_seen, today) }));

  return { additions, evidenceBumps, promotions, conflicts, staleFlags, supersedes };
}

/**
 * Apply a proposed diff to produce the next ProfileDoc. Pure (returns a new doc).
 * Because primary and alt ids are byte-distinct, no two facts ever share an id, so
 * the bump/promote targets are unambiguous and the doc stays well-formed (SC3).
 *
 * SB-S3b supersede: a retired fact is REPLACED IN PLACE (re-minted to an archival id +
 * `status: superseded`, kept in its layer, never bumped/promoted — supersede wins) and
 * the winner is installed under the canonical key-id. A state-check (oldId present, still
 * active, value still === oldValue) makes re-apply a no-op and a stale diff safe.
 */
export function applyDiff(current: ProfileDoc, diff: ProfileDiff): ProfileDoc {
  // Resolve supersessions that still apply against THIS doc (idempotency + stale-diff guard).
  const allById = new Map([...current.static, ...current.dynamic].map((f) => [f.id, f]));
  const archive = new Map<string, ProfileFact>(); // oldId → archival replacement
  const winners: ProfileFact[] = [];
  for (const s of diff.supersedes ?? []) {
    const f = allById.get(s.oldId);
    if (!f || f.status === "superseded" || f.value !== s.oldValue) continue; // skip: re-apply / stale
    archive.set(s.oldId, { ...f, id: archivalId(s.oldId, s.oldValue), status: "superseded" });
    winners.push({ ...s.winner });
  }

  const bumpMap = new Map(diff.evidenceBumps.map((b) => [b.id, b]));
  // A superseded fact is replaced in place and NEVER bumped (supersede wins, SC12).
  const transform = (f: ProfileFact): ProfileFact => {
    const replaced = archive.get(f.id);
    if (replaced) return replaced;
    const b = bumpMap.get(f.id);
    return b ? { ...f, evidence_count: b.newCount, last_seen: b.last_seen } : { ...f };
  };

  let staticF = current.static.map(transform);
  let dynamicF = current.dynamic.map(transform);

  // Promote active dynamic facts only — a just-superseded fact is never promoted (SC12).
  const promoteIds = new Set(diff.promotions.map((p) => p.id));
  const promoting = (f: ProfileFact) => promoteIds.has(f.id) && f.status === "active";
  const promoted = dynamicF.filter(promoting);
  dynamicF = dynamicF.filter((f) => !promoting(f));
  staticF = [...staticF, ...promoted];

  // Additions + supersede winners land in the dynamic layer (winners re-earn promotion).
  dynamicF = [...dynamicF, ...diff.additions.map((f) => ({ ...f })), ...winners];

  return { schemaVersion: SCHEMA_VERSION, static: staticF, dynamic: dynamicF };
}

// ── consolidation-state sidecar (the only IO; brain data-root, reachable by both
//    the CLI via dataRoot and the session-start hook via getDataRoot) ────────────

const STATE_SUB = "brain/consolidation-state.json";

/** Read the last-run date; absent/malformed → {last_run:null}. */
export function readConsolidationState(): { last_run: string | null } {
  const p = dataRoot(STATE_SUB);
  if (!existsSync(p)) return { last_run: null };
  try {
    const j = JSON.parse(readFileSync(p, "utf8"));
    return { last_run: typeof j?.last_run === "string" ? j.last_run : null };
  } catch {
    return { last_run: null };
  }
}

/** Record the last-run date (called only by the gated --apply path). */
export function writeConsolidationState(date: string): void {
  const p = dataRoot(STATE_SUB);
  mkdirSync(dirname(p), { recursive: true });
  writeFileSync(p, JSON.stringify({ last_run: date }, null, 2) + "\n", "utf8");
}