From 00b95667c9237946c44590b11bd5a33de78c5f48 Mon Sep 17 00:00:00 2001
From: Admin <dmparrow@gmail.com>
Date: Tue, 29 Sep 2026 06:46:04 +0000
Subject: [PATCH] Sync source failure Agent Activity messages

---
 src/db.ts   | 37 +++++++++++++++++++++++++++++++++++++
 src/sync.ts | 33 ++++++++++++++++++++++++++++++---
 2 files changed, 67 insertions(+), 3 deletions(-)

diff --git a/src/db.ts b/src/db.ts
index f4b0be3..36b3c45 100644
--- a/src/db.ts
+++ b/src/db.ts
@@ -99,6 +99,7 @@ export async function migrate() {
       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()
@@ -127,6 +128,8 @@ export async function migrate() {
 
   await pool.query("ALTER TABLE sources ADD COLUMN IF NOT EXISTS baseline_complete BOOLEAN NOT NULL DEFAULT FALSE");
   await pool.query("ALTER TABLE 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)
@@ -525,3 +528,37 @@ export async function acknowledgeAgentMessage(id: string, acknowledgedBy: string
   );
   return result.rows[0];
 }
+
+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
+    ]
+  );
+
+  return result.rows[0] ?? null;
+}
diff --git a/src/sync.ts b/src/sync.ts
index 9dc8c0f..d267fb7 100644
--- a/src/sync.ts
+++ b/src/sync.ts
@@ -1,5 +1,5 @@
-import { randomUUID } from "node:crypto";
-import { pool, storeItem } from "./db.js";
+import { createHash, randomUUID } from "node:crypto";
+import { pool, postSystemMessageOnce, storeItem } from "./db.js";
 import { fetchSource, SOURCES } from "./sources.js";
 import { scoreItem } from "./scoring.js";
 import type { SourceId } from "./types.js";
@@ -14,12 +14,14 @@ export async function syncSource(sourceId: SourceId) {
     "INSERT INTO source_runs(id,source_id,status) VALUES($1,$2,'running')",
     [runId, sourceId]
   );
+  let previousError: string | null = null;
   try {
     const sourceState = await pool.query(
-      "SELECT baseline_complete FROM sources WHERE id=$1",
+      "SELECT baseline_complete,last_error FROM sources WHERE id=$1",
       [sourceId]
     );
     const baseline = !Boolean(sourceState.rows[0]?.baseline_complete);
+    previousError = sourceState.rows[0]?.last_error ?? null;
 
     const items = await fetchSource(sourceId);
     let created = 0;
@@ -43,6 +45,19 @@ export async function syncSource(sourceId: SourceId) {
       [sourceId, baseline && items.length > 0]
     );
 
+    if (previousError) {
+      const errorHash = createHash("sha256").update(previousError).digest("hex").slice(0, 16);
+      await postSystemMessageOnce({
+        dedupeKey: `source-recovered:${sourceId}:${errorHash}`,
+        title: `${source.name} recovered`,
+        body: `The source sync is healthy again after the previous error: ${previousError}`,
+        severity: "notice",
+        kind: "update",
+        relatedType: "source",
+        relatedId: sourceId
+      });
+    }
+
     return {
       sourceId,
       mode: baseline ? "baseline" : "live",
@@ -60,6 +75,18 @@ export async function syncSource(sourceId: SourceId) {
       "UPDATE sources SET last_synced_at=NOW(),last_error=$2 WHERE id=$1",
       [sourceId, message]
     );
+
+    const errorHash = createHash("sha256").update(message).digest("hex").slice(0, 16);
+    await postSystemMessageOnce({
+      dedupeKey: `source-error:${sourceId}:${errorHash}`,
+      title: `${source.name} sync failed`,
+      body: message,
+      severity: "warning",
+      kind: "warning",
+      relatedType: "source",
+      relatedId: sourceId
+    });
+
     throw error;
   }
 }