#!/usr/bin/env python3
import datetime
import hashlib
import json
import os
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
RUN_ID = "run-002"
REVISION = "v4"
FREEZE_KEYS = [
    "candidate",
    "outline",
    "story_brief",
    "characters",
    "ledger",
    "identity_registry",
    "reveal_ledger",
    "candidate_validation",
    "originality_receipt",
]
AUTHORITY_KEYS = [
    "story_brief",
    "characters",
    "outline",
    "ledger",
    "identity_registry",
    "reveal_ledger",
]


def sha256(path):
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(8 * 1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def audit_set_digest(rows):
    payload = json.dumps(
        {row["path"]: row["sha256"] for row in rows},
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def atomic_bytes(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    part = path.with_suffix(path.suffix + ".part")
    with open(part, "xb") as handle:
        handle.write(data)
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(part, path)


def atomic_json(path, value):
    atomic_bytes(path, (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8"))


def fail(message):
    raise RuntimeError(message)


def main():
    manifest_path = ROOT / "script/project-manifest.json"
    story_manifest_path = ROOT / "script/story-manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    story_manifest = json.loads(story_manifest_path.read_text(encoding="utf-8"))
    if story_manifest.get("run_id") != RUN_ID or story_manifest.get("active_version") != REVISION:
        fail("story manifest run/revision is not promotion target")
    if manifest.get("active_run") != RUN_ID or manifest.get("active_version") != REVISION:
        fail("project run/revision is not promotion target")
    if manifest.get("status") != "promotion_ready":
        fail("project is not promotion-ready")
    active = manifest["active_paths"]

    candidate_path = ROOT / active["candidate"]
    validation_path = ROOT / active["candidate_validation"]
    originality_path = ROOT / active["originality_receipt"]
    lock_path = ROOT / active["audit_set_lock"]
    semantic_path = ROOT / active["semantic_review"]
    producer_path = ROOT / active["producer_close"]
    canon_path = ROOT / active["story_canon"]
    source_path = ROOT / active["spoken_narration_source"]
    tts_assembly_path = ROOT / active["tts_semantic_assembly_receipt"]
    promotion_path = ROOT / active["promotion_receipt"]

    for output in (canon_path, source_path, tts_assembly_path, promotion_path):
        if output.exists() or output.with_suffix(output.suffix + ".part").exists():
            fail(f"promotion output must be virgin: {output}")
    for prerequisite in (candidate_path, validation_path, originality_path, lock_path, semantic_path, producer_path):
        if not prerequisite.exists():
            fail(f"promotion prerequisite missing: {prerequisite}")

    lock_bytes_hash = sha256(lock_path)
    lock = json.loads(lock_path.read_text(encoding="utf-8"))
    if lock.get("status") != "FROZEN" or lock.get("verified") is not True:
        fail("audit-set lock is not terminal FROZEN")
    if lock.get("run_id") != RUN_ID or lock.get("revision") != REVISION:
        fail("audit-set lock run/revision drift")
    locked_rows = lock.get("locked_files", [])
    if lock.get("locked_file_count") != 9 or len(locked_rows) != 9:
        fail("audit-set lock must contain exactly nine files")
    if [row.get("key") for row in locked_rows] != FREEZE_KEYS:
        fail("audit-set key order/inventory drift")
    expected_paths = {key: active[key] for key in FREEZE_KEYS}
    if {row.get("key"): row.get("path") for row in locked_rows} != expected_paths:
        fail("audit-set paths differ from active routing")

    current_rows = []
    for row in locked_rows:
        path = ROOT / row["path"]
        if not path.exists():
            fail(f"frozen file missing: {row['path']}")
        current_rows.append({
            "key": row["key"],
            "path": row["path"],
            "bytes": path.stat().st_size,
            "mtime_ns": path.stat().st_mtime_ns,
            "sha256": sha256(path),
        })
    if current_rows != locked_rows:
        fail("frozen file fingerprint drift")
    current_audit_set = audit_set_digest(current_rows)
    if current_audit_set != lock.get("audit_set_sha256"):
        fail("audit_set_sha256 does not reproduce")

    candidate_sha = sha256(candidate_path)
    if lock.get("candidate_sha256") != candidate_sha:
        fail("lock candidate hash drift")
    validation = json.loads(validation_path.read_text(encoding="utf-8"))
    originality = json.loads(originality_path.read_text(encoding="utf-8"))
    semantic = json.loads(semantic_path.read_text(encoding="utf-8"))
    producer = json.loads(producer_path.read_text(encoding="utf-8"))
    for name, receipt in (("validation", validation), ("originality", originality), ("producer-close", producer)):
        if receipt.get("status") != "PASS" or receipt.get("verified") is not True:
            fail(f"{name} not terminal PASS")
        if receipt.get("candidate_sha256") != candidate_sha:
            fail(f"{name} candidate hash drift")
    if validation.get("revision") != REVISION or producer.get("revision") != REVISION:
        fail("validation/producer candidate revision drift")
    if originality.get("candidate_path") != active["candidate"]:
        fail("originality candidate path drift")
    if producer.get("manual_full_read_verified") is not True:
        fail("producer-close lacks manual full read")
    if originality.get("corpus_project_count") != 17 or originality.get("semantic_architecture_verdict") != "PASS":
        fail("originality gate not terminal across 17 projects")

    if semantic.get("status") != "PASS" or semantic.get("verified") is not True:
        fail("semantic aggregate is not terminal PASS")
    if semantic.get("run_id") != RUN_ID or semantic.get("revision") != REVISION:
        fail("semantic aggregate run/revision drift")
    if semantic.get("candidate_sha256") != candidate_sha:
        fail("semantic aggregate candidate drift")
    if semantic.get("audit_set_sha256") != current_audit_set:
        fail("semantic aggregate audit-set drift")
    if semantic.get("audit_lock_receipt_sha256") != lock_bytes_hash:
        fail("semantic aggregate lock-receipt drift")
    if semantic.get("stale") is not False or semantic.get("provenance_verdict") != "PASS":
        fail("semantic aggregate currentness/provenance failed")
    if semantic.get("semantic_verdict") != "PASS" or semantic.get("blocker_count") != 0 or semantic.get("high_count") != 0:
        fail("semantic aggregate did not return 0 BLOCKER / 0 HIGH")
    if semantic.get("independent_auditor_count") != 2 or semantic.get("auditor_disagreement") is not False:
        fail("two-auditor terminal agreement missing")
    reports = semantic.get("auditor_reports", [])
    if len(reports) != 2 or any(row.get("terminal") is not True for row in reports):
        fail("two terminal auditor source reports required")
    if len({row.get("report_sha256") for row in reports}) != 2:
        fail("auditor reports are not independently identified")
    expected_hashes = {row["path"]: row["sha256"] for row in locked_rows}
    if semantic.get("observed_hashes") != expected_hashes:
        fail("semantic observed hashes do not equal exact frozen set")

    word_count = len(candidate_path.read_text(encoding="utf-8").split())
    if validation.get("candidate_words") != word_count:
        fail("candidate word count is not reproducible")
    target = validation.get("target_words", [])
    if len(target) != 2 or not target[0] <= word_count <= target[1] or not 9240 <= word_count <= 13860:
        fail("candidate outside locked/40-60-minute word range")

    candidate_bytes = candidate_path.read_bytes()
    atomic_bytes(canon_path, candidate_bytes)
    part_rows = validation.get("assembly", {}).get("parts", [])
    if len(part_rows) != 4:
        fail("TTS semantic assembly requires exact four source parts")
    part_bytes = []
    for row in part_rows:
        path = ROOT / row["path"]
        raw = path.read_bytes()
        if hashlib.sha256(raw).hexdigest() != row["sha256"]:
            fail(f"TTS source part drift: {row['path']}")
        if not raw.endswith(b"\n") or raw.endswith(b"\n\n"):
            fail(f"TTS source part LF contract failed: {row['path']}")
        part_bytes.append(raw)
    # Each part already ends in one LF; three added LFs create exactly three blank lines.
    tts_source_bytes = b"\n\n\n".join(part_bytes)
    atomic_bytes(source_path, tts_source_bytes)
    if sha256(canon_path) != candidate_sha:
        fail("canon/candidate byte equality failed")
    tts_source_sha = sha256(source_path)
    now = datetime.datetime.now(datetime.timezone.utc).isoformat()
    tts_assembly = {
        "status": "PASS",
        "verified": True,
        "run_id": RUN_ID,
        "revision": REVISION,
        "semantic_source_path": active["spoken_narration_source"],
        "semantic_source_sha256": tts_source_sha,
        "source_candidate_sha256": candidate_sha,
        "representation_contract": "TTS semantic source is distinct from candidate aggregate and has its own separator receipt",
        "separator_contract": "exactly three blank lines between source parts",
        "separator_bytes_added_between_parts_hex": "0a0a0a",
        "boundary_total_consecutive_lf": 4,
        "source_parts": part_rows,
        "source_part_hashes_verified": True,
        "semantic_content_preserved": True,
        "created_at": now,
    }
    atomic_json(tts_assembly_path, tts_assembly)
    authority = {key: sha256(ROOT / active[key]) for key in AUTHORITY_KEYS}
    promotion = {
        "status": "completed",
        "verified": True,
        "project_id": manifest["project_id"],
        "run_id": RUN_ID,
        "revision": REVISION,
        "source_candidate_path": active["candidate"],
        "source_candidate_sha256": candidate_sha,
        "source_canon_path": active["story_canon"],
        "source_canon_sha256": candidate_sha,
        "spoken_narration_source_path": active["spoken_narration_source"],
        "spoken_narration_source_sha256": tts_source_sha,
        "spoken_narration_path": active["spoken_narration"],
        "spoken_narration_sha256": None,
        "pronunciation_status": "pending",
        "word_count": word_count,
        "predicted_duration_minutes": round(word_count / 231, 4),
        "audit_set_sha256": current_audit_set,
        "audit_lock_receipt_sha256": lock_bytes_hash,
        "gates": {
            "candidate_validation": True,
            "semantic_review": True,
            "originality_17_projects": True,
            "audit_set_lock": True,
            "producer_full_read": True,
            "canon_candidate_byte_equality": True,
            "tts_semantic_source_three_blank_line_assembly": True,
        },
        "receipts": {
            "candidate_validation": {"path": active["candidate_validation"], "sha256": sha256(validation_path)},
            "semantic_review": {"path": active["semantic_review"], "sha256": sha256(semantic_path)},
            "originality": {"path": active["originality_receipt"], "sha256": sha256(originality_path)},
            "audit_set_lock": {"path": active["audit_set_lock"], "sha256": lock_bytes_hash},
            "producer_close": {"path": active["producer_close"], "sha256": sha256(producer_path)},
            "tts_semantic_assembly": {"path": active["tts_semantic_assembly_receipt"], "sha256": sha256(tts_assembly_path)},
        },
        "authority_snapshot": authority,
        "completed_at": now,
    }
    atomic_json(promotion_path, promotion)

    story_manifest.update({
        "status": "promoted",
        "verified": True,
        "run_id": RUN_ID,
        "revision": REVISION,
        "candidate_sha256": candidate_sha,
        "story_canon_path": active["story_canon"],
        "story_canon_sha256": candidate_sha,
        "spoken_narration_source_path": active["spoken_narration_source"],
        "spoken_narration_source_sha256": tts_source_sha,
        "spoken_narration_path": active["spoken_narration"],
        "spoken_narration_sha256": None,
        "word_count": word_count,
        "predicted_duration_minutes_at_231_wpm": round(word_count / 231, 4),
        "promoted_at": now,
    })
    atomic_json(story_manifest_path, story_manifest)

    manifest["status"] = "story_promoted_pronunciation_pending"
    manifest["candidate_sha256"] = candidate_sha
    manifest["candidate_bytes"] = len(candidate_bytes)
    manifest["candidate_words"] = word_count
    manifest["canon_sha256"] = candidate_sha
    manifest["spoken_narration_source_sha256"] = tts_source_sha
    manifest["spoken_narration_sha256"] = None
    manifest["steps"]["story"] = "completed"
    manifest["steps"]["pronunciation"] = "pending"
    manifest["story"] = {
        "status": "completed",
        "verified": True,
        "promotion_receipt": active["promotion_receipt"],
        "canon_sha256": candidate_sha,
        "word_count": word_count,
        "completed_at": now,
    }
    manifest["updated_at"] = now
    atomic_json(manifest_path, manifest)
    print(json.dumps({"status": "promoted", "canon_sha256": candidate_sha, "word_count": word_count}, ensure_ascii=False))


if __name__ == "__main__":
    main()