Merge PR #1849: record declared backtests in strategy bench history

agent/src/skills/strategy-dev-manager/SKILL.md

@@ -84,13 +84,13 @@ After ingesting a paper via `read_document`, check the `ocr_quality` field in th
 Build the SignalEngine, run the backtest, and link results.
 
 1. Call `create_hypothesis(title, thesis, universe, signal_definition)` to create a research hypothesis that tracks this work.
-2. Call `generate_backtest_config(hypothesis_id, start_date, end_date)` to produce the `config.json` for the backtest runner.
+2. Call `generate_backtest_config(hypothesis_id, start_date, end_date)` to produce the `config.json` for the backtest runner. For an artifact registered with `sdm_register`, add its `artifact_id` to `config.json` so the backtest is recorded in the artifact's bench history. Add `"bench_type": "initial"` for the first evaluation run and `"periodic"` for scheduled re-benches; without it the run is recorded as `manual`, which decay monitoring ignores.
 3. Call `scaffold_signal_engine(hypothesis_id, run_dir)` to generate the skeleton `signal_engine.py` in the run directory.
 4. Implement the full `signal_engine.py` using the appropriate template from `templates/`:
    - Factor artifacts → `templates/factor_signal_engine.py`
    - Strategy artifacts → `templates/strategy_signal_engine.py`
 5. Validate syntax: `bash("python -c \"import ast; ast.parse(open('code/signal_engine.py').read()); print('OK')\"")`
-6. Call `backtest(run_dir)` to execute the backtest.
+6. Call `backtest(run_dir)` to execute the backtest. When `config.json` declares an `artifact_id`, the result carries a `bench_record` that reports whether the run was `recorded`, `skipped` (failed run, no trades, missing metrics, or an identical row already stored) or `failed` (for example an unregistered `artifact_id`), with a `reason`.
 7. Call `link_autopilot_backtest(hypothesis_id, run_dir)` to link the run results back to the hypothesis.
 8. Call `sdm_status(action="detail", artifact_id=...)` and update the artifact status to "benching".
 
@@ -116,7 +116,7 @@ Judge the backtest output against quality thresholds.
    - Update status to "disabled" with a reason string
    - Record what failed (for example "IC mean 0.012, below 0.03 threshold")
 
-5. Record bench results via `sdm_status` update so the history is queryable.
+5. Check the bench record: the backtest itself writes the bench row when `config.json` declares the `artifact_id`. Confirm `bench_record.status` is `recorded` in the backtest result, or list the rows with `sdm_status(action="detail", artifact_id=...)`. If the run was not recorded, fix the config and re-run the backtest.
 
 ### Phase 5: MONITOR
 
@@ -205,7 +205,7 @@ LLMs may generate plausible-looking formulas that do not appear in the paper. AL
 IC > 0.99 means the factor is a duplicate. IC between 0.90 and 0.99 may be a variant. Use judgment: if the formula is structurally different but produces similar signals, note it as a variant rather than rejecting it outright.
 
 ### Decay Baseline
-Decay monitoring requires at least 3 bench history entries to establish a baseline. A newly registered artifact with only one backtest cannot be meaningfully scanned for decay.
+Decay monitoring requires at least 3 `initial` or `periodic` bench history entries to establish a baseline; `manual` entries are kept for reference and do not count. A newly registered artifact with only one backtest cannot be meaningfully scanned for decay.
 
 ### Template Mismatch
 Strategy-type artifacts need the strategy SignalEngine template (with entry/exit/position logic), not the factor template. Using the wrong template produces a SignalEngine that compiles but generates meaningless signals.

agent/src/skills/strategy-generate/SKILL.md

@@ -164,6 +164,7 @@ Self-check after writing `signal_engine.py`:
   - Rule of thumb: `"rebalance"` when the target weight itself carries the strategy (optimizers, risk budgets, continuous scaling); `"hold"` when entries and exits carry it and the weight in between is incidental.
 - `rebalance_mask`: optional execution schedule used only under `"rebalance"`. Use a pandas offset alias such as `"MS"`, `"W-FRI"`, or `"QS"`, or an explicit ISO-date list such as `["2026-01-02", "2026-02-02"]`. Each period/date selects the first aligned trading bar on or after it; ordinary bars HOLD even when the dense target is zero. An alias must not be finer than the aligned bar interval; `W-FRI` starts a Friday-anchored period and normally executes on the following Monday. Omit it to preserve every-bar execution. Do not combine it with `"hold"`.
 - `rebalance_tolerance`: drift band around the target, as a fraction of it, used only under `"rebalance"`. A resize executes once the held weight has moved further than this from its target; a **changed** target breaches any sane band on its own, so target changes always execute. Default `0.0` means no band, and then the resize test is decided by the slippage width alone — measured on a constant 20% target over 60 bars, `0.0` produced 60 fills, `0.02` produced 12, and `0.05` produced 5 while the weight never left 0.21. Use `rebalance_mask`, not tolerance, to express a strategy's execution cadence. `0.05` is a reasonable starting point, not a recommendation with evidence behind it — it is your modelling choice and the report records the value the run used.
