1961 lines
67 KiB
TypeScript
1961 lines
67 KiB
TypeScript
import pg from "pg";
|
|
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";
|
|
import { publishNewsroomEvent } from "./realtime.js";
|
|
|
|
const { Pool } = pg;
|
|
|
|
if (!process.env.DATABASE_URL) {
|
|
console.warn("DATABASE_URL is not set; database operations will fail until Arrowstack provisions it.");
|
|
}
|
|
|
|
export const pool = new Pool({
|
|
connectionString: process.env.DATABASE_URL,
|
|
ssl: process.env.DATABASE_SSL === "require" ? { rejectUnauthorized: false } : undefined,
|
|
max: Number(process.env.DB_POOL_MAX ?? 8)
|
|
});
|
|
|
|
export async function migrate() {
|
|
await pool.query(`
|
|
CREATE TABLE IF NOT EXISTS sources (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
source_type TEXT NOT NULL,
|
|
url TEXT NOT NULL,
|
|
enabled BOOLEAN NOT NULL DEFAULT TRUE,
|
|
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',
|
|
source_category TEXT NOT NULL DEFAULT 'other',
|
|
independence_group TEXT NOT NULL DEFAULT '',
|
|
authority_tier INTEGER NOT NULL DEFAULT 3,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS source_runs (
|
|
id TEXT PRIMARY KEY,
|
|
source_id TEXT NOT NULL REFERENCES sources(id),
|
|
started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
finished_at TIMESTAMPTZ,
|
|
status TEXT NOT NULL,
|
|
item_count INTEGER NOT NULL DEFAULT 0,
|
|
new_signal_count INTEGER NOT NULL DEFAULT 0,
|
|
error TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS signals (
|
|
id TEXT PRIMARY KEY,
|
|
source_id TEXT NOT NULL REFERENCES sources(id),
|
|
external_id TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
url TEXT NOT NULL,
|
|
published_at TIMESTAMPTZ,
|
|
detected_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
content_hash TEXT NOT NULL,
|
|
change_kind TEXT NOT NULL DEFAULT 'new',
|
|
topics JSONB NOT NULL DEFAULT '[]'::jsonb,
|
|
raw_json JSONB NOT NULL,
|
|
UNIQUE(source_id, external_id, content_hash)
|
|
);
|
|
|
|
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,
|
|
snapshot_kind TEXT NOT NULL DEFAULT 'normalized',
|
|
canonical_url TEXT,
|
|
document_text TEXT,
|
|
http_status INTEGER,
|
|
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 '',
|
|
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 (
|
|
id TEXT PRIMARY KEY,
|
|
signal_id TEXT NOT NULL UNIQUE REFERENCES signals(id) ON DELETE CASCADE,
|
|
status TEXT NOT NULL DEFAULT 'new',
|
|
score INTEGER NOT NULL DEFAULT 0,
|
|
reasons JSONB NOT NULL DEFAULT '[]'::jsonb,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_opportunities_status_created ON opportunities(status, created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS notes (
|
|
id TEXT PRIMARY KEY,
|
|
opportunity_id TEXT NOT NULL REFERENCES opportunities(id) ON DELETE CASCADE,
|
|
note TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
|
|
CREATE TABLE IF NOT EXISTS story_threads (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
question TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT 'researching',
|
|
created_by_sub TEXT NOT NULL,
|
|
created_by_name TEXT NOT NULL,
|
|
created_by_client_id TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_story_threads_status_updated
|
|
ON story_threads(status, updated_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS story_nodes (
|
|
id TEXT PRIMARY KEY,
|
|
thread_id TEXT NOT NULL REFERENCES story_threads(id) ON DELETE CASCADE,
|
|
node_type TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
source_id TEXT REFERENCES sources(id) ON DELETE SET NULL,
|
|
signal_id TEXT REFERENCES signals(id) ON DELETE SET NULL,
|
|
evidence_id TEXT REFERENCES evidence(id) ON DELETE SET NULL,
|
|
opportunity_id TEXT REFERENCES opportunities(id) ON DELETE SET NULL,
|
|
message_id TEXT,
|
|
confidence NUMERIC(4,3),
|
|
review_state TEXT NOT NULL DEFAULT 'active',
|
|
created_by_sub TEXT NOT NULL,
|
|
created_by_name TEXT NOT NULL,
|
|
created_by_client_id TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_story_nodes_thread
|
|
ON story_nodes(thread_id, created_at ASC);
|
|
|
|
CREATE TABLE IF NOT EXISTS story_edges (
|
|
id TEXT PRIMARY KEY,
|
|
thread_id TEXT NOT NULL REFERENCES story_threads(id) ON DELETE CASCADE,
|
|
from_node_id TEXT NOT NULL REFERENCES story_nodes(id) ON DELETE CASCADE,
|
|
to_node_id TEXT NOT NULL REFERENCES story_nodes(id) ON DELETE CASCADE,
|
|
relationship TEXT NOT NULL,
|
|
reason TEXT NOT NULL DEFAULT '',
|
|
created_by_sub TEXT NOT NULL,
|
|
created_by_name TEXT NOT NULL,
|
|
created_by_client_id TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
UNIQUE(thread_id, from_node_id, to_node_id, relationship)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_story_edges_thread
|
|
ON story_edges(thread_id, created_at ASC);
|
|
|
|
CREATE TABLE IF NOT EXISTS story_research_checks (
|
|
id TEXT PRIMARY KEY,
|
|
thread_id TEXT NOT NULL REFERENCES story_threads(id) ON DELETE CASCADE,
|
|
check_type TEXT NOT NULL DEFAULT 'counter_evidence',
|
|
result TEXT NOT NULL,
|
|
sources_checked JSONB NOT NULL DEFAULT '[]'::jsonb,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
completed_by_sub TEXT NOT NULL,
|
|
completed_by_name TEXT NOT NULL,
|
|
completed_by_client_id TEXT NOT NULL,
|
|
completed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
UNIQUE(thread_id, check_type)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS agent_messages (
|
|
id TEXT PRIMARY KEY,
|
|
actor_sub TEXT NOT NULL,
|
|
actor_name TEXT NOT NULL,
|
|
actor_client_id TEXT NOT NULL,
|
|
severity TEXT NOT NULL DEFAULT 'info',
|
|
kind TEXT NOT NULL DEFAULT 'update',
|
|
title TEXT NOT NULL,
|
|
body TEXT NOT NULL,
|
|
related_type TEXT,
|
|
related_id TEXT,
|
|
dedupe_key TEXT UNIQUE,
|
|
acknowledged_at TIMESTAMPTZ,
|
|
acknowledged_by TEXT,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_agent_messages_created
|
|
ON agent_messages(created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_agent_messages_unread
|
|
ON agent_messages(acknowledged_at, created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS app_meta (
|
|
key TEXT PRIMARY KEY,
|
|
value JSONB NOT NULL DEFAULT '{}'::jsonb,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS audit_events (
|
|
id TEXT PRIMARY KEY,
|
|
event_type TEXT NOT NULL,
|
|
object_type TEXT NOT NULL,
|
|
object_id TEXT NOT NULL,
|
|
payload JSONB NOT NULL DEFAULT '{}'::jsonb,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS source_trust_policy (
|
|
id TEXT PRIMARY KEY,
|
|
domain TEXT NOT NULL UNIQUE,
|
|
name TEXT NOT NULL,
|
|
trust_weight NUMERIC(4,3) NOT NULL CHECK (trust_weight >= 0 AND trust_weight <= 1),
|
|
independence_group TEXT NOT NULL,
|
|
source_class TEXT NOT NULL DEFAULT 'independent',
|
|
notes TEXT NOT NULL DEFAULT '',
|
|
enabled BOOLEAN NOT NULL DEFAULT TRUE,
|
|
source_id TEXT REFERENCES sources(id) ON DELETE SET NULL,
|
|
created_by TEXT NOT NULL DEFAULT 'system',
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_source_trust_policy_enabled_weight
|
|
ON source_trust_policy(enabled, trust_weight DESC);
|
|
`);
|
|
|
|
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 source_category TEXT NOT NULL DEFAULT 'other'");
|
|
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("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("ALTER TABLE story_nodes ADD COLUMN IF NOT EXISTS review_state TEXT NOT NULL DEFAULT 'active'");
|
|
|
|
const payloadMarker = await pool.query(
|
|
`INSERT INTO app_meta(key,value)
|
|
VALUES('evidence_payload_quality_v1','{"applied":true}'::jsonb)
|
|
ON CONFLICT(key) DO NOTHING
|
|
RETURNING key`
|
|
);
|
|
if (payloadMarker.rowCount) {
|
|
await pool.query(`
|
|
UPDATE source_snapshots ss
|
|
SET snapshot_kind='source_payload'
|
|
FROM sources src
|
|
WHERE ss.source_id=src.id
|
|
AND src.source_type IN ('rss','geojson')
|
|
AND src.authority_tier=1
|
|
AND ss.snapshot_kind='normalized'
|
|
`);
|
|
|
|
await pool.query(`
|
|
UPDATE evidence e
|
|
SET evidence_kind='source_payload',
|
|
locator=COALESCE(locator,'official feed/API item')
|
|
FROM source_snapshots ss, sources src
|
|
WHERE e.snapshot_id=ss.id
|
|
AND ss.source_id=src.id
|
|
AND ss.snapshot_kind='source_payload'
|
|
AND src.authority_tier=1
|
|
AND e.evidence_kind='normalized_excerpt'
|
|
`);
|
|
}
|
|
|
|
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)");
|
|
|
|
const marker = await pool.query(
|
|
`INSERT INTO app_meta(key,value)
|
|
VALUES('noise_baseline_v1','{"applied":true}'::jsonb)
|
|
ON CONFLICT(key) DO NOTHING
|
|
RETURNING key`
|
|
);
|
|
|
|
if (marker.rowCount) {
|
|
await pool.query(
|
|
`UPDATE opportunities
|
|
SET status='closed',
|
|
reasons = reasons || '["Archived during initial baseline cleanup"]'::jsonb,
|
|
updated_at=NOW()
|
|
WHERE status='new'`
|
|
);
|
|
}
|
|
}
|
|
|
|
export async function seedSources(sources: SourceDefinition[]) {
|
|
for (const source of sources) {
|
|
await pool.query(
|
|
`INSERT INTO sources(
|
|
id,name,source_type,url,publisher,source_class,source_category,independence_group,authority_tier
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
|
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,
|
|
source_category=EXCLUDED.source_category,
|
|
independence_group=EXCLUDED.independence_group,
|
|
authority_tier=EXCLUDED.authority_tier`,
|
|
[
|
|
source.id, source.name, source.type, source.url,
|
|
source.publisher, source.sourceClass, source.category, source.independenceGroup, source.authorityTier
|
|
]
|
|
);
|
|
}
|
|
|
|
await pool.query(
|
|
`UPDATE sources s
|
|
SET baseline_complete=TRUE
|
|
WHERE baseline_complete=FALSE
|
|
AND EXISTS (SELECT 1 FROM signals sig WHERE sig.source_id=s.id)`
|
|
);
|
|
}
|
|
|
|
|
|
function normaliseTrustDomain(value: string) {
|
|
const raw = String(value ?? "").trim().toLowerCase();
|
|
if (!raw) throw new Error("Source domain is required");
|
|
let hostname = raw;
|
|
try {
|
|
const url = new URL(raw.includes("://") ? raw : `https://${raw}`);
|
|
hostname = url.hostname.toLowerCase();
|
|
} catch {
|
|
throw new Error("Invalid source domain or URL");
|
|
}
|
|
hostname = hostname.replace(/^www\./, "").replace(/\.$/, "");
|
|
if (!hostname || !hostname.includes(".")) throw new Error("Invalid source domain");
|
|
return hostname;
|
|
}
|
|
|
|
function defaultTrustWeight(sourceClass: string, authorityTier: number) {
|
|
const byClass: Record<string, number> = {
|
|
official_primary: 1.0,
|
|
official_data: 0.98,
|
|
official_registry: 0.97,
|
|
official_notice: 0.95,
|
|
official_analysis: 0.93,
|
|
official_market_notice: 0.99,
|
|
official_market_data: 1.0,
|
|
corporate_primary: 0.95,
|
|
catalogue: 0.75
|
|
};
|
|
if (sourceClass in byClass) return byClass[sourceClass];
|
|
if (authorityTier === 1) return 0.92;
|
|
if (authorityTier === 2) return 0.78;
|
|
return 0.65;
|
|
}
|
|
|
|
function defaultTrustNote(source: SourceDefinition) {
|
|
if (source.sourceClass === "corporate_primary") {
|
|
return "Strong primary evidence of what the company announced or claimed; not independent corroboration of the company's performance or impact claims.";
|
|
}
|
|
if (source.sourceClass === "official_market_notice") {
|
|
return "Primary exchange-hosted issuer disclosure. Strong evidence that the issuer made the disclosure; not an independent assessment of the issuer's claims.";
|
|
}
|
|
if (source.sourceClass === "official_market_data") {
|
|
return "Direct market-data source. Strong for observed price, volume and market-state facts within the feed's licensing and timing constraints.";
|
|
}
|
|
return "Seeded from configured ingestion source. Weight is an editorial prior, not a truth guarantee.";
|
|
}
|
|
|
|
export async function seedSourceTrustPolicy(sources: SourceDefinition[]) {
|
|
for (const source of sources) {
|
|
const domain = normaliseTrustDomain(source.url);
|
|
const weight = defaultTrustWeight(source.sourceClass, source.authorityTier);
|
|
await pool.query(
|
|
`INSERT INTO source_trust_policy(
|
|
id,domain,name,trust_weight,independence_group,source_class,notes,
|
|
enabled,source_id,created_by
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,TRUE,$8,'system-seed')
|
|
ON CONFLICT(domain) DO UPDATE SET
|
|
source_id=COALESCE(source_trust_policy.source_id,EXCLUDED.source_id),
|
|
name=CASE WHEN source_trust_policy.created_by='system-seed' THEN EXCLUDED.name ELSE source_trust_policy.name END,
|
|
independence_group=CASE WHEN source_trust_policy.created_by='system-seed' THEN EXCLUDED.independence_group ELSE source_trust_policy.independence_group END,
|
|
source_class=CASE WHEN source_trust_policy.created_by='system-seed' THEN EXCLUDED.source_class ELSE source_trust_policy.source_class END,
|
|
updated_at=NOW()`,
|
|
[
|
|
randomUUID(),
|
|
domain,
|
|
source.publisher || source.name,
|
|
weight,
|
|
source.independenceGroup || domain,
|
|
source.sourceClass,
|
|
defaultTrustNote(source),
|
|
source.id
|
|
]
|
|
);
|
|
}
|
|
}
|
|
|
|
export async function listSourceTrustPolicy(args: { enabledOnly?: boolean } = {}) {
|
|
const result = await pool.query(
|
|
`SELECT *
|
|
FROM source_trust_policy
|
|
${args.enabledOnly ? "WHERE enabled=TRUE" : ""}
|
|
ORDER BY trust_weight DESC, name ASC`
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
export async function upsertSourceTrustEntry(
|
|
args: {
|
|
domain: string;
|
|
name: string;
|
|
trustWeight: number;
|
|
independenceGroup?: string;
|
|
sourceClass?: string;
|
|
notes?: string;
|
|
enabled?: boolean;
|
|
},
|
|
actorLabel: string
|
|
) {
|
|
const domain = normaliseTrustDomain(args.domain);
|
|
const name = String(args.name ?? "").trim();
|
|
const weight = Number(args.trustWeight);
|
|
const independenceGroup = String(args.independenceGroup ?? domain).trim().toLowerCase();
|
|
const sourceClass = String(args.sourceClass ?? "independent").trim().toLowerCase();
|
|
const notes = String(args.notes ?? "").trim();
|
|
|
|
if (!name) throw new Error("Source name is required");
|
|
if (!Number.isFinite(weight) || weight < 0 || weight > 1) {
|
|
throw new Error("Trust weight must be between 0 and 1");
|
|
}
|
|
if (!independenceGroup) throw new Error("Independence group is required");
|
|
if (name.length > 180 || notes.length > 4000) throw new Error("Source trust entry is too long");
|
|
|
|
const existing = await pool.query(
|
|
"SELECT id FROM source_trust_policy WHERE domain=$1",
|
|
[domain]
|
|
);
|
|
const id = existing.rows[0]?.id ?? randomUUID();
|
|
|
|
const result = await pool.query(
|
|
`INSERT INTO source_trust_policy(
|
|
id,domain,name,trust_weight,independence_group,source_class,notes,
|
|
enabled,created_by
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
|
ON CONFLICT(domain) DO UPDATE SET
|
|
name=EXCLUDED.name,
|
|
trust_weight=EXCLUDED.trust_weight,
|
|
independence_group=EXCLUDED.independence_group,
|
|
source_class=EXCLUDED.source_class,
|
|
notes=EXCLUDED.notes,
|
|
enabled=EXCLUDED.enabled,
|
|
created_by=EXCLUDED.created_by,
|
|
updated_at=NOW()
|
|
RETURNING *`,
|
|
[
|
|
id,domain,name,weight,independenceGroup,sourceClass,notes,
|
|
args.enabled !== false,actorLabel
|
|
]
|
|
);
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'source_trust.updated','source_trust',$2,$3::jsonb)`,
|
|
[randomUUID(), result.rows[0].id, JSON.stringify({
|
|
domain, trustWeight: weight, independenceGroup, actor: actorLabel
|
|
})]
|
|
);
|
|
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("source_trust.updated", { entry: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function evaluateSourceTrust(
|
|
inputs: Array<{ url?: string; domain?: string; sourceId?: string }>
|
|
) {
|
|
if (!Array.isArray(inputs) || !inputs.length) {
|
|
throw new Error("At least one source is required");
|
|
}
|
|
|
|
const policy = await listSourceTrustPolicy({ enabledOnly: true });
|
|
const bySourceId = new Map(
|
|
policy.filter((p: any) => p.source_id).map((p: any) => [String(p.source_id), p])
|
|
);
|
|
|
|
const matched = inputs.slice(0, 50).map((input) => {
|
|
let entry: any = null;
|
|
let domain: string | null = null;
|
|
|
|
if (input.sourceId && bySourceId.has(String(input.sourceId))) {
|
|
entry = bySourceId.get(String(input.sourceId));
|
|
domain = String(entry.domain);
|
|
} else {
|
|
const value = input.url || input.domain;
|
|
if (value) {
|
|
try { domain = normaliseTrustDomain(value); } catch { domain = null; }
|
|
}
|
|
if (domain) {
|
|
entry = policy
|
|
.filter((p: any) => domain === p.domain || domain!.endsWith(`.${p.domain}`))
|
|
.sort((a: any, b: any) => String(b.domain).length - String(a.domain).length)[0] ?? null;
|
|
}
|
|
}
|
|
|
|
return {
|
|
input,
|
|
matched: Boolean(entry),
|
|
domain,
|
|
policy_id: entry?.id ?? null,
|
|
name: entry?.name ?? null,
|
|
weight: entry ? Number(entry.trust_weight) : null,
|
|
independence_group: entry?.independence_group ?? null,
|
|
source_class: entry?.source_class ?? null
|
|
};
|
|
});
|
|
|
|
const groups = new Map<string, any>();
|
|
for (const item of matched.filter((x) => x.matched)) {
|
|
const group = String(item.independence_group || item.domain);
|
|
const current = groups.get(group);
|
|
if (!current || Number(item.weight) > Number(current.weight)) groups.set(group, item);
|
|
}
|
|
|
|
const independent = Array.from(groups.values()).sort(
|
|
(a, b) => Number(b.weight) - Number(a.weight)
|
|
);
|
|
const strongest = independent.length ? Number(independent[0].weight) : 0;
|
|
const corroborationBonus = independent
|
|
.slice(1)
|
|
.reduce((sum, item) => sum + Number(item.weight) * 0.05, 0);
|
|
const effectiveScore = Math.min(1, strongest + Math.min(0.15, corroborationBonus));
|
|
const independentCount = independent.length;
|
|
|
|
let confidenceBand = "unrated";
|
|
if (independentCount >= 2 && effectiveScore >= 0.9) confidenceBand = "strong";
|
|
else if (independentCount >= 2 && effectiveScore >= 0.75) confidenceBand = "moderate";
|
|
else if (independentCount === 1 && strongest >= 0.85) confidenceBand = "strong_single_source";
|
|
else if (independentCount >= 1) confidenceBand = "limited";
|
|
|
|
return {
|
|
policy_version: "source-trust-v1",
|
|
methodology: {
|
|
description: "Use the strongest configured source from each independence group. Repeated sources from the same publisher/group do not add corroboration. Additional independent groups contribute a small capped corroboration bonus.",
|
|
strongest_source_weight: strongest,
|
|
corroboration_bonus: Number(Math.min(0.15, corroborationBonus).toFixed(3)),
|
|
maximum_independence_bonus: 0.15
|
|
},
|
|
effective_score: Number(effectiveScore.toFixed(3)),
|
|
confidence_band: confidenceBand,
|
|
independent_source_count: independentCount,
|
|
meets_independence_minimum: independentCount >= 2,
|
|
policy_coverage: Number((matched.filter((x) => x.matched).length / matched.length).toFixed(3)),
|
|
independent_groups: independent.map((x) => ({
|
|
group: x.independence_group || x.domain,
|
|
representative: x.name,
|
|
weight: x.weight,
|
|
domain: x.domain
|
|
})),
|
|
sources: matched,
|
|
guidance: independentCount >= 2
|
|
? "Independent corroboration present. Evaluate whether the sources genuinely support the same claim and check for shared upstream evidence."
|
|
: "Independent corroboration is not yet present. Seek at least one source from a different independence group before treating the claim as robust."
|
|
};
|
|
}
|
|
|
|
|
|
export function contentHash(item: NormalizedItem) {
|
|
return createHash("sha256")
|
|
.update(JSON.stringify({
|
|
title: item.title,
|
|
summary: item.summary,
|
|
url: item.url,
|
|
publishedAt: item.publishedAt,
|
|
raw: item.raw
|
|
}))
|
|
.digest("hex");
|
|
}
|
|
|
|
export async function storeItem(
|
|
item: NormalizedItem,
|
|
scored: ScoredItem,
|
|
options: { baseline?: boolean } = {}
|
|
): Promise<{
|
|
created: boolean;
|
|
signalId?: string;
|
|
opportunityId?: string;
|
|
changeKind?: "baseline" | "new" | "updated";
|
|
}> {
|
|
const hash = contentHash(item);
|
|
const previous = await pool.query(
|
|
`SELECT id,content_hash
|
|
FROM signals
|
|
WHERE source_id=$1 AND external_id=$2
|
|
ORDER BY detected_at DESC
|
|
LIMIT 1`,
|
|
[item.sourceId, item.externalId]
|
|
);
|
|
|
|
if (previous.rowCount && previous.rows[0].content_hash === hash) {
|
|
return { created: false };
|
|
}
|
|
|
|
const changeKind: "baseline" | "new" | "updated" =
|
|
options.baseline ? "baseline" : previous.rowCount ? "updated" : "new";
|
|
|
|
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);
|
|
|
|
let recentEnough = true;
|
|
if (item.publishedAt) {
|
|
const published = new Date(item.publishedAt).getTime();
|
|
if (!Number.isNaN(published)) {
|
|
recentEnough = Date.now() - published <= maxAgeDays * 86_400_000;
|
|
}
|
|
}
|
|
|
|
const sourceMeta = await pool.query(
|
|
"SELECT source_type,authority_tier,source_class FROM sources WHERE id=$1",
|
|
[item.sourceId]
|
|
);
|
|
const sourceClass = String(sourceMeta.rows[0]?.source_class ?? "");
|
|
const suppressNormalizationUpdate =
|
|
changeKind === "updated" &&
|
|
new Set(["corporate_primary","official_market_notice"]).has(sourceClass);
|
|
|
|
const shouldCreateOpportunity =
|
|
!options.baseline &&
|
|
!suppressNormalizationUpdate &&
|
|
recentEnough &&
|
|
scored.score >= minScore;
|
|
|
|
const directPayload =
|
|
Number(sourceMeta.rows[0]?.authority_tier) === 1 &&
|
|
new Set(["rss","geojson"]).has(String(sourceMeta.rows[0]?.source_type ?? ""));
|
|
const snapshotKind = directPayload ? "source_payload" : "normalized";
|
|
const evidenceKind = directPayload ? "source_payload" : "normalized_excerpt";
|
|
const locator = directPayload ? "official feed/API item" : "title+summary";
|
|
|
|
await pool.query("BEGIN");
|
|
try {
|
|
await pool.query(
|
|
`INSERT INTO signals(
|
|
id,source_id,external_id,title,summary,url,published_at,content_hash,change_kind,topics,raw_json
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,$11::jsonb)`,
|
|
[
|
|
signalId,
|
|
item.sourceId,
|
|
item.externalId,
|
|
item.title,
|
|
item.summary,
|
|
item.url,
|
|
item.publishedAt,
|
|
hash,
|
|
changeKind,
|
|
JSON.stringify(item.topics),
|
|
JSON.stringify(item.raw)
|
|
]
|
|
);
|
|
|
|
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 source_snapshots(
|
|
id,source_id,signal_id,source_url,published_at,content_hash,
|
|
content_type,byte_length,snapshot_json,snapshot_kind
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10)`,
|
|
[
|
|
snapshotId,
|
|
item.sourceId,
|
|
signalId,
|
|
item.url,
|
|
item.publishedAt,
|
|
hash,
|
|
directPayload
|
|
? "application/vnd.media-intelligence.source-payload+json"
|
|
: "application/vnd.media-intelligence.normalized+json",
|
|
Buffer.byteLength(snapshotText),
|
|
snapshotText,
|
|
snapshotKind
|
|
]
|
|
);
|
|
|
|
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,$7,$8)`,
|
|
[
|
|
evidenceId,
|
|
signalId,
|
|
item.url,
|
|
hash,
|
|
`${item.title}\n${item.summary}`.slice(0, 2000),
|
|
snapshotId,
|
|
evidenceKind,
|
|
locator
|
|
]
|
|
);
|
|
|
|
if (shouldCreateOpportunity) {
|
|
await pool.query(
|
|
`INSERT INTO opportunities(id,signal_id,status,score,reasons)
|
|
VALUES($1,$2,'new',$3,$4::jsonb)`,
|
|
[opportunityId, signalId, scored.score, JSON.stringify(scored.reasons)]
|
|
);
|
|
}
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'signal.ingested','signal',$2,$3::jsonb)`,
|
|
[
|
|
randomUUID(),
|
|
signalId,
|
|
JSON.stringify({
|
|
sourceId: item.sourceId,
|
|
externalId: item.externalId,
|
|
score: scored.score,
|
|
changeKind,
|
|
opportunityCreated: shouldCreateOpportunity,
|
|
recentEnough
|
|
})
|
|
]
|
|
);
|
|
|
|
await pool.query("COMMIT");
|
|
|
|
return {
|
|
created: true,
|
|
signalId,
|
|
opportunityId: shouldCreateOpportunity ? opportunityId : undefined,
|
|
changeKind
|
|
};
|
|
} catch (error) {
|
|
await pool.query("ROLLBACK");
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
export async function listSources() {
|
|
const result = await pool.query("SELECT * FROM sources ORDER BY source_category, name");
|
|
return result.rows;
|
|
}
|
|
|
|
export async function sourceStatus() {
|
|
const result = await pool.query(`
|
|
SELECT s.*,
|
|
(SELECT COUNT(DISTINCT sig.external_id)::int
|
|
FROM signals sig
|
|
WHERE sig.source_id=s.id) AS signal_count,
|
|
(SELECT COUNT(*)::int
|
|
FROM opportunities o
|
|
JOIN signals sig ON sig.id=o.signal_id
|
|
WHERE sig.source_id=s.id
|
|
AND o.status <> 'closed') AS opportunity_count
|
|
FROM sources s
|
|
ORDER BY s.source_category, s.name
|
|
`);
|
|
return result.rows;
|
|
}
|
|
|
|
export async function listOpportunities(args: { status?: string; since?: string; limit?: number }) {
|
|
const limit = Math.min(Math.max(Number(args.limit ?? 25), 1), 100);
|
|
const base = `SELECT o.id,o.status,o.score,o.reasons,o.created_at,o.updated_at,
|
|
s.id AS signal_id,s.source_id,s.title,s.summary,s.url,s.published_at,s.detected_at,s.topics
|
|
FROM opportunities o
|
|
JOIN signals s ON s.id=o.signal_id`;
|
|
|
|
if (args.status && args.since) {
|
|
const result = await pool.query(
|
|
`${base}
|
|
WHERE o.status=$1 AND o.created_at >= $2::timestamptz
|
|
ORDER BY o.created_at DESC
|
|
LIMIT $3`,
|
|
[args.status, args.since, limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
if (args.status) {
|
|
const result = await pool.query(
|
|
`${base}
|
|
WHERE o.status=$1
|
|
ORDER BY o.created_at DESC
|
|
LIMIT $2`,
|
|
[args.status, limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
if (args.since) {
|
|
const result = await pool.query(
|
|
`${base}
|
|
WHERE o.status <> 'closed' AND o.created_at >= $1::timestamptz
|
|
ORDER BY o.created_at DESC
|
|
LIMIT $2`,
|
|
[args.since, limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
const result = await pool.query(
|
|
`${base}
|
|
WHERE o.status <> 'closed'
|
|
ORDER BY o.created_at DESC
|
|
LIMIT $1`,
|
|
[limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
export async function getOpportunity(id: string) {
|
|
const result = await pool.query(
|
|
`SELECT o.*, s.source_id,s.external_id,s.title,s.summary,s.url,s.published_at,s.detected_at,s.topics,s.raw_json
|
|
FROM opportunities o JOIN signals s ON s.id=o.signal_id WHERE o.id=$1`,
|
|
[id]
|
|
);
|
|
if (!result.rowCount) return null;
|
|
const row = result.rows[0];
|
|
const evidence = await pool.query("SELECT * FROM evidence WHERE signal_id=$1 ORDER BY retrieved_at DESC", [row.signal_id]);
|
|
const notes = await pool.query("SELECT * FROM notes WHERE opportunity_id=$1 ORDER BY created_at ASC", [id]);
|
|
return { ...row, evidence: evidence.rows, notes: notes.rows };
|
|
}
|
|
|
|
export async function markOpportunity(id: string, status: string) {
|
|
const allowed = new Set(["new", "watch", "investigate", "develop", "ignore", "closed"]);
|
|
if (!allowed.has(status)) throw new Error(`Invalid status: ${status}`);
|
|
const result = await pool.query(
|
|
"UPDATE opportunities SET status=$2, updated_at=NOW() WHERE id=$1 RETURNING *",
|
|
[id, status]
|
|
);
|
|
if (!result.rowCount) return null;
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'opportunity.status_changed','opportunity',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ status })]
|
|
);
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("opportunity.updated", { opportunity: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
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,
|
|
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;
|
|
}
|
|
|
|
export async function searchEvidence(query: string, limit = 20) {
|
|
const result = await pool.query(
|
|
`WITH latest_signal AS (
|
|
SELECT DISTINCT ON (source_id,external_id) id
|
|
FROM signals
|
|
ORDER BY source_id,external_id,detected_at DESC
|
|
)
|
|
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 latest_signal ls
|
|
JOIN signals s ON s.id=ls.id
|
|
JOIN evidence e ON e.signal_id=s.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)]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
|
|
export interface StoryActor {
|
|
sub: string;
|
|
name: string;
|
|
clientId: string;
|
|
}
|
|
|
|
const STORY_NODE_TYPES = new Set([
|
|
"signal", "evidence", "claim", "counterclaim", "trend", "forecast",
|
|
"actual", "question", "context", "organisation", "person", "place", "metric"
|
|
]);
|
|
|
|
const STORY_RELATIONSHIPS = new Set([
|
|
"supports", "contradicts", "qualifies", "explains", "correlates_with",
|
|
"precedes", "updates", "supersedes", "same_subject",
|
|
"regional_breakdown_of", "forecast_for", "actual_for", "investigates"
|
|
]);
|
|
|
|
const STORY_STATUSES = new Set(["researching", "review", "develop", "closed"]);
|
|
|
|
export async function createStoryThread(
|
|
args: { title: string; question?: string },
|
|
actor: StoryActor
|
|
) {
|
|
const title = String(args.title ?? "").trim();
|
|
const question = String(args.question ?? "").trim();
|
|
if (!title) throw new Error("Story thread title is required");
|
|
if (title.length > 220) throw new Error("Story thread title is too long");
|
|
if (question.length > 1200) throw new Error("Story thread question is too long");
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_threads(
|
|
id,title,question,status,created_by_sub,created_by_name,created_by_client_id
|
|
) VALUES($1,$2,$3,'researching',$4,$5,$6)
|
|
RETURNING *`,
|
|
[id, title, question, actor.sub, actor.name, actor.clientId]
|
|
);
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'story_thread.created','story_thread',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ actor, title, question })]
|
|
);
|
|
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("story_thread.created", { story: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function listStoryThreads(args: {
|
|
status?: string;
|
|
limit?: number;
|
|
} = {}) {
|
|
const limit = Math.min(Math.max(Number(args.limit ?? 30), 1), 100);
|
|
|
|
if (args.status) {
|
|
if (!STORY_STATUSES.has(args.status)) throw new Error("Invalid story status");
|
|
const result = await pool.query(
|
|
`SELECT t.*,
|
|
(SELECT COUNT(*)::int FROM story_nodes n WHERE n.thread_id=t.id) AS node_count,
|
|
(SELECT COUNT(*)::int FROM story_edges e WHERE e.thread_id=t.id) AS edge_count,
|
|
(SELECT result FROM story_research_checks c
|
|
WHERE c.thread_id=t.id AND c.check_type='counter_evidence'
|
|
LIMIT 1) AS counter_evidence_result
|
|
FROM story_threads t
|
|
WHERE t.status=$1
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT $2`,
|
|
[args.status, limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
const result = await pool.query(
|
|
`SELECT t.*,
|
|
(SELECT COUNT(*)::int FROM story_nodes n WHERE n.thread_id=t.id) AS node_count,
|
|
(SELECT COUNT(*)::int FROM story_edges e WHERE e.thread_id=t.id) AS edge_count,
|
|
(SELECT result FROM story_research_checks c
|
|
WHERE c.thread_id=t.id AND c.check_type='counter_evidence'
|
|
LIMIT 1) AS counter_evidence_result
|
|
FROM story_threads t
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT $1`,
|
|
[limit]
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
function buildStorySourceAudit(story: any) {
|
|
const allNodes = Array.isArray(story.nodes) ? story.nodes : [];
|
|
const nodes = allNodes.filter((n: any) => n.review_state !== "irrelevant");
|
|
const nodeIds = new Set(nodes.map((n: any) => n.id));
|
|
const edges = (Array.isArray(story.edges) ? story.edges : [])
|
|
.filter((e: any) => nodeIds.has(e.from_node_id) && nodeIds.has(e.to_node_id));
|
|
|
|
const evidenceBacked = nodes.filter((n: any) => {
|
|
if (!n.evidence_id || !n.snapshot_id || !n.provenance_source_id) return false;
|
|
|
|
const documentBacked =
|
|
n.snapshot_kind === "source_document" &&
|
|
n.evidence_kind === "source_document" &&
|
|
Number(n.snapshot_text_length ?? 0) >= 500;
|
|
|
|
const payloadBacked =
|
|
n.snapshot_kind === "source_payload" &&
|
|
n.evidence_kind === "source_payload" &&
|
|
Number(n.provenance_authority_tier) === 1 &&
|
|
Number(n.snapshot_payload_length ?? 0) >= 350 &&
|
|
Number(n.evidence_excerpt_length ?? 0) >= 160;
|
|
|
|
return documentBacked || payloadBacked;
|
|
});
|
|
|
|
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 directEvidenceTypes = new Set(["evidence","actual","signal"]);
|
|
const sourceTaggedButUnbacked = nodes
|
|
.filter((n: any) => directEvidenceTypes.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 assertionTypes = new Set(["claim","counterclaim","forecast","trend","metric"]);
|
|
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) => assertionTypes.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_assertions_have_direct_evidence: unsupportedClaims.length === 0,
|
|
no_source_tag_only_direct_evidence_nodes: sourceTaggedButUnbacked.length === 0
|
|
};
|
|
|
|
return {
|
|
robust: Object.values(checks).every(Boolean),
|
|
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,
|
|
evidence_kind: n.evidence_kind,
|
|
snapshot_kind: n.snapshot_kind,
|
|
snapshot_text_length: n.snapshot_text_length,
|
|
snapshot_payload_length: n.snapshot_payload_length,
|
|
evidence_excerpt_length: n.evidence_excerpt_length,
|
|
url: n.evidence_url ?? n.signal_url
|
|
})),
|
|
unsupported_claims: unsupportedClaims,
|
|
source_tag_only_nodes: sourceTaggedButUnbacked,
|
|
stale_evidence: staleEvidence
|
|
};
|
|
}
|
|
|
|
function buildStoryResearchChecklist(story: any, sourceAudit: any) {
|
|
const nodes = Array.isArray(story.nodes) ? story.nodes : [];
|
|
const activeNodes = nodes.filter((n: any) => n.review_state !== "irrelevant");
|
|
const counterDone = (story.research_checks ?? []).some(
|
|
(c: any) => c.check_type === "counter_evidence" && ["found","none_found"].includes(c.result)
|
|
);
|
|
const unresolvedQuestions = activeNodes.filter(
|
|
(n: any) => n.node_type === "question" && n.review_state !== "resolved"
|
|
);
|
|
|
|
const items = [
|
|
{
|
|
key: "primary_evidence",
|
|
label: "Primary/substantive evidence captured",
|
|
done: Number(sourceAudit?.counts?.evidence_backed_nodes ?? 0) >= 1
|
|
},
|
|
{
|
|
key: "counter_evidence",
|
|
label: "Counter-evidence pass completed",
|
|
done: counterDone
|
|
},
|
|
{
|
|
key: "independent_corroboration",
|
|
label: "At least two independent source groups",
|
|
done: Number(sourceAudit?.counts?.distinct_independence_groups ?? 0) >= 2
|
|
},
|
|
{
|
|
key: "assertions_supported",
|
|
label: "All active assertions connected to substantive evidence",
|
|
done: Number(sourceAudit?.counts?.unsupported_claims ?? 0) === 0
|
|
},
|
|
{
|
|
key: "questions_resolved",
|
|
label: "Open research questions resolved or explicitly parked",
|
|
done: unresolvedQuestions.length === 0
|
|
}
|
|
];
|
|
|
|
return {
|
|
ready_for_editorial: Boolean(sourceAudit?.robust && counterDone),
|
|
completed: items.filter((item) => item.done).length,
|
|
total: items.length,
|
|
unresolved_questions: unresolvedQuestions.map((n: any) => ({
|
|
id: n.id,
|
|
title: n.title,
|
|
summary: n.summary
|
|
})),
|
|
items
|
|
};
|
|
}
|
|
|
|
export async function getStoryThread(id: string) {
|
|
const threadResult = await pool.query(
|
|
`SELECT t.*,
|
|
(SELECT result FROM story_research_checks c
|
|
WHERE c.thread_id=t.id AND c.check_type='counter_evidence'
|
|
LIMIT 1) AS counter_evidence_result
|
|
FROM story_threads t
|
|
WHERE t.id=$1`,
|
|
[id]
|
|
);
|
|
if (!threadResult.rowCount) return null;
|
|
|
|
const [nodes, edges, checks, requests] = await Promise.all([
|
|
pool.query(
|
|
`SELECT n.*,
|
|
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,
|
|
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,
|
|
e.evidence_kind AS evidence_kind,
|
|
e.locator AS evidence_locator,
|
|
e.excerpt AS evidence_excerpt,
|
|
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,
|
|
LENGTH(COALESCE(ss.document_text,''))::int AS snapshot_text_length,
|
|
LENGTH(COALESCE(ss.snapshot_json::text,''))::int AS snapshot_payload_length,
|
|
LENGTH(COALESCE(e.excerpt,''))::int AS evidence_excerpt_length
|
|
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]
|
|
),
|
|
pool.query(
|
|
"SELECT * FROM story_edges WHERE thread_id=$1 ORDER BY created_at ASC",
|
|
[id]
|
|
),
|
|
pool.query(
|
|
"SELECT * FROM story_research_checks WHERE thread_id=$1 ORDER BY completed_at DESC",
|
|
[id]
|
|
),
|
|
pool.query(
|
|
"SELECT * FROM agent_messages WHERE related_type='story' AND related_id=$1 ORDER BY created_at DESC",
|
|
[id]
|
|
)
|
|
]);
|
|
|
|
const story = {
|
|
...threadResult.rows[0],
|
|
nodes: nodes.rows,
|
|
edges: edges.rows,
|
|
research_checks: checks.rows,
|
|
research_requests: requests.rows
|
|
};
|
|
|
|
const sourceAudit = buildStorySourceAudit(story);
|
|
return {
|
|
...story,
|
|
source_audit: sourceAudit,
|
|
research_checklist: buildStoryResearchChecklist(story, sourceAudit)
|
|
};
|
|
}
|
|
|
|
export async function addStoryNode(
|
|
threadId: string,
|
|
args: {
|
|
nodeType: string;
|
|
title: string;
|
|
summary?: string;
|
|
sourceId?: string | null;
|
|
signalId?: string | null;
|
|
evidenceId?: string | null;
|
|
opportunityId?: string | null;
|
|
messageId?: string | null;
|
|
confidence?: number | null;
|
|
},
|
|
actor: StoryActor
|
|
) {
|
|
const nodeType = String(args.nodeType ?? "");
|
|
if (!STORY_NODE_TYPES.has(nodeType)) throw new Error("Invalid story node type");
|
|
|
|
const title = String(args.title ?? "").trim();
|
|
const summary = String(args.summary ?? "").trim();
|
|
if (!title) throw new Error("Story node title is required");
|
|
if (title.length > 240) throw new Error("Story node title is too long");
|
|
if (summary.length > 8000) throw new Error("Story node summary is too long");
|
|
|
|
const confidence =
|
|
args.confidence == null ? null : Number(args.confidence);
|
|
if (confidence != null && (!Number.isFinite(confidence) || confidence < 0 || confidence > 1)) {
|
|
throw new Error("confidence must be between 0 and 1");
|
|
}
|
|
|
|
const thread = await pool.query("SELECT id FROM story_threads WHERE id=$1", [threadId]);
|
|
if (!thread.rowCount) throw new Error("Story thread not found");
|
|
|
|
if (args.evidenceId) {
|
|
await enrichEvidenceDocument(args.evidenceId);
|
|
}
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_nodes(
|
|
id,thread_id,node_type,title,summary,source_id,signal_id,evidence_id,
|
|
opportunity_id,message_id,confidence,
|
|
created_by_sub,created_by_name,created_by_client_id
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
|
|
RETURNING *`,
|
|
[
|
|
id, threadId, nodeType, title, summary,
|
|
args.sourceId ?? null,
|
|
args.signalId ?? null,
|
|
args.evidenceId ?? null,
|
|
args.opportunityId ?? null,
|
|
args.messageId ?? null,
|
|
confidence,
|
|
actor.sub, actor.name, actor.clientId
|
|
]
|
|
);
|
|
|
|
await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]);
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'story_node.added','story_node',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ threadId, nodeType, actor })]
|
|
);
|
|
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("story_node.added", { node: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function setStoryNodeReviewState(
|
|
threadId: string,
|
|
nodeId: string,
|
|
reviewState: string,
|
|
actor: StoryActor
|
|
) {
|
|
const allowed = new Set(["active","resolved","irrelevant"]);
|
|
if (!allowed.has(reviewState)) throw new Error("Invalid story node review state");
|
|
|
|
const result = await pool.query(
|
|
`UPDATE story_nodes
|
|
SET review_state=$3
|
|
WHERE id=$2 AND thread_id=$1
|
|
RETURNING *`,
|
|
[threadId,nodeId,reviewState]
|
|
);
|
|
if (!result.rowCount) throw new Error("Story node not found");
|
|
|
|
await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]);
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'story_node.review_state_changed','story_node',$2,$3::jsonb)`,
|
|
[randomUUID(), nodeId, JSON.stringify({ threadId, reviewState, actor })]
|
|
);
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("story_node.updated", { node: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function addStoryEdge(
|
|
threadId: string,
|
|
args: {
|
|
fromNodeId: string;
|
|
toNodeId: string;
|
|
relationship: string;
|
|
reason?: string;
|
|
},
|
|
actor: StoryActor
|
|
) {
|
|
const relationship = String(args.relationship ?? "");
|
|
if (!STORY_RELATIONSHIPS.has(relationship)) throw new Error("Invalid story relationship");
|
|
if (args.fromNodeId === args.toNodeId) throw new Error("Story edge cannot point to itself");
|
|
|
|
const nodeCheck = await pool.query(
|
|
`SELECT id FROM story_nodes
|
|
WHERE thread_id=$1 AND id = ANY($2::text[])`,
|
|
[threadId, [args.fromNodeId, args.toNodeId]]
|
|
);
|
|
if (nodeCheck.rowCount !== 2) {
|
|
throw new Error("Both story nodes must belong to this thread");
|
|
}
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_edges(
|
|
id,thread_id,from_node_id,to_node_id,relationship,reason,
|
|
created_by_sub,created_by_name,created_by_client_id
|
|
)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
|
ON CONFLICT(thread_id,from_node_id,to_node_id,relationship)
|
|
DO UPDATE SET reason=EXCLUDED.reason
|
|
RETURNING *`,
|
|
[
|
|
id, threadId, args.fromNodeId, args.toNodeId, relationship,
|
|
String(args.reason ?? "").slice(0, 2000),
|
|
actor.sub, actor.name, actor.clientId
|
|
]
|
|
);
|
|
|
|
await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]);
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("story_edge.updated", { edge: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function completeCounterEvidenceCheck(
|
|
threadId: string,
|
|
args: { result: string; sourcesChecked: string[]; summary: string },
|
|
actor: StoryActor
|
|
) {
|
|
const resultValue = String(args.result ?? "");
|
|
if (!new Set(["found", "none_found"]).has(resultValue)) {
|
|
throw new Error("Counter-evidence result must be found or none_found");
|
|
}
|
|
|
|
const sources = Array.from(
|
|
new Set((args.sourcesChecked ?? []).map((v) => String(v).trim()).filter(Boolean))
|
|
).slice(0, 50);
|
|
if (!sources.length) throw new Error("At least one checked source is required");
|
|
|
|
const summary = String(args.summary ?? "").trim();
|
|
if (!summary) throw new Error("Counter-evidence summary is required");
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO story_research_checks(
|
|
id,thread_id,check_type,result,sources_checked,summary,
|
|
completed_by_sub,completed_by_name,completed_by_client_id
|
|
)
|
|
VALUES($1,$2,'counter_evidence',$3,$4::jsonb,$5,$6,$7,$8)
|
|
ON CONFLICT(thread_id,check_type)
|
|
DO UPDATE SET
|
|
result=EXCLUDED.result,
|
|
sources_checked=EXCLUDED.sources_checked,
|
|
summary=EXCLUDED.summary,
|
|
completed_by_sub=EXCLUDED.completed_by_sub,
|
|
completed_by_name=EXCLUDED.completed_by_name,
|
|
completed_by_client_id=EXCLUDED.completed_by_client_id,
|
|
completed_at=NOW()
|
|
RETURNING *`,
|
|
[
|
|
id, threadId, resultValue, JSON.stringify(sources), summary,
|
|
actor.sub, actor.name, actor.clientId
|
|
]
|
|
);
|
|
|
|
await pool.query("UPDATE story_threads SET updated_at=NOW() WHERE id=$1", [threadId]);
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("story_thread.research_check", { check: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function markStoryThreadStatus(
|
|
id: string,
|
|
status: string,
|
|
actorLabel: string
|
|
) {
|
|
if (!STORY_STATUSES.has(status)) throw new Error("Invalid story status");
|
|
|
|
if (status === "review" || status === "develop") {
|
|
const check = await pool.query(
|
|
`SELECT result FROM story_research_checks
|
|
WHERE thread_id=$1 AND check_type='counter_evidence'`,
|
|
[id]
|
|
);
|
|
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(
|
|
"UPDATE story_threads SET status=$2,updated_at=NOW() WHERE id=$1 RETURNING *",
|
|
[id, status]
|
|
);
|
|
if (!result.rowCount) return null;
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'story_thread.status_changed','story_thread',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ status, actorLabel })]
|
|
);
|
|
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("story_thread.status_changed", { story: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
|
|
export async function auditStorySources(threadId: string) {
|
|
const story = await getStoryThread(threadId);
|
|
if (!story) throw new Error("Story thread not found");
|
|
return { thread_id: threadId, ...(story as any).source_audit };
|
|
}
|
|
|
|
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.evidence_kind === "source_payload") &&
|
|
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) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
const errorHash = createHash("sha256").update(message).digest("hex").slice(0, 16);
|
|
await postSystemMessageOnce({
|
|
dedupeKey: `evidence-enrich-error:${row.evidence_id}:${errorHash}`,
|
|
title: "Source document enrichment failed",
|
|
body: `Evidence ${row.evidence_id} could not be upgraded from discovery metadata to a frozen source document: ${message}`,
|
|
severity: "warning",
|
|
kind: "warning",
|
|
relatedType: "evidence",
|
|
relatedId: row.evidence_id
|
|
});
|
|
enriched.push({
|
|
evidence_id: row.evidence_id,
|
|
ok: false,
|
|
error: message
|
|
});
|
|
}
|
|
}
|
|
return enriched;
|
|
}
|
|
|
|
export async function seedInitialStoryThread() {
|
|
const existing = await pool.query(
|
|
"SELECT id FROM story_threads WHERE title=$1 LIMIT 1",
|
|
["NZ labour market — August 2026"]
|
|
);
|
|
if (existing.rowCount) return existing.rows[0];
|
|
|
|
const evidence = await pool.query(
|
|
`SELECT e.id AS evidence_id,e.signal_id,s.source_id,s.title,s.summary
|
|
FROM evidence e
|
|
JOIN signals s ON s.id=e.signal_id
|
|
WHERE s.url ILIKE '%employment-indicators-august-2026%'
|
|
ORDER BY e.retrieved_at DESC
|
|
LIMIT 1`
|
|
);
|
|
if (!evidence.rowCount) return null;
|
|
|
|
const message = await pool.query(
|
|
`SELECT id,title,body
|
|
FROM agent_messages
|
|
WHERE title='Counter-evidence pass: August employment lead'
|
|
ORDER BY created_at DESC
|
|
LIMIT 1`
|
|
);
|
|
|
|
const actor: StoryActor = {
|
|
sub: "system:story-bootstrap",
|
|
name: "Story Bootstrap",
|
|
clientId: "media-intelligence-worker"
|
|
};
|
|
|
|
await pool.query("BEGIN");
|
|
try {
|
|
const thread = await createStoryThread({
|
|
title: "NZ labour market — August 2026",
|
|
question: "How should rising filled jobs be interpreted alongside unemployment, underutilisation and economic forecasts?"
|
|
}, actor);
|
|
|
|
const primary = await addStoryNode(thread.id, {
|
|
nodeType: "evidence",
|
|
title: "Stats NZ: Employment indicators August 2026",
|
|
summary: "Primary stored employment signal for this research thread.",
|
|
sourceId: evidence.rows[0].source_id,
|
|
signalId: evidence.rows[0].signal_id,
|
|
evidenceId: evidence.rows[0].evidence_id
|
|
}, actor);
|
|
|
|
const claim = await addStoryNode(thread.id, {
|
|
nodeType: "claim",
|
|
title: "Filled jobs are increasing",
|
|
summary: "Candidate interpretation from the August filled-jobs signal; this is not a final editorial conclusion."
|
|
}, actor);
|
|
|
|
const counter = await addStoryNode(thread.id, {
|
|
nodeType: "counterclaim",
|
|
title: "Filled jobs alone do not establish labour-market strengthening",
|
|
summary: "Counter-hypothesis: unemployment, underutilisation, participation, population and forward-looking indicators may qualify the headline signal."
|
|
}, actor);
|
|
|
|
const question = await addStoryNode(thread.id, {
|
|
nodeType: "question",
|
|
title: "What evidence contradicts the apparent improvement?",
|
|
summary: "Required 360° research branch before this thread can move to editorial review."
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: primary.id,
|
|
toNodeId: claim.id,
|
|
relationship: "supports",
|
|
reason: "The Stats NZ filled-jobs signal is the primary evidence behind the candidate claim."
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: counter.id,
|
|
toNodeId: claim.id,
|
|
relationship: "qualifies",
|
|
reason: "Filled-jobs growth needs to be interpreted alongside wider labour-market measures."
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: question.id,
|
|
toNodeId: counter.id,
|
|
relationship: "investigates",
|
|
reason: "The thread explicitly requires a counter-evidence search."
|
|
}, actor);
|
|
|
|
if (message.rowCount) {
|
|
const context = await addStoryNode(thread.id, {
|
|
nodeType: "context",
|
|
title: "Counter-evidence research pass",
|
|
summary: message.rows[0].body,
|
|
messageId: message.rows[0].id
|
|
}, actor);
|
|
|
|
await addStoryEdge(thread.id, {
|
|
fromNodeId: context.id,
|
|
toNodeId: counter.id,
|
|
relationship: "supports",
|
|
reason: "Prior research identified wider labour-market measures that qualify the filled-jobs interpretation."
|
|
}, actor);
|
|
|
|
await completeCounterEvidenceCheck(thread.id, {
|
|
result: "found",
|
|
sourcesChecked: ["Stats NZ", "Treasury", "BNZ"],
|
|
summary: "Material qualifying evidence was identified. Broader RBNZ and MBIE cross-checking remains desirable when those upstream sources are reachable."
|
|
}, actor);
|
|
}
|
|
|
|
await pool.query("COMMIT");
|
|
|
|
await postSystemMessageOnce({
|
|
dedupeKey: `story-thread-bootstrap:${thread.id}`,
|
|
title: "Story graph initialized",
|
|
body: `Created the first 360° research thread: ${thread.title}. The graph includes the primary Stats NZ evidence, candidate claim, counterclaim, explicit counter-evidence question and persisted research check.`,
|
|
severity: "notice",
|
|
kind: "update",
|
|
relatedType: "general",
|
|
relatedId: thread.id
|
|
});
|
|
|
|
return thread;
|
|
} catch (error) {
|
|
await pool.query("ROLLBACK");
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
|
|
export interface AgentMessageActor {
|
|
sub: string;
|
|
name: string;
|
|
clientId: string;
|
|
}
|
|
|
|
export async function postAgentMessage(
|
|
args: {
|
|
title: string;
|
|
body: string;
|
|
severity?: string;
|
|
kind?: string;
|
|
relatedType?: string | null;
|
|
relatedId?: string | null;
|
|
},
|
|
actor: AgentMessageActor
|
|
) {
|
|
const title = String(args.title ?? "").trim();
|
|
const body = String(args.body ?? "").trim();
|
|
if (!title || !body) throw new Error("Message title and body are required");
|
|
if (title.length > 180) throw new Error("Message title is too long");
|
|
if (body.length > 8000) throw new Error("Message body is too long");
|
|
|
|
const severity = String(args.severity ?? "info");
|
|
const kind = String(args.kind ?? "update");
|
|
const allowedSeverities = new Set(["info", "notice", "warning", "error"]);
|
|
const allowedKinds = new Set(["update", "finding", "request", "warning", "error"]);
|
|
const allowedRelated = new Set(["opportunity", "source", "evidence", "story", "general"]);
|
|
|
|
if (!allowedSeverities.has(severity)) throw new Error("Invalid message severity");
|
|
if (!allowedKinds.has(kind)) throw new Error("Invalid message kind");
|
|
|
|
const relatedType = args.relatedType ? String(args.relatedType) : null;
|
|
if (relatedType && !allowedRelated.has(relatedType)) throw new Error("Invalid related_type");
|
|
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO agent_messages(
|
|
id,actor_sub,actor_name,actor_client_id,severity,kind,title,body,related_type,related_id
|
|
) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
|
|
RETURNING *`,
|
|
[
|
|
id,
|
|
actor.sub,
|
|
actor.name,
|
|
actor.clientId,
|
|
severity,
|
|
kind,
|
|
title,
|
|
body,
|
|
relatedType,
|
|
args.relatedId ? String(args.relatedId) : null
|
|
]
|
|
);
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'agent_message.posted','agent_message',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({
|
|
actorSub: actor.sub,
|
|
actorName: actor.name,
|
|
actorClientId: actor.clientId,
|
|
severity,
|
|
kind,
|
|
relatedType,
|
|
relatedId: args.relatedId ?? null
|
|
})]
|
|
);
|
|
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("agent_message.created", { message: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function listAgentMessages(args: {
|
|
acknowledged?: boolean;
|
|
severity?: string;
|
|
since?: string;
|
|
limit?: number;
|
|
} = {}) {
|
|
const values: unknown[] = [];
|
|
const where: string[] = [];
|
|
|
|
if (typeof args.acknowledged === "boolean") {
|
|
where.push(args.acknowledged ? "acknowledged_at IS NOT NULL" : "acknowledged_at IS NULL");
|
|
}
|
|
if (args.severity) {
|
|
values.push(args.severity);
|
|
where.push(`severity=$${values.length}`);
|
|
}
|
|
if (args.since) {
|
|
values.push(args.since);
|
|
where.push(`created_at >= $${values.length}::timestamptz`);
|
|
}
|
|
|
|
const limit = Math.min(Math.max(Number(args.limit ?? 25), 1), 100);
|
|
values.push(limit);
|
|
|
|
const result = await pool.query(
|
|
`SELECT * FROM agent_messages
|
|
${where.length ? "WHERE " + where.join(" AND ") : ""}
|
|
ORDER BY created_at DESC
|
|
LIMIT $${values.length}`,
|
|
values
|
|
);
|
|
return result.rows;
|
|
}
|
|
|
|
export async function getAgentMessage(id: string) {
|
|
const result = await pool.query(
|
|
"SELECT * FROM agent_messages WHERE id=$1",
|
|
[id]
|
|
);
|
|
return result.rows[0] ?? null;
|
|
}
|
|
|
|
export async function acknowledgeAgentMessage(id: string, acknowledgedBy: string) {
|
|
const result = await pool.query(
|
|
`UPDATE agent_messages
|
|
SET acknowledged_at=COALESCE(acknowledged_at,NOW()),
|
|
acknowledged_by=COALESCE(acknowledged_by,$2)
|
|
WHERE id=$1
|
|
RETURNING *`,
|
|
[id, acknowledgedBy]
|
|
);
|
|
if (!result.rowCount) return null;
|
|
|
|
await pool.query(
|
|
`INSERT INTO audit_events(id,event_type,object_type,object_id,payload)
|
|
VALUES($1,'agent_message.acknowledged','agent_message',$2,$3::jsonb)`,
|
|
[randomUUID(), id, JSON.stringify({ acknowledgedBy })]
|
|
);
|
|
const realtimeRow = result.rows[0];
|
|
publishNewsroomEvent("agent_message.acknowledged", { message: realtimeRow });
|
|
return realtimeRow;
|
|
}
|
|
|
|
export async function postSystemMessageOnce(args: {
|
|
dedupeKey: string;
|
|
title: string;
|
|
body: string;
|
|
severity?: string;
|
|
kind?: string;
|
|
relatedType?: string | null;
|
|
relatedId?: string | null;
|
|
}) {
|
|
const id = randomUUID();
|
|
const result = await pool.query(
|
|
`INSERT INTO agent_messages(
|
|
id,actor_sub,actor_name,actor_client_id,severity,kind,title,body,
|
|
related_type,related_id,dedupe_key
|
|
)
|
|
VALUES($1,'system:source-monitor','Source Monitor','media-intelligence-worker',
|
|
$2,$3,$4,$5,$6,$7,$8)
|
|
ON CONFLICT(dedupe_key) DO NOTHING
|
|
RETURNING *`,
|
|
[
|
|
id,
|
|
args.severity ?? "warning",
|
|
args.kind ?? "warning",
|
|
args.title,
|
|
args.body,
|
|
args.relatedType ?? null,
|
|
args.relatedId ?? null,
|
|
args.dedupeKey
|
|
]
|
|
);
|
|
|
|
const message = result.rows[0] ?? null;
|
|
if (message) publishNewsroomEvent("agent_message.created", { message });
|
|
return message;
|
|
}
|