feat(cdc): add Re-sync sources button + auto-heal for Debezium connectors

- streaming_ops: add connector status discovery, manual resync endpoint
  (POST /api/pipeline/streaming/resync), connectors status endpoint, and a
  background connector_autoheal_loop that restarts FAILED tasks automatically
- main: wire connector_autoheal_loop into app lifespan
- api.ts: add resyncSources() helper
- ChangesView: add "Re-sync sources" header button with live status

api/main.py

@@ -47,6 +47,7 @@ from elasticsearch_api import router as elasticsearch_router
 from sql_console import router as sql_router
 from agent_ops import router as agent_ops_router, agent_dml_loop, etl_agent_loop, custodian_offload_loop
 from agent_activity import agent_activity_loop
+from streaming_ops import connector_autoheal_loop
 from cdc_consumer import router as cdc_router, cdc_consumer_loop
 from movements import router as movements_router
 from movements import MOVEMENT_BY_ID, trigger_and_watch
@@ -737,6 +738,7 @@ async def lifespan(app: FastAPI):
     etl_task = asyncio.create_task(etl_agent_loop())
     cust_task = asyncio.create_task(custodian_offload_loop())
     act_task = asyncio.create_task(agent_activity_loop())
+    heal_task = asyncio.create_task(connector_autoheal_loop())
     add_feed("infra-sentinel", "ATC Command Center API online", "info")
     yield
     task.cancel()
@@ -745,6 +747,7 @@ async def lifespan(app: FastAPI):
     etl_task.cancel()
     cust_task.cancel()
     act_task.cancel()
+    heal_task.cancel()
     if redis_client:
         await redis_client.close()
 

api/streaming_ops.py

@@ -433,3 +433,126 @@ async def resume_kafka_connector(name: str) -> JSONResponse:
         return JSONResponse({"ok": True, "connector": name, "action": "resume"})
     except Exception as exc:
         return JSONResponse({"ok": False, "error": str(exc)}, status_code=500)
