sovereign_fail_safe_report.py

# sovereign_fail_safe_report.py
from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from sovereign_fail_safe_profile import generate_fail_safe_profile


FAIL_SAFE_OUTPUT_DIR = "sovereign_fail_safe_outputs"


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _ensure_dir(path: str) -> None:
    os.makedirs(path, exist_ok=True)


def _safe(v: Any, default: Any = "") -> Any:
    return v if v is not None else default


def _render_header(bundle: Dict[str, Any]) -> str:
    profile = bundle.get("fail_safe_profile", {}) if isinstance(bundle, dict) else {}

    return f"""# Sovereign Fail-Safe Report

Generated at: {_safe((profile or {}).get("generated_at"), _utc_now())}

## 1. Profile Identity
- Profile Type: {_safe(profile.get("profile_type"), "Sovereign Fail-Safe Profile")}
- Domain: {_safe(profile.get("domain"), "banking_and_financial_institutions")}
- Default Fail Posture: {_safe(profile.get("default_fail_posture"), "unknown")}
- Fail-Open Policy: {_safe(profile.get("fail_open_policy"), "unknown")}
- Ambiguity Policy: {_safe(profile.get("ambiguity_policy"), "unknown")}
- Tamper Policy: {_safe(profile.get("tamper_policy"), "unknown")}
- Timeout Policy: {_safe(profile.get("timeout_policy"), "unknown")}
- Manual Override Policy: {_safe(profile.get("manual_override_policy"), "unknown")}
"""


def _render_summary(bundle: Dict[str, Any]) -> str:
    profile = bundle.get("fail_safe_profile", {}) if isinstance(bundle, dict) else {}
    summary = profile.get("summary", {}) if isinstance(profile, dict) else {}

    return f"""## 2. Executive Summary
- Total Failure Modes: {_safe(summary.get("total_failure_modes"), 0)}
- Failure Modes Defaulting to FREEZE: {_safe(summary.get("freeze_defaults"), 0)}
- Failure Modes Defaulting to BLOCK: {_safe(summary.get("block_defaults"), 0)}

### Practical Interpretation
- Critical-path silent fail-open is treated as unacceptable.
- Ambiguous execution states are expected to degrade to FREEZE.
- Explicit integrity or tamper break conditions are expected to degrade to BLOCK.
- Human recovery remains controlled and audited, not implicit.
"""


def _render_failure_modes(bundle: Dict[str, Any]) -> str:
    profile = bundle.get("fail_safe_profile", {}) if isinstance(bundle, dict) else {}
    failure_modes = profile.get("failure_modes", []) if isinstance(profile, dict) else []

    lines: List[str] = ["## 3. Failure Modes"]

    if not failure_modes:
        lines.append("- No failure modes recorded.")
        return "\n".join(lines)

    for item in failure_modes:
        lines.append(f"### {item.get('failure_mode', 'unknown_failure_mode')}")
        lines.append(f"- Category: {_safe(item.get('category'))}")
        lines.append(f"- Default Action: {_safe(item.get('default_action'))}")
        lines.append(f"- Posture: {_safe(item.get('posture'))}")
        lines.append(f"- Rationale: {_safe(item.get('rationale'))}")
        lines.append("")

    return "\n".join(lines).rstrip() + "\n"


def _render_principles(bundle: Dict[str, Any]) -> str:
    profile = bundle.get("fail_safe_profile", {}) if isinstance(bundle, dict) else {}
    principles = profile.get("operational_principles", []) if isinstance(profile, dict) else []

    lines: List[str] = ["## 4. Operational Principles"]

    if not principles:
        lines.append("- No operational principles recorded.")
        return "\n".join(lines)

    for item in principles:
        lines.append(f"### {item.get('principle', 'unknown_principle')}")
        lines.append(f"- Status: {_safe(item.get('status'))}")
        lines.append(f"- Description: {_safe(item.get('description'))}")
        lines.append("")

    return "\n".join(lines).rstrip() + "\n"


def _render_artifact_status(bundle: Dict[str, Any]) -> str:
    seal = bundle.get("crypto_seal", {}) if isinstance(bundle, dict) else {}

    return f"""## 5. Artifact Status
- Sealed: {bool(bundle.get("sealed", False))}
- Seal Digest: {_safe(seal.get("seal_digest"), "")}
- Seal Scheme: {_safe(seal.get("scheme"), "")}
- Seal Algorithm: {_safe(seal.get("algorithm"), "")}
- Signer Hint: {_safe(seal.get("signer_hint"), "")}
"""


def _render_verdict(bundle: Dict[str, Any]) -> str:
    profile = bundle.get("fail_safe_profile", {}) if isinstance(bundle, dict) else {}
    notes = profile.get("notes", []) if isinstance(profile, dict) else []

    verdict = (
        "Sovereign currently expresses a clear bank-grade fail-safe posture: "
        "critical-path fail-open behavior is forbidden, ambiguity degrades to FREEZE, "
        "and trust/integrity break conditions degrade to BLOCK."
    )

    lines = [
        "## 6. Preliminary Bank-Facing Verdict",
        verdict,
        "",
        "## 7. Notes",
    ]

    if notes:
        for item in notes:
            lines.append(f"- {item}")
    else:
        lines.append("- No additional notes provided.")

    return "\n".join(lines)


def build_fail_safe_markdown(bundle: Dict[str, Any]) -> str:
    parts = [
        _render_header(bundle),
        _render_summary(bundle),
        _render_failure_modes(bundle),
        _render_principles(bundle),
        _render_artifact_status(bundle),
        _render_verdict(bundle),
    ]
    return "\n\n".join(parts).strip() + "\n"


def save_fail_safe_report(
    bundle: Dict[str, Any],
    output_dir: str = FAIL_SAFE_OUTPUT_DIR,
    base_name: Optional[str] = None,
) -> Dict[str, str]:
    _ensure_dir(output_dir)

    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    base = base_name or f"sovereign_fail_safe_report_{ts}"

    json_path = os.path.join(output_dir, f"{base}.json")
    md_path = os.path.join(output_dir, f"{base}.md")

    with open(json_path, "w", encoding="utf-8") as f:
        json.dump(bundle, f, ensure_ascii=False, indent=2)

    markdown = build_fail_safe_markdown(bundle)
    with open(md_path, "w", encoding="utf-8") as f:
        f.write(markdown)

    return {
        "json_report": json_path,
        "markdown_report": md_path,
    }


def generate_fail_safe_report(
    output_dir: str = FAIL_SAFE_OUTPUT_DIR,
    base_name: Optional[str] = None,
) -> Dict[str, Any]:
    bundle = generate_fail_safe_profile()

    files = save_fail_safe_report(
        bundle=bundle,
        output_dir=output_dir,
        base_name=base_name,
    )

    return {
        "ok": True,
        "generated_at": _utc_now(),
        "bundle": bundle,
        "files": files,
    }


if __name__ == "__main__":
    out = generate_fail_safe_report()
    print(json.dumps(out, indent=2, ensure_ascii=False))