diff --git a/src/db.ts b/src/db.ts index cd108e8..f963fa6 100644 --- a/src/db.ts +++ b/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", diff --git a/src/documents.ts b/src/documents.ts new file mode 100644 index 0000000..312c28b --- /dev/null +++ b/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 { + 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) || + /\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() + }; +} diff --git a/src/index.ts b/src/index.ts index 79f0b89..ddf623b 100644 --- a/src/index.ts +++ b/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() diff --git a/src/mcp.ts b/src/mcp.ts index d9d0550..fe0b931 100644 --- a/src/mcp.ts +++ b/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,