feat: live pipeline activity log + on-shift roster

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
   }
 ]

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)."""

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

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(),
+            },
+        )
+    )

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)

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"]

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",

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 (
+    <div className={styles.card}>
+      <div className={styles.cardHead}>Activity log</div>
+      {rows.length === 0 ? (
+        <div className={styles.feedEmpty}>No activity yet</div>
+      ) : (
+        <div className={styles.activityList}>
+          {rows.map((e, i) => (
+            <div key={i} className={styles.activityRow}>
+              <span className={styles.activityTime}>{clockTime(e.ts)}</span>
+              <span className={`${styles.activityStage} ${styles[e.stage] || ""}`}>
+                {STAGE_LABEL[e.stage] || e.stage}
+              </span>
+              <span className={styles.activityText}>
+                {e.machine} {e.fault_code}
+                {e.detail ? ` — ${e.detail}` : ""}
+              </span>
+            </div>
+          ))}
+        </div>
+      )}
+    </div>
+  );
+}
+
 function EvalCard({ data }) {
   if (!data || !data.configs?.length) return null;
   return (
@@ -541,6 +612,8 @@ export default function Dashboard() {
 
         <LiveFeed faults={sortedFaults} t={t} />
 
+        <ActivityLog />
+
         <div className={styles.grid}>
           {loading ? (
             <>

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 {