From fac15545b09d649d792896bd7c6ae289647685eb Mon Sep 17 00:00:00 2001 From: yaojingang Date: Sun, 14 Jun 2026 11:42:54 +0800 Subject: [PATCH] feat: derive atlas no-route opportunities from telemetry --- scripts/build_skill_atlas.py | 62 ++++++++++++++++++++++++++++++++++-- tests/verify_skill_atlas.py | 15 +++++++-- 2 files changed, 72 insertions(+), 5 deletions(-) diff --git a/scripts/build_skill_atlas.py b/scripts/build_skill_atlas.py index d88a2b7..7d8344c 100644 --- a/scripts/build_skill_atlas.py +++ b/scripts/build_skill_atlas.py @@ -451,7 +451,7 @@ def owner_review_gaps(skills: list[dict[str, Any]]) -> list[dict[str, Any]]: return gaps -def no_route_opportunities(workspace_root: Path) -> list[dict[str, Any]]: +def failure_case_no_route_opportunities(workspace_root: Path) -> list[dict[str, Any]]: opportunities = [] for path in sorted(workspace_root.rglob("failure-cases.md")): if should_skip(path, workspace_root): @@ -463,10 +463,51 @@ def no_route_opportunities(workspace_root: Path) -> list[dict[str, Any]]: continue lowered = stripped.casefold() if "no_route" in lowered or "no route" in lowered or "missed" in lowered or "under-trigger" in lowered: - opportunities.append({"source": safe_rel(workspace_root, path), "note": stripped.lstrip("- ").strip()}) + opportunities.append( + { + "source_type": "failure-case", + "source": safe_rel(workspace_root, path), + "note": stripped.lstrip("- ").strip(), + "actionable": True, + "privacy_contract": "source note only; raw prompts are not required", + } + ) return opportunities[:50] +def telemetry_no_route_opportunities(drift_signals: list[dict[str, Any]]) -> list[dict[str, Any]]: + opportunities = [] + for signal in drift_signals: + signal_types = {str(item) for item in signal.get("signal_types", [])} + if not {"missed trigger", "under trigger"} & signal_types: + continue + summary = signal.get("summary", {}) if isinstance(signal.get("summary"), dict) else {} + opportunities.append( + { + "source_type": "telemetry", + "source": str(signal.get("source", "")), + "skill": str(signal.get("name", "")), + "path": str(signal.get("path", "")), + "signal": "missed trigger", + "missed_trigger_count": int(summary.get("missed_trigger_count") or 0), + "recommendation": str( + signal.get("recommendation") + or "Add missed prompts to trigger eval and evaluate whether a new skill route is needed." + ), + "actionable": bool(signal.get("actionable")), + "scope": str(signal.get("scope", "")), + "privacy_contract": "metadata-only telemetry; no raw prompt, output, transcript, or note is stored", + } + ) + return opportunities + + +def no_route_opportunities(workspace_root: Path, drift_signals: list[dict[str, Any]]) -> list[dict[str, Any]]: + opportunities = failure_case_no_route_opportunities(workspace_root) + opportunities.extend(telemetry_no_route_opportunities(drift_signals)) + return opportunities[:80] + + def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) fields = ["skill_a", "skill_b", "path_a", "path_b", "score", "status", "actionable", "scope_a", "scope_b"] @@ -502,6 +543,16 @@ def render_html(payload: dict[str, Any]) -> str: f"
  • {html.escape(item.get('name', item.get('skill_a', 'issue')))} {html.escape(item.get('reason', item.get('status', ', '.join(item.get('missing', item.get('signal_types', []))))))}
    • " for item in blockers ) + opportunity_items = "".join( + ( + "
  • " + f"{html.escape(item.get('skill') or item.get('source_type', 'opportunity'))} " + f"{html.escape(item.get('note') or item.get('recommendation') or item.get('signal', 'no-route opportunity'))}" + f"
    {html.escape(item.get('source', ''))}" + "
    • " + ) + for item in payload["no_route_opportunities"][:20] + ) return f""" @@ -540,6 +591,11 @@ def render_html(payload: dict[str, Any]) -> str:

    Actionable Issues

    +
    +

    No-Route Opportunities

    +

    Missed-trigger telemetry and explicit failure cases become candidate routing work without storing raw prompts or outputs.

    + +

    Full Portfolio Counts

    All scanned skills remain visible: {summary['route_collision_count']} total route collisions, {summary['owner_gap_count']} total owner gaps, {summary['stale_count']} total stale signals, and {summary['drift_signal_count']} telemetry drift signals.

    @@ -575,7 +631,7 @@ def build_atlas(workspace_root: Path, output_dir: Path, report_html: Path, repor graph = dependency_graph(skills) stale = stale_skills(skills, today) owner_gaps = owner_review_gaps(skills) - opportunities = no_route_opportunities(workspace_root) + opportunities = no_route_opportunities(workspace_root, drift_signals) actionable_skills = [skill for skill in skills if skill.get("actionable")] actionable_collisions = [item for item in collisions if item.get("actionable")] actionable_stale = [item for item in stale if item.get("actionable")] diff --git a/tests/verify_skill_atlas.py b/tests/verify_skill_atlas.py index 8d2e979..92bed68 100644 --- a/tests/verify_skill_atlas.py +++ b/tests/verify_skill_atlas.py @@ -190,7 +190,7 @@ def main() -> None: assert summary["owner_gap_count"] >= 1, summary assert summary["stale_count"] >= 1, summary assert summary["shared_resource_count"] >= 1, summary - assert summary["no_route_opportunity_count"] >= 1, summary + assert summary["no_route_opportunity_count"] >= 2, summary assert summary["drift_signal_count"] == 2, summary assert summary["actionable_drift_signal_count"] == 1, summary assert summary["telemetry_report_count"] == 1, summary @@ -200,13 +200,15 @@ def main() -> None: assert (output_dir / "stale_skills.json").exists(), output_dir assert (output_dir / "owner_review_gaps.json").exists(), output_dir assert (output_dir / "drift_signals.json").exists(), output_dir - assert (output_dir / "no_route_opportunities.json").exists(), output_dir + no_route_path = output_dir / "no_route_opportunities.json" + assert no_route_path.exists(), output_dir html = report_html.read_text(encoding="utf-8") assert "Skill Atlas" in html, html[:1000] assert "Route Collisions" in html, html[:3000] assert "Actionable Issues" in html, html[:3000] assert "Drift Signals" in html, html[:3000] assert "No-Route Opportunities" in html, html[:3000] + assert "Missed-trigger telemetry" in html, html catalog = json.loads((output_dir / "catalog.json").read_text(encoding="utf-8")) assert catalog["summary"]["skill_count"] == 3, catalog alpha_catalog = next(item for item in catalog["skills"] if item["name"] == alpha.name) @@ -214,6 +216,15 @@ def main() -> None: drift = json.loads((output_dir / "drift_signals.json").read_text(encoding="utf-8")) assert any(item["name"] == alpha.name and item["signal_types"] == ["no telemetry"] for item in drift), drift assert any(item["name"] == beta.name and "bad output" in item["signal_types"] and not item["actionable"] for item in drift), drift + no_route = json.loads(no_route_path.read_text(encoding="utf-8")) + assert {item["source_type"] for item in no_route} >= {"failure-case", "telemetry"}, no_route + telemetry_opportunity = next(item for item in no_route if item["source_type"] == "telemetry") + assert telemetry_opportunity["skill"] == beta.name, telemetry_opportunity + assert telemetry_opportunity["signal"] == "missed trigger", telemetry_opportunity + assert telemetry_opportunity["missed_trigger_count"] == 1, telemetry_opportunity + assert telemetry_opportunity["actionable"] is False, telemetry_opportunity + assert "metadata-only telemetry" in telemetry_opportunity["privacy_contract"], telemetry_opportunity + assert "raw prompt" in telemetry_opportunity["privacy_contract"], telemetry_opportunity assert payload["scope_policy"]["present"], payload["scope_policy"] print(json.dumps({"ok": True}, ensure_ascii=False, indent=2))