// The Masked Relay mini-fixture — deterministic node bodies for IT-2
// (tests/masked-relay.md; INTEGRATION-TESTS-PLAN.md §3 IT-2).
//
// This is the heaviest exercise of the upstream-read + read-isolation seam, scaled
// to ~10 nodes. Like the Counter fixture (`fake-render.ts`), every render body is a
// PURE deterministic function of (upstream truth read BY REFERENCE, own prior) —
// zero model calls — so the offline suite gates the commit and replay is free. The
// live sibling (`masked-relay.live.test.ts`) swaps these fakes for the live
// `createAgentRender` adapter; the reconciler cannot tell them apart.
//
//   Signal Inbox (gateway; facet `ledger`)
//        └─► Signal Ledger ──► Scout[price] ┐
//                          ├──► Scout[friction] ├─(peer-blind: no scout reads a sibling)
//                          └──► Scout[desire]  ┘
//                                  │ (all three)
//                                  ▼
//                          Viewport Masker (deterministic seed ⇒ visible/hidden per consumer)
//                                  │ (facet `view_e1` for Expander 1, `view_e2` for Expander 2)
//                    ┌─────────────┴─────────────┐
//             Expander 1 (⊂ view_e1)      Expander 2 (⊂ view_e2)   ← each a DIFFERENT masked view
//                    └─────────────┬─────────────┘
//                            ┌─────┴─────┐
//                      Critic[strong]  Critic[weak]   (⊂ both expanders)
//                            └─────┬─────┘
//                                  ▼
//                        Insight Synthesizer (⊂ scouts + expanders + critics; full provenance)
//                                  ▼
//                        Diversity Auditor (terminal diagnostic; ⊂ synthesizer + masker)
//
// The Viewport Masker is the determinism boundary in miniature: from the full set
// of scout claim-ids it computes, for each consumer slot, a VISIBLE subset and a
// HIDDEN subset using a deterministic seed (a stable hash mod over the seed +
// consumer + claim-id). Same seed + same scout claims ⇒ identical masked set ⇒ the
// run replays. Each expander's masked-view facet only moves when ITS visible subset
// moves, so an expander wakes only when its assigned view changes.

import { createHash } from "node:crypto";

import { ATOMIC_FACET, asFacet, asFingerprint, type Facet } from "../shapes";
import {
  fingerprintArtifact,
  files,
  jsonFile,
  readTextFile,
  InMemoryWorldModelStore,
  type Canonicalizer,
  type WorldModelFiles,
  type WorldModelStore,
} from "../world-model";
import { zeroCost, type RenderContext } from "../sdk/render-atom";
import { type MountedRender } from "../sdk/mounted-dag";
import {
  buildScenario,
  injectExternalReceipt,
  materialFingerprint,
  readJson,
  type NodeDecl,
  type ReconcileResult,
  type Scenario,
} from "./fixture";

// ---------------------------------------------------------------------------
// Identities + facets
// ---------------------------------------------------------------------------

export const SOURCE = "ingress.signal-inbox"; // the system's edge (phantom)
export const GATEWAY = "gateway.signal-inbox";
export const SIGNAL_LEDGER = "responsibility.signal-ledger";

export const SCOUT_PRICE = "responsibility.scout-price";
export const SCOUT_FRICTION = "responsibility.scout-friction";
export const SCOUT_DESIRE = "responsibility.scout-desire";
export const SCOUTS = [SCOUT_PRICE, SCOUT_FRICTION, SCOUT_DESIRE] as const;

export const VIEWPORT_MASKER = "responsibility.viewport-masker";

export const EXPANDER_1 = "responsibility.expander-1";
export const EXPANDER_2 = "responsibility.expander-2";
export const EXPANDERS = [EXPANDER_1, EXPANDER_2] as const;

export const CRITIC_STRONG = "responsibility.critic-strong";
export const CRITIC_WEAK = "responsibility.critic-weak";
export const CRITICS = [CRITIC_STRONG, CRITIC_WEAK] as const;

export const SYNTHESIZER = "responsibility.insight-synthesizer";
export const AUDITOR = "responsibility.diversity-auditor";

export const LEDGER = asFacet("ledger");
export const INBOX = asFacet("inbox");
/** The masker exposes one masked-view facet PER consumer slot (partial visibility). */
export const VIEW_E1 = asFacet("view_e1");
export const VIEW_E2 = asFacet("view_e2");

/** The deterministic mask seed (the determinism boundary; replay-stable). */
export const MASK_SEED = 1729;

