feat(cost): task name snapshot pattern + backfill (#530) (#536)

* feat(cost): task name snapshot pattern + backfill (#530)

Cato dashboard's Tokens by task panel currently shows opaque hex
IDs (be31d8c7…) instead of real kanban task names. Mirrors the
project_names snapshot pattern from PR #515.

Schema (cost_store.py)
- New task_names table: task_id PRIMARY KEY, name, first_seen,
  last_seen. Created via SCHEMA_SQL with IF NOT EXISTS — no
  migration script needed beyond a fresh CREATE on existing DBs.
- New record_task_name(task_id, name) + get_task_name(task_id)
  on CostStore. UPSERT-on-conflict semantics, identical shape
  to upsert_project_name.

Decomposer wiring (integrations/nlp_tools.py)
- The decomposer's tasks-with-real-ids loop now snapshots each
  task's name into costs.db::task_names. Failure is logged and
  swallowed — cost tracking is optional and must never block
  project creation.

Aggregator (cost_aggregator.py)
- by_task in run_summary and project_summary now LEFT JOINs
  task_names and emits task_name on every slice. NULL when the
  task_id was never snapshotted.

Backfill (scripts/backfill_task_names.py)
- One-shot script reading marcus.db::task_metadata into
  costs.db::task_names. Two passes:
  1. Parent tasks: direct copy of (task_id, name).
  2. Subtasks: synthesize "<parent name> — subtask N" for IDs
     like <parent>_sub_N that appear in token_events but not in
     task_metadata.
- Idempotent. UPSERT.

Real-data result on a 68,974-row costs.db:
  Backfilled 292 parent task names from data/marcus.db
  Backfilled 157 subtask names derived from parents
  task_id match coverage in token_events: 220 / 893 (was 0 / 893)

Tests: 4 in test_cost_store.py (TestTaskNamesSnapshot) + 2 in
test_cost_aggregator.py (TestTaskNamesJoin).

Refs: #527, #530, #532 (About + Planning task name follow-up)

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

* fix(cost): move task-name snapshot into shared create_tasks_on_board (Codex P2)

Codex P2 review on #536: the original Phase 3 wiring put the
record_task_name call inside
NaturalLanguageProjectCreator.create_project_from_description only.
The feature-adder flow
(NaturalLanguageFeatureAdder.add_feature_from_description) also
creates kanban tasks through the same
NaturalLanguageTaskCreator.create_tasks_on_board shared method but
never executed the snapshot block. Result: agents working on
feature-added tasks would render as opaque hex IDs in the Cato
dashboard, defeating Phase 3 for that flow.

Move the snapshot into the shared create_tasks_on_board in
src/integrations/nlp_base.py, right after the existing
task_metadata persistence write. Both flows now covered
automatically; the project-creator-specific snapshot block in
nlp_tools.py is now redundant and removed.

Verifying the recorder failure path is non-fatal — cost tracking
is optional and must never block kanban task creation.

Tests: 2 new in tests/unit/integrations/test_task_name_snapshot.py.
- test_record_task_name_called_for_each_created_task: exercises
  the shared method with a mock kanban client, asserts every
  created task lands in task_names with the right (id, name) pair.
- test_snapshot_failure_does_not_break_task_creation: broken
  recorder doesn't propagate into kanban creation.

Baseline: 1449 pass under pytest -m unit. mypy clean.

Refs: #536 Codex review, #530

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

---------

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

Author: lwgray

scripts/backfill_task_names.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""
+One-shot backfill of ``task_names`` from ``marcus.db::task_metadata``.
+
+Marcus's `marcus.db` (collection ``task_metadata`` inside the
+key-value ``persistence`` table) is the source of truth for task
+names. The cost-tracking dashboard reads from `costs.db::task_names`.
+After Marcus #530 added the snapshot path, new tasks land in both
+DBs at creation time. Historical tasks created before #530 only
+exist in `marcus.db`; the dashboard renders their opaque hex IDs.
+
+This script reads every row in the ``task_metadata`` collection,
+extracts ``(task_id, name)``, and calls
+:meth:`CostStore.record_task_name` for each. The store's UPSERT
+makes the operation idempotent — re-running is safe and a no-op
+when nothing has changed.
+
+Usage
+-----
+.. code-block:: console
+
+    $ python scripts/backfill_task_names.py
+    Backfilled 287 task names from /Users/.../data/marcus.db
+    Skipped 5 rows (malformed JSON or missing name)
+
+Pass ``--marcus-db PATH`` to point at a non-default Marcus DB.
+Pass ``--costs-db PATH`` to override the cost-store path (defaults
+to ``~/.marcus/costs.db``).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Tuple
+
+# Allow running directly from the repo root without installing.
+_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(
+        "--marcus-db",
+        type=Path,
+        default=_REPO_ROOT / "data" / "marcus.db",
+        help="Path to Marcus's marcus.db (default: ./data/marcus.db).",
+    )
+    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(marcus_db: Path, costs_db: Path) -> Tuple[int, int, int]:
+    """Copy ``(task_id, name)`` from ``marcus.db`` into ``costs.db``.
+
+    Two passes:
+
+    1. **Parent tasks.** Read every row in ``marcus.db::task_metadata``
+       and snapshot its ``(task_id, name)`` into ``task_names``.
+    2. **Subtasks.** Marcus decomposes some tasks into subtasks with
+       IDs like ``<parent_hex>_sub_1`` / ``_sub_2``. Those IDs land in
+       ``token_events.task_id`` but never appear in ``task_metadata``
+       (only parents do). Walk every distinct ``_sub_N`` ID in
+       ``token_events``, look up the parent's snapshotted name, and
+       write a derived label (``"<parent name> — subtask N"``) so the
+       dashboard surfaces something readable instead of a hex tail.
+
+    Returns
+    -------
+    (parents, subtasks, skipped) : tuple of int
+        Counts for parent-task names recorded, subtask-derived names
+        recorded, and rows skipped (malformed JSON, missing name).
+    """
+    import sqlite3
+
+    if not marcus_db.exists():
+        raise FileNotFoundError(f"marcus.db not found at {marcus_db}")
+
+    store = CostStore(db_path=costs_db)
+
+    src = sqlite3.connect(str(marcus_db))
+    rows = src.execute(
+        "SELECT key, data FROM persistence WHERE collection = 'task_metadata'"
+    ).fetchall()
+    src.close()
+
+    # Pass 1: parent-task names from marcus.db.
+    parents = 0
+    skipped = 0
+    for key, raw in rows:
+        try:
+            parsed = json.loads(raw) if isinstance(raw, str) else raw
+        except (json.JSONDecodeError, ValueError, TypeError):
+            skipped += 1
+            continue
+
+        if not isinstance(parsed, dict):
+            skipped += 1
+            continue
+
+        task_id = parsed.get("task_id") or key
+        name = parsed.get("name")
+        if not isinstance(task_id, str) or not isinstance(name, str) or not name:
+            skipped += 1
+            continue
+
+        store.record_task_name(task_id, name)
+        parents += 1
+
+    # Pass 2: subtask derived names. Look for ``_sub_N`` task_ids that
+    # appear in token_events but have no entry in task_names yet, and
+    # synthesize a label from the parent's snapshotted name.
+    subtasks = 0
+    sub_rows = store.conn.execute("""
+        SELECT DISTINCT te.task_id
+        FROM token_events te
+        LEFT JOIN task_names tn USING (task_id)
+        WHERE te.task_id IS NOT NULL
+          AND tn.task_id IS NULL
+          AND instr(te.task_id, '_sub_') > 0
+        """).fetchall()
+    for (sub_id,) in sub_rows:
+        suffix_at = sub_id.find("_sub_")
+        parent_id = sub_id[:suffix_at]
+        suffix = sub_id[suffix_at + len("_sub_") :]
+        parent_name = store.get_task_name(parent_id)
+        if not parent_name:
+            continue
+        store.record_task_name(sub_id, f"{parent_name} — subtask {suffix}")
+        subtasks += 1
+
+    return parents, subtasks, skipped
+
+
+def main() -> int:
+    """Parse CLI args, run the backfill, and print a short summary."""
+    args = _parse_args()
+    parents, subtasks, skipped = backfill(args.marcus_db, args.costs_db)
+    print(f"Backfilled {parents} parent task names from {args.marcus_db}")
+    if subtasks > 0:
+        print(f"Backfilled {subtasks} subtask names derived from parents")
+    if skipped > 0:
+        print(f"Skipped {skipped} rows (malformed JSON or missing name)")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/cost_tracking/cost_aggregator.py

@@ -170,15 +170,21 @@ def run_summary(self, run_id: str) -> Optional[Dict[str, Any]]:
             (run_id,),
         )
 
