From b6282243fa9a3ad4c2b1d348176cd1c0a75eee0c Mon Sep 17 00:00:00 2001 From: Billo Rani Date: Wed, 10 Jun 2026 16:25:53 +0000 Subject: [PATCH] feat: add manual trigger buttons for background processes - POST /api/processes/trigger-consolidation - POST /api/processes/trigger-extraction - POST /api/processes/trigger-decay - Dashboard buttons with status feedback - Added refresh() to usePolling hook --- crates/perspective-core/src/server.rs | 71 +++++++++++++++++++++ dashboard/src/hooks.ts | 11 +++- dashboard/src/pages/Processes.tsx | 91 ++++++++++++++++++++++++++- 3 files changed, 171 insertions(+), 2 deletions(-) diff --git a/crates/perspective-core/src/server.rs b/crates/perspective-core/src/server.rs index f1e973b..bc8b4c5 100644 --- a/crates/perspective-core/src/server.rs +++ b/crates/perspective-core/src/server.rs @@ -378,6 +378,71 @@ async fn handle_reflect(engine: &PerspectiveEngine, body: &str) -> (String, Stri ) } +/// Handle a POST /api/processes/trigger-consolidation request. +async fn handle_trigger_consolidation(engine: &PerspectiveEngine) -> (String, String) { + let tenants = engine.list_tenants().await.unwrap_or_default(); + let mut reports = Vec::new(); + for tenant_id in &tenants { + match engine.run_consolidation(tenant_id).await { + Ok(report) => { + reports.push(serde_json::json!({ + "tenant": tenant_id, + "duplicates": report.duplicates_found, + "promotable": report.promotable_count, + "communities": report.communities, + })); + } + Err(e) => { + tracing::warn!("trigger consolidation: {tenant_id} failed: {e}"); + reports.push(serde_json::json!({ + "tenant": tenant_id, + "error": format!("{e}"), + })); + } + } + } + let resp = serde_json::json!({ "ok": true, "reports": reports }); + let json = serde_json::to_string(&resp).unwrap_or_default(); + ( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nAccess-Control-Allow-Origin: *\r\n".into(), + json, + ) +} + +/// Handle a POST /api/processes/trigger-extraction request. +async fn handle_trigger_extraction(engine: &PerspectiveEngine) -> (String, String) { + match engine.process_extraction_batch().await { + Ok(count) => { + let resp = serde_json::json!({ "ok": true, "extracted": count }); + let json = serde_json::to_string(&resp).unwrap_or_default(); + ( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nAccess-Control-Allow-Origin: *\r\n".into(), + json, + ) + } + Err(e) => { + tracing::warn!("trigger extraction: failed: {e}"); + let resp = serde_json::json!({ "ok": false, "error": format!("{e}") }); + let json = serde_json::to_string(&resp).unwrap_or_default(); + ( + "HTTP/1.1 500 Internal Server Error\r\nContent-Type: application/json\r\nAccess-Control-Allow-Origin: *\r\n".into(), + json, + ) + } + } +} + +/// Handle a POST /api/processes/trigger-decay request. +async fn handle_trigger_decay(engine: &PerspectiveEngine) -> (String, String) { + engine.run_decay_tick().await; + let resp = serde_json::json!({ "ok": true }); + let json = serde_json::to_string(&resp).unwrap_or_default(); + ( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nAccess-Control-Allow-Origin: *\r\n".into(), + json, + ) +} + /// Handle a GET /api/logs request. /// Reads the last N lines from the perspective.log file. /// Supports ?limit=N (default 100) and ?filter=keyword query params. @@ -727,6 +792,12 @@ pub fn start_background_with_config( json, ) } + } else if method == "POST" && path == "/api/processes/trigger-consolidation" { + handle_trigger_consolidation(&engine).await + } else if method == "POST" && path == "/api/processes/trigger-extraction" { + handle_trigger_extraction(&engine).await + } else if method == "POST" && path == "/api/processes/trigger-decay" { + handle_trigger_decay(&engine).await } else if method == "GET" && path.starts_with("/api/memories") { // Parse optional ?q= and ?limit= query params let (q, limit) = if let Some(qs) = path.split_once('?') { diff --git a/dashboard/src/hooks.ts b/dashboard/src/hooks.ts index db61a76..9a981b8 100644 --- a/dashboard/src/hooks.ts +++ b/dashboard/src/hooks.ts @@ -8,6 +8,15 @@ function usePolling(fetcher: () => Promise, intervalMs = 5000) { const fetcherRef = useRef(fetcher); fetcherRef.current = fetcher; + const refreshNow = useRef(async () => { + try { + const result = await fetcherRef.current(); + setData(result); setError(null); + } catch (e: unknown) { + setError(e instanceof Error ? e.message : 'Unknown error'); + } + }); + useEffect(() => { let alive = true; @@ -27,7 +36,7 @@ function usePolling(fetcher: () => Promise, intervalMs = 5000) { return () => { alive = false; clearInterval(id); }; }, [intervalMs]); - return { data, error, loading }; + return { data, error, loading, refresh: refreshNow.current }; } function useOnce(fetcher: () => Promise) { diff --git a/dashboard/src/pages/Processes.tsx b/dashboard/src/pages/Processes.tsx index 1996fdd..b8654b1 100644 --- a/dashboard/src/pages/Processes.tsx +++ b/dashboard/src/pages/Processes.tsx @@ -1,3 +1,4 @@ +import { useState } from 'react'; import { useProcesses } from '../hooks'; function StatusDot({ active }: { active: boolean }) { @@ -10,6 +11,27 @@ function StatusDot({ active }: { active: boolean }) { ); } +function TriggerButton({ onClick, label }: { onClick: () => void; label: string }) { + return ( + + ); +} + +function TriggerFeedback({ status }: { status: string | null }) { + if (!status) return null; + const isError = status.startsWith('Error'); + return ( +
+ {status} +
+ ); +} + function SectionCard({ title, children }: { title: string; children: React.ReactNode }) { return (
@@ -29,7 +51,10 @@ function KV({ label, value }: { label: string; value: string | number | null }) } export default function Processes() { - const { data, error, loading } = useProcesses(); + const { data, error, loading, refresh } = useProcesses(); + const [consolidationStatus, setConsolidationStatus] = useState(null); + const [extractionStatus, setExtractionStatus] = useState(null); + const [decayStatus, setDecayStatus] = useState(null); if (loading) return
Loading...
; if (error) return
{error}
; @@ -37,6 +62,58 @@ export default function Processes() { const qLen = data.extraction_queue.length; + async function triggerConsolidation() { + setConsolidationStatus('Running...'); + try { + const resp = await fetch('/api/processes/trigger-consolidation', { method: 'POST' }); + const json = await resp.json(); + if (json.ok) { + const reports = json.reports ?? []; + setConsolidationStatus(`Done: ${reports.map((r: any) => `${r.tenant}: ${r.duplicates ?? 0} dupes`).join(', ') || 'no tenants'}`); + } else { + setConsolidationStatus(`Error: ${json.error ?? 'unknown'}`); + } + } catch (e: any) { + setConsolidationStatus(`Error: ${e.message}`); + } + setTimeout(() => setConsolidationStatus(null), 5000); + refresh(); + } + + async function triggerExtraction() { + setExtractionStatus('Running...'); + try { + const resp = await fetch('/api/processes/trigger-extraction', { method: 'POST' }); + const json = await resp.json(); + if (json.ok) { + setExtractionStatus(`Done: extracted ${json.extracted} items`); + } else { + setExtractionStatus(`Error: ${json.error ?? 'unknown'}`); + } + } catch (e: any) { + setExtractionStatus(`Error: ${e.message}`); + } + setTimeout(() => setExtractionStatus(null), 5000); + refresh(); + } + + async function triggerDecay() { + setDecayStatus('Running...'); + try { + const resp = await fetch('/api/processes/trigger-decay', { method: 'POST' }); + const json = await resp.json(); + if (json.ok) { + setDecayStatus('Done: decay tick completed'); + } else { + setDecayStatus(`Error: ${json.error ?? 'unknown'}`); + } + } catch (e: any) { + setDecayStatus(`Error: ${e.message}`); + } + setTimeout(() => setDecayStatus(null), 5000); + refresh(); + } + return (

Background Processes

@@ -54,6 +131,10 @@ export default function Processes() { +
+ + +
@@ -71,6 +152,10 @@ export default function Processes() { {qLen === 0 && !data.extraction_loop_active && (

Extraction loop not started

)} +
+ + +
@@ -85,6 +170,10 @@ export default function Processes() { +
+ + +