fix(cost): close only latest open run per project on end_experiment (#537)

The cost schema allows multiple ``runs`` rows per ``project_id``
(retries, repeated traversals — ``run_id`` is the PK, project views
group/list runs). The bulk close in
``close_open_runs_for_project`` stamped every historical open row
for the project with the just-finished experiment's ``ended_at``
and final counters, corrupting older runs.

Replace with ``close_latest_open_run_for_project``: scope to the
``MAX(started_at)`` open run only. Older open rows are left to the
periodic ``backfill_run_close_state.py`` script, which derives
``ended_at`` from each row's own ``MAX(token_events.timestamp)`` —
the on-disk approximation that's correct per-run rather than a
batch-overwrite from the live hook.

Return type tightened from ``int`` (count) to ``Optional[str]``
(the closed run_id) so callers can log which row was touched.

Caught by Codex P2 on PR #538.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: lwgray

src/cost_tracking/cost_store.py

@@ -992,7 +992,7 @@ def close_run(
         self.conn.commit()
         return cur.rowcount > 0
 
-    def close_open_runs_for_project(
+    def close_latest_open_run_for_project(
         self,
         project_id: str,
         *,
@@ -1001,46 +1001,59 @@ def close_open_runs_for_project(
         completed_tasks: Optional[int] = None,
         blocked_tasks: Optional[int] = None,
         num_agents: Optional[int] = None,
-    ) -> int:
-        """Close every open run attributed to one project (Marcus #537).
+    ) -> Optional[str]:
+        """Close the most recently started open run for a project (#537).
+
+        Live-close hook for ``end_experiment``. The schema allows
+        multiple ``runs`` rows per ``project_id`` (retries, repeated
+        traversals), so a bulk close would stamp every historical
+        open row with the just-finished experiment's
+        ``ended_at`` / counters — corrupting older runs.
 
-        Wraps :meth:`close_run` over every ``runs`` row where
-        ``project_id`` matches and ``ended_at IS NULL``. Used as the
-        live-close hook from ``end_experiment``, where we know the
-        monitor's project_id but not the cost-tracking run_id.
+        Scope the close to the latest open run, identified by
+        ``MAX(started_at)``. Older open rows for the same project are
+        left to the periodic ``backfill_run_close_state.py``, which
+        derives ``ended_at`` from each row's own
+        ``MAX(token_events.timestamp)``.
 
         The incoming ``project_id`` is normalized via
         :func:`src.cost_tracking.cost_recorder.canonical_project_id`
         so callers can pass either dashed UUID form (ProjectRegistry)
         or dashless hex form (SQLiteKanban auto-discovery) — both
         resolve to the dashless form actually stored in ``runs``.
-        Without this, the live close from ``end_experiment`` silently
-        no-ops whenever the monitor was handed a dashed project_id.
+        Without this, the live close silently no-ops whenever the
+        monitor was handed a dashed project_id.
 
         Returns
         -------
-        int
-            Number of runs closed.
+        Optional[str]
+            The closed ``run_id``, or ``None`` if the project had no
+            open runs.
         """
         from src.cost_tracking.cost_recorder import canonical_project_id
 
         normalized = canonical_project_id(project_id) or project_id
-        rows = self.conn.execute(
-            "SELECT run_id FROM runs WHERE project_id = ? AND ended_at IS NULL",
+        row = self.conn.execute(
+            """
+            SELECT run_id FROM runs
+             WHERE project_id = ? AND ended_at IS NULL
+             ORDER BY started_at DESC
+             LIMIT 1
+            """,
             (normalized,),
-        ).fetchall()
-        closed = 0
-        for (run_id,) in rows:
-            if self.close_run(
-                run_id,
-                ended_at=ended_at,
-                total_tasks=total_tasks,
-                completed_tasks=completed_tasks,
-                blocked_tasks=blocked_tasks,
-                num_agents=num_agents,
-            ):
-                closed += 1
-        return closed
+        ).fetchone()
+        if row is None:
+            return None
+        (run_id,) = row
+        closed = self.close_run(
+            run_id,
+            ended_at=ended_at,
+            total_tasks=total_tasks,
+            completed_tasks=completed_tasks,
+            blocked_tasks=blocked_tasks,
+            num_agents=num_agents,
+        )
+        return run_id if closed else None
 
     def record_price(self, price: ModelPrice) -> None:
         """Insert one ``model_prices`` row.

src/marcus_mcp/tools/experiments.py

@@ -252,18 +252,18 @@ async def end_experiment() -> Dict[str, Any]:
 
                 _cost_store = get_recorder().store
                 final = result.get("final_metrics", {}) or {}
-                closed = _cost_store.close_open_runs_for_project(
+                closed_run_id = _cost_store.close_latest_open_run_for_project(
                     project_id=project_id,
                     ended_at=datetime.now(timezone.utc),
                     total_tasks=final.get("total_tasks"),
                     completed_tasks=final.get("total_task_completions"),
                     blocked_tasks=final.get("total_blockers"),
                     num_agents=final.get("total_registered_agents"),
                 )
-                if closed > 0:
+                if closed_run_id:
                     logger.info(
-                        "Closed %d open run(s) for project %s on end_experiment",
-                        closed,
+                        "Closed run %s for project %s on end_experiment",
+                        closed_run_id,
                         project_id,
                     )
             except Exception as _close_err:  # pragma: no cover

tests/unit/cost_tracking/test_cost_store.py

@@ -1034,22 +1034,29 @@ def test_none_fields_preserve_existing_values(self, store: CostStore) -> None:
         assert row[0] == 15  # original value preserved
 
 
-class TestCloseOpenRunsForProject:
-    """``close_open_runs_for_project`` is the bulk-close helper (#537)."""
+class TestCloseLatestOpenRunForProject:
+    """``close_latest_open_run_for_project`` is the live-close helper (#537)."""
 
-    def test_closes_all_open_runs_for_project(self, store: CostStore) -> None:
-        """Two open runs for proj_x both get closed; closed runs are untouched."""
+    def test_closes_only_latest_open_run(self, store: CostStore) -> None:
+        """Older open run is preserved; only MAX(started_at) is stamped.
+
+        Regression for Codex P2 on PR #538: the schema allows multiple
+        open runs per project (retries, repeated traversals). A bulk
+        close would stamp historical rows with the just-finished
+        experiment's ended_at + counters, corrupting them. The live
+        hook must scope to one row.
+        """
         store.record_run(
             Run(
-                run_id="r1",
+                run_id="r_old",
                 project_id="proj_x",
                 project_name="x",
                 started_at=datetime(2026, 5, 13, 10, tzinfo=timezone.utc),
             )
         )
         store.record_run(
             Run(
-                run_id="r2",
+                run_id="r_new",
                 project_id="proj_x",
                 project_name="x",
                 started_at=datetime(2026, 5, 13, 11, tzinfo=timezone.utc),
@@ -1058,24 +1065,49 @@ def test_closes_all_open_runs_for_project(self, store: CostStore) -> None:
         # Different project — must NOT be touched.
         store.record_run(
             Run(
-                run_id="r3",
+                run_id="r_other",
                 project_id="other",
                 project_name="o",
                 started_at=datetime(2026, 5, 13, 10, tzinfo=timezone.utc),
             )
         )
 
         end = datetime(2026, 5, 13, 12, tzinfo=timezone.utc)
-        closed = store.close_open_runs_for_project("proj_x", ended_at=end)
-        assert closed == 2
+        closed_id = store.close_latest_open_run_for_project(
+            "proj_x", ended_at=end, total_tasks=8
+        )
+        assert closed_id == "r_new"
+
+        rows = {
+            run_id: (ended_at, total_tasks)
+            for run_id, ended_at, total_tasks in store.conn.execute(
+                "SELECT run_id, ended_at, total_tasks FROM runs ORDER BY run_id"
+            )
+        }
+        # Latest stamped with the experiment's counters.
+        assert rows["r_new"] == (end.isoformat(), 8)
+        # Older row left untouched for the periodic backfill.
+        assert rows["r_old"] == (None, None)
+        # Different project untouched.
+        assert rows["r_other"] == (None, None)
+
+    def test_returns_none_when_no_open_runs(self, store: CostStore) -> None:
+        """Empty / fully-closed project returns None — caller treats as no-op."""
+        end = datetime(2026, 5, 13, 12, tzinfo=timezone.utc)
+        # Project never recorded.
+        assert store.close_latest_open_run_for_project("ghost", ended_at=end) is None
 
-        # proj_x's runs are closed; other's stays open.
-        rows = dict(
-            store.conn.execute("SELECT run_id, ended_at FROM runs ORDER BY run_id")
+        # Project with a row that's already closed.
+        store.record_run(
+            Run(
+                run_id="r_done",
+                project_id="proj_y",
+                project_name="y",
+                started_at=datetime(2026, 5, 13, 10, tzinfo=timezone.utc),
+                ended_at=datetime(2026, 5, 13, 11, tzinfo=timezone.utc),
+            )
         )
-        assert rows["r1"] == end.isoformat()
-        assert rows["r2"] == end.isoformat()
-        assert rows["r3"] is None
+        assert store.close_latest_open_run_for_project("proj_y", ended_at=end) is None
 
     def test_dashed_project_id_canonicalizes_to_dashless(
         self, store: CostStore
@@ -1103,8 +1135,8 @@ def test_dashed_project_id_canonicalizes_to_dashless(
 
         end = datetime(2026, 5, 13, 12, tzinfo=timezone.utc)
         # Caller hands in the dashed form — must still close the row.
-        closed = store.close_open_runs_for_project(dashed, ended_at=end)
-        assert closed == 1
+        closed_id = store.close_latest_open_run_for_project(dashed, ended_at=end)
+        assert closed_id == "r_canon"
 
         row = store.conn.execute(
             "SELECT ended_at FROM runs WHERE run_id = ?", ("r_canon",)