+        # by_task: LEFT JOIN to task_names so the dashboard can render
+        # human-readable task names instead of opaque hex IDs (Marcus
+        # #530). Rows without a name entry come back with task_name=NULL
+        # and the dashboard falls back to the truncated id.
         by_task = self._rows(
             """
-            SELECT task_id,
+            SELECT t.task_id,
+                   n.name                            AS task_name,
                    COUNT(*)                          AS events,
-                   COALESCE(SUM(total_tokens), 0)    AS tokens,
-                   COALESCE(SUM(cost_usd), 0)        AS cost_usd
-            FROM v_event_cost_inclusive
-            WHERE run_id = ? AND task_id IS NOT NULL
-            GROUP BY task_id
+                   COALESCE(SUM(t.total_tokens), 0)  AS tokens,
+                   COALESCE(SUM(t.cost_usd), 0)      AS cost_usd
+            FROM v_event_cost_inclusive t
+            LEFT JOIN task_names n USING (task_id)
+            WHERE t.run_id = ? AND t.task_id IS NOT NULL
+            GROUP BY t.task_id, n.name
             ORDER BY cost_usd DESC
             """,
             (run_id,),
@@ -598,15 +604,18 @@ def project_summary(self, project_id: str) -> Optional[Dict[str, Any]]:
             (project_id,),
         )
 
