frostbyte_neo/yao-meta-skill
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat: derive atlas no-route opportunities from telemetry

Browse Source
This commit is contained in:
yaojingang
2026-06-14 11:42:54 +08:00
parent a25be312bc
commit fac15545b0
2 changed files with 72 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/build_skill_atlas.py
+59 -3
View File
@@ -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"<li><strong>{html.escape(item.get('name', item.get('skill_a', 'issue')))}</strong> {html.escape(item.get('reason', item.get('status', ', '.join(item.get('missing', item.get('signal_types', []))))))}</li>"
for item in blockers
)
opportunity_items = "".join(
(
"<li>"
f"<strong>{html.escape(item.get('skill') or item.get('source_type', 'opportunity'))}</strong> "
f"{html.escape(item.get('note') or item.get('recommendation') or item.get('signal', 'no-route opportunity'))}"
f"<br><small>{html.escape(item.get('source', ''))}</small>"
"</li>"
)
for item in payload["no_route_opportunities"][:20]
)
return f"""<!DOCTYPE html>
<html lang="en">
<head>
@@ -540,6 +591,11 @@ def render_html(payload: dict[str, Any]) -> str:
<h2>Actionable Issues</h2>
<ul>{blocker_items or '<li>No blocking portfolio issues detected.</li>'}</ul>
</section>
<section>
<h2>No-Route Opportunities</h2>
<p>Missed-trigger telemetry and explicit failure cases become candidate routing work without storing raw prompts or outputs.</p>
<ul>{opportunity_items or '<li>No no-route opportunities detected.</li>'}</ul>
</section>
<section>
<h2>Full Portfolio Counts</h2>
<p>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.</p>
@@ -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")]
tests/verify_skill_atlas.py
+13 -2
View File
@@ -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))