Merge pull request #97 from zestones/29-m47-memory-flex-scene

29 m47 memory flex scene

Author: zestones

backend/agents/investigator.py

@@ -556,7 +556,7 @@ def _spawn_work_order_generator(work_order_id: int) -> None:
     instead of a throwaway stub.
     """
     try:
-        from agents.work_order_generator import ( # type: ignore[import-not-found]
+        from agents.work_order_generator import (  # type: ignore[import-not-found]
             run_work_order_generator,
         )
     except ImportError:

backend/core/config.py

@@ -35,6 +35,10 @@ class Settings(BaseSettings):
     # `model_for()` currently ignores this and routes per use case.
     aria_model: str = "sonnet"  # "sonnet" | "opus"
 
+    # Mounts `modules.demo.router` when true — off by default so production
+    # deployments do not expose `/api/v1/demo/trigger-memory-scene`. See #29.
+    aria_demo_enabled: bool = False
+
     @property
     def database_dsn(self) -> str:
         return (

backend/main.py

@@ -91,6 +91,14 @@ async def health() -> dict[str, str]:
     app.include_router(kb_router)
     app.include_router(events_router)
 
+    if settings.aria_demo_enabled:
+        # Demo-only routes (#29 memory-flex scene). Mounted behind a flag
+        # so production deployments do not expose the scene triggers.
+        from modules.demo.router import router as demo_router
+
+        app.include_router(demo_router)
+        log.info("Demo endpoints enabled at /api/v1/demo/*")
+
     app.mount("/mcp", mcp_http_app)
 
     return app

backend/modules/demo/__init__.py

@@ -0,0 +1,169 @@
+"""Demo-only routes — mounted in ``main.py`` only when ``ARIA_DEMO_ENABLED=true``.
+
+Issue #29 (M4.7 — Memory flex scene). Provides an idempotent, re-playable
+endpoint that:
+
+1. Clears any recent ``failure_history`` for the target cell (so the scene
+   can be re-triggered during demo rehearsal without accumulating rows).
+2. Seeds one past failure dated 3 months ago with a ``signal_patterns``
+   JSON matching the current anomaly the operator is about to inject.
+3. Cancels any still-open agent-generated work order on the cell (so
+   Sentinel's per-WO debounce window does not swallow the fresh anomaly).
+4. Inserts a short burst of high-vibration readings into
+   ``process_signal_data`` so Sentinel's next 30-second tick opens a
+   ``work_order(status='detected')`` and spawns the Investigator.
+
+The Investigator then loads ``get_failure_history(cell_id, limit=5)`` on
+startup (#25) and — if the current signal pattern matches the seeded past
+failure — cites it in ``submit_rca.similar_past_failure``. That chain is
+the "knowledge doesn't retire" demo scene.
+
+Intentionally throwaway: this router is kept separate from production
+routers, guarded by an env flag, and carries no Pydantic schemas outside
+the inline response dict. Do not import from it in production code.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+import asyncpg
+from core.database import get_db
+from core.security import get_current_user
+from fastapi import APIRouter, Body, Depends, HTTPException
+
+log = logging.getLogger("aria.demo")
+
+router = APIRouter(
+    prefix="/api/v1/demo",
+    tags=["demo"],
+    dependencies=[Depends(get_current_user)],
+)
+
+
+_PAST_FAILURE_PATTERN: dict[str, Any] = {"vibration_mm_s": {"peak": 5.4, "duration_min": 14}}
+# Short burst of readings — 5 x 30s apart, each above the 4.5 mm/s alert
+# threshold seeded in migration 007's P-02 KB. Sentinel's 5-minute window
+# captures all of them.
+_FRESH_READINGS = [5.05, 5.12, 5.18, 5.22, 5.15]
+
+
+@router.post("/trigger-memory-scene")
+async def trigger_memory_scene(
+    cell_name: str = Body("P-02", embed=True),
+    conn: asyncpg.Connection = Depends(get_db),
+) -> dict[str, Any]:
+    """Prime the "memory flex" demo scene — see module docstring.
+
+    Safe to call repeatedly: step 1 clears prior scene state and step 3
+    cancels any open agent WO so Sentinel's debounce is reset.
+    """
+    cell_row = await conn.fetchrow("SELECT id FROM cell WHERE name = $1", cell_name)
+    if cell_row is None:
+        raise HTTPException(status_code=404, detail=f"cell {cell_name!r} not found")
+    cell_id: int = cell_row["id"]
+
+    # Find the signal def that maps to vibration_mm_s — must exist for the
+    # Sentinel → Investigator chain to pick up the injected readings.
+    sig_row = await conn.fetchrow(
+        """
+        SELECT id FROM process_signal_definition
+        WHERE cell_id = $1 AND kb_threshold_key = 'vibration_mm_s'
+        LIMIT 1
+        """,
+        cell_id,
+    )
+    if sig_row is None:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"cell {cell_name!r} has no process_signal_definition with "
+                "kb_threshold_key='vibration_mm_s'"
+            ),
+        )
+    signal_def_id: int = sig_row["id"]
+
+    async with conn.transaction():
+        # 1. Clear recent failure_history so the seeded past-failure is the
+        #    only one the Investigator sees from the last week.
+        await conn.execute(
+            "DELETE FROM failure_history WHERE cell_id = $1 "
+            "AND failure_time > NOW() - INTERVAL '7 days'",
+            cell_id,
+        )
+
+        # 2. Insert the past-failure row at t-3 months. `encode_fields` is
+        #    overkill here — raw jsonb literal is fine in a demo route.
+        past_row = await conn.fetchrow(
+            """
+            INSERT INTO failure_history (
+                cell_id, failure_time, resolved_time,
+                failure_mode, root_cause, signal_patterns
+            ) VALUES (
+                $1,
+                NOW() - INTERVAL '3 months',
+                NOW() - INTERVAL '3 months' + INTERVAL '4 hours',
+                'bearing_wear',
+                'Discharge bearing wear near end-of-life — replaced under PM-2026-01-18.',
+                $2::jsonb
+            )
+            RETURNING id
+            """,
+            cell_id,
+            json.dumps(_PAST_FAILURE_PATTERN),
+        )
+        past_failure_id: int = past_row["id"] if past_row is not None else 0
+
+        # 3. Cancel any still-open agent-generated WO on this cell so the
+        #    Sentinel per-(cell, signal) debounce window does not swallow
+        #    the fresh anomaly. `detected` and `analyzed` are the statuses
+        #    that block detection per M4.2.
+        cancelled = await conn.execute(
+            """
+            UPDATE work_order
+               SET status = 'cancelled', completed_at = NOW()
+             WHERE cell_id = $1
+               AND generated_by_agent = TRUE
+               AND status IN ('detected', 'analyzed', 'open', 'in_progress')
+            """,
+            cell_id,
+        )
+
+        # 4. Seed a burst of high-vibration readings backdated 1..5 minutes.
+        #    Sentinel's 5-minute look-back will pick them up on the next tick.
+        for i, value in enumerate(_FRESH_READINGS):
+            await conn.execute(
+                """
+                INSERT INTO process_signal_data (time, cell_id, signal_def_id, raw_value)
+                VALUES (NOW() - (INTERVAL '30 seconds' * $1), $2, $3, $4)
+                ON CONFLICT (time, signal_def_id) DO NOTHING
+                """,
+                len(_FRESH_READINGS) - i,
+                cell_id,
+                signal_def_id,
+                value,
+            )
+
+    log.info(
+        "memory-scene triggered cell=%s cell_id=%d past_failure_id=%d "
+        "signal_def_id=%d readings=%d cancelled_wos=%s",
+        cell_name,
+        cell_id,
+        past_failure_id,
+        signal_def_id,
+        len(_FRESH_READINGS),
+        cancelled,
+    )
+
+    return {
+        "ok": True,
+        "cell_id": cell_id,
+        "cell_name": cell_name,
+        "past_failure_id": past_failure_id,
+        "signal_def_id": signal_def_id,
+        "readings_inserted": len(_FRESH_READINGS),
+        # Sentinel tick is 30s; worst-case latency = one full tick + DB roundtrip.
+        "expect_anomaly_within_seconds": 35,
+    }