+        # See run_summary for the LEFT JOIN rationale (Marcus #530).
         by_task = self._rows(
             """
-            SELECT task_id,
+            SELECT t.task_id,
+                   n.name                            AS task_name,
                    COUNT(*)                          AS events,
-                   COALESCE(SUM(total_tokens), 0)    AS tokens,
-                   COALESCE(SUM(cost_usd), 0)        AS cost_usd
-            FROM v_event_cost_inclusive
-            WHERE project_id = ? AND task_id IS NOT NULL
-            GROUP BY task_id
+                   COALESCE(SUM(t.total_tokens), 0)  AS tokens,
+                   COALESCE(SUM(t.cost_usd), 0)      AS cost_usd
+            FROM v_event_cost_inclusive t
+            LEFT JOIN task_names n USING (task_id)
+            WHERE t.project_id = ? AND t.task_id IS NOT NULL
+            GROUP BY t.task_id, n.name
             ORDER BY cost_usd DESC
             """,
             (project_id,),

src/cost_tracking/cost_store.py

@@ -154,6 +154,20 @@
                 DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
 );
 
+-- Per-task name snapshot (Marcus #530). Mirrors project_names. Marcus
+-- writes the kanban task name here when the decomposer first creates
+-- the task; Cato joins by task_id in by_task SQL so the dashboard
+-- shows real names instead of opaque hex IDs. Idempotent UPSERT keeps
+-- the most recent name when a task gets renamed mid-flight.
+CREATE TABLE IF NOT EXISTS task_names (
+  task_id       TEXT PRIMARY KEY,
+  name          TEXT NOT NULL,
+  first_seen    TIMESTAMP NOT NULL
+                DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+  last_seen     TIMESTAMP NOT NULL
+                DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
 -- Project-level budget caps. Set by the dashboard so users can compare
 -- spend against a target without needing MLflow experiments. Stored in
 -- the cost DB (not ProjectRegistry) because the cap is a cost concept,
@@ -955,6 +969,42 @@ def get_project_name(self, project_id: str) -> Optional[str]:
         ).fetchone()
         return row[0] if row else None
 
+    def record_task_name(self, task_id: str, name: str) -> None:
+        """Snapshot a kanban task's name into cost storage (Marcus #530).
+
+        Mirrors :meth:`upsert_project_name` for tasks. Called by the
+        decomposer at the moment a task is first created, so the
+        ``by_task`` chart on Cato shows human-readable names instead
+        of opaque hex IDs. Idempotent: same ``(task_id, name)`` is a
+        no-op after the first call; a name change updates in place
+        and bumps ``last_seen``.
+
+        Tasks whose name isn't snapshotted still appear in ``by_task``
+        (the aggregator LEFT JOINs); the dashboard falls back to the
+        truncated ID for those rows.
+        """
+        if not task_id or not name:
+            return
+        self.conn.execute(
+            """
+            INSERT INTO task_names (task_id, name)
+            VALUES (?, ?)
+            ON CONFLICT(task_id) DO UPDATE SET
+                name = excluded.name,
+                last_seen = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+            """,
+            (task_id, name),
+        )
+        self.conn.commit()
+
+    def get_task_name(self, task_id: str) -> Optional[str]:
+        """Return the snapshotted name for ``task_id``, or None."""
+        row = self.conn.execute(
+            "SELECT name FROM task_names WHERE task_id = ?",
+            (task_id,),
+        ).fetchone()
+        return row[0] if row else None
+
     def rebind_project_id(self, *, from_id: str, to_id: str) -> int:
         """Re-attribute every token_events row from one project to another.
 

src/integrations/nlp_base.py

@@ -184,6 +184,31 @@ async def create_tasks_on_board(
                         f"Failed to log task metadata for '{task.name}': {log_error}"
                     )
 
+                # Snapshot the kanban task's human name into costs.db so the
+                # Cato dashboard can render real names in "Tokens by task"
+                # instead of opaque hex IDs (Marcus #530). Lives in the
+                # shared task-creation method so BOTH create_project and
+                # add_feature flows are covered automatically — Codex P2 on
+                # PR #536 caught that the project-only snapshot left the
+                # feature-adder path silently dropping names.
+                #
+                # Failure is logged + swallowed: cost tracking is optional
+                # and must never block kanban task creation.
+                try:
+                    from src.cost_tracking.cost_recorder import (
+                        get_recorder as _cost_get_recorder,
+                    )
+
+                    _cost_store = _cost_get_recorder().store
+                    if kanban_task.id and task.name:
+                        _cost_store.record_task_name(str(kanban_task.id), task.name)
+                except Exception as _name_err:  # pragma: no cover
+                    logger.debug(
+                        "Failed to snapshot task name for %s: %s",
+                        kanban_task.id,
+                        _name_err,
+                    )
+
             except Exception as e:
                 from src.core.error_framework import (
                     ErrorContext,

src/integrations/nlp_tools.py

@@ -1596,8 +1596,11 @@ async def create_project_from_description(
                         f"Failed to stage planning phase metadata: {_plan_err}"
                     )
 
-                # NOW create About task AFTER decomposition with real task IDs
-                # Map created tasks to original tasks to preserve details
+                # NOW create About task AFTER decomposition with real task IDs.
+                # Map created tasks to original tasks to preserve details.
+                # Task-name snapshotting for the cost dashboard happens in
+                # the shared ``create_tasks_on_board`` (nlp_base.py) so
+                # both this flow and the feature-adder flow get covered.
                 tasks_with_real_ids = []
                 for i, created in enumerate(created_tasks):
                     if i < len(safe_tasks):

tests/unit/cost_tracking/test_cost_aggregator.py

@@ -668,3 +668,31 @@ def test_project_summary_includes_by_tool(self, populated_store: CostStore) -> N
         assert result is not None
         intents = {r["tool_intent"] for r in result["by_tool"]}
         assert "worker_bash" in intents
+
+
+class TestTaskNamesJoin:
+    """``by_task`` LEFT JOINs to task_names so the dashboard sees names
+    (Marcus #530). Rows without a name entry still appear with NULL.
+    """
+
+    def test_run_summary_by_task_includes_task_name(
+        self, populated_store: CostStore
+    ) -> None:
+        """When a task_name row exists, by_task surfaces it."""
+        populated_store.record_task_name("t_1", "Implement scoring logic")
+        result = CostAggregator(store=populated_store).run_summary("exp_1")
+        assert result is not None
+        rows = {r["task_id"]: r for r in result["by_task"]}
+        assert rows["t_1"]["task_name"] == "Implement scoring logic"
+        # t_2 has no name entry — LEFT JOIN returns NULL, not a row drop.
+        assert rows["t_2"]["task_name"] is None
+
+    def test_project_summary_by_task_includes_task_name(
+        self, populated_store: CostStore
+    ) -> None:
+        """Same LEFT JOIN on the project-scoped summary."""
+        populated_store.record_task_name("t_1", "Implement scoring logic")
+        result = CostAggregator(store=populated_store).project_summary("proj_1")
+        assert result is not None
+        rows = {r["task_id"]: r for r in result["by_task"]}
+        assert rows["t_1"]["task_name"] == "Implement scoring logic"

tests/unit/cost_tracking/test_cost_store.py

@@ -13,6 +13,8 @@
 
 import pytest
 
+pytestmark = pytest.mark.unit
+
 from src.cost_tracking.cost_store import (
     CostStore,
     ModelPrice,
@@ -878,3 +880,32 @@ def test_load_seed_idempotent(self, store: CostStore) -> None:
             "SELECT COUNT(*) FROM model_prices"
         ).fetchone()[0]
         assert first_count == second_count
+
+
+class TestTaskNamesSnapshot:
+    """``task_names`` table mirrors the project_names snapshot pattern (#530)."""
+
+    def test_record_and_retrieve_task_name(self, tmp_path: Path) -> None:
+        """Round-trip a single task name."""
+        s = CostStore(db_path=tmp_path / "costs.db")
+        s.record_task_name("abc123", "Implement scoring logic")
+        assert s.get_task_name("abc123") == "Implement scoring logic"
+
+    def test_upsert_on_conflict(self, tmp_path: Path) -> None:
+        """A name change updates in place."""
+        s = CostStore(db_path=tmp_path / "costs.db")
+        s.record_task_name("abc123", "Old name")
+        s.record_task_name("abc123", "New name")
+        assert s.get_task_name("abc123") == "New name"
+
+    def test_empty_inputs_are_no_op(self, tmp_path: Path) -> None:
+        """Empty task_id or name silently does nothing — doesn't crash."""
+        s = CostStore(db_path=tmp_path / "costs.db")
+        s.record_task_name("", "name")
+        s.record_task_name("abc", "")
+        assert s.get_task_name("abc") is None
+
+    def test_missing_task_returns_none(self, tmp_path: Path) -> None:
+        """get_task_name on an unrecorded id returns None, not an error."""
+        s = CostStore(db_path=tmp_path / "costs.db")
+        assert s.get_task_name("never_recorded") is None