// ---------------------------------------------------------------------------
// The external signal payload + harness deps the fake renders close over
// ---------------------------------------------------------------------------

export interface Signal {
  readonly id: string;
  readonly source: "customer_call" | "support_ticket" | "lost_deal" | "competitor";
  readonly text: string;
}

export interface MaskedDeps {
  readonly store: WorldModelStore;
  /** Raw external arrivals, in order (the gateway dedups by id). */
  readonly inbox: Signal[];
  /** Per-node render invocation counts — proves a memo-skip never calls a render. */
  readonly renders: Record<string, number>;
  /** The mask seed (mutable so a test can prove a seed change re-masks). */
  seed: number;
}

function freshDeps(store: WorldModelStore): MaskedDeps {
  return { store, inbox: [], renders: {}, seed: MASK_SEED };
}

function tick(deps: MaskedDeps, node: string): void {
  deps.renders[node] = (deps.renders[node] ?? 0) + 1;
}

// ---------------------------------------------------------------------------
// The deterministic mask — the determinism boundary in miniature
// ---------------------------------------------------------------------------

/**
 * Decide whether a claim-id is VISIBLE to a consumer slot under a seed. A stable
 * hash over (seed, consumer, claimId) reduced mod 3 — keep 2/3, hide 1/3. PURE +
 * deterministic: the SAME (seed, consumer, claims) always yields the SAME split,
 * so the run replays and an unchanged seed never re-masks (world-model.md §3).
 */
export function isVisible(
  seed: number,
  consumer: string,
  claimId: string,
): boolean {
  const h = createHash("sha256")
    .update(`${seed} ${consumer} ${claimId}`)
    .digest();
  return h[0]! % 3 !== 0;
}

/** The masked view a consumer slot sees: its visible + hidden claim-ids. */
export interface MaskedView {
  readonly consumer: string;
  readonly seed: number;
  readonly visible: readonly string[];
  readonly hidden_hashes: readonly string[];
}

function maskFor(
  seed: number,
  consumer: string,
  claimIds: readonly string[],
): MaskedView {
  const visible: string[] = [];
  const hidden_hashes: string[] = [];
  for (const id of claimIds) {
    if (isVisible(seed, consumer, id)) {
      visible.push(id);
    } else {
      // Hidden claims are referenced only by an opaque hash — never their content.
      hidden_hashes.push(
        createHash("sha256").update(id).digest("hex").slice(0, 12),
      );
    }
  }
  return { consumer, seed, visible: visible.sort(), hidden_hashes: hidden_hashes.sort() };
}

// ---------------------------------------------------------------------------
// Canonicalizers
// ---------------------------------------------------------------------------

const atomicTruth: Canonicalizer = (fm) => ({
  [ATOMIC_FACET]: asFingerprint(fingerprintArtifact(fm)),
});

function readTruth(fm: WorldModelFiles): Record<string, unknown> {
  const bytes = fm["truth.json"];
  return bytes === undefined
    ? {}
    : (JSON.parse(readTextFile(bytes)) as Record<string, unknown>);
}

/** Gateway: the deduped signal ledger is the single `ledger` facet. */
const gatewayCanon: Canonicalizer = (fm) => {
  const t = readTruth(fm);
  return {
    [ATOMIC_FACET]: asFingerprint(fingerprintArtifact(fm)),
    [LEDGER]: materialFingerprint(t["items"] ?? []),
  };
};

/** Ingress (phantom source): the raw inbox is the single `inbox` facet. */
const ingressCanon: Canonicalizer = (fm) => {
  const bytes = fm["inbox.json"];
  const inbox = bytes === undefined ? [] : JSON.parse(readTextFile(bytes));
  return {
    [ATOMIC_FACET]: asFingerprint(fingerprintArtifact(fm)),
    [INBOX]: materialFingerprint(inbox),
  };
};

/**
 * Viewport Masker: ONE masked-view facet per consumer slot (partial visibility).
 * Each facet token is over ONLY that consumer's visible subset — so Expander 1's
 * `view_e1` facet moves iff its OWN visible set moves, independent of Expander 2's.
 * This is what makes "wake only when YOUR masked view changes" a fingerprint fact.
 */
export const maskerCanon: Canonicalizer = (fm) => {
  const t = readTruth(fm);
  const views = (t["views"] ?? {}) as Record<string, MaskedView>;
  return {
    [ATOMIC_FACET]: asFingerprint(fingerprintArtifact(fm)),
    [VIEW_E1]: materialFingerprint(views[EXPANDER_1]?.visible ?? []),
    [VIEW_E2]: materialFingerprint(views[EXPANDER_2]?.visible ?? []),
  };
};

