From cb35741a8946f9e1b2ac970b8a445638fed468b0 Mon Sep 17 00:00:00 2001 From: goodnight Date: Sat, 18 Jul 2026 19:56:12 +0100 Subject: [PATCH] feat: live pipeline activity log (SSE) + roster with real on-shift techs --- data/shifts.json | 26 +++++------ src/torq/api/routes.py | 24 ++++++++++ src/torq/dispatch/approval.py | 11 +++++ src/torq/events/live.py | 31 +++++++++++++ src/torq/pipeline.py | 4 ++ tests/test_activity_log.py | 38 ++++++++++++++++ web/src/api.js | 3 ++ web/src/pages/Dashboard.jsx | 73 ++++++++++++++++++++++++++++++ web/src/pages/Dashboard.module.css | 59 ++++++++++++++++++++++++ 9 files changed, 256 insertions(+), 13 deletions(-) create mode 100644 tests/test_activity_log.py diff --git a/data/shifts.json b/data/shifts.json index 1ec10c5..61bdbf5 100644 --- a/data/shifts.json +++ b/data/shifts.json @@ -1,37 +1,37 @@ [ { "name": "Mohamed Arbi Nsibi", - "phone": "+21620000001", + "phone": "+21629051246", "lang": "ar", "skills": ["electromechanical", "general"], "on_shift": true }, { - "name": "Aziz Sayadi", - "phone": "+21620000002", + "name": "Adam Ridene", + "phone": "+21694420276", "lang": "fr", "skills": ["packaging", "general"], "on_shift": true }, { - "name": "Adam Ridene", - "phone": "+21620000003", + "name": "Youcef", + "phone": "+21695905163", "lang": "fr", - "skills": ["electromechanical", "packaging"], + "skills": ["electromechanical", "packaging", "general"], "on_shift": true }, + { + "name": "Aziz Sayadi", + "phone": "+21620000002", + "lang": "fr", + "skills": ["packaging", "general"], + "on_shift": false + }, { "name": "Mahdi Bani", "phone": "+21620000004", "lang": "en", "skills": ["general"], - "on_shift": true - }, - { - "name": "Youcef", - "phone": "+21620000005", - "lang": "fr", - "skills": ["electromechanical", "general"], "on_shift": false } ] diff --git a/src/torq/api/routes.py b/src/torq/api/routes.py index edfd9bc..c44e3a4 100644 --- a/src/torq/api/routes.py +++ b/src/torq/api/routes.py @@ -154,6 +154,30 @@ def recent_events(): return [event for _seq, event in live.RECENT_FAULTS] +@router.get("/events/activity/stream") +async def activity_stream(request: Request): + """SSE endpoint: streams pipeline activity (received -> diagnosed -> dispatched).""" + + async def generate(): + last_seq = 0 + while True: + for seq, event in list(live.RECENT_ACTIVITY): + if seq > last_seq: + yield f"event: activity\ndata: {json.dumps(event)}\n\n" + last_seq = seq + if await request.is_disconnected(): + break + await asyncio.sleep(0.5) + + return StreamingResponse(generate(), media_type="text/event-stream") + + +@router.get("/events/activity/recent") +def recent_activity(): + """Return the recent pipeline activity entries (oldest first).""" + return [event for _seq, event in live.RECENT_ACTIVITY] + + @router.get("/eval") def eval_results(): """Precomputed retrieval-eval results (dense vs hybrid vs hybrid+rerank).""" diff --git a/src/torq/dispatch/approval.py b/src/torq/dispatch/approval.py index 5caccde..04a4dde 100644 --- a/src/torq/dispatch/approval.py +++ b/src/torq/dispatch/approval.py @@ -3,6 +3,7 @@ from torq.agent.schemas import WorkOrder, _now from torq.db import models from torq.dispatch import notify, routing +from torq.events import live from torq.workorder.pdf import render_pdf @@ -23,6 +24,7 @@ def reject(wo_id: str) -> WorkOrder | None: return None wo.status = "rejected" models.save(wo) + live.push_activity("rejected", wo.machine, wo.fault_code, wo_id=wo.id) return wo @@ -31,10 +33,15 @@ def approve(wo_id: str) -> tuple[WorkOrder, dict] | None: wo = models.get(wo_id) if not wo: return None + live.push_activity("approved", wo.machine, wo.fault_code, wo_id=wo.id) tech = routing.choose_technician(wo) if not tech: wo.status = "approved" # approved but no one on shift to take it models.save(wo) + live.push_activity( + "dispatch_failed", wo.machine, wo.fault_code, + detail="no technician on shift", wo_id=wo.id, + ) return wo, {"channel": "none", "error": "no technician available"} try: @@ -47,4 +54,8 @@ def approve(wo_id: str) -> tuple[WorkOrder, dict] | None: wo.assigned_to = tech.get("name") wo.dispatched_at = _now() models.save(wo) + live.push_activity( + "dispatched", wo.machine, wo.fault_code, + detail=f"{delivery.get('channel', '?')} to {tech.get('name', '?')}", wo_id=wo.id, + ) return wo, delivery diff --git a/src/torq/events/live.py b/src/torq/events/live.py index de1c611..d16bb15 100644 --- a/src/torq/events/live.py +++ b/src/torq/events/live.py @@ -1,6 +1,7 @@ """Thread-safe shared state between the MQTT listener and the SSE stream.""" from collections import deque +from datetime import datetime, timezone from itertools import count from torq.events.schemas import MachineFaultEvent @@ -16,3 +17,33 @@ def push(event: MachineFaultEvent) -> None: """Push a validated fault event into the recent buffer.""" RECENT_FAULTS.append((next(_seq), event.model_dump())) + + +# Pipeline activity log: one entry per stage of the fault -> fix flow, streamed +# to the dashboard so operators watch faults move through diagnosis and dispatch +# in near real-time. Same (seq, dict) shape as RECENT_FAULTS. +RECENT_ACTIVITY: deque = deque(maxlen=100) +_act_seq = count(1) + + +def push_activity( + stage: str, + machine: str = "", + fault_code: str = "", + detail: str = "", + wo_id: str = "", +) -> None: + """Record one pipeline stage (received, diagnosing, dispatched, ...).""" + RECENT_ACTIVITY.append( + ( + next(_act_seq), + { + "stage": stage, + "machine": machine, + "fault_code": fault_code, + "detail": detail, + "wo_id": wo_id, + "ts": datetime.now(timezone.utc).isoformat(), + }, + ) + ) diff --git a/src/torq/pipeline.py b/src/torq/pipeline.py index c59cdc7..c050190 100644 --- a/src/torq/pipeline.py +++ b/src/torq/pipeline.py @@ -6,6 +6,7 @@ from torq.agent.schemas import WorkOrder from torq.db import models from torq.dispatch import approval +from torq.events import live from torq.workorder.generate import build_work_order @@ -20,6 +21,9 @@ def handle_fault( if fault_arrived_at is None: fault_arrived_at = datetime.now(timezone.utc).isoformat() models.init_db() + live.push_activity("fault_received", machine, fault_code) + live.push_activity("diagnosing", machine, fault_code, detail="reading manuals + repair history") diag = diagnose(fault_code, machine, context) wo = build_work_order(diag, translate=translate, fault_arrived_at=fault_arrived_at) + live.push_activity("work_order_created", machine, fault_code, detail=diag.root_cause, wo_id=wo.id) return approval.submit(wo) diff --git a/tests/test_activity_log.py b/tests/test_activity_log.py new file mode 100644 index 0000000..3c09206 --- /dev/null +++ b/tests/test_activity_log.py @@ -0,0 +1,38 @@ +"""The pipeline emits activity-log stages that the dashboard SSE stream serves.""" + +from unittest.mock import MagicMock, patch + +from torq.agent.schemas import Diagnosis, WorkOrder +from torq.events import live + + +def _stages(): + return [e["stage"] for _seq, e in live.RECENT_ACTIVITY] + + +def test_push_activity_records_shape(): + live.RECENT_ACTIVITY.clear() + live.push_activity("dispatched", "CM-350", "E-201", detail="whatsapp to Adam", wo_id="ab12") + _seq, e = live.RECENT_ACTIVITY[-1] + assert e["stage"] == "dispatched" + assert e["machine"] == "CM-350" + assert e["fault_code"] == "E-201" + assert e["wo_id"] == "ab12" + assert e["ts"] # timestamp present + + +@patch("torq.pipeline.approval.submit", side_effect=lambda wo: wo) +@patch("torq.pipeline.build_work_order") +@patch("torq.pipeline.diagnose") +@patch("torq.pipeline.models.init_db") +def test_handle_fault_emits_stages(_init, mock_diag, mock_build, _submit): + from torq.pipeline import handle_fault + + live.RECENT_ACTIVITY.clear() + mock_diag.return_value = Diagnosis(fault_code="E-201", root_cause="seized bearing") + mock_build.return_value = WorkOrder(id="wo1", fault_code="E-201", root_cause="seized bearing") + + handle_fault("E-201", "CM-350") + + stages = _stages() + assert stages == ["fault_received", "diagnosing", "work_order_created"] diff --git a/web/src/api.js b/web/src/api.js index e50a5aa..0e35a27 100644 --- a/web/src/api.js +++ b/web/src/api.js @@ -17,6 +17,9 @@ export const getTrend = () => export const getFaultsPerMachine = () => j("/metrics/faults-per-machine").catch(() => null); +export const getRecentActivity = () => + j("/events/activity/recent").catch(() => []); + export const reportFault = (body) => j("/faults", { method: "POST", diff --git a/web/src/pages/Dashboard.jsx b/web/src/pages/Dashboard.jsx index 49aa0af..56ae67f 100644 --- a/web/src/pages/Dashboard.jsx +++ b/web/src/pages/Dashboard.jsx @@ -167,6 +167,77 @@ function formatTime(ts, t) { return new Date(ts).toLocaleDateString(); } +const STAGE_LABEL = { + fault_received: "Fault detected", + diagnosing: "Diagnosing", + work_order_created: "Work order ready", + approved: "Approved", + dispatched: "Dispatched", + dispatch_failed: "Dispatch failed", + rejected: "Rejected", +}; + +function clockTime(ts) { + if (!ts) return ""; + return new Date(ts).toLocaleTimeString([], { + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + }); +} + +// Live pipeline log. Backfills from REST, then streams new stages over SSE +// (EventSource auto-reconnects on drop, so no polling and no manual retry). +function ActivityLog() { + const [entries, setEntries] = useState([]); + + useEffect(() => { + let alive = true; + api.getRecentActivity().then((rows) => { + if (alive && Array.isArray(rows)) setEntries(rows); + }); + const es = new EventSource("/api/events/activity/stream"); + es.addEventListener("activity", (ev) => { + try { + const e = JSON.parse(ev.data); + setEntries((prev) => [...prev, e].slice(-100)); + } catch { + /* ignore malformed frame */ + } + }); + return () => { + alive = false; + es.close(); + }; + }, []); + + const rows = [...entries].reverse().slice(0, 20); // newest first + + return ( +
+
Activity log
+ {rows.length === 0 ? ( +
No activity yet
+ ) : ( +
+ {rows.map((e, i) => ( +
+ {clockTime(e.ts)} + + {STAGE_LABEL[e.stage] || e.stage} + + + {e.machine} {e.fault_code} + {e.detail ? ` — ${e.detail}` : ""} + +
+ ))} +
+ )} +
+ ); +} + function EvalCard({ data }) { if (!data || !data.configs?.length) return null; return ( @@ -541,6 +612,8 @@ export default function Dashboard() { + +
{loading ? ( <> diff --git a/web/src/pages/Dashboard.module.css b/web/src/pages/Dashboard.module.css index b2d87d2..555d9da 100644 --- a/web/src/pages/Dashboard.module.css +++ b/web/src/pages/Dashboard.module.css @@ -750,6 +750,65 @@ transition: color 0.35s; } +/* ── Activity log ── */ + +.activityList { + display: flex; + flex-direction: column; + max-height: 280px; + overflow-y: auto; +} + +.activityRow { + display: flex; + align-items: baseline; + gap: 10px; + padding: 7px 0; + border-bottom: 1px solid var(--border-color); + font-family: "ALTGumbo", system-ui, sans-serif; + font-size: 13px; + transition: border-color 0.35s; +} + +.activityRow:last-child { + border-bottom: none; +} + +.activityTime { + flex: 0 0 auto; + font-family: "SF Mono", "Cascadia Code", "Consolas", monospace; + font-size: 11px; + color: var(--text-muted); + opacity: 0.7; +} + +.activityStage { + flex: 0 0 auto; + font-weight: 700; + font-size: 10px; + text-transform: uppercase; + letter-spacing: 0.05em; + padding: 2px 8px; + border-radius: 6px; + background: var(--bg-secondary); + color: var(--text-muted); +} + +.activityStage.fault_received { background: rgba(248, 81, 73, 0.14); color: #dc3545; } +.activityStage.diagnosing { background: rgba(88, 133, 240, 0.14); color: #5885f0; } +.activityStage.work_order_created { background: rgba(88, 133, 240, 0.14); color: #5885f0; } +.activityStage.approved { background: rgba(46, 160, 67, 0.14); color: #2ea043; } +.activityStage.dispatched { background: rgba(46, 160, 67, 0.18); color: #2ea043; } +.activityStage.dispatch_failed, +.activityStage.rejected { background: rgba(248, 81, 73, 0.14); color: #dc3545; } + +.activityText { + flex: 1; + color: var(--text-primary); + line-height: 1.5; + transition: color 0.35s; +} + /* ── Confidence badge ── */ .lowConfidence {