Refactor: manual operator reporting as primary fault ingestion path

README.md

@@ -21,13 +21,25 @@
 
 TORQ is an event-driven predictive-maintenance engine that turns machine faults into grounded diagnoses, approval-ready work orders, and technician dispatches.
 
-A fault arrives over REST or MQTT. TORQ retrieves relevant OEM guidance and past repairs, asks an OpenAI-compatible model for a structured diagnosis, creates an English/French/Arabic work order, and queues it for supervisor review. Resolved repairs return to the knowledge base so the next diagnosis can reuse what worked.
+A fault arrives — from a floor operator, a REST API call, or an MQTT event. TORQ retrieves relevant OEM guidance and past repairs, asks an OpenAI-compatible model for a structured diagnosis, creates an English/French/Arabic work order, and queues it for supervisor review. Resolved repairs return to the knowledge base so the next diagnosis can reuse what worked.
+
+### Three-tier fault ingestion
+
+TORQ is designed to work in plants with *any* level of digital connectivity. Faults enter through one of three tiers, each tagged with a `source` label visible in the dashboard:
+
+| Tier | Method | source tag | Hardware needed | For plants that… |
+|------|--------|------------|-----------------|------------------|
+| **1 — Manual** | Dashboard form or REST `POST /api/faults` | `manual` | None | Have no digital machine connectivity at all — operators report what they see, hear, or smell. |
+| **2 — REST API** | Any system that can `POST /api/faults` | `rest` | Existing IT systems | Already use a CMMS, ERP, or custom maintenance tool that can emit fault events. |
+| **3 — MQTT** | MQTT broker subscription | `mqtt` | ESP32 bridge (~$20) or existing SCADA | Want automated ingestion — a $20 ESP32 with clamp-on sensors bridges legacy machines to the broker, or modern PLCs publish directly to MQTT. |
+
+**Tier 1 is the default.** No retrofitting, no sensors, no PLC integration required. The dashboard's fault-report form is the primary ingestion path. Tiers 2 and 3 layer on as plants grow their digital footprint.
 
 ## Why TORQ?
 
 | Traditional maintenance flow | TORQ |
 | --- | --- |
-| Faults wait to be noticed and triaged | REST and MQTT fault ingestion starts the workflow immediately |
+| Faults wait to be noticed and triaged | Operators report faults from the dashboard in seconds — REST and MQTT are optional add-ons |
 | Technicians search manuals and logs by hand | Dense + BM25 retrieval surfaces relevant guidance and repair history |
 | Diagnosis quality depends on who is available | Structured, source-aware AI diagnosis creates a consistent starting point |
 | Work orders are manually written and translated | Approval-ready EN/FR/AR work orders and PDFs are generated automatically |
@@ -139,7 +151,7 @@ When a repair is marked resolved, TORQ records the actual fix, technician notes,
 
 ### Supervisor and operations views
 
-The built-in dashboard supports fault simulation, approval, rejection, resolution, and downtime metrics. A separate React dashboard adds work-order details, multilingual content, PDF downloads, retrieval evaluation charts, and an editable ROI calculator.
+The built-in dashboard supports manual fault reporting, approval, rejection, resolution, and downtime metrics. A separate React dashboard adds work-order details, multilingual content (with source badges — manual vs. mqtt), PDF downloads, retrieval evaluation charts, and an editable ROI calculator.
 
 ### Multiple integration surfaces
 

src/torq/agent/schemas.py

@@ -38,6 +38,7 @@ class WorkOrder(BaseModel):
     status: str = "pending"  # pending | approved | rejected | dispatched | resolved | failed
     assigned_to: str | None = None
     confidence: float = 0.5
+    source: str = "manual"
     fault_arrived_at: str | None = None
     created_at: str = Field(default_factory=_now)
     dispatched_at: str | None = None

src/torq/api/main.py

@@ -6,6 +6,8 @@
 from contextlib import asynccontextmanager
 from typing import AsyncIterator
 
+from pathlib import Path
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import HTMLResponse
@@ -24,9 +26,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 
     models.init_db()
     mqtt_client = None
-    if settings.enable_fallbacks:
-        print("[MQTT] fallbacks enabled, skipping broker connection")
-    else:
+    if settings.mqtt_broker_url:
         mqtt_client = listener.build_client()
         live.mqtt_client = mqtt_client
         try:
@@ -36,7 +36,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
                 keepalive=60,
             )
             mqtt_client.loop_start()
-            print("[MQTT] background listener started")
+            print(f"[MQTT] listening on {settings.mqtt_broker_url}:{settings.mqtt_port}")
         except Exception as exc:
             print(f"[MQTT] broker unreachable ({exc}), continuing without live feed")
             mqtt_client = None
@@ -64,6 +64,11 @@ def dashboard() -> str:
     return DASHBOARD_HTML
 
 