+- `artifact_id`: optional. Set it only when the run benches an artifact registered with `sdm_register`; the backtest then records its metrics in that artifact's bench history. `bench_type` (`"initial"` / `"periodic"` / `"manual"`, default `"manual"`) states whether the run counts towards decay monitoring, which reads only `initial` and `periodic` runs. Omit both for exploratory runs.
 - `initial_cash`: default 1,000,000
 - `commission`: default 0.1%
 - `hk_lot_sizes`: optional, Hong Kong only — `{"00700.HK": 100, "02800.HK": 500}`. HK board lots are set per issuer, so without this table every HK order is rounded to 100 shares and a name whose real lot is smaller than its target trades zero times. Codes match with or without leading zeros. Every symbol with no configured lot size uses 100 shares and is listed under `lot_assumptions` in the metrics, including runs without a table. Reports must identify these as assumptions rather than verified exchange lots.

agent/src/strategy_store/metrics.py

@@ -12,6 +12,10 @@ from typing import Any
 
 _DECAY_INPUT_KEYS = ("ic_ratio", "rolling_ir", "ic_positive_ratio", "rolling_sharpe")
 
+#: Bench types that measure an artifact over time. ``manual`` rows record
+#: ad-hoc reruns and are kept out of decay so they cannot shift the baseline.
+DECAY_BENCH_TYPES = ("initial", "periodic")
+
 
 def has_decay_inputs(metrics: dict[str, float | None]) -> bool:
     """Return True if at least one evaluate_decay input metric is available.

agent/src/strategy_store/models.py

@@ -134,6 +134,10 @@ class Artifact:
     validation_date: str | None = None
 
 
+#: Bench types the ``bench_history`` schema accepts.
+BENCH_TYPES = ("initial", "periodic", "manual")
+
+
 @dataclass(frozen=True)
 class BenchResult:
     """Backtest result for an artifact."""

agent/src/strategy_store/sqlite_store.py

@@ -620,63 +620,109 @@ class SqliteStrategyStore:
 
     # -- Bench history ------------------------------------------------------
 
+    @staticmethod
+    def _bench_values(result: BenchResult) -> dict[str, object]:
+        """Column values of a bench row other than ``id`` and ``created_at``.
+
+        One list serves both the insert and the duplicate check, so a column
+        added to one cannot be missing from the other.
+        """
+        return {
+            "artifact_id": result.artifact_id,
+            "bench_type": result.bench_type,
+            "ic_mean": result.ic_mean,
+            "ic_std": result.ic_std,
+            "ir": result.ir,
+            "ic_positive_ratio": result.ic_positive_ratio,
+            "t_stat": result.t_stat,
+            "sharpe": result.sharpe,
+            "annual_return": result.annual_return,
+            "max_drawdown": result.max_drawdown,
+            "calmar": result.calmar,
+            "category": result.category.value if result.category else None,
+            "train_start": result.train_start,
+            "train_end": result.train_end,
+            "test_start": result.test_start,
+            "test_end": result.test_end,
+            "run_dir": result.run_dir,
+        }
+
     @_synchronized
     def record_bench(self, result: BenchResult) -> int:
         """Record a bench result.  Returns the result ID."""
-        now = _now_iso()
-        created_at = result.created_at or now
-        category_value = result.category.value if result.category else None
+        values = self._bench_values(result)
+        values["created_at"] = result.created_at or _now_iso()
+        columns = ", ".join(values)
+        placeholders = ", ".join("?" for _ in values)
 
         with self._write_transaction():
             cursor = self._conn.execute(
-                """
-                INSERT INTO bench_history (
-                    artifact_id, bench_type, ic_mean, ic_std, ir,
-                    ic_positive_ratio, t_stat, sharpe, annual_return,
-                    max_drawdown, calmar, category,
-                    train_start, train_end, test_start, test_end,
-                    run_dir, created_at
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-                """,
-                (
-                    result.artifact_id,
-                    result.bench_type,
-                    result.ic_mean,
-                    result.ic_std,
-                    result.ir,
-                    result.ic_positive_ratio,
-                    result.t_stat,
-                    result.sharpe,
-                    result.annual_return,
-                    result.max_drawdown,
-                    result.calmar,
-                    category_value,
-                    result.train_start,
-                    result.train_end,
-                    result.test_start,
-                    result.test_end,
-                    result.run_dir,
-                    created_at,
-                ),
+                f"INSERT INTO bench_history ({columns}) VALUES ({placeholders})",
+                list(values.values()),
             )
             return cursor.lastrowid  # type: ignore[return-value]
 
     @_synchronized
     def get_bench_history(
-        self, artifact_id: str, *, limit: int = 50
+        self,
+        artifact_id: str,
+        *,
+        limit: int = 50,
+        bench_types: Sequence[str] | None = None,
     ) -> Sequence[BenchResult]:
-        """Get bench history for an artifact, newest first."""
+        """Get bench history for an artifact, newest first.
+
+        Args:
+            artifact_id: Artifact whose bench rows are returned.
+            limit: Maximum number of rows.
+            bench_types: When given, only rows of these bench types are
+                returned, and ``limit`` applies after the filter.
+
+        Returns:
+            Matching bench results, newest first.
+        """
+        type_clause = ""
+        params: list[object] = [artifact_id]
+        if bench_types is not None:
+            types = list(bench_types)
+            if not types:
+                return []
+            type_clause = f"AND bench_type IN ({', '.join('?' for _ in types)})"
+            params.extend(types)
+        params.append(limit)
         rows = self._conn.execute(
-            """
+            f"""
             SELECT * FROM bench_history
-            WHERE artifact_id = ?
+            WHERE artifact_id = ? {type_clause}
             ORDER BY created_at DESC, id DESC
             LIMIT ?
             """,
-            (artifact_id, limit),
+            params,
         ).fetchall()
         return [self._row_to_bench_result(row) for row in rows]
 
