Admin/media-intelligence-deploy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sync robust sourcing and story source audit

Browse Source
This commit is contained in:
Admin 2026-09-29 07:57:49 +00:00
parent 701ebce0d4
commit 6cdd27188a
4 changed files with 307 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

264
src/db.ts
View File

@ -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",

12
src/mcp.ts
View File

@ -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<typeof actorF
relationship: args.relationship,
reason: args.reason
}, actor));
if (name === "story_source_audit") return result(await auditStorySources(args.thread_id));
if (name === "story_counter_check_complete") return result(await completeCounterEvidenceCheck(args.thread_id, {
result: args.result,
sourcesChecked: args.sources_checked,

40
src/sources.ts
View File

@ -9,60 +9,100 @@ export const SOURCES: SourceDefinition[] = [
name: "GeoNet earthquakes",
type: "geojson",
url: "https://api.geonet.org.nz/quake?MMI=3"
publisher: "GeoNet",
sourceClass: "official_data",
independenceGroup: "geonet",
authorityTier: 1,
},
{
id: "parliament",
name: "NZ Parliament evidence and submissions",
type: "html",
url: "https://www3.parliament.nz/en/pb/sc/evidence-submissions/"
publisher: "New Zealand Parliament",
sourceClass: "official_primary",
independenceGroup: "parliament",
authorityTier: 1,
},
{
id: "stats-nz",
name: "Stats NZ information releases",
type: "sitemap",
url: "https://www.stats.govt.nz/sitemap.xml"
publisher: "Stats NZ",
sourceClass: "official_data",
independenceGroup: "stats-nz",
authorityTier: 1,
},
{
id: "data-govt-nz",
name: "data.govt.nz catalogue",
type: "ckan",
url: "https://catalogue.data.govt.nz/api/3/action/package_search?sort=metadata_modified%20desc&rows=25"
publisher: "New Zealand Government Data Catalogue",
sourceClass: "catalogue",
independenceGroup: "data-govt-nz",
authorityTier: 2,
},
{
id: "treasury",
name: "New Zealand Treasury publications",
type: "rss",
url: "https://www.treasury.govt.nz/feeds/publications"
publisher: "New Zealand Treasury",
sourceClass: "official_analysis",
independenceGroup: "treasury",
authorityTier: 1,
},
{
id: "rbnz",
name: "Reserve Bank of New Zealand statistics",
type: "html",
url: "https://www.rbnz.govt.nz/statistics/series/data-file-index-page"
publisher: "Reserve Bank of New Zealand",
sourceClass: "official_data",
independenceGroup: "rbnz",
authorityTier: 1,
},
{
id: "mbie-jobs-online",
name: "MBIE Jobs Online",
type: "html",
url: "https://www.mbie.govt.nz/business-and-employment/employment-and-skills/labour-market-reports-data-and-analysis/jobs-online"
publisher: "Ministry of Business, Innovation and Employment",
sourceClass: "official_data",
independenceGroup: "mbie",
authorityTier: 1,
},
{
id: "companies-office",
name: "Companies Office company statistics",
type: "html",
url: "https://www.companiesoffice.govt.nz/insights-and-articles/latest-company-statistics/"
publisher: "New Zealand Companies Office",
sourceClass: "official_registry",
independenceGroup: "companies-office",
authorityTier: 1,
},
{
id: "nz-gazette",
name: "New Zealand Gazette notices",
type: "html",
url: "https://gazette.govt.nz/home/search?from=0&sortField=publish_date&sortOrder=desc"
publisher: "New Zealand Gazette",
sourceClass: "official_notice",
independenceGroup: "nz-gazette",
authorityTier: 1,
},
{
id: "electricity-authority",
name: "Electricity Authority security of supply",
type: "html",
url: "https://www.ea.govt.nz/your-power/keeping-the-lights-on/electricity-security-of-supply/"
publisher: "Electricity Authority",
sourceClass: "official_data",
independenceGroup: "electricity-authority",
authorityTier: 1,
}
];

4
src/types.ts
View File

@ -5,6 +5,10 @@ export interface SourceDefinition {
name: string;
type: "geojson" | "html" | "sitemap" | "ckan" | "rss";
url: string;
publisher: string;
sourceClass: "official_primary" | "official_data" | "official_registry" | "official_notice" | "official_analysis" | "catalogue";
independenceGroup: string;
authorityTier: 1 | 2 | 3;
}
export interface NormalizedItem {