fix: persist sub-agent execution metrics

Fixes #84

Author: EvanProgramming

event_store.py

@@ -318,7 +318,7 @@ def integer(value: int | None) -> int | None:
 
     def list_usage_attempts(
             self, *, workspace_id: str | None = None, session_id: str | None = None,
-            user_id: str | None = None, limit: int = 1000,
+            user_id: str | None = None, run_id: str | None = None, limit: int = 1000,
     ) -> list[dict[str, Any]]:
         clauses = ["1=1"]
         params: list[Any] = []
@@ -331,6 +331,9 @@ def list_usage_attempts(
         if user_id is not None:
             clauses.append("user_id=?")
             params.append(user_id)
+        if run_id is not None:
+            clauses.append("run_id=?")
+            params.append(run_id)
         params.append(max(1, min(limit, 10000)))
         with self.connection() as db:
             rows = db.execute(
@@ -341,13 +344,13 @@ def list_usage_attempts(
 
     def usage_totals(
             self, *, workspace_id: str | None = None, session_id: str | None = None,
-            user_id: str | None = None, since: str | None = None,
+            user_id: str | None = None, run_id: str | None = None, since: str | None = None,
     ) -> dict[str, Any]:
         """Aggregate immutable usage records for one explicit durable scope."""
         clauses = ["1=1"]
         params: list[Any] = []
         for field, value in (("workspace_id", workspace_id), ("session_id", session_id),
-                             ("user_id", user_id)):
+                             ("user_id", user_id), ("run_id", run_id)):
             if value is not None:
                 clauses.append(f"{field}=?")
                 params.append(value)

learning_engine.py

@@ -150,16 +150,18 @@ def artifact_context(self, run: dict[str, str]) -> tuple[str, list[dict[str, Any
         return "\n".join(lines), receipts
 
     def complete_run(self, run: dict[str, str], *, result: str, receipts: list[dict[str, Any]],
-                     tools: list[dict[str, Any]], tokens: int, latency: float,
+                     tools: list[dict[str, Any]], tokens: int | None, latency: float | None,
                      acceptance: list[dict[str, Any]] | None = None,
                      provider_error: bool = False) -> None:
         errors = sum(1 for item in tools if not item.get("success"))
         contains_secret = bool(_SECRET_RE.search(json.dumps(
             {"task": run.get("task"), "result": result, "tools": tools}, ensure_ascii=False,
         )))
         payload = {**run, "result": str(result)[:4000], "receipts": receipts, "tools": tools[:50],
-                   "tool_calls": len(tools), "errors": errors, "tokens": max(0, int(tokens)),
-                   "latency": max(0.0, float(latency)), "provider_error": bool(provider_error),
+                   "tool_calls": len(tools), "errors": errors,
+                   "tokens": None if tokens is None else max(0, int(tokens)),
+                   "latency": None if latency is None else max(0.0, float(latency)),
+                   "provider_error": bool(provider_error),
                    "acceptance_evidence": (acceptance or [])[:20],
                    "contains_secret": contains_secret,
                    "eligible": not provider_error and not contains_secret and len(tools) >= 2}
@@ -652,7 +654,7 @@ def pending_reviews(self, *, min_age_seconds: float = 30.0, limit: int = 1) -> l
             recurrence = sum(item["payload"].get("task_signature") == payload.get("task_signature")
                              for item in completed)
             score = recurrence * 2 + int(payload["run_id"] in requested) * 5 + int(payload.get("errors", 0)) * 2
-            score += min(5, int(payload.get("tokens", 0)) // 1000)
+            score += min(5, int(payload.get("tokens") or 0) // 1000)
             candidates.append((score, event["created_at"], payload))
         candidates.sort(key=lambda item: (-item[0], item[1]))
         return [payload for _, _, payload in candidates[:limit]]
@@ -803,8 +805,8 @@ def metrics(self, profile: str | None = None) -> dict[str, Any]:
             "correction_rate": (sum(bool(item.get("correction")) for item in verified) / len(verified)) if verified else None,
             "repeated_error_rate": (sum(item.get("errors", 0) > 1 for item in completed) / len(completed)) if completed else None,
             "tool_calls": sum(int(item.get("tool_calls", 0)) for item in completed),
-            "tokens": sum(int(item.get("tokens", 0)) for item in completed),
-            "latency": sum(float(item.get("latency", 0.0)) for item in completed),
+            "tokens": sum(int(item.get("tokens") or 0) for item in completed),
+            "latency": sum(float(item.get("latency") or 0.0) for item in completed),
             "task_families": families,
             "context_chars": sum(sum(int(receipt.get("chars", 0)) for receipt in item.get("receipts", []))
                                  for item in outcomes),

main.py

@@ -2348,7 +2348,7 @@ def _run_subagent_tool(profile: AgentProfile, action: str, args: Any, tools: set
     }
 
 
-def _run_subagent_llm(profile: AgentProfile, task: str, context: list[dict[str, Any]], tools: set[str]) -> dict[str, Any]:
+def _run_subagent_llm_result(profile: AgentProfile, task: str, context: list[dict[str, Any]], tools: set[str]) -> dict[str, Any]:
     memory_lines = "\n".join(
         f"- kind={item.get('kind')} confidence={item.get('confidence', 0):.2f}: {str(item.get('content', ''))[:500]}"
         for item in context
@@ -2421,7 +2421,66 @@ def _run_subagent_llm(profile: AgentProfile, task: str, context: list[dict[str,
     return {"result": response, "tool_records": tool_records, "evidence": evidence}
 
 
-subagent_manager = SubAgentManager(memory_bank, runner=_run_subagent_llm, learning_engine=learning_engine)
+def _subagent_usage_metrics(run_id: str, started: float) -> dict[str, Any]:
+    """Read one sub-agent's actual provider usage from the durable ledger."""
+    attempts = memory_bank.store.list_usage_attempts(
+        user_id=memory_bank.user_id, workspace_id=memory_bank.workspace_id, run_id=run_id,
+    )
+    totals = memory_bank.store.usage_totals(
+        user_id=memory_bank.user_id, workspace_id=memory_bank.workspace_id, run_id=run_id,
+    )
+
+    def known_total(field: str) -> int | None:
+        if not attempts or any(item.get(field) is None for item in attempts):
+            return None
+        return int(totals[field])
+
+    provider_models = sorted({f"{item['provider']}:{item['model']}" for item in attempts})
+    statuses = {str(item.get("usage_status") or "unknown") for item in attempts}
+    return {
+        "provider_model": provider_models[0] if len(provider_models) == 1 else (
+            "mixed" if provider_models else "unknown"
+        ),
+        "provider_models": provider_models,
+        "attempts": int(totals["attempts"]),
+        "prompt_tokens": known_total("prompt_tokens"),
+        "completion_tokens": known_total("completion_tokens"),
+        "reasoning_tokens": known_total("reasoning_tokens"),
+        "tokens": (
+            int(totals["prompt_tokens"]) + int(totals["completion_tokens"])
+            if known_total("prompt_tokens") is not None and known_total("completion_tokens") is not None else None
+        ),
+        "latency_ms": round((time.monotonic() - started) * 1000, 3),
+        "usage_status": (
+            next(iter(statuses)) if len(statuses) == 1 else "mixed" if statuses else "unknown"
+        ),
+    }
+
+
+def _run_subagent_llm(profile: AgentProfile, task: str, context: list[dict[str, Any]], tools: set[str], *,
+                      run_id: str | None = None) -> dict[str, Any]:
+    started = time.monotonic()
+    token = _active_usage_run_id.set(run_id) if run_id else None
+    try:
+        result = _run_subagent_llm_result(profile, task, context, tools)
+        if run_id:
+            result["metrics"] = _subagent_usage_metrics(run_id, started)
+        return result
+    finally:
+        if token is not None:
+            _active_usage_run_id.reset(token)
+
+
+def _subagent_provider_model() -> str:
+    if _provider_config and DEEPSEEK_MODEL:
+        return f"{_provider_config.provider}:{DEEPSEEK_MODEL}"
+    return "unknown"
+
+
+subagent_manager = SubAgentManager(
+    memory_bank, runner=_run_subagent_llm, learning_engine=learning_engine,
+    provider_model=_subagent_provider_model,
+)
 
 # ---- Tool to let agent examine its own memory ----
 def _check_stored_data(args: str) -> str:

subagents.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import re
+import time
 import uuid
 from dataclasses import dataclass, field
 from typing import Any, Callable, TYPE_CHECKING
@@ -27,11 +28,13 @@ class AgentProfile:
 
 
 class SubAgentManager:
-    def __init__(self, memory: MemoryBank, *, runner: Callable[[AgentProfile, str, list[dict[str, Any]], set[str]], str | dict[str, Any]] | None = None,
-                 learning_engine: "LearningEngine | None" = None):
+    def __init__(self, memory: MemoryBank, *, runner: Callable[..., str | dict[str, Any]] | None = None,
+                 learning_engine: "LearningEngine | None" = None,
+                 provider_model: Callable[[], str] | None = None):
         self.memory = memory
         self.runner = runner
         self.learning_engine = learning_engine
+        self.provider_model = provider_model
         self.profiles: dict[str, AgentProfile] = {}
         self.register(AgentProfile("researcher", "Research the task, cite evidence, and separate facts from assumptions.", "readonly"))
         self.register(AgentProfile("coder", "Inspect and modify code only when requested; verify every change.", "workspace"))
@@ -71,10 +74,11 @@ def run(self, profile_name: str, task: str, *, workspace_id: str | None = None)
             task, n_results=5, profile=profile.name,
             task_signature=self.learning_engine.task_signature(profile.name, task) if self.learning_engine else None,
         ) if profile.memory_scope != "none" else [])
+        expected_provider_model = self.provider_model() if self.provider_model else "unknown"
         learning_run = None
         receipts: list[dict[str, Any]] = []
         if self.learning_engine and profile.evolution_enabled and profile.name in {"coder", "researcher"}:
-            learning_run = self.learning_engine.begin_run(profile.name, task)
+            learning_run = self.learning_engine.begin_run(profile.name, task, provider_model=expected_provider_model)
             guidance, receipts = self.learning_engine.artifact_context(learning_run)
             if guidance:
                 context.append({"kind": "learned_guidance", "confidence": 1.0, "content": guidance})
@@ -83,21 +87,51 @@ def run(self, profile_name: str, task: str, *, workspace_id: str | None = None)
                                        user_id=self.memory.user_id, workspace_id=workspace_id, session_id=run_id)
         tool_records: list[dict[str, Any]] = []
         evidence: list[dict[str, Any]] = []
+        metrics: dict[str, Any] = {
+            "provider_model": expected_provider_model or "unknown", "provider_models": [], "attempts": None,
+            "prompt_tokens": None, "completion_tokens": None, "reasoning_tokens": None, "tokens": None,
+            "latency_ms": None, "usage_status": "unknown",
+        }
         provider_error = False
+        started = time.monotonic()
         if self.runner is None:
             result = "Sub-agent runner is not configured."
         else:
             try:
-                raw_result = self.runner(profile, task, context, tools)
+                raw_result = self.runner(profile, task, context, tools, run_id=run_id)
                 if isinstance(raw_result, dict):
                     result = str(raw_result.get("result", ""))
                     tool_records = [item for item in raw_result.get("tool_records", []) if isinstance(item, dict)]
                     evidence = [item for item in raw_result.get("evidence", []) if isinstance(item, dict)]
+                    if isinstance(raw_result.get("metrics"), dict):
+                        metrics.update({key: raw_result["metrics"][key] for key in metrics
+                                        if key in raw_result["metrics"]})
                 else:
                     result = str(raw_result)
             except Exception as exc:
                 provider_error = True
                 result = f"Sub-agent failed: {type(exc).__name__}: {exc}"
+        if not isinstance(metrics.get("provider_model"), str) or not metrics["provider_model"].strip():
+            metrics["provider_model"] = expected_provider_model or "unknown"
+        metrics["provider_model"] = str(metrics["provider_model"]).strip()[:192]
+        metrics["provider_models"] = [str(item)[:192] for item in metrics["provider_models"]] if isinstance(
+            metrics["provider_models"], (list, tuple, set)
+        ) else []
+        if metrics["usage_status"] not in {"authoritative", "estimated", "mixed", "unknown"}:
+            metrics["usage_status"] = "unknown"
+        for field in ("attempts", "prompt_tokens", "completion_tokens", "reasoning_tokens", "tokens"):
+            try:
+                metrics[field] = None if metrics[field] is None else max(0, int(metrics[field]))
+            except (TypeError, ValueError):
+                metrics[field] = None
+        try:
+            metrics["latency_ms"] = None if metrics["latency_ms"] is None else max(0.0, float(metrics["latency_ms"]))
+        except (TypeError, ValueError):
+            metrics["latency_ms"] = None
+        if metrics.get("latency_ms") is None:
+            metrics["latency_ms"] = (time.monotonic() - started) * 1000
+        if learning_run:
+            learning_run["provider_model"] = metrics["provider_model"]
 
         for record in tool_records:
             payload = {"run_id": run_id, **record}
@@ -119,13 +153,16 @@ def run(self, profile_name: str, task: str, *, workspace_id: str | None = None)
                                                                "result": str(result)[:4000],
                                                                "tool_receipts": tool_records[:50],
                                                                "acceptance_evidence": evidence[:20],
+                                                               "metrics": metrics,
                                                                "provider_error": provider_error},
                                        user_id=self.memory.user_id, workspace_id=workspace_id, session_id=run_id)
         scoped_memory.add_log(str(result), kind="episodic", status="active",
                               metadata={"subagent_run_id": run_id, "profile": profile.name})
         if learning_run:
             self.learning_engine.complete_run(learning_run, result=str(result), receipts=receipts,
-                                              tools=tool_records, tokens=0, latency=0.0,
+                                              tools=tool_records, tokens=metrics.get("tokens"),
+                                              latency=(float(metrics["latency_ms"]) / 1000
+                                                       if metrics.get("latency_ms") is not None else None),
                                               acceptance=evidence, provider_error=provider_error)
             if evidence:
                 self.learning_engine.record_outcome(
@@ -135,4 +172,4 @@ def run(self, profile_name: str, task: str, *, workspace_id: str | None = None)
                 )
         return {"run_id": run_id, "profile": profile.name, "result": str(result),
                 "tools": sorted(tools), "memory_scope": profile.memory_scope,
-                "tool_receipts": tool_records, "evidence": evidence}
+                "tool_receipts": tool_records, "evidence": evidence, "metrics": metrics}

tests/test_server.py

@@ -55,6 +55,11 @@ def test_subagent_api_executes_allowed_file_action_and_rejects_disallowed_action
             self.assertFalse((root / "denied.txt").exists())
             self.assertFalse(denied.json()["tool_receipts"][0]["success"])
             self.assertFalse(denied.json()["tool_receipts"][0]["authorized"])
+            completed = next(event for event in manager.memory.store.list_events(
+                "subagent.completed", workspace_id=server._agent.memory_bank.workspace_id,
+                session_id=allowed.json()["run_id"], limit=10,
+            ))
+            self.assertEqual(completed["payload"]["metrics"], allowed.json()["metrics"])
             events = manager.memory.store.list_events(
                 workspace_id=server._agent.memory_bank.workspace_id,
                 session_id=denied.json()["run_id"], limit=20,