121 lines
3.4 KiB
Python
121 lines
3.4 KiB
Python
|
"""Dual-run flags and checkout-mutation meter (RMGR-WP-0002 / ArchitectureBlueprint Stage B).
|
||
|
|
|
||
|
|
Environment variables (shared with State Hub adapter):
|
||
|
|
|
||
|
|
RM_WRITEBACK=1|true|yes — file+git task writeback via repo-manager
|
||
|
|
RM_RECONCILE=1|true|yes — reconcile path prefers repo-manager for pilot repos
|
||
|
|
RM_PILOT_REPOS=slug1,slug2 — if set, flags only apply to these slugs; empty = all
|
||
|
|
RM_METER_PATH=~/.repo-manager/mutation-meter.jsonl — append-only meter log
|
||
|
|
|
||
|
|
Rollback: unset flags or set to 0 → State Hub native path only.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Literal
|
||
|
|
|
||
|
|
Source = Literal["state-hub", "repo-manager"]
|
||
|
|
|
||
|
|
|
||
|
|
def _truthy(name: str) -> bool:
|
||
|
|
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
|
||
|
|
|
||
|
|
|
||
|
|
def writeback_enabled() -> bool:
|
||
|
|
return _truthy("RM_WRITEBACK")
|
||
|
|
|
||
|
|
|
||
|
|
def reconcile_enabled() -> bool:
|
||
|
|
return _truthy("RM_RECONCILE")
|
||
|
|
|
||
|
|
|
||
|
|
def pilot_slugs() -> set[str] | None:
|
||
|
|
"""None means all repos; empty set after parse of blank list means none."""
|
||
|
|
raw = os.environ.get("RM_PILOT_REPOS")
|
||
|
|
if raw is None:
|
||
|
|
return None
|
||
|
|
raw = raw.strip()
|
||
|
|
if not raw:
|
||
|
|
return set()
|
||
|
|
return {s.strip() for s in raw.split(",") if s.strip()}
|
||
|
|
|
||
|
|
|
||
|
|
def slug_allowed(slug: str | None) -> bool:
|
||
|
|
if not slug:
|
||
|
|
return False
|
||
|
|
pilots = pilot_slugs()
|
||
|
|
if pilots is None:
|
||
|
|
return True
|
||
|
|
return slug in pilots
|
||
|
|
|
||
|
|
|
||
|
|
def writeback_for_repo(slug: str | None) -> bool:
|
||
|
|
return writeback_enabled() and slug_allowed(slug)
|
||
|
|
|
||
|
|
|
||
|
|
def reconcile_for_repo(slug: str | None) -> bool:
|
||
|
|
return reconcile_enabled() and slug_allowed(slug)
|
||
|
|
|
||
|
|
|
||
|
|
def meter_path() -> Path:
|
||
|
|
raw = os.environ.get("RM_METER_PATH", "~/.repo-manager/mutation-meter.jsonl")
|
||
|
|
return Path(raw).expanduser()
|
||
|
|
|
||
|
|
|
||
|
|
def record_mutation(
|
||
|
|
*,
|
||
|
|
source: Source,
|
||
|
|
kind: str,
|
||
|
|
repo_slug: str | None,
|
||
|
|
detail: dict[str, Any] | None = None,
|
||
|
|
) -> None:
|
||
|
|
"""Append one meter line. Best-effort; never raises to callers."""
|
||
|
|
try:
|
||
|
|
path = meter_path()
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
row = {
|
||
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"source": source,
|
||
|
|
"kind": kind,
|
||
|
|
"repo_slug": repo_slug,
|
||
|
|
"detail": detail or {},
|
||
|
|
}
|
||
|
|
with path.open("a", encoding="utf-8") as fh:
|
||
|
|
fh.write(json.dumps(row, default=str) + "\n")
|
||
|
|
except OSError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def meter_summary(path: Path | None = None) -> dict[str, int]:
|
||
|
|
path = path or meter_path()
|
||
|
|
counts: dict[str, int] = {"state-hub": 0, "repo-manager": 0, "total": 0}
|
||
|
|
if not path.is_file():
|
||
|
|
return counts
|
||
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
||
|
|
line = line.strip()
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
row = json.loads(line)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
continue
|
||
|
|
src = row.get("source")
|
||
|
|
if src in counts:
|
||
|
|
counts[src] += 1
|
||
|
|
counts["total"] += 1
|
||
|
|
return counts
|
||
|
|
|
||
|
|
|
||
|
|
def flags_status() -> dict[str, Any]:
|
||
|
|
return {
|
||
|
|
"RM_WRITEBACK": writeback_enabled(),
|
||
|
|
"RM_RECONCILE": reconcile_enabled(),
|
||
|
|
"RM_PILOT_REPOS": sorted(pilot_slugs()) if pilot_slugs() is not None else None,
|
||
|
|
"RM_METER_PATH": str(meter_path()),
|
||
|
|
"meter": meter_summary(),
|
||
|
|
}
|