+@app.get("/operator/report", response_class=HTMLResponse)
+def operator_report() -> str:
+    return Path(__file__).parent.parent.joinpath("operator", "report.html").read_text(encoding="utf-8")
+
+
 DASHBOARD_HTML = r"""
 <!doctype html><html><head><meta charset="utf-8">
 <title>TORQ Dashboard</title>
@@ -101,15 +106,18 @@ def dashboard() -> str:
 </style></head><body>
 <header>
   <h1>TORQ <span>Fault-to-Fix</span> &mdash; Supervisor Dashboard</h1>
-  <span class="mqtt-status"><span class="status-dot disconnected" id="mqtt-dot"></span> MQTT <span id="mqtt-label">disconnected</span></span>
+  <span class="mqtt-status" id="mqtt-status"></span>
 </header>
 <main>
   <div class="tiles" id="tiles"></div>
 
   <h2>&#9889; LIVE FAULT FEED</h2>
   <div id="live-feed"><div class="feed-empty">Waiting for faults&hellip;</div></div>
 
-  <button class="sim" onclick="simulate()">&#9889; Simulate fault (E-471, CM-350 Line 2)</button>
+  <div style="background:#111c28;border:1px solid #223;border-radius:10px;padding:14px 16px;margin-bottom:16px;display:flex;gap:12px;align-items:center;flex-wrap:wrap">
+    <a href="/operator/report" target="_blank" style="color:#e6edf3;font-size:14px;font-weight:700;text-decoration:none;background:#1f6feb;padding:8px 16px;border-radius:6px">&#128241; Report a fault (open on phone)</a>
+    <span style="color:#8b98a5;font-size:13px">Select machine → enter fault code → submitted</span>
+  </div>
   <h2>PENDING APPROVAL</h2><table id="pending"></table>
   <h2>ALL WORK ORDERS</h2><table id="all"></table>
 </main>
@@ -118,11 +126,6 @@ def dashboard() -> str:
 const post = p => fetch('/api'+p,{method:'POST'}).then(r=>r.json());
 const esc = s => String(s).replace(/[&<>"]/g,function(m){return{'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[m]||m});
 
-async function simulate(){
-  await fetch('/api/faults',{method:'POST',headers:{'Content-Type':'application/json'},
-    body:JSON.stringify({fault_code:'E-471',machine:'CM-350 Line 2',context:'Motor tripped after hours running.'})});
-  load();
-}
 async function approve(id){await post('/work-orders/'+id+'/approve');load();}
 async function reject(id){await post('/work-orders/'+id+'/reject');load();}
 async function resolve(id){
@@ -186,5 +189,18 @@ def dashboard() -> str:
       '<td>'+(w.status==='dispatched'?'<button onclick="resolve(\''+w.id+'\')">Mark fixed</button>':'')+'</td></tr>').join('');
 }
 load(); setInterval(load, 4000);
+
+// Show MQTT status (configured/connected/not configured)
+fetch('/api/health').then(r=>r.json()).then(h=>{
+  const m = h.integrations?.mqtt_broker || {};
+  const el = document.getElementById('mqtt-status');
+  if (!m.configured) {
+    el.innerHTML = '<span style="opacity:0.4">MQTT — not configured</span>';
+  } else {
+    const dot = m.connected ? 'connected' : 'disconnected';
+    const label = m.connected ? 'connected' : 'disconnected';
+    el.innerHTML = '<span class="status-dot '+dot+'"></span> MQTT <span>'+label+'</span>';
+  }
+}).catch(()=>{});
 </script></body></html>
 """

src/torq/api/routes.py

@@ -19,6 +19,8 @@
 
 router = APIRouter()
 
+_FAULT_CODES_CACHE: list[dict[str, str]] | None = None
+
 
 class MachineIn(BaseModel):
     id: str = Field(max_length=100)
@@ -31,6 +33,7 @@ class FaultIn(BaseModel):
     machine: str = ""
     context: str = ""
     translate: bool = True
+    source: str = "manual"
 
 
 class OutcomeIn(BaseModel):
@@ -53,6 +56,31 @@ def machine(machine_id: str) -> dict[str, Any]:
     return registered
 
 
+@router.get("/fault-codes")
+def fault_codes() -> list[dict[str, str]]:
+    """Return known fault codes from the scenarios file — used by the operator report form."""
+    global _FAULT_CODES_CACHE
+    if _FAULT_CODES_CACHE is not None:
+        return _FAULT_CODES_CACHE
+    try:
+        scenarios = json.loads(settings.scenarios_file.read_text(encoding="utf-8"))
+        seen: set[str] = set()
+        result: list[dict[str, str]] = []
+        for s in scenarios:
+            code = s.get("fault_code", "").strip()
+            if code and code not in seen:
+                seen.add(code)
+                result.append({
+                    "fault_code": code,
+                    "machine": s.get("machine", ""),
+                    "description": s.get("expected_topic", ""),
+                })
+        _FAULT_CODES_CACHE = result
+        return result
+    except (OSError, json.JSONDecodeError):
+        return []
+
+
 @router.post("/machines", status_code=201)
 def create_machine(machine: MachineIn) -> dict[str, Any]:
     machine_id = machine.id.strip()
