Skip to content

feat: add manual trigger buttons for background processes #27

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
swapkats merged 1 commit into from
Jun 10, 2026
Merged

feat: add manual trigger buttons for background processes #27

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 71 additions & 0 deletions crates/perspective-core/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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('?') {
Expand Down
11 changes: 10 additions & 1 deletion dashboard/src/hooks.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,15 @@ function usePolling<T>(fetcher: () => Promise<T>, 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;

Expand All @@ -27,7 +36,7 @@ function usePolling<T>(fetcher: () => Promise<T>, intervalMs = 5000) {
return () => { alive = false; clearInterval(id); };
}, [intervalMs]);

return { data, error, loading };
return { data, error, loading, refresh: refreshNow.current };
}

function useOnce<T>(fetcher: () => Promise<T>) {
Expand Down
91 changes: 90 additions & 1 deletion dashboard/src/pages/Processes.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { useState } from 'react';
import { useProcesses } from '../hooks';

function StatusDot({ active }: { active: boolean }) {
Expand All @@ -10,6 +11,27 @@ function StatusDot({ active }: { active: boolean }) {
);
}

function TriggerButton({ onClick, label }: { onClick: () => void; label: string }) {
return (
<button
onClick={onClick}
className="px-3 py-1.5 text-xs font-medium bg-zinc-800 hover:bg-zinc-700 border border-zinc-700 hover:border-zinc-600 rounded-lg text-zinc-300 hover:text-zinc-100 transition-all duration-150 active:scale-95"
>
{label}
</button>
);
}

function TriggerFeedback({ status }: { status: string | null }) {
if (!status) return null;
const isError = status.startsWith('Error');
return (
<div className={`text-xs mt-2 px-2 py-1 rounded ${isError ? 'text-red-400 bg-red-400/10' : 'text-emerald-400 bg-emerald-400/10'}`}>
{status}
</div>
);
}

function SectionCard({ title, children }: { title: string; children: React.ReactNode }) {
return (
<div className="bg-zinc-900 border border-zinc-800 rounded-xl p-5">
Expand All @@ -29,14 +51,69 @@ 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<string | null>(null);
const [extractionStatus, setExtractionStatus] = useState<string | null>(null);
const [decayStatus, setDecayStatus] = useState<string | null>(null);

if (loading) return <div className="text-zinc-500 animate-pulse">Loading...</div>;
if (error) return <div className="text-red-400">{error}</div>;
if (!data) return null;

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 (
<div className="space-y-6">
<h2 className="text-xl font-bold">Background Processes</h2>
Expand All @@ -54,6 +131,10 @@ export default function Processes() {
<KV label="Items processed" value={data.consolidation.items_processed} />
<KV label="Merges" value={data.consolidation.merges} />
<KV label="Promotions" value={data.consolidation.promotions} />
<div className="mt-3 pt-3 border-t border-zinc-800/50">
<TriggerButton onClick={triggerConsolidation} label="Run Now" />
<TriggerFeedback status={consolidationStatus} />
</div>
</SectionCard>

<SectionCard title="Extraction Queue">
Expand All @@ -71,6 +152,10 @@ export default function Processes() {
{qLen === 0 && !data.extraction_loop_active && (
<p className="text-zinc-600 text-xs py-4 text-center">Extraction loop not started</p>
)}
<div className="mt-3 pt-3 border-t border-zinc-800/50">
<TriggerButton onClick={triggerExtraction} label="Run Now" />
<TriggerFeedback status={extractionStatus} />
</div>
</SectionCard>

<SectionCard title="Decay & GC">
Expand All @@ -85,6 +170,10 @@ export default function Processes() {
<KV label="GC candidates" value={data.decay.gc_candidates} />
<KV label="Items collected" value={data.decay.items_collected} />
<KV label="Last GC" value={data.decay.last_gc_run} />
<div className="mt-3 pt-3 border-t border-zinc-800/50">
<TriggerButton onClick={triggerDecay} label="Run Now" />
<TriggerFeedback status={decayStatus} />
</div>
</SectionCard>
</div>
</div>
Expand Down
Loading