// ---------------------------------------------------------------------------
// The render bodies (factories closing over deps)
// ---------------------------------------------------------------------------

function dedupById(items: readonly Signal[]): Signal[] {
  const seen = new Set<string>();
  const out: Signal[] = [];
  for (const s of items) {
    if (seen.has(s.id)) continue;
    seen.add(s.id);
    out.push(s);
  }
  return out;
}

function commit(truth: unknown, ctx: RenderContext) {
  return {
    world_model: files({ "truth.json": jsonFile(truth) }),
    cost: zeroCost(ctx.wake.source),
  };
}

function gatewayRender(deps: MaskedDeps): MountedRender {
  return (ctx) => {
    tick(deps, GATEWAY);
    const inbox = (readJson<Signal[]>(deps.store, SOURCE, "inbox.json") ??
      []) as Signal[];
    const items = dedupById(inbox);
    return commit({ items, count: items.length }, ctx);
  };
}

function signalLedgerRender(deps: MaskedDeps): MountedRender {
  return (ctx) => {
    tick(deps, SIGNAL_LEDGER);
    const gw = readJson(deps.store, GATEWAY);
    const items = (gw?.["items"] ?? []) as Signal[];
    // The ledger normalizes each signal into a stable {id, source, dedupe_key}.
    const ledger = items.map((s) => ({
      id: s.id,
      source: s.source,
      dedupe_key: `${s.source}:${s.id}`,
      observed_at: s.id, // deterministic, not wall-clock
    }));
    return commit({ items: ledger, stable_fingerprint: materialFingerprint(ledger) }, ctx);
  };
}

/**
 * A Stage-1 Scout. Reads ONLY the Signal Ledger (peer-blind: its `requires` lists
 * the ledger, never a sibling scout — so its resolved inbound edges, and thus the
 * read-isolation pin, exclude every sibling). Emits persona-tagged claims, one per
 * ledger item, with a stable claim-id derived from the persona + item id.
 */
function scoutRender(deps: MaskedDeps, persona: string, node: string): MountedRender {
  return (ctx) => {
    tick(deps, node);
    const ledger = readJson(deps.store, SIGNAL_LEDGER);
    const items = (ledger?.["items"] ?? []) as { id: string }[];
    const claims = items.map((it) => ({
      claim_id: `${persona}:${it.id}`,
      persona,
      evidence_ref: it.id,
      confidence: 0.5,
    }));
    return commit({ persona, claims, claim_ids: claims.map((c) => c.claim_id) }, ctx);
  };
}

/**
 * The Viewport Masker. Reads ALL THREE scouts (its subscriptions), gathers every
 * claim-id, and computes a DETERMINISTIC masked view per consumer slot from the
 * seed. Same seed + same scout claims ⇒ identical views (the determinism boundary).
 */
function maskerRender(deps: MaskedDeps): MountedRender {
  return (ctx) => {
    tick(deps, VIEWPORT_MASKER);
    const allClaimIds: string[] = [];
    for (const scout of SCOUTS) {
      const s = readJson(deps.store, scout);
      for (const id of (s?.["claim_ids"] ?? []) as string[]) {
        allClaimIds.push(id);
      }
    }
    allClaimIds.sort();
    const views: Record<string, MaskedView> = {};
    for (const consumer of EXPANDERS) {
      views[consumer] = maskFor(deps.seed, consumer, allClaimIds);
    }
    const coverage_matrix = Object.fromEntries(
      EXPANDERS.map((c) => [c, views[c]!.visible.length]),
    );
    return commit(
      {
        seed: deps.seed,
        consumers: [...EXPANDERS],
        views,
        coverage_matrix,
        policy_reason: "deterministic 2/3 keep, 1/3 hide per (seed, consumer, claim)",
      },
      ctx,
    );
  };
}

/**
 * A Stage-2 Expander. Reads ONLY its assigned masked view from the Masker (partial
 * visibility) + the Signal Ledger — never a peer expander, never the OTHER masked
 * view. Expands each VISIBLE claim; records the hidden hashes it could NOT see.
 */
