mirror of
https://github.com/blackboxprogramming/BlackRoad-Operating-System.git
synced 2026-03-17 07:57:19 -05:00
9644737ba7
## Domain Architecture - Complete domain-to-service mapping for 16 verified domains - Subdomain architecture for blackroad.systems and blackroad.io - GitHub organization mapping (BlackRoad-OS repos) - Railway service-to-domain configuration - DNS configuration templates for Cloudflare ## Extracted Services ### AIops Service (services/aiops/) - Canary analysis for deployment validation - Config drift detection - Event correlation engine - Auto-remediation with runbook mapping - SLO budget management ### Analytics Service (services/analytics/) - Rule-based anomaly detection with safe expression evaluation - Cohort analysis with multi-metric aggregation - Decision engine with credit budget constraints - Narrative report generation ### Codex Governance (services/codex/) - 82+ governance principles (entries) - Codex Pantheon with 48+ agent archetypes - Manifesto defining ethical framework ## Integration Points - AIops → infra.blackroad.systems (blackroad-os-infra) - Analytics → core.blackroad.systems (blackroad-os-core) - Codex → operator.blackroad.systems (blackroad-os-operator) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
import yaml
|
|
|
|
from .utils import increment, log_event, validate
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ART = ROOT / "artifacts" / "decisions"
|
|
|
|
HEURISTICS = {
|
|
"uptime": {
|
|
"action": "Roll out SRE mitigation",
|
|
"bot": "sre_bot",
|
|
"owner": "bob",
|
|
"credits": 8,
|
|
"impact": 4,
|
|
"goals": ["improve_uptime"],
|
|
},
|
|
"revenue": {
|
|
"action": "Adjust Pricing",
|
|
"bot": "pricing_bot",
|
|
"owner": "alice",
|
|
"credits": 10,
|
|
"impact": 5,
|
|
"goals": ["increase_gm_pct", "reduce_churn"],
|
|
},
|
|
}
|
|
|
|
|
|
def choose_actions(goals: List[str], constraints: Dict[str, int], candidates: List[Dict]) -> Dict:
|
|
budget = constraints["credit_budget"]
|
|
k = constraints["max_concurrent"]
|
|
chosen: List[Dict] = []
|
|
spent = 0
|
|
goal_set = {goal.lower() for goal in goals}
|
|
|
|
def _matches_goal(candidate: Dict) -> bool:
|
|
if not goal_set:
|
|
return True
|
|
candidate_goals = {g.lower() for g in candidate.get("goals", [])}
|
|
return bool(candidate_goals & goal_set) or not candidate_goals
|
|
|
|
for cand in sorted(candidates, key=lambda c: (-c["impact"], c["action"])):
|
|
if not _matches_goal(cand):
|
|
continue
|
|
if spent + cand["credits"] <= budget and len(chosen) < k:
|
|
chosen.append(cand)
|
|
spent += cand["credits"]
|
|
raci = {c["bot"]: c["owner"] for c in chosen}
|
|
return {"actions": chosen, "raci": raci}
|
|
|
|
|
|
def plan_actions(anomalies_path: Path, goals_path: Path, constraints_path: Path) -> Path:
|
|
anomalies = json.loads(anomalies_path.read_text())
|
|
goals = yaml.safe_load(goals_path.read_text()).get("goals", [])
|
|
constraints = yaml.safe_load(constraints_path.read_text())
|
|
candidates: List[Dict] = []
|
|
for anom in anomalies:
|
|
h = HEURISTICS.get(anom["metric"])
|
|
if h:
|
|
candidates.append(h.copy())
|
|
plan = choose_actions(goals, constraints, candidates)
|
|
validate(plan, "plan.schema.json")
|
|
ts = datetime.utcnow().strftime("%Y%m%d%H%M%S")
|
|
ART.mkdir(parents=True, exist_ok=True)
|
|
out_path = ART / f"plan_{ts}.json"
|
|
out_path.write_text(json.dumps(plan, indent=2))
|
|
increment("decision_plan")
|
|
log_event({"type": "decision_plan", "anomalies": str(anomalies_path)})
|
|
return out_path
|