feat(llm): approval-gated movement executor + rag-api agent wiring + agent UI

- decide_approval now executes the underlying data movement when an approved request carries an executor=movement payload (real human-gated executor). - rag-api service gets OPENMETADATA_* (via atc.env) + COMMAND_CENTER_URL so it can sync the catalog and call platform tools. - Knowledge Chat gains an Agent-mode toggle (SSE tool-loop with step chips) and a 'Sync catalog' button.
2026-06-27 03:28:18 +02:00
parent 8263a03190
commit 9fb5b0a780
3 changed files with 152 additions and 5 deletions
@@ -45,6 +45,7 @@ from sql_console import router as sql_router
 from agent_ops import router as agent_ops_router, agent_dml_loop, etl_agent_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
 from dataflow import router as dataflow_router
 from pii_catalog import router as pii_router
 from ssh_terminal import ssh_session
@@ -1084,6 +1085,16 @@ async def decide_approval(approval_id: str, body: ApprovalDecision):
        )
    if not item:
        return {"error": "not found"}
+
+    # Approval-gated executor: run the underlying action once a supervisor approves it.
+    if item.get("status") == "approved":
+        payload = item.get("payload") or {}
+        if isinstance(payload, dict) and payload.get("executor") == "movement":
+            mid = payload.get("movement_id")
+            if mid in MOVEMENT_BY_ID:
+                add_feed("etl-guardian", f"Approval {approval_id} granted — executing movement '{mid}'", "info")
+                asyncio.create_task(trigger_and_watch(mid, payload.get("conf")))
+
    return {"ok": True, "approval": item}


@@ -87,6 +87,8 @@ services:
  rag-api:
    build: ../atc-data-quality/rag-api
    restart: unless-stopped
+    env_file:
+      - atc.env
    environment:
      CHROMA_HOST: chromadb
      CHROMA_PORT: 8000
@@ -95,11 +97,14 @@ services:
      LLM_MODEL: gpt-4o
      LLM_API_KEY: sk-local
      RAG_DATA_DIR: /data
+      OPENMETADATA_URL: ${OPENMETADATA_URL:-http://10.0.21.47:8585}
+      COMMAND_CENTER_URL: http://api:3201
    volumes:
      - rag_data:/data
    depends_on:
      - chromadb
      - docling-serve
+      - api

  dq-api:
    build: ../atc-data-quality/dq-api
@@ -1,5 +1,5 @@
 import { useCallback, useEffect, useRef, useState } from 'react'
-import { BookOpen, FileText, Loader2, MessageSquare, RefreshCw, RotateCcw, Send, Upload } from 'lucide-react'
+import { Bot, BookOpen, Database, FileText, Loader2, MessageSquare, RefreshCw, RotateCcw, Send, Upload, Wrench } from 'lucide-react'
 import { cn } from '../../lib/utils'
 import { subTabActive, subTabIdle } from '../../lib/tabActive'

@@ -14,7 +14,8 @@ type StoredDoc = {
  bytes?: number
 }
 type Source = { source?: string; chunk?: number; preview?: string }
-type ChatMsg = { role: 'user' | 'assistant'; content: string; sources?: Source[] }
+type AgentStep = { tool: string; input?: Record<string, unknown> }
+type ChatMsg = { role: 'user' | 'assistant'; content: string; sources?: Source[]; steps?: AgentStep[] }

 type Health = { ok: boolean; chroma: boolean; docling: boolean; llm: boolean; embed_model?: string }

@@ -34,6 +35,9 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) {
  const [selectedDocId, setSelectedDocId] = useState<string | null>(null)
  const [summarizing, setSummarizing] = useState(false)
  const [reindexing, setReindexing] = useState<string | null>(null)
+  const [agentMode, setAgentMode] = useState(false)
+  const [syncing, setSyncing] = useState(false)
+  const [catalogDocs, setCatalogDocs] = useState<number | null>(null)
  const bottomRef = useRef<HTMLDivElement>(null)

  const loadMeta = useCallback(async () => {
@@ -57,9 +61,19 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) {
    }
  }, [])

