diff --git a/src/ace/sidecar/dashboard_render.py b/src/ace/sidecar/dashboard_render.py index 7b9eb96..14e5684 100644 --- a/src/ace/sidecar/dashboard_render.py +++ b/src/ace/sidecar/dashboard_render.py @@ -714,6 +714,295 @@ def _activity_svg(daily: List[Dict[str, Any]], commits: bool) -> str: ) +def _quality(qm: Optional[Dict[str, Any]]) -> str: + """§ 02 — Code quality, verification hygiene, and agent execution reliability.""" + if not qm or not qm.get("available"): + return ( + "
~/ace/code_quality" + "NO SESSIONS
" + "
No session data available in this scope to compute code quality metrics. " + "Metrics will populate as coding agent sessions run and edit workspace files.
" + ) + + score = qm.get("quality_score", 100) + grade = qm.get("grade", "A") + c_rate = qm.get("task_completion_rate_pct", 100.0) + v_rate = qm.get("verification_rate_pct", 100.0) + fsr = qm.get("first_pass_success_rate_pct", 100.0) + err_rate = qm.get("tool_error_rate_pct", 0.0) + thrash_cnt = qm.get("thrashed_files_count", 0) + recovery_turns = qm.get("avg_error_recovery_turns", 1.0) + redundant_reads = qm.get("redundant_reads_count", 0) + test_code_ratio = qm.get("test_to_code_ratio", 1.0) + sessions_edits = qm.get("sessions_with_edits", 0) + sessions_tests = qm.get("sessions_with_tests", 0) + clean_completed = qm.get("clean_completed_sessions", 0) + + score_color = ( + "var(--mint)" + if score >= 80 + else ("var(--gold)" if score >= 60 else "var(--crit)") + ) + c_cls = "" if c_rate >= 80 else ("warn" if c_rate >= 60 else "crit") + v_cls = "" if v_rate >= 75 else ("warn" if v_rate >= 50 else "crit") + fsr_cls = "" if fsr >= 85 else ("warn" if fsr >= 70 else "crit") + thrash_cls = "" if thrash_cnt == 0 else ("warn" if thrash_cnt <= 5 else "crit") + + tiles = [ + _st( + "quality_score", + f"{score}/ 100", + f"Grade {grade}", + delta="COMPOSITE", + title="Weighted reliability index across task completion (35%), verification hygiene (30%), first-pass tool success (20%), and edit stability (15%).", + ), + _st( + "task_completion", + f"{c_rate}%", + f"{clean_completed} verified sessions", + delta="TASK RESOLUTION", + dcls=c_cls, + title="Percentage of sessions that resolved cleanly without trailing tool errors or unverified changes.", + ), + _st( + "verification_rate", + f"{v_rate}%", + f"{sessions_tests} of {sessions_edits} edit sessions", + delta="TEST HYGIENE", + dcls=v_cls, + title="Percentage of sessions containing file modifications that executed an automated test runner or linter (pytest, npm test, ruff, etc.).", + ), + _st( + "first_pass_success", + f"{fsr}%", + f"{err_rate}% error rate", + delta="TOOL RELIABILITY", + dcls=fsr_cls, + title="Share of tool executions that succeeded on their first attempt without returning execution errors or non-zero exit codes.", + ), + _st( + "edit_thrash_files", + f"{thrash_cnt}", + f"{qm.get('total_edits', 0)} total file edits", + delta="REWORK CHURN", + dcls=thrash_cls, + title="Files edited 3 or more times within the same session, indicating thrashing or lack of convergence.", + ), + _st( + "healing_latency", + f"{recovery_turns} turns", + f"{redundant_reads} redundant reads", + delta="ERROR HEALING", + title="Average number of conversation turns required for the agent to resolve a failed tool execution and resume forward progress.", + ), + ] + + thrashed_files_list = qm.get("thrashed_files_list") or [] + thrash_html = "" + if thrashed_files_list: + thrashed_items = "".join( + f"
  • {escape(_mask_home(f))}
    • " + for f in thrashed_files_list + ) + thrash_html = ( + f"
    " + f"⚠️ Repeatedly Modified Files (Thrashing Detected):" + f"" + f"
    " + ) + + breakdown_rows = [] + # Agent breakdown rows + by_agent = qm.get("by_agent") or {} + for ak, a_info in by_agent.items(): + a_score = a_info.get("quality_score", 100) + a_grade = a_info.get("grade", "A") + a_comp = a_info.get("task_completion_rate_pct", 100.0) + a_v_rate = a_info.get("verification_rate_pct", 100.0) + a_fsr = a_info.get("first_pass_success_rate_pct", 100.0) + a_thrash = a_info.get("thrashed_files_count", 0) + a_rec = a_info.get("avg_error_recovery_turns", 1.0) + a_sess = a_info.get("sessions", 0) + badge_style = ( + "color:var(--mint);border-color:#1d3b2e;background:#0F231A" + if ak == "antigravity" + else ( + "color:var(--blue);border-color:#1e355b;background:#0d1c33" + if ak == "claude" + else "color:var(--purple, #c084fc);border-color:#3b1e5b;background:#1a0d33" + ) + ) + score_badge = ( + "color:var(--mint);border-color:#1d3b2e;background:#0F231A" + if a_score >= 80 + else ( + "color:var(--gold);border-color:#3d3014;background:#241D0E" + if a_score >= 60 + else "color:var(--crit);border-color:#4a1e17;background:#2a110e" + ) + ) + breakdown_rows.append( + f"" + f"{escape(a_info.get('label', ak))}" + f"{a_score} ({a_grade})" + f"{a_comp}%" + f"{a_v_rate}%" + f"{a_fsr}%" + f"{'' + str(a_thrash) + '' if a_thrash > 0 else '0'}" + f"{a_rec} turns" + f"{a_sess}" + f"" + ) + + # Model breakdown rows + by_model = qm.get("by_model") or [] + for m_info in by_model: + m_name = m_info.get("model", "unknown") + m_score = m_info.get("quality_score", 100) + m_grade = m_info.get("grade", "A") + m_comp = m_info.get("task_completion_rate_pct", 100.0) + m_v_rate = m_info.get("verification_rate_pct", 100.0) + m_fsr = m_info.get("first_pass_success_rate_pct", 100.0) + m_thrash = m_info.get("thrashed_files_count", 0) + m_rec = m_info.get("avg_error_recovery_turns", 1.0) + m_sess = m_info.get("sessions", 0) + score_badge = ( + "color:var(--mint);border-color:#1d3b2e;background:#0F231A" + if m_score >= 80 + else ( + "color:var(--gold);border-color:#3d3014;background:#241D0E" + if m_score >= 60 + else "color:var(--crit);border-color:#4a1e17;background:#2a110e" + ) + ) + breakdown_rows.append( + f"" + f"{escape(m_name)}" + f"{m_score} ({m_grade})" + f"{m_comp}%" + f"{m_v_rate}%" + f"{m_fsr}%" + f"{'' + str(m_thrash) + '' if m_thrash > 0 else '0'}" + f"{m_rec} turns" + f"{m_sess}" + f"" + ) + + matrix_table = "" + if breakdown_rows: + matrix_table = ( + f"
    " + f"
    " + f"ENGINE & MODEL RELIABILITY COMPARISON" + f"
    " + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"{''.join(breakdown_rows)}" + f"
    ENGINE / MODELSCORECOMPLETIONVERIFICATIONFIRST-PASS SUCCESSTHRASH FILESHEALING TURNSSESSIONS
    " + f"
    " + ) + + category_table = "" + by_category = qm.get("by_category") or {} + category_rows = [] + for ck, c_info in by_category.items(): + c_sessions = c_info.get("sessions", 0) + if c_sessions == 0: + continue + c_label = c_info.get("label", ck.capitalize()) + c_icon = c_info.get("icon", "💻") + c_desc = c_info.get("desc", "") + c_score = c_info.get("quality_score", 100) + c_grade = c_info.get("grade", "A") + c_comp = c_info.get("task_completion_rate_pct", 100.0) + c_verif = c_info.get("verification_rate_pct", 100.0) + c_fsr = c_info.get("first_pass_success_rate_pct", 100.0) + c_thrash = c_info.get("thrashed_files_count", 0) + c_share = c_info.get("share_pct", 0.0) + best_agent = c_info.get("best_agent", "—") + best_model = c_info.get("best_model", "—") + + score_badge = ( + "color:var(--mint);border-color:#1d3b2e;background:#0F231A" + if c_score >= 80 + else ( + "color:var(--gold);border-color:#3d3014;background:#241D0E" + if c_score >= 60 + else "color:var(--crit);border-color:#4a1e17;background:#2a110e" + ) + ) + category_rows.append( + f"" + f"" + f"
    " + f"{c_icon}{escape(c_label)}" + f"({c_share}%)" + f"
    " + f"
    {escape(c_desc)}
    " + f"" + f"{c_score} ({c_grade})" + f"{c_comp}%" + f"{c_verif}%" + f"{c_fsr}%" + f"{'' + str(c_thrash) + '' if c_thrash > 0 else '0'}" + f"{c_sessions}" + f"" + f"
    {escape(best_agent)}
    " + f"
    {escape(best_model)}
    " + f"" + f"" + ) + + if category_rows: + category_table = ( + f"
    " + f"
    " + f"CAPABILITY & PERFORMANCE BY CODING TASK DOMAIN" + f"
    " + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"" + f"{''.join(category_rows)}" + f"
    TASK CATEGORYSCORECOMPLETIONVERIFICATIONFIRST-PASS SUCCESSTHRASH FILESSESSIONSBEST FIT ENGINE / MODEL
    " + f"
    " + ) + + return ( + f"
    {''.join(tiles)}
    " + f"
    " + f"
    ~/ace/quality_breakdownLOCAL VERIFIED
    " + f"
    " + f"
    " + f"
    {sessions_tests} test-verified sessions
    " + f"
    {sessions_edits} editing sessions
    " + f"
    {qm.get('total_tool_calls', 0)} total tool executions
    " + f"
    {redundant_reads} redundant duplicate file reads
    " + f"
    " + f"{thrash_html}" + f"{matrix_table}" + f"{category_table}" + f"
    Measures how safely and stably coding agents operate in your repository. Correlates cost against first-pass tool correctness, test diligence, and domain task fit.
    " + f"
    " + ) + + def _fleet(f: Optional[Dict[str, Any]]) -> str: """§ 01 — the eleven fleet metrics from docs/22 §0, on this machine's transcripts. @@ -1597,6 +1886,7 @@ def _prometheus_section(d: Dict[str, Any]) -> str: # the same reasoning that keeps _sec's id keyed off the section number. _NAV = ( ("◫", "Overview", "01"), + ("🎯", "Code Quality", "02"), ("⇄", "Strategies", "04"), ("✦", "Recommendations", "06"), ("⚡", "Workflow Skills", "07"), @@ -1822,7 +2112,7 @@ def _lever_note(d: Dict[str, Any]) -> str: def _rail(d: Dict[str, Any]) -> str: - live = d["live"] + live = d.get("live") or {"turns": 0} # Each entry jumps to a section already on the page -- one document, not five views. # Anchors rather than divs: clickable without JS. nav = "".join( @@ -1981,11 +2271,23 @@ def render(d: Dict[str, Any]) -> str: # on to "what it cost" and "what to do about it". b.append(_fleet(d.get("fleet"))) - peak = h.get("peak_context") or 0 - # § 02 — spend + # § 02 — code quality & reliability b.append( _sec( "02", + "CODE QUALITY & RELIABILITY", + "Agent execution stability & test hygiene.", + "Verification rate, rework thrash, and error recovery.", + "LOCAL", + ) + ) + b.append(_quality(d.get("quality"))) + + peak = h.get("peak_context") or 0 + # § 03 — spend + b.append( + _sec( + "03", "SPEND", "Where the money goes.", "List price on your transcripts.", diff --git a/src/ace/sidecar/insights.py b/src/ace/sidecar/insights.py index 96e5063..8c0598e 100644 --- a/src/ace/sidecar/insights.py +++ b/src/ace/sidecar/insights.py @@ -148,6 +148,219 @@ def _sig(name: str, tool_input: Dict[str, Any]) -> str: return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] +_TEST_CMD_RE = re.compile( + r"\b(pytest|npm\s+(?:run\s+)?test|vitest|jest|cargo\s+test|go\s+test|dotnet\s+test|ctest|ruff|eslint|mypy|flake8|pylint|black\s+--check|tsc\s+--noEmit|bundle\s+exec\s+rspec)\b", + re.IGNORECASE, +) + +_ERR_MSG_RE = re.compile( + r"(encountered error in tool execution|the command exited with code [1-9]|exit code [1-9]|operation not permitted|command failed|fatal:|traceback \(most recent call last\)|syntaxerror|typeerror|keyerror|assertionerror|modulenotfounderror|permission denied|no such file or directory)", + re.IGNORECASE, +) + +_TEST_FILE_RE = re.compile( + r"(^|[/\\])(tests?|spec|__tests__)[/\\]|(\.|_)(test|spec)\.[a-zA-Z0-9]+$", + re.IGNORECASE, +) + +_SOURCE_FILE_EXTS = ( + ".py", + ".js", + ".jsx", + ".ts", + ".tsx", + ".go", + ".rs", + ".java", + ".c", + ".cpp", + ".h", + ".hpp", + ".cs", + ".rb", + ".php", + ".swift", + ".kt", + ".scala", + ".sh", + ".html", + ".css", + ".vue", +) + +TASK_CAT_UI = "ui" +TASK_CAT_BACKEND = "backend" +TASK_CAT_TESTING = "testing" +TASK_CAT_DOCS = "docs" +TASK_CAT_RESEARCH = "research" + +TASK_CATEGORIES: Dict[str, Dict[str, str]] = { + TASK_CAT_UI: { + "label": "UI & Frontend", + "icon": "🎨", + "desc": "User interfaces, web layouts, components, templates, and styling.", + }, + TASK_CAT_BACKEND: { + "label": "Backend & Systems", + "icon": "⚙️", + "desc": "Server logic, database models, APIs, pipelines, and algorithms.", + }, + TASK_CAT_TESTING: { + "label": "Testing & QA", + "icon": "🧪", + "desc": "Unit/integration tests, test harnesses, assertions, and linter auto-fixes.", + }, + TASK_CAT_DOCS: { + "label": "Docs & Specs", + "icon": "📝", + "desc": "Architecture documentation, implementation plans, and walkthroughs.", + }, + TASK_CAT_RESEARCH: { + "label": "Codebase Research", + "icon": "🔍", + "desc": "Codebase navigation, symbol search, architecture analysis, and comprehension.", + }, +} + +_UI_EXTS = ( + ".tsx", + ".jsx", + ".vue", + ".svelte", + ".css", + ".scss", + ".sass", + ".html", + ".svg", +) +_DOC_EXTS = (".md", ".mdx", ".rst", ".txt", ".adoc") + + +def _classify_session_task_category(session: Dict[str, Any]) -> str: + """Classifies a coding session into a primary task domain based on tool calls and target files.""" + turns = session.get("turns") or [] + edits: List[str] = [] + has_test_runs = False + + for t in turns: + for c in t.get("calls") or []: + if c.get("is_edit"): + tgt = str(c.get("raw_target") or c.get("target") or "").lower() + edits.append(tgt) + if c.get("is_test_run"): + has_test_runs = True + + if not edits: + return TASK_CAT_RESEARCH + + ui_count = sum( + 1 + for e in edits + if any(e.endswith(ext) for ext in _UI_EXTS) + or any( + k in e + for k in ( + "/ui/", + "/frontend/", + "/components/", + "/views/", + "/styles/", + "/web/", + "/templates/", + "dashboard_render", + ) + ) + ) + test_count = sum(1 for e in edits if _TEST_FILE_RE.search(e)) + doc_count = sum( + 1 + for e in edits + if any(e.endswith(ext) for ext in _DOC_EXTS) + or any(k in e for k in ("doc", "readme", "walkthrough", "plan")) + ) + backend_count = len(edits) - ui_count - test_count - doc_count + + counts = { + TASK_CAT_UI: ui_count, + TASK_CAT_TESTING: test_count, + TASK_CAT_DOCS: doc_count, + TASK_CAT_BACKEND: max(0, backend_count), + } + top_cat = max(counts.items(), key=lambda kv: kv[1]) + if top_cat[1] > 0: + return top_cat[0] + return TASK_CAT_BACKEND + + +def _classify_call(name: str, tool_input: Dict[str, Any]) -> Dict[str, Any]: + target_raw = None + for k in ( + "file_path", + "path", + "TargetFile", + "target_file", + "filename", + "file", + "notebook_path", + "AbsolutePath", + ): + v = tool_input.get(k) + if isinstance(v, str) and v: + target_raw = v + break + + cmd = None + for k in ("command", "CommandLine", "cmd", "input"): + v = tool_input.get(k) + if isinstance(v, str) and v: + cmd = v + break + + name_lower = name.lower() + is_test_run = False + if cmd and _TEST_CMD_RE.search(cmd): + is_test_run = True + + is_edit = name_lower in ( + "edit", + "str_replace_editor", + "write_to_file", + "replace_file_content", + "create_file", + "modify_file_content", + "save_file", + "patch", + ) or ( + name_lower.startswith("edit") + or name_lower.startswith("write") + or name_lower.startswith("replace") + ) + is_view = name_lower in ( + "view", + "view_file", + "read_file", + "cat", + "open_file", + "get_file_contents", + ) or (name_lower.startswith("view") or name_lower.startswith("read")) + + is_test_file = bool(target_raw and _TEST_FILE_RE.search(target_raw)) + is_src_file = bool( + target_raw + and any(target_raw.lower().endswith(ext) for ext in _SOURCE_FILE_EXTS) + and not is_test_file + ) + + return { + "raw_target": target_raw, + "is_test_run": is_test_run, + "is_edit": is_edit, + "is_view": is_view, + "is_test_file": is_test_file, + "is_src_file": is_src_file, + } + + def _measure(body: Any) -> int: """A tool_result body -> its size in byte-equivalents, with images priced as images. @@ -256,6 +469,7 @@ def _scan(root: str) -> List[Dict[str, Any]]: # tool_use_id -> short hash of the result content, so the de-dup lever can prove # "already in context" instead of inferring it from a matching path. result_digests: Dict[str, str] = {} + result_errors: Dict[str, bool] = {} seen_ids: Dict[str, int] = {} # message id -> index of its turn in `turns` cwds: List[str] = [] # The timeline `time_budget` and `parked` read: (start, kind, tool names, end). Only @@ -304,10 +518,16 @@ def _scan(root: str) -> List[Dict[str, Any]]: ): saw_result = True body = b.get("content") - result_bytes[b.get("tool_use_id")] = _measure(body) - dg = _digest(body) - if dg: - result_digests[b.get("tool_use_id")] = dg + tid = b.get("tool_use_id") + if tid: + result_bytes[tid] = _measure(body) + dg = _digest(body) + if dg: + result_digests[tid] = dg + if b.get("is_error") or str( + b.get("status", "") + ).lower() in ("error", "failed"): + result_errors[tid] = True # A tool finishing and a human typing are both "user" records and # they mean opposite things about who the session is waiting on. at = _epoch(rec.get("timestamp")) @@ -330,12 +550,19 @@ def _scan(root: str) -> List[Dict[str, Any]]: if bt == "tool_use": ti = b.get("input") or {} nm = b.get("name") or "?" + cl = _classify_call(nm, ti) calls.append( { "id": b.get("id"), "name": nm, "target": _target(ti), "sig": _sig(nm, ti), + "raw_target": cl["raw_target"], + "is_test_run": cl["is_test_run"], + "is_edit": cl["is_edit"], + "is_view": cl["is_view"], + "is_test_file": cl["is_test_file"], + "is_src_file": cl["is_src_file"], } ) turn = { @@ -382,12 +609,15 @@ def _scan(root: str) -> List[Dict[str, Any]]: for t in turns: for c in t["calls"]: cid = c.pop("id", None) - n = result_bytes.get(cid) - if n is not None: - c["result_bytes"] = n - dg = result_digests.get(cid) - if dg: - c["digest"] = dg + if cid is not None: + n = result_bytes.get(cid) + if n is not None: + c["result_bytes"] = n + dg = result_digests.get(cid) + if dg: + c["digest"] = dg + if cid in result_errors: + c["is_error"] = True # Assistant events are derived here rather than inside the loop above: a turn's # ``tool_use`` blocks can be spread across several records sharing one message id (see # the docstring), so the complete call list only exists once the join is done. Reading @@ -464,6 +694,7 @@ def _scan_antigravity(root: str) -> List[Dict[str, Any]]: result_bytes: Dict[str, int] = {} result_digests: Dict[str, str] = {} + result_errors: Dict[str, bool] = {} first_snippet: str = "" try: @@ -489,22 +720,7 @@ def _scan_antigravity(root: str) -> List[Dict[str, Any]]: if text and not text.startswith("<"): first_snippet = text[:110] events.append((at, "prompt", (), at)) - continue - - if stype in ("TOOL_RESULT", "SYSTEM_RESULT"): - body = ( - rec.get("content") or rec.get("output") or rec.get("result") - ) - tid = ( - rec.get("tool_use_id") - or rec.get("call_id") - or f"call_{len(events)}" - ) - result_bytes[tid] = _measure(body) - dg = _digest(body) - if dg: - result_digests[tid] = dg - events.append((at, "tool_result", (), at)) + current_turn_calls = None continue if ( @@ -524,13 +740,19 @@ def _scan_antigravity(root: str) -> List[Dict[str, Any]]: if isinstance(tc, dict): nm = tc.get("name") or "tool" args = tc.get("args") or tc.get("input") or {} - call_id = tc.get("id") or f"call_{len(turns)}_{idx_c}" + cl = _classify_call(nm, args) calls.append( { - "id": call_id, "name": nm, "target": _target(args), "sig": _sig(nm, args), + "raw_target": cl["raw_target"], + "is_test_run": cl["is_test_run"], + "is_edit": cl["is_edit"], + "is_view": cl["is_view"], + "is_test_file": cl["is_test_file"], + "is_src_file": cl["is_src_file"], + "is_error": False, } ) @@ -580,24 +802,40 @@ def _scan_antigravity(root: str) -> List[Dict[str, Any]]: ), } turns.append(turn) + current_turn_calls = calls if calls else None events.append( (at, "assistant", tuple(c["name"] for c in calls), at) ) + continue + + if current_turn_calls and ( + stype in ("GENERIC", "TOOL_RESULT", "SYSTEM_MESSAGE", "SYSTEM_RESULT") + or ssource in ("SYSTEM", "MODEL") + ): + body = rec.get("content") or rec.get("output") or rec.get("result") or "" + st = str(rec.get("status", "")).upper() + is_err = ( + st in ("ERROR", "FAILED") + or bool(rec.get("error")) + or bool(rec.get("is_error")) + or bool(_ERR_MSG_RE.search(str(body))) + ) + for c in current_turn_calls: + if is_err: + c["is_error"] = True + c["result_bytes"] = _measure(body) + dg = _digest(body) + if dg: + c["digest"] = dg + events.append((at, "tool_result", (), at)) + continue except Exception: continue if turns: - for t in turns: - for c in t["calls"]: - cid_tag = c.pop("id", None) - if cid_tag and cid_tag in result_bytes: - c["result_bytes"] = result_bytes[cid_tag] - if cid_tag and cid_tag in result_digests: - c["digest"] = result_digests[cid_tag] - events.sort(key=lambda e: e[0]) name = f"agy_{cid[:8]}" if cid else f"agy_{len(sessions)+1}" - session: Dict[str, Any] = { + session = { "session": name, "kind": "main", "agent_type": AGENT_ANTIGRAVITY, @@ -634,6 +872,7 @@ def _scan_codex(root: str) -> List[Dict[str, Any]]: result_bytes: Dict[str, int] = {} result_digests: Dict[str, str] = {} + result_errors: Dict[str, bool] = {} first_snippet: str = "" try: @@ -732,12 +971,19 @@ def _scan_codex(root: str) -> List[Dict[str, Any]]: or payload.get("id") or f"call_{len(turns)}_{len(current_calls)}" ) + cl = _classify_call(nm, args) current_calls.append( { "id": cid, "name": nm, "target": _target(args), "sig": _sig(nm, args), + "raw_target": cl["raw_target"], + "is_test_run": cl["is_test_run"], + "is_edit": cl["is_edit"], + "is_view": cl["is_view"], + "is_test_file": cl["is_test_file"], + "is_src_file": cl["is_src_file"], } ) continue @@ -750,6 +996,13 @@ def _scan_codex(root: str) -> List[Dict[str, Any]]: dg = _digest(out_body) if dg: result_digests[cid] = dg + if ( + payload.get("exit_code") not in (None, 0) + or bool(payload.get("is_error")) + or str(payload.get("status", "")).lower() + in ("error", "failed") + ): + result_errors[cid] = True events.append((at, "tool_result", (), at)) continue @@ -842,6 +1095,12 @@ def _scan_codex(root: str) -> List[Dict[str, Any]]: dg = _digest(body) if dg: result_digests[tid] = dg + if ( + rec.get("exit_code") not in (None, 0) + or bool(rec.get("is_error")) + or str(rec.get("status", "")).lower() in ("error", "failed") + ): + result_errors[tid] = True events.append((at, "tool_result", (), at)) continue @@ -883,12 +1142,19 @@ def _scan_codex(root: str) -> List[Dict[str, Any]]: except Exception: args = {"raw": args} call_id = tc.get("id") or f"call_{len(turns)}_{idx_c}" + cl = _classify_call(nm, args) calls.append( { "id": call_id, "name": nm, "target": _target(args), "sig": _sig(nm, args), + "raw_target": cl["raw_target"], + "is_test_run": cl["is_test_run"], + "is_edit": cl["is_edit"], + "is_view": cl["is_view"], + "is_test_file": cl["is_test_file"], + "is_src_file": cl["is_src_file"], } ) @@ -965,6 +1231,8 @@ def _scan_codex(root: str) -> List[Dict[str, Any]]: c["result_bytes"] = result_bytes[cid_tag] if cid_tag and cid_tag in result_digests: c["digest"] = result_digests[cid_tag] + if cid_tag and cid_tag in result_errors: + c["is_error"] = True if not first_snippet: try: @@ -1810,6 +2078,346 @@ def totals(sess: List[Dict[str, Any]]) -> Dict[str, Any]: return agg +# ------------------------------------------------- code quality & reliability metrics + + +def _calc_quality_block(sess: List[Dict[str, Any]]) -> Dict[str, Any]: + total_sessions = len(sess) + if not total_sessions: + return { + "available": False, + "quality_score": 100, + "grade": "A", + "task_completion_rate": 1.0, + "task_completion_rate_pct": 100.0, + "verification_rate": 1.0, + "verification_rate_pct": 100.0, + "first_pass_success_rate": 1.0, + "first_pass_success_rate_pct": 100.0, + "tool_error_rate": 0.0, + "tool_error_rate_pct": 0.0, + "total_edits": 0, + "total_tests": 0, + "total_tool_calls": 0, + "thrashed_files_count": 0, + "thrashed_files_list": [], + "rework_thrash_rate": 0.0, + "rework_thrash_rate_pct": 0.0, + "edit_stability": 1.0, + "edit_stability_pct": 100.0, + "redundant_reads_count": 0, + "avg_error_recovery_turns": 1.0, + "test_to_code_ratio": 1.0, + "sessions_with_edits": 0, + "sessions_with_tests": 0, + "clean_completed_sessions": 0, + } + + sessions_with_edits = 0 + sessions_with_tests = 0 + clean_completed_sessions = 0 + total_edits = 0 + total_tests = 0 + total_tool_calls = 0 + failed_tool_calls = 0 + redundant_reads_count = 0 + test_edits_count = 0 + src_edits_count = 0 + + all_thrashed_files = set() + recovery_turns_list = [] + + for s in sess: + turns = s.get("turns") or [] + session_has_edit = False + session_has_test = False + session_file_edits: Dict[str, int] = {} + last_view_sig: Optional[str] = None + pending_error_turn: Optional[int] = None + last_turn_had_error = False + + for turn_idx, t in enumerate(turns): + turn_has_error = False + for c in t.get("calls") or []: + total_tool_calls += 1 + is_err = bool(c.get("is_error")) + if is_err: + failed_tool_calls += 1 + turn_has_error = True + + if c.get("is_test_run"): + session_has_test = True + total_tests += 1 + + if c.get("is_edit"): + session_has_edit = True + total_edits += 1 + raw_t = c.get("raw_target") or c.get("target") or "unknown" + is_artifact = ( + "/.gemini/antigravity/brain/" in raw_t + or "/.system_generated/" in raw_t + or raw_t.endswith("walkthrough.md") + or raw_t.endswith("implementation_plan.md") + ) + if not is_artifact: + session_file_edits[raw_t] = session_file_edits.get(raw_t, 0) + 1 + if c.get("is_test_file"): + test_edits_count += 1 + elif c.get("is_src_file"): + src_edits_count += 1 + # A file edit invalidates previous view cache + last_view_sig = None + + if c.get("is_view"): + view_sig = ( + c.get("sig") + or c.get("digest") + or c.get("raw_target") + or c.get("target") + ) + if view_sig and view_sig == last_view_sig: + redundant_reads_count += 1 + last_view_sig = view_sig + + if turn_has_error: + if pending_error_turn is None: + pending_error_turn = turn_idx + elif pending_error_turn is not None: + recovery_turns_list.append(max(1, turn_idx - pending_error_turn)) + pending_error_turn = None + + if turns and any(c.get("is_error") for c in turns[-1].get("calls") or []): + last_turn_had_error = True + + if session_has_edit: + sessions_with_edits += 1 + if session_has_test and not last_turn_had_error: + clean_completed_sessions += 1 + else: + if not last_turn_had_error: + clean_completed_sessions += 1 + + if session_has_test: + sessions_with_tests += 1 + + for fpath, count in session_file_edits.items(): + if count >= 3: + all_thrashed_files.add(fpath) + + thrashed_files_count = len(all_thrashed_files) + verification_rate = ( + (sessions_with_tests / sessions_with_edits) + if sessions_with_edits > 0 + else (1.0 if not total_edits else 0.0) + ) + + first_pass_success_rate = ( + ((total_tool_calls - failed_tool_calls) / total_tool_calls) + if total_tool_calls > 0 + else 1.0 + ) + + tool_error_rate = ( + (failed_tool_calls / total_tool_calls) + if total_tool_calls > 0 + else 0.0 + ) + + task_completion_rate = ( + (clean_completed_sessions / total_sessions) + if total_sessions > 0 + else 1.0 + ) + + rework_thrash_rate = ( + (thrashed_files_count / max(1, len(all_thrashed_files) + total_edits)) + if total_edits > 0 + else 0.0 + ) + + thrash_ratio = (thrashed_files_count / max(1, sessions_with_edits)) if sessions_with_edits > 0 else 0.0 + edit_stability = max(0.0, 1.0 - (thrash_ratio * 1.0)) + + test_to_code_ratio = ( + (test_edits_count / src_edits_count) + if src_edits_count > 0 + else (1.0 if test_edits_count > 0 else 0.5) + ) + + avg_error_recovery_turns = ( + (sum(recovery_turns_list) / len(recovery_turns_list)) + if recovery_turns_list + else 1.0 + ) + + # Balanced 0-100 score: + # 35% Verified Task Completion, 30% Verification Diligence, 20% First-Pass Tool Success, 15% Edit Stability (Thrash-Free) + raw_score = ( + 0.35 * task_completion_rate + + 0.30 * verification_rate + + 0.20 * first_pass_success_rate + + 0.15 * edit_stability + ) * 100.0 + + quality_score = max(0, min(100, int(round(raw_score)))) + if quality_score >= 90: + grade = "A" + elif quality_score >= 80: + grade = "B" + elif quality_score >= 70: + grade = "C" + elif quality_score >= 60: + grade = "D" + else: + grade = "F" + + return { + "available": True, + "quality_score": quality_score, + "grade": grade, + "task_completion_rate": round(task_completion_rate, 4), + "task_completion_rate_pct": round(task_completion_rate * 100.0, 1), + "verification_rate": round(verification_rate, 4), + "verification_rate_pct": round(verification_rate * 100.0, 1), + "first_pass_success_rate": round(first_pass_success_rate, 4), + "first_pass_success_rate_pct": round(first_pass_success_rate * 100.0, 1), + "tool_error_rate": round(tool_error_rate, 4), + "tool_error_rate_pct": round(tool_error_rate * 100.0, 1), + "total_edits": total_edits, + "total_tests": total_tests, + "total_tool_calls": total_tool_calls, + "thrashed_files_count": thrashed_files_count, + "thrashed_files_list": sorted(list(all_thrashed_files))[:10], + "rework_thrash_rate": round(rework_thrash_rate, 4), + "rework_thrash_rate_pct": round(rework_thrash_rate * 100.0, 1), + "edit_stability": round(edit_stability, 4), + "edit_stability_pct": round(edit_stability * 100.0, 1), + "redundant_reads_count": redundant_reads_count, + "avg_error_recovery_turns": round(avg_error_recovery_turns, 1), + "test_to_code_ratio": round(test_to_code_ratio, 2), + "sessions_with_edits": sessions_with_edits, + "sessions_with_tests": sessions_with_tests, + "clean_completed_sessions": clean_completed_sessions, + } + + +def quality_metrics(sess: List[Dict[str, Any]]) -> Dict[str, Any]: + """Calculates unified code quality, verification hygiene, and reliability metrics. + + Includes top-line metrics along with breakdowns: + - by_agent: Quality scores partitioned per agent engine (Claude Code, Antigravity, Codex). + - by_model: Quality scores partitioned per LLM model. + - by_category: Quality and capability metrics partitioned per coding task domain (UI, Backend, Testing, Docs, Research). + """ + overall = _calc_quality_block(sess) + if not sess: + overall["by_agent"] = {} + overall["by_model"] = [] + overall["by_category"] = {} + return overall + + # Group by agent + by_agent: Dict[str, Any] = {} + agent_groups: Dict[str, List[Dict[str, Any]]] = {} + for s in sess: + ak = s.get("agent_type") or AGENT_CLAUDE + agent_groups.setdefault(ak, []).append(s) + + for ak, a_sess in agent_groups.items(): + block = _calc_quality_block(a_sess) + by_agent[ak] = { + "agent": ak, + "label": AGENTS.get(ak, ak.capitalize()), + "sessions": len(a_sess), + **block, + } + + # Group by model + model_sessions: Dict[str, List[Dict[str, Any]]] = {} + for s in sess: + models_in_s = set(t.get("model") for t in s.get("turns", []) if t.get("model")) + for m in models_in_s: + projected_turns = [t for t in s.get("turns", []) if t.get("model") == m] + if projected_turns: + model_sessions.setdefault(m, []).append( + { + "session": s.get("session"), + "agent_type": s.get("agent_type"), + "turns": projected_turns, + "events": s.get("events", []), + } + ) + + by_model: List[Dict[str, Any]] = [] + for m_name, m_sess in sorted(model_sessions.items(), key=lambda kv: -len(kv[1])): + block = _calc_quality_block(m_sess) + by_model.append( + { + "model": m_name, + "sessions": len(m_sess), + **block, + } + ) + + # Group by task category + by_category: Dict[str, Any] = {} + category_groups: Dict[str, List[Dict[str, Any]]] = {} + for s in sess: + ck = _classify_session_task_category(s) + category_groups.setdefault(ck, []).append(s) + + for ck, cat_meta in TASK_CATEGORIES.items(): + c_sess = category_groups.get(ck) or [] + block = _calc_quality_block(c_sess) + + best_agent = "—" + best_agent_score = -1 + for ak in (AGENT_CLAUDE, AGENT_ANTIGRAVITY, AGENT_CODEX): + sub_ak = [s for s in c_sess if (s.get("agent_type") or AGENT_CLAUDE) == ak] + if sub_ak: + sc = _calc_quality_block(sub_ak)["quality_score"] + if sc > best_agent_score: + best_agent_score = sc + best_agent = AGENTS.get(ak, ak.capitalize()) + + best_model = "—" + best_model_score = -1 + cat_models = set( + t.get("model") + for s in c_sess + for t in s.get("turns", []) + if t.get("model") and not str(t.get("model")).startswith("<") + ) + for m in cat_models: + sub_m = [] + for s in c_sess: + proj = [t for t in s.get("turns", []) if t.get("model") == m] + if proj: + sub_m.append({"turns": proj, "events": s.get("events", [])}) + if sub_m: + sc = _calc_quality_block(sub_m)["quality_score"] + if sc > best_model_score: + best_model_score = sc + best_model = m + + by_category[ck] = { + "category": ck, + "label": cat_meta["label"], + "icon": cat_meta["icon"], + "desc": cat_meta["desc"], + "sessions": len(c_sess), + "share_pct": round(len(c_sess) / max(1, len(sess)) * 100.0, 1), + "best_agent": best_agent, + "best_model": best_model, + **block, + } + + overall["by_agent"] = by_agent + overall["by_model"] = by_model + overall["by_category"] = by_category + return overall + + # ------------------------------------------------- time budget (docs/analysis_docs §1 and §2) # Declared order is presentation order for ties; the payload sorts by size. @@ -2140,6 +2748,37 @@ def recommendations( ) ) + # 4. Code Quality & Test Hygiene Recommendations + if sess is not None: + qm = quality_metrics(sess) + if ( + qm.get("sessions_with_edits", 0) > 0 + and qm.get("verification_rate", 1.0) < 0.5 + ): + recs.append( + _rec( + "Low test verification hygiene in agent sessions", + f"Only {qm.get('verification_rate_pct', 0)}% of sessions with code edits ran automated test suites. " + f"Running test/lint passes before finishing turns reduces runtime bugs and catches regressions early.", + f"{qm.get('sessions_with_tests', 0)} of {qm.get('sessions_with_edits', 0)} editing sessions verified", + "LOW", + f"{round((1.0 - qm.get('verification_rate', 0.0)) * 100, 1)}% unverified", + "of editing sessions", + ) + ) + if qm.get("thrashed_files_count", 0) >= 3: + recs.append( + _rec( + f"File edit thrashing detected on {qm.get('thrashed_files_count')} files", + "Agent modified the same files 3+ times in single sessions. Providing more explicit prompt instructions, " + "specifying test fixtures, or decomposing tasks into smaller subagents reduces edit churn.", + f"{qm.get('thrashed_files_count')} thrashed files across scope", + "MED", + f"{qm.get('rework_thrash_rate_pct')}% thrash", + "churn rate", + ) + ) + if not recs: recs.append( _rec( @@ -2471,6 +3110,7 @@ def _build_payload( "agent_breakdown": ab, "span": span(range_key, window, agg), "historical": agg, + "quality": quality_metrics(scoped), "time": time_budget(scoped), "parked": parked(scoped), "fleet": _fleet, @@ -2686,6 +3326,60 @@ def format_prometheus_metrics(d: Dict[str, Any]) -> str: lines.append("# TYPE ace_installed_skills_total gauge") lines.append(f'ace_installed_skills_total {len(skills)}') + # Code Quality & Reliability Metrics + qm = d.get("quality") or {} + lines.append("# HELP ace_quality_score Composite code quality and verification score (0-100).") + lines.append("# TYPE ace_quality_score gauge") + lines.append(f'ace_quality_score{{agent="all"}} {qm.get("quality_score", 100)}') + for agent_id, q_info in (qm.get("by_agent") or {}).items(): + lines.append(f'ace_quality_score{{agent="{agent_id}"}} {q_info.get("quality_score", 100)}') + for m_info in (qm.get("by_model") or []): + m_name = m_info.get("model", "unknown") + lines.append(f'ace_quality_score{{model="{m_name}"}} {m_info.get("quality_score", 100)}') + + lines.append("# HELP ace_quality_verification_rate Share of edited sessions that ran automated tests or linters.") + lines.append("# TYPE ace_quality_verification_rate gauge") + lines.append(f'ace_quality_verification_rate{{agent="all"}} {qm.get("verification_rate", 1.0)}') + for agent_id, q_info in (qm.get("by_agent") or {}).items(): + lines.append(f'ace_quality_verification_rate{{agent="{agent_id}"}} {q_info.get("verification_rate", 1.0)}') + + lines.append("# HELP ace_quality_first_pass_success_rate Share of tool calls that succeeded on first pass.") + lines.append("# TYPE ace_quality_first_pass_success_rate gauge") + lines.append(f'ace_quality_first_pass_success_rate{{agent="all"}} {qm.get("first_pass_success_rate", 1.0)}') + for agent_id, q_info in (qm.get("by_agent") or {}).items(): + lines.append(f'ace_quality_first_pass_success_rate{{agent="{agent_id}"}} {q_info.get("first_pass_success_rate", 1.0)}') + + lines.append("# HELP ace_quality_tool_error_rate Share of tool executions that returned errors.") + lines.append("# TYPE ace_quality_tool_error_rate gauge") + lines.append(f'ace_quality_tool_error_rate {qm.get("tool_error_rate", 0.0)}') + + lines.append("# HELP ace_quality_thrashed_files_total Number of files edited 3 or more times in a single session.") + lines.append("# TYPE ace_quality_thrashed_files_total counter") + lines.append(f'ace_quality_thrashed_files_total {qm.get("thrashed_files_count", 0)}') + + lines.append("# HELP ace_quality_redundant_reads_total Count of consecutive duplicate file reads.") + lines.append("# TYPE ace_quality_redundant_reads_total counter") + lines.append(f'ace_quality_redundant_reads_total {qm.get("redundant_reads_count", 0)}') + + lines.append("# HELP ace_quality_error_recovery_turns_avg Average turns to recover from an execution error.") + lines.append("# TYPE ace_quality_error_recovery_turns_avg gauge") + lines.append(f'ace_quality_error_recovery_turns_avg {qm.get("avg_error_recovery_turns", 1.0)}') + + lines.append("# HELP ace_quality_test_to_code_ratio Ratio of test file edits to source file edits.") + lines.append("# TYPE ace_quality_test_to_code_ratio gauge") + lines.append(f'ace_quality_test_to_code_ratio {qm.get("test_to_code_ratio", 1.0)}') + + # Task Category Performance Metrics + lines.append("# HELP ace_quality_category_score Quality score partitioned by task category.") + lines.append("# TYPE ace_quality_category_score gauge") + for cat_id, cat_info in (qm.get("by_category") or {}).items(): + lines.append(f'ace_quality_category_score{{category="{cat_id}"}} {cat_info.get("quality_score", 100)}') + + lines.append("# HELP ace_quality_category_completion_rate Task completion rate partitioned by task category.") + lines.append("# TYPE ace_quality_category_completion_rate gauge") + for cat_id, cat_info in (qm.get("by_category") or {}).items(): + lines.append(f'ace_quality_category_completion_rate{{category="{cat_id}"}} {cat_info.get("task_completion_rate", 1.0)}') + return "\n".join(lines) + "\n" diff --git a/tests/test_quality_metrics.py b/tests/test_quality_metrics.py new file mode 100644 index 0000000..33b5db3 --- /dev/null +++ b/tests/test_quality_metrics.py @@ -0,0 +1,496 @@ +"""Tests for ace.sidecar code quality, verification hygiene, and reliability metrics.""" + +from __future__ import annotations + +from typing import Any, Dict, List + +import pytest + +from ace.sidecar.dashboard_render import render +from ace.sidecar.insights import ( + _classify_call, + _build_payload, + format_prometheus_metrics, + quality_metrics, +) + + +def test_classify_call_test_commands() -> None: + # Pytest + cl = _classify_call("Bash", {"command": "pytest -v tests/"}) + assert cl["is_test_run"] is True + assert cl["is_edit"] is False + + # npm test + cl = _classify_call("run_command", {"CommandLine": "npm test"}) + assert cl["is_test_run"] is True + + # ruff check + cl = _classify_call("Bash", {"command": "ruff check --fix ."}) + assert cl["is_test_run"] is True + + # cargo test + cl = _classify_call("exec_command", {"command": "cargo test --all"}) + assert cl["is_test_run"] is True + + # non-test bash command + cl = _classify_call("Bash", {"command": "git status"}) + assert cl["is_test_run"] is False + + +def test_classify_call_edits_and_views() -> None: + # Edit tool + cl = _classify_call("Edit", {"file_path": "src/main.py"}) + assert cl["is_edit"] is True + assert cl["is_view"] is False + assert cl["raw_target"] == "src/main.py" + assert cl["is_src_file"] is True + assert cl["is_test_file"] is False + + # write_to_file on test file + cl = _classify_call("write_to_file", {"TargetFile": "/app/tests/test_api.py"}) + assert cl["is_edit"] is True + assert cl["is_test_file"] is True + assert cl["is_src_file"] is False + + # replace_file_content + cl = _classify_call("replace_file_content", {"TargetFile": "web/app.tsx"}) + assert cl["is_edit"] is True + assert cl["is_src_file"] is True + + # View / read_file + cl = _classify_call("view_file", {"AbsolutePath": "/app/README.md"}) + assert cl["is_view"] is True + assert cl["is_edit"] is False + assert cl["raw_target"] == "/app/README.md" + + +def test_quality_metrics_empty() -> None: + qm = quality_metrics([]) + assert qm["available"] is False + assert qm["quality_score"] == 100 + assert qm["grade"] == "A" + assert qm["verification_rate_pct"] == 100.0 + assert qm["first_pass_success_rate_pct"] == 100.0 + assert qm["thrashed_files_count"] == 0 + + +def test_quality_metrics_clean_verified_session() -> None: + sess: List[Dict[str, Any]] = [ + { + "session": "s1", + "agent_type": "claude", + "turns": [ + { + "model": "claude-sonnet-4-6", + "calls": [ + { + "name": "view_file", + "raw_target": "src/app.py", + "sig": "sig_v1", + "is_view": True, + "is_edit": False, + "is_test_run": False, + }, + { + "name": "replace_file_content", + "raw_target": "src/app.py", + "sig": "sig_e1", + "is_view": False, + "is_edit": True, + "is_test_run": False, + "is_src_file": True, + "is_test_file": False, + }, + { + "name": "write_to_file", + "raw_target": "tests/test_app.py", + "sig": "sig_e2", + "is_view": False, + "is_edit": True, + "is_test_run": False, + "is_src_file": False, + "is_test_file": True, + }, + { + "name": "Bash", + "raw_target": "pytest", + "sig": "sig_t1", + "is_view": False, + "is_edit": False, + "is_test_run": True, + "is_error": False, + }, + ], + } + ], + } + ] + qm = quality_metrics(sess) + assert qm["available"] is True + assert qm["verification_rate"] == 1.0 + assert qm["verification_rate_pct"] == 100.0 + assert qm["first_pass_success_rate"] == 1.0 + assert qm["tool_error_rate"] == 0.0 + assert qm["thrashed_files_count"] == 0 + assert qm["redundant_reads_count"] == 0 + assert qm["sessions_with_edits"] == 1 + assert qm["sessions_with_tests"] == 1 + assert qm["quality_score"] >= 90 + assert qm["grade"] == "A" + + +def test_quality_metrics_unverified_and_thrashed_session() -> None: + # 1 session editing a file 4 times (thrashing), zero tests, 1 tool error + sess: List[Dict[str, Any]] = [ + { + "session": "s2", + "agent_type": "antigravity", + "turns": [ + { + "model": "gemini-3.6-flash", + "calls": [ + { + "name": "view_file", + "raw_target": "src/flaky.py", + "sig": "v1", + "is_view": True, + }, + { + "name": "view_file", + "raw_target": "src/flaky.py", + "sig": "v1", + "is_view": True, # Redundant read + }, + { + "name": "edit", + "raw_target": "src/flaky.py", + "is_edit": True, + "is_src_file": True, + "is_error": True, # Error 1 + }, + ], + }, + { + "model": "gemini-3.6-flash", + "calls": [ + { + "name": "edit", + "raw_target": "src/flaky.py", + "is_edit": True, + "is_src_file": True, + }, + { + "name": "edit", + "raw_target": "src/flaky.py", + "is_edit": True, + "is_src_file": True, + }, + { + "name": "edit", + "raw_target": "src/flaky.py", + "is_edit": True, + "is_src_file": True, + }, + ], + }, + ], + } + ] + qm = quality_metrics(sess) + assert qm["available"] is True + assert qm["verification_rate"] == 0.0 # Zero tests run + assert qm["sessions_with_edits"] == 1 + assert qm["sessions_with_tests"] == 0 + assert qm["thrashed_files_count"] == 1 + assert "src/flaky.py" in qm["thrashed_files_list"] + assert qm["redundant_reads_count"] == 1 + assert qm["tool_error_rate"] > 0.0 + assert qm["quality_score"] < 70 + assert qm["grade"] in ("C", "D", "F") + + +def test_quality_in_payload_and_prometheus() -> None: + sess: List[Dict[str, Any]] = [ + { + "session": "s1", + "agent_type": "claude", + "cwds": ["/test/repo"], + "turns": [ + { + "model": "claude-sonnet-4-6", + "input_tokens": 1000, + "output_tokens": 200, + "cache_read_input_tokens": 800, + "cache_creation_input_tokens": 100, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "blocks": {"text": 1, "tool_use": 2}, + "calls": [ + { + "name": "Edit", + "raw_target": "app.py", + "is_edit": True, + "is_src_file": True, + }, + { + "name": "Bash", + "command": "pytest", + "is_test_run": True, + }, + ], + "ts": "2026-08-27T12:00:00Z", + } + ], + "events": [ + (1787832000.0, "prompt", (), 1787832000.0), + (1787832005.0, "assistant", ("Edit", "Bash"), 1787832005.0), + ], + "path": "/dummy/s1.jsonl", + "bytes": 500, + "mtime": 1787832010.0, + "snippet": "fix bug and test", + } + ] + + payload = _build_payload(sess, capture=None, range_key="all", agent="all", store_path=None) + assert "quality" in payload + qm = payload["quality"] + assert qm["available"] is True + assert qm["verification_rate_pct"] == 100.0 + assert qm["quality_score"] >= 80 + + # Prometheus export check + prom_text = format_prometheus_metrics(payload) + assert "ace_quality_score" in prom_text + assert 'ace_quality_verification_rate{agent="all"} 1.0' in prom_text + assert 'ace_quality_first_pass_success_rate{agent="all"} 1.0' in prom_text + assert "ace_quality_thrashed_files_total 0" in prom_text + assert "ace_quality_redundant_reads_total 0" in prom_text + + # Render dashboard check + html = render(payload) + assert "CODE QUALITY & RELIABILITY" in html or "CODE QUALITY & RELIABILITY" in html + assert "quality_score" in html + assert "verification_rate" in html + assert "first_pass_success" in html + + +def test_quality_metrics_by_agent_and_model() -> None: + sess: List[Dict[str, Any]] = [ + # Session 1: Claude using Sonnet - verified, high quality + { + "session": "s1", + "agent_type": "claude", + "cwds": ["/test"], + "turns": [ + { + "model": "claude-sonnet-4-6", + "input_tokens": 1000, + "output_tokens": 200, + "cache_read_input_tokens": 800, + "cache_creation_input_tokens": 0, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "calls": [ + {"name": "Edit", "raw_target": "src/a.py", "is_edit": True, "is_src_file": True}, + {"name": "Bash", "command": "pytest", "is_test_run": True, "is_error": False}, + ], + } + ], + "events": [], + }, + # Session 2: Antigravity using Gemini Flash - unverified, error + { + "session": "s2", + "agent_type": "antigravity", + "cwds": ["/test"], + "turns": [ + { + "model": "gemini-3.6-flash", + "input_tokens": 500, + "output_tokens": 100, + "cache_read_input_tokens": 300, + "cache_creation_input_tokens": 0, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "calls": [ + {"name": "write_to_file", "raw_target": "src/b.py", "is_edit": True, "is_src_file": True, "is_error": True}, + ], + } + ], + "events": [], + }, + ] + + qm = quality_metrics(sess) + assert "by_agent" in qm + assert "claude" in qm["by_agent"] + assert "antigravity" in qm["by_agent"] + + claude_q = qm["by_agent"]["claude"] + assert claude_q["verification_rate_pct"] == 100.0 + assert claude_q["quality_score"] >= 85 + + agy_q = qm["by_agent"]["antigravity"] + assert agy_q["verification_rate_pct"] == 0.0 + assert agy_q["first_pass_success_rate_pct"] == 0.0 + + assert "by_model" in qm + models = [m["model"] for m in qm["by_model"]] + assert "claude-sonnet-4-6" in models + assert "gemini-3.6-flash" in models + + # Dashboard render with comparative table + payload = _build_payload(sess, capture=None, range_key="all", agent="all", store_path=None) + html = render(payload) + assert "Claude Code" in html or "claude" in html + assert "claude-sonnet-4-6" in html + assert "gemini-3.6-flash" in html + assert "ENGINE / MODEL" in html + + +def test_quality_metrics_by_task_category() -> None: + from ace.sidecar.insights import ( + _classify_session_task_category, + TASK_CAT_UI, + TASK_CAT_BACKEND, + TASK_CAT_TESTING, + TASK_CAT_DOCS, + TASK_CAT_RESEARCH, + ) + + # 1. UI session + s_ui = { + "session": "ui_sess", + "turns": [ + { + "input_tokens": 100, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "output_tokens": 50, + "model": "claude-sonnet-4-6", + "calls": [ + {"name": "write_to_file", "raw_target": "src/components/Header.tsx", "is_edit": True}, + {"name": "write_to_file", "raw_target": "src/styles/app.css", "is_edit": True}, + ], + } + ], + "events": [], + } + assert _classify_session_task_category(s_ui) == TASK_CAT_UI + + # 2. Testing session + s_test = { + "session": "test_sess", + "turns": [ + { + "input_tokens": 100, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "output_tokens": 50, + "model": "claude-sonnet-4-6", + "calls": [ + {"name": "write_to_file", "raw_target": "tests/test_api.py", "is_edit": True, "is_test_file": True}, + {"name": "Bash", "command": "pytest", "is_test_run": True}, + ], + } + ], + "events": [], + } + assert _classify_session_task_category(s_test) == TASK_CAT_TESTING + + # 3. Docs session + s_docs = { + "session": "docs_sess", + "turns": [ + { + "input_tokens": 100, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "output_tokens": 50, + "model": "claude-sonnet-4-6", + "calls": [ + {"name": "write_to_file", "raw_target": "docs/architecture.md", "is_edit": True}, + {"name": "replace_file_content", "raw_target": "README.md", "is_edit": True}, + ], + } + ], + "events": [], + } + assert _classify_session_task_category(s_docs) == TASK_CAT_DOCS + + # 4. Research session (read-only) + s_res = { + "session": "res_sess", + "turns": [ + { + "input_tokens": 100, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "output_tokens": 50, + "model": "gemini-3.6-flash", + "calls": [ + {"name": "view_file", "raw_target": "src/server.py", "is_view": True}, + {"name": "grep_search", "raw_target": "main", "is_view": False}, + ], + } + ], + "events": [], + } + assert _classify_session_task_category(s_res) == TASK_CAT_RESEARCH + + # 5. Backend session + s_back = { + "session": "back_sess", + "turns": [ + { + "input_tokens": 100, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + "output_tokens": 50, + "model": "gpt-5.3-codex", + "calls": [ + {"name": "replace_file_content", "raw_target": "src/ace/gateway/proxy.py", "is_edit": True, "is_src_file": True}, + ], + } + ], + "events": [], + } + assert _classify_session_task_category(s_back) == TASK_CAT_BACKEND + + # Multi-session aggregation in quality_metrics + sess_all = [s_ui, s_test, s_docs, s_res, s_back] + qm = quality_metrics(sess_all) + assert "by_category" in qm + assert TASK_CAT_UI in qm["by_category"] + assert TASK_CAT_BACKEND in qm["by_category"] + assert TASK_CAT_TESTING in qm["by_category"] + assert TASK_CAT_DOCS in qm["by_category"] + assert TASK_CAT_RESEARCH in qm["by_category"] + + ui_cat = qm["by_category"][TASK_CAT_UI] + assert ui_cat["sessions"] == 1 + assert ui_cat["label"] == "UI & Frontend" + + # Prometheus export + payload = _build_payload(sess_all, capture=None, range_key="all", agent="all", store_path=None) + prom_text = format_prometheus_metrics(payload) + assert 'ace_quality_category_score{category="ui"}' in prom_text + assert 'ace_quality_category_completion_rate{category="ui"}' in prom_text + + # Render dashboard + html = render(payload) + assert "CAPABILITY & PERFORMANCE BY CODING TASK DOMAIN" in html or "CAPABILITY & PERFORMANCE BY CODING TASK DOMAIN" in html + assert "UI & Frontend" in html or "UI & Frontend" in html +