function expanderRender(
  deps: MaskedDeps,
  node: string,
  slot: string,
): MountedRender {
  return (ctx) => {
    tick(deps, node);
    const mask = readJson(deps.store, VIEWPORT_MASKER);
    const views = (mask?.["views"] ?? {}) as Record<string, MaskedView>;
    const myView = views[slot] ?? { visible: [], hidden_hashes: [] };
    const expanded = (myView.visible ?? []).map((id) => ({
      from_claim: id,
      hypothesis: `expanded(${id})`,
    }));
    return commit(
      {
        slot,
        visible_count: (myView.visible ?? []).length,
        hidden_count: (myView.hidden_hashes ?? []).length,
        expanded_claims: expanded,
        preserved_minorities: [],
      },
      ctx,
    );
  };
}

/** A Stage-3 Critic. Reads BOTH expanders (its subscriptions). */
function criticRender(deps: MaskedDeps, node: string, mode: string): MountedRender {
  return (ctx) => {
    tick(deps, node);
    let total = 0;
    for (const exp of EXPANDERS) {
      const e = readJson(deps.store, exp);
      total += ((e?.["expanded_claims"] ?? []) as unknown[]).length;
    }
    return commit({ mode, claims_reviewed: total, critique: `${mode}:${total}` }, ctx);
  };
}

/**
 * The Insight Synthesizer. Reads the FULL receipt trail — scouts + expanders +
 * critics. Its memo records WHICH upstream producers it consumed (full provenance):
 * the harness threads its consumed input fingerprints into the receipt, and the
 * body records the upstream node-ids it read so the memo names what moved.
 */
function synthesizerRender(deps: MaskedDeps): MountedRender {
  return (ctx) => {
    tick(deps, SYNTHESIZER);
    const cited: string[] = [];
    let evidence = 0;
    for (const up of [...SCOUTS, ...EXPANDERS, ...CRITICS]) {
      const u = readJson(deps.store, up);
      if (u !== null) {
        cited.push(up);
        evidence += 1;
      }
    }
    return commit(
      {
        headline: `insight over ${evidence} upstream ledgers`,
        evidence_refs: cited,
        // Full provenance: the synthesizer names which upstream receipts it saw.
        changed_since_last: ctx.input_fingerprints.slice(),
        minority_threads: [],
        best_objection: "consensus may be premature",
        recommended_probe: "interview a hidden-claim source",
      },
      ctx,
    );
  };
}

/** The terminal Diversity Auditor. Reads the synthesizer + the masker (diagnostic). */
function auditorRender(deps: MaskedDeps): MountedRender {
  return (ctx) => {
    tick(deps, AUDITOR);
    const memo = readJson(deps.store, SYNTHESIZER);
    const mask = readJson(deps.store, VIEWPORT_MASKER);
    const cov = (mask?.["coverage_matrix"] ?? {}) as Record<string, number>;
    return commit(
      {
        convergence_score: ((memo?.["evidence_refs"] ?? []) as unknown[]).length,
        coverage_matrix: cov,
        mask_rate_recommendation: "hold",
        show_all_baseline_recommendation: "no",
      },
      ctx,
    );
  };
}

// ---------------------------------------------------------------------------
// The fixture + driver
// ---------------------------------------------------------------------------

export interface MaskedScenario extends Scenario {
  readonly deps: MaskedDeps;
}