+  const loadCatalog = useCallback(async () => {
+    try {
+      const r = await fetch('/rag/catalog/status')
+      if (r.ok) setCatalogDocs((await r.json()).documents ?? 0)
+    } catch {
+      setCatalogDocs(null)
+    }
+  }, [])
+
  useEffect(() => {
    loadMeta()
-  }, [loadMeta])
+    loadCatalog()
+  }, [loadMeta, loadCatalog])

  useEffect(() => {
    bottomRef.current?.scrollIntoView({ behavior: 'smooth' })
@@ -146,12 +160,95 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) {
    }
  }

+  const onSyncCatalog = async () => {
+    setSyncing(true)
+    setError(null)
+    try {
+      const r = await fetch('/rag/catalog/sync', { method: 'POST' })
+      const j = await r.json()
+      if (!r.ok || !j.ok) {
+        setError(j.error || 'Catalog sync failed')
+        return
+      }
+      const c = j.counts || {}
+      setMessages((m) => [...m, {
+        role: 'assistant',
+        content: `Catalog synced — ${j.documents} entries (${c.tables ?? 0} tables, ${c.pii_datasets ?? 0} PII datasets, ${c.flows ?? 0} lineage edges, ${c.movements ?? 0} movements). Agent mode can now answer catalog/PII/lineage questions.`,
+      }])
+      loadCatalog()
+    } catch {
+      setError('Catalog sync request failed')
+    } finally {
+      setSyncing(false)
+    }
+  }
+
+  const onSendAgent = async (msg: string) => {
+    setLoading(true)
+    const steps: AgentStep[] = []
+    let answer = ''
+    let idx = -1
+    setMessages((m) => {
+      idx = m.length
+      return [...m, { role: 'assistant', content: '', steps: [] }]
+    })
+    const patch = () =>
+      setMessages((m) => m.map((x, i) => (i === idx ? { ...x, content: answer, steps: [...steps] } : x)))
+    try {
+      const r = await fetch('/rag/agent', {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ message: msg, max_steps: 5 }),
+      })
+      if (!r.ok || !r.body) {
+        setError('Agent unavailable')
+        return
+      }
+      const reader = r.body.getReader()
+      const decoder = new TextDecoder()
+      let buf = ''
+      for (;;) {
+        const { done, value } = await reader.read()
+        if (done) break
+        buf += decoder.decode(value, { stream: true })
+        const events = buf.split('\n\n')
+        buf = events.pop() || ''
+        for (const ev of events) {
+          const eMatch = ev.match(/^event: (.+)$/m)
+          const dMatch = ev.match(/^data: (.+)$/m)
+          if (!eMatch || !dMatch) continue
+          const type = eMatch[1].trim()
+          let data: Record<string, unknown> = {}
+          try { data = JSON.parse(dMatch[1]) } catch { continue }
+          if (type === 'step') {
+            steps.push({ tool: String(data.tool), input: data.input as Record<string, unknown> })
+            patch()
+          } else if (type === 'token') {
+            answer += String(data.t || '')
+            patch()
+          } else if (type === 'error') {
+            setError(String(data.error || 'Agent error'))
+          }
+        }
+      }
+    } catch {
+      setError('Failed to reach agent')
+    } finally {
+      patch()
+      setLoading(false)
+    }
+  }
+
  const onSend = async () => {
    const msg = input.trim()
    if (!msg || loading) return
    setInput('')
    setError(null)
    setMessages((m) => [...m, { role: 'user', content: msg }])
+    if (agentMode) {
+      await onSendAgent(msg)
+      return
+    }
    setLoading(true)
    try {
      const r = await fetch('/rag/chat', {
@@ -233,6 +330,31 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) {
            <button type="button" onClick={createCollection} className={cn('shrink-0 rounded px-2 py-1 text-[10px]', subTabIdle)}>Add</button>
          </div>

+          <h3 className="mb-2 text-[10px] font-semibold uppercase tracking-wider text-foreground-faint">Platform agent</h3>
+          <button
+            type="button"
+            onClick={() => setAgentMode((v) => !v)}
+            className={cn(
+              'mb-1 flex w-full items-center justify-between rounded border px-2 py-1.5 text-[11px] transition-colors',
+              agentMode ? 'border-docker/50 bg-docker/15 text-docker' : 'border-border text-foreground-muted',
+            )}
+          >
+            <span className="flex items-center gap-1"><Bot className="h-3.5 w-3.5" />Agent mode {agentMode ? 'on' : 'off'}</span>
+            <span className={cn('h-2 w-2 rounded-full', agentMode ? 'bg-docker' : 'bg-foreground-faint/40')} />
+          </button>
+          <p className="mb-2 text-[9px] text-foreground-faint">
+            Live tools: catalog, PII, CDC, movements + approval-gated triggers.
+          </p>
+          <button
+            type="button"
+            onClick={onSyncCatalog}
+            disabled={syncing}
+            className={cn('mb-4 flex w-full items-center justify-center gap-1 rounded px-2 py-1 text-[10px]', subTabIdle)}
+          >
+            {syncing ? <Loader2 className="h-3 w-3 animate-spin" /> : <Database className="h-3 w-3" />}
+            Sync catalog{catalogDocs != null ? ` (${catalogDocs})` : ''}
+          </button>
+
          <h3 className="mb-2 text-[10px] font-semibold uppercase tracking-wider text-foreground-faint">Ingest documents</h3>
          <label className={cn('flex cursor-pointer flex-col items-center rounded-lg border-2 border-dashed border-border px-3 py-4 text-center hover:border-docker/40', ingesting && 'opacity-50')}>
            <Upload className="mb-1 h-6 w-6 text-docker opacity-60" />
@@ -308,7 +430,16 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) {
            )}
            {messages.map((m, i) => (
              <div key={i} className={cn('mb-3 max-w-[90%] rounded-lg px-3 py-2 text-[12px]', m.role === 'user' ? 'ml-auto bg-docker/20 text-foreground' : 'bg-surface-overlay text-foreground-muted')}>
-                <p className="whitespace-pre-wrap leading-relaxed">{m.content}</p>
+                {m.steps && m.steps.length > 0 && (
+                  <div className="mb-2 flex flex-wrap gap-1">
+                    {m.steps.map((s, j) => (
+                      <span key={j} className="inline-flex items-center gap-1 rounded-full bg-docker/15 px-2 py-0.5 text-[9px] text-docker">
+                        <Wrench className="h-2.5 w-2.5" />{s.tool}
+                      </span>
+                    ))}
+                  </div>
+                )}
+                <p className="whitespace-pre-wrap leading-relaxed">{m.content || (loading && i === messages.length - 1 ? '…' : '')}</p>
                {m.sources && m.sources.length > 0 && (
                  <div className="mt-2 border-t border-border pt-2">
                    <p className="mb-1 text-[9px] font-semibold uppercase text-foreground-faint">Sources</p>
@@ -335,7 +466,7 @@ export function KnowledgeChatView({ onGpuActivity }: Props = {}) {
              value={input}
              onChange={(e) => setInput(e.target.value)}
              onKeyDown={(e) => e.key === 'Enter' && !e.shiftKey && (e.preventDefault(), onSend())}
-              placeholder="Ask a question about your ingested data…"
+              placeholder={agentMode ? 'Ask the platform agent (catalog, PII, CDC, run a movement…)' : 'Ask a question about your ingested data…'}
              className="min-w-0 flex-1 rounded-lg border border-border bg-surface-overlay px-3 py-2 text-[12px]"
              disabled={loading}
            />