+    @_synchronized
+    def has_bench_result(self, result: BenchResult) -> bool:
+        """Return whether an identical bench row is already stored.
+
+        ``IS`` rather than ``=`` makes two missing values compare equal, so a
+        metric absent from both rows still counts as a match.
+
+        Args:
+            result: Candidate bench result; ``id`` and ``created_at`` are
+                ignored.
+
+        Returns:
+            ``True`` when a stored row matches ``result`` on every other field.
+        """
+        values = self._bench_values(result)
+        where = " AND ".join(f"{column} IS ?" for column in values)
+        row = self._conn.execute(
+            f"SELECT 1 FROM bench_history WHERE {where} LIMIT 1",
+            list(values.values()),
+        ).fetchone()
+        return row is not None
+
     # -- Decay snapshots ----------------------------------------------------
 
     @_synchronized

agent/src/strategy_store/store.py

@@ -117,9 +117,37 @@ class StrategyStoreProtocol(Protocol):
         ...
 
     def get_bench_history(
-        self, artifact_id: str, *, limit: int = 50
+        self,
+        artifact_id: str,
+        *,
+        limit: int = 50,
+        bench_types: Sequence[str] | None = None,
     ) -> Sequence[BenchResult]:
-        """Get bench history for an artifact, newest first."""
+        """Get bench history for an artifact, newest first.
+
+        Args:
+            artifact_id: Artifact whose bench rows are returned.
+            limit: Maximum number of rows.
+            bench_types: When given, only rows of these bench types are
+                returned, and ``limit`` applies after the filter.
+
+        Returns:
+            Matching bench results, newest first.
+        """
+        ...
+
+    def has_bench_result(self, result: BenchResult) -> bool:
+        """Return whether an identical bench row is already stored.
+
+        Rows are identical when every field except ``id`` and ``created_at``
+        is equal, so a rerun that changed any metric or window is not a match.
+
+        Args:
+            result: Candidate bench result.
+
+        Returns:
+            ``True`` when a stored row matches ``result``.
+        """
         ...
 
     # --- Decay snapshots ---
@@ -299,14 +327,50 @@ class InMemoryStrategyStore:
 
     @_synchronized
     def get_bench_history(
-        self, artifact_id: str, *, limit: int = 50
+        self,
+        artifact_id: str,
+        *,
+        limit: int = 50,
+        bench_types: Sequence[str] | None = None,
     ) -> Sequence[BenchResult]:
-        """Return bench results for an artifact, newest first."""
-        matched = [r for r in self._bench_results if r.artifact_id == artifact_id]
+        """Return bench results for an artifact, newest first.
+
+        Args:
+            artifact_id: Artifact whose bench rows are returned.
+            limit: Maximum number of rows.
+            bench_types: When given, only rows of these bench types are
+                returned, and ``limit`` applies after the filter.
+
+        Returns:
+            Matching bench results, newest first.
+        """
+        matched = [
+            r
+            for r in self._bench_results
+            if r.artifact_id == artifact_id
+            and (bench_types is None or r.bench_type in bench_types)
+        ]
         # id ascends with insertion, so it breaks timestamp ties newest first.
         matched.sort(key=lambda r: (r.created_at or "", r.id or 0), reverse=True)
         return matched[:limit]
 
