fix(sdm): deduplicate copied bench measurements atomically

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

@@ -90,7 +90,7 @@ Build the SignalEngine, run the backtest, and link results.
    - 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. 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`.
+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 the same measurement already stored (even in another directory)) 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".
 

agent/src/strategy_store/sqlite_store.py

@@ -650,17 +650,33 @@ class SqliteStrategyStore:
     @_synchronized
     def record_bench(self, result: BenchResult) -> int:
         """Record a bench result.  Returns the result ID."""
+        with self._write_transaction():
+            return self._insert_bench(result)
+
+    def _insert_bench(self, result: BenchResult) -> int:
+        """Insert within the caller's transaction."""
         values = self._bench_values(result)
         values["created_at"] = result.created_at or _now_iso()
         columns = ", ".join(values)
         placeholders = ", ".join("?" for _ in values)
 
+        cursor = self._conn.execute(
+            f"INSERT INTO bench_history ({columns}) VALUES ({placeholders})",
+            list(values.values()),
+        )
+        return cursor.lastrowid  # type: ignore[return-value]
+
+    @_synchronized
+    def record_bench_once(self, result: BenchResult) -> int | None:
+        """Atomically record a measurement, returning None for a duplicate.
+
+        BEGIN IMMEDIATE serializes this check and insert across connections.
+        Directory, timestamp, bench label and grading do not change evidence.
+        """
         with self._write_transaction():
-            cursor = self._conn.execute(
-                f"INSERT INTO bench_history ({columns}) VALUES ({placeholders})",
-                list(values.values()),
-            )
-            return cursor.lastrowid  # type: ignore[return-value]
+            if self.has_bench_result(result):
+                return None
+            return self._insert_bench(result)
 
     @_synchronized
     def get_bench_history(
@@ -709,13 +725,15 @@ class SqliteStrategyStore:
         metric absent from both rows still counts as a match.
 
         Args:
-            result: Candidate bench result; ``id`` and ``created_at`` are
-                ignored.
+            result: Candidate measurement; location, time and grading metadata
+                are ignored.
 
         Returns:
-            ``True`` when a stored row matches ``result`` on every other field.
+            ``True`` when a stored measurement has the same artifact, windows and metrics.
         """
         values = self._bench_values(result)
+        for metadata in ("run_dir", "bench_type", "category"):
+            values.pop(metadata)
         where = " AND ".join(f"{column} IS ?" for column in values)
         row = self._conn.execute(
             f"SELECT 1 FROM bench_history WHERE {where} LIMIT 1",

agent/src/strategy_store/store.py

@@ -12,7 +12,6 @@ backends.
 
 from __future__ import annotations
 
-import copy
 import threading
 import uuid
 from dataclasses import replace
@@ -27,7 +26,6 @@ from src.strategy_store.models import (
     ArtifactType,
     BenchResult,
     DecaySnapshot,
-    DecaySignal,
     ValidationStatus,
     validate_validation_status_transition,
 )
@@ -116,6 +114,10 @@ class StrategyStoreProtocol(Protocol):
         """Record a bench result. Returns the result ID."""
         ...
 
+    def record_bench_once(self, result: BenchResult) -> int | None:
+        """Atomically record new evidence; ignore location and grading metadata."""
+        ...
+
     def get_bench_history(
         self,
         artifact_id: str,
@@ -139,8 +141,8 @@ class StrategyStoreProtocol(Protocol):
     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.
+        Identity uses the artifact, evaluation windows and measured metrics.
+        IDs, timestamps, directories, bench labels and grading are metadata.
 
         Args:
             result: Candidate bench result.
@@ -325,6 +327,13 @@ class InMemoryStrategyStore:
         self._bench_results.append(stored)
         return stored.id
 
+    @_synchronized
+    def record_bench_once(self, result: BenchResult) -> int | None:
+        """Record new evidence once under the same lock as the duplicate check."""
+        if self.has_bench_result(result):
+            return None
+        return self.record_bench(result)
+
     @_synchronized
     def get_bench_history(
         self,
@@ -359,15 +368,15 @@ class InMemoryStrategyStore:
         """Return whether an identical bench row is already stored.
 
         Args:
-            result: Candidate bench result; ``id`` and ``created_at`` are
-                ignored.
+            result: Candidate measurement; location, time and grading metadata
+                are ignored.
 
         Returns:
-            ``True`` when a stored row matches ``result`` on every other field.
+            ``True`` when a stored measurement has the same artifact, windows and metrics.
         """
-        probe = replace(result, id=None, created_at="")
+        probe = replace(result, id=None, created_at="", run_dir=None, bench_type="", category=None)
         return any(
-            replace(stored, id=None, created_at="") == probe
+            replace(stored, id=None, created_at="", run_dir=None, bench_type="", category=None) == probe
             for stored in self._bench_results
         )
 

agent/src/tools/backtest_tool.py

@@ -116,8 +116,8 @@ def _record_bench_if_declared(
     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
+    measurement is skipped atomically, even from another directory or
+    concurrent process, so copying a run cannot count as independent evidence; a rerun of the same directory whose metrics or window
     changed is a new measurement and is recorded.
 
     Args:
@@ -183,9 +183,8 @@ def _record_bench_if_declared(
             run_dir=str(run_path),
             **metrics,
         )
-        if store.has_bench_result(result):
+        if store.record_bench_once(result) is None:
             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}")

agent/src/tools/trade_journal_parsers.py

@@ -17,6 +17,8 @@ from typing import Any
 
 import pandas as pd
 
+from src.market_data import FIAT_CODES as _FIAT_CODES
+
 # Broker CSV/Excel cells often include ISO codes or currency glyphs around the
 # number (Schwab/IBKR "$1,234.56", JP/CN "¥1000"). Commas are already stripped;
 # without stripping these tokens float() fails and we silently store 0.0.
@@ -24,9 +26,6 @@ _CURRENCY_TOKEN_RE = re.compile(
     r"(?i)(?<![A-Za-z])(?:USDT|USDC|USD|EUR|GBP|JPY|CNY|HKD)(?![A-Za-z])|[$€£¥￥]"
 )
 
-# ISO codes of the fiat currencies a broker export can quote as a pair
-# (EUR/USD); a pair with two fiat legs is forex, never crypto.
-from src.market_data import FIAT_CODES as _FIAT_CODES
 
 FormatName = str  # "tonghuashun" | "eastmoney" | "futu" | "generic" | "unknown"
 

agent/tests/test_backtest_records_bench.py

@@ -12,6 +12,7 @@ from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Any
 from unittest.mock import patch
+from concurrent.futures import ThreadPoolExecutor
 
 import pytest
 
@@ -231,6 +232,24 @@ def test_identical_rerun_is_not_recorded_twice(
     assert len(store.get_bench_history(artifact_id)) == 1
 
 
+def test_copied_runs_cannot_meet_the_decay_floor(run_root, store, artifact_id):
+    for index in range(3):
+        copied = _make_run_dir(run_root, f"copy-{index}")
+        _write_config(copied, artifact_id=artifact_id, bench_type="periodic")
+        result = _run(copied)
+        assert result["bench_record"]["status"] == ("recorded" if index == 0 else "skipped")
+    assert len(store.get_bench_history(artifact_id)) == 1
+    assert _decay_check(artifact_id)["signal"] == "insufficient_data"
+
+
+def test_same_metrics_on_a_new_window_are_new_evidence(run_root, store, artifact_id):
+    for index in range(2):
+        run = _make_run_dir(run_root, f"window-{index}")
+        _write_config(run, artifact_id=artifact_id, bench_type="periodic", end_date=f"2026-09-{25 + index}")
+        assert _run(run)["bench_record"]["status"] == "recorded"
+    assert len(store.get_bench_history(artifact_id)) == 2
+
+
 def test_rerun_with_changed_metrics_is_recorded(
     run_dir: Path, store: Any, artifact_id: str
 ) -> None:
@@ -327,7 +346,7 @@ def test_store_failure_is_reported_without_failing_the_backtest(
     _write_config(run_dir, artifact_id=artifact_id)
 
     with patch(
-        "src.strategy_store.sqlite_store.SqliteStrategyStore.record_bench",
+        "src.strategy_store.sqlite_store.SqliteStrategyStore.record_bench_once",
         side_effect=RuntimeError("disk full"),
     ):
         envelope = _run(run_dir)
@@ -378,8 +397,8 @@ def test_has_bench_result_matches_only_identical_rows(backend: Any) -> None:
     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"))
+    assert backend.has_bench_result(replace(stored, run_dir="/runs/b"))
+    assert backend.has_bench_result(replace(stored, bench_type="periodic"))
 
 
 def test_bench_history_filters_by_type_before_limit(backend: Any) -> None:
@@ -397,3 +416,63 @@ def test_bench_history_filters_by_type_before_limit(backend: Any) -> None:
 
     assert [row.sharpe for row in rows] == [1.0]
     assert len(backend.get_bench_history(artifact_id)) == 2
+
+
+def test_concurrent_copies_are_reserved_once(backend):
+    artifact_id = _register(backend)
+    stores = ([SqliteStrategyStore(backend.db_path) for _ in range(8)]
+              if isinstance(backend, SqliteStrategyStore) else [backend] * 8)
+    def write(index):
+        return stores[index].record_bench_once(BenchResult(
+            artifact_id=artifact_id, bench_type="periodic", sharpe=0.5,
+            test_start="2026-01-01", test_end="2026-09-30", run_dir=f"/runs/{index}",
+        ))
+    try:
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            ids = list(pool.map(write, range(8)))
+        assert sum(value is not None for value in ids) == 1
+        assert len(backend.get_bench_history(artifact_id)) == 1
+    finally:
+        if isinstance(backend, SqliteStrategyStore):
+            for store in stores:
+                store._conn.close()
+
+
+def test_real_local_perpetual_backtest_records_one_measurement(run_root, store, artifact_id, monkeypatch, tmp_path):
+    """Exercise real tool -> subprocess -> crypto engine -> bench persistence."""
+    import pandas as pd
+    import shutil
+    from src.core import runner as core_runner
+
+    fixture_home = tmp_path / "data-home"
+    bridge = fixture_home / ".vibe-trading" / "data-bridge"
+    bridge.mkdir(parents=True)
+    prices = bridge / "prices.csv"
+    dates = pd.date_range("2026-01-01", periods=40)
+    pd.DataFrame({"date": dates, "open": [100 + i for i in range(40)],
+                  "high": [102 + i for i in range(40)], "low": [99 + i for i in range(40)],
+                  "close": [101 + i for i in range(40)], "volume": 100000}).to_csv(prices, index=False)
+    (bridge / "config.yaml").write_text(json.dumps({"sources": [
+        {"symbol": "1000PEPE-USDT-PERP", "type": "csv", "path": str(prices)},
+    ]}))
+    prepare = core_runner._prepare_sandbox_home
+    monkeypatch.setattr(core_runner, "_prepare_sandbox_home", lambda _home: prepare(fixture_home))
+    run = _make_run_dir(run_root, "real-engine")
+    _write_config(run, artifact_id=artifact_id, bench_type="periodic", source="local",
+                  codes=["1000PEPE-USDT-PERP"], start_date="2026-01-01", end_date="2026-02-09", initial_cash=10000)
+    (run / "code" / "signal_engine.py").write_text(
+        'import pandas as pd\nclass SignalEngine:\n'
+        '    def __init__(self, **kwargs): pass\n'
+        '    def generate(self, data_map):\n'
+        '        return {k: pd.Series([1 if i % 10 < 5 else -1 for i in range(len(v))], index=v.index) for k,v in data_map.items()}\n'
+    )
+    outcome = json.loads(run_backtest(str(run)))
+    assert outcome["status"] == "ok", outcome
+    assert outcome["bench_record"]["status"] == "recorded", outcome
+    assert pd.read_csv(run / "artifacts" / "metrics.csv").iloc[0]["trade_count"] > 0
+    copied = run_root / "real-copy"
+    shutil.copytree(run, copied)
+    repeated = json.loads(run_backtest(str(copied)))
+    assert repeated["status"] == "ok", repeated
+    assert repeated["bench_record"]["status"] == "skipped", repeated
+    assert len(store.get_bench_history(artifact_id)) == 1