Sync source document freezing for story evidence

src/db.ts

@@ -3,6 +3,7 @@ import { randomUUID } from "node:crypto";
 import { createHash } from "node:crypto";
 import type { NormalizedItem, SourceDefinition } from "./types.js";
 import type { ScoredItem } from "./types.js";
+import { fetchSourceDocument } from "./documents.js";
 
 const { Pool } = pg;
 
@@ -75,6 +76,10 @@ export async function migrate() {
       content_type TEXT NOT NULL DEFAULT 'application/vnd.media-intelligence.normalized+json',
       byte_length INTEGER NOT NULL DEFAULT 0,
       snapshot_json JSONB NOT NULL,
+      snapshot_kind TEXT NOT NULL DEFAULT 'normalized',
+      canonical_url TEXT,
+      document_text TEXT,
+      http_status INTEGER,
       UNIQUE(signal_id, content_hash)
     );
 
@@ -226,6 +231,10 @@ export async function migrate() {
   await pool.query("ALTER TABLE evidence ADD COLUMN IF NOT EXISTS snapshot_id TEXT REFERENCES source_snapshots(id) ON DELETE SET NULL");
   await pool.query("ALTER TABLE evidence ADD COLUMN IF NOT EXISTS evidence_kind TEXT NOT NULL DEFAULT 'normalized_excerpt'");
   await pool.query("ALTER TABLE evidence ADD COLUMN IF NOT EXISTS locator TEXT");
+  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS snapshot_kind TEXT NOT NULL DEFAULT 'normalized'");
+  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS canonical_url TEXT");
+  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS document_text TEXT");
+  await pool.query("ALTER TABLE source_snapshots ADD COLUMN IF NOT EXISTS http_status INTEGER");
 
   await pool.query(`
     INSERT INTO source_snapshots(
@@ -700,7 +709,10 @@ export async function getStoryThread(id: string) {
               src.source_class AS provenance_source_class,
               src.independence_group AS provenance_independence_group,
               src.authority_tier AS provenance_authority_tier,
+              e.evidence_kind AS evidence_kind,
+              e.locator AS evidence_locator,
               ss.id AS snapshot_id,
+              ss.snapshot_kind AS snapshot_kind,
               ss.retrieved_at AS snapshot_retrieved_at,
               ss.published_at AS snapshot_published_at
        FROM story_nodes n
@@ -930,7 +942,13 @@ export async function auditStorySources(threadId: string) {
   const edges = Array.isArray(story.edges) ? story.edges : [];
 
   const evidenceBacked = nodes.filter((n: any) =>
-    Boolean(n.evidence_id && n.snapshot_id && n.provenance_source_id)
+    Boolean(
+      n.evidence_id &&
+      n.snapshot_id &&
+      n.provenance_source_id &&
+      n.snapshot_kind === "source_document" &&
+      n.evidence_kind === "source_document"
+    )
   );
 
   const distinctSourceIds = Array.from(new Set(
@@ -1020,6 +1038,131 @@ export async function auditStorySources(threadId: string) {
   };
 }
 
+export async function enrichEvidenceDocument(evidenceId: string) {
+  const current = await pool.query(
+    `SELECT e.*,s.source_id,s.url,s.published_at
+     FROM evidence e
+     JOIN signals s ON s.id=e.signal_id
+     WHERE e.id=$1`,
+    [evidenceId]
+  );
+  if (!current.rowCount) throw new Error("Evidence not found");
+
+  const row = current.rows[0];
+  if (row.evidence_kind === "source_document" && row.snapshot_id) {
+    return getEvidence(evidenceId);
+  }
+
+  const document = await fetchSourceDocument(row.url);
+  const snapshotId = randomUUID();
+
+  await pool.query("BEGIN");
+  try {
+    await pool.query(
+      `INSERT INTO source_snapshots(
+         id,source_id,signal_id,source_url,retrieved_at,published_at,content_hash,
+         content_type,byte_length,snapshot_json,snapshot_kind,canonical_url,
+         document_text,http_status
+       )
+       VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,'source_document',$11,$12,$13)
+       ON CONFLICT(signal_id,content_hash) DO NOTHING`,
+      [
+        snapshotId,
+        row.source_id,
+        row.signal_id,
+        row.url,
+        document.retrievedAt,
+        row.published_at,
+        document.contentHash,
+        document.contentType,
+        document.byteLength,
+        JSON.stringify({
+          requestedUrl: document.requestedUrl,
+          canonicalUrl: document.canonicalUrl,
+          title: document.title,
+          retrievedAt: document.retrievedAt,
+          contentHash: document.contentHash
+        }),
+        document.canonicalUrl,
+        document.text,
+        document.httpStatus
+      ]
+    );
+
+    const snapshot = await pool.query(
+      `SELECT id FROM source_snapshots
+       WHERE signal_id=$1 AND content_hash=$2
+       ORDER BY retrieved_at DESC
+       LIMIT 1`,
+      [row.signal_id, document.contentHash]
+    );
+
+    const resolvedSnapshotId = snapshot.rows[0]?.id;
+    if (!resolvedSnapshotId) throw new Error("Failed to persist source document snapshot");
+
+    await pool.query(
+      `UPDATE evidence
+       SET snapshot_id=$2,
+           content_hash=$3,
+           excerpt=$4,
+           evidence_kind='source_document',
+           locator='main document body'
+       WHERE id=$1`,
+      [
+        evidenceId,
+        resolvedSnapshotId,
+        document.contentHash,
+        document.text.slice(0, 8000)
+      ]
+    );
+
+    await pool.query(
+      `INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
+       VALUES($1,'evidence.source_document_frozen','evidence',$2,$3::jsonb)`,
+      [randomUUID(), evidenceId, JSON.stringify({
+        sourceId: row.source_id,
+        url: document.canonicalUrl,
+        snapshotId: resolvedSnapshotId,
+        contentHash: document.contentHash
+      })]
+    );
+
+    await pool.query("COMMIT");
+    return getEvidence(evidenceId);
+  } catch (error) {
+    await pool.query("ROLLBACK");
+    throw error;
+  }
+}
+
+export async function enrichStoryEvidenceDocuments(limit = 20) {
+  const result = await pool.query(
+    `SELECT DISTINCT n.evidence_id
+     FROM story_nodes n
+     JOIN evidence e ON e.id=n.evidence_id
+     WHERE n.evidence_id IS NOT NULL
+       AND e.evidence_kind <> 'source_document'
+     ORDER BY n.evidence_id
+     LIMIT $1`,
+    [Math.min(Math.max(limit, 1), 100)]
+  );
+
+  const enriched: any[] = [];
+  for (const row of result.rows) {
+    try {
+      const evidence = await enrichEvidenceDocument(row.evidence_id);
+      enriched.push({ evidence_id: row.evidence_id, ok: true, snapshot_id: evidence?.snapshot_id ?? null });
+    } catch (error) {
+      enriched.push({
+        evidence_id: row.evidence_id,
+        ok: false,
+        error: error instanceof Error ? error.message : String(error)
+      });
+    }
+  }
+  return enriched;
+}
+
 export async function seedInitialStoryThread() {
   const existing = await pool.query(
     "SELECT id FROM story_threads WHERE title=$1 LIMIT 1",

src/documents.ts

@@ -0,0 +1,99 @@
+import { createHash } from "node:crypto";
+import { load } from "cheerio";
+
+const DOCUMENT_UA =
+  "Mozilla/5.0 (compatible; ArrowstackMediaIntelligence/0.1; +https://media.arrowstack.dev)";
+
+function normaliseText(value: string) {
+  return value.replace(/\u00a0/g, " ").replace(/\s+/g, " ").trim();
+}
+
+export interface SourceDocumentSnapshot {
+  requestedUrl: string;
+  canonicalUrl: string;
+  title: string;
+  contentType: string;
+  httpStatus: number;
+  byteLength: number;
+  contentHash: string;
+  text: string;
+  retrievedAt: string;
+}
+
+export async function fetchSourceDocument(url: string): Promise<SourceDocumentSnapshot> {
+  const response = await fetch(url, {
+    headers: {
+      accept: "text/html,application/xhtml+xml,application/json,text/plain;q=0.8,*/*;q=0.4",
+      "accept-language": "en-NZ,en;q=0.9",
+      "user-agent": DOCUMENT_UA
+    },
+    redirect: "follow",
+    signal: AbortSignal.timeout(25000)
+  });
+
+  if (!response.ok) {
+    throw new Error(`Source document HTTP ${response.status} ${response.statusText} for ${url}`);
+  }
+
+  const bytes = Buffer.from(await response.arrayBuffer());
+  const contentType = String(response.headers.get("content-type") ?? "application/octet-stream").toLowerCase();
+  const contentHash = createHash("sha256").update(bytes).digest("hex");
+  let title = "";
+  let canonicalUrl = response.url || url;
+  let text = "";
+
+  if (contentType.includes("html") || contentType.includes("xhtml")) {
+    const html = bytes.toString("utf8");
+    if (
+      /incapsula incident id/i.test(html) ||
+      /request unsuccessful/i.test(html) ||
+      /<title>\s*radware page\s*<\/title>/i.test(html)
+    ) {
+      throw new Error(`Source document blocked by upstream WAF for ${url}`);
+    }
+
+    const $ = load(html);
+    $("script,style,noscript,svg,nav,footer,header").remove();
+    title = normaliseText($("h1").first().text() || $("title").first().text());
+
+    const canonical = $('link[rel="canonical"]').attr("href");
+    if (canonical) {
+      try { canonicalUrl = new URL(canonical, canonicalUrl).toString(); } catch {}
+    }
+
+    const candidates = [
+      $("main").first(),
+      $("article").first(),
+      $('[role="main"]').first(),
+      $("#content").first(),
+      $(".main-content").first(),
+      $("body").first()
+    ].filter((node) => node.length);
+
+    text = candidates
+      .map((node) => normaliseText(node.text()))
+      .sort((a, b) => b.length - a.length)[0] ?? "";
+  } else if (
+    contentType.includes("json") ||
+    contentType.startsWith("text/") ||
+    contentType.includes("xml")
+  ) {
+    text = normaliseText(bytes.toString("utf8"));
+  }
+
+  if (!text) {
+    throw new Error(`Source document is not text-extractable (${contentType}) for ${url}`);
+  }
+
+  return {
+    requestedUrl: url,
+    canonicalUrl,
+    title,
+    contentType,
+    httpStatus: response.status,
+    byteLength: bytes.length,
+    contentHash,
+    text: text.slice(0, 120000),
+    retrievedAt: new Date().toISOString()
+  };
+}

src/index.ts

@@ -1,6 +1,7 @@
 import express from "express";
 import {
   acknowledgeAgentMessage,
+  enrichStoryEvidenceDocuments,
   getOpportunity,
   getStoryThread,
   listAgentMessages,
@@ -438,6 +439,16 @@ async function start() {
     Number(process.env.SYNC_INTERVAL_MINUTES ?? 15),
     5
   );
+  setTimeout(
+    () =>
+      enrichStoryEvidenceDocuments()
+        .then((result) =>
+          console.log(JSON.stringify({ event: "story_evidence_enrichment", result }))
+        )
+        .catch((error) => console.error(error)),
+    2500
+  );
+
   setTimeout(
     () =>
       syncAll()

src/mcp.ts

@@ -6,6 +6,7 @@ import {
   addStoryNode,
   completeCounterEvidenceCheck,
   createStoryThread,
+  enrichEvidenceDocument,
   getAgentMessage,
   getEvidence,
   getOpportunity,
@@ -88,6 +89,16 @@ const tools = [
       additionalProperties: false
     }
   },
+  {
+    name: "evidence_enrich",
+    description: "Fetch and freeze the actual source document behind an evidence record, replacing discovery-only evidence with a source-document snapshot when possible.",
+    inputSchema: {
+      type: "object",
+      properties: { id: { type: "string" } },
+      required: ["id"],
+      additionalProperties: false
+    }
+  },
   {
     name: "evidence_search",
     description: "Search stored evidence and source text.",
@@ -280,6 +291,7 @@ async function callTool(name: string, args: any, actor: ReturnType<typeof actorF
   if (name === "opportunity_get") return result(await getOpportunity(args.id));
   if (name === "opportunity_mark") return result(await markOpportunity(args.id, args.status));
   if (name === "evidence_get") return result(await getEvidence(args.id));
+  if (name === "evidence_enrich") return result(await enrichEvidenceDocument(args.id));
   if (name === "evidence_search") return result(await searchEvidence(args.query, args.limit));
   if (name === "messages_list") return result(await listAgentMessages({
     acknowledged: args?.acknowledged,