+
+
+# ── Source CDC re-sync + autonomous self-heal ───────────────────────────────
+# When a source database briefly drops, the Debezium tasks land in FAILED (or
+# stall while still reporting RUNNING) and never recover on their own. The
+# re-sync restarts the connectors so CDC catches up; the auto-heal loop does the
+# same automatically for FAILED tasks.
+SOURCE_CONNECTORS = [c for c in os.getenv(
+    "CDC_SOURCE_CONNECTORS",
+    "postgres-sales-connector,mysql-hr-connector,mongodb-supplychain-connector",
+).split(",") if c.strip()]
+
+
+async def _term(agent_id: str, text: str, level: str = "info", phase: str = "resync") -> None:
+    try:
+        from agent_terminal import terminal_log
+        await terminal_log(agent_id, text, level=level, phase=phase)
+    except Exception:
+        pass
+
+
+async def _connector_status(client: httpx.AsyncClient, name: str) -> dict[str, Any]:
+    try:
+        r = await client.get(f"{KAFKA_CONNECT_URL}/connectors/{name}/status")
+        if r.status_code >= 400:
+            return {"name": name, "state": "MISSING", "tasks": [], "failed": []}
+        st = r.json()
+        tasks = st.get("tasks") or []
+        return {
+            "name": name,
+            "state": (st.get("connector") or {}).get("state"),
+            "tasks": [{"id": t.get("id"), "state": t.get("state")} for t in tasks],
+            "failed": [t.get("id") for t in tasks if t.get("state") == "FAILED"],
+        }
+    except Exception as exc:
+        return {"name": name, "state": "ERROR", "error": str(exc)[:120], "tasks": [], "failed": []}
+
+
+async def _discover_connectors(client: httpx.AsyncClient) -> list[str]:
+    try:
+        r = await client.get(f"{KAFKA_CONNECT_URL}/connectors")
+        if r.status_code < 400 and isinstance(r.json(), list) and r.json():
+            return r.json()
+    except Exception:
+        pass
+    return list(SOURCE_CONNECTORS)
+
+
+async def resync_connectors(force: bool = True, names: list[str] | None = None) -> dict[str, Any]:
+    """force=True → restart every connector incl. all tasks (full re-sync, also
+    recovers stalled-but-RUNNING tasks). force=False → only restart connectors
+    that have a FAILED connector or task (self-heal)."""
+    result: dict[str, Any] = {"ok": True, "restarted": [], "skipped": [], "before": []}
+    async with httpx.AsyncClient(timeout=20.0) as client:
+        targets = names or await _discover_connectors(client)
+        for name in targets:
+            before = await _connector_status(client, name)
+            result["before"].append(before)
+            unhealthy = before.get("state") in ("FAILED", "ERROR", "MISSING") or before.get("failed")
+            if not force and not unhealthy:
+                result["skipped"].append(name)
+                continue
+            try:
+                qs = "includeTasks=true" if force else "includeTasks=true&onlyFailed=true"
+                rr = await client.post(f"{KAFKA_CONNECT_URL}/connectors/{name}/restart?{qs}")
+                if rr.status_code < 400:
+                    result["restarted"].append(name)
+                    await _term("etl-guardian", f"$ kafka-connect restart {name}  ({'full re-sync' if force else 'failed tasks'})", level="cmd")
+                else:
+                    result["ok"] = False
+                    await _term("etl-guardian", f"  ✗ {name}: HTTP {rr.status_code}", level="err")
+            except Exception as exc:
+                result["ok"] = False
+                await _term("etl-guardian", f"  ✗ {name}: {str(exc)[:100]}", level="err")
+    _cache["ts"] = 0
+    return result
+
+
+@router.get("/connectors")
+async def connectors_status() -> JSONResponse:
+    async with httpx.AsyncClient(timeout=12.0) as client:
+        names = await _discover_connectors(client)
+        items = [await _connector_status(client, n) for n in names]
+    healthy = sum(1 for c in items if c.get("state") == "RUNNING" and not c.get("failed"))
+    return JSONResponse({"ok": True, "connect_url": KAFKA_CONNECT_URL,
+                         "healthy": healthy, "total": len(items), "connectors": items})
+
+
+@router.post("/resync")
+async def resync_sources(body: dict[str, Any] = Body(default={})) -> JSONResponse:
+    force = bool(body.get("force", True))
+    _feed("etl-guardian", f"[kafka] Source CDC re-sync requested ({'full' if force else 'failed only'})", "warn")
+    await _term("etl-guardian", "═══ Source CDC re-sync — restarting Debezium connectors ═══", level="info")
+    res = await resync_connectors(force=force)
+    await asyncio.sleep(3)  # let tasks transition
+    async with httpx.AsyncClient(timeout=15.0) as client:
+        after = [await _connector_status(client, c["name"]) for c in res["before"]]
+    res["after"] = after
+    healthy = sum(1 for c in after if c.get("state") == "RUNNING" and not c.get("failed"))
+    res["healthy"] = healthy
+    res["total"] = len(after)
+    await _term("etl-guardian", f"  ← re-sync requested for {len(res['restarted'])} connectors · {healthy}/{len(after)} healthy", level="ok")
+    return JSONResponse(res)
+
+
+async def connector_autoheal_loop() -> None:
+    """Autonomously restart FAILED Debezium tasks so CDC recovers after a source
+    DB outage without operator action."""
+    await asyncio.sleep(40)
+    interval = max(30.0, float(os.getenv("CONNECTOR_AUTOHEAL_SECONDS", "60")))
+    while True:
+        try:
+            async with httpx.AsyncClient(timeout=15.0) as client:
+                for name in await _discover_connectors(client):
+                    st = await _connector_status(client, name)
+                    if st.get("state") in ("FAILED", "ERROR") or st.get("failed"):
+                        await client.post(f"{KAFKA_CONNECT_URL}/connectors/{name}/restart?includeTasks=true&onlyFailed=true")
+                        _feed("etl-guardian", f"[kafka] auto-heal restarted {name}", "warn")
+                        await _term("etl-guardian", f"  ⟳ auto-heal: restarted FAILED task(s) on {name}", level="warn", phase="autoheal")
+            _cache["ts"] = 0
+        except Exception:
+            pass
+        await asyncio.sleep(interval)

ui/src/components/features/ChangesView.tsx

