diff --git a/src/db.ts b/src/db.ts index 9877f00..cd108e8 100644 --- a/src/db.ts +++ b/src/db.ts @@ -27,6 +27,10 @@ export async function migrate() { last_synced_at TIMESTAMPTZ, last_error TEXT, baseline_complete BOOLEAN NOT NULL DEFAULT FALSE, + publisher TEXT NOT NULL DEFAULT '', + source_class TEXT NOT NULL DEFAULT 'catalogue', + independence_group TEXT NOT NULL DEFAULT '', + authority_tier INTEGER NOT NULL DEFAULT 3, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); @@ -60,13 +64,33 @@ export async function migrate() { CREATE INDEX IF NOT EXISTS idx_signals_source_detected ON signals(source_id, detected_at DESC); CREATE INDEX IF NOT EXISTS idx_signals_external ON signals(source_id, external_id); + CREATE TABLE IF NOT EXISTS source_snapshots ( + id TEXT PRIMARY KEY, + source_id TEXT NOT NULL REFERENCES sources(id), + signal_id TEXT NOT NULL REFERENCES signals(id) ON DELETE CASCADE, + source_url TEXT NOT NULL, + retrieved_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + published_at TIMESTAMPTZ, + content_hash TEXT NOT NULL, + content_type TEXT NOT NULL DEFAULT 'application/vnd.media-intelligence.normalized+json', + byte_length INTEGER NOT NULL DEFAULT 0, + snapshot_json JSONB NOT NULL, + UNIQUE(signal_id, content_hash) + ); + + CREATE INDEX IF NOT EXISTS idx_source_snapshots_signal + ON source_snapshots(signal_id, retrieved_at DESC); + CREATE TABLE IF NOT EXISTS evidence ( id TEXT PRIMARY KEY, signal_id TEXT NOT NULL REFERENCES signals(id) ON DELETE CASCADE, source_url TEXT NOT NULL, retrieved_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), content_hash TEXT NOT NULL, - excerpt TEXT NOT NULL DEFAULT '' + excerpt TEXT NOT NULL DEFAULT '', + snapshot_id TEXT REFERENCES source_snapshots(id) ON DELETE SET NULL, + evidence_kind TEXT NOT NULL DEFAULT 'normalized_excerpt', + locator TEXT ); CREATE TABLE IF NOT EXISTS opportunities ( @@ -195,6 +219,48 @@ export async function migrate() { `); await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS baseline_complete BOOLEAN NOT NULL DEFAULT FALSE"); + await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS publisher TEXT NOT NULL DEFAULT ''"); + await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS source_class TEXT NOT NULL DEFAULT 'catalogue'"); + await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS independence_group TEXT NOT NULL DEFAULT ''"); + await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS authority_tier INTEGER NOT NULL DEFAULT 3"); + 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(` + INSERT INTO source_snapshots( + id,source_id,signal_id,source_url,retrieved_at,published_at,content_hash, + content_type,byte_length,snapshot_json + ) + SELECT + 'snapshot-' || e.id, + s.source_id, + s.id, + e.source_url, + e.retrieved_at, + s.published_at, + e.content_hash, + 'application/vnd.media-intelligence.normalized+json', + LENGTH(s.raw_json::text), + jsonb_build_object( + 'title',s.title, + 'summary',s.summary, + 'url',s.url, + 'publishedAt',s.published_at, + 'topics',s.topics, + 'raw',s.raw_json + ) + FROM evidence e + JOIN signals s ON s.id=e.signal_id + WHERE e.snapshot_id IS NULL + ON CONFLICT(id) DO NOTHING + `); + await pool.query(` + UPDATE evidence + SET snapshot_id='snapshot-' || id + WHERE snapshot_id IS NULL + AND EXISTS (SELECT 1 FROM source_snapshots ss WHERE ss.id='snapshot-' || evidence.id) + `); await pool.query("ALTER TABLE signals ADD COLUMN IF NOT EXISTS change_kind TEXT NOT NULL DEFAULT 'new'"); await pool.query("ALTER TABLE agent_messages ADD COLUMN IF NOT EXISTS dedupe_key TEXT"); await pool.query("CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_messages_dedupe_key ON agent_messages(dedupe_key)"); @@ -220,10 +286,22 @@ export async function migrate() { export async function seedSources(sources: SourceDefinition[]) { for (const source of sources) { await pool.query( - `INSERT INTO sources(id,name,source_type,url) - VALUES($1,$2,$3,$4) - ON CONFLICT(id) DO UPDATE SET name=EXCLUDED.name, source_type=EXCLUDED.source_type, url=EXCLUDED.url`, - [source.id, source.name, source.type, source.url] + `INSERT INTO sources( + id,name,source_type,url,publisher,source_class,independence_group,authority_tier + ) + VALUES($1,$2,$3,$4,$5,$6,$7,$8) + ON CONFLICT(id) DO UPDATE SET + name=EXCLUDED.name, + source_type=EXCLUDED.source_type, + url=EXCLUDED.url, + publisher=EXCLUDED.publisher, + source_class=EXCLUDED.source_class, + independence_group=EXCLUDED.independence_group, + authority_tier=EXCLUDED.authority_tier`, + [ + source.id, source.name, source.type, source.url, + source.publisher, source.sourceClass, source.independenceGroup, source.authorityTier + ] ); } @@ -277,6 +355,7 @@ export async function storeItem( const signalId = randomUUID(); const opportunityId = randomUUID(); const evidenceId = randomUUID(); + const snapshotId = randomUUID(); const minScore = Number(process.env.OPPORTUNITY_MIN_SCORE ?? 2); const maxAgeDays = Math.max(Number(process.env.MAX_OPPORTUNITY_AGE_DAYS ?? 14), 1); @@ -316,15 +395,48 @@ export async function storeItem( ] ); + const snapshotPayload = { + title: item.title, + summary: item.summary, + url: item.url, + publishedAt: item.publishedAt, + topics: item.topics, + metrics: item.metrics ?? null, + raw: item.raw + }; + const snapshotText = JSON.stringify(snapshotPayload); + await pool.query( - `INSERT INTO evidence(id,signal_id,source_url,content_hash,excerpt) - VALUES($1,$2,$3,$4,$5)`, + `INSERT INTO source_snapshots( + id,source_id,signal_id,source_url,published_at,content_hash, + content_type,byte_length,snapshot_json + ) + VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb)`, + [ + snapshotId, + item.sourceId, + signalId, + item.url, + item.publishedAt, + hash, + "application/vnd.media-intelligence.normalized+json", + Buffer.byteLength(snapshotText), + snapshotText + ] + ); + + await pool.query( + `INSERT INTO evidence( + id,signal_id,source_url,content_hash,excerpt,snapshot_id,evidence_kind,locator + ) + VALUES($1,$2,$3,$4,$5,$6,'normalized_excerpt','title+summary')`, [ evidenceId, signalId, item.url, hash, - `${item.title}\n${item.summary}`.slice(0, 2000) + `${item.title}\n${item.summary}`.slice(0, 2000), + snapshotId ] ); @@ -446,8 +558,17 @@ export async function markOpportunity(id: string, status: string) { export async function getEvidence(id: string) { const result = await pool.query( - `SELECT e.*, s.source_id,s.title,s.summary,s.url,s.published_at - FROM evidence e JOIN signals s ON s.id=e.signal_id WHERE e.id=$1`, + `SELECT e.*, s.source_id,s.title,s.summary,s.url,s.published_at, + ss.retrieved_at AS snapshot_retrieved_at, + ss.published_at AS snapshot_published_at, + ss.content_type AS snapshot_content_type, + ss.byte_length AS snapshot_byte_length, + src.publisher,src.source_class,src.independence_group,src.authority_tier + FROM evidence e + JOIN signals s ON s.id=e.signal_id + LEFT JOIN source_snapshots ss ON ss.id=e.snapshot_id + LEFT JOIN sources src ON src.id=s.source_id + WHERE e.id=$1`, [id] ); return result.rows[0] ?? null; @@ -455,8 +576,13 @@ export async function getEvidence(id: string) { export async function searchEvidence(query: string, limit = 20) { const result = await pool.query( - `SELECT e.*, s.source_id,s.title,s.summary,s.url,s.published_at - FROM evidence e JOIN signals s ON s.id=e.signal_id + `SELECT e.*, s.source_id,s.title,s.summary,s.url,s.published_at, + src.publisher,src.source_class,src.independence_group,src.authority_tier, + ss.retrieved_at AS snapshot_retrieved_at + FROM evidence e + JOIN signals s ON s.id=e.signal_id + LEFT JOIN sources src ON src.id=s.source_id + LEFT JOIN source_snapshots ss ON ss.id=e.snapshot_id WHERE e.excerpt ILIKE $1 OR s.title ILIKE $1 OR s.summary ILIKE $1 ORDER BY e.retrieved_at DESC LIMIT $2`, [`%${query}%`, Math.min(Math.max(limit, 1), 100)] @@ -569,11 +695,20 @@ export async function getStoryThread(id: string) { COALESCE(n.source_id,s.source_id) AS provenance_source_id, e.source_url AS evidence_url, s.url AS signal_url, - m.title AS message_title + m.title AS message_title, + src.publisher AS provenance_publisher, + src.source_class AS provenance_source_class, + src.independence_group AS provenance_independence_group, + src.authority_tier AS provenance_authority_tier, + ss.id AS snapshot_id, + ss.retrieved_at AS snapshot_retrieved_at, + ss.published_at AS snapshot_published_at FROM story_nodes n LEFT JOIN evidence e ON e.id=n.evidence_id LEFT JOIN signals s ON s.id=COALESCE(n.signal_id,e.signal_id) LEFT JOIN agent_messages m ON m.id=n.message_id + LEFT JOIN sources src ON src.id=COALESCE(n.source_id,s.source_id) + LEFT JOIN source_snapshots ss ON ss.id=e.snapshot_id WHERE n.thread_id=$1 ORDER BY n.created_at ASC`, [id] @@ -765,6 +900,10 @@ export async function markStoryThreadStatus( if (!check.rowCount || !new Set(["found", "none_found"]).has(check.rows[0].result)) { throw new Error("Counter-evidence research must be completed before review/develop"); } + const sourceAudit = await auditStorySources(id); + if (!sourceAudit.robust) { + throw new Error("Story sourcing is not robust enough for review/develop; run story_source_audit for gaps"); + } } const result = await pool.query( @@ -782,6 +921,105 @@ export async function markStoryThreadStatus( return result.rows[0]; } + +export async function auditStorySources(threadId: string) { + const story = await getStoryThread(threadId); + if (!story) throw new Error("Story thread not found"); + + const nodes = Array.isArray(story.nodes) ? story.nodes : []; + 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) + ); + + const distinctSourceIds = Array.from(new Set( + evidenceBacked.map((n: any) => String(n.provenance_source_id)).filter(Boolean) + )); + const distinctGroups = Array.from(new Set( + evidenceBacked.map((n: any) => String(n.provenance_independence_group)).filter(Boolean) + )); + const publishers = Array.from(new Set( + evidenceBacked.map((n: any) => String(n.provenance_publisher)).filter(Boolean) + )); + const tierOneCount = evidenceBacked.filter((n: any) => + Number(n.provenance_authority_tier) === 1 + ).length; + + const evidentiaryTypes = new Set(["evidence","actual","forecast","trend","signal","metric"]); + const sourceTaggedButUnbacked = nodes + .filter((n: any) => evidentiaryTypes.has(n.node_type) && n.source_id && !n.evidence_id) + .map((n: any) => ({ id: n.id, title: n.title, node_type: n.node_type, source_id: n.source_id })); + + const claimTypes = new Set(["claim","counterclaim"]); + const evidenceNodeIds = new Set(evidenceBacked.map((n: any) => n.id)); + const supportingRelationships = new Set([ + "supports","contradicts","qualifies","forecast_for","actual_for","regional_breakdown_of" + ]); + + const unsupportedClaims = nodes + .filter((n: any) => claimTypes.has(n.node_type)) + .filter((claim: any) => !edges.some((edge: any) => + edge.to_node_id === claim.id && + evidenceNodeIds.has(edge.from_node_id) && + supportingRelationships.has(edge.relationship) + )) + .map((n: any) => ({ id: n.id, title: n.title, node_type: n.node_type })); + + const staleCutoff = Date.now() - 45 * 86_400_000; + const staleEvidence = evidenceBacked + .filter((n: any) => { + const value = n.snapshot_published_at ?? n.snapshot_retrieved_at; + const ts = value ? new Date(value).getTime() : NaN; + return Number.isFinite(ts) && ts < staleCutoff; + }) + .map((n: any) => ({ id: n.id, title: n.title, source_id: n.provenance_source_id })); + + const checks = { + at_least_two_independent_sources: distinctGroups.length >= 2, + at_least_two_evidence_backed_nodes: evidenceBacked.length >= 2, + at_least_one_tier_one_source: tierOneCount >= 1, + all_claims_have_direct_evidence: unsupportedClaims.length === 0, + no_source_tag_only_evidence_nodes: sourceTaggedButUnbacked.length === 0 + }; + + const robust = Object.values(checks).every(Boolean); + + return { + thread_id: threadId, + robust, + checks, + counts: { + evidence_backed_nodes: evidenceBacked.length, + distinct_sources: distinctSourceIds.length, + distinct_independence_groups: distinctGroups.length, + publishers: publishers.length, + tier_one_evidence_nodes: tierOneCount, + unsupported_claims: unsupportedClaims.length, + source_tag_only_nodes: sourceTaggedButUnbacked.length, + stale_evidence: staleEvidence.length + }, + sources: evidenceBacked.map((n: any) => ({ + node_id: n.id, + node_type: n.node_type, + title: n.title, + source_id: n.provenance_source_id, + publisher: n.provenance_publisher, + source_class: n.provenance_source_class, + independence_group: n.provenance_independence_group, + authority_tier: n.provenance_authority_tier, + evidence_id: n.evidence_id, + snapshot_id: n.snapshot_id, + snapshot_retrieved_at: n.snapshot_retrieved_at, + snapshot_published_at: n.snapshot_published_at, + url: n.evidence_url ?? n.signal_url + })), + unsupported_claims: unsupportedClaims, + source_tag_only_nodes: sourceTaggedButUnbacked, + stale_evidence: staleEvidence + }; +} + export async function seedInitialStoryThread() { const existing = await pool.query( "SELECT id FROM story_threads WHERE title=$1 LIMIT 1", diff --git a/src/mcp.ts b/src/mcp.ts index c1bd62a..d9d0550 100644 --- a/src/mcp.ts +++ b/src/mcp.ts @@ -2,6 +2,7 @@ import type { Request, Response } from "express"; import { acknowledgeAgentMessage, addStoryEdge, + auditStorySources, addStoryNode, completeCounterEvidenceCheck, createStoryThread, @@ -223,6 +224,16 @@ const tools = [ additionalProperties: false } }, + { + name: "story_source_audit", + description: "Audit a story thread for robust sourcing: immutable evidence snapshots, independent source coverage, unsupported claims, source-tag-only nodes and stale evidence.", + inputSchema: { + type: "object", + properties: { thread_id: { type: "string" } }, + required: ["thread_id"], + additionalProperties: false + } + }, { name: "story_counter_check_complete", description: "Record the mandatory counter-evidence pass for a story thread, including sources checked and whether material counter-evidence was found.", @@ -312,6 +323,7 @@ async function callTool(name: string, args: any, actor: ReturnType