backend/modules/demo/router.py

@@ -0,0 +1,166 @@
+"""Tests for ``modules.demo.router`` (issue #29 / M4.7 memory flex scene).
+
+Covers the handler's SQL orchestration without touching a live database:
+
+- Returns 404 when the cell does not exist.
+- Returns 400 when the cell has no vibration signal_def mapped to
+  ``kb_threshold_key='vibration_mm_s'``.
+- Happy path inside a transaction: DELETE recent failure_history, INSERT
+  one past failure, CANCEL open agent WOs, INSERT the burst of fresh
+  readings, returns the expected envelope.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from modules.demo import router as demo_router
+
+
+# ---------------------------------------------------------------------------
+# Fake asyncpg.Connection surface — only the methods the handler uses.
+# ---------------------------------------------------------------------------
+
+
+class _FakeTxn:
+    async def __aenter__(self) -> "_FakeTxn":
+        return self
+
+    async def __aexit__(self, *a: Any) -> None:
+        return None
+
+
+class _FakeConn:
+    def __init__(
+        self,
+        *,
+        cell_row: dict[str, Any] | None,
+        sig_row: dict[str, Any] | None,
+        past_id: int = 77,
+    ) -> None:
+        self._cell_row = cell_row
+        self._sig_row = sig_row
+        self._past_id = past_id
+        self.fetchrow_calls: list[tuple[str, tuple[Any, ...]]] = []
+        self.execute_calls: list[tuple[str, tuple[Any, ...]]] = []
+
+    def transaction(self) -> _FakeTxn:
+        return _FakeTxn()
+
+    async def fetchrow(self, query: str, *args: Any) -> dict[str, Any] | None:
+        self.fetchrow_calls.append((query, args))
+        q = " ".join(query.split())
+        if "FROM cell WHERE name" in q:
+            return self._cell_row
+        if "FROM process_signal_definition" in q:
+            return self._sig_row
+        if "INSERT INTO failure_history" in q:
+            return {"id": self._past_id}
+        return None
+
+    async def execute(self, query: str, *args: Any) -> str:
+        self.execute_calls.append((query, args))
+        q = " ".join(query.split())
+        if "DELETE FROM failure_history" in q:
+            return "DELETE 0"
+        if "UPDATE work_order" in q:
+            return "UPDATE 1"
+        if "INSERT INTO process_signal_data" in q:
+            return "INSERT 0 1"
+        return "OK"
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_returns_404_when_cell_missing() -> None:
+    conn = _FakeConn(cell_row=None, sig_row=None)
+    with pytest.raises(Exception) as excinfo:
+        await demo_router.trigger_memory_scene(
+            cell_name="UNKNOWN-99", conn=conn  # type: ignore[arg-type]
+        )
+    # FastAPI HTTPException — inspect via attribute.
+    assert getattr(excinfo.value, "status_code", None) == 404
+
+
+@pytest.mark.asyncio
+async def test_returns_400_when_signal_def_missing() -> None:
+    conn = _FakeConn(cell_row={"id": 2}, sig_row=None)
+    with pytest.raises(Exception) as excinfo:
+        await demo_router.trigger_memory_scene(
+            cell_name="P-02", conn=conn  # type: ignore[arg-type]
+        )
+    assert getattr(excinfo.value, "status_code", None) == 400
+    # Nothing written when signal lookup fails (we return before the txn).
+    assert conn.execute_calls == []
+
+
+@pytest.mark.asyncio
+async def test_happy_path_orchestrates_all_four_steps() -> None:
+    conn = _FakeConn(cell_row={"id": 2}, sig_row={"id": 55}, past_id=99)
+
+    resp = await demo_router.trigger_memory_scene(
+        cell_name="P-02", conn=conn  # type: ignore[arg-type]
+    )
+
+    assert resp["ok"] is True
+    assert resp["cell_id"] == 2
+    assert resp["cell_name"] == "P-02"
+    assert resp["past_failure_id"] == 99
+    assert resp["signal_def_id"] == 55
+    # 5 readings seeded by the handler.
+    assert resp["readings_inserted"] == 5
+    assert resp["expect_anomaly_within_seconds"] == 35
+
+    # Step order inside the txn: DELETE, (fetchrow INSERT failure), UPDATE,
+    # then 5 x INSERT process_signal_data.
+    exec_queries = [" ".join(q.split()) for q, _ in conn.execute_calls]
+    assert any(q.startswith("DELETE FROM failure_history") for q in exec_queries)
+    assert any(q.startswith("UPDATE work_order") for q in exec_queries)
+    readings_inserts = [q for q in exec_queries if "INSERT INTO process_signal_data" in q]
+    assert len(readings_inserts) == 5
+
+
+@pytest.mark.asyncio
+async def test_happy_path_past_failure_row_carries_signal_patterns_jsonb() -> None:
+    conn = _FakeConn(cell_row={"id": 2}, sig_row={"id": 55})
+
+    await demo_router.trigger_memory_scene(
+        cell_name="P-02", conn=conn  # type: ignore[arg-type]
+    )
+
+    insert_calls = [
+        (q, args)
+        for q, args in conn.fetchrow_calls
+        if "INSERT INTO failure_history" in " ".join(q.split())
+    ]
+    assert len(insert_calls) == 1
+    _, args = insert_calls[0]
+    # args[0] = cell_id, args[1] = jsonb string
+    assert args[0] == 2
+    import json as _json
+
+    patterns = _json.loads(args[1])
+    assert "vibration_mm_s" in patterns
+    assert patterns["vibration_mm_s"]["peak"] == pytest.approx(5.4)
+
+
+@pytest.mark.asyncio
+async def test_readings_all_above_alert_threshold() -> None:
+    conn = _FakeConn(cell_row={"id": 2}, sig_row={"id": 55})
+    await demo_router.trigger_memory_scene(
+        cell_name="P-02", conn=conn  # type: ignore[arg-type]
+    )
+    # All seeded readings must exceed the P-02 vibration alert (4.5 mm/s)
+    # otherwise Sentinel will not open a work_order and the scene dies.
+    reading_values = [
+        args[-1]
+        for q, args in conn.execute_calls
+        if "INSERT INTO process_signal_data" in " ".join(q.split())
+    ]
+    assert reading_values  # at least one
+    assert all(v > 4.5 for v in reading_values)