#!/usr/bin/env python3
"""驗證 IwoooS / S4.9 owner gate 維持只讀收件邊界。

本 guard 只讀取 repo 內的 S4.9 / source-control owner response snapshot
與 Markdown 規範，不送 request、不收 response、不呼叫 Gitea / GitHub /
AwoooP、不修改 repo / refs / workflow / secret / runner，也不開 runtime gate。
"""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


REQUIRED_DOCS = [
    "docs/security/S4-9-OWNER-RESPONSE-GATE-CURRENT-GAP-AUDIT.md",
    "docs/security/S4-9-CANONICAL-OWNER-RESPONSE-ENVELOPE.md",
    "docs/security/S4-9-OWNER-RESPONSE-INTAKE-FORM.md",
    "docs/security/S4-9-REVIEWER-VALIDATION-CHECKLIST.md",
    "docs/security/S4-9-SECURITY-ACCEPTANCE-RECORD-TEMPLATE.md",
    "docs/security/GITEA-INVENTORY-OWNER-ATTESTATION-RESPONSE.md",
    "docs/security/SOURCE-CONTROL-OWNER-RESPONSE-VALIDATION-ROLLUP.md",
    "docs/security/GITHUB-TARGET-OWNER-DECISION-RESPONSE.md",
    "docs/security/SOURCE-CONTROL-REF-TRUTH-OWNER-RESPONSE.md",
    "docs/security/SOURCE-CONTROL-WORKFLOW-SECRET-NAME-OWNER-RESPONSE.md",
]

CANONICAL_FIELDS = [
    "owner_role_or_team",
    "decision",
    "decision_reason",
    "affected_scope",
    "redacted_evidence_refs",
    "followup_owner",
]

S4_9_TEMPLATES = [
    "response-public-only-vs-local-gitea-gap",
    "response-org-user-endpoint-identity",
    "response-internal-110-adjacent-scope",
    "response-repo-owner-canonical-scope",
    "response-legacy-or-inaccessible-disposition",
]

OWNER_PACKET_SPECS = [
    {
        "label": "s4.9 gitea owner attestation response",
        "path": "docs/security/gitea-inventory-owner-attestation-response.snapshot.json",
        "schema": "gitea_inventory_owner_attestation_response_v1",
        "status": "draft_waiting_owner_response",
        "template_count": 5,
        "expected_template_ids": S4_9_TEMPLATES,
        "summary_counts": {
            "owner_response_request_packet_count": 1,
            "response_template_count": 5,
            "owner_response_template_status_count": 5,
            "intake_preflight_check_count": 6,
            "intake_outcome_lane_count": 5,
            "acceptance_check_count": 8,
            "rejection_rule_count": 10,
            "received_response_count": 0,
            "accepted_response_count": 0,
            "rejected_response_count": 0,
            "owner_response_metadata_intake_required_count": 6,
            "owner_response_metadata_intake_received_count": 0,
            "owner_response_metadata_intake_accepted_count": 0,
            "owner_response_metadata_intake_runtime_gate_count": 0,
            "owner_response_intake_handoff_queue_count": 5,
            "owner_response_intake_handoff_queue_received_count": 0,
            "owner_response_intake_handoff_queue_accepted_count": 0,
            "owner_response_intake_handoff_queue_runtime_gate_count": 0,
        },
        "false_flags": [
            "runtime_execution_authorized",
            "action_buttons_allowed",
            "token_value_collection_allowed",
            "raw_secret_allowed",
            "repo_write_allowed",
            "refs_sync_allowed",
            "github_primary_switch_authorized",
            "owner_response_metadata_intake_raw_payload_allowed",
            "owner_response_metadata_intake_secret_plaintext_allowed",
            "owner_response_metadata_intake_action_buttons_allowed",
            "owner_response_intake_handoff_queue_raw_payload_allowed",
            "owner_response_intake_handoff_queue_action_buttons_allowed",
        ],
    },
    {
        "label": "s4.10 github target owner decision response",
        "path": "docs/security/github-target-owner-decision-response.snapshot.json",
        "schema": "github_target_owner_decision_response_v1",
        "status": "draft_waiting_owner_response",
        "template_count": 9,
        "summary_counts": {
            "owner_response_request_packet_count": 1,
            "response_template_count": 9,
            "owner_response_template_status_count": 9,
            "intake_preflight_check_count": 6,
            "acceptance_check_count": 8,
            "rejection_rule_count": 10,
            "received_response_count": 0,
            "accepted_response_count": 0,
            "rejected_response_count": 0,
        },
        "false_flags": [
            "runtime_execution_authorized",
            "action_buttons_allowed",
            "repo_creation_authorized",
            "visibility_change_authorized",
            "refs_sync_authorized",
            "github_primary_switch_authorized",
            "secret_value_collection_allowed",
            "target_owner_request_dispatch_authorized",
        ],
    },
    {
        "label": "s4.11 ref truth owner response",
        "path": "docs/security/source-control-ref-truth-owner-response.snapshot.json",
        "schema": "source_control_ref_truth_owner_response_v1",
        "status": "draft_waiting_owner_response",
        "template_count": 5,
        "summary_counts": {
            "owner_response_request_packet_count": 1,
            "response_template_count": 5,
            "owner_response_template_status_count": 5,
            "intake_preflight_check_count": 6,
            "acceptance_check_count": 8,
            "rejection_rule_count": 10,
            "received_response_count": 0,
            "accepted_response_count": 0,
            "rejected_response_count": 0,
        },
        "false_flags": [
            "runtime_execution_authorized",
            "action_buttons_allowed",
            "refs_sync_authorized",
            "refs_delete_authorized",
            "force_push_authorized",
            "github_primary_switch_authorized",
            "secret_value_collection_allowed",
        ],
    },
    {
        "label": "s4.12 workflow secret owner response",
        "path": "docs/security/source-control-workflow-secret-name-owner-response.snapshot.json",
        "schema": "source_control_workflow_secret_name_owner_response_v1",
        "status": "draft_waiting_owner_response",
        "template_count": 5,
        "summary_counts": {
            "owner_response_request_packet_count": 1,
            "response_template_count": 5,
            "owner_response_template_status_count": 5,
            "intake_preflight_check_count": 6,
            "acceptance_check_count": 8,
            "rejection_rule_count": 10,
            "received_response_count": 0,
            "accepted_response_count": 0,
            "rejected_response_count": 0,
        },
        "false_flags": [
            "runtime_execution_authorized",
            "action_buttons_allowed",
            "workflow_modification_authorized",
            "repo_secret_change_authorized",
            "runner_change_authorized",
            "webhook_modification_authorized",
            "branch_protection_change_authorized",
            "deploy_key_change_authorized",
            "github_hosted_runner_enable_authorized",
            "secret_value_collection_allowed",
            "secret_value_or_hash_collection_allowed",
            "workflow_secret_owner_request_dispatch_authorized",
            "write_token_allowed",
        ],
    },
]


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def fail(message: str) -> None:
    raise SystemExit(f"BLOCKED {message}")


