feat(agents): schedule mainstream agent version governance
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled

This commit is contained in:
ogt
2026-07-10 10:51:18 +08:00
parent fa68f35e5b
commit 210d7db338
16 changed files with 1519 additions and 239 deletions

View File

@@ -1,9 +1,12 @@
#!/usr/bin/env python3
"""
Classify market discovery repositories using primary GitHub metadata.
Classify market discovery repositories using supplied primary-source metadata.
The command is read-only. It does not add watch-registry entries, install SDKs,
call LLMs, approve paid provider use, enter replay, or change production.
Under the GitHub account-safety freeze it does not fetch GitHub metadata; callers
must provide sanitized metadata explicitly, or drafts are marked policy-blocked
for offline review.
"""
from __future__ import annotations
@@ -14,7 +17,6 @@ import json
import sys
from pathlib import Path
from typing import Any
from urllib.request import Request, urlopen
ROOT = Path(__file__).resolve().parents[2]
@@ -33,7 +35,7 @@ def main() -> int:
metadata = (
_read_json(Path(args.metadata))
if args.metadata
else _fetch_repository_metadata(discovery_review, args.timeout_seconds)
else _policy_blocked_repository_metadata(discovery_review)
)
service = _load_service()
report = service.run_agent_market_discovery_classification(
@@ -49,9 +51,8 @@ def main() -> int:
return 0
def _fetch_repository_metadata(
def _policy_blocked_repository_metadata(
discovery_review: dict[str, Any],
timeout_seconds: int,
) -> dict[str, dict[str, Any]]:
metadata: dict[str, dict[str, Any]] = {}
for draft in discovery_review.get("candidate_drafts") or []:
@@ -60,43 +61,17 @@ def _fetch_repository_metadata(
repo = str(draft.get("repository_full_name", ""))
if not repo:
continue
try:
metadata[repo] = _fetch_one_repository(repo, timeout_seconds)
except Exception as exc: # noqa: BLE001
metadata[repo] = {
"full_name": repo,
"html_url": draft.get("html_url"),
"description": None,
"topics": [],
"stargazers_count": draft.get("stargazers_count_max"),
"error": str(exc),
}
metadata[repo] = {
"full_name": repo,
"html_url": draft.get("html_url"),
"description": None,
"topics": [],
"stargazers_count": draft.get("stargazers_count_max"),
"policy_blocker": "github_global_freeze_metadata_must_be_supplied_offline",
}
return metadata
def _fetch_one_repository(repo: str, timeout_seconds: int) -> dict[str, Any]:
request = Request(
f"https://api.github.com/repos/{repo}",
headers={
"User-Agent": "awoooi-agent-market-discovery-classifier/1.0",
"Accept": "application/vnd.github+json",
},
)
with urlopen(request, timeout=timeout_seconds) as response: # noqa: S310
payload = json.loads(response.read().decode("utf-8"))
return {
"full_name": str(payload.get("full_name") or repo),
"html_url": payload.get("html_url"),
"description": payload.get("description"),
"homepage": payload.get("homepage"),
"topics": list(payload.get("topics") or []),
"language": payload.get("language"),
"stargazers_count": payload.get("stargazers_count"),
"pushed_at": payload.get("pushed_at"),
"archived": bool(payload.get("archived", False)),
}
def _read_json(path: Path) -> dict[str, Any]:
with path.open(encoding="utf-8") as handle:
payload = json.load(handle)