Sync Parliament fix and agent activity layer

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];
+}

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, "&lt;")
+    .replace(/>/g, "&gt;")
+    .replace(/"/g, "&quot;")
+    .replace(/'/g, "&#39;");
+}
+
+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}
 </style></head><body>
 <header>
 <div><h1>Media Intelligence MVP</h1><p>Human-led NZ public-source monitoring behind a dedicated MCP.</p></div>
-<div class="user"><strong>${identity}</strong><br><a href="/auth/logout">Sign out</a></div>
+<div class="user"><strong>${escapeHtml(identity)}</strong><br><a href="/auth/logout">Sign out</a></div>
 </header>
 <div class="grid">
 <div class="card"><strong class="big">${c.signals ?? 0}</strong>signals</div>
 <div class="card"><strong class="big">${c.opportunities ?? 0}</strong>opportunities</div>
 <div class="card"><strong class="big">${c.new_opportunities ?? 0}</strong>new</div>
+<div class="card"><strong class="big">${c.unread_messages ?? 0}</strong>agent messages</div>
 </div>
+<h2>Agent Activity</h2>
+${messages.length
+  ? messages.map((m:any)=>`<div class="card message ${escapeHtml(m.severity)}">
+      <div class="message-head">
+        <div><strong>${escapeHtml(m.title)}</strong><br><span class="meta">${escapeHtml(m.kind)} · ${escapeHtml(m.severity)}</span></div>
+        <div class="meta">${escapeHtml(nzTime(m.created_at))}</div>
+      </div>
+      <p class="message-body">${escapeHtml(m.body)}</p>
+      <div class="meta">From ${escapeHtml(m.actor_name)} via <code>${escapeHtml(m.actor_client_id)}</code>${m.related_type ? ` · ${escapeHtml(m.related_type)} <code>${escapeHtml(m.related_id ?? "")}</code>` : ""}</div>
+      <form method="post" action="/messages/${encodeURIComponent(m.id)}/ack"><button type="submit">Acknowledge</button></form>
+    </div>`).join("")
+  : '<div class="card"><span class="meta">No unread agent messages.</span></div>'}
 <h2>Sources</h2>
 ${sourceRows.rows.map((s:any)=>`<div class="card"><strong>${s.name}</strong><br><code>${s.id}</code><br>${s.signal_count} stored signals<br><span class="${s.last_error?'bad':'ok'}">${s.last_error ? 'Error: '+s.last_error : (s.last_synced_at ? 'Last sync '+new Date(s.last_synced_at).toLocaleString() : 'Waiting for first sync')}</span></div>`).join("")}
 <p>MCP endpoint: <code>/api/mcp</code> — OAuth protected separately from this browser session.</p>

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<typeof actorFromRequest>) {
   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({

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",
+    type: "html",
-    url: "https://www3.parliament.nz/en/channellisting/rss/9343/e37f168d-4521-4437-a6fd-6bb7ff54e800"
+    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<string, number> = {
+    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<NormalizedItem[]> {
   const source = SOURCES.find((s) => s.id === "parliament")!;
-  const xml = await getText(source.url, "application/atom+xml, application/xml, text/xml");
+  const html = await getText(source.url, "text/html");
-  const parser = new XMLParser({ ignoreAttributes: false, attributeNamePrefix: "@_" });
+  const $ = load(html);
-  const parsed = parser.parse(xml);
+  const seen = new Set<string>();
+  const items: NormalizedItem[] = [];
 
-  let list = asArray<any>(parsed?.feed?.entry ?? parsed?.rss?.channel?.item);
+  $('a[href*="/en/pb/sc/submissions-and-advice/document/"]').each((_index, element) => {
-  if (!list.length) list = rawFeedItems(xml);
+    if (items.length >= 30) return;
 
-  return list.slice(0, 30).map((entry: any, index: number) => {
+    const href = $(element).attr("href");
-    const title = cleanXmlText(entry.title?.["#text"] ?? entry.title ?? "Parliament item");
+    const title = $(element).text().replace(/\s+/g, " ").trim();
-    const summary = cleanXmlText(
+    if (!href || !title || title.length < 2) return;
-      entry.summary?.["#text"] ?? entry.summary ?? entry.description ?? entry.content ?? ""
+
-    );
+    const url = new URL(href, source.url).toString().split("?")[0];
-    const url = firstLink(entry.link) || String(entry.link ?? source.url);
+    if (seen.has(url)) return;
-    return {
+    seen.add(url);
-      sourceId: "parliament" as const,
+
-      externalId: String(entry.id?.["#text"] ?? entry.id ?? entry.guid ?? url ?? index),
+    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:
+      publishedAt: parseParliamentDate(dateText),
-        cleanXmlText(entry.updated ?? entry.published ?? entry.pubDate ?? "") || null,
+      topics: [
-      topics: ["parliament", "public-policy"],
+        "parliament",
-      raw: entry
+        "select-committee",
-    };
+        "evidence",
+        ...(committee ? [committee.toLowerCase().replace(/\s+/g, "-")] : [])
+      ],
+      raw: { title, summary, url, committee }
     });
+  });
+
+  return items;
 }
 
 interface SitemapEntry {

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;
 }