Add event-triggered monitoring and durable private history

.env.example

@@ -15,3 +15,14 @@ ALPHA_VANTAGE_API_KEY=
 PORT=7860
 
 API_PORT=8000
+
+# Market scans are scheduled; paid research is event-triggered.
+MONITOR_INTERVAL_MINUTES=60
+# Optional: private dataset repository for Space history backups. Keep the token
+# in local .env or Hugging Face Settings > Secrets, never in source control.
+MONITOR_BACKUP_REPO=
+MONITOR_BACKUP_TOKEN=
+PRICE_MOVE_TRIGGER_PCT=1.0
+VOLATILITY_TRIGGER_MULTIPLIER=1.5
+SENTIMENT_SHIFT_THRESHOLD=0.25
+RESEARCH_COOLDOWN_MINUTES=30

app/agents/market_research_agent.py

@@ -11,6 +11,7 @@ import math
 import re
 import threading
 import time
+import uuid
 import warnings
 from urllib.parse import parse_qsl, unquote, urlencode, urlsplit, urlunsplit
 
@@ -18,6 +19,7 @@ from dateutil import parser as date_parser
 
 from app.schemas.agent_models import EvidenceItem, ResearchResult
 from app.services.tavily_service import TavilyService, prediction_cutoff, unavailable
+from app.services.agent_config import redact_values
 
 PRIMARY = {
     "federalreserve.gov": "Federal Reserve", "sec.gov": "SEC", "treasury.gov": "US Treasury",
@@ -161,15 +163,30 @@ def evidence_category(headline, snippet):
 
 
 class MarketResearchAgent:
-    def __init__(self, settings, tavily=None):
+    def __init__(self, settings, tavily=None, store=None):
         self.settings = settings
         self.tavily = tavily if tavily is not None else TavilyService(settings)
         self._cache = OrderedDict()
         self._inflight = {}
         self._guard = threading.Lock()
+        self.store = store
 
     @staticmethod
-    def query_plan(question, mode="live"):
+    def query_plan(question, mode="live", trigger_types=None):
+        if trigger_types:
+            topics = []
+            rules = {
+                "large_price_move": [TOPICS[2][1], TOPICS[0][1]],
+                "unusual_volatility": [TOPICS[2][1], TOPICS[1][1]],
+                "regime_change": [TOPICS[1][1], TOPICS[2][1]],
+                "sentiment_shift": [TOPICS[0][1], TOPICS[3][1]],
+                "forecast_state_change": [TOPICS[0][1], TOPICS[1][1]],
+            }
+            for reason in trigger_types:
+                for query in rules.get(reason, []):
+                    if query not in topics:
+                        topics.append(query)
+            return ["Bitcoin BTC latest market news price drivers"] + topics[:2]
         text = str(question).lower()[:2000]
         chosen = [query for pattern, query in TOPICS if re.search(pattern, text)][:2]
         for _, query in TOPICS[:2]:
@@ -180,11 +197,19 @@ class MarketResearchAgent:
         prefix = "Bitcoin BTC latest market news price drivers" if mode == "live" else "Bitcoin BTC market news price drivers"
         return [prefix] + chosen
 
-    def _search(self, query, mode, cutoff):
+    def _search(self, query, mode, cutoff, fresh=False):
         key = (query, mode, cutoff.isoformat() if cutoff else None)
+        disk_key = hashlib.sha256(repr((key, self.settings.news_max_age_hours)).encode()).hexdigest()
+        if self.store and not fresh:
+            try:
+                disk = self.store.cache_get(disk_key)
+            except Exception:
+                return {**unavailable("The shared research cache is unavailable."), "retrieved_at": _now()}, False
+            if disk is not None:
+                return disk, True
         with self._guard:
             entry = self._cache.get(key)
-            if entry and entry[0] > time.monotonic():
+            if not fresh and entry and entry[0] > time.monotonic():
                 self._cache.move_to_end(key)
                 return deepcopy(entry[1]), True
             future = self._inflight.get(key)
@@ -197,6 +222,27 @@ class MarketResearchAgent:
                 return deepcopy(future.result(timeout=self.settings.tavily_timeout_seconds * 2 + 5)), True
             except FutureTimeout:
                 return {**unavailable("The shared research request is still in progress."), "retrieved_at": _now()}, False
+        lease_owner = uuid.uuid4().hex
+        if self.store:
+            acquired, disk = False, None
+            try:
+                acquired = self.store.acquire("query:" + disk_key, lease_owner, ttl=self.settings.tavily_timeout_seconds + 30)
+                # Another process may have filled the cache after our first read.
+                if not fresh or not acquired:
+                    disk = self.store.cache_get(disk_key)
+            except Exception:
+                acquired = False
+            if disk is not None or not acquired:
+                payload = disk or {**unavailable("The shared research request is in progress or unavailable."), "retrieved_at": _now()}
+                if acquired:
+                    try:
+                        self.store.release("query:" + disk_key, lease_owner)
+                    except Exception:
+                        pass
+                with self._guard:
+                    self._inflight.pop(key, None)
+                    future.set_result(deepcopy(payload))
+                return payload, True
         try:
             payload = self.tavily.search(query, mode=mode, prediction_timestamp=cutoff)
             if not isinstance(payload, dict) or payload.get("status") not in ("ok", "empty", "unavailable"):
@@ -207,6 +253,21 @@ class MarketResearchAgent:
         ttl = max(0., self.settings.news_cache_minutes * 60)
         if payload["status"] == "unavailable":
             ttl = min(ttl, 30.)
+        if self.store:
+            try:
+                # Retain bounded snippets only; do not persist whole article bodies.
+                disk_payload = deepcopy(payload)
+                disk_payload["results"] = [{k: (v[:1000] if isinstance(v, str) else v)
+                    for k, v in item.items() if k in {"title", "url", "content", "score", "published_at", "published_date"}}
+                    for item in payload.get("results", [])[:5] if isinstance(item, dict)]
+                self.store.cache_put(disk_key, redact_values(disk_payload), ttl)
+            except Exception:
+                pass  # A cache write failure must not strand in-flight waiters.
+            finally:
+                try:
+                    self.store.release("query:" + disk_key, lease_owner)
+                except Exception:
+                    pass  # The lease expires; waiting callers still receive a result.
         with self._guard:
             self._cache[key] = (time.monotonic() + ttl, deepcopy(payload))
             self._cache.move_to_end(key)
@@ -248,17 +309,17 @@ class MarketResearchAgent:
                             direction=headline_direction(title), category=evidence_category(title, content),
                             source_quality=quality, supports_model_direction=None, recency_verified=False)
 
-    def run(self, question, mode="live", prediction_timestamp=None):
+    def run(self, question, mode="live", prediction_timestamp=None, fresh=False, trigger_types=None):
         if mode not in ("live", "historical"):
             return ResearchResult(status="unavailable", reason="Unsupported research mode.")
         cutoff = prediction_cutoff(prediction_timestamp) if mode == "historical" else None
         if mode == "historical" and cutoff is None:
             return ResearchResult(status="unavailable", reason="Historical research requires a timezone-aware prediction timestamp.")
-        queries = self.query_plan(question, mode)
+        queries = self.query_plan(question, mode, trigger_types)
         payloads, cached_flags, evidence = [], [], []
         rejected = 0
         for query in queries:
-            payload, cached = self._search(query, mode, cutoff)
+            payload, cached = self._search(query, mode, cutoff, fresh=fresh)
             payloads.append(payload)
             cached_flags.append(cached)
             results = payload.get("results", [])

app/agents/orchestrator.py

@@ -20,6 +20,8 @@ def needs_research(question):
     if (re.search(r"(?:what|which).*(?:model|feature|input)|(?:does|is|are).*(?:model|feature|input).*(?:use|include|support)", text)
             and not re.search(r"(?:latest|current).*(?:news|headline)|search|research", text)):
         return False