+    @_synchronized
+    def has_bench_result(self, result: BenchResult) -> bool:
+        """Return whether an identical bench row is already stored.
+
+        Args:
+            result: Candidate bench result; ``id`` and ``created_at`` are
+                ignored.
+
+        Returns:
+            ``True`` when a stored row matches ``result`` on every other field.
+        """
+        probe = replace(result, id=None, created_at="")
+        return any(
+            replace(stored, id=None, created_at="") == probe
+            for stored in self._bench_results
+        )
+
     # -- Decay snapshots -----------------------------------------------------
 
     @_synchronized

agent/src/tools/backtest_tool.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import csv
 import json
 import subprocess
 from pathlib import Path
@@ -14,6 +15,7 @@ from src.config.accessor import get_env_config
 from src.config.limits import TOOL_RESULT_LIMIT
 from src.core.runner import Runner
 from src.core.state import RunStateStore
+from src.strategy_store.models import BENCH_TYPES
 from src.tools.backtest_summary import collect_ohlcv_paths, try_build_backtest_summary
 from src.tools.path_utils import safe_run_dir
 
@@ -60,6 +62,135 @@ def _persist_timeout_output(run_path: Path, exc: subprocess.TimeoutExpired) -> d
     return output
 
 
+#: Strategy metrics copied from ``artifacts/metrics.csv`` into a bench row. A
+#: strategy backtest has no cross-sectional IC, so the factor columns stay
+#: ``None`` rather than being filled with a stand-in.
+_BENCH_METRIC_COLUMNS = ("sharpe", "annual_return", "max_drawdown", "calmar")
+
+#: Bench type of a run whose config does not declare one; it is left out of
+#: decay monitoring (see ``DECAY_BENCH_TYPES``).
+_DEFAULT_BENCH_TYPE = "manual"
+
+
+def _read_bench_metrics(run_path: Path) -> dict[str, float] | None:
+    """Read the strategy metrics a bench row records from ``metrics.csv``.
+
+    Args:
+        run_path: Run directory containing ``artifacts/metrics.csv``.
+
+    Returns:
+        Finite bench metric values keyed by column name, or ``None`` when the
+        file is missing or has no data row. NaN and infinities are dropped
+        because decay monitoring treats them as missing observations.
+    """
+    from src.strategy_discovery.run_artifacts import parse_finite_float
+
+    metrics_path = run_path / "artifacts" / "metrics.csv"
+    if not metrics_path.is_file():
+        return None
+    with metrics_path.open(encoding="utf-8-sig", newline="") as handle:
+        row = next(csv.DictReader(handle), None)
+    if not row:
+        return None
+    # Same header normalisation as read_metrics_trade_count, so the two readers
+    # of this file cannot disagree about a column.
+    row = {(name or "").strip().lower(): value for name, value in row.items()}
+    metrics: dict[str, float] = {}
+    for column in _BENCH_METRIC_COLUMNS:
+        value = parse_finite_float(row.get(column))
+        if value is not None:
+            metrics[column] = value
+    return metrics
+
+
+def _record_bench_if_declared(
+    config: dict[str, Any], run_path: Path, *, success: bool
+) -> dict[str, Any] | None:
+    """Record the run in the bench history of the artifact its config declares.
+
+    A backtest cannot tell which registered artifact it serves, so the link is
+    declared through ``artifact_id`` in ``config.json``. The bench type comes
+    from ``bench_type`` in the config and defaults to ``manual`` (see
+    ``DECAY_BENCH_TYPES``). ``category`` stays ``None``: grading a run alive or
+    dead belongs to the evaluation step, not to the runner that produced the
+    numbers.
+
+    One row is kept per distinct measurement. A rerun that reproduces a stored
+    row exactly is skipped, so repeating a run cannot count as an independent
+    confirmation; a rerun of the same directory whose metrics or window
+    changed is a new measurement and is recorded.
+
+    Args:
+        config: Parsed ``config.json`` of the run.
+        run_path: Resolved run directory.
+        success: Whether the backtest engine finished successfully.
+
+    Returns:
+        ``None`` when the config declares no ``artifact_id``. Otherwise a dict
+        with ``status`` (``recorded``, ``skipped`` or ``failed``) and
+        ``artifact_id``, plus ``bench_type`` when recorded or ``reason``
+        otherwise. Errors are reported here instead of raised, so bookkeeping
+        never turns a successful backtest into a failed one.
+    """
+    artifact_id = str(config.get("artifact_id") or "").strip()
+    if not artifact_id:
+        return None
+
+    def _outcome(status: str, **detail: str) -> dict[str, Any]:
+        return {"status": status, "artifact_id": artifact_id, **detail}
+
+    if not success:
+        return _outcome("skipped", reason="backtest did not succeed")
+
+    bench_type = str(config.get("bench_type") or _DEFAULT_BENCH_TYPE).strip()
+    if bench_type not in BENCH_TYPES:
+        return _outcome(
+            "failed",
+            reason=f"bench_type must be one of {', '.join(BENCH_TYPES)}, "
+            f"got {bench_type!r}",
+        )
+
+    try:
+        from src.strategy_discovery.run_artifacts import read_metrics_trade_count
+        from src.strategy_store._shared import get_store
+        from src.strategy_store.models import BenchResult
+
+        metrics = _read_bench_metrics(run_path)
+        if metrics is None:
+            return _outcome("skipped", reason="artifacts/metrics.csv missing or empty")
+        if not metrics:
+            return _outcome(
+                "skipped",
+                reason="metrics.csv has no finite " + ", ".join(_BENCH_METRIC_COLUMNS),
+            )
+        # A run without trades has nothing to measure decay against, and its
+        # zero-valued metrics would read as a real collapse in performance.
+        trade_count = read_metrics_trade_count(run_path)
+        if trade_count is None:
+            return _outcome("skipped", reason="metrics.csv has no usable trade_count")
+        if trade_count <= 0:
+            return _outcome("skipped", reason="backtest produced no trades")
+
+        store = get_store()
+        if store.get_artifact(artifact_id) is None:
+            return _outcome("failed", reason="artifact_id is not registered")
+
+        result = BenchResult(
+            artifact_id=artifact_id,
+            bench_type=bench_type,
+            test_start=str(config.get("start_date") or "") or None,
+            test_end=str(config.get("end_date") or "") or None,
+            run_dir=str(run_path),
+            **metrics,
+        )
+        if store.has_bench_result(result):
+            return _outcome("skipped", reason="identical bench already recorded")
+        store.record_bench(result)
+        return _outcome("recorded", bench_type=bench_type)
+    except Exception as exc:  # noqa: BLE001
+        return _outcome("failed", reason=f"{type(exc).__name__}: {exc}")
+
+
 def run_backtest(run_dir: str) -> str:
     """Run backtest: validate config.json + signal_engine.py, invoke built-in engine.
 
@@ -153,6 +284,8 @@ def run_backtest(run_dir: str) -> str:
     else:
         state_store.mark_failure(run_path, f"backtest engine exited with code {result.exit_code}")
 
+    bench_record = _record_bench_if_declared(config, run_path, success=result.success)
+
     emit_progress("finalize", message="collecting artifacts")
     artifacts_found = {name: str(path) for name, path in result.artifacts.items()}
     envelope: dict[str, Any] = {
@@ -163,6 +296,8 @@ def run_backtest(run_dir: str) -> str:
         "artifacts": artifacts_found,
         "run_dir": run_dir,
     }
+    if bench_record is not None:
+        envelope["bench_record"] = bench_record
     if result.success:
         # Small envelopes retain the legacy fields unchanged. Oversized
         # envelopes are fitted below with explicit omission notices. A missing

agent/src/tools/sdm_decay_scan_tool.py

@@ -7,7 +7,11 @@ from typing import Any
 
 from src.agent.tools import BaseTool
 from src.strategy_store.decay import DecayEvaluator
-from src.strategy_store.metrics import compute_decay_metrics, has_decay_inputs
+from src.strategy_store.metrics import (
+    DECAY_BENCH_TYPES,
+    compute_decay_metrics,
+    has_decay_inputs,
+)
 from src.strategy_store.models import (
     ArtifactStatus,
     ArtifactType,
@@ -96,7 +100,9 @@ class SdmDecayScanTool(BaseTool):
             for artifact in targets:
                 counts["total_scanned"] += 1
                 bench_history = list(
-                    store.get_bench_history(artifact.id, limit=20)
+                    store.get_bench_history(
+                        artifact.id, limit=20, bench_types=DECAY_BENCH_TYPES
+                    )
                 )
 
                 if len(bench_history) < 3:

agent/src/tools/sdm_status_tool.py

@@ -10,7 +10,11 @@ from typing import Any
 
 from src.agent.tools import BaseTool
 from src.strategy_store.decay import DecayEvaluator
-from src.strategy_store.metrics import compute_decay_metrics, has_decay_inputs
+from src.strategy_store.metrics import (
+    DECAY_BENCH_TYPES,
+    compute_decay_metrics,
+    has_decay_inputs,
+)
 from src.strategy_store.models import (
     ArtifactStatus,
     ArtifactType,
@@ -221,12 +225,19 @@ class SdmStatusTool(BaseTool):
         if artifact is None:
             return _error(ValueError(f"Artifact not found: {artifact_id}"))
 
-        bench_history = list(store.get_bench_history(artifact_id, limit=10))
+        bench_history = list(
+            store.get_bench_history(
+                artifact_id, limit=10, bench_types=DECAY_BENCH_TYPES
+            )
+        )
         if len(bench_history) < 3:
             return _ok({
                 "artifact_id": artifact_id,
                 "signal": "insufficient_data",
-                "message": f"Need at least 3 bench results, have {len(bench_history)}",
+                "message": (
+                    "Need at least 3 initial or periodic bench results, "
+                    f"have {len(bench_history)}"
+                ),
             })
 
         metrics = compute_decay_metrics(bench_history)

agent/tests/test_backtest_records_bench.py

@@ -0,0 +1,399 @@
+"""A backtest whose config declares an artifact lands in that artifact's bench history.
+
+Decay monitoring reads ``bench_history``, so the rows written here decide what
+it sees: which runs count towards decay, whether a rerun is a new measurement,
+and whether an idle run can pose as one.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field, replace
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from src.strategy_store.metrics import DECAY_BENCH_TYPES
+from src.strategy_store.models import Artifact, ArtifactType, BenchResult
+from src.strategy_store.sqlite_store import SqliteStrategyStore
+from src.strategy_store.store import InMemoryStrategyStore
+from src.tools.backtest_tool import run_backtest
+from src.tools.sdm_status_tool import SdmStatusTool
+
+_METRICS_HEADER = "sharpe,annual_return,max_drawdown,calmar,trade_count\n"
+_METRICS_ROW = "0.7845,0.1067,-0.2123,0.5026,1427\n"
+
+
+@dataclass
+class _FakeRunResult:
+    success: bool
+    exit_code: int
+    stdout: str = ""
+    stderr: str = ""
+    artifacts: dict[str, Path] = field(default_factory=dict)
+
+
+@pytest.fixture()
+def store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Any:
+    """A fresh SQLite store per test, shared with the tools through the singleton."""
+    monkeypatch.setenv(
+        "VIBE_TRADING_STRATEGY_STORE_DB_PATH", str(tmp_path / "store.db")
+    )
+    import src.strategy_store._shared as shared
+
+    # The env var alone does not isolate a test once the singleton exists.
+    monkeypatch.setattr(shared, "_store", None)
+    return shared.get_store()
+
+
+@pytest.fixture()
+def artifact_id(store: Any) -> str:
+    """A registered strategy artifact."""
+    return store.register_artifact(
+        Artifact(
+            id="",
+            type=ArtifactType.STRATEGY,
+            name="inverse vol weighting",
+            universe="equity_hk",
+        )
+    )
+
+
+@pytest.fixture()
+def run_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Directory the backtest tool is allowed to run in."""
+    root = tmp_path / "runs"
+    root.mkdir()
+    monkeypatch.setenv("VIBE_TRADING_ALLOWED_RUN_ROOTS", str(root))
+    return root
+
+
+@pytest.fixture()
+def run_dir(run_root: Path) -> Path:
+    """A run directory holding the metrics of a run that traded."""
+    return _make_run_dir(run_root, "run")
+
+
+def _make_run_dir(root: Path, name: str, metrics_row: str = _METRICS_ROW) -> Path:
+    run_dir = root / name
+    (run_dir / "code").mkdir(parents=True)
+    (run_dir / "artifacts").mkdir(parents=True)
+    (run_dir / "code" / "signal_engine.py").write_text("", encoding="utf-8")
+    _write_metrics(run_dir, metrics_row)
+    return run_dir
+
+
+def _write_metrics(run_dir: Path, row: str, header: str = _METRICS_HEADER) -> None:
+    metrics_path = run_dir / "artifacts" / "metrics.csv"
+    metrics_path.write_text(header + row, encoding="utf-8")
+
+
+def _write_config(run_dir: Path, **extra: Any) -> None:
+    config = {
+        "source": "yfinance",
+        "start_date": "2021-01-04",
+        "end_date": "2026-09-25",
+        **extra,
+    }
+    (run_dir / "config.json").write_text(json.dumps(config), encoding="utf-8")
+
+
+def _run(run_dir: Path, *, success: bool = True, exit_code: int = 0) -> dict[str, Any]:
+    with (
+        patch("src.tools.backtest_tool.emit_progress"),
+        patch("src.tools.backtest_tool.Runner") as runner_cls,
+    ):
+        runner_cls.return_value.execute.return_value = _FakeRunResult(
+            success=success, exit_code=exit_code
+        )
+        return json.loads(run_backtest(str(run_dir)))
+
+
+def _decay_check(artifact_id: str) -> dict[str, Any]:
+    return json.loads(
+        SdmStatusTool().execute(action="decay_check", artifact_id=artifact_id)
+    )
+
+
+# -- recording ---------------------------------------------------------------
+
+
+def test_declared_run_is_recorded_as_manual_by_default(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    _write_config(run_dir, artifact_id=artifact_id)
+
+    envelope = _run(run_dir)
+
+    assert envelope["status"] == "ok", envelope
+    assert envelope["bench_record"] == {
+        "status": "recorded",
+        "artifact_id": artifact_id,
+        "bench_type": "manual",
+    }
+    (row,) = store.get_bench_history(artifact_id)
+    assert row.bench_type == "manual"
+    assert row.sharpe == pytest.approx(0.7845)
+    assert row.annual_return == pytest.approx(0.1067)
+    assert row.max_drawdown == pytest.approx(-0.2123)
+    assert row.calmar == pytest.approx(0.5026)
+    assert (row.test_start, row.test_end) == ("2021-01-04", "2026-09-25")
+    assert row.run_dir == str(run_dir)
+
+
+def test_runner_leaves_grading_and_factor_metrics_empty(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    """Alive or dead is the evaluation step's verdict, and a strategy has no IC."""
+    _write_config(run_dir, artifact_id=artifact_id)
+
+    _run(run_dir)
+
+    (row,) = store.get_bench_history(artifact_id)
+    assert row.category is None
+    assert row.ic_mean is None
+    assert row.ic_std is None
+    assert row.ir is None
+    assert row.ic_positive_ratio is None
+    assert row.t_stat is None
+
+
+@pytest.mark.parametrize("bench_type", ["initial", "periodic", "manual"])
+def test_declared_bench_type_is_recorded(
+    run_dir: Path, store: Any, artifact_id: str, bench_type: str
+) -> None:
+    _write_config(run_dir, artifact_id=artifact_id, bench_type=bench_type)
+
+    envelope = _run(run_dir)
+
+    assert envelope["bench_record"]["bench_type"] == bench_type
+    (row,) = store.get_bench_history(artifact_id)
+    assert row.bench_type == bench_type
+
+
+def test_unknown_bench_type_is_refused(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    _write_config(run_dir, artifact_id=artifact_id, bench_type="weekly")
+
+    envelope = _run(run_dir)
+
+    assert envelope["status"] == "ok"
+    assert envelope["bench_record"]["status"] == "failed"
+    assert store.get_bench_history(artifact_id) == []
+
+
+def test_only_declared_periodic_runs_feed_decay(
+    run_root: Path, store: Any, artifact_id: str
+) -> None:
+    """Default runs are bookkeeping; decay moves only on runs declared for it."""
+    for index in range(3):
+        manual = _make_run_dir(
+            run_root, f"manual{index}", f"0.{index + 1},0.1,-0.2,0.5,10\n"
+        )
+        _write_config(manual, artifact_id=artifact_id)
+        _run(manual)
+
+    assert len(store.get_bench_history(artifact_id)) == 3
+    assert _decay_check(artifact_id)["signal"] == "insufficient_data"
+
+    for index in range(3):
+        periodic = _make_run_dir(
+            run_root, f"periodic{index}", f"1.{index + 1},0.1,-0.2,0.5,10\n"
+        )
+        _write_config(periodic, artifact_id=artifact_id, bench_type="periodic")
+        _run(periodic)
+
+    result = _decay_check(artifact_id)
+    assert result["signal"] != "insufficient_data", result
+    assert result["metrics"]["baseline_sharpe"] == pytest.approx(1.2)
+
+
+# -- reruns ------------------------------------------------------------------
+
+
+def test_identical_rerun_is_not_recorded_twice(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    """The same measurement counted twice would look like two confirmations."""
+    _write_config(run_dir, artifact_id=artifact_id)
+    _run(run_dir)
+
+    envelope = _run(run_dir)
+
+    assert envelope["bench_record"] == {
+        "status": "skipped",
+        "artifact_id": artifact_id,
+        "reason": "identical bench already recorded",
+    }
+    assert len(store.get_bench_history(artifact_id)) == 1
+
+
+def test_rerun_with_changed_metrics_is_recorded(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    """A rerun of the same directory that measured something new is kept."""
+    _write_config(run_dir, artifact_id=artifact_id)
+    _run(run_dir)
+    _write_metrics(run_dir, "0.6012,0.0911,-0.2500,0.3644,1388\n")
+
+    envelope = _run(run_dir)
+
+    assert envelope["bench_record"]["status"] == "recorded"
+    sharpes = [row.sharpe for row in store.get_bench_history(artifact_id)]
+    assert sharpes == pytest.approx([0.6012, 0.7845])
+
+
+# -- runs that must not be recorded ------------------------------------------
+
+
+def test_undeclared_run_keeps_legacy_envelope_and_writes_nothing(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    _write_config(run_dir)
+
+    envelope = _run(run_dir)
+
+    assert envelope["status"] == "ok", envelope
+    assert "bench_record" not in envelope
+    assert store.get_bench_history(artifact_id) == []
+
+
+def test_failed_backtest_records_no_bench(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    _write_config(run_dir, artifact_id=artifact_id)
+
+    envelope = _run(run_dir, success=False, exit_code=3)
+
+    assert envelope["status"] == "error"
+    assert envelope["bench_record"] == {
+        "status": "skipped",
+        "artifact_id": artifact_id,
+        "reason": "backtest did not succeed",
+    }
+    assert store.get_bench_history(artifact_id) == []
+
+
+def test_zero_trade_run_is_skipped(run_dir: Path, store: Any, artifact_id: str) -> None:
+    """Zero-valued metrics from an idle run would read as a performance collapse."""
+    _write_config(run_dir, artifact_id=artifact_id)
+    _write_metrics(run_dir, "0,0,0,0,0\n")
+
+    envelope = _run(run_dir)
+
+    assert envelope["bench_record"] == {
+        "status": "skipped",
+        "artifact_id": artifact_id,
+        "reason": "backtest produced no trades",
+    }
+    assert store.get_bench_history(artifact_id) == []
+
+
+def test_metrics_without_trade_count_are_skipped(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    _write_config(run_dir, artifact_id=artifact_id)
+    _write_metrics(
+        run_dir,
+        "0.7845,0.1067,-0.2123,0.5026\n",
+        header="sharpe,annual_return,max_drawdown,calmar\n",
+    )
+
+    envelope = _run(run_dir)
+
+    assert envelope["bench_record"]["status"] == "skipped"
+    assert store.get_bench_history(artifact_id) == []
+
+
+def test_missing_metrics_csv_is_skipped(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    """An empty row would count towards the decay floor with nothing to measure."""
+    _write_config(run_dir, artifact_id=artifact_id)
+    (run_dir / "artifacts" / "metrics.csv").unlink()
+
+    envelope = _run(run_dir)
+
+    assert envelope["bench_record"]["status"] == "skipped"
+    assert store.get_bench_history(artifact_id) == []
+
+
+def test_store_failure_is_reported_without_failing_the_backtest(
+    run_dir: Path, store: Any, artifact_id: str
+) -> None:
+    _write_config(run_dir, artifact_id=artifact_id)
+
+    with patch(
+        "src.strategy_store.sqlite_store.SqliteStrategyStore.record_bench",
+        side_effect=RuntimeError("disk full"),
+    ):
+        envelope = _run(run_dir)
+
+    assert envelope["status"] == "ok"
+    assert envelope["bench_record"]["status"] == "failed"
+    assert "disk full" in envelope["bench_record"]["reason"]
+    assert store.get_bench_history(artifact_id) == []
+
+
+def test_unregistered_artifact_id_is_refused(run_dir: Path, store: Any) -> None:
+    """A mistyped id must surface as a failure, not as an orphaned row."""
+    _write_config(run_dir, artifact_id="never-registered")
+
+    envelope = _run(run_dir)
+
+    assert envelope["status"] == "ok"
+    assert envelope["bench_record"]["status"] == "failed"
+    assert store.get_bench_history("never-registered") == []
+
+
+# -- store contract ----------------------------------------------------------
+
+
+@pytest.fixture(params=["memory", "sqlite"])
+def backend(request: pytest.FixtureRequest, tmp_path: Path) -> Any:
+    """An empty store of each backend."""
+    if request.param == "memory":
+        return InMemoryStrategyStore()
+    return SqliteStrategyStore(db_path=tmp_path / "contract.db")
+
+
+def _register(backend: Any) -> str:
+    return backend.register_artifact(
+        Artifact(
+            id="", type=ArtifactType.STRATEGY, name="contract", universe="equity_us"
+        )
+    )
+
+
+def test_has_bench_result_matches_only_identical_rows(backend: Any) -> None:
+    artifact_id = _register(backend)
+    stored = BenchResult(
+        artifact_id=artifact_id, bench_type="manual", sharpe=0.5, run_dir="/runs/a"
+    )
+    backend.record_bench(stored)
+
+    assert backend.has_bench_result(stored)
+    assert backend.has_bench_result(replace(stored, created_at="2020-01-01"))
+    assert not backend.has_bench_result(replace(stored, sharpe=0.6))
+    assert not backend.has_bench_result(replace(stored, run_dir="/runs/b"))
+    assert not backend.has_bench_result(replace(stored, bench_type="periodic"))
+
+
+def test_bench_history_filters_by_type_before_limit(backend: Any) -> None:
+    artifact_id = _register(backend)
+    backend.record_bench(
+        BenchResult(artifact_id=artifact_id, bench_type="periodic", sharpe=1.0)
+    )
+    backend.record_bench(
+        BenchResult(artifact_id=artifact_id, bench_type="manual", sharpe=2.0)
+    )
+
+    rows = backend.get_bench_history(
+        artifact_id, limit=1, bench_types=DECAY_BENCH_TYPES
+    )
+
+    assert [row.sharpe for row in rows] == [1.0]
+    assert len(backend.get_bench_history(artifact_id)) == 2