lwgray
diff --git a/‎scripts/backfill_run_close_state.py‎
Lines changed: 103 additions & 0 deletions b/‎scripts/backfill_run_close_state.py‎
Lines changed: 103 additions & 0 deletions
diff --git a/‎src/cost_tracking/cost_aggregator.py‎
Lines changed: 39 additions & 4 deletions b/‎src/cost_tracking/cost_aggregator.py‎
Lines changed: 39 additions & 4 deletions
diff --git a/‎src/cost_tracking/cost_store.py‎
Lines changed: 150 additions & 0 deletions b/‎src/cost_tracking/cost_store.py‎
Lines changed: 150 additions & 0 deletions
diff --git a/‎src/marcus_mcp/coordinator/subtask_assignment.py‎
Lines changed: 33 additions & 0 deletions b/‎src/marcus_mcp/coordinator/subtask_assignment.py‎
Lines changed: 33 additions & 0 deletions
@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""
+One-shot backfill of ``runs.ended_at`` from ``token_events.timestamp``.
+
+The ``runs`` table has lifecycle columns (``ended_at``, ``total_tasks``,
+``completed_tasks``, etc.) that the schema clearly anticipates, but
+until Marcus #537 no Python code ever wrote them. Every row in ``runs``
+has been "open" since insertion. This blocks any dashboard query
+filtering on ``ended_at IS NOT NULL`` and prevents wall-clock-time
+analysis (cost-per-minute, coordination tax rate, run duration).
+
+Going forward, ``end_experiment`` closes runs live for the
+``path=marcus`` and ``path=posidonius`` flows. This backfill closes
+the rest: historical runs that pre-date the fix, and runs from
+``path=direct`` usage where the user never invokes ``end_experiment``.
+
+Heuristic
+---------
+For each ``runs`` row where ``ended_at IS NULL``, set ``ended_at`` to
+``MAX(timestamp) FROM token_events WHERE run_id = ?`` — the timestamp
+of the last LLM call attributed to that run. That's the best on-disk
+approximation we have for unattended close. Runs with zero token
+events are left open (genuinely empty runs are ambiguous and likely
+indicate the run died before any LLM call).
+
+Idempotent: re-runnable. Rows with ``ended_at`` already set are
+skipped (the CostStore's UPDATE only fills NULL fields).
+
+Usage
+-----
+.. code-block:: console
+
+    $ python scripts/backfill_run_close_state.py
+    Closed 23 runs (10 had zero events and were skipped).
+
+Pass ``--costs-db PATH`` to override the cost-store path (defaults
+to ``~/.marcus/costs.db``).
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+from typing import Tuple
+
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(_REPO_ROOT))
+
+from src.cost_tracking.cost_store import CostStore  # noqa: E402
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    parser.add_argument(
+        "--costs-db",
+        type=Path,
+        default=Path.home() / ".marcus" / "costs.db",
+        help="Path to costs.db (default: ~/.marcus/costs.db).",
+    )
+    return parser.parse_args()
+
+
+def backfill(costs_db: Path) -> Tuple[int, int]:
+    """Close every open run whose token_events have a max timestamp.
+
+    Returns
+    -------
+    (closed, skipped) : tuple of int
+        Counts for runs successfully closed and runs left open
+        because they had zero token_events.
+    """
+    store = CostStore(db_path=costs_db)
+
+    open_rows = store.conn.execute(
+        "SELECT run_id FROM runs WHERE ended_at IS NULL"
+    ).fetchall()
+
+    closed = 0
+    skipped = 0
+    for (run_id,) in open_rows:
+        # ``close_run`` returns False when there are no events to
+        # derive ended_at from — those runs stay open.
+        if store.close_run(run_id):
+            closed += 1
+        else:
+            skipped += 1
+    return closed, skipped
+
+
+def main() -> int:
+    """Parse CLI args, run the backfill, and print a summary."""
+    args = _parse_args()
+    closed, skipped = backfill(args.costs_db)
+    print(
+        f"Closed {closed} runs"
+        + (f" ({skipped} had zero events and were skipped)." if skipped > 0 else ".")
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -361,24 +361,59 @@ def run_audit(self, run_id: str) -> Dict[str, Any]:
         or ``agent_id``. A healthy audit shows ``reconciles=True`` and
         zero ``worker_events_without_task_id``.
 
+        Also reports run-lifecycle status (Marcus #537): ``run_open``
+        is True when the run's ``ended_at`` is NULL — i.e. the row
+        was written at create_project time but never closed. A
+        complete audit cares about both token reconciliation AND
+        lifecycle closure.
+
         Returns
         -------
         dict
             ``total_events``, ``total_tokens``, ``by_role_total_tokens``,
             ``reconciles`` (bool), ``tokens_outside_known_roles``,
             ``planner_events``, ``worker_events``,
             ``worker_events_without_task_id``,
-            ``worker_events_without_agent_id``.
+            ``worker_events_without_agent_id``, ``run_open`` (bool).
         """
-        return self._audit_query("run_id = ?", (run_id,))
+        audit = self._audit_query("run_id = ?", (run_id,))
+        row = self._row("SELECT ended_at FROM runs WHERE run_id = ?", (run_id,))
+        # ``run_open`` is True only when a run exists and has no
+        # ended_at. An unknown run_id yields False to avoid noise.
+        audit["run_open"] = row is not None and row.get("ended_at") is None
+        return audit
 
     def project_audit(self, project_id: str) -> Dict[str, Any]:
         """Token-attribution audit for one project (Marcus #527).
 
         Like :meth:`run_audit` but scoped to all runs for one
-        ``project_id``. Same return shape — see that docstring.
+        ``project_id``. Reports the same token-attribution fields plus
+        project-level lifecycle counts (Marcus #537): ``runs_total``
+        and ``runs_open`` for the runs attributed to this project.
+
+        Returns
+        -------
+        dict
+            All fields from :meth:`run_audit` minus ``run_open``, plus
+            ``runs_total`` and ``runs_open``.
         """
-        return self._audit_query("project_id = ?", (project_id,))
+        audit = self._audit_query("project_id = ?", (project_id,))
+        row = (
+            self._row(
+                """
+            SELECT COUNT(*) AS total,
+                   COALESCE(SUM(CASE WHEN ended_at IS NULL THEN 1 ELSE 0 END), 0)
+                                                                AS open_count
+            FROM runs
+            WHERE project_id = ?
+            """,
+                (project_id,),
+            )
+            or {"total": 0, "open_count": 0}
+        )
+        audit["runs_total"] = int(row.get("total", 0) or 0)
+        audit["runs_open"] = int(row.get("open_count", 0) or 0)
+        return audit
 
     def cache_hit_rate_by_agent(self, run_id: str) -> List[Dict[str, Any]]:
         """Per-agent cache hit rate within one experiment.
 
@@ -905,6 +905,156 @@ def record_run(self, run: Run) -> None:
         )
         self.conn.commit()
 
+    def close_run(
+        self,
+        run_id: str,
+        *,
+        ended_at: Optional[datetime] = None,
+        total_tasks: Optional[int] = None,
+        completed_tasks: Optional[int] = None,
+        blocked_tasks: Optional[int] = None,
+        num_agents: Optional[int] = None,
+    ) -> bool:
+        """Stamp final-state lifecycle fields on a run (Marcus #537).
+
+        Closes the data-quality gap where every ``runs`` row stayed
+        open forever because the only writer
+        (``create_project`` → ``record_run``) only wrote the open
+        state. The dataclass and UPSERT have always supported the
+        close fields; this method is the missing caller.
+
+        When ``ended_at`` is ``None``, falls back to the timestamp of
+        the most recent ``token_events`` row attributed to this run —
+        the best on-disk approximation for unattended close
+        (``path=direct`` usage where the user never calls
+        ``end_experiment``). When no events exist for the run, the
+        method returns False without updating, since "closed with no
+        activity" is ambiguous and likely indicates the run died
+        before any LLM call.
+
+        Uses a targeted UPDATE rather than the upsert path so passing
+        ``None`` for a field leaves the existing column value
+        untouched (COALESCE-guarded). Safe to re-run; idempotent on
+        already-closed rows.
+
+        Parameters
+        ----------
+        run_id : str
+            The run to close.
+        ended_at : datetime, optional
+            Wall-clock end time. When None, derives from the run's
+            last event timestamp.
+        total_tasks, completed_tasks, blocked_tasks, num_agents : int, optional
+            Final lifecycle counts. None leaves the existing value.
+
+        Returns
+        -------
+        bool
+            True if a row was updated, False if the run_id was
+            unknown or had no events to back into ``ended_at``.
+        """
+        # Resolve ended_at via the last event timestamp when the
+        # caller didn't supply one. This is the unattended-close path
+        # — backfill scripts and direct-MCP runs land here.
+        resolved_ended_at_iso: Optional[str]
+        if ended_at is not None:
+            resolved_ended_at_iso = ended_at.isoformat()
+        else:
+            row = self.conn.execute(
+                "SELECT MAX(timestamp) FROM token_events WHERE run_id = ?",
+                (run_id,),
+            ).fetchone()
+            if row is None or row[0] is None:
+                # No events for this run — can't infer ended_at.
+                # Caller should pass an explicit value or skip.
+                return False
+            resolved_ended_at_iso = str(row[0])
+
+        cur = self.conn.execute(
+            """
+            UPDATE runs
+               SET ended_at        = COALESCE(?, ended_at),
+                   total_tasks     = COALESCE(?, total_tasks),
+                   completed_tasks = COALESCE(?, completed_tasks),
+                   blocked_tasks   = COALESCE(?, blocked_tasks),
+                   num_agents      = COALESCE(?, num_agents)
+             WHERE run_id = ?
+            """,
+            (
+                resolved_ended_at_iso,
+                total_tasks,
+                completed_tasks,
+                blocked_tasks,
+                num_agents,
+                run_id,
+            ),
+        )
+        self.conn.commit()
+        return cur.rowcount > 0
+
+    def close_latest_open_run_for_project(
+        self,
+        project_id: str,
+        *,
+        ended_at: Optional[datetime] = None,
+        total_tasks: Optional[int] = None,
+        completed_tasks: Optional[int] = None,
+        blocked_tasks: Optional[int] = None,
+        num_agents: Optional[int] = None,
+    ) -> 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.
+
+        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 silently no-ops whenever the
+        monitor was handed a dashed project_id.
+
+        Returns
+        -------
+        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
+        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,),
+        ).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.
 
 
@@ -328,6 +328,39 @@ async def check_and_complete_parent_task(
         # task to the agent that drove it to completion (Bug 3 fix).
         # Without this, the parent appears as a ghost completion with no
         # agent record — invisible in experiment analytics.
+        #
+        # DELIBERATELY NOT WRITING task_outcomes HERE — see contract below.
+        #
+        # Auto-completion contract (do not change without an ADR):
+        #
+        #   Parents-with-subtasks are coordination containers. The work data
+        #   (started_at, completed_at, actual_hours, agent_id) lives on the
+        #   subtask outcomes — which are real per-agent records written via
+        #   Memory.record_task_completion when each subtask finishes.
+        #
+        #   Writing a synthetic parent outcome here would corrupt agent
+        #   learning:
+        #     * Memory._update_agent_profile (memory.py) ticks
+        #       total_tasks + skill_success_rates EMA per outcome row;
+        #       a parent row attributed to completing_agent_id would
+        #       double-count their work.
+        #     * analysis.aggregator._build_agent_histories sums
+        #       outcome.actual_hours into total_hours, also double-counting.
+        #     * Scheduler tests (test_scheduler_prototype_mode.py) already
+        #       assert "parent with subtasks is a container — should be
+        #       excluded" from CPM. Outcome accounting must match that.
+        #
+        #   Contrast with About/design auto-completion in nlp_tools.py:
+        #   those write task_outcomes with agent_id="system" or "Marcus",
+        #   synthetic IDs that don't correspond to a real worker and so
+        #   don't pollute learning. Those tasks are LEAVES with no
+        #   subtasks — the outcome write is the only place their data
+        #   lives. Subtask-driven parents are DERIVATIVE — the data
+        #   already lives on the subtasks.
+        #
+        #   Downstream observers (Cato) aggregate subtask data for parent
+        #   display rather than relying on a parent outcome row. See
+        #   cato_src/core/aggregator._apply_subtask_rollup.
         if completing_agent_id and state and hasattr(state, "log_event"):
             state.log_event(
                 "task_assignment",