+    if re.search(r"why.*(?:model|predict|forecast).*(?:up|down)|what happened.*market|explain.*(?:latest|current).*mov|current events|external.*evidence|fresh.*research", text):
+        return True
     if re.search(r"news|headlines|moving|market.moving|etf|fed\b|macro|inflation|cpi\b|employment|jobs report|regulat|liquidat|funding|open interest|whale|institution|bullish.*bearish|bearish.*bullish|factors|geopolit|hack", text):
         return True
     if (re.search(r"\b(?:btc|bitcoin)\b", text) and re.search(r"\b(?:why|today|now|current)\b", text)
@@ -32,7 +34,7 @@ class AgentOrchestrator:
         self.monitoring = monitoring
         self.settings = settings or AgentSettings.from_env()
         self.forecast_agent = ForecastContextAgent(self.settings)
-        self.research_agent = research_agent or MarketResearchAgent(self.settings)
+        self.research_agent = research_agent or MarketResearchAgent(self.settings, store=getattr(monitoring, "event_store", None))
         self.validation_agent = ValidationAgent(self.settings)
         self.decision_agent = DecisionAgent(self.settings, self.validation_agent, llm=llm)
         self._lock = threading.RLock()
@@ -68,6 +70,13 @@ class AgentOrchestrator:
     def evidence(self):
         with self._lock:
             latest = self._evidence.model_copy(deep=True)
+        store = getattr(self.monitoring, "event_store", None)
+        if store:
+            saved = store.latest_event(with_evidence=True)
+            candidates = [saved["result"] if saved else None, self.monitoring.store.get_state("analyst_evidence")]
+            for candidate in candidates:
+                if candidate and (candidate.get("retrieved_at") or "") > (latest.retrieved_at or ""):
+                    latest = ResearchResult.model_validate(candidate)
         # Revalidate cached evidence at read time; a source can age out while the
         # dashboard stays open. No new provider calls are made by GET requests.
         snapshot = copy.deepcopy(self.monitoring.snapshot())
@@ -92,12 +101,15 @@ class AgentOrchestrator:
         snapshot = copy.deepcopy(self.monitoring.snapshot())
         context = self.forecast_agent.run(snapshot)
         self._set(request_id, forecast="done", research="running")
-        research = self.research_agent.run(redact(request.question), mode=request.mode, prediction_timestamp=request.prediction_timestamp)
+        options = {"fresh": True} if request.fresh_research else {}
+        research = self.research_agent.run(redact(request.question), mode=request.mode, prediction_timestamp=request.prediction_timestamp, **options)
         self._set(request_id, research="unavailable" if research.status == "unavailable" else "done", validation="running")
         _, research, validation = self.validation_agent.validate(context, snapshot, research, mode=request.mode, prediction_timestamp=request.prediction_timestamp)
         if request.mode == "live":
             with self._lock:
                 self._evidence = research.model_copy(deep=True)
+            if getattr(self.monitoring, "event_store", None):
+                self.monitoring.store.set_state("analyst_evidence", research.model_dump())
         self._set(request_id, validation="done")
         return {"research": research.model_dump(), "validation": validation.model_dump(), "request_id": request_id,
                 "agent_status": self.activity(request_id)["agent_status"]}
@@ -113,9 +125,10 @@ class AgentOrchestrator:
             snapshot = copy.deepcopy(self.monitoring.snapshot())
             context = self.forecast_agent.run(snapshot)
             self._set(request_id, forecast="done")
-            if needs_research(question):
+            if needs_research(question) or request.fresh_research:
                 self._set(request_id, research="running")
-                research = self.research_agent.run(question)
+                fresh = request.fresh_research or bool(re.search(r"\b(?:fresh|new)\s+(?:web\s+)?(?:search|research)|search again|ignore.*cache", question, re.I))
+                research = self.research_agent.run(question, **({"fresh": True} if fresh else {}))
                 self._set(request_id, research="unavailable" if research.status == "unavailable" else "done")
             else:
                 research = ResearchResult()
@@ -125,6 +138,8 @@ class AgentOrchestrator:
             if research.status != "not_requested":
                 with self._lock:
                     self._evidence = research.model_copy(deep=True)
+                if getattr(self.monitoring, "event_store", None):
+                    self.monitoring.store.set_state("analyst_evidence", research.model_dump())
             self._set(request_id, validation="done", decision="running")
             answer, llm_status, sources = self.decision_agent.run(question, history, context, research, validation)
             self._set(request_id, decision="done" if llm_status == "ok" else "unavailable")

app/agents/sentiment_agent.py

@@ -30,6 +30,8 @@ class SentimentAgent:
             cmd.append("--skip-gdelt")
 
         env = os.environ.copy()
+        for key in ("MONITOR_BACKUP_TOKEN", "HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN"):
+            env.pop(key, None)
         if self.alpha_vantage_key:
             env["ALPHA_VANTAGE_API_KEY"] = self.alpha_vantage_key
 

app/api.py

@@ -6,7 +6,7 @@ from typing import Literal
 from fastapi import BackgroundTasks, FastAPI, HTTPException, Query
 from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from .monitoring_service import MonitoringService
 
 PROJECT = Path(os.getenv("BTC_PROJECT_ROOT", str(Path(__file__).resolve().parents[1])))
@@ -47,9 +47,11 @@ class RefreshRequest(BaseModel):
 
 class MonitoringRunRequest(RefreshRequest):
     trigger: Literal["manual", "hourly"] = "manual"
+    worker_id: str | None = Field(None, pattern=r"^[a-f0-9]{32}$")
 
 class HeartbeatRequest(BaseModel):
     next_run_at: str | None = None
+    worker_id: str | None = Field(None, pattern=r"^[a-f0-9]{32}$")
 
 @app.get("/monitor", include_in_schema=False)
 @app.get("/monitor/", include_in_schema=False)
@@ -132,10 +134,24 @@ def monitoring_run(req: MonitoringRunRequest, tasks: BackgroundTasks, background
     if background:
         if service.system_status()["active_run"]:
             return JSONResponse({"status": "busy", "detail": "A refresh is already running."}, status_code=202)
-        tasks.add_task(service.run, req.trigger, req.use_gdelt)
+        tasks.add_task(service.run, req.trigger, req.use_gdelt, req.worker_id)
         return JSONResponse({"status": "queued"}, status_code=202)
-    return service.run(req.trigger, req.use_gdelt)
+    return service.run(req.trigger, req.use_gdelt, req.worker_id)
 
 @app.post("/monitoring/heartbeat", include_in_schema=False)
 def monitoring_heartbeat(req: HeartbeatRequest):
-    return get_service().heartbeat(req.next_run_at)
+    return get_service().heartbeat(req.next_run_at, req.worker_id)
+
+@app.post("/monitoring/worker/release", include_in_schema=False)
+def release_worker(req: HeartbeatRequest):
+    if req.worker_id:
+        get_service().event_store.release("scheduler", req.worker_id)
+    return {"released": True}
+
+@app.get("/monitoring/activity")
+def monitoring_activity():
+    return get_service().monitoring_activity()
+
+@app.get("/monitoring/scans")
+def monitoring_scans(limit: int = Query(50, ge=1, le=200)):
+    return {"items": get_service().event_store.scans(limit)}

app/monitor_worker.py

@@ -1,51 +1,5 @@
-"""Hourly scheduler; FastAPI is the single inference and database owner."""
-from __future__ import annotations
-import os
-import signal
-import threading
-from datetime import datetime, timedelta, timezone
-import requests
-
-UTC = timezone.utc
-
-def next_hourly_run(now):
-    slot = now.replace(minute=2, second=0, microsecond=0)
-    return slot if now < slot else slot + timedelta(hours=1)
-
-def main():
-    api = os.getenv("BTC_API_URL", "http://btc-api:8000").rstrip("/")
-    stop = threading.Event()
-    signal.signal(signal.SIGTERM, lambda *_: stop.set())
-    signal.signal(signal.SIGINT, lambda *_: stop.set())
-    next_run = [datetime.now(UTC)]
-
-    def heartbeat():
-        while not stop.is_set():
-            try:
-                requests.post(f"{api}/monitoring/heartbeat", json={"next_run_at": next_run[0].isoformat()}, timeout=10).raise_for_status()
-            except requests.RequestException:
-                pass
-            stop.wait(25)
-
-    threading.Thread(target=heartbeat, daemon=True).start()
-    while not stop.is_set():
-        now = datetime.now(UTC)
-        if now < next_run[0]:
-            stop.wait(min(30, (next_run[0] - now).total_seconds()))
-            continue
-        try:
-            response = requests.post(f"{api}/monitoring/run", json={"use_gdelt": os.getenv("MONITOR_USE_GDELT", "true").lower() == "true", "trigger": "hourly"}, timeout=1500)
-            response.raise_for_status()
-            status = response.json()["status"]
-            print(f"Monitoring cycle: {status}", flush=True)
-            if status in {"error", "busy"}:
-                next_run[0] = datetime.now(UTC) + timedelta(minutes=5)
-            else:
-                next_run[0] = next_hourly_run(datetime.now(UTC))
-        except (requests.RequestException, ValueError, KeyError) as exc:
-            print(f"Monitoring API unavailable ({type(exc).__name__}); retry in 60s.", flush=True)
-            next_run[0] = datetime.now(UTC) + timedelta(seconds=60)
-    print("Monitoring scheduler stopped.", flush=True)
+"""Stable entry point for the browser-independent monitoring worker."""
+from .services.monitor_worker_runtime import MonitoringWorker, main, next_hourly_run
 
 if __name__ == "__main__":
     main()

app/monitoring_service.py

@@ -15,6 +15,10 @@ import pandas as pd
 import yaml
 
 from .monitoring_store import MonitoringStore
+from .services.agent_config import AgentSettings
+from .services.monitoring_events_store import EventStore
+from .services.monitor_schedule import next_cycle
+from .services.monitoring_backup import MonitoringBackup
 
 UTC = timezone.utc
 
@@ -24,19 +28,23 @@ def utcnow():
 
 
 def safe_error(exc):
-    message = str(exc)
-    for name in ("ALPHA_VANTAGE_API_KEY", "OPENAI_API_KEY"):
-        secret = os.getenv(name)
-        if secret:
-            message = message.replace(secret, "[redacted]")
-    return re.sub(r"(?i)(apikey|api_key|token)=([^&\s]+)", r"\1=[redacted]", message)[-600:]
+    from .services.agent_config import redact
+    return redact(str(exc))[-600:]
 
 
 class MonitoringService:
     def __init__(self, project_root, db_path=None):
         self.root = Path(project_root)
-        self.store = MonitoringStore(db_path or os.getenv("BTC_MONITOR_DB", str(self.root / "runtime/monitoring.sqlite3")))
+        path = db_path or os.getenv("BTC_MONITOR_DB", str(self.root / "runtime/monitoring.sqlite3"))
+        self.backup = MonitoringBackup()
+        restored = self.backup.restore(path)
+        self.store = MonitoringStore(path)
+        if restored:
+            self.store.set_state("backup", {"status": "RESTORED", "restored_at": utcnow().isoformat()})
         self.work_dir = Path(self.store.path).parent
+        self.settings = AgentSettings.from_env()
+        self.event_store = EventStore(self.store)
+        self._market_monitor = None
         self._pipeline = None
         self._pipeline_lock = threading.Lock()
         self._run_lock = threading.Lock()
@@ -70,10 +78,54 @@ class MonitoringService:
     def snapshot(self):
         return self.store.latest_snapshot() or {}
 
-    def heartbeat(self, next_run_at=None):
+    def heartbeat(self, next_run_at=None, worker_id=None):
+        if worker_id and not self.event_store.acquire("scheduler", worker_id, ttl=90):
+            return {"accepted": False, "next_cycle_due": self.store.get_state("next_cycle_due")}
+        due = self.store.get_state("next_cycle_due")
+        started = self.store.get_state("last_cycle_started")
+        completed = self.store.get_state("last_cycle_finished") or self.store.get_state("last_cycle_completed")
+        active = self.store.get_state("active_run") or {}
+        if started and (not completed or started > completed):
+            last_beat = active.get("heartbeat_at", active.get("started_at", started))
+            if (utcnow() - datetime.fromisoformat(last_beat)).total_seconds() > 120:
+                due = utcnow().isoformat()
+                self.store.set_state("next_cycle_due", due)
+        if due is None:
+            last = self.store.get_state("last_cycle_completed") or self.store.get_state("last_success")
+            due = next_cycle(datetime.fromisoformat(last), self.settings.monitor_interval_minutes).isoformat() if last else utcnow().isoformat()
+            self.store.set_state("next_cycle_due", due)
+        if worker_id:
+            next_run_at = due
         state = {"at": utcnow().isoformat(), "next_run_at": next_run_at}
+        if worker_id:
+            state["worker_id"] = worker_id
         self.store.set_state("worker", state)
-        return state
+        return {**state, "accepted": True, "next_cycle_due": due, "monitor_interval_minutes": self.settings.monitor_interval_minutes,
+                "last_cycle_started": self.store.get_state("last_cycle_started"),
+                "last_cycle_completed": self.store.get_state("last_cycle_completed"),
+                "last_research_trigger": self.store.get_state("last_research_trigger"),
+                "last_error": self.store.get_state("last_error")}
+
+    def market_monitor(self):
+        if self._market_monitor is None:
+            from .services.market_monitoring import MarketMonitoringAgent
+            self._market_monitor = MarketMonitoringAgent(self)
+        return self._market_monitor
+
+    def monitoring_activity(self):
+        system = self.system_status()
+        event = self.event_store.latest_event()
+        return {"worker_alive": system["worker_alive"],
+                "last_cycle_started": self.store.get_state("last_cycle_started"),
+                "last_cycle_completed": self.store.get_state("last_cycle_completed"),
+                "last_market_scan": self.store.get_state("last_market_scan"),
+                "last_successful_prediction_cycle": self.store.get_state("last_success"),
+                "next_cycle_due": self.store.get_state("next_cycle_due"),
+                "last_research_trigger": self.store.get_state("last_research_trigger"),
+                "research": self.store.get_state("research_operation", {"status": "IDLE"}),
+                "last_research": event, "last_error": system["last_error"],
+                "backup": self.store.get_state("backup", {"status": "CONFIGURED" if self.backup.repo else "LOCAL_ONLY"}),
+                "monitor_interval_minutes": self.settings.monitor_interval_minutes}
 
     def system_status(self):
         worker = self.store.get_state("worker", {}) or {}
@@ -124,7 +176,7 @@ class MonitoringService:
             targets = [t for t in targets if pd.Timestamp(t) >= boundary]
         return result
 
-    def run(self, trigger="manual", use_gdelt=True):
+    def run(self, trigger="manual", use_gdelt=True, worker_id=None):
         owner = uuid.uuid4().hex
         if not self._run_lock.acquire(blocking=False):
             return {"status": "busy", "detail": "A refresh is already running."}
@@ -136,6 +188,12 @@ class MonitoringService:
         if not acquired:
             self._run_lock.release()
             return {"status": "busy", "detail": "A refresh is already running."}
+        if trigger == "hourly":
+            due = self.store.get_state("next_cycle_due")
+            if (worker_id and not self.event_store.owns("scheduler", worker_id)) or (due and utcnow() < datetime.fromisoformat(due)):
+                self.store.release_lock(owner)
+                self._run_lock.release()
+                return {"status": "not_due", "next_cycle_due": due}
         started = utcnow()
         clock_start = time.monotonic()
         renew_stop, lease_lost = threading.Event(), threading.Event()
@@ -158,6 +216,9 @@ class MonitoringService:
         renewal = threading.Thread(target=renew, daemon=True)
         try:
             run_id = self.store.start_run(trigger)
+            self.store.set_state("last_cycle_started", started.isoformat())
+            self.store.set_state("next_cycle_due", next_cycle(started, self.settings.monitor_interval_minutes).isoformat())
+            self.store.set_state("research_operation", {"status": "IDLE", "monitor": "running", "forecast": "running", "trigger": "idle", "validation": "idle", "decision": "idle"})
             old_active = self.store.get_state("active_run")
             if old_active and old_active.get("id"):
                 self.store.finish_run(old_active["id"], "failed", detail="Previous process stopped; retrying with saved history.")
@@ -189,7 +250,10 @@ class MonitoringService:
                 for h, forecast in data["forecasts"].items():
                     forecast["model_version"] = self.models[int(h)]["model_version"]
                     forecast["issued_at"] = issued_at.isoformat()
-                recorded = self.store.record_forecasts(data["forecasts"], issued_at=issued_at)
+                quantitative = self.market_monitor().prepare(data, market, now=issued_at)
+                data["market"]["regime"] = quantitative["regime"]
+                recorded = self.store.record_forecasts(data["forecasts"], issued_at=issued_at,
+                    context_by_horizon=quantitative["forecasts"], regime=quantitative["regime"])
                 # Recover the first-issued values after a crash between record
                 # insertion and snapshot saving, without rewriting predictions.
                 for h, forecast in data["forecasts"].items():
@@ -206,18 +270,28 @@ class MonitoringService:
                 self.store.save_snapshot(json.loads(json.dumps(data, default=str)))
                 self.store.set_state("last_success", issued_at.isoformat())
                 status = "success"
+            try:
+                research_trigger = self.market_monitor().process(run_id, self.snapshot(), market)
+            except Exception:
+                research_trigger = {"should_research": False, "triggered_by": [], "error": "Monitoring context processing failed; forecasts remain saved."}
+                self.store.set_state("research_operation", {"status": "ERROR", "error": research_trigger["error"]})
             self.store.set_state("last_error", None)
+            self.store.set_state("last_cycle_completed", utcnow().isoformat())
             duration = round((time.monotonic() - clock_start) * 1000)
             self.store.finish_run(run_id, "skipped" if status == "up_to_date" else status, duration_ms=duration)
-            return {"status": status, "recorded": recorded, "settled": settled, "duration_ms": duration}
+            return {"status": status, "recorded": recorded, "settled": settled, "duration_ms": duration, "research_trigger": research_trigger}
         except Exception as exc:
             detail = safe_error(exc)
             self.store.set_state("last_error", detail)
+            from datetime import timedelta
+            self.store.set_state("next_cycle_due", (utcnow() + timedelta(minutes=min(5, self.settings.monitor_interval_minutes))).isoformat())
+            self.store.set_state("research_operation", {"status": "IDLE", "monitor": "error", "forecast": "error", "trigger": "idle", "validation": "idle", "decision": "idle"})
             if run_id is not None:
                 self.store.finish_run(run_id, "error", detail=detail, duration_ms=round((time.monotonic() - clock_start) * 1000))
             return {"status": "error", "detail": detail}
         finally:
             try:
+                self.store.set_state("last_cycle_finished", utcnow().isoformat())
                 renew_stop.set()
                 if renewal.is_alive():
                     renewal.join(timeout=2)
@@ -227,3 +301,4 @@ class MonitoringService:
                 self.store.release_lock(owner)
             finally:
                 self._run_lock.release()
+            self.backup.save(self.store)

app/monitoring_store.py

@@ -93,6 +93,11 @@ class MonitoringStore:
                 CREATE TABLE IF NOT EXISTS worker_lock (
                     id INTEGER PRIMARY KEY CHECK(id = 1), owner TEXT NOT NULL, expires_at TEXT NOT NULL
                 );
+                CREATE TABLE IF NOT EXISTS forecast_annotations (
+                    forecast_id INTEGER PRIMARY KEY, reliability REAL,
+                    signal_state TEXT NOT NULL, regime TEXT,
+                    FOREIGN KEY(forecast_id) REFERENCES forecasts(id)
+                );
             """)
 
     @contextmanager
@@ -126,7 +131,7 @@ class MonitoringStore:
     def latest_snapshot(self):
         return self.get_state("latest_snapshot")
 
-    def record_forecasts(self, forecasts, issued_at=None):
+    def record_forecasts(self, forecasts, issued_at=None, context_by_horizon=None, regime=None):
         issued = _utc(issued_at)
         rows = []
         for key, forecast in forecasts.items():
@@ -163,6 +168,10 @@ class MonitoringStore:
                      model,feature_set,status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                      ON CONFLICT(asset,model_version,horizon,origin_open) DO NOTHING""", row)
                 if cursor.rowcount:
+                    context = (context_by_horizon or {}).get(str(row[2]))
+                    if context is not None:
+                        db.execute("INSERT INTO forecast_annotations VALUES(?,?,?,?)",
+                                   (cursor.lastrowid, context.get("reliability"), context.get("signal_state", "UNKNOWN"), regime))
                     result["inserted"] += 1
                     result[row[-1]] += 1
                 else:
@@ -311,7 +320,7 @@ class MonitoringStore:
                                     "AND issued_at<=? ORDER BY issued_at DESC,id DESC LIMIT 1",
                                     (horizon, _iso(current))).fetchone()
                 model_version = latest[0] if latest else None