@@ -71,7 +99,7 @@ def report_fault(f: FaultIn) -> WorkOrder:
     arrival = datetime.now(timezone.utc).isoformat()
     return handle_fault(
         f.fault_code, f.machine, f.context,
-        translate=f.translate, fault_arrived_at=arrival,
+        translate=f.translate, fault_arrived_at=arrival, source=f.source,
     )
 
 
@@ -280,13 +308,14 @@ async def _check_twilio() -> bool:
         return False
 
 
-async def _check_mqtt() -> bool:
-    if live.mqtt_client is None:
-        return False
+async def _check_mqtt() -> dict[str, bool]:
+    configured = bool(settings.mqtt_broker_url)
+    if not configured:
+        return {"configured": False, "connected": False}
     try:
-        return live.mqtt_client.is_connected()
+        return {"configured": True, "connected": live.mqtt_client is not None and live.mqtt_client.is_connected()}
     except Exception:
-        return False
+        return {"configured": True, "connected": False}
 
 
 @router.get("/health")
@@ -302,7 +331,7 @@ async def health_check() -> dict[str, Any]:
     qdrant_ok = await qdrant_task
     llm_ok = await llm_task
     twilio_ok = await twilio_task
-    mqtt_ok = await mqtt_task
+    mqtt_status = await mqtt_task
 
     overall_healthy = db_ok
 
@@ -324,9 +353,9 @@ async def health_check() -> dict[str, Any]:
                 "connected": twilio_ok,
             },
             "mqtt_broker": {
-                "connected": mqtt_ok,
-                "broker": settings.mqtt_broker_url,
-                "fallbacks_active": settings.enable_fallbacks,
+                "configured": mqtt_status["configured"],
+                "connected": mqtt_status["connected"],
+                "broker": settings.mqtt_broker_url if settings.mqtt_broker_url else "(not configured)",
             },
         },
     }

src/torq/config.py

@@ -46,11 +46,10 @@ def _ensure_llm_key(self):
     twilio_auth_token: str = ""
     twilio_whatsapp_from: str = ""
 
-    # MQTT (machine fault events). Public broker by default; $0, no auth.
-    # Public unauthenticated broker + fixed topic means anyone can publish fake
-    # faults (each spawns a diagnosis run) or read plant data. For production,
-    # point at an authenticated broker over TLS and set credentials.
-    mqtt_broker_url: str = "broker.hivemq.com"
+    # MQTT (machine fault events). Empty by default — purely opt-in.
+    # Set MQTT_BROKER_URL to enable the listener for plants with digital
+    # machine connectivity. For authenticated brokers, also set credentials.
+    mqtt_broker_url: str = ""
     mqtt_port: int = 1883
     mqtt_topic: str = "torq/demo/faults"
 

src/torq/db/models.py

@@ -1,7 +1,7 @@
 """Persistence for work orders and the machine registry."""
 
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any
 
 from torq.agent.schemas import WorkOrder
@@ -122,6 +122,25 @@ def list_all() -> list[WorkOrder]:
     return [WorkOrder.model_validate_json(r["data"]) for r in rows]
 
 
+def find_recent(machine: str, fault_code: str, window_sec: int = 30) -> WorkOrder | None:
+    """Return the most recent work order for (machine, fault_code) within window_sec, if any."""
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT data FROM work_orders WHERE machine = ? AND fault_code = ? ORDER BY rowid DESC LIMIT 1",
+            (machine, fault_code),
+        ).fetchall()
+    if not rows:
+        return None
+    wo = WorkOrder.model_validate_json(rows[0]["data"])
+    if not wo.created_at:
+        return None
+    try:
+        age = (datetime.now(timezone.utc) - datetime.fromisoformat(wo.created_at)).total_seconds()
+    except ValueError:
+        return None
+    return wo if age <= window_sec else None
+
+
 def _secs(start: str | None, end: str | None) -> float | None:
     if not start or not end:
         return None

src/torq/events/listener.py

@@ -39,6 +39,7 @@ def handle_payload(payload: bytes) -> WorkOrder | None:
         machine=event.machine_id,
         context=event.context,
         fault_arrived_at=arrival,
+        source="mqtt",
     )
     sev = MachineFaultEvent.SEVERITY_LABELS.get(event.severity, "unknown")
     print(