YAO
31ce04c655
Merge the beta-ready Yao Meta Skill architecture, report, evidence gate, and release-boundary updates.\n\nRelease boundary: beta/public testing is allowed; formal world-class, fully reviewed, or superiority claims remain blocked until the pending evidence gates are accepted.
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""Shared runtime helpers for the Yao CLI."""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT_INTERFACE = "internal-module"
|
|
SCRIPT_INTERFACE_REASON = "Imported by yao.py and command modules for shared subprocess execution and JSON payload parsing."
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCRIPTS = ROOT / "scripts"
|
|
|
|
|
|
def script_path(name: str) -> str:
|
|
return str(SCRIPTS / name)
|
|
|
|
|
|
def load_json_maybe(text: str) -> dict | None:
|
|
text = text.strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
|
|
def run_script(name: str, args: list[str], cwd: Path | None = None) -> dict:
|
|
proc = subprocess.run(
|
|
[sys.executable, script_path(name), *args],
|
|
cwd=cwd or ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
payload = load_json_maybe(proc.stdout)
|
|
return {
|
|
"command": f"{name} {' '.join(args)}".strip(),
|
|
"returncode": proc.returncode,
|
|
"ok": proc.returncode == 0,
|
|
"stdout": proc.stdout,
|
|
"stderr": proc.stderr,
|
|
"payload": payload,
|
|
}
|
|
|
|
|
|
def allow_report_status(result: dict, *, allowed_returncodes: set[int] | None = None) -> dict:
|
|
allowed = allowed_returncodes or {2}
|
|
if result.get("ok") or result.get("returncode") not in allowed or result.get("payload") is None:
|
|
return result
|
|
normalized = dict(result)
|
|
normalized["ok"] = True
|
|
normalized["soft_status_returncode"] = result.get("returncode")
|
|
normalized["soft_status_reason"] = "report generated with a non-passing evidence decision"
|
|
return normalized
|
|
|
|
|
|
def run_adoption_drift_if_source_exists(skill_dir: Path | None = None) -> dict:
|
|
target = (skill_dir or ROOT).resolve()
|
|
events_path = target / "reports" / "telemetry_events.jsonl"
|
|
if events_path.exists():
|
|
return run_script("render_adoption_drift_report.py", [str(target)])
|
|
return {
|
|
"command": "render_adoption_drift_report.py skipped: missing reports/telemetry_events.jsonl",
|
|
"returncode": 0,
|
|
"ok": True,
|
|
"stdout": "",
|
|
"stderr": "",
|
|
"payload": {
|
|
"ok": True,
|
|
"skipped": True,
|
|
"reason": "raw telemetry event logs are local-only; keeping committed adoption_drift_report artifacts",
|
|
},
|
|
}
|