signal_log: outcomes for every setup, event dedup, interned news; fix over-scan, dead outlook, panel OPPOSE bias

- seed an advisory outcome on every setup row; same-setup re-logs within 60 min
  are DUPLICATE rows (advisory_event_id); resolver rebuilds bar streams across gaps
- headlines/events interned into news_snapshots (DB 667 MB -> 262 MB); readers
  use signals_with_news / read_news and mode=ro (immutable=1 missed WAL rows)
- post-gate truth columns: gate_reason, actionable, cleaned_warnings, scan_kind,
  config_fingerprint; SELL-hour gates fail closed
- telegram: breach rescan reuses this bar's report; grade dropped from alert
  fingerprint (upgrade-only re-alert)
- market_outlook: hour-bucket dedup (old rule skipped alternate hours), skip heartbeats
- agent_panel: macro-anchored OPPOSE guard (opposed trades were +30R over 3 weeks)
- analyze_signal_log --stored-candles/--parity; backfill + news migration scripts
- tests: Module 25 (analyze_signal_log) + 27 (observability); 429 total

Claude-Session: https://claude.ai/code/session_01FC4s6VCURFte4dub4xKxYL

signal_log.py

@@ -27,16 +27,47 @@ Fail-safe contract: logging must NEVER break a scan. Every public entry
 point catches all exceptions and downgrades them to a warning log line.
 Report/context objects are duck-typed (no imports from signal_aggregator
 or news_engine) so this module stays at the bottom of the import graph.