def assert_equal(label: str, actual: Any, expected: Any) -> None:
    if actual != expected:
        fail(f"{label}: expected {expected!r}, got {actual!r}")


def assert_contains(label: str, values: list[Any], expected: Any) -> None:
    if expected not in values:
        fail(f"{label}: missing {expected!r}")


def assert_text_contains(label: str, text: str, expected: str) -> None:
    if expected not in text:
        fail(f"{label}: missing {expected!r}")


def assert_path_exists(root: Path, relative_path: str) -> None:
    if not (root / relative_path).exists():
        fail(f"path missing: {relative_path}")


def summary_value(data: dict[str, Any], key: str) -> Any:
    summary = data.get("summary", {})
    if key in summary:
        return summary[key]
    return data.get(key)


def assert_false_summary_flag(label: str, data: dict[str, Any], key: str) -> None:
    value = summary_value(data, key)
    assert_equal(f"{label}.{key}", value, False)


def validate_docs(root: Path) -> None:
    for relative_path in REQUIRED_DOCS:
        assert_path_exists(root, relative_path)

    canonical_text = (root / "docs/security/S4-9-CANONICAL-OWNER-RESPONSE-ENVELOPE.md").read_text(
        encoding="utf-8"
    )
    intake_text = (root / "docs/security/S4-9-OWNER-RESPONSE-INTAKE-FORM.md").read_text(
        encoding="utf-8"
    )
    gap_text = (root / "docs/security/S4-9-OWNER-RESPONSE-GATE-CURRENT-GAP-AUDIT.md").read_text(
        encoding="utf-8"
    )
    for field in CANONICAL_FIELDS:
        assert_text_contains("canonical_envelope.fields", canonical_text, field)
        assert_text_contains("intake_form.fields", intake_text, field)
    for template_id in S4_9_TEMPLATES:
        assert_text_contains("canonical_envelope.templates", canonical_text, template_id)
        assert_text_contains("intake_form.templates", intake_text, template_id)
    for marker in [
        "request_sent=false",
        "received_response_count=0",
        "accepted_response_count=0",
        "runtime_execution_authorized=false",
        "action_buttons_allowed=false",
    ]:
        assert_text_contains("canonical_envelope.zero_boundary", canonical_text, marker)
        assert_text_contains("intake_form.zero_boundary", intake_text, marker)
    assert_text_contains("gap_audit.owner_gate_zero", gap_text, "S4.9 owner response gate 仍是 `0%`")


