From d9142312fca342625d71e4384695be98557ae4e7 Mon Sep 17 00:00:00 2001 From: Admin Date: Tue, 29 Sep 2026 06:25:54 +0000 Subject: [PATCH] Sync Parliament fix and agent activity layer --- src/db.ts | 154 +++++++++++++++++++++++++++++++++++++++++++++++++ src/index.ts | 64 +++++++++++++++++++- src/mcp.ts | 95 +++++++++++++++++++++++++++++- src/sources.ts | 82 +++++++++++++++++++------- src/types.ts | 2 +- 5 files changed, 369 insertions(+), 28 deletions(-) diff --git a/src/db.ts b/src/db.ts index de68cd4..17822e9 100644 --- a/src/db.ts +++ b/src/db.ts @@ -86,6 +86,27 @@ export async function migrate() { created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); + 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, + 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 audit_events ( id TEXT PRIMARY KEY, event_type TEXT NOT NULL, @@ -259,3 +280,136 @@ export async function searchEvidence(query: string, limit = 20) { ); return result.rows; } + +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", "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 + })] + ); + + return result.rows[0]; +} + +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 })] + ); + return result.rows[0]; +} diff --git a/src/index.ts b/src/index.ts index 89a95a8..a41be0e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,11 @@ import express from "express"; -import { migrate, pool, seedSources } from "./db.js"; +import { + acknowledgeAgentMessage, + listAgentMessages, + migrate, + pool, + seedSources +} from "./db.js"; import { beginWebLogin, finishWebLogin, @@ -15,6 +21,29 @@ import { syncAll } from "./sync.js"; const app = express(); app.set("trust proxy", true); app.use(express.json({ limit: "2mb" })); +app.use(express.urlencoded({ extended: false })); + +function escapeHtml(value: unknown) { + return String(value ?? "") + .replace(/&/g, "&") + .replace(//g, ">") + .replace(/"/g, """) + .replace(/'/g, "'"); +} + +function nzTime(value: unknown) { + if (!value) return ""; + try { + return new Intl.DateTimeFormat("en-NZ", { + timeZone: "Pacific/Auckland", + dateStyle: "medium", + timeStyle: "short" + }).format(new Date(String(value))); + } catch { + return String(value); + } +} app.get("/health", async (_req, res) => { try { @@ -74,6 +103,15 @@ app.get("/api/mcp", requireMcpAuth, (_req, res) => { }); }); +app.post("/messages/:id/ack", requireWebAuth, async (req, res) => { + const user = (req as any).user ?? {}; + const identity = String( + user.preferred_username ?? user.email ?? user.name ?? user.sub ?? "web-user" + ); + await acknowledgeAgentMessage(req.params.id, `web:${identity}`); + return res.redirect("/"); +}); + app.get("/", requireWebAuth, async (req, res) => { const sourceRows = await pool.query( `SELECT s.id,s.name,s.last_synced_at,s.last_error, @@ -87,8 +125,10 @@ app.get("/", requireWebAuth, async (req, res) => { SELECT (SELECT COUNT(*) FROM signals) AS signals, (SELECT COUNT(*) FROM opportunities) AS opportunities, - (SELECT COUNT(*) FROM opportunities WHERE status='new') AS new_opportunities + (SELECT COUNT(*) FROM opportunities WHERE status='new') AS new_opportunities, + (SELECT COUNT(*) FROM agent_messages WHERE acknowledged_at IS NULL) AS unread_messages `); + const messages = await listAgentMessages({ acknowledged: false, limit: 10 }); const c = counts.rows[0] ?? {}; const user = (req as any).user ?? {}; const identity = user.name || user.preferred_username || user.email || "Signed-in user"; @@ -106,16 +146,34 @@ strong.big{font-size:2rem;display:block} code{background:#f3f4f6;padding:2px 6px;border-radius:5px} a{color:inherit} .user{font-size:.9rem;color:#4b5563;text-align:right} +.message{border-left-width:5px}.message.notice{border-left-color:#2563eb}.message.warning{border-left-color:#d97706}.message.error{border-left-color:#b91c1c} +.message-head{display:flex;align-items:flex-start;justify-content:space-between;gap:16px} +.message-body{white-space:pre-wrap;line-height:1.45} +.meta{font-size:.84rem;color:#6b7280} +button{margin-top:12px;padding:7px 11px;border:1px solid #9ca3af;border-radius:7px;background:#fff;cursor:pointer}

Media Intelligence MVP

Human-led NZ public-source monitoring behind a dedicated MCP.

-
${identity}
Sign out
+
${escapeHtml(identity)}
Sign out
${c.signals ?? 0}signals
${c.opportunities ?? 0}opportunities
${c.new_opportunities ?? 0}new
+
${c.unread_messages ?? 0}agent messages
+

Agent Activity

+${messages.length + ? messages.map((m:any)=>`
+
+
${escapeHtml(m.title)}
${escapeHtml(m.kind)} · ${escapeHtml(m.severity)}
+
${escapeHtml(nzTime(m.created_at))}
+
+

${escapeHtml(m.body)}

+
From ${escapeHtml(m.actor_name)} via ${escapeHtml(m.actor_client_id)}${m.related_type ? ` · ${escapeHtml(m.related_type)} ${escapeHtml(m.related_id ?? "")}` : ""}
+
+
`).join("") + : '
No unread agent messages.
'}

Sources

${sourceRows.rows.map((s:any)=>`
${s.name}
${s.id}
${s.signal_count} stored signals
${s.last_error ? 'Error: '+s.last_error : (s.last_synced_at ? 'Last sync '+new Date(s.last_synced_at).toLocaleString() : 'Waiting for first sync')}
`).join("")}

MCP endpoint: /api/mcp — OAuth protected separately from this browser session.

diff --git a/src/mcp.ts b/src/mcp.ts index fee8462..7afc2a0 100644 --- a/src/mcp.ts +++ b/src/mcp.ts @@ -1,10 +1,14 @@ import type { Request, Response } from "express"; import { + acknowledgeAgentMessage, + getAgentMessage, getEvidence, getOpportunity, + listAgentMessages, listOpportunities, listSources, markOpportunity, + postAgentMessage, searchEvidence, sourceStatus } from "./db.js"; @@ -89,6 +93,57 @@ const tools = [ required: ["query"], additionalProperties: false } + }, + { + name: "messages_list", + description: "List agent activity messages for the human media team.", + inputSchema: { + type: "object", + properties: { + acknowledged: { type: "boolean" }, + severity: { type: "string", enum: ["info", "notice", "warning", "error"] }, + since: { type: "string", description: "ISO-8601 timestamp" }, + limit: { type: "integer", minimum: 1, maximum: 100 } + }, + additionalProperties: false + } + }, + { + name: "message_get", + description: "Get one agent activity message.", + inputSchema: { + type: "object", + properties: { id: { type: "string" } }, + required: ["id"], + additionalProperties: false + } + }, + { + name: "message_post", + description: "Post a human-readable activity update from the authenticated MCP agent. The author identity is taken from OAuth and cannot be supplied by the caller.", + inputSchema: { + type: "object", + properties: { + title: { type: "string", maxLength: 180 }, + body: { type: "string", maxLength: 8000 }, + severity: { type: "string", enum: ["info", "notice", "warning", "error"] }, + kind: { type: "string", enum: ["update", "finding", "request", "warning", "error"] }, + related_type: { type: "string", enum: ["opportunity", "source", "evidence", "general"] }, + related_id: { type: "string" } + }, + required: ["title", "body"], + additionalProperties: false + } + }, + { + name: "message_acknowledge", + description: "Acknowledge an agent activity message.", + inputSchema: { + type: "object", + properties: { id: { type: "string" } }, + required: ["id"], + additionalProperties: false + } } ]; @@ -104,7 +159,16 @@ function result(data: unknown) { }; } -async function callTool(name: string, args: any) { +function actorFromRequest(req: Request) { + const auth = (req as any).auth ?? {}; + return { + sub: String(auth.sub ?? "unknown"), + name: String(auth.preferred_username ?? auth.email ?? auth.name ?? auth.sub ?? "OAuth agent"), + clientId: String(auth.azp ?? auth.client_id ?? process.env.KEYCLOAK_CLIENT_ID ?? "unknown") + }; +} + +async function callTool(name: string, args: any, actor: ReturnType) { if (name === "sources_list") return result(await listSources()); if (name === "sources_status") return result(await sourceStatus()); if (name === "sources_sync") return result(await syncAll(args?.source_id as SourceId | undefined)); @@ -113,6 +177,25 @@ async function callTool(name: string, args: any) { if (name === "opportunity_mark") return result(await markOpportunity(args.id, args.status)); if (name === "evidence_get") return result(await getEvidence(args.id)); if (name === "evidence_search") return result(await searchEvidence(args.query, args.limit)); + if (name === "messages_list") return result(await listAgentMessages({ + acknowledged: args?.acknowledged, + severity: args?.severity, + since: args?.since, + limit: args?.limit + })); + if (name === "message_get") return result(await getAgentMessage(args.id)); + if (name === "message_post") return result(await postAgentMessage({ + title: args.title, + body: args.body, + severity: args.severity, + kind: args.kind, + relatedType: args.related_type, + relatedId: args.related_id + }, actor)); + if (name === "message_acknowledge") return result(await acknowledgeAgentMessage( + args.id, + `${actor.clientId}:${actor.name}` + )); throw new Error(`Unknown tool: ${name}`); } @@ -138,7 +221,11 @@ export async function handleMcp(req: Request, res: Response) { } else if (method === "tools/list") { response = { tools }; } else if (method === "tools/call") { - response = await callTool(body.params?.name, body.params?.arguments ?? {}); + response = await callTool( + body.params?.name, + body.params?.arguments ?? {}, + actorFromRequest(req) + ); } else if (method === "resources/list") { response = { resources: [ @@ -153,7 +240,8 @@ export async function handleMcp(req: Request, res: Response) { response = { resourceTemplates: [ { uriTemplate: "media://opportunities/{id}", name: "Media opportunity", mimeType: "application/json" }, - { uriTemplate: "media://evidence/{id}", name: "Source evidence", mimeType: "application/json" } + { uriTemplate: "media://evidence/{id}", name: "Source evidence", mimeType: "application/json" }, + { uriTemplate: "media://messages/{id}", name: "Agent activity message", mimeType: "application/json" } ] }; } else if (method === "resources/read") { @@ -161,6 +249,7 @@ export async function handleMcp(req: Request, res: Response) { if (uri === "media://sources") response = { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(await listSources(), null, 2) }] }; else if (uri.startsWith("media://opportunities/")) response = { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(await getOpportunity(uri.split("/").pop()!), null, 2) }] }; else if (uri.startsWith("media://evidence/")) response = { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(await getEvidence(uri.split("/").pop()!), null, 2) }] }; + else if (uri.startsWith("media://messages/")) response = { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(await getAgentMessage(uri.split("/").pop()!), null, 2) }] }; else throw new Error("Unknown resource URI"); } else { return res.status(404).json({ diff --git a/src/sources.ts b/src/sources.ts index 50ad23a..9871938 100644 --- a/src/sources.ts +++ b/src/sources.ts @@ -1,3 +1,4 @@ +import { load } from "cheerio"; import { XMLParser } from "fast-xml-parser"; import type { NormalizedItem, SourceDefinition, SourceId } from "./types.js"; @@ -11,8 +12,8 @@ export const SOURCES: SourceDefinition[] = [ { id: "parliament", name: "NZ Parliament evidence and submissions", - type: "atom", - url: "https://www3.parliament.nz/en/channellisting/rss/9343/e37f168d-4521-4437-a6fd-6bb7ff54e800" + type: "html", + url: "https://www3.parliament.nz/en/pb/sc/evidence-submissions/" }, { id: "stats-nz", @@ -138,33 +139,72 @@ function rawFeedItems(xml: string): any[] { }); } +function parseParliamentDate(text: string): string | null { + const match = text.match(/\b(\d{1,2})\s+(January|February|March|April|May|June|July|August|September|October|November|December)\s+(20\d{2})\b/i); + if (!match) return null; + const months: Record = { + january: 0, february: 1, march: 2, april: 3, may: 4, june: 5, + july: 6, august: 7, september: 8, october: 9, november: 10, december: 11 + }; + const value = new Date(Date.UTC( + Number(match[3]), + months[match[2].toLowerCase()], + Number(match[1]), + 0, 0, 0 + )); + return Number.isNaN(value.getTime()) ? null : value.toISOString(); +} + async function fetchParliament(): Promise { const source = SOURCES.find((s) => s.id === "parliament")!; - const xml = await getText(source.url, "application/atom+xml, application/xml, text/xml"); - const parser = new XMLParser({ ignoreAttributes: false, attributeNamePrefix: "@_" }); - const parsed = parser.parse(xml); + const html = await getText(source.url, "text/html"); + const $ = load(html); + const seen = new Set(); + const items: NormalizedItem[] = []; - let list = asArray(parsed?.feed?.entry ?? parsed?.rss?.channel?.item); - if (!list.length) list = rawFeedItems(xml); + $('a[href*="/en/pb/sc/submissions-and-advice/document/"]').each((_index, element) => { + if (items.length >= 30) return; - return list.slice(0, 30).map((entry: any, index: number) => { - const title = cleanXmlText(entry.title?.["#text"] ?? entry.title ?? "Parliament item"); - const summary = cleanXmlText( - entry.summary?.["#text"] ?? entry.summary ?? entry.description ?? entry.content ?? "" - ); - const url = firstLink(entry.link) || String(entry.link ?? source.url); - return { - sourceId: "parliament" as const, - externalId: String(entry.id?.["#text"] ?? entry.id ?? entry.guid ?? url ?? index), + const href = $(element).attr("href"); + const title = $(element).text().replace(/\s+/g, " ").trim(); + if (!href || !title || title.length < 2) return; + + const url = new URL(href, source.url).toString().split("?")[0]; + if (seen.has(url)) return; + seen.add(url); + + const container = $(element).closest("tr").length + ? $(element).closest("tr") + : $(element).closest("li, article, .list-item, div"); + + const surrounding = container.text().replace(/\s+/g, " ").trim(); + const summary = surrounding.replace(title, "").replace(/View more View less/gi, "").trim().slice(0, 1400); + + const committee = + surrounding.match(/Committee:\s*([^|]+?)(?:Date:|$)/i)?.[1]?.trim() ?? + ""; + const dateText = + surrounding.match(/Date:\s*(\d{1,2}\s+[A-Za-z]+\s+20\d{2})/i)?.[1] ?? + surrounding; + + items.push({ + sourceId: "parliament", + externalId: url, title, summary, url, - publishedAt: - cleanXmlText(entry.updated ?? entry.published ?? entry.pubDate ?? "") || null, - topics: ["parliament", "public-policy"], - raw: entry - }; + publishedAt: parseParliamentDate(dateText), + topics: [ + "parliament", + "select-committee", + "evidence", + ...(committee ? [committee.toLowerCase().replace(/\s+/g, "-")] : []) + ], + raw: { title, summary, url, committee } + }); }); + + return items; } interface SitemapEntry { diff --git a/src/types.ts b/src/types.ts index 5359cc0..4108a34 100644 --- a/src/types.ts +++ b/src/types.ts @@ -3,7 +3,7 @@ export type SourceId = "geonet" | "parliament" | "stats-nz" | "data-govt-nz"; export interface SourceDefinition { id: SourceId; name: string; - type: "geojson" | "atom" | "sitemap" | "ckan"; + type: "geojson" | "html" | "sitemap" | "ckan"; url: string; }