-            rows = db.execute("SELECT * FROM forecasts WHERE horizon=? AND model_version=? "
+            rows = db.execute("SELECT f.*,a.reliability,a.signal_state,a.regime FROM forecasts f LEFT JOIN forecast_annotations a ON a.forecast_id=f.id WHERE horizon=? AND model_version=? "
                               "AND issued_at>=? AND issued_at<=? ORDER BY issued_at",
                               (horizon, model_version, _iso(cutoff - timedelta(days=days)),
                                _iso(current))).fetchall()
@@ -327,7 +336,29 @@ class MonitoringStore:
                            **self._metrics(subset, end)})
         return {"horizon": horizon, "days": days, "model_version": model_version,
                 "as_of": _iso(current), "window_start": _iso(cutoff),
-                **self._metrics(in_window, current), "series": series}
+                **self._metrics(in_window, current), "series": series,
+                "qualification": self._qualification(in_window, current)}
+
+    @classmethod
+    def _qualification(cls, rows, now):
+        # Legacy forecasts have NULL annotations; never backfill issue-time gates.
+        annotated = [r for r in rows if r["signal_state"] is not None and r["status"] != "excluded_late"]
+        qualified = [r for r in annotated if r["signal_state"] in {"UP", "DOWN"}]
+        def grouped(field):
+            groups = {}
+            for row in annotated:
+                value = row[field]
+                if field == "reliability":
+                    value = "unavailable" if value is None else "0.00-0.49" if value < .5 else "0.50-0.74" if value < .75 else "0.75-1.00"
+                groups.setdefault(value or "unavailable", []).append(row)
+            return {key: cls._metrics(items, now) for key, items in groups.items()}
+        n = len(annotated)
+        return {"annotated_forecasts": n, "legacy_unannotated": sum(r["signal_state"] is None for r in rows),
+                "signal_coverage": len(qualified) / n if n else None,
+                "no_signal_frequency": sum(r["signal_state"] == "NO_SIGNAL" for r in annotated) / n if n else None,
+                "unknown_frequency": sum(r["signal_state"] == "UNKNOWN" for r in annotated) / n if n else None,
+                "qualified_performance": cls._metrics(qualified, now),
+                "reliability_buckets": grouped("reliability"), "regimes": grouped("regime")}
 
     def latest_forecasts(self):
         snapshot = self.latest_snapshot()

app/schemas/agent_models.py

@@ -101,11 +101,13 @@ class SafeChatMessage(Contract):
     content: str = Field(max_length=6000)
 
 class ChatRequest(Contract):
+    fresh_research: bool = False
     message: str = Field(min_length=1, max_length=2000)
     history: list[SafeChatMessage] = Field(default_factory=list, max_length=12)
     request_id: str | None = Field(None, pattern=r"^[a-zA-Z0-9_-]{1,64}$")
 
 class ResearchRequest(Contract):
+    fresh_research: bool = False
     question: str = Field(default="Bitcoin latest market news today", min_length=1, max_length=2000)
     mode: Literal["live", "historical"] = "live"
     prediction_timestamp: datetime | None = None

app/services/agent_config.py

@@ -24,6 +24,11 @@ class AgentSettings:
     context_max_age_minutes: float
     tavily_timeout_seconds: float
     openai_timeout_seconds: float
+    monitor_interval_minutes: float = 60
+    price_move_trigger_pct: float = 1.0
+    volatility_trigger_multiplier: float = 1.5
+    sentiment_shift_threshold: float = .25
+    research_cooldown_minutes: float = 30
 
     @classmethod
     def from_env(cls):
@@ -31,17 +36,24 @@ class AgentSettings:
                    os.getenv("TAVILY_API_KEY", "").strip(), bounded_env("RELIABILITY_THRESHOLD", .50, 0, 1),
                    bounded_env("NEWS_CACHE_MINUTES", 10, 0, 120), bounded_env("NEWS_MAX_AGE_HOURS", 72, 1, 720),
                    bounded_env("CONTEXT_MAX_AGE_MINUTES", 90, 1, 1440),
-                   bounded_env("TAVILY_TIMEOUT_SECONDS", 12, 1, 60), bounded_env("OPENAI_TIMEOUT_SECONDS", 45, 1, 120))
+                   bounded_env("TAVILY_TIMEOUT_SECONDS", 12, 1, 60), bounded_env("OPENAI_TIMEOUT_SECONDS", 45, 1, 120),
+                   monitor_interval_minutes=bounded_env("MONITOR_INTERVAL_MINUTES", 60, 1, 1440),
+                   price_move_trigger_pct=bounded_env("PRICE_MOVE_TRIGGER_PCT", 1, .01, 100),
+                   volatility_trigger_multiplier=bounded_env("VOLATILITY_TRIGGER_MULTIPLIER", 1.5, 1.01, 20),
+                   sentiment_shift_threshold=bounded_env("SENTIMENT_SHIFT_THRESHOLD", .25, .001, 2),
+                   research_cooldown_minutes=bounded_env("RESEARCH_COOLDOWN_MINUTES", 30, 0, 1440))
 
     def public_status(self):
         return {"openai_configured": bool(self.openai_key and self.openai_model),
                 "tavily_configured": bool(self.tavily_key), "openai_model": self.openai_model or None,
                 "agent_framework": "OpenAI Agents SDK",
-                "reliability_threshold": self.reliability_threshold, "news_cache_minutes": self.news_cache_minutes}
+                "reliability_threshold": self.reliability_threshold, "news_cache_minutes": self.news_cache_minutes,
+                "monitor_interval_minutes": self.monitor_interval_minutes,
+                "research_cooldown_minutes": self.research_cooldown_minutes}
 
 def redact(value):
     text = str(value)
-    for name in ("OPENAI_API_KEY", "TAVILY_API_KEY", "ALPHA_VANTAGE_API_KEY"):
+    for name in ("OPENAI_API_KEY", "TAVILY_API_KEY", "ALPHA_VANTAGE_API_KEY", "HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN", "MONITOR_BACKUP_TOKEN"):
         key = os.getenv(name)
         if key:
             text = text.replace(key, "[redacted]")

app/services/market_monitoring.py

@@ -0,0 +1,82 @@
+"""Persist quantitative scans, then optionally research a deterministic event."""
+from __future__ import annotations
+import copy
+from datetime import datetime, timezone
+from app.agents.forecast_context_agent import ForecastContextAgent
+from app.agents.market_research_agent import MarketResearchAgent
+from app.agents.validation_agent import ValidationAgent
+from app.agents.decision_agent import DecisionAgent
+from app.schemas.agent_models import ResearchResult
+from .research_triggers import ResearchTriggerEngine, market_metrics
+
+
+class MarketMonitoringAgent:
+    def __init__(self, monitoring, research_agent=None, llm=None):
+        self.monitoring = monitoring
+        self.settings = monitoring.settings
+        self.store = monitoring.event_store
+        self.forecast = ForecastContextAgent(self.settings)
+        self.engine = ResearchTriggerEngine(self.settings)
+        self.research = research_agent or MarketResearchAgent(self.settings, store=self.store)
+        self.validation = ValidationAgent(self.settings)
+        self.decision = DecisionAgent(self.settings, self.validation, llm=llm)
+
+    def prepare(self, snapshot, candles, now=None):
+        metrics = market_metrics(candles, snapshot, self.settings.volatility_trigger_multiplier)
+        context = self.forecast.run(snapshot, now=now)
+        sentiment = context.sentiment.score
+        if context.sentiment.article_count == 0:
+            sentiment = None
+        return {**metrics, "data_fresh": context.data_fresh, "sentiment": sentiment,
+                "sentiment_scale": "weighted signed score [-1, 1]",
+                "forecasts": {h: f.model_dump() for h, f in context.forecasts.items()}}
+
+    def state(self, **values):
+        current = self.monitoring.store.get_state("research_operation", {})
+        self.monitoring.store.set_state("research_operation", {**current, **values})
+
+    def process(self, run_id, snapshot, candles, now=None):
+        now = now or datetime.now(timezone.utc)
+        current = self.prepare(snapshot, candles, now)
+        previous = self.store.scans(1)
+        trigger = self.engine.evaluate(current, previous[0]["payload"] if previous else None)
+        scan_id = self.store.scan(run_id, current, trigger, now)
+        self.monitoring.store.set_state("last_market_scan", now.isoformat())
+        # No paid provider or Decision Agent call on the ordinary path.
+        if not trigger["should_research"]:
+            self.state(status="IDLE", monitor="done", forecast="done", trigger="done", validation="idle", decision="idle")
+            return trigger
+        self.monitoring.store.set_state("last_research_trigger", {"timestamp": now.isoformat(), **trigger})
+        event, leader = self.store.claim(trigger["fingerprint"], current["market_timestamp"], trigger["triggered_by"], self.settings.research_cooldown_minutes, now)
+        self.store.link(scan_id, event["id"])
+        if not leader:
+            # Validate reused evidence again. Cooldown never extends source freshness.
+            cached = ResearchResult.model_validate(event["result"]) if event.get("result") else ResearchResult()
+            context = self.forecast.run(snapshot, now=now)
+            _, cached, _ = self.validation.validate(context, snapshot, cached, now=now)
+            self.state(status="CACHED" if cached.evidence else "IDLE", monitor="done", forecast="done", trigger="done", validation="done", decision="idle",
+                       event_id=event["id"], cache_reason="Repeated event suppressed; no provider calls.")
+            return {**trigger, "cached": bool(cached.evidence), "suppressed": True}
+        self.state(status="SEARCHING", monitor="done", forecast="done", trigger="done", validation="idle", decision="idle", event_id=event["id"])
+        try:
+            question = "Explain the latest Bitcoin market context. Research topics: " + ", ".join(trigger["triggered_by"])
+            result = self.research.run(question, trigger_types=trigger["triggered_by"])
+            self.state(status="CACHED" if result.cached else "SEARCHING", validation="running", decision="idle", event_id=event["id"])
+            context = self.forecast.run(snapshot, now=now)
+            context, result, validation = self.validation.validate(context, snapshot, result)
+            useful = any(item.recency_verified and item.source_quality in {"primary", "established"} for item in result.evidence)
+            summary, llm_status = None, None
+            if useful:
+                self.state(status="CACHED" if result.cached else "SEARCHING", validation="done", decision="running", event_id=event["id"])
+                answer, llm_status, _ = self.decision.run(question, [], context, result, validation)
+                if llm_status == "ok":
+                    summary = answer
+            failed = result.status == "unavailable" or llm_status in {"unavailable", "invalid_output"}
+            self.store.finish(event["id"], "error" if failed else "done" if useful else "empty", result.model_dump(), summary, llm_status)
+            self.state(status="ERROR" if failed else "CACHED" if result.cached else "IDLE", monitor="done", forecast="done", trigger="done", validation="done",
+                       decision="error" if failed and useful else "done" if useful else "idle", event_id=event["id"])
+        except Exception:
+            # Forecasts and the scan already committed; research must never undo them.
+            self.store.finish(event["id"], "error")
+            self.state(status="ERROR", monitor="done", forecast="done", trigger="done", validation="error", decision="idle", event_id=event["id"], error="Context research failed; market forecasts remain saved.")
+        return trigger