/** Build the Masked Relay mini-fixture mounted over the real reconciler. */
export function maskedRelayScenario(): MaskedScenario {
  const realStore = new InMemoryWorldModelStore();
  const deps = freshDeps(realStore);

  const scoutDecl = (
    id: string,
    persona: string,
    name: string,
  ): NodeDecl => ({
    id,
    kind: "responsibility",
    name,
    // Peer-blind: a scout subscribes ONLY to the Signal Ledger, NEVER a sibling.
    requires: [{ producer: SIGNAL_LEDGER }],
    maintains: ["scout_ledger"],
    continuity: "input-driven",
    render: scoutRender(deps, persona, id),
    canonicalizer: atomicTruth,
  });

  const expanderDecl = (id: string, slot: string, view: Facet): NodeDecl => ({
    id,
    kind: "responsibility",
    name: `Stage 2 Expander ${slot}`,
    // Partial visibility + "wake only when YOUR masked view changes": the expander
    // SUBSCRIBES to ONLY its assigned masked-view facet — never the other view,
    // never a peer expander, and NOT the ledger's whole-truth facet (subscribing to
    // the ledger atomic would wake it on EVERY signal, defeating selective wake;
    // its masked view already carries the visible claim-ids it expands). This keeps
    // "wakes iff its view moved" an honest fingerprint fact, not a confound.
    requires: [{ producer: VIEWPORT_MASKER, facet: view }],
    maintains: ["expansion_ledger"],
    continuity: "input-driven",
    render: expanderRender(deps, id, slot),
    canonicalizer: atomicTruth,
  });

  const criticDecl = (id: string, mode: string): NodeDecl => ({
    id,
    kind: "responsibility",
    name: `Stage 3 Critic ${mode}`,
    requires: EXPANDERS.map((e) => ({ producer: e })),
    maintains: ["critic_ledger"],
    continuity: "input-driven",
    render: criticRender(deps, id, mode),
    canonicalizer: atomicTruth,
  });

  const decls: NodeDecl[] = [
    {
      id: GATEWAY,
      kind: "gateway",
      name: "Signal Inbox",
      requires: [{ producer: SOURCE, facet: INBOX }],
      maintains: ["ledger"],
      continuity: "external",
      render: gatewayRender(deps),
      canonicalizer: gatewayCanon,
    },
    {
      id: SIGNAL_LEDGER,
      kind: "responsibility",
      name: "Signal Ledger",
      requires: [{ producer: GATEWAY, facet: LEDGER }],
      maintains: ["signal_ledger"],
      continuity: "input-driven",
      render: signalLedgerRender(deps),
      canonicalizer: atomicTruth,
    },
    scoutDecl(SCOUT_PRICE, "price", "Stage 1 Scout: price anxiety"),
    scoutDecl(SCOUT_FRICTION, "friction", "Stage 1 Scout: workflow friction"),
    scoutDecl(SCOUT_DESIRE, "desire", "Stage 1 Scout: latent desire"),
    {
      id: VIEWPORT_MASKER,
      kind: "responsibility",
      name: "Viewport Masker",
      requires: SCOUTS.map((s) => ({ producer: s })),
      maintains: ["view_e1", "view_e2"],
      continuity: "input-driven",
      render: maskerRender(deps),
      canonicalizer: maskerCanon,
    },
    expanderDecl(EXPANDER_1, EXPANDER_1, VIEW_E1),
    expanderDecl(EXPANDER_2, EXPANDER_2, VIEW_E2),
    criticDecl(CRITIC_STRONG, "strong"),
    criticDecl(CRITIC_WEAK, "weak"),
    {
      id: SYNTHESIZER,
      kind: "responsibility",
      name: "Insight Synthesizer",
      // Full provenance: subscribes to the whole upstream trail.
      requires: [...SCOUTS, ...EXPANDERS, ...CRITICS].map((p) => ({ producer: p })),
      maintains: ["insight_memo"],
      continuity: "input-driven",
      render: synthesizerRender(deps),
      canonicalizer: atomicTruth,
    },
    {
      id: AUDITOR,
      kind: "responsibility",
      name: "Diversity Auditor",
      requires: [{ producer: SYNTHESIZER }, { producer: VIEWPORT_MASKER }],
      maintains: ["diversity_audit"],
      continuity: "input-driven",
      render: auditorRender(deps),
      canonicalizer: atomicTruth,
    },
  ];

  const scn = buildScenario(decls, { store: realStore });
  return { ...scn, deps };
}

/**
 * Deliver one external signal and drain to quiescence. Mirrors the Counter
 * fixture's `deliverEvent`: append to the raw inbox, inject the ingress receipt
 * (moving the gateway's input), then wake the gateway and let propagation flow.
 */
export function deliverSignal(
  scn: MaskedScenario,
  signal: Signal,
): readonly ReconcileResult[] {
  scn.deps.inbox.push(signal);
  injectExternalReceipt(
    scn,
    SOURCE,
    files({ "inbox.json": jsonFile(scn.deps.inbox) }),
    ingressCanon,
  );
  return scn.dag.ingest(GATEWAY);
}

/**
 * Re-wake the gateway with the EXACT current inbox (no new signal): the raw inbox
 * array is byte-identical, so the ingress `inbox` facet is UNMOVED, the gateway's
 * memo key does not move, and the whole relay memo-SKIPS. This is the honest
 * no-change re-run (distinct from re-delivering a duplicate id, which appends to
 * the inbox array and DOES move the raw fingerprint).
 */
export function reWakeUnchanged(
  scn: MaskedScenario,
): readonly ReconcileResult[] {
  injectExternalReceipt(
    scn,
    SOURCE,
    files({ "inbox.json": jsonFile(scn.deps.inbox) }),
    ingressCanon,
  );
  return scn.dag.ingest(GATEWAY);
}

export { readJson };