+
+2026-09 observability additions (signal-log audit):
+
+* Every row with a trade setup is an advisory-outcome candidate, not only
+  panel rows. Re-logs of the same setup (same ``setup_key`` — tf,
+  direction, entry, stop, strategy — within OUTCOME_DEDUP_GAP_SECONDS,
+  from ANY surface) are marked ``DUPLICATE`` and point at the canonical
+  row through ``advisory_event_id``; only the canonical row is seeded and
+  advanced, so summing ``advisory_r_multiple`` over
+  ``advisory_outcome_status='RESOLVED'`` counts each trade once.
+* Pending outcomes advance from the inserting scan's own snapshot first;
+  when that snapshot no longer overlaps the outcome's last processed bar
+  (downtime, weekend straddle) the resolver rebuilds the bar stream from
+  every stored snapshot of that timeframe, so a >14-bar gap in one
+  process's scans is not a terminal ``UNRESOLVABLE``.
+* News blobs are content-addressed: ``headlines_json``/``events_json`` are
+  interned once into ``news_snapshots`` and rows carry ``headlines_hash``
+  / ``events_hash`` (the legacy columns stay NULL on new rows; the
+  ``signals_with_news`` view and ``read_news`` resolve both eras).
+* Post-gate truth columns: ``gate_reason`` (the warning that suppressed
+  the setup), ``actionable`` (grade >= C AND setup present — what an
+  alert surface would act on; the ``action`` text is written BEFORE the
+  gate chain and can disagree), ``cleaned_warnings_json`` (warnings the
+  XAU_CP acceptance removed from ``warnings_json``), ``bar_range_median``
+  (the value historically stored as ``spread`` — it is the median
+  High-Low of the primary frame, NOT a broker spread), ``scan_kind``
+  (``scan`` / ``blocked`` for weekend/day early returns, which carry no
+  news or candles) and ``config_fingerprint`` (hash of the gate config the
+  producing process ran with — surfaces config drift between processes).
 """
 
+import hashlib
 import json
 import logging
 import math
 import os
 import sqlite3
 import threading
+import time
 from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterable, List, Optional, Tuple
 
 import app_config
 from advisory_outcome import (
@@ -90,6 +121,13 @@ CREATE TABLE IF NOT EXISTS signals (
 );
 CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts
     ON signals (symbol, timeframe, logged_at);
+CREATE TABLE IF NOT EXISTS news_snapshots (
+    hash TEXT PRIMARY KEY,
+    kind TEXT NOT NULL,
+    body TEXT NOT NULL,
+    first_seen TEXT NOT NULL,
+    last_seen TEXT
+);
 CREATE TABLE IF NOT EXISTS outlooks (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     logged_at TEXT NOT NULL,
@@ -120,25 +158,72 @@ _ADDITIVE_SIGNAL_COLS = (
     ("advisory_outcome_status", "TEXT"),
     ("advisory_r_multiple", "REAL"),
     ("advisory_outcome_json", "TEXT"),
+    # 2026-09 observability columns (see module docstring).
+    ("setup_key", "TEXT"),
+    ("advisory_event_id", "INTEGER"),
+    ("headlines_hash", "TEXT"),
+    ("events_hash", "TEXT"),
+    ("gate_reason", "TEXT"),
+    ("actionable", "INTEGER"),
+    ("cleaned_warnings_json", "TEXT"),
+    ("bar_range_median", "REAL"),
+    ("scan_kind", "TEXT"),
+    ("config_fingerprint", "TEXT"),
 )
 
+# Read-side compatibility view `signals_with_news`: legacy rows keep
+# their inline news blobs, new rows resolve them through news_snapshots.
+# Recreated on every logger init (SignalLogger._ensure_news_view) so it
+# always exposes the current column set.
+
+# Same-setup re-logs (cross-surface double logs, persisting-signal
+# re-fires) within this gap are ONE advisory event. Mirrors
+# analyze_signal_log's 60-minute clustering.
+OUTCOME_DEDUP_GAP_SECONDS = 3600
+
+# Warning prefixes that mean "a gate removed the setup", checked in gate
+# order (live wrapper first, then the post-grade chain, then the setup
+# generators). Kept here — signal_log must not import signal_aggregator.
+_GATE_PREFIXES = (
+    "WEEKEND FILTER",
+    "DAY FILTER",
+    "SESSION FILTER",
+    "Primary timeframe stale",
+    "Primary timeframe data unavailable",
+    "REGIME FILTER",
+    "EXHAUSTION BLOCK",
+    "EXHAUSTED TREND FILTER",
+    "SELL FILTER",
+    "MTF OPPOSITION",
+    "BIG CANDLE FILTER",
+    "M5 FILTER",
+    "LOW ATR FILTER",
+    "SPREAD GATE",
+)
+_ACTIONABLE_GRADES = ("A+", "A", "B", "C")
+
 _ADDITIVE_OUTLOOK_COLS = (
     ("news_json", "TEXT"),
     ("indicators_json", "TEXT"),
     ("candles_json", "TEXT"),
 )
 
-_INSERT = """
-INSERT INTO signals (
-    logged_at, source, symbol, timeframe, signal, score, confidence,
-    quality, action, price, spread, market_regime, volatility_regime,
-    mtf_json, setup_json, reasons_json, warnings_json,
-    event_risk, event_description, news_sentiment, news_description,
-    news_summary, events_json, headlines_json,
-    indicators_json, candles_json,
-    panel_verdict, panel_confidence, panel_rationale, panel_takes_json
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-"""
+_SIGNAL_INSERT_COLS = (
+    "logged_at", "source", "symbol", "timeframe", "signal", "score",
+    "confidence", "quality", "action", "price", "spread", "market_regime",
+    "volatility_regime", "mtf_json", "setup_json", "reasons_json",
+    "warnings_json", "event_risk", "event_description", "news_sentiment",
+    "news_description", "news_summary", "events_json", "headlines_json",
+    "indicators_json", "candles_json", "panel_verdict", "panel_confidence",
+    "panel_rationale", "panel_takes_json",
+    "setup_key", "headlines_hash", "events_hash", "gate_reason",
+    "actionable", "cleaned_warnings_json", "bar_range_median", "scan_kind",
+    "config_fingerprint",
+)
+_INSERT = (
+    "INSERT INTO signals (" + ", ".join(_SIGNAL_INSERT_COLS) + ") VALUES ("
+    + ", ".join("?" for _ in _SIGNAL_INSERT_COLS) + ")"
+)
 
 _OUTLOOK_INSERT = """
 INSERT INTO outlooks (
@@ -173,6 +258,57 @@ def _dumps(obj: Any) -> str:
     return json.dumps(obj, default=str, allow_nan=False)
 
 
+def _blob_hash(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def setup_key(setup: Optional[Dict[str, Any]], timeframe: Any) -> Optional[str]:
+    """Identity of a logged trade setup for event dedup.
+
+    ``"<tf>|<direction>|<entry>|<stop>|<strategy>"`` — the same tuple
+    analyze_signal_log clusters on (tf, direction, round(entry, 2),
+    round(SL, 2), strategy). None when the setup lacks a usable price."""
+    if not isinstance(setup, dict):
+        return None
+    entry = _finite(setup.get("entry"))
+    stop = _finite(setup.get("stop_loss"))
+    if entry is None or stop is None:
+        return None
+    return "|".join((
+        str(getattr(timeframe, "value", timeframe) or ""),
+        str(setup.get("direction") or ""),
+        f"{entry:.2f}",
+        f"{stop:.2f}",
+        str(setup.get("strategy") or ""),
+    ))
+
+
+def gate_reason(
+    warnings: Iterable[str],
+    cleaned_warnings: Iterable[str],
+    quality: str,
+    has_setup: bool,
+) -> Optional[str]:
+    """The warning that explains why a row carries no setup, or None.
+
+    Rows WITH a setup have no gate reason. Rows without one report the
+    first gate-prefixed warning (in gate order); an actionable grade with
+    no setup and no gate warning is the setup generator returning None
+    (RR floor / geometry), which the aggregator does not warn about."""
+    if has_setup:
+        return None
+    pool = list(warnings or []) + list(cleaned_warnings or [])
+    for prefix in _GATE_PREFIXES:
+        for warning in pool:
+            if str(warning).startswith(prefix):
+                return str(warning)
+    if quality in _ACTIONABLE_GRADES:
+        return "SETUP GENERATOR: no setup (RR floor / geometry)"
+    if quality:
+        return f"GRADE {quality}: below setup threshold"
+    return None
+
+
 # ── Serializers (duck-typed; deliberately NOT imported from            ──
 # ── dashboard_publisher — the dashboard's hard independence rule)      ──
 
@@ -322,7 +458,12 @@ def _report_row(
     source: str,
     timeframe: str,
     dataframes: Any = None,
-) -> tuple:
+    extra: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """Column-name → value mapping for one scan row (see
+    _SIGNAL_INSERT_COLS). ``extra`` carries producer-side metadata
+    (``scan_kind``, ``config_fingerprint``)."""
+    extra = extra or {}
     mtf_signals = getattr(report, "mtf_signals", None) or {}
     mtf_json = _dumps({
         "agreement": bool(getattr(report, "mtf_agreement", False)),
@@ -363,38 +504,56 @@ def _report_row(
     signal = getattr(report, "final_signal", None)
     panel_takes = getattr(report, "panel_takes", None) or []
     panel_takes_json = _dumps(panel_takes) if panel_takes else None
-    return (
-        datetime.now(timezone.utc).isoformat(),
-        source,
-        str(getattr(report, "symbol", "")),
-        timeframe,
-        getattr(signal, "name", str(signal)),
-        _finite(getattr(report, "final_score", None)),
-        _finite(getattr(report, "final_confidence", None)),
-        getattr(report, "signal_quality", "") or "",
-        getattr(report, "action_recommendation", "") or "",
-        _finite(getattr(report, "current_price", None)),
-        _finite(getattr(report, "spread_estimate", None)),
-        getattr(report, "market_regime", "") or "",
-        getattr(report, "volatility_regime", "") or "",
-        mtf_json,
-        setup_json,
-        reasons_json,
-        warnings_json,
-        getattr(report, "event_risk", "") or "",
-        getattr(report, "event_description", "") or "",
-        _finite(getattr(report, "news_sentiment", None)),
-        getattr(report, "news_description", "") or "",
-        getattr(report, "news_summary", "") or "",
-        events_json,
-        headlines_json,
-        _report_indicators_json(report),
-        _candles_json(dataframes),
-        getattr(report, "panel_verdict", None),
-        _finite(getattr(report, "panel_confidence", None)),
-        getattr(report, "panel_rationale", None),
-        panel_takes_json,
-    )
+    quality = getattr(report, "signal_quality", "") or ""
+    warnings = list(getattr(report, "warnings", None) or [])
+    cleaned = list(getattr(report, "cleaned_warnings", None) or [])
+    setup_dict = json.loads(setup_json) if setup_json else None
+    bar_range = _finite(getattr(report, "spread_estimate", None))
+    return {
+        "logged_at": datetime.now(timezone.utc).isoformat(),
+        "source": source,
+        "symbol": str(getattr(report, "symbol", "")),
+        "timeframe": timeframe,
+        "signal": getattr(signal, "name", str(signal)),
+        "score": _finite(getattr(report, "final_score", None)),
+        "confidence": _finite(getattr(report, "final_confidence", None)),
+        "quality": quality,
+        "action": getattr(report, "action_recommendation", "") or "",
+        "price": _finite(getattr(report, "current_price", None)),
+        "spread": bar_range,
+        "market_regime": getattr(report, "market_regime", "") or "",
+        "volatility_regime": getattr(report, "volatility_regime", "") or "",
+        "mtf_json": mtf_json,
+        "setup_json": setup_json,
+        "reasons_json": reasons_json,
+        "warnings_json": warnings_json,
+        "event_risk": getattr(report, "event_risk", "") or "",
+        "event_description": getattr(report, "event_description", "") or "",
+        "news_sentiment": _finite(getattr(report, "news_sentiment", None)),
+        "news_description": getattr(report, "news_description", "") or "",
+        "news_summary": getattr(report, "news_summary", "") or "",
+        # Interned by SignalLogger.log into news_snapshots; the inline
+        # columns stay NULL on new rows (legacy rows keep theirs).
+        "events_json": events_json,
+        "headlines_json": headlines_json,
+        "indicators_json": _report_indicators_json(report),
+        "candles_json": _candles_json(dataframes),
+        "panel_verdict": getattr(report, "panel_verdict", None),
+        "panel_confidence": _finite(getattr(report, "panel_confidence", None)),
+        "panel_rationale": getattr(report, "panel_rationale", None),
+        "panel_takes_json": panel_takes_json,
+        "setup_key": setup_key(setup_dict, timeframe),
+        "headlines_hash": None,
+        "events_hash": None,
+        "gate_reason": gate_reason(warnings, cleaned, quality,
+                                   setup_dict is not None),
+        "actionable": int(setup_dict is not None
+                          and quality in _ACTIONABLE_GRADES),
+        "cleaned_warnings_json": _dumps(cleaned) if cleaned else None,
+        "bar_range_median": bar_range,
+        "scan_kind": str(extra.get("scan_kind") or "scan"),
+        "config_fingerprint": extra.get("config_fingerprint"),
+    }
 
 
 class SignalLogger:
@@ -408,13 +567,53 @@ class SignalLogger:
         if parent:
             os.makedirs(parent, exist_ok=True)
         self._conn = sqlite3.connect(path, check_same_thread=False)
-        self._conn.execute("PRAGMA journal_mode=WAL")
+        # busy_timeout BEFORE the journal-mode switch: several processes
+        # (bot, dashboard, tests) open the same file at once and the WAL
+        # switch plus the additive migration need brief exclusive locks.
         self._conn.execute("PRAGMA busy_timeout=5000")
         self._lock = threading.Lock()
         with self._lock:
-            self._conn.executescript(_SCHEMA)
-            self._migrate_signal_columns()
-            self._migrate_outlook_columns()
+            self._with_lock_retry(
+                lambda: self._conn.execute("PRAGMA journal_mode=WAL"))
+            self._with_lock_retry(
+                lambda: self._conn.executescript(_SCHEMA))
+            self._with_lock_retry(self._migrate_signal_columns)
+            self._with_lock_retry(self._migrate_outlook_columns)
+            self._with_lock_retry(self._ensure_news_view)
+
+    def _ensure_news_view(self) -> None:
+        """(Re)create the read-side view atomically. Several processes
+        open the DB at once; DROP+CREATE inside one IMMEDIATE transaction
+        serialises them, and a view another opener created in between is
+        not an error."""
+        try:
+            self._conn.execute("BEGIN IMMEDIATE")
+            self._conn.execute("DROP VIEW IF EXISTS signals_with_news")
+            self._conn.execute(
+                "CREATE VIEW signals_with_news AS"
+                " SELECT s.*,"
+                " COALESCE(s.headlines_json, nh.body) AS headlines_resolved,"
+                " COALESCE(s.events_json, ne.body) AS events_resolved"
+                " FROM signals s"
+                " LEFT JOIN news_snapshots nh ON nh.hash = s.headlines_hash"
+                " LEFT JOIN news_snapshots ne ON ne.hash = s.events_hash")
+            self._conn.commit()
+        except sqlite3.OperationalError as exc:
+            self._conn.rollback()
+            if "already exists" not in str(exc).lower():
+                raise
+
+    @staticmethod
+    def _with_lock_retry(fn, attempts: int = 8, delay: float = 0.05):
+        """Run ``fn`` again on 'database is locked' (concurrent openers
+        racing the same schema step); other errors propagate at once."""
+        for attempt in range(attempts):
+            try:
+                return fn()
+            except sqlite3.OperationalError as exc:
+                if "locked" not in str(exc).lower() or attempt == attempts - 1:
+                    raise
+                time.sleep(delay * (attempt + 1))
 
     @property
     def path(self) -> str:
@@ -436,6 +635,12 @@ class SignalLogger:
         self._conn.execute(
             "CREATE INDEX IF NOT EXISTS idx_signals_advisory_pending"
             " ON signals (symbol, advisory_outcome_status, id)")
+        self._conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_signals_setup_key"
+            " ON signals (symbol, setup_key, id)")
+        self._conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_signals_event"
+            " ON signals (advisory_event_id)")
         self._conn.commit()
 
     def _migrate_outlook_columns(self) -> None:
@@ -460,20 +665,109 @@ class SignalLogger:
         source: str = "live",
         timeframe: str = "",
         dataframes: Any = None,
+        extra: Optional[Dict[str, Any]] = None,
     ) -> int:
-        """Insert one scan row; returns the rowid (panel-update key)."""
-        row = _report_row(report, context, source, timeframe, dataframes)
+        """Insert one scan row; returns the rowid (panel-update key).
+
+        One transaction interns the news blobs, inserts the row and seeds
+        (or dedups) its advisory outcome; a seeding failure is logged and
+        the row still commits. Pending outcomes are then advanced from
+        this scan's candle snapshot."""
+        row = _report_row(report, context, source, timeframe, dataframes,
+                          extra=extra)
+        now_iso = row["logged_at"]
         with self._lock:
-            cur = self._conn.execute(_INSERT, row)
-            self._conn.commit()
-        rowid = int(cur.lastrowid)
+            self._conn.execute("BEGIN IMMEDIATE")
+            try:
+                row["headlines_hash"] = self._intern_news(
+                    "headlines", row.pop("headlines_json"), now_iso)
+                row["events_hash"] = self._intern_news(
+                    "events", row.pop("events_json"), now_iso)
+                row["headlines_json"] = None
+                row["events_json"] = None
+                cur = self._conn.execute(
+                    _INSERT, tuple(row[c] for c in _SIGNAL_INSERT_COLS))
+                rowid = int(cur.lastrowid)
+                try:
+                    self._seed_outcome(rowid, row)
+                except Exception:
+                    logger.warning("signal_log: outcome seeding failed",
+                                   exc_info=True)
+                self._conn.commit()
+            except Exception:
+                self._conn.rollback()
+                raise
         try:
-            self.resolve_pending(str(row[2]), row[25])
+            self.resolve_pending(str(row["symbol"]), row["candles_json"])
         except Exception:
             logger.warning("signal_log: advisory outcome resolver failed",
                            exc_info=True)
         return rowid
 
+    # ── news interning ────────────────────────────────────────────
+
+    def _intern_news(self, kind: str, body: Optional[str],
+                     now_iso: str) -> Optional[str]:
+        """Store one news blob once; returns its content hash."""
+        if not body:
+            return None
+        digest = _blob_hash(body)
+        self._conn.execute(
+            "INSERT OR IGNORE INTO news_snapshots"
+            " (hash, kind, body, first_seen, last_seen)"
+            " VALUES (?, ?, ?, ?, ?)",
+            (digest, kind, body, now_iso, now_iso))
+        self._conn.execute(
+            "UPDATE news_snapshots SET last_seen=? WHERE hash=?",
+            (now_iso, digest))
+        return digest
+
+    def news_for(self, rowid: int) -> Tuple[Optional[str], Optional[str]]:
+        """(events_json, headlines_json) for a row, legacy or interned."""
+        with self._lock:
+            return read_news(self._conn, rowid)
+
+    # ── advisory-event seeding / dedup ────────────────────────────
+
+    def _seed_outcome(self, rowid: int, row: Dict[str, Any]) -> None:
+        """Seed a PENDING outcome for a new setup row, or mark it a
+        DUPLICATE of the most recent same-setup row within the gap.
+        Runs inside log()'s transaction."""
+        key = row.get("setup_key")
+        if not key or not row.get("setup_json"):
+            return
+        prior = self._conn.execute(
+            "SELECT id, advisory_event_id, logged_at FROM signals"
+            " WHERE symbol=? AND setup_key=? AND id<?"
+            " ORDER BY id DESC LIMIT 1",
+            (row["symbol"], key, rowid)).fetchone()
+        if prior is not None:
+            try:
+                gap = (datetime.fromisoformat(row["logged_at"])
+                       - datetime.fromisoformat(prior[2])).total_seconds()
+            except (TypeError, ValueError):
+                gap = None
+            if gap is not None and 0 <= gap <= OUTCOME_DEDUP_GAP_SECONDS:
+                event_id = int(prior[1] or prior[0])
+                self._conn.execute(
+                    "UPDATE signals SET advisory_event_id=?,"
+                    " advisory_outcome_status='DUPLICATE',"
+                    " advisory_outcome_json=? WHERE id=?",
+                    (event_id,
+                     _dumps({"model": "advisory-event-dedup-v1",
+                             "status": "DUPLICATE", "event_id": event_id}),
+                     rowid))
+                return
+        setup = json.loads(row["setup_json"])
+        indicators = (json.loads(row["indicators_json"])
+                      if row.get("indicators_json") else None)
+        status, state = initial_outcome(setup, indicators, row["timeframe"])
+        self._conn.execute(
+            "UPDATE signals SET advisory_event_id=?,"
+            " advisory_outcome_status=?, advisory_outcome_json=?"
+            " WHERE id=?",
+            (rowid, status, _dumps(state), rowid))
+
     def update_panel(
         self,
         rowid: int,
@@ -486,7 +780,8 @@ class SignalLogger:
         cache_status: Optional[str] = None,
         payload_hash: Optional[str] = None,
     ) -> None:
-        """Fill panel provenance and initialize its advisory outcome."""
+        """Fill panel provenance; seeds an outcome only for rows that
+        log() did not already seed (rows written by an older process)."""
         with self._lock:
             row = self._conn.execute(
                 "SELECT setup_json, indicators_json, timeframe,"
@@ -520,7 +815,15 @@ class SignalLogger:
     def resolve_pending(
         self, symbol: str, candles_json: Optional[str], limit: int = 100
     ) -> int:
-        """Advance pending advisory outcomes from a new candle snapshot."""
+        """Advance pending advisory outcomes from a new candle snapshot.
+
+        The inserting scan's own snapshot is tried first (cheap, covers the
+        steady state). When it does not overlap an outcome's last processed
+        bar — that process missed >14 bars of this timeframe, a weekend
+        straddle, a restart — the resolver rebuilds the bar stream for the
+        timeframe from every stored snapshot after that bar (any surface)
+        and retries, so the outcome is not marked UNRESOLVABLE for a gap
+        the database can fill. Each row commits on its own."""
         if not candles_json:
             return 0
         snapshots = json.loads(candles_json)
@@ -528,42 +831,248 @@ class SignalLogger:
             return 0
         changed = 0
         with self._lock:
-            self._conn.execute("BEGIN IMMEDIATE")
-            try:
-                rows = self._conn.execute(
-                    "SELECT id, timeframe, advisory_outcome_json FROM signals"
-                    " WHERE symbol=? AND advisory_outcome_status='PENDING'"
-                    " ORDER BY id",
-                    (symbol,),
-                ).fetchall()
-                eligible = 0
-                for rowid, timeframe, state_json in rows:
-                    bars = timeframe_value(snapshots, timeframe)
-                    if not bars or not state_json:
+            rows = self._conn.execute(
+                "SELECT id, timeframe FROM signals"
+                " WHERE symbol=? AND advisory_outcome_status='PENDING'"
+                " ORDER BY id",
+                (symbol,),
+            ).fetchall()
+            eligible = 0
+            for rowid, timeframe in rows:
+                bars = timeframe_value(snapshots, timeframe)
+                if not bars:
+                    continue
+                if eligible >= max(1, int(limit)):
+                    break
+                eligible += 1
+                # One IMMEDIATE transaction per row: the state is re-read
+                # under the write lock, so the bot and the dashboard (each
+                # advancing from its own snapshot) serialise instead of
+                # overwriting each other's progress.
+                self._conn.execute("BEGIN IMMEDIATE")
+                try:
+                    hit = self._conn.execute(
+                        "SELECT advisory_outcome_json FROM signals"
+                        " WHERE id=? AND advisory_outcome_status='PENDING'",
+                        (rowid,)).fetchone()
+                    if hit is None or not hit[0]:
+                        self._conn.rollback()
                         continue
-                    if eligible >= max(1, int(limit)):
-                        break
-                    eligible += 1
-                    state = json.loads(state_json)
-                    before = _dumps(state)
-                    state = advance_outcome(state, bars)
-                    after = _dumps(state)
+                    before = _dumps(json.loads(hit[0]))
+                    advanced = advance_outcome(json.loads(before), bars)
+                    if (advanced.get("status") == "UNRESOLVABLE"
+                            and "continuity" in str(advanced.get("reason", ""))):
+                        union = self._stream_since(
+                            symbol, timeframe,
+                            json.loads(before)["progress"]["last_processed_bar"])
+                        if union:
+                            advanced = advance_outcome(json.loads(before), union)
+                    after = _dumps(advanced)
                     if after == before:
+                        self._conn.rollback()
                         continue
-                    status = state.get("status", "UNRESOLVABLE")
+                    status = advanced.get("status", "UNRESOLVABLE")
                     self._conn.execute(
                         "UPDATE signals SET advisory_outcome_status=?,"
                         " advisory_r_multiple=?, advisory_outcome_json=?"
-                        " WHERE id=? AND advisory_outcome_status='PENDING'",
-                        (status, canonical_r(state), after, rowid),
+                        " WHERE id=?",
+                        (status, canonical_r(advanced), after, rowid),
                     )
+                    self._conn.commit()
                     changed += 1
-                self._conn.commit()
-            except Exception:
-                self._conn.rollback()
-                raise
+                except Exception:
+                    self._conn.rollback()
+                    raise
         return changed
 
+    def _stream_since(self, symbol: str, timeframe: str,
+                      last_bar: str, max_rows: int = 4000) -> List[Dict[str, Any]]:
+        """Union of every stored closed bar of ``timeframe`` from
+        ``last_bar`` onward, across all rows/surfaces (oldest first).
+        Only the requested timeframe's array is pulled out of each
+        candles_json (json_extract), so the cost is bounded by the number
+        of rows logged since that bar, not by blob size."""
+        tf = str(getattr(timeframe, "value", timeframe))
+        try:
+            since = datetime.fromisoformat(
+                str(last_bar).replace(" ", "T")).isoformat()
+        except (TypeError, ValueError):
+            return []
+        bars: Dict[str, Dict[str, Any]] = {}
+        for key in _tf_storage_keys(tf):
+            rows = self._conn.execute(
+                "SELECT json_extract(candles_json, ?) FROM signals"
+                " WHERE symbol=? AND candles_json IS NOT NULL"
+                " AND logged_at >= ? ORDER BY id LIMIT ?",
+                (f'$."{key}"', symbol, since, int(max_rows))).fetchall()
+            for (blob,) in rows:
+                if not blob:
+                    continue
+                try:
+                    for bar in json.loads(blob):
+                        stamp = bar.get("timestamp")
+                        if stamp and stamp >= since:
+                            bars.setdefault(str(stamp), bar)
+                except (TypeError, ValueError, AttributeError):
+                    continue
+        return [bars[k] for k in sorted(bars)]
+
+    def advance_with_bars(self, rowid: int, bars: Iterable[Dict[str, Any]],
+                          evidence: str = "external_bars") -> str:
+        """Advance ONE pending outcome from caller-supplied bars (backfill
+        from provider history). Returns the resulting status."""
+        with self._lock:
+            row = self._conn.execute(
+                "SELECT advisory_outcome_status, advisory_outcome_json"
+                " FROM signals WHERE id=?", (rowid,)).fetchone()
+            if row is None or row[0] != "PENDING" or not row[1]:
+                return str(row[0]) if row else "MISSING"
+            state = advance_outcome(json.loads(row[1]), list(bars))
+            state["evidence"] = evidence
+            status = state.get("status", "UNRESOLVABLE")
+            self._conn.execute(
+                "UPDATE signals SET advisory_outcome_status=?,"
+                " advisory_r_multiple=?, advisory_outcome_json=?"
+                " WHERE id=? AND advisory_outcome_status='PENDING'",
+                (status, canonical_r(state), _dumps(state), rowid))
+            self._conn.commit()
+            return status
+
+    def backfill_seeds(self, limit: Optional[int] = None) -> Dict[str, int]:
+        """Assign setup_key / advisory_event_id to legacy setup rows and
+        seed the canonical ones that carry no outcome yet. Existing
+        outcomes (panel-seeded rows) are never overwritten. Safe to run
+        beside live writers (short per-row transactions)."""
+        stats = {"keyed": 0, "duplicates": 0, "seeded": 0}
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT id, symbol, timeframe, logged_at, setup_json,"
+                " indicators_json, advisory_outcome_status, advisory_event_id"
+                " FROM signals WHERE setup_json IS NOT NULL"
+                " AND setup_json != 'null' ORDER BY id"
+                + (f" LIMIT {int(limit)}" if limit else "")).fetchall()
+            last_by_key: Dict[Tuple[str, str], Tuple[int, int, datetime]] = {}
+            for (rowid, symbol, tf, logged_at, setup_json, ind_json,
+                 status, event_id) in rows:
+                setup = json.loads(setup_json)
+                key = setup_key(setup, tf)
+                if not key:
+                    continue
+                try:
+                    ts = datetime.fromisoformat(logged_at)
+                except (TypeError, ValueError):
+                    continue
+                prior = last_by_key.get((symbol, key))
+                if (prior is not None
+                        and 0 <= (ts - prior[2]).total_seconds()
+                        <= OUTCOME_DEDUP_GAP_SECONDS):
+                    # Same setup re-logged inside the gap: a duplicate of
+                    # the canonical event regardless of its legacy status
+                    # (a panel-seeded RESOLVED re-log measured the very
+                    # same trade — keeping it would double count R).
+                    canonical = prior[1]
+                    if status != "DUPLICATE" or event_id != canonical:
+                        self._conn.execute(
+                            "UPDATE signals SET setup_key=?,"
+                            " advisory_event_id=?,"
+                            " advisory_outcome_status='DUPLICATE',"
+                            " advisory_r_multiple=NULL,"
+                            " advisory_outcome_json=? WHERE id=?",
+                            (key, canonical,
+                             _dumps({"model": "advisory-event-dedup-v1",
+                                     "status": "DUPLICATE",
+                                     "event_id": canonical,
+                                     "superseded_status": status}), rowid))
+                        stats["duplicates"] += 1
+                    last_by_key[(symbol, key)] = (rowid, canonical, ts)
+                    continue
+                canonical = int(event_id or rowid)
+                if status is None:
+                    indicators = json.loads(ind_json) if ind_json else None
+                    new_status, state = initial_outcome(setup, indicators, tf)
+                    self._conn.execute(
+                        "UPDATE signals SET setup_key=?, advisory_event_id=?,"
+                        " advisory_outcome_status=?, advisory_outcome_json=?"
+                        " WHERE id=?",
+                        (key, canonical, new_status, _dumps(state), rowid))
+                    stats["seeded"] += 1
+                else:
+                    self._conn.execute(
+                        "UPDATE signals SET setup_key=?, advisory_event_id=?"
+                        " WHERE id=?", (key, canonical, rowid))
+                stats["keyed"] += 1
+                last_by_key[(symbol, key)] = (rowid, canonical, ts)
+            self._conn.commit()
+        return stats
+
+    def resolve_from_store(self, symbol: str, limit: int = 500) -> int:
+        """Advance every PENDING outcome from the stored bar streams alone
+        (no provider). Used by the backfill script after seeding."""
+        changed = 0
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT id, timeframe, advisory_outcome_json FROM signals"
+                " WHERE symbol=? AND advisory_outcome_status='PENDING'"
+                " ORDER BY id LIMIT ?", (symbol, int(limit))).fetchall()
+            for rowid, timeframe, state_json in rows:
+                if not state_json:
+                    continue
+                state = json.loads(state_json)
+                union = self._stream_since(
+                    symbol, timeframe, state["progress"]["last_processed_bar"])
+                if not union:
+                    continue
+                before = _dumps(state)
+                advanced = advance_outcome(json.loads(before), union)
+                after = _dumps(advanced)
+                if after == before:
+                    continue
+                self._conn.execute(
+                    "UPDATE signals SET advisory_outcome_status=?,"
+                    " advisory_r_multiple=?, advisory_outcome_json=?"
+                    " WHERE id=? AND advisory_outcome_status='PENDING'",
+                    (advanced.get("status", "UNRESOLVABLE"),
+                     canonical_r(advanced), after, rowid))
+                self._conn.commit()
+                changed += 1
+        return changed
+
+    def reseed_unresolvable(self, max_reseeds: int = 1) -> List[int]:
+        """Bounded retry for continuity-broken outcomes: re-freeze the
+        seed from the row's own setup/indicators (at most ``max_reseeds``
+        times per row) so the next resolver pass can walk the now-filled
+        stream. Invalid-seed rows stay terminal."""
+        reseeded: List[int] = []
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT id, timeframe, setup_json, indicators_json,"
+                " advisory_outcome_json FROM signals"
+                " WHERE advisory_outcome_status='UNRESOLVABLE'"
+                " AND advisory_event_id = id").fetchall()
+            for rowid, tf, setup_json, ind_json, state_json in rows:
+                old = json.loads(state_json) if state_json else {}
+                if "continuity" not in str(old.get("reason", "")) and \
+                        "gap" not in str(old.get("reason", "")):
+                    continue
+                count = int((old.get("reseed") or {}).get("count", 0))
+                if count >= max_reseeds or not setup_json:
+                    continue
+                indicators = json.loads(ind_json) if ind_json else None
+                status, state = initial_outcome(
+                    json.loads(setup_json), indicators, tf)
+                if status != "PENDING":
+                    continue
+                state["reseed"] = {
+                    "count": count + 1,
+                    "at": datetime.now(timezone.utc).isoformat()}
+                self._conn.execute(
+                    "UPDATE signals SET advisory_outcome_status=?,"
+                    " advisory_r_multiple=NULL, advisory_outcome_json=?"
+                    " WHERE id=?", (status, _dumps(state), rowid))
+                reseeded.append(int(rowid))
+            self._conn.commit()
+        return reseeded
+
     def log_outlook(self, outlook: Any, surface: str,
                     payload: Any = None) -> int:
         """Insert one advisory market outlook; returns its rowid."""
@@ -596,6 +1105,37 @@ class SignalLogger:
             self._conn.close()
 
 
+def _tf_storage_keys(tf: str) -> List[str]:
+    """Keys under which a timeframe's bars may be stored in candles_json
+    (enum value plus the D1 alias used by older rows)."""
+    keys = [tf]
+    if tf in ("1d", "1day"):
+        keys = ["1d", "1day"]
+    return keys
+
+
+def read_news(con: sqlite3.Connection, rowid: int
+              ) -> Tuple[Optional[str], Optional[str]]:
+    """(events_json, headlines_json) for ``rowid`` on ANY connection —
+    resolves interned snapshots and falls back to the legacy inline
+    columns. Read-only."""
+    row = con.execute(
+        "SELECT events_json, headlines_json, events_hash, headlines_hash"
+        " FROM signals WHERE id=?", (rowid,)).fetchone()
+    if row is None:
+        return None, None
+    events, headlines, ev_hash, hl_hash = row
+    if events is None and ev_hash:
+        hit = con.execute("SELECT body FROM news_snapshots WHERE hash=?",
+                          (ev_hash,)).fetchone()
+        events = hit[0] if hit else None
+    if headlines is None and hl_hash:
+        hit = con.execute("SELECT body FROM news_snapshots WHERE hash=?",
+                          (hl_hash,)).fetchone()
+        headlines = hit[0] if hit else None
+    return events, headlines
+
+
 # ── Module-level entry point (config-driven, fail-safe) ───────────────
 
 _loggers: Dict[str, SignalLogger] = {}
@@ -608,14 +1148,20 @@ def log_signal(
     source: str = "live",
     timeframe: str = "",
     dataframes: Any = None,
+    extra: Optional[Dict[str, Any]] = None,
 ) -> None:
     """Persist one scan if the signal_log config block enables it.
     Stashes the rowid on the report (``_signal_log_rowid``) so a post-hoc
     panel attach can update this same row. NEVER raises — a logging
-    failure must not break a live scan."""
+    failure must not break a live scan. ``extra`` = producer metadata
+    (``scan_kind``, ``config_fingerprint``); blocked scans are skipped
+    when ``signal_log.log_blocked_scans`` is false."""
     try:
         if not app_config.signal_log_enabled():
             return
+        if ((extra or {}).get("scan_kind") == "blocked"
+                and not app_config.signal_log_blocked_scans()):
+            return
         path = app_config.signal_log_path()
         with _loggers_lock:
             lg = _loggers.get(path)
@@ -623,7 +1169,8 @@ def log_signal(
                 lg = SignalLogger(path)
                 _loggers[path] = lg
         rowid = lg.log(report, context=context, source=source,
-                       timeframe=timeframe, dataframes=dataframes)
+                       timeframe=timeframe, dataframes=dataframes,
+                       extra=extra)
         try:
             report._signal_log_rowid = rowid
         except Exception: