aios/services/aios-autonomous-daemon/daemon.py

#!/usr/bin/env python3
"""AIOS Autonomous GitOps Daemon.

Low-impact sidecar daemon for:
- reading GitOps events from a file-backed event bus;
- running safe validation/self-healing commands;
- sending owner notifications through a runtime-injected WeChat/WeCom webhook;
- creating audit records and request artifacts instead of bypassing production gates.

No credentials are stored in this file. Runtime secrets must be injected by the host
through environment variables or secret-mounted files.
"""

from __future__ import annotations

import argparse
import datetime as _dt
import hashlib
import hmac
import json
import os
import pathlib
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request
from typing import Any, Dict, Iterable, List, Optional, Tuple

ROOT = pathlib.Path(os.environ.get("AIOS_REPO_ROOT", ".")).resolve()
CONFIG_PATH = ROOT / "config" / "codex" / "self-optimizing-orchestrator.json"
PROD_POLICY_PATH = ROOT / "config" / "codex" / "autonomous-production-policy.json"
MERGE_POLICY_PATH = ROOT / "config" / "codex" / "autonomous-merge-policy.json"
EVENT_DIR = pathlib.Path(os.environ.get("AIOS_EVENT_DIR", str(ROOT / "runtime" / "gitops" / "events")))
AUDIT_DIR = pathlib.Path(os.environ.get("AIOS_AUDIT_DIR", str(ROOT / "runtime" / "gitops" / "audit")))
REQUEST_DIR = pathlib.Path(os.environ.get("AIOS_REQUEST_DIR", str(ROOT / "runtime" / "gitops" / "requests")))
REPORT_DIR = pathlib.Path(os.environ.get("AIOS_REPORT_DIR", str(ROOT / "reports" / "codex")))
PREAPPROVAL_DIR = pathlib.Path(os.environ.get("AIOS_PREAPPROVAL_DIR", str(ROOT / "approvals" / "preapproved")))

SAFE_AUTONOMOUS_EVENTS = {
    "git_push",
    "patch_created",
    "validation_failed",
    "validation_passed",
    "merge_candidate_created",
    "runtime_health_degraded",
}

OWNER_NOTIFY_EVENTS = {
    "main_version_changed",
    "baseline_changed",
    "merge_candidate_created",
    "validation_failed",
    "rollback_request_created",
    "approval_required",
    "daily_digest",
}

PRODUCTION_ACTIONS = {
    "merge_to_main",
    "create_release_tag",
    "activate_new_baseline",
    "production_runtime_change",
    "production_rollback_execution",
}

ALWAYS_BLOCKED = {
    "force_push",
    "purge_baseline",
    "delete_version_history",
    "delete_audit_records",
    "commit_real_secret",
    "disable_auth_or_gitea_protection",
}


def utc_now() -> str:
    return _dt.datetime.now(_dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def load_json(path: pathlib.Path, default: Any) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return default
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Invalid JSON: {path}: {exc}") from exc


def write_json(path: pathlib.Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    tmp.replace(path)


def event_id(event: Dict[str, Any]) -> str:
    raw = json.dumps(event, ensure_ascii=False, sort_keys=True).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()[:16]


def audit(action: str, status: str, detail: Dict[str, Any]) -> Dict[str, Any]:
    record = {
        "ts": utc_now(),
        "action": action,
        "status": status,
        "detail": redact(detail),
    }
    write_json(AUDIT_DIR / f"{record['ts'].replace(':', '-')}-{action}.json", record)
    print(json.dumps(record, ensure_ascii=False), flush=True)
    return record


def redact(value: Any) -> Any:
    if isinstance(value, list):
        return [redact(v) for v in value]
    if isinstance(value, dict):
        out: Dict[str, Any] = {}
        for key, val in value.items():
            lowered = key.lower()
            sensitive = any(word in lowered for word in ["api_key", "apikey", "password", "secret", "private_key", "credential", "webhook"])
            if sensitive and not (isinstance(val, str) and val.startswith("secret_ref://")):
                out[key] = "[REDACTED]"
            else:
                out[key] = redact(val)
        return out
    return value


def run_cmd(cmd: List[str], timeout: int = 300) -> Tuple[int, str, str]:
    result = subprocess.run(
        cmd,
        cwd=str(ROOT),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=timeout,
        check=False,
    )
    return result.returncode, result.stdout, result.stderr


def read_event_files(limit: int = 100) -> List[pathlib.Path]:
    EVENT_DIR.mkdir(parents=True, exist_ok=True)
    return sorted(EVENT_DIR.glob("*.json"), key=lambda p: p.stat().st_mtime)[:limit]


def archive_event(path: pathlib.Path, status: str) -> None:
    target = EVENT_DIR / status / path.name
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.move(str(path), str(target))


def get_webhook_url() -> Optional[str]:
    # Runtime-only injection. Do not commit real values.
    return os.environ.get("AIOS_OWNER_WECHAT_WEBHOOK_URL") or os.environ.get("AIOS_OWNER_WECOM_WEBHOOK_URL")


def sign_payload(payload: Dict[str, Any]) -> Optional[str]:
    secret = os.environ.get("AIOS_OWNER_ACTION_SIGNING_SECRET")
    if not secret:
        return None
    raw = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")
    return hmac.new(secret.encode("utf-8"), raw, hashlib.sha256).hexdigest()


def send_owner_wechat(event: Dict[str, Any], summary: Dict[str, Any]) -> Dict[str, Any]:
    url = get_webhook_url()
    if not url:
        return {"status": "skipped", "reason": "wechat_webhook_not_configured"}

    content = format_owner_message(event, summary)
    body = {
        "msgtype": "markdown",
        "markdown": {"content": content},
    }
    signature = sign_payload(body)
    if signature:
        body["aios_signature"] = signature

    req = urllib.request.Request(
        url,
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = resp.read().decode("utf-8", errors="replace")
            return {"status": "sent", "http_status": resp.status, "response": data[:500]}
    except urllib.error.URLError as exc:
        return {"status": "failed", "error": str(exc)}


def format_owner_message(event: Dict[str, Any], summary: Dict[str, Any]) -> str:
    title = event.get("type", "gitops_event")
    branch = summary.get("branch") or event.get("branch") or "-"
    from_version = summary.get("from_version") or event.get("from_version") or "-"
    to_version = summary.get("to_version") or event.get("to_version") or "-"
    changed = summary.get("changed_features") or event.get("changed_features") or []
    if isinstance(changed, list):
        changed_text = ", ".join(str(x) for x in changed[:8]) or "-"
    else:
        changed_text = str(changed)
    return "\n".join([
        f"### AIOS GitOps 通报：{title}",
        f"- Branch: `{branch}`",
        f"- Version: `{from_version}` → `{to_version}`",
        f"- Code Δ: +{summary.get('code_lines_added', 0)} / -{summary.get('code_lines_deleted', 0)}",
        f"- Config Δ: +{summary.get('config_lines_added', 0)} / -{summary.get('config_lines_deleted', 0)}",
        f"- Elapsed: {summary.get('elapsed_since_previous_main_change', '-')}",
        f"- Risk: {summary.get('risk', event.get('risk', 'unknown'))}",
        f"- Test: {summary.get('test_status', '-')}",
        f"- Rollback: `{summary.get('rollback_target', 'V1R1C4')}`",
        f"- Changed: {changed_text}",
        f"- Web UI: {summary.get('web_ui_audit_url', '/portal/me/#gitops-audit')}",
    ])


def diff_stats(base: str, branch: str) -> Dict[str, int]:
    code_add = code_del = config_add = config_del = 0
    rc, stdout, _ = run_cmd(["git", "diff", "--numstat", f"{base}...{branch}"], timeout=60)
    if rc != 0:
        return {"code_lines_added": 0, "code_lines_deleted": 0, "config_lines_added": 0, "config_lines_deleted": 0}
    config_ext = {".json", ".yaml", ".yml", ".toml", ".ini", ".conf", ".service"}
    for line in stdout.splitlines():
        parts = line.split("\t")
        if len(parts) < 3 or parts[0] == "-" or parts[1] == "-":
            continue
        add, delete, path = int(parts[0]), int(parts[1]), parts[2]
        if pathlib.Path(path).suffix in config_ext or path.startswith("config/"):
            config_add += add
            config_del += delete
        else:
            code_add += add
            code_del += delete
    return {
        "code_lines_added": code_add,
        "code_lines_deleted": code_del,
        "config_lines_added": config_add,
        "config_lines_deleted": config_del,
    }


def build_summary(event: Dict[str, Any]) -> Dict[str, Any]:
    branch = event.get("branch") or "patch/web-ui-complete-gitops-dashboard-20260706"
    base = event.get("base") or "main"
    stats = diff_stats(base, branch)
    return {
        "event_id": event.get("id") or event_id(event),
        "branch": branch,
        "from_rev": event.get("from_rev", base),
        "to_rev": event.get("to_rev", branch),
        "from_version": event.get("from_version", "V1R1C4"),
        "to_version": event.get("to_version", "V1R1C5"),
        "elapsed_since_previous_main_change": event.get("elapsed_since_previous_main_change", "unknown"),
        "changed_features": event.get("changed_features", ["GitOps autonomous orchestration", "Owner notifications", "event bus", "safe daemon"]),
        "test_status": event.get("test_status", "pending"),
        "risk": event.get("risk", "high"),
        "approval_status": event.get("approval_status", "required"),
        "rollback_target": event.get("rollback_target", "V1R1C4"),
        "web_ui_audit_url": event.get("web_ui_audit_url", "/portal/me/#gitops-audit"),
        **stats,
    }


def has_preapproval(action: str, event: Dict[str, Any]) -> bool:
    target_version = event.get("to_version") or "V1R1C5"
    rollback_target = event.get("rollback_target") or "V1R1C4"
    now_dt = _dt.datetime.now(_dt.timezone.utc)
    for path in PREAPPROVAL_DIR.glob("*.json"):
        try:
            doc = json.loads(path.read_text(encoding="utf-8"))
        except Exception:
            continue
        allowed = doc.get("allowed_actions", [])
        if action not in allowed:
            continue
        if doc.get("target_version") != target_version:
            continue
        if doc.get("rollback_target") != rollback_target:
            continue
        if doc.get("approved_by_role") not in ["owner", "admin"]:
            continue
        expires_at = doc.get("expires_at")
        if not expires_at:
            continue
        try:
            exp = _dt.datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
        except ValueError:
            continue
        if exp > now_dt:
            return True
    return False


def create_request(event: Dict[str, Any], reason: str) -> Dict[str, Any]:
    req = {
        "id": f"req-{utc_now().replace(':', '-')}-{event_id(event)}",
        "created_at": utc_now(),
        "event": redact(event),
        "reason": reason,
        "status": "pending_owner_review",
        "web_ui": "/portal/me/#gitops-owner-actions",
    }
    write_json(REQUEST_DIR / f"{req['id']}.json", req)
    return req


def run_validation(event: Dict[str, Any]) -> Dict[str, Any]:
    branch = event.get("branch") or "patch/web-ui-complete-gitops-dashboard-20260706"
    base = event.get("base") or "main"
    script = ROOT / "scripts" / "validate-gitops-platform-ui.sh"
    if not script.exists():
        return {"status": "blocked", "reason": "validation_script_missing"}
    rc, out, err = run_cmd(["bash", str(script), branch, base], timeout=600)
    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    (REPORT_DIR / "validation-daemon.log").write_text(out + "\n" + err, encoding="utf-8")
    return {"status": "passed" if rc == 0 else "failed", "exit_code": rc, "log": "reports/codex/validation-daemon.log"}


def run_merge_gate(event: Dict[str, Any]) -> Dict[str, Any]:
    branch = event.get("branch") or "patch/web-ui-complete-gitops-dashboard-20260706"
    base = event.get("base") or "main"
    script = ROOT / "scripts" / "autonomous-merge-gate.sh"
    if not script.exists():
        return {"status": "blocked", "reason": "merge_gate_script_missing"}
    rc, out, err = run_cmd(["bash", str(script), branch, base], timeout=600)
    (REPORT_DIR / "merge-gate-daemon.log").write_text(out + "\n" + err, encoding="utf-8")
    return {"status": "passed" if rc == 0 else "failed", "exit_code": rc, "log": "reports/codex/merge-gate-daemon.log"}


def handle_event(event: Dict[str, Any]) -> Dict[str, Any]:
    event_type = event.get("type", "unknown")
    action = event.get("action", event_type)

    if action in ALWAYS_BLOCKED:
        req = create_request(event, f"Blocked permanently by autonomous production policy: {action}")
        audit("event.blocked", "blocked", {"event": event, "request": req})
        return {"status": "blocked", "request": req}

    summary = build_summary(event)

    if event_type in OWNER_NOTIFY_EVENTS:
        notify_result = send_owner_wechat(event, summary)
        audit("owner.notify", notify_result.get("status", "unknown"), {"event": event, "summary": summary, "result": notify_result})

    if event_type in {"git_push", "patch_created", "validation_failed"}:
        validation = run_validation(event)
        audit("codex.validation", validation["status"], {"event": event, "validation": validation})
        if validation["status"] != "passed":
            req = create_request(event, "Validation failed; GPT repair loop required.")
            return {"status": "validation_failed", "request": req, "validation": validation}
        gate = run_merge_gate(event)
        audit("codex.merge_gate", gate["status"], {"event": event, "gate": gate})
        return {"status": "merge_candidate" if gate["status"] == "passed" else "gate_failed", "validation": validation, "gate": gate}

    if action in PRODUCTION_ACTIONS:
        if has_preapproval(action, event):
            # This daemon deliberately records intent. The actual protected write must be
            # performed by configured GitOps Actions with server-side authorization.
            req = create_request(event, f"Preapproval found for {action}; execute through protected GitOps Action.")
            audit("production.preapproved", "request_created", {"event": event, "request": req})
            return {"status": "preapproved_request_created", "request": req}
        req = create_request(event, f"Production action requires preapproved runbook: {action}")
        audit("production.approval_required", "pending", {"event": event, "request": req})
        return {"status": "approval_required", "request": req}

    audit("event.record", "ok", {"event": event, "summary": summary})
    return {"status": "recorded", "summary": summary}


def process_once() -> int:
    count = 0
    for path in read_event_files():
        try:
            event = json.loads(path.read_text(encoding="utf-8"))
            result = handle_event(event)
            archive_event(path, "done")
            count += 1
            audit("event.processed", "ok", {"path": str(path), "result": result})
        except Exception as exc:  # noqa: BLE001 - daemon must not crash on single event
            archive_event(path, "failed")
            audit("event.processed", "failed", {"path": str(path), "error": str(exc)})
    return count


def emit_event(args: argparse.Namespace) -> None:
    event = {
        "id": args.id or f"evt-{utc_now().replace(':', '-')}",
        "type": args.type,
        "action": args.action or args.type,
        "branch": args.branch,
        "base": args.base,
        "from_version": args.from_version,
        "to_version": args.to_version,
        "risk": args.risk,
        "created_at": utc_now(),
    }
    write_json(EVENT_DIR / f"{event['id']}.json", event)
    print(json.dumps(redact(event), ensure_ascii=False, indent=2))


def main(argv: Optional[Iterable[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="AIOS Autonomous GitOps Daemon")
    sub = parser.add_subparsers(dest="cmd", required=True)

    run = sub.add_parser("run")
    run.add_argument("--interval", type=int, default=30)
    run.add_argument("--once", action="store_true")

    emit = sub.add_parser("emit")
    emit.add_argument("--id")
    emit.add_argument("--type", required=True)
    emit.add_argument("--action")
    emit.add_argument("--branch", default="patch/web-ui-complete-gitops-dashboard-20260706")
    emit.add_argument("--base", default="main")
    emit.add_argument("--from-version", default="V1R1C4")
    emit.add_argument("--to-version", default="V1R1C5")
    emit.add_argument("--risk", default="high")

    args = parser.parse_args(list(argv) if argv is not None else None)

    if args.cmd == "emit":
        emit_event(args)
        return 0

    if args.cmd == "run":
        EVENT_DIR.mkdir(parents=True, exist_ok=True)
        AUDIT_DIR.mkdir(parents=True, exist_ok=True)
        REQUEST_DIR.mkdir(parents=True, exist_ok=True)
        if args.once:
            return 0 if process_once() >= 0 else 1
        audit("daemon.start", "ok", {"event_dir": str(EVENT_DIR)})
        while True:
            process_once()
            time.sleep(args.interval)

    return 0


if __name__ == "__main__":
    sys.exit(main())