app/services/monitor_schedule.py

@@ -0,0 +1,10 @@
+"""UTC cadence, aligned two minutes after closed hourly candle boundaries."""
+from datetime import datetime, timedelta, timezone
+import math
+
+
+def next_cycle(now, interval_minutes=60):
+    anchor = datetime(1970, 1, 1, 0, 2, tzinfo=timezone.utc)
+    seconds = float(interval_minutes) * 60
+    slot = math.floor((now - anchor).total_seconds() / seconds) + 1
+    return anchor + timedelta(seconds=slot * seconds)

app/services/monitor_worker_runtime.py

@@ -0,0 +1,76 @@
+"""Browser-independent scheduler; API owns its durable lease and due time."""
+from __future__ import annotations
+import os
+import signal
+import threading
+import uuid
+from datetime import datetime, timezone
+import requests
+from .monitor_schedule import next_cycle
+
+UTC = timezone.utc
+
+
+def next_hourly_run(now):
+    return next_cycle(now, 60)
+
+
+class MonitoringWorker:
+    def __init__(self, api, session=None, now=None):
+        self.api = api.rstrip("/")
+        # Per-call sessions keep heartbeat and the long scan request independent.
+        self.session = session or requests
+        self.now = now or (lambda: datetime.now(UTC))
+        self.worker_id = uuid.uuid4().hex
+        self.stop = threading.Event()
+
+    def heartbeat(self):
+        response = self.session.post(self.api + "/monitoring/heartbeat", json={"worker_id": self.worker_id}, timeout=10)
+        response.raise_for_status()
+        return response.json()
+
+    def step(self):
+        status = self.heartbeat()
+        if not status.get("accepted"):
+            return "standby"
+        due = status.get("next_cycle_due")
+        if due and self.now() < datetime.fromisoformat(due):
+            return "waiting"
+        response = self.session.post(self.api + "/monitoring/run", json={
+            "trigger": "hourly", "worker_id": self.worker_id,
+            "use_gdelt": os.getenv("MONITOR_USE_GDELT", "true").lower() == "true"}, timeout=1500)
+        response.raise_for_status()
+        return response.json()["status"]
+
+    def run(self):
+        def renew():
+            while not self.stop.wait(25):
+                try:
+                    self.heartbeat()
+                except (requests.RequestException, ValueError):
+                    pass
+        thread = threading.Thread(target=renew, daemon=True)
+        thread.start()
+        try:
+            while not self.stop.is_set():
+                try:
+                    result = self.step()
+                    if result not in {"waiting", "standby"}:
+                        print(f"Monitoring cycle: {result}", flush=True)
+                except (requests.RequestException, ValueError, KeyError) as error:
+                    print(f"Monitoring API unavailable ({type(error).__name__}); retry shortly.", flush=True)
+                self.stop.wait(10)
+        finally:
+            self.stop.set()
+            thread.join(timeout=12)
+            try:
+                self.session.post(self.api + "/monitoring/worker/release", json={"worker_id": self.worker_id}, timeout=5)
+            except requests.RequestException:
+                pass
+
+
+def main():
+    worker = MonitoringWorker(os.getenv("BTC_API_URL", "http://btc-api:8000"))
+    signal.signal(signal.SIGTERM, lambda *_: worker.stop.set())
+    signal.signal(signal.SIGINT, lambda *_: worker.stop.set())
+    worker.run()

app/services/monitoring_backup.py

@@ -0,0 +1,91 @@
+"""Optional private Hub snapshots for Spaces with ephemeral local disks.
+
+SQLite remains on local disk. Only consistent backup files are uploaded;
+we never run SQLite/WAL directly on an object-storage mount.
+"""
+from __future__ import annotations
+import os
+from pathlib import Path
+import shutil
+import sqlite3
+import tempfile
+import threading
+from contextlib import closing
+from datetime import datetime, timezone
+
+
+class MonitoringBackup:
+    def __init__(self, repo=None, client=None, downloader=None):
+        self.repo = repo if repo is not None else os.getenv("MONITOR_BACKUP_REPO", "").strip()
+        self.token = os.getenv("MONITOR_BACKUP_TOKEN") or os.getenv("HF_TOKEN")
+        self.client = client
+        self.downloader = downloader
+        self.revision = None
+        self.lock = threading.Lock()
+
+    def api(self):
+        if self.client is None:
+            from huggingface_hub import HfApi
+            self.client = HfApi(token=self.token)
+        return self.client
+
+    def restore(self, path):
+        path = Path(path)
+        if not self.repo or (path.exists() and path.stat().st_size):
+            return False
+        path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            info = self.api().repo_info(self.repo, repo_type="dataset")
+            if not info.private:
+                raise RuntimeError("Monitoring backup repository must be private")
+            self.revision = info.sha
+            download = self.downloader
+            if download is None:
+                from huggingface_hub import hf_hub_download
+                download = hf_hub_download
+            source = download(repo_id=self.repo, filename="monitoring.sqlite3", repo_type="dataset",
+                              revision=self.revision, token=self.token, etag_timeout=15)
+            # Validate before replacing anything; a failed restore cannot create
+            # an empty database that would later overwrite the durable history.
+            with closing(sqlite3.connect(f"file:{Path(source).as_posix()}?mode=ro", uri=True)) as db:
+                if db.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
+                    raise RuntimeError("Invalid monitoring backup")
+                db.execute("SELECT id,probability_up,issued_at FROM forecasts LIMIT 1")
+            temporary = path.with_suffix(".restoring")
+            shutil.copyfile(source, temporary)
+            os.replace(temporary, path)
+            return True
+        except Exception:
+            raise RuntimeError("Monitoring history restore failed. Check private backup access; refusing to start with an empty history.") from None
+
+    def save(self, store):
+        if not self.repo or not self.lock.acquire(blocking=False):
+            return
+        try:
+            if self.revision is None:
+                info = self.api().repo_info(self.repo, repo_type="dataset")
+                if not info.private:
+                    raise RuntimeError("Monitoring backup must be private")
+                self.revision = info.sha
+            with tempfile.TemporaryDirectory() as directory:
+                target = Path(directory) / "monitoring.sqlite3"
+                with store._connection() as source, closing(sqlite3.connect(target)) as destination:
+                    source.backup(destination)
+                    # A downloadable snapshot must be self-contained; source
+                    # WAL mode otherwise leaves cleanup writes in a sidecar.
+                    destination.execute("PRAGMA journal_mode=DELETE")
+                    # Runtime ownership cannot survive a container replacement.
+                    destination.execute("DELETE FROM worker_lock")
+                    destination.execute("DELETE FROM operation_leases")
+                    destination.execute("DELETE FROM state WHERE key IN ('worker','active_run')")
+                    destination.execute("UPDATE research_events SET status='error' WHERE status='running'")
+                    destination.commit()
+                commit = self.api().upload_file(path_or_fileobj=str(target), path_in_repo="monitoring.sqlite3",
+                    repo_id=self.repo, repo_type="dataset", parent_commit=self.revision,
+                    commit_message="Save monitoring history and event state")
+                self.revision = commit.oid
+            store.set_state("backup", {"status": "OK", "saved_at": datetime.now(timezone.utc).isoformat()})
+        except Exception:
+            store.set_state("backup", {"status": "ERROR", "error": "Durable backup unavailable; local history remains saved. Check access or conflicting writers."})
+        finally:
+            self.lock.release()

app/services/monitoring_events_store.py

@@ -0,0 +1,111 @@
+"""Additive SQLite storage for scans, research references, caches and leases."""
+from __future__ import annotations
+import json
+import uuid
+from datetime import timedelta
+from app.monitoring_store import _utc, _iso, _json
+
+
+class EventStore:
+    def __init__(self, store):
+        self.store = store
+        with store._connection() as db:
+            db.executescript("""
+                CREATE TABLE IF NOT EXISTS monitoring_scans (
+                    id INTEGER PRIMARY KEY, run_id INTEGER UNIQUE NOT NULL,
+                    scanned_at TEXT NOT NULL, market_timestamp TEXT, payload TEXT NOT NULL,
+                    trigger_result TEXT NOT NULL, research_event_id TEXT
+                );
+                CREATE TABLE IF NOT EXISTS research_events (
+                    id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, market_timestamp TEXT,
+                    created_at TEXT NOT NULL, status TEXT NOT NULL, reasons TEXT NOT NULL,
+                    result TEXT, summary TEXT, llm_status TEXT, completed_at TEXT
+                );
+                CREATE INDEX IF NOT EXISTS research_fingerprint ON research_events(fingerprint,created_at);
+                CREATE TABLE IF NOT EXISTS research_query_cache (
+                    key TEXT PRIMARY KEY, expires_at TEXT NOT NULL, payload TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS operation_leases (
+                    name TEXT PRIMARY KEY, owner TEXT NOT NULL, expires_at TEXT NOT NULL
+                );
+            """)
+
+    def acquire(self, name, owner, ttl=120, now=None):
+        now = _utc(now)
+        with self.store._connection() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute("SELECT * FROM operation_leases WHERE name=?", (name,)).fetchone()
+            if row and row["owner"] != owner and row["expires_at"] > _iso(now):
+                return False
+            db.execute("INSERT INTO operation_leases VALUES(?,?,?) ON CONFLICT(name) DO UPDATE SET owner=excluded.owner,expires_at=excluded.expires_at",
+                       (name, owner, _iso(now + timedelta(seconds=ttl))))
+        return True
+
+    def owns(self, name, owner, now=None):
+        with self.store._connection() as db:
+            row = db.execute("SELECT 1 FROM operation_leases WHERE name=? AND owner=? AND expires_at>?", (name, owner, _iso(now))).fetchone()
+        return bool(row)
+
+    def release(self, name, owner):
+        with self.store._connection() as db:
+            db.execute("DELETE FROM operation_leases WHERE name=? AND owner=?", (name, owner))
+
+    def scan(self, run_id, payload, trigger, now=None):
+        with self.store._connection() as db:
+            db.execute("INSERT INTO monitoring_scans(run_id,scanned_at,market_timestamp,payload,trigger_result) VALUES(?,?,?,?,?) ON CONFLICT(run_id) DO NOTHING",
+                       (run_id, _iso(now), payload.get("market_timestamp"), _json(payload), _json(trigger)))
+            return db.execute("SELECT id FROM monitoring_scans WHERE run_id=?", (run_id,)).fetchone()[0]
+
+    def scans(self, limit=50):
+        with self.store._connection() as db:
+            rows = db.execute("SELECT * FROM monitoring_scans ORDER BY id DESC LIMIT ?", (max(1, min(int(limit), 200)),)).fetchall()
+        return [{**dict(row), "payload": json.loads(row["payload"]), "trigger_result": json.loads(row["trigger_result"])} for row in rows]
+
+    def link(self, scan_id, event_id):
+        with self.store._connection() as db:
+            db.execute("UPDATE monitoring_scans SET research_event_id=? WHERE id=?", (event_id, scan_id))
+
+    @staticmethod
+    def event(row):
+        if not row:
+            return None
+        result = dict(row)
+        for field in ("reasons", "result"):
+            result[field] = json.loads(result[field]) if result[field] else None
+        return result
+
+    def claim(self, fingerprint, market_timestamp, reasons, cooldown_minutes, now=None):
+        now = _utc(now)
+        with self.store._connection() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute("SELECT * FROM research_events WHERE fingerprint=? ORDER BY created_at DESC LIMIT 1", (fingerprint,)).fetchone()
+            if row and (row["market_timestamp"] == market_timestamp or row["created_at"] > _iso(now - timedelta(minutes=cooldown_minutes))):
+                return self.event(row), False
+            event_id = uuid.uuid4().hex
+            db.execute("INSERT INTO research_events(id,fingerprint,market_timestamp,created_at,status,reasons) VALUES(?,?,?,?,?,?)",
+                       (event_id, fingerprint, market_timestamp, _iso(now), "running", _json(reasons)))
+            row = db.execute("SELECT * FROM research_events WHERE id=?", (event_id,)).fetchone()
+        return self.event(row), True
+
+    def finish(self, event_id, status, result=None, summary=None, llm_status=None, now=None):
+        with self.store._connection() as db:
+            db.execute("UPDATE research_events SET status=?,result=?,summary=?,llm_status=?,completed_at=? WHERE id=?",
+                       (status, _json(result) if result is not None else None, summary, llm_status, _iso(now), event_id))
+
+    def latest_event(self, with_evidence=False):
+        with self.store._connection() as db:
+            row = db.execute("SELECT * FROM research_events " + ("WHERE result IS NOT NULL " if with_evidence else "") + "ORDER BY created_at DESC LIMIT 1").fetchone()
+        return self.event(row)
+
+    def cache_get(self, key, now=None):
+        with self.store._connection() as db:
+            row = db.execute("SELECT payload FROM research_query_cache WHERE key=? AND expires_at>?", (key, _iso(now))).fetchone()
+        return json.loads(row[0]) if row else None
+
+    def cache_put(self, key, payload, seconds, now=None):
+        now = _utc(now)
+        with self.store._connection() as db:
+            db.execute("DELETE FROM research_query_cache WHERE expires_at<=?", (_iso(now),))
+            db.execute("INSERT INTO research_query_cache VALUES(?,?,?) ON CONFLICT(key) DO UPDATE SET expires_at=excluded.expires_at,payload=excluded.payload",
+                       (key, _iso(now + timedelta(seconds=seconds)), _json(payload)))
+            db.execute("DELETE FROM research_query_cache WHERE key NOT IN (SELECT key FROM research_query_cache ORDER BY expires_at DESC LIMIT 256)")

