lwgray
diff --git a/‎src/cost_tracking/cost_aggregator.py‎
Lines changed: 36 additions & 0 deletions b/‎src/cost_tracking/cost_aggregator.py‎
Lines changed: 36 additions & 0 deletions
diff --git a/‎src/cost_tracking/cost_store.py‎
Lines changed: 94 additions & 2 deletions b/‎src/cost_tracking/cost_store.py‎
Lines changed: 94 additions & 2 deletions
diff --git a/‎src/cost_tracking/worker_ingester.py‎
Lines changed: 127 additions & 3 deletions b/‎src/cost_tracking/worker_ingester.py‎
Lines changed: 127 additions & 3 deletions
@@ -222,6 +222,25 @@ def run_summary(self, run_id: str) -> Optional[Dict[str, Any]]:
             (run_id,),
         )
 
+        # by_tool: spend grouped by Claude Code tool the agent invoked on
+        # each turn (Marcus #527 Phase 2). Worker-only — planner rows
+        # have tool_intent NULL. Coordination overhead (worker_marcus_call)
+        # surfaces here as its own slice, which is the metric users most
+        # want to see and the chart can't display today.
+        by_tool = self._rows(
+            """
+            SELECT tool_intent,
+                   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 tool_intent IS NOT NULL
+            GROUP BY tool_intent
+            ORDER BY cost_usd DESC
+            """,
+            (run_id,),
+        )
+
         return {
             **meta,
             "summary": totals,
@@ -230,6 +249,7 @@ def run_summary(self, run_id: str) -> Optional[Dict[str, Any]]:
             "by_task": by_task,
             "by_operation": by_operation,
             "by_model": by_model,
+            "by_tool": by_tool,
             "audit": self.run_audit(run_id),
         }
 
@@ -658,6 +678,21 @@ def project_summary(self, project_id: str) -> Optional[Dict[str, Any]]:
             (project_id,),
         )
 
+        # by_tool: same shape as run_summary's by_tool. See #527 Phase 2.
+        by_tool = self._rows(
+            """
+            SELECT tool_intent,
+                   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 tool_intent IS NOT NULL
+            GROUP BY tool_intent
+            ORDER BY cost_usd DESC
+            """,
+            (project_id,),
+        )
+
         return {
             "project_id": project_id,
             "summary": totals,
@@ -666,5 +701,6 @@ def project_summary(self, project_id: str) -> Optional[Dict[str, Any]]:
             "by_task": by_task,
             "by_operation": by_operation,
             "by_model": by_model,
+            "by_tool": by_tool,
             "audit": self.project_audit(project_id),
         }
@@ -90,6 +90,13 @@
   task_id               TEXT,
   subtask_id            TEXT,
   operation             TEXT NOT NULL,