def validate_gap_audit(root: Path) -> None:
    gap = load_json(root / "docs/security/s4-9-owner-response-gap-audit.snapshot.json")
    summary = gap["summary"]
    assert_equal("gap.schema_version", gap["schema_version"], "s4_9_owner_response_gap_audit_v1")
    assert_equal("gap.status", gap["status"], "gap_audit_ready_owner_gate_zero")
    assert_equal("gap.summary.current_requirement_gap_count", summary["current_requirement_gap_count"], 8)
    assert_equal("gap.summary.new_rule_count", summary["new_rule_count"], 7)
    assert_equal("gap.summary.rule_adjustment_count", summary["rule_adjustment_count"], 7)
    assert_equal("gap.summary.priority_work_item_count", summary["priority_work_item_count"], 9)
    for key in [
        "request_sent_count",
        "owner_response_received_count",
        "owner_response_accepted_count",
        "owner_response_rejected_count",
        "runtime_gate_count",
    ]:
        assert_equal(f"gap.summary.{key}", summary[key], 0)
    assert_equal("gap.summary.public_surface_raw_namespace_allowed", summary["public_surface_raw_namespace_allowed"], False)
    assert_equal(
        "gap.summary.work_session_transcript_public_allowed",
        summary["work_session_transcript_public_allowed"],
        False,
    )
    false_boundaries = gap["false_boundaries"]
    for key, value in false_boundaries.items():
        if value is not False:
            fail(f"gap.false_boundaries.{key}: expected false, got {value!r}")


def validate_owner_packet(root: Path, spec: dict[str, Any]) -> None:
    data = load_json(root / spec["path"])
    label = spec["label"]
    assert_equal(f"{label}.schema_version", data.get("schema_version"), spec["schema"])
    assert_equal(f"{label}.status", data.get("status"), spec["status"])
    assert_equal(f"{label}.runtime_execution_authorized", data.get("runtime_execution_authorized"), False)
    for key, expected in spec["summary_counts"].items():
        assert_equal(f"{label}.summary.{key}", summary_value(data, key), expected)
    for key in spec["false_flags"]:
        assert_false_summary_flag(label, data, key)

    templates = data.get("response_templates", [])
    if not isinstance(templates, list):
        fail(f"{label}.response_templates: expected list")
    assert_equal(f"{label}.response_templates.count", len(templates), spec["template_count"])
    if "expected_template_ids" in spec:
        template_ids = [item.get("template_id") or item.get("id") for item in templates]
        for template_id in spec["expected_template_ids"]:
            assert_contains(f"{label}.response_templates", template_ids, template_id)


def validate_rollup(root: Path) -> None:
    rollup = load_json(root / "docs/security/source-control-owner-response-validation-rollup.snapshot.json")
    summary = rollup["summary"]
    assert_equal("rollup.schema_version", rollup["schema_version"], "source_control_owner_response_validation_rollup_v1")
    assert_equal("rollup.status", rollup["status"], "draft_waiting_owner_responses")
    assert_equal("rollup.summary.response_packet_count", summary["response_packet_count"], 4)
    assert_equal("rollup.summary.total_response_template_count", summary["total_response_template_count"], 24)
    assert_equal("rollup.summary.total_acceptance_check_count", summary["total_acceptance_check_count"], 32)
    assert_equal("rollup.summary.total_rejection_rule_count", summary["total_rejection_rule_count"], 40)
    assert_equal("rollup.summary.validation_lane_count", summary["validation_lane_count"], 4)
    assert_equal("rollup.summary.owner_response_validation_reviewer_checklist_count", summary["owner_response_validation_reviewer_checklist_count"], 9)
    assert_equal("rollup.summary.owner_response_validation_reviewer_outcome_lane_count", summary["owner_response_validation_reviewer_outcome_lane_count"], 7)
    for key in [
        "total_received_response_count",
        "total_accepted_response_count",
        "total_rejected_response_count",
        "primary_ready_count",
    ]:
        assert_equal(f"rollup.summary.{key}", summary[key], 0)
    for key in [
        "runtime_execution_authorized",
        "action_buttons_allowed",
        "repo_creation_authorized",
        "visibility_change_authorized",
        "refs_sync_authorized",
        "refs_delete_authorized",
        "workflow_modification_authorized",
        "runner_enablement_authorized",
        "secret_value_collection_allowed",
        "github_primary_switch_authorized",
        "force_push_authorized",
        "write_token_allowed",
    ]:
        assert_false_summary_flag("rollup", rollup, key)


def validate(root: Path) -> None:
    validate_docs(root)
    validate_gap_audit(root)
    validate_rollup(root)
    for spec in OWNER_PACKET_SPECS:
        validate_owner_packet(root, spec)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--root",
        default=Path(__file__).resolve().parents[2],
        type=Path,
        help="Repository root. Defaults to the current script's repository.",
    )
    args = parser.parse_args()
    validate(args.root.resolve())
    print("IWOOOS_OWNER_GATE_GUARD_OK")


if __name__ == "__main__":
    main()