app/services/research_triggers.py

@@ -0,0 +1,100 @@
+"""Deterministic monitoring features and event rules; never model inputs."""
+from __future__ import annotations
+import hashlib
+import json
+import math
+import pandas as pd
+
+
+def finite(value):
+    try:
+        value = float(value)
+        return value if math.isfinite(value) else None
+    except (TypeError, ValueError):
+        return None
+
+
+def market_metrics(candles, snapshot, volatility_multiplier=1.5):
+    # Match existing market features: sample std of hourly log returns and
+    # fractional close-to-close momentum. Use completed, contiguous candles.
+    frame = candles.sort_values("timestamp").drop_duplicates("timestamp").copy()
+    times = pd.to_datetime(frame["timestamp"], utc=True)
+    close = pd.Series(pd.to_numeric(frame["close"], errors="coerce").values, index=times)
+    close = close.asfreq("h")
+    returns = close.map(lambda x: math.log(x) if pd.notna(x) and x > 0 else float("nan")).diff()
+    def change(hours):
+        if len(close) <= hours or pd.isna(close.iloc[-hours-1]) or close.iloc[-hours-1] <= 0:
+            return None
+        return finite((close.iloc[-1] / close.iloc[-hours-1] - 1) * 100)
+    volatility = finite(returns.rolling(24, min_periods=24).std().iloc[-1])
+    # Prior 168 hourly returns: baseline excludes the current 24-hour window.
+    baseline = finite(returns.shift(24).rolling(168, min_periods=168).std().iloc[-1])
+    move24 = change(24)
+    market = snapshot.get("market", {})
+    regime = market.get("regime")
+    basis = "upstream" if regime else "monitor-only: 24h momentum versus 24h realized volatility"
+    if not regime and volatility is not None and baseline is not None and baseline > 0 and move24 is not None:
+        # Fixed descriptive regime boundaries, independent of research thresholds.
+        # These labels never alter model features, probabilities or reliability.
+        if volatility >= baseline * volatility_multiplier:
+            regime = "HIGH_VOL"
+        elif move24 / 100 > volatility * math.sqrt(24):
+            regime = "BULL"
+        elif move24 / 100 < -volatility * math.sqrt(24):
+            regime = "BEAR"
+        else:
+            regime = "RANGE"
+    return {"market_timestamp": times.iloc[-1].isoformat(), "price": finite(close.iloc[-1]),
+            "price_change_1h": change(1), "price_change_24h": move24,
+            "volatility": volatility, "volatility_baseline": baseline,
+            "momentum": None if move24 is None else move24 / 100,
+            "regime": regime, "regime_basis": basis}
+
+
+class ResearchTriggerEngine:
+    def __init__(self, settings):
+        self.settings = settings
+
+    def evaluate(self, current, previous=None):
+        previous = previous or {}
+        reasons, details = [], {}
+        # Repeated scans of the same closed candle do not constitute new events.
+        fresh = current.get("data_fresh") is True
+        changed_candle = current.get("market_timestamp") != previous.get("market_timestamp")
+        if not fresh or not changed_candle:
+            return {"should_research": False, "triggered_by": [], "details": {}, "fingerprint": None}
+        move = finite(current.get("price_change_1h"))
+        if move is not None and abs(move) >= self.settings.price_move_trigger_pct:
+            reasons.append("large_price_move")
+            details["price_change_1h"] = move
+        old_regime, regime = previous.get("regime"), current.get("regime")
+        if old_regime and regime and old_regime != regime:
+            reasons.append("regime_change")
+            details.update(previous_regime=old_regime, current_regime=regime)
+        vol, baseline = finite(current.get("volatility")), finite(current.get("volatility_baseline"))
+        if vol is not None and baseline is not None and baseline > 0 and vol >= baseline * self.settings.volatility_trigger_multiplier:
+            reasons.append("unusual_volatility")
+            details.update(volatility=vol, volatility_baseline=baseline)
+        sentiment, old_sentiment = finite(current.get("sentiment")), finite(previous.get("sentiment"))
+        if sentiment is not None and old_sentiment is not None:
+            delta = sentiment - old_sentiment
+            # Meaningful reversal requires both sides outside a small neutral band.
+            reversal = sentiment * old_sentiment < 0 and min(abs(sentiment), abs(old_sentiment)) >= self.settings.sentiment_shift_threshold / 2
+            if abs(delta) >= self.settings.sentiment_shift_threshold or reversal:
+                reasons.append("sentiment_shift")
+                details.update(sentiment_change=delta, sentiment=sentiment)
+        transitions = {}
+        for horizon, forecast in current.get("forecasts", {}).items():
+            old = previous.get("forecasts", {}).get(horizon, {})
+            before, after = old.get("signal_state"), forecast.get("signal_state")
+            if before and after != before and after in {"UP", "DOWN", "NO_SIGNAL"} and (before in {"UP", "DOWN"} or after in {"UP", "DOWN"}):
+                transitions[horizon] = {"before": before, "after": after}
+        if transitions:
+            reasons.append("forecast_state_change")
+            details["signal_transitions"] = transitions
+        signature = {"reasons": sorted(reasons), "regime": regime,
+                     "move_bucket": None if move is None else math.trunc(move / self.settings.price_move_trigger_pct),
+                     "sentiment_direction": None if sentiment is None else (0 if abs(sentiment) < self.settings.sentiment_shift_threshold / 2 else 1 if sentiment > 0 else -1),
+                     "signals": {h: f.get("signal_state") for h, f in current.get("forecasts", {}).items()}}
+        return {"should_research": bool(reasons), "triggered_by": reasons, "details": details,
+                "fingerprint": hashlib.sha256(json.dumps(signature, sort_keys=True).encode()).hexdigest() if reasons else None}

app/space_server.py

@@ -22,7 +22,7 @@ def create_app():
         worker = None
         if os.getenv("SPACE_START_MONITOR", "true").lower() == "true":
             env = dict(os.environ, BTC_API_URL=local_api)
-            for key in ("OPENAI_API_KEY", "TAVILY_API_KEY", "ALPHA_VANTAGE_API_KEY"):
+            for key in ("OPENAI_API_KEY", "TAVILY_API_KEY", "ALPHA_VANTAGE_API_KEY", "MONITOR_BACKUP_TOKEN", "HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN"):
                 env.pop(key, None)
             worker = subprocess.Popen([sys.executable, "-m", "app.monitor_worker"], env=env)
         try:

app/static/monitor/analyst.css