+  -- ``tool_intent`` (Marcus #527 Phase 2): which Claude Code tool the
+  -- agent invoked on this turn. Populated by ``worker_intent.py`` from
+  -- the JSONL ``message.content`` tool_use blocks. NULL on planner
+  -- rows and on legacy worker rows from before the parser landed.
+  -- Values: worker_marcus_call / worker_mcp_call / worker_edit /
+  -- worker_bash / worker_search / worker_read / worker_text / unknown.
+  tool_intent           TEXT,
   provider              TEXT NOT NULL,
   model                 TEXT NOT NULL,
   input_tokens          INTEGER NOT NULL DEFAULT 0,
@@ -119,6 +126,7 @@
 CREATE INDEX IF NOT EXISTS idx_te_run_agent ON token_events(run_id, agent_id);
 CREATE INDEX IF NOT EXISTS idx_te_task      ON token_events(task_id);
 CREATE INDEX IF NOT EXISTS idx_te_op_model  ON token_events(operation, model);
+CREATE INDEX IF NOT EXISTS idx_te_intent    ON token_events(tool_intent);
 CREATE INDEX IF NOT EXISTS idx_te_timestamp ON token_events(timestamp);
 -- Partial unique index for dedup. Worker JSONL ingestion may sweep the
 -- same session files repeatedly (e.g. Cato's 30s dashboard poll re-runs
@@ -276,6 +284,7 @@ class TokenEvent:
     parent_agent_id: Optional[str] = None
     task_id: Optional[str] = None
     subtask_id: Optional[str] = None
+    tool_intent: Optional[str] = None
     latency_ms: Optional[int] = None
     session_id: Optional[str] = None
     turn_index: Optional[int] = None
@@ -570,6 +579,7 @@ def _init_schema(self) -> None:
         """
         self._runs_rename_migration()
         self._dedup_pre_index_migration()
+        self._tool_intent_migration()
         self.conn.executescript(SCHEMA_SQL)
         self.conn.commit()
 
@@ -643,6 +653,37 @@ def _runs_rename_migration(self) -> None:
         )
         self.conn.commit()
 
+    def _tool_intent_migration(self) -> None:
+        """Add ``token_events.tool_intent`` to existing DBs (Marcus #527 Phase 2).
+
+        Fresh installs get the column via SCHEMA_SQL. For an existing
+        cost DB the column doesn't exist; this migration adds it as
+        nullable so old rows survive with NULL until a backfill runs.
+        Idempotent: detects the column via ``PRAGMA table_info`` and
+        skips when already present.
+
+        Designed to be cheap and lock-friendly: a single ``ALTER TABLE
+        ADD COLUMN`` is a metadata-only operation in SQLite, so it
+        doesn't rewrite the table even on a multi-million-row
+        ``token_events``.
+        """
+        cols = {
+            row[1]
+            for row in self.conn.execute("PRAGMA table_info(token_events)").fetchall()
+        }
+        if "token_events" not in {
+            r[0]
+            for r in self.conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table'"
+            ).fetchall()
+        }:
+            # Fresh install — SCHEMA_SQL will create the column.
+            return
+        if "tool_intent" in cols:
+            return
+        self.conn.execute("ALTER TABLE token_events ADD COLUMN tool_intent TEXT")
+        self.conn.commit()
+
     def _dedup_pre_index_migration(self) -> None:
         """Compact duplicate ``request_id`` rows before the index lands.
 
@@ -698,11 +739,11 @@ def record_event(self, event: TokenEvent) -> int:
             """
             INSERT OR IGNORE INTO token_events (
                 run_id, project_id, agent_id, agent_role,
-                parent_agent_id, task_id, subtask_id, operation,
+                parent_agent_id, task_id, subtask_id, operation, tool_intent,
                 provider, model, input_tokens, cache_creation_tokens,
                 cache_read_tokens, output_tokens, latency_ms, session_id,
                 turn_index, request_id, status, error_type, timestamp
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                       COALESCE(?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')))
             """,
             (
@@ -714,6 +755,7 @@ def record_event(self, event: TokenEvent) -> int:
                 event.task_id,
                 event.subtask_id,
                 event.operation,
+                event.tool_intent,
                 event.provider,
                 event.model,
                 event.input_tokens,
@@ -745,6 +787,56 @@ def record_event(self, event: TokenEvent) -> int:
             raise RuntimeError("sqlite3 did not return lastrowid")
         return event_id
 
+    def update_attribution(
+        self,
+        request_id: str,
+        *,
+        task_id: Optional[str] = None,
+        tool_intent: Optional[str] = None,
+    ) -> int:
+        """Backfill ``task_id`` and/or ``tool_intent`` on an existing row.
+
+        Used by the worker JSONL backfill path (Marcus #527 Phase 1.5
+        + 2): historical rows were written before the ingester knew
+        how to populate these fields. A re-walk of the JSONL files
+        runs the parser and calls this helper to fill in the gaps.
+
+        Idempotent and side-effect-light: only updates columns whose
+        current value is NULL, so re-running on already-backfilled
+        rows is a no-op. Returns the number of rows updated (0 or 1)
+        so callers can report progress.
+
+        Parameters
+        ----------
+        request_id : str
+            The Claude Code per-call ID used as the upsert key.
+        task_id : str, optional
+            New ``task_id``. Only written if the row's current
+            ``task_id`` is NULL.
+        tool_intent : str, optional
+            New ``tool_intent``. Only written if the row's current
+            ``tool_intent`` is NULL.
+
+        Returns
+        -------
+        int
+            ``cur.rowcount`` from the UPDATE — 0 if no row matched
+            the ``request_id``, 1 if a row was updated. Note: SQLite
+            still reports 1 when the WHERE clause matched but the
+            COALESCE guards left every column unchanged.
+        """
+        cur = self.conn.execute(
+            """
+            UPDATE token_events
+               SET task_id     = COALESCE(task_id, ?),
+                   tool_intent = COALESCE(tool_intent, ?)
+             WHERE request_id = ?
+            """,
+            (task_id, tool_intent, request_id),
+        )
+        self.conn.commit()
+        return cur.rowcount
+
     def record_run(self, run: Run) -> None:
         """Upsert one ``runs`` row keyed by ``run_id``.
 
 
@@ -39,6 +39,10 @@
 
 from src.cost_tracking.cost_recorder import canonical_project_id
 from src.cost_tracking.cost_store import CostStore, TokenEvent
+from src.cost_tracking.worker_intent import (
+    classify_tool_intent,
+    extract_task_id_from_user_message,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -117,6 +121,13 @@ def __init__(
         self.resolve_binding = resolve_binding
         self._turn_counter: Dict[str, int] = defaultdict(int)
         self._seen_uuids: Set[str] = set()
+        # Per-session running task tracker (Marcus #527 Phase 1.5).
+        # Updated when a user-record carries a Marcus MCP tool_result
+        # echoing a task_id; consumed when the next assistant record on
+        # the same session emits a token_events row. Lives per-instance
+        # so a single ingest_file call walks the file in order with the
+        # tracker following the agent's task lifecycle.
+        self._session_task: Dict[str, str] = {}
 
     # -- file-level API ---------------------------------------------------
 
@@ -155,18 +166,119 @@ def ingest_directory(self, dir_path: Path) -> int:
             total += self.ingest_file(jsonl)
         return total
 
+    def backfill_file(self, path: Path) -> int:
+        """Walk a JSONL file and backfill ``task_id`` / ``tool_intent``.
+
+        Used to populate the new columns on rows that were ingested
+        before the parser landed (Marcus #527 Phase 1.5 + 2). Walks
+        the file in order, maintains the same per-session task
+        tracker as :meth:`ingest_file`, and calls
+        :meth:`CostStore.update_attribution` for each assistant
+        record carrying a ``request_id``. Skips rows that don't exist
+        in the DB (the UPDATE rowcount comes back zero) and rows
+        whose columns are already populated (the store's
+        ``COALESCE`` guards no-op them).
+
+        Idempotent: safe to re-run on the same file. The store's
+        ``update_attribution`` only fills NULL columns, so a second
+        pass after manual edits will not overwrite curated values.
+
+        Returns
+        -------
+        int
+            Number of rows whose ``task_id`` and/or ``tool_intent``
+            were filled in by this call.
+        """
+        # Use a fresh per-call tracker so backfill state doesn't bleed
+        # into a subsequent ingest_file call (or vice versa).
+        tracker: Dict[str, str] = {}
+        updated = 0
+        with path.open("r", encoding="utf-8") as fh:
+            for raw in fh:
+                line = raw.strip()
+                if not line:
+                    continue
+                try:
+                    record = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+
+                session_id = record.get("sessionId") or "unknown"
+                record_type = record.get("type")
+
+                if record_type == "user":
+                    message = record.get("message") or {}
+                    content = (
+                        message.get("content") if isinstance(message, dict) else None
+                    )
+                    tid = extract_task_id_from_user_message(content)
+                    if tid is not None:
+                        tracker[session_id] = tid
+                    continue
+
+                if record_type != "assistant":
+                    continue
+
+                request_id = record.get("requestId")
+                if not request_id:
+                    continue
+
+                message = record.get("message") or {}
+                content = message.get("content") if isinstance(message, dict) else None
+                tool_intent = classify_tool_intent(content)
+                task_id = tracker.get(session_id)
+
+                # classify_tool_intent always returns a non-None string, so
+                # there's always at least one column to potentially fill.
+                # update_attribution's COALESCE guards turn this into a
+                # no-op when the row already has those values populated.
+                rowcount = self.store.update_attribution(
+                    request_id=request_id,
+                    task_id=task_id,
+                    tool_intent=tool_intent,
+                )
+                if rowcount > 0:
+                    updated += 1
+        return updated
+
+    def backfill_directory(self, dir_path: Path) -> int:
+        """Recursively backfill every ``*.jsonl`` file under a directory."""
+        total = 0
+        for jsonl in sorted(dir_path.rglob("*.jsonl")):
+            total += self.backfill_file(jsonl)
+        return total
+
     # -- record-level helpers --------------------------------------------
 
     def _ingest_record(self, record: Dict[str, Any]) -> bool:
         """Insert one token_events row if the record is an assistant turn.
 
+        Also walks ``user`` records (tool_result carriers) to update
+        the per-session task tracker (Marcus #527 Phase 1.5) — those
+        records produce no row themselves but update the state used by
+        the next assistant record on the same session.
+
         Returns
         -------
         bool
             True if a row was inserted, False if the record was skipped
             (wrong type, missing usage, deduped, or binding rejected).
         """
-        if record.get("type") != "assistant":
+        session_id = record.get("sessionId") or "unknown"
+        record_type = record.get("type")
+
+        # User records carry tool_results — scan them for Marcus MCP
+        # task_id echoes and update the tracker. No row written. The
+        # tracker is per-session so concurrent agents don't bleed.
+        if record_type == "user":
+            message = record.get("message") or {}
+            content = message.get("content") if isinstance(message, dict) else None
+            tid = extract_task_id_from_user_message(content)
+            if tid is not None:
+                self._session_task[session_id] = tid
+            return False
+
+        if record_type != "assistant":
             return False
         message = record.get("message") or {}
         usage = message.get("usage") if isinstance(message, dict) else None
@@ -181,10 +293,21 @@ def _ingest_record(self, record: Dict[str, Any]) -> bool:
         if binding is None:
             return False
 
-        session_id = record.get("sessionId") or "unknown"
         self._turn_counter[session_id] += 1
         turn_index = self._turn_counter[session_id]
 
+        # task_id: prefer caller-supplied binding, fall back to the
+        # tracker. The fallback covers the common case where the
+        # resolver only knows agent_id/project_id from the cwd and
+        # never sets task_id; the tracker fills it in from MCP
+        # tool-result history.
+        resolved_task_id = binding.task_id or self._session_task.get(session_id)
+
+        # tool_intent: classify the assistant message's content (#527
+        # Phase 2). Pure function in worker_intent.py.
+        content = message.get("content") if isinstance(message, dict) else None
+        tool_intent = classify_tool_intent(content)
+
         # Normalize project_id to the cost-data canonical form (dashless
         # hex). Bindings come from spawn_agents.py via project_info.json,
         # which may carry either dashed or dashless UUIDs depending on
@@ -199,7 +322,8 @@ def _ingest_record(self, record: Dict[str, Any]) -> bool:
             agent_id=binding.agent_id,
             agent_role="worker",
             parent_agent_id=binding.parent_agent_id,
-            task_id=binding.task_id,
+            task_id=resolved_task_id,
+            tool_intent=tool_intent,
             operation="turn",
             provider="anthropic",
             model=str(message.get("model", "unknown")),