@@ -1,6 +1,6 @@
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
-import { Activity, Radio, RefreshCw, PlusCircle, Pencil, Trash2, Database, Layers, TrendingUp } from 'lucide-react'
-import { fetchChanges, fetchChangeStats } from '../../lib/api'
+import { Activity, Radio, RefreshCw, PlusCircle, Pencil, Trash2, Database, Layers, TrendingUp, Cable } from 'lucide-react'
+import { fetchChanges, fetchChangeStats, resyncSources } from '../../lib/api'
 import type { CdcChange, CdcStats } from '../../types'
 import { cn } from '../../lib/utils'
 
@@ -224,6 +224,8 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) {
   const [expanded, setExpanded] = useState<string | null>(null)
   const [connected, setConnected] = useState(false)
   const [flash, setFlash] = useState(false)
+  const [resyncing, setResyncing] = useState(false)
+  const [resyncMsg, setResyncMsg] = useState<string | null>(null)
   // Live overlay: CDC events counted straight off the WebSocket stream since the
   // last server stats snapshot. The top KPIs/charts = authoritative server stats
   // (refreshed every 2.5s) + this overlay, so they move in lock-step with the
@@ -246,6 +248,25 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) {
     applyStats(s)
   }, [applyStats])
 
+  const doResync = useCallback(async () => {
+    setResyncing(true)
+    setResyncMsg('Restarting Debezium connectors…')
+    try {
+      const res = await resyncSources(true)
+      if (res) {
+        setResyncMsg(`Re-synced · ${res.healthy ?? 0}/${res.total ?? 0} connectors healthy`)
+        setTimeout(() => load(), 2500)
+      } else {
+        setResyncMsg('Re-sync failed — check ETL Guardian terminal')
+      }
+    } catch {
+      setResyncMsg('Re-sync failed — check ETL Guardian terminal')
+    } finally {
+      setResyncing(false)
+      setTimeout(() => setResyncMsg(null), 7000)
+    }
+  }, [load])
+
   useEffect(() => {
     load()
     const iv = setInterval(() => fetchChangeStats(15).then((s) => applyStats(s)), 2500)
@@ -346,6 +367,18 @@ export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) {
             connected ? 'border-emerald-500/40 bg-emerald-500/10 text-emerald-300' : 'border-rose-500/40 bg-rose-500/10 text-rose-300')}>
             <Radio className={cn('h-3 w-3', connected && 'animate-pulse')} /> {connected ? 'STREAMING' : 'OFFLINE'}
           </span>
+          {resyncMsg && (
+            <span className="hidden text-[10px] text-amber-300/90 md:inline">{resyncMsg}</span>
+          )}
+          <button
+            type="button"
+            onClick={doResync}
+            disabled={resyncing}
+            title="Restart all Debezium source connectors so CDC catches up after a database outage"
+            className="flex items-center gap-1 rounded border border-amber-500/40 bg-amber-500/10 px-2 py-1 text-[10px] font-medium text-amber-300 hover:bg-amber-500/20 disabled:opacity-60"
+          >
+            <Cable className={cn('h-3 w-3', resyncing && 'animate-spin')} /> {resyncing ? 'Re-syncing…' : 'Re-sync sources'}
+          </button>
           <button type="button" onClick={load} className="flex items-center gap-1 rounded border border-border/60 px-2 py-1 text-[10px] text-foreground-muted hover:text-docker">
             <RefreshCw className="h-3 w-3" /> Refresh
           </button>

ui/src/lib/api.ts

@@ -115,6 +115,18 @@ export async function fetchChangeStats(minutes = 15): Promise<CdcStats | null> {
   return fetchJson<CdcStats>(`/api/changes/stats?minutes=${minutes}`, 8000)
 }
 
+export type ConnectorState = { name: string; state?: string; failed?: number[] }
+export type ResyncResult = { ok: boolean; restarted?: string[]; healthy?: number; total?: number; after?: ConnectorState[] }
+
+export async function resyncSources(force = true): Promise<ResyncResult | null> {
+  const r = await fetch('/api/pipeline/streaming/resync', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ force }),
+  })
+  return r.ok ? ((await r.json()) as ResyncResult) : null
+}
+
 export async function fetchAgentOpsStatus() {
   return fetchJson<Record<string, unknown>>('/api/agent-ops/status', 8000)
 }