@@ -1 +1,3 @@
 .ai-heading{padding-bottom:24px}.ai-heading h1{font-size:34px}.live-state{font:10px var(--mono);letter-spacing:1px;color:var(--muted);border:1px solid var(--border);padding:8px 10px;border-radius:4px;white-space:nowrap}.live-state.live{color:var(--green);background:#152b24;border-color:#28513f}.market-strip{display:grid;grid-template-columns:1.25fr 1fr 1fr 1.2fr;border:1px solid var(--border);border-radius:7px;padding:20px 22px;background:linear-gradient(110deg,#142126,#13191d);gap:18px;align-items:center}.market-strip>div{min-width:0}.market-strip span{font:8px var(--mono);letter-spacing:.7px;color:#8c9ba4;display:block;line-height:1.6}.market-strip strong{display:block;font-size:12px;font-weight:500;margin-top:9px;overflow-wrap:anywhere}.market-price strong{font-size:27px;font-weight:550;letter-spacing:-.7px;color:#e9f5f8;font-variant-numeric:tabular-nums}.section-heading{display:flex;align-items:center;justify-content:space-between;margin:28px 0 16px;gap:15px}.section-heading h2{font-size:15px;font-weight:550;margin:0}.section-heading p{font-size:11px;color:var(--muted);margin:7px 0 0}.forecast-cards{display:grid;grid-template-columns:repeat(3,minmax(0,1fr));gap:16px}.signal-card{padding:20px;border:1px solid var(--border);border-radius:7px;background:var(--panel);min-width:0;position:relative}.signal-card::before{content:'';position:absolute;top:-1px;left:20px;width:36px;height:2px;background:#5c717b}.signal-card.qualified-up::before{background:var(--green)}.signal-card.qualified-down::before{background:var(--red)}.signal-card-top{display:flex;align-items:center;justify-content:space-between;gap:8px;font:10px var(--mono);color:#b3c5cd;letter-spacing:.7px}.signal-state{font:8px var(--mono);border:1px solid #3b454b;color:#9aaab3;border-radius:3px;padding:4px 6px}.signal-title{font-size:19px;line-height:1.3;font-weight:500;letter-spacing:-.4px;margin:21px 0 9px;color:#c7d4db}.qualified-up .signal-title{color:var(--green)}.qualified-down .signal-title{color:var(--red)}.signal-reason{color:#8398a4;font-size:10px;line-height:1.7;margin:0 0 18px;min-height:34px}.signal-numbers{display:grid;grid-template-columns:1fr 1fr;border-top:1px solid #283138;border-bottom:1px solid #283138;padding:14px 0;gap:10px}.signal-numbers span{display:block;color:#8b9fa9;font-size:10px}.signal-numbers strong{display:block;font:20px var(--mono);margin-top:8px;letter-spacing:-.5px;color:#d0dce2}.signal-facts{margin:15px 0 0;font-size:10px;display:grid;gap:10px}.signal-facts>div{display:flex;justify-content:space-between;gap:12px}.signal-facts dt{color:#77909f;flex-shrink:0}.signal-facts dd{margin:0;text-align:right;color:#afc1cc;overflow-wrap:anywhere}.signal-note{font-size:10px;color:#7893a2;line-height:1.8;margin:13px 0 24px}.analyst-layout{display:grid;grid-template-columns:minmax(0,1.5fr) minmax(300px,1fr);gap:18px;align-items:start}.analyst-panel .eyebrow,.evidence-panel .eyebrow{display:block;font-size:8px;letter-spacing:1.3px;margin-bottom:9px}.analyst-panel .panel-heading h2,.evidence-panel .panel-heading h2{font-size:16px}.provider-status{font:9px var(--mono);line-height:1.7;color:#85a3b2;padding:0 22px 15px;overflow-wrap:anywhere}.agent-strip{display:flex;gap:7px;flex-wrap:wrap;padding:12px 22px;border-top:1px solid #243139;border-bottom:1px solid #243139;background:#121b20}.agent-strip span{font:9px var(--mono);color:#8499a6;border:1px solid #2b3b44;border-radius:4px;padding:5px 7px}.agent-strip span[data-state=running]{color:var(--cyan);border-color:#2a6570;background:#133139}.agent-strip span[data-state=done]{color:#a2c6b9}.agent-strip span[data-state=unavailable],.agent-strip span[data-state=error]{color:#c4b591}.chat-log{min-height:300px;max-height:590px;overflow-y:auto;padding:22px;scrollbar-color:#3b525f transparent}.chat-welcome{display:flex;align-items:center;flex-direction:column;text-align:center;padding:24px 20px}.analyst-orb{display:grid;place-items:center;width:53px;height:53px;background:radial-gradient(circle,#1e4148,#142329);border:1px solid #2b525d;border-radius:14px;font-size:29px;color:var(--cyan);box-shadow:0 0 32px #21bad00c}.chat-welcome h3{font-size:19px;font-weight:500;letter-spacing:-.4px;margin:18px 0 8px}.chat-welcome p{font-size:11px;line-height:1.9;max-width:280px;margin:0;color:#91a5b2}.chat-welcome>span{font:8px var(--mono);color:#657e8d;letter-spacing:.6px;margin-top:19px}.chat-message{margin:0 0 20px}.chat-message:last-child{margin-bottom:0}.message-label{font:8px var(--mono);color:#7c96a6;letter-spacing:1px;margin-bottom:8px;display:flex;justify-content:space-between;gap:10px}.message-content{white-space:pre-wrap;overflow-wrap:anywhere;font-size:12px;line-height:1.85;color:#c2d2dc;padding:15px 17px;background:#172229;border:1px solid #283b46;border-radius:7px}.chat-message.user{margin-left:42px}.chat-message.user .message-content{background:#152b32;border-color:#244752;color:#d1e9ef}.chat-message.assistant .message-label{color:var(--cyan)}.message-sources{display:flex;flex-wrap:wrap;gap:7px;margin-top:10px}.source-link{font-size:10px;color:#9dcedb;border:1px solid #33515e;background:#172830;border-radius:4px;padding:6px 8px}.source-link:hover{background:#213c47}.quick-questions{display:flex;gap:7px;flex-wrap:wrap;padding:0 22px 18px}.quick-questions button{font-size:9px;color:#9bb5c4;background:#19252c;border:1px solid #2c3e49;padding:7px 9px;border-radius:5px;text-align:left}.quick-questions button:hover{color:#c6eff4;border-color:#386b79;background:#19313a}#chat-form{border-top:1px solid #293741;padding:17px 22px 16px}.chat-compose{display:flex;gap:10px;align-items:flex-end;border:1px solid #365261;background:#101a20;border-radius:6px;padding:10px}.chat-compose:focus-within{border-color:#47a3b6;box-shadow:0 0 0 2px #28c4d413}.chat-compose textarea{background:transparent;border:0;color:#d8e6ed;font:12px var(--sans);line-height:1.7;resize:vertical;min-height:43px;max-height:150px;width:100%;padding:3px 4px;outline:none}.chat-compose textarea::placeholder{color:#66818f}.chat-compose textarea:focus-visible{outline:none}.chat-compose .primary-button{font-size:10px;flex-shrink:0}.chat-foot{display:flex;justify-content:space-between;gap:12px;font-size:8px;line-height:1.7;color:#6d8c9e;margin-top:10px}.evidence-disclaimer{margin:0 20px 16px;padding:12px 13px;border-left:2px solid #376c7b;background:#13242b;font-size:10px;line-height:1.8;color:#8fb0c0}.evidence-list{max-height:670px;overflow-y:auto;padding:0 20px 4px;scrollbar-color:#3b525f transparent}.evidence-item{padding:17px 0;border-top:1px solid #2a363e}.evidence-tags{display:flex;gap:6px;align-items:center;flex-wrap:wrap;margin-bottom:10px}.evidence-tag{font:8px var(--mono);letter-spacing:.4px;padding:4px 6px;border-radius:3px;background:#23323b;color:#92afc0}.evidence-tag.bullish{background:#19382e;color:#88c9b2}.evidence-tag.bearish{background:#392a30;color:#d5a0a9}.evidence-item h3{font-size:12px;font-weight:500;line-height:1.7;margin:0 0 9px}.evidence-item h3 a:hover{color:var(--cyan)}.evidence-meta{font:8px var(--mono);line-height:1.8;color:#859eac;overflow-wrap:anywhere}.evidence-item p{font-size:10px;color:#9bb0be;line-height:1.85;margin:10px 0}.evidence-item .source-link{display:inline-block;margin-top:4px}.evidence-empty{padding:35px 8px 42px;text-align:center;display:flex;align-items:center;flex-direction:column}.evidence-empty>span{color:#487888;font-size:33px;margin-bottom:17px}.evidence-empty strong{font-size:13px;font-weight:500}.evidence-empty p{font-size:11px;line-height:1.9;color:#7c99aa;max-width:260px}.evidence-footer{border-top:1px solid #27363f;padding:13px 20px;font-size:9px;line-height:1.7;color:#7996a6}.monitoring-heading{padding-top:34px}.monitoring-heading h2{font-size:23px;font-weight:500;letter-spacing:-.5px;margin:0;display:flex;align-items:center;gap:12px}.prediction-direction,.prediction-direction.down{color:#becfd8}.sr-only{position:absolute;width:1px;height:1px;padding:0;margin:-1px;overflow:hidden;clip:rect(0,0,0,0);white-space:nowrap;border:0}@media(max-width:1150px){.signal-card{padding:17px 15px}.signal-title{font-size:16px}.signal-facts{font-size:9px}.analyst-layout{grid-template-columns:minmax(0,1.3fr) minmax(280px,1fr)}.chat-foot>span:last-child{display:none}}@media(max-width:1000px){.analyst-layout{grid-template-columns:1fr}.evidence-list{max-height:450px}.market-strip{grid-template-columns:repeat(2,minmax(0,1fr))}.forecast-cards{gap:10px}.signal-facts>div{display:block}.signal-facts dd{text-align:left;margin-top:4px}.signal-title{min-height:42px}.ai-heading h1{font-size:28px}}@media(max-width:700px){.forecast-cards{grid-template-columns:1fr}.signal-title{min-height:0}.signal-reason{min-height:0}.signal-facts>div{display:flex}.signal-facts dd{margin-top:0;text-align:right}.signal-card{padding:19px}.signal-title{font-size:21px}.market-strip{padding:18px}.ai-heading h1{font-size:25px}.ai-heading p{line-height:1.9}.live-state{font-size:8px;padding:7px}.chat-log{padding:17px;max-height:470px}.quick-questions{padding:0 17px 16px}#chat-form{padding:15px 17px}.agent-strip{padding:12px 17px}.provider-status{padding:0 17px 13px}.section-heading .subtle-tag{display:none}.chat-message.user{margin-left:20px}.monitoring-heading h2{font-size:21px}.sidebar{overflow-y:auto}}
+
+.continuous-grid{display:grid;grid-template-columns:1fr 1fr;gap:18px;margin:24px 0 18px}.continuous-facts{padding:0 22px 14px;margin:0}.continuous-facts div{display:flex;justify-content:space-between;gap:18px;padding:7px 0}.continuous-facts dt{color:#91a8b5}.continuous-facts dd{margin:0;text-align:right}.continuous-context{padding:22px;margin-bottom:20px}.continuous-context p{line-height:1.7;white-space:pre-wrap}.continuous-grid .signal-note{padding:0 22px 15px}#continuous-agent-status{display:flex;flex-wrap:wrap;gap:10px;margin-top:18px}#continuous-agent-status span{border:1px solid #29414e;border-radius:5px;padding:6px 9px;font-size:12px}#qualification-groups{display:flex;gap:25px;flex-wrap:wrap;color:#91a8b5;font-size:12px}@media(max-width:780px){.continuous-grid{grid-template-columns:1fr}.continuous-facts div{display:block}.continuous-facts dd{text-align:left;margin-top:5px}}

app/static/monitor/analyst.js

@@ -137,7 +137,7 @@
       finally { polling = false; }
     }, 1100);
     try {
-      const data = await api('/' + kind, kind === 'chat' ? {message:question,history:previous,request_id:requestId} : {question,request_id:requestId});
+      const data = await api('/' + kind, kind === 'chat' ? {message:question,history:previous,request_id:requestId} : {question,request_id:requestId,fresh_research:true});
       if (generation !== epoch) return;
       activity(data.agent_status);
       if (kind === 'chat') {

app/static/monitor/dashboard.js

@@ -15,6 +15,16 @@ function query(history=false){const p=new URLSearchParams({horizon:state.horizon
 async function readJson(url){const r=await fetch(url,{cache:'no-store',signal:AbortSignal.timeout(20000)});if(!r.ok)throw new Error('API returned '+r.status);return r.json();}
 
 function renderSummary(data){
+  const q=data.qualification;
+  if(q){
+    setText('qualification-performance','Signal coverage: '+percent(q.signal_coverage)+' | NO_SIGNAL: '+percent(q.no_signal_frequency)+' | UNKNOWN: '+percent(q.unknown_frequency)+' | Qualified accuracy: '+percent(q.qualified_performance.accuracy));
+    const groups=$('qualification-groups');groups.replaceChildren();
+    for(const [label,values] of [['Reliability',q.reliability_buckets],['Regime',q.regimes]]){
+      for(const [key,m] of Object.entries(values)){groups.append(element('span',label+' '+key+': '+percent(m.accuracy)+' accuracy / '+m.scored+' scored'));}
+    }
+    if(q.legacy_unannotated)groups.append(element('span',q.legacy_unannotated+' older forecasts have no saved issue-time gate; excluded from coverage.'));
+  }
+
   const sys=data.system, model=data.model, latest=data.latest;
   setText('crumb','btc-direction-'+state.horizon+'h');
   setText('model-name',modelName(latest?.model || model.model));

app/static/monitor/index.html

@@ -8,6 +8,7 @@
   <script src="/monitor/assets/dashboard.js" defer></script>
   <link rel="stylesheet" href="/monitor/assets/analyst.css">
   <script src="/monitor/assets/analyst.js" defer></script>
+  <script src="/monitor/assets/monitoring-events.js" defer></script>
 </head>
 <body>
 <aside class="sidebar">
@@ -39,6 +40,10 @@
     <article class="panel analyst-panel" id="analyst"><div class="panel-heading"><div><span class="eyebrow">RESEARCH WORKSPACE</span><h2>AI Market Analyst</h2><p>Ask about your models and the news around Bitcoin.</p></div><button class="text-button" id="clear-chat" type="button">Clear chat</button></div><div class="provider-status" id="provider-status">Checking provider configuration…</div><div class="agent-strip" aria-label="Agent activity"><span id="agent-forecast">Forecast · idle</span><span id="agent-research">Research · idle</span><span id="agent-validation">Validation · idle</span><span id="agent-decision">Decision · idle</span></div><div class="chat-log" id="chat-log" role="log" aria-label="Chat history" aria-live="polite"><div class="chat-welcome"><div class="analyst-orb">✧</div><h3>A clearer view of BTC.</h3><p>Compare model signals, inspect reliability, or explore current market evidence with linked sources.</p><span>Grounded in your saved forecasts</span></div></div><div class="quick-questions" id="quick-questions" aria-label="Suggested questions"></div><form id="chat-form"><label class="sr-only" for="chat-input">Ask the AI market analyst</label><div class="chat-compose"><textarea id="chat-input" maxlength="2000" rows="2" placeholder="Ask about BTC, a forecast, or current news…" required></textarea><button class="primary-button" id="send-chat" type="submit">Send ↗</button></div><div class="chat-foot"><span id="chat-status" role="status">Ready · research runs only when needed</span><span>Enter to send · Shift + Enter for a new line</span></div></form></article>
     <article class="panel evidence-panel" id="evidence"><div class="panel-heading"><div><span class="eyebrow">TAVILY RESEARCH</span><h2>Live Market Evidence</h2><p id="evidence-status">No research requested yet</p></div><button class="text-button" id="research-news" type="button" aria-label="Research latest Bitcoin news">↻ Research</button></div><div class="evidence-disclaimer">External web evidence provides market context and does not necessarily explain the forecasting model's prediction.</div><div class="evidence-list" id="evidence-list"><div class="evidence-empty"><span>◎</span><strong>Follow the evidence.</strong><p>Ask a news question or select Research to fetch current sources. Refreshing this dashboard does not trigger paid research.</p></div></div><div class="evidence-footer" id="evidence-footer">News never increases model reliability.</div></article>
   </section>
+<section class="continuous-grid" aria-label="Continuous market monitoring">
+<article class="panel"><div class="panel-heading"><div><h2>Market monitor</h2><p id="scan-cadence">Scheduled independently of this browser</p></div><span class="badge" id="scan-worker">CONNECTING</span></div><dl class="continuous-facts"><div><dt>Last market scan</dt><dd id="scan-last">Not yet available</dd></div><div><dt>Next market scan</dt><dd id="scan-next">Not yet available</dd></div><div><dt>Last successful prediction cycle</dt><dd id="scan-success">Not yet available</dd></div><div><dt>History backup</dt><dd id="scan-backup">Checking</dd></div></dl><p class="signal-note" id="scan-error"></p></article>
+<article class="panel"><div class="panel-heading"><div><h2>Market research</h2><p>Research runs when meaningful changes occur</p></div><span class="badge" id="research-state">IDLE</span></div><dl class="continuous-facts"><div><dt>Last research</dt><dd id="research-last-time">Not yet available</dd></div><div><dt>Trigger</dt><dd id="research-trigger-reason">No event yet</dd></div></dl><p class="signal-note" id="research-event-note">Web research is event-triggered to reduce unnecessary API calls.</p></article>
+</section><section class="panel continuous-context"><div class="panel-heading"><h2>Latest event context</h2></div><p id="stored-market-context">Awaiting validated research.</p><div id="continuous-agent-status" aria-label="Operational agent status"></div></section>
   <section class="page-heading monitoring-heading"><div><div class="eyebrow">FORWARD PERFORMANCE</div><h2>Model monitoring <span class="badge" id="model-badge">COLLECTING</span></h2><p>Every forecast recorded. Every outcome accounted for.</p></div><button class="primary-button" id="refresh-button">↻ <span>Run refresh</span></button></section>
   <div class="toolbar"><div class="model-description"><strong id="model-name">VAE-Transformer</strong><span class="separator">/</span><span id="feature-set">Market only</span><span class="small-badge" id="horizon-description">1h horizon</span></div><div class="period" aria-label="Accuracy window"><button data-days="7" class="selected" aria-pressed="true">7 days</button><button data-days="30" aria-pressed="false">30 days</button></div></div>
   <div class="notice" id="notice" role="status" hidden></div>
@@ -48,6 +53,7 @@
     <article class="metric-card"><div class="metric-label">PENDING OUTCOMES <span>◷</span></div><div class="metric-value" id="pending">0</div><div class="metric-detail" id="pending-detail">Waiting for the first forecast</div><div class="metric-foot" id="missing-detail">Predictions are saved before evaluation</div></article>
     <article class="metric-card"><div class="metric-label">BRIER SCORE <span>◎</span></div><div class="metric-value" id="brier">—</div><div class="metric-detail">Probability quality · lower is better</div><div class="metric-foot">0 is perfect · 0.25 for constant 50%</div></article>
   </section>
+  <section class="panel continuous-context"><h2>Signal qualification performance</h2><p id="qualification-performance">Awaiting issue-time signal records.</p><div id="qualification-groups"></div></section>
   <section class="chart-grid">
     <article class="panel performance-panel"><div class="panel-heading"><div><h2>Accuracy over time</h2><p id="chart-description">Trailing 7-day accuracy · scored forecasts only</p></div><span class="subtle-tag">LIVE EVALUATION</span></div><div class="chart-legend"><span><i class="legend-line cyan-line"></i> Model</span><span><i class="legend-line muted-line"></i> Always UP baseline</span></div><div class="chart-wrap"><canvas id="accuracy-chart" aria-label="Rolling model accuracy and always UP baseline" role="img"></canvas><div class="chart-empty" id="chart-empty"><span class="empty-symbol">⌁</span><strong>Good evidence takes time.</strong><p>Your accuracy curve appears as forecasts reach their target time.</p></div></div><div class="chart-footer"><span id="window-note">No historical tests are included in live accuracy.</span><span>UTC</span></div></article>
     <article class="panel latest-panel"><div class="panel-heading"><div><h2>Latest raw prediction</h2><p>Original forecast, saved at issue time</p></div><span class="dot" id="latest-dot"></span></div><div id="latest-empty" class="latest-empty"><span class="empty-symbol">↗</span><strong>Ready for the first forecast</strong><p>The hourly worker collects market data and records predictions automatically.</p></div><div id="latest-content" hidden><div class="prediction-direction" id="direction">—</div><div class="probability-heading"><span>Probability of UP</span><b id="probability">—</b></div><div class="probability-track"><div id="probability-fill"></div></div><dl class="prediction-facts"><div><dt>Reference close</dt><dd id="reference-price">—</dd></div><div><dt>Issued at</dt><dd id="issued-at">—</dd></div><div><dt>Outcome due</dt><dd id="target-at">—</dd></div><div><dt>Status</dt><dd id="latest-status">Pending</dd></div></dl></div><div class="model-version"><label for="version-select">Model version</label><select id="version-select"><option value="">Current deployment</option></select></div></article>

app/static/monitor/monitoring-events.js

@@ -0,0 +1,45 @@
+(() => {
+  'use strict';
+  const $ = id => document.getElementById(id);
+  const stamp = value => value && Number.isFinite(Date.parse(value)) ? new Date(value).toISOString().slice(0,16).replace('T',' ') + ' UTC' : 'Not yet available';
+  let busy = false;
+  function text(id, value) { $(id).textContent = value; }
+  async function refresh() {
+    if (busy) return;
+    busy = true;
+    try {
+      const response = await fetch('/monitoring/activity', {cache:'no-store', signal:AbortSignal.timeout(15000)});
+      if (!response.ok) throw new Error('Monitoring API unavailable');
+      const data = await response.json(), event = data.last_research, operation = data.research || {};
+      text('scan-worker', data.worker_alive ? 'ACTIVE' : 'STOPPED');
+      text('scan-last', stamp(data.last_market_scan));
+      text('scan-next', stamp(data.next_cycle_due));
+      text('scan-success', stamp(data.last_successful_prediction_cycle));
+      text('scan-backup', (data.backup?.status || 'UNKNOWN') + (data.backup?.saved_at ? ' · ' + stamp(data.backup.saved_at) : ''));
+      text('scan-cadence', `Every ${data.monitor_interval_minutes} minutes; runs while this page is closed.`);
+      text('research-state', operation.status || 'IDLE');
+      text('research-last-time', stamp(event?.result?.retrieved_at || event?.completed_at));
+      const trigger = data.last_research_trigger;
+      let reason = trigger?.triggered_by?.map(value => value.replaceAll('_',' ')).join(', ') || 'No meaningful change detected';
+      if (trigger?.details?.price_change_1h != null) reason += ` (${trigger.details.price_change_1h.toFixed(2)}% in 1h)`;
+      text('research-trigger-reason', reason);
+      text('research-event-note', operation.cache_reason || (event?.status === 'error' ? 'Last context research was unavailable. Saved market forecasts are preserved.' : 'Web research is event-triggered to reduce unnecessary API calls.'));
+      text('stored-market-context', event?.summary || 'A contextual summary appears after an event produces useful validated evidence.');
+      const agents = [['Market monitor','monitor'],['Forecast agent','forecast'],['Trigger engine','trigger'],['Research agent',null],['Validation agent','validation'],['Decision agent','decision']];
+      const fragment = document.createDocumentFragment();
+      for (const [label,key] of agents) {
+        const node = document.createElement('span');
+        node.textContent = label + ': ' + (key ? operation[key] || 'idle' : operation.status || 'IDLE');
+        fragment.append(node);
+      }
+      $('continuous-agent-status').replaceChildren(fragment);
+      text('scan-error', [data.last_error, data.backup?.error].filter(Boolean).join(' '));
+    } catch {
+      text('scan-worker','UNAVAILABLE');
+      text('research-state','UNAVAILABLE');
+      text('scan-error','Cannot confirm current monitoring health; displayed timestamps may be stale.');
+    } finally { busy = false; }
+  }
+  window.addEventListener('focus', refresh);
+  refresh(); setInterval(refresh, 30000);
+})();

requirements-agents.txt

@@ -1,2 +1,3 @@
 # Official framework used by the conversational Decision Agent.
 openai-agents==0.22.1
+huggingface-hub>=1.0,<2

scripts/check_space_bundle.py

@@ -11,6 +11,10 @@ def check_bundle(root):
         "app/agents/validation_agent.py", "app/agents/answer_guard.py",
         "app/services/llm_service.py", "app/services/agent_config.py",
         "app/services/tavily_service.py", "app/schemas/agent_models.py",
+        "app/services/research_triggers.py", "app/services/market_monitoring.py",
+        "app/services/monitoring_events_store.py", "app/services/monitoring_backup.py",
+        "app/services/monitor_schedule.py", "app/services/monitor_worker_runtime.py",
+        "app/static/monitor/monitoring-events.js",
         "app/static/monitor/index.html", "app/static/monitor/analyst.js",
         "app/static/monitor/analyst.css", "app/static/monitor/dashboard.js",
         "app/static/monitor/styles.css", "requirements-agents.txt",

tests/test_event_monitoring.py

@@ -0,0 +1,282 @@
+"""Scheduled scans and research gating with real SQLite and offline providers."""
+import copy
+import tempfile
+import unittest
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import pandas as pd
+from fastapi.testclient import TestClient
+from app import api
+from app.monitoring_service import MonitoringService
+from app.monitor_worker import MonitoringWorker
+from app.services.market_monitoring import MarketMonitoringAgent
+from app.services.monitoring_events_store import EventStore
+from app.services.research_triggers import ResearchTriggerEngine, market_metrics
+from app.services.monitor_schedule import next_cycle
+from app.schemas.agent_models import ResearchResult, ChatRequest, DecisionPlan
+from app.agents.market_research_agent import MarketResearchAgent
+from app.agents.orchestrator import AgentOrchestrator, needs_research
+from test_analyst_integration import settings, saved_snapshot, article, ResearchStub, PlanStub
+
+ROOT = Path(__file__).resolve().parents[1]
+UTC = timezone.utc
+
+
+class EventMonitoringTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.service = MonitoringService(ROOT, Path(self.temp.name) / "events.sqlite3")
+        self.service.settings = settings()
+        self.raw = saved_snapshot(p=.1)
+        one = self.raw["forecasts"]["1"]
+        one["current_price"] = 100.
+        self.raw["forecasts"] = {str(h): {**copy.deepcopy(one), "horizon_hours": h} for h in (1, 6, 24)}
+        self.raw["market"].update(price=100., regime="RANGE")
+        self.origin = datetime.fromisoformat(one["latest_candle"])
+        self.market = pd.DataFrame({"timestamp": pd.date_range(end=self.origin, periods=500, freq="h"), "close": [100.] * 500})
+        self.research = ResearchStub(ResearchResult(status="ok", evidence=[article()]))
+        selected = DecisionPlan(focus_horizons=["1h"], fact_ids=[], evidence_ids=["bull"], interpretation="bullish_context", answer_style="concise",
+                                answer="The report describes ETF inflows. {{source:bull}}")
+        self.llm = PlanStub(selected)
+        self.monitor = MarketMonitoringAgent(self.service, self.research, self.llm)
+        self.service._market_monitor = self.monitor
+        self.pipeline = Mock()
+        self.pipeline.market_agent.fetch_closed_candles.side_effect = lambda **_: self.market.copy()
+        self.pipeline.refresh.side_effect = lambda **_: copy.deepcopy(self.raw)
+        self.service._pipeline = self.pipeline
+        self.old_service = api._service
+        api._service = self.service
+        self.client = TestClient(api.app)
+        self.seed_previous()
+
+    def tearDown(self):
+        api._service = self.old_service
+        self.temp.cleanup()
+
+    def seed_previous(self, transform=None):
+        previous = self.monitor.prepare(self.raw, self.market)
+        previous["market_timestamp"] = (self.origin - timedelta(hours=1)).isoformat()
+        if transform:
+            transform(previous)
+        # Replace only this fixture's baseline scan; never production history.
+        with self.service.store._connection() as db:
+            db.execute("DELETE FROM monitoring_scans")
+        self.service.event_store.scan(-1, previous, {"should_research": False, "triggered_by": []})
+
+    def price_move(self):
+        self.market.loc[self.market.index[-1], "close"] = 102.
+        self.raw["market"]["price"] = 102.
+        for forecast in self.raw["forecasts"].values():
+            forecast["current_price"] = 102.
+
+    def test_ordinary_cycle_records_scan_without_paid_calls(self):
+        result = self.service.run()
+        self.assertEqual(result["status"], "success")
+        self.assertFalse(result["research_trigger"]["should_research"])
+        self.assertEqual(self.research.calls, [])
+        self.assertEqual(self.llm.calls, [])
+        self.assertEqual(len(self.service.event_store.scans()), 2)
+        self.assertTrue(self.service.monitoring_activity()["last_cycle_completed"])
+
+    def test_large_price_move_calls_research_and_saves_context_reference(self):
+        self.price_move()
+        result = self.service.run()
+        self.assertIn("large_price_move", result["research_trigger"]["triggered_by"])
+        self.assertEqual(len(self.research.calls), 1)
+        self.assertEqual(len(self.llm.calls), 1)
+        event = self.service.event_store.latest_event()
+        self.assertEqual(event["status"], "done")
+        self.assertIn("ETF inflows", event["summary"])
+        scan = self.service.event_store.scans(1)[0]
+        self.assertEqual(scan["research_event_id"], event["id"])
+        self.assertNotIn("evidence", scan["payload"])
+
+    def test_regime_transition_triggers_research(self):
+        self.raw["market"]["regime"] = "HIGH_VOL"
+        self.assertIn("regime_change", self.service.run()["research_trigger"]["triggered_by"])
+        self.assertEqual(len(self.research.calls), 1)
+
+    def test_signed_internal_sentiment_shift_triggers_research(self):
+        self.raw["sentiment"]["latest_sentiment_score"] = .4
+        self.seed_previous()
+        self.raw["sentiment"]["latest_sentiment_score"] = -.4
+        self.assertIn("sentiment_shift", self.service.run()["research_trigger"]["triggered_by"])
+        self.assertEqual(len(self.research.calls), 1)
+
+    def test_small_probability_change_does_not_trigger(self):
+        for forecast in self.raw["forecasts"].values():
+            forecast["probability_up"] = .51
+        self.seed_previous()
+        for forecast in self.raw["forecasts"].values():
+            forecast["probability_up"] = .52
+        self.assertFalse(self.service.run()["research_trigger"]["should_research"])
+        self.assertEqual(self.research.calls, [])
+
+    def test_reliability_crossing_no_signal_to_down_triggers(self):
+        for forecast in self.raw["forecasts"].values():
+            forecast["validated_reliability"] = .4
+        self.seed_previous()
+        for forecast in self.raw["forecasts"].values():
+            forecast["validated_reliability"] = .8
+        self.assertIn("forecast_state_change", self.service.run()["research_trigger"]["triggered_by"])
+        self.assertEqual(len(self.research.calls), 1)
+
+    def test_same_candle_never_researches_again_even_after_restart(self):
+        self.price_move()
+        self.service.run()
+        self.service._market_monitor = MarketMonitoringAgent(self.service, self.research, self.llm)
+        result = self.service.run()
+        self.assertEqual(result["status"], "up_to_date")
+        self.assertEqual(len(self.research.calls), 1)
+        self.assertEqual(len(self.llm.calls), 1)
+
+    def test_persistent_cooldown_reuses_event_without_research_or_decision(self):
+        self.price_move()
+        self.service.run()
+        current = self.monitor.prepare(self.service.snapshot(), self.market)
+        current["market_timestamp"] = (self.origin + timedelta(hours=1)).isoformat()
+        reopened = MarketMonitoringAgent(self.service, self.research, self.llm)
+        with patch.object(reopened, "prepare", return_value=current):
+            result = reopened.process(999, self.service.snapshot(), self.market)
+        self.assertTrue(result["suppressed"])
+        self.assertEqual(len(self.research.calls), 1)
+        self.assertEqual(len(self.llm.calls), 1)
+        self.assertEqual(self.service.monitoring_activity()["research"]["status"], "CACHED")
+
+    def test_tavily_failure_does_not_fail_market_cycle(self):
+        self.price_move()
+        self.research.result = ResearchResult(status="unavailable")
+        result = self.service.run()
+        self.assertEqual(result["status"], "success")
+        self.assertEqual(result["recorded"]["inserted"], 3)
+        self.assertEqual(self.llm.calls, [])
+        self.assertIsNone(self.service.store.get_state("last_error"))
+        self.assertEqual(self.service.monitoring_activity()["research"]["status"], "ERROR")
+
+    def test_openai_failure_preserves_forecasts_and_scan(self):
+        self.price_move()
+        self.llm.value = None
+        result = self.service.run()
+        self.assertEqual(result["status"], "success")
+        self.assertEqual(result["recorded"]["inserted"], 3)
+        self.assertEqual(self.service.event_store.latest_event()["status"], "error")
+        self.assertIsNone(self.service.event_store.latest_event()["summary"])
+
+    def test_bearish_news_cannot_promote_unreliable_model(self):
+        self.price_move()
+        for forecast in self.raw["forecasts"].values():
+            forecast["validated_reliability"] = .42
+        self.research.result = ResearchResult(status="ok", evidence=[article("bear", headline="Bitcoin ETF outflows rise", direction="bearish")])
+        self.service.run()
+        for view in self.service.event_store.scans(1)[0]["payload"]["forecasts"].values():
+            self.assertEqual(view["signal_state"], "NO_SIGNAL")
+            self.assertEqual(view["reliability"], .42)
+        self.assertEqual(self.service.summary()["qualification"]["no_signal_frequency"], 1.)
+
+    def test_no_useful_validated_evidence_means_no_decision_call(self):
+        self.price_move()
+        self.research.result = ResearchResult(status="ok", evidence=[article(published_at=None)])
+        self.service.run()
+        self.assertEqual(self.llm.calls, [])
+
+    def test_forward_outcome_resolution_keeps_original_prediction_and_gate(self):
+        self.raw["forecasts"]["1"]["validated_reliability"] = .8
+        self.service.run()
+        original = self.service.store.history(horizon=1)["items"][0]
+        now = self.origin + timedelta(hours=2, minutes=1)
+        self.service.store.settle([{"timestamp": self.origin + timedelta(hours=1), "close": 90}], now=now)
+        row = self.service.store.history(horizon=1, now=now)["items"][0]
+        self.assertEqual(row["probability_up"], original["probability_up"])
+        self.assertEqual(row["issued_at"], original["issued_at"])
+        self.assertEqual(row["correct"], 1)
+        metric = self.service.store.summary(horizon=1, now=now)
+        self.assertEqual(metric["qualification"]["qualified_performance"]["accuracy"], 1.)
+        self.assertAlmostEqual(metric["brier_score"], .01)
+
+    def test_two_workers_and_restart_execute_without_browser_or_duplicates(self):
+        client = self.client
+        class Transport:
+            def post(self, url, json, timeout):
+                return client.post(url.replace("http://testserver", ""), json=json)
+        first = MonitoringWorker("http://testserver", session=Transport())
+        second = MonitoringWorker("http://testserver", session=Transport())
+        self.assertEqual(first.step(), "success")
+        self.assertEqual(second.step(), "standby")
+        self.assertEqual(first.step(), "waiting")
+        self.service.event_store.release("scheduler", first.worker_id)
+        self.assertEqual(second.step(), "waiting")
+        self.assertEqual(first.step(), "standby")
+        self.assertEqual(self.pipeline.refresh.call_count, 1)
+        self.assertEqual(self.research.calls, [])
+
+    def test_explicit_fresh_why_research_bypasses_monitoring_cooldown(self):
+        self.price_move()
+        self.service.run()
+        analyst = AgentOrchestrator(self.service, self.service.settings, self.research, self.llm)
+        analyst.chat(ChatRequest(message="Why is the model predicting DOWN?", fresh_research=True))
+        self.assertEqual(len(self.research.calls), 2)
+        self.assertTrue(self.research.calls[-1][1]["fresh"])
+
+
+class TriggerUnitTests(unittest.TestCase):
+    def test_volatility_trigger_and_missing_baseline(self):
+        engine = ResearchTriggerEngine(settings())
+        point = {"data_fresh": True, "market_timestamp": "new", "volatility": .03, "volatility_baseline": .01}
+        self.assertIn("unusual_volatility", engine.evaluate(point)["triggered_by"])
+        point["volatility_baseline"] = None
+        self.assertFalse(engine.evaluate(point)["should_research"])
+
+    def test_stale_context_never_triggers_paid_research(self):
+        point = {"data_fresh": False, "market_timestamp": "new", "price_change_1h": 20}
+        self.assertFalse(ResearchTriggerEngine(settings()).evaluate(point)["should_research"])
+
+    def test_price_threshold_is_configurable(self):
+        point = {"data_fresh": True, "market_timestamp": "new", "price_change_1h": 1.5}
+        self.assertFalse(ResearchTriggerEngine(settings(price_move_trigger_pct=2)).evaluate(point)["should_research"])
+
+    def test_query_plans_are_bounded_relevant_and_direction_neutral(self):
+        for trigger, keyword in (("large_price_move", "liquidations"), ("unusual_volatility", "liquidations"), ("sentiment_shift", "regulation")):
+            queries = MarketResearchAgent.query_plan("model DOWN", trigger_types=[trigger])
+            self.assertEqual(len(queries), 3)
+            self.assertIn(keyword, " ".join(queries))
+            self.assertNotIn("will fall", " ".join(queries))
+
+    def test_requested_market_explanations_route_to_research(self):
+        for question in ("Why is BTC moving?", "Why is the model predicting DOWN?", "What happened in the market?", "What news is affecting BTC?", "Explain the latest move."):
+            self.assertTrue(needs_research(question), question)
+
+    def test_configurable_cadence_and_midnight_rollover(self):
+        now = datetime(2026, 9, 8, 23, 59, tzinfo=UTC)
+        self.assertEqual(next_cycle(now, 60), datetime(2026, 9, 9, 0, 2, tzinfo=UTC))
+        self.assertEqual(next_cycle(now.replace(hour=9, minute=5), 15), now.replace(hour=9, minute=17))
+
+    def test_metrics_do_not_invent_returns_across_missing_candles(self):
+        times = pd.date_range("2026-01-01", periods=26, freq="h", tz="UTC")
+        frame = pd.DataFrame({"timestamp": times, "close": [100.] * 26}).drop(index=24)
+        point = market_metrics(frame, {})
+        self.assertIsNone(point["price_change_1h"])
+        self.assertIsNone(point["volatility"])
+
+
+class DurableQueryCacheTests(unittest.TestCase):
+    def test_restart_reuses_query_cache_but_explicit_fresh_bypasses_it(self):
+        with tempfile.TemporaryDirectory() as directory:
+            service = MonitoringService(ROOT, Path(directory) / "cache.sqlite3")
+            tavily = Mock()
+            tavily.search.return_value = {"status": "empty", "results": []}
+            config = settings()
+            first = MarketResearchAgent(config, tavily, service.event_store)
+            first.run("Bitcoin latest market news")
+            self.assertEqual(tavily.search.call_count, 3)
+            second = MarketResearchAgent(config, tavily, EventStore(service.store))
+            self.assertTrue(second.run("Bitcoin latest market news").cached)
+            self.assertEqual(tavily.search.call_count, 3)
+            second.run("Bitcoin latest market news", fresh=True)
+            self.assertEqual(tavily.search.call_count, 6)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_monitoring_persistence.py

@@ -0,0 +1,131 @@
+import shutil
+import sqlite3
+import tempfile
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from app.monitoring_store import MonitoringStore
+from app.monitoring_service import MonitoringService
+from app.services.monitoring_backup import MonitoringBackup
+from app.services.monitoring_events_store import EventStore
+
+ROOT = Path(__file__).resolve().parents[1]
+UTC = timezone.utc
+
+
+class PersistenceTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.path = Path(self.temp.name)
+        self.store = MonitoringStore(self.path / 'live.sqlite3')
+        self.events = EventStore(self.store)
+        self.store.set_state('preserve', {'probability_up': .4123})
+        self.client = Mock()
+        self.client.repo_info.return_value = SimpleNamespace(private=True, sha='revision-a')
+        self.client.upload_file.return_value = SimpleNamespace(oid='revision-b')
+        self.backup = MonitoringBackup('owner/private', self.client)
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def test_consistent_snapshot_restore_preserves_data_and_clears_ownership(self):
+        self.store.try_acquire_lock('old')
+        self.events.acquire('scheduler', 'old')
+        self.store.set_state('worker', {'worker_id': 'old'})
+        self.events.claim('fingerprint', 'candle', ['price'], 30)
+        archive = self.path / 'archive.sqlite3'
+        def upload(**kwargs):
+            self.assertEqual(kwargs['parent_commit'], 'revision-a')
+            shutil.copyfile(kwargs['path_or_fileobj'], archive)
+            return SimpleNamespace(oid='revision-b')
+        self.client.upload_file.side_effect = upload
+        self.backup.save(self.store)
+        self.assertEqual(self.store.get_state('backup')['status'], 'OK')
+        restored = self.path / 'restored.sqlite3'
+        restore = MonitoringBackup('owner/private', self.client, lambda **_: str(archive))
+        self.assertTrue(restore.restore(restored))
+        db = MonitoringStore(restored)
+        self.assertEqual(db.get_state('preserve'), {'probability_up': .4123})
+        self.assertIsNone(db.get_state('worker'))
+        self.assertTrue(EventStore(db).acquire('scheduler', 'new'))
+        self.assertEqual(EventStore(db).latest_event()['status'], 'error')
+        self.assertTrue(db.try_acquire_lock('new'))
+        with sqlite3.connect(restored) as connection:
+            self.assertEqual(connection.execute('PRAGMA integrity_check').fetchone()[0], 'ok')
+
+    def test_failed_upload_preserves_local_history_and_reports_error(self):
+        self.client.upload_file.side_effect = RuntimeError('credential should not be exposed')
+        self.backup.save(self.store)
+        self.assertEqual(self.store.get_state('preserve')['probability_up'], .4123)
+        status = self.store.get_state('backup')
+        self.assertEqual(status['status'], 'ERROR')
+        self.assertNotIn('credential', str(status))
+
+    def test_public_repository_is_never_uploaded_to(self):
+        self.client.repo_info.return_value.private = False
+        self.backup.save(self.store)
+        self.client.upload_file.assert_not_called()
+        self.assertEqual(self.store.get_state('backup')['status'], 'ERROR')
+
+    def test_corrupt_restore_fails_closed(self):
+        corrupt = self.path / 'corrupt'
+        corrupt.write_text('not a database')
+        target = self.path / 'new.sqlite3'
+        backup = MonitoringBackup('owner/private', self.client, lambda **_: str(corrupt))
+        with self.assertRaisesRegex(RuntimeError, 'refusing to start'):
+            backup.restore(target)
+        self.assertFalse(target.exists())
+
+    def test_existing_local_history_is_not_replaced(self):
+        self.assertFalse(self.backup.restore(self.store.path))
+        self.client.repo_info.assert_not_called()
+
+    def test_disabled_backup_uses_no_network(self):
+        backup = MonitoringBackup('', self.client)
+        self.assertFalse(backup.restore(self.path / 'absent'))
+        backup.save(self.store)
+        self.client.repo_info.assert_not_called()
+
+    def test_concurrent_workers_only_one_owns_scheduler(self):
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            results = list(pool.map(lambda n: self.events.acquire('scheduler', str(n)), range(8)))
+        self.assertEqual(sum(results), 1)
+
+    def test_query_cache_expires_without_extending_on_reads(self):
+        now = datetime.now(UTC)
+        self.events.cache_put('query', {'results': []}, 600, now)
+        self.assertIsNotNone(self.events.cache_get('query', now + timedelta(minutes=9)))
+        self.assertIsNone(self.events.cache_get('query', now + timedelta(minutes=10)))
+
+    def test_fingerprint_cooldown_survives_new_store_and_expires(self):
+        now = datetime.now(UTC)
+        _, first = self.events.claim('same', 'candle-a', [], 30, now)
+        restarted = EventStore(MonitoringStore(self.store.path))
+        _, recent = restarted.claim('same', 'candle-b', [], 30, now + timedelta(minutes=20))
+        _, expired = restarted.claim('same', 'candle-b', [], 30, now + timedelta(minutes=31))
+        self.assertEqual((first, recent, expired), (True, False, True))
+
+    def test_finished_failure_keeps_retry_due_and_does_not_claim_success(self):
+        service = MonitoringService(ROOT, self.path / 'service.sqlite3')
+        now = datetime.now(UTC)
+        started = now - timedelta(minutes=4)
+        due = (now + timedelta(minutes=3)).isoformat()
+        service.store.set_state('last_cycle_started', started.isoformat())
+        service.store.set_state('last_cycle_finished', (started + timedelta(seconds=1)).isoformat())
+        service.store.set_state('next_cycle_due', due)
+        status = service.heartbeat(worker_id='new')
+        self.assertEqual(status['next_cycle_due'], due)
+        self.assertIsNone(status['last_cycle_completed'])
+
+    def test_interrupted_cycle_is_rescheduled_after_lease_expires(self):
+        service = MonitoringService(ROOT, self.path / 'service.sqlite3')
+        now = datetime.now(UTC)
+        service.store.set_state('last_cycle_started', (now - timedelta(minutes=4)).isoformat())
+        service.store.set_state('next_cycle_due', (now + timedelta(minutes=30)).isoformat())
+        status = service.heartbeat(worker_id='new')
+        self.assertLess(datetime.fromisoformat(status['next_cycle_due']), now + timedelta(seconds=10))
+        self.assertIsNone(status['last_cycle_completed'])