lib/packet_builder.py

from __future__ import annotations

import csv
import json
import mimetypes
import os
import re
import shutil
import tempfile
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

from .canonical_json import build_deterministic_zip, canonical_json_bytes, write_canonical_json, write_text_lf
from .claim_ladder import bound_claims
from .dpio_reader import build_dpio_read
from .hashing import sha256_bytes, sha256_file
from .metric_analysis import METRIC_HEADERS, analyze_distribution, distribution_report_markdown, parse_metric_receipts
from .sensitive_data import scan_file, scan_text
from .temporal_ratchet import sort_events
from .validation import gate_receipt, validate_capsule, validate_causal_arc

MAX_FILE_BYTES = 100 * 1024 * 1024
MAX_TOTAL_BYTES = 500 * 1024 * 1024
ALLOWED_EXTENSIONS = {
    ".png", ".jpg", ".jpeg", ".webp", ".gif",
    ".mp4", ".mov", ".webm",
    ".csv", ".json", ".txt", ".md", ".pdf", ".zip",
}

DEFAULT_ACKNOWLEDGEMENTS = [
    "I understand this prototype preserves evidence but does not prove suppression, targeting, motive, or intent.",
    "I control whether the generated packet is shared.",
    "I have reviewed uploads for sensitive or unnecessary third-party information.",
]

CLAIMS_NOT_TO_MAKE = [
    "This packet alone proves shadowbanning or deliberate suppression.",
    "This packet alone identifies an internal classifier, employee, executive, or authorization chain.",
    "A missing notification alone proves deliberate withholding.",
    "A label removal alone proves full downstream restoration.",
    "A single account establishes motive or intent.",
]



def cleanup_old_workspaces(max_age_seconds: int = 6 * 60 * 60) -> None:
    temp_root = Path(tempfile.gettempdir())
    now = time.time()
    for candidate in temp_root.glob("substrate_creator_packet_*"):
        try:
            if candidate.is_dir() and now - candidate.stat().st_mtime > max_age_seconds:
                shutil.rmtree(candidate, ignore_errors=True)
        except OSError:
            continue

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


def parse_lines(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(x).strip() for x in value if str(x).strip()]
    text = str(value)
    return [line.strip(" \t-•") for line in text.splitlines() if line.strip(" \t-•")]


def safe_slug(value: str, fallback: str = "CREATOR") -> str:
    cleaned = re.sub(r"[^A-Za-z0-9_.-]+", "_", value.strip()).strip("._-")
    return cleaned or fallback


def capsule_id_from_seed(platform: str, handle: str, created_at: str, topology: str) -> str:
    seed = {
        "platform": platform.strip(),
        "handle": handle.strip(),
        "created_at": created_at.strip(),
        "creator_exact_description": topology,
    }
    suffix = sha256_bytes(canonical_json_bytes(seed))[:12].upper()
    prefix = safe_slug(f"{platform}_{handle}", "CREATOR").upper()
    return f"CREATOR_{prefix}_{suffix}"


def _cell(row: Any, index: int, default: Any = "") -> Any:
    if row is None:
        return default
    if isinstance(row, dict):
        return default
    if index >= len(row):
        return default
    value = row[index]
    if value is None:
        return default
    return value


def rows_from_dataframe(data: Any) -> list[list[Any]]:
    if data is None:
        return []
    if hasattr(data, "values"):
        return data.fillna("").values.tolist()
    if isinstance(data, dict) and "data" in data:
        return data["data"] or []
    if isinstance(data, list):
        return data
    return []


def parse_events(data: Any) -> list[dict[str, Any]]:
    rows = rows_from_dataframe(data)
    events: list[dict[str, Any]] = []
    for row_number, row in enumerate(rows, start=1):
        description = str(_cell(row, 4)).strip()
        if not description:
            continue
        raw_index = _cell(row, 0, row_number - 1)
        try:
            sequence_index = int(float(raw_index))
        except (TypeError, ValueError):
            sequence_index = row_number - 1
        event_id = str(_cell(row, 9)).strip() or f"EVENT_{sequence_index:04d}"
        evidence_basis = parse_lines(str(_cell(row, 6)).replace(";", "\n"))
        corrects = str(_cell(row, 7)).strip() or None
        predecessors = parse_lines(str(_cell(row, 10)).replace(";", "\n"))
        event = {
            "event_id": event_id,
            "sequence_index": sequence_index,
            "event_type": str(_cell(row, 1, "OTHER")).strip() or "OTHER",
            "event_time": str(_cell(row, 2)).strip(),
            "observed_time": str(_cell(row, 3)).strip(),
            "description": description,
            "evidence_basis": evidence_basis,
            "predecessor_event_ids": predecessors,
            "corrects_event_id": corrects,
            "state": str(_cell(row, 5, "CREATOR_REPORTED")).strip() or "CREATOR_REPORTED",
            "uncertainty": str(_cell(row, 8)).strip(),
        }
        # Empty optional strings are valid under the schema, but omit no fields
        # that are used in deterministic exports.
        events.append(event)
    return sort_events(events)


def parse_claims(data: Any) -> list[dict[str, Any]]:
    rows = rows_from_dataframe(data)
    claims: list[dict[str, Any]] = []
    for row_number, row in enumerate(rows, start=1):
        statement = str(_cell(row, 1)).strip()
        if not statement:
            continue
        claims.append({
            "claim_id": str(_cell(row, 6)).strip() or f"CLAIM_{row_number:04d}",
            "claim_level": str(_cell(row, 0, "L1_DIRECT_OBSERVATION")).strip() or "L1_DIRECT_OBSERVATION",
            "statement": statement,
            "basis": parse_lines(str(_cell(row, 2)).replace(";", "\n")),
            "state": str(_cell(row, 3, "CREATOR_REPORTED")).strip() or "CREATOR_REPORTED",
            "uncertainty": str(_cell(row, 4)).strip(),
            "human_review_required": str(_cell(row, 5)).strip().lower() in {"true", "yes", "1", "required"},
        })
    return bound_claims(claims)


def _normalize_upload_paths(uploaded_files: Any) -> list[Path]:
    if uploaded_files is None:
        return []
    items = uploaded_files if isinstance(uploaded_files, list) else [uploaded_files]
    paths: list[Path] = []
    for item in items:
        if item is None:
            continue
        if isinstance(item, str):
            path = Path(item)
        elif hasattr(item, "name"):
            path = Path(item.name)
        elif isinstance(item, dict) and item.get("path"):
            path = Path(item["path"])
        else:
            raise ValueError(f"Unsupported upload object: {type(item)!r}")
        if not path.exists() or not path.is_file():
            raise ValueError(f"Uploaded file is unavailable: {path}")
        paths.append(path)
    return paths


def _asset_metadata_map(metadata_json: str) -> dict[str, dict[str, Any]]:
    if not str(metadata_json or "").strip():
        return {}
    parsed = json.loads(metadata_json)
    if not isinstance(parsed, dict):
        raise ValueError("Per-file metadata JSON must be an object keyed by original filename.")
    return parsed


def _copy_and_manifest_assets(
    uploads: list[Path],
    evidence_dir: Path,
    created_at: str,
    event_ids: set[str],
    common: dict[str, Any],
    metadata_map: dict[str, dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[str], bool]:
    total = sum(p.stat().st_size for p in uploads)
    if total > MAX_TOTAL_BYTES:
        raise ValueError(f"Total uploads exceed {MAX_TOTAL_BYTES // (1024*1024)} MB.")

    manifests: list[dict[str, Any]] = []
    warnings: list[str] = []
    must_stop = False
    used_names: set[str] = set()
    evidence_dir.mkdir(parents=True, exist_ok=True)

    for index, source in enumerate(uploads, start=1):
        size = source.stat().st_size
        if size > MAX_FILE_BYTES:
            raise ValueError(f"{source.name} exceeds the per-file limit of {MAX_FILE_BYTES // (1024*1024)} MB.")
        extension = source.suffix.lower()
        if extension not in ALLOWED_EXTENSIONS:
            raise ValueError(f"Unsupported file type for {source.name}: {extension or '<none>'}")

        original_name = source.name
        safe_name = safe_slug(original_name, f"asset_{index:03d}{extension}")
        candidate = f"{index:03d}_{safe_name}"
        while candidate in used_names:
            candidate = f"{index:03d}_{sha256_file(source)[:8]}_{safe_name}"
        used_names.add(candidate)
        destination = evidence_dir / candidate
        shutil.copyfile(source, destination)

        if sha256_file(source) != sha256_file(destination):
            raise RuntimeError(f"Source immutability verification failed for {original_name}.")

        override = metadata_map.get(original_name, {})
        linked_ids = override.get("linked_event_ids", common.get("linked_event_ids", [])) or []
        linked_ids = [x for x in linked_ids if x in event_ids]
        public_status = override.get("public_status", common.get("public_status", "UNKNOWN"))
        redaction_state = override.get("redaction_state", common.get("redaction_state", "REDACTION_REQUIRED"))
        findings = scan_file(source)
        if findings:
            warnings.append(f"{original_name}: sensitive-data scanner flagged {', '.join(findings)}.")
            if public_status == "PUBLIC" and redaction_state not in {"REDACTED", "BLOCKED_SENSITIVE"}:
                redaction_state = "BLOCKED_SENSITIVE"
                must_stop = True
        if redaction_state == "BLOCKED_SENSITIVE":
            must_stop = True

        media_type = override.get("media_type") or mimetypes.guess_type(original_name)[0] or "application/octet-stream"
        manifests.append({
            "asset_id": f"ASSET_{index:04d}",
            "original_filename": original_name,
            "sha256": sha256_file(source),
            "byte_size": size,
            "media_type": media_type,
            "source_surface": str(override.get("source_surface", common.get("source_surface", "Creator-provided source"))),
            "source_url": str(override.get("source_url", common.get("source_url", ""))),
            "creator_description": str(override.get("creator_description", common.get("creator_description", "Creator-provided evidence asset"))),
            "date_represented": str(override.get("date_represented", common.get("date_represented", ""))),
            "uploaded_at": created_at,
            "redaction_state": redaction_state,
            "public_status": public_status,
            "linked_event_ids": linked_ids,
            "execution_blocked": True,
        })
    return manifests, warnings, must_stop


def _write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames, lineterminator="\n", extrasaction="ignore")
        writer.writeheader()
        for row in rows:
            rendered = {}
            for field in fieldnames:
                value = row.get(field, "")
                if isinstance(value, (list, dict)):
                    rendered[field] = json.dumps(value, ensure_ascii=False, sort_keys=True)
                elif value is None:
                    rendered[field] = ""
                else:
                    rendered[field] = value
            writer.writerow(rendered)


def _md_list(items: Iterable[str], empty: str = "None recorded.") -> str:
    clean = [str(item).strip() for item in items if str(item).strip()]
    return "\n".join(f"- {item}" for item in clean) if clean else f"- {empty}"


def _dpio_markdown(read: dict[str, Any]) -> str:
    sections = [
        "# DPIO Procedural Read v0.1.0",
        "",
        f"**Capsule:** `{read['creator_capsule_id']}`  ",
        f"**Procedure:** `{read['procedure']}`  ",
        f"**Claim ceiling:** `{read['claim_ceiling']}`  ",
        "**Human review:** Required",
        "",
        "## Observed facts",
        _md_list(read["observed_facts"]),
        "",
        "## Creator-reported context",
        _md_list(read["creator_reported_context"]),
        "",
        "## Supported or provisional inferences",
        _md_list(read["supported_inferences"]),
        "",
        "## Unresolved causes",
        _md_list(read["unresolved_causes"]),
        "",
        "## Controls and competing conditions",
        _md_list(read["controls_and_competing_conditions"]),
        "",
        "## Fixed execution order",
        _md_list([f"{item['sequence']}. {item['stage']} — {item['state']}" for item in read["execution_order_receipt"]]),
        "",
        "## Collapsed causal families",
        _md_list([f"{item['causal_family_id']}: {item['description']} Members: {', '.join(item['member_event_ids'])}" for item in read["causal_families"]]),
        "",
        "## Frozen discriminator predictions",
        _md_list([f"{item['prediction_id']}: {item['statement']} [{item['state']}]" for item in read["frozen_discriminator_predictions"]]),
        "",
        "## Competing hypotheses",
    ]
    for hypothesis in read["competing_hypotheses"]:
        sections.extend([
            f"### {hypothesis['hypothesis']}",
            f"**Current fit:** `{hypothesis['fit']}`",
            "",
            "Supporting observations:",
            _md_list(hypothesis["supporting_observations"]),
            "",
            "Falsifiers:",
            _md_list(hypothesis["falsifiers"]),
            "",
        ])
    sections.extend(["## Source-return requests", _md_list(read["source_return_requests"]), ""])
    return "\n".join(sections)


def _causal_arc_markdown(packet: dict[str, Any]) -> str:
    return "\n".join([
        "# Bounded Causal Arc Packet v0.1.0",
        "",
        f"**Packet:** `{packet['packet_id']}`  ",
        f"**Creator capsule:** `{packet['creator_capsule_id']}`  ",
        f"**Loop state:** `{packet['loop_state']}`  ",
        "**Human review:** Required",
        "",
        "## OBSERVED_FACTS",
        _md_list(packet["observed_facts"]),
        "",
        "## CREATOR_REPORTED_CONTEXT",
        _md_list(packet["creator_reported_context"]),
        "",
        "## SUPPORTED_INFERENCES",
        _md_list(packet["supported_inferences"]),
        "",
        "## UNRESOLVED_CAUSES",
        _md_list(packet["unresolved_causes"]),
        "",
        "## Source-return requests",
        _md_list(packet["source_return_requests"]),
        "",
        "## Claims not to make",
        _md_list(packet["claims_not_to_make"]),
        "",
    ])


def build_packet(config: dict[str, Any], schema_dir: str | Path, output_root: str | Path | None = None) -> dict[str, Any]:
    cleanup_old_workspaces()
    created_at = str(config.get("created_at") or utc_now_iso()).strip()
    events = parse_events(config.get("events"))
    claims = parse_claims(config.get("claims"))
    metric_receipts = parse_metric_receipts(config.get("metric_receipts"))
    uploads = _normalize_upload_paths(config.get("uploaded_files"))
    metadata_map = _asset_metadata_map(str(config.get("asset_metadata_json", "")))

    platform = str(config.get("platform", "")).strip() or "UNSPECIFIED_PLATFORM"
    handle = str(config.get("handle_or_pseudonym", "")).strip() or "UNSPECIFIED_CREATOR"
    exact_description = str(config.get("creator_exact_description", "")).strip()
    if not exact_description:
        raise ValueError("The creator's exact content description is required and cannot be system-inferred.")

    capsule_id = str(config.get("capsule_id", "")).strip() or capsule_id_from_seed(platform, handle, created_at, exact_description)
    consent_state = str(config.get("consent_state", "PENDING"))
    consent_scope = str(config.get("consent_scope", "PRIVATE_PACKET_ONLY"))
    event_ids = {e["event_id"] for e in events}
    if len(event_ids) != len(events):
        raise ValueError("Event IDs must be unique.")

    workspace_root = Path(output_root) if output_root else Path(tempfile.mkdtemp(prefix="substrate_creator_packet_"))
    packet_dir = workspace_root / f"CREATOR_PACKET_{safe_slug(capsule_id)}"
    if packet_dir.exists():
        shutil.rmtree(packet_dir)
    packet_dir.mkdir(parents=True)

    linked_event_ids = parse_lines(str(config.get("asset_linked_event_ids", "")).replace(";", "\n"))
    common_asset = {
        "source_surface": str(config.get("asset_source_surface", "Creator-provided source")),
        "source_url": str(config.get("asset_source_url", "")),
        "creator_description": str(config.get("asset_creator_description", "Creator-provided evidence asset")),
        "date_represented": str(config.get("asset_date_represented", "")),
        "redaction_state": str(config.get("asset_redaction_state", "REDACTION_REQUIRED")),
        "public_status": str(config.get("asset_public_status", "UNKNOWN")),
        "linked_event_ids": linked_event_ids,
    }
    assets, sensitive_warnings, must_stop = _copy_and_manifest_assets(
        uploads, packet_dir / "evidence", created_at, event_ids, common_asset, metadata_map
    )
    text_scan_fields = [
        exact_description, str(config.get("creator_reported_context", "")),
        str(config.get("controls", "")), str(config.get("source_return_requests", "")),
        str(config.get("account_url", "")), str(config.get("asset_source_url", "")),
    ]
    text_findings = sorted(set(item for text in text_scan_fields for item in scan_text(text)))
    if text_findings:
        sensitive_warnings.append("Packet text fields flagged: " + ", ".join(text_findings) + ".")
        if "possible_secret_or_token" in text_findings or consent_scope == "PUBLIC_EXHIBIT_CANDIDATE":
            must_stop = True

    requested_loop_state = str(config.get("loop_state", "ACTIVE_REVIEW"))
    closure_blockers = parse_lines(config.get("closure_blockers"))
    if must_stop:
        loop_state = "MUST_STOP"
        closure_blockers.append("Sensitive-data gate requires redaction or removal before further routing.")
    elif requested_loop_state == "CLOSED_FOR_CURRENT_SCOPE" and closure_blockers:
        loop_state = "FALSE_CLOSURE_RISK"
    else:
        loop_state = requested_loop_state

    consent = {
        "consent_state": consent_state,
        "scope": consent_scope,
        "attribution_name": str(config.get("attribution_name", "")),
        "granted_at": created_at if consent_state == "GRANTED" else None,
        "withdrawn_at": created_at if consent_state == "WITHDRAWN" else None,
        "creator_acknowledgements": DEFAULT_ACKNOWLEDGEMENTS,
        "publication_requires_additional_review": True,
    }

    source_returns = parse_lines(config.get("source_return_requests"))
    capsule = {
        "capsule_id": capsule_id,
        "schema_version": "v0.1.0",
        "created_at": created_at,
        "creator": {
            "display_name": str(config.get("display_name", "")).strip(),
            "platform": platform,
            "handle_or_pseudonym": handle,
            "account_url": str(config.get("account_url", "")).strip(),
            "account_created": str(config.get("account_created", "")).strip(),
            "followers": int(config["followers"]) if config.get("followers") not in (None, "") else None,
            "followers_observed_at": str(config.get("followers_observed_at", "")).strip(),
            "attribution_preference": str(config.get("attribution_preference", "PSEUDONYMOUS")),
        },
        "consent": consent,
        "content_topology": {
            "creator_exact_description": exact_description,
            "recurring_subjects": parse_lines(config.get("recurring_subjects")),
            "criticized_institutions_or_conduct": parse_lines(config.get("criticized_institutions_or_conduct")),
            "modes": parse_lines(config.get("modes")),
            "formats": parse_lines(config.get("formats")),
            "posting_cadence": str(config.get("posting_cadence", "")).strip(),
            "subject_change_near_event": str(config.get("subject_change_near_event", "")).strip(),
            "system_inferred_labels": [],
        },
        "account_baseline": {
            "baseline_start": str(config.get("baseline_start", "")).strip(),
            "baseline_end": str(config.get("baseline_end", "")).strip(),
            "typical_impressions": float(config["typical_impressions"]) if config.get("typical_impressions") not in (None, "") else None,
            "typical_engagement_rate": float(config["typical_engagement_rate"]) if config.get("typical_engagement_rate") not in (None, "") else None,
            "prior_high_reach_examples": parse_lines(config.get("prior_high_reach_examples")),
            "follower_delivery_baseline": str(config.get("follower_delivery_baseline", "")).strip(),
            "notification_baseline": str(config.get("notification_baseline", "")).strip(),
            "verification_state": str(config.get("verification_state", "")).strip(),
            "subscription_state": str(config.get("subscription_state", "")).strip(),
        },
        "temporal_events": events,
        "evidence_assets": assets,
        "claims": claims,
        "loop_state": loop_state,
        "review_pack": {
            "closure_blockers": sorted(set(closure_blockers)),
            "claims_not_to_make": CLAIMS_NOT_TO_MAKE,
            "evidence_to_reduce_uncertainty": parse_lines(config.get("evidence_to_reduce_uncertainty")),
            "repair_deltas": parse_lines(config.get("repair_deltas")),
            "next_reviewer_targets": parse_lines(config.get("next_reviewer_targets")),
        },
        "parent_capsule_ids": parse_lines(config.get("parent_capsule_ids")),
        "correction_history": [],
        "source_return_request": source_returns,
    }

    distribution_report = analyze_distribution(
        metric_receipts,
        float(config["typical_impressions"]) if config.get("typical_impressions") not in (None, "") else None,
    )
    truth_surface_ledger = {
        "version": "v0.2.0",
        "source_account_truth": metric_receipts,
        "creator_reported_truth": parse_lines(config.get("creator_reported_context")),
        "system_inference_boundary": [
            "Unknown mechanism cannot erase a documented platform-displayed state.",
            "A displayed metric receipt does not independently establish unique-human count, internal counting method, mechanism, authorization, or intent.",
            "Truth is downstream from honest preservation of the source account and causal order.",
        ],
    }
    holographic_precondition = {
        "version": "v0.2.0",
        "capsule_id": capsule_id,
        "individual_trace_review_required": True,
        "source_bound_receipts_present": bool(assets or metric_receipts),
        "temporal_order_present": bool(events),
        "uncertainty_preserved": True,
        "comparison_consent_scope": consent_scope,
        "comparison_ready_now": False,
        "human_review_required": True,
        "boundary": "No trace may be stacked into a holographic comparison before individual review, consent eligibility, checksum continuity, and false-closure review.",
    }

    controls = parse_lines(config.get("controls"))
    creator_context = parse_lines(config.get("creator_reported_context"))
    dpio_read = build_dpio_read(capsule, controls, creator_context)
    causal_arc = {
        "packet_id": f"CAUSAL_ARC_{capsule_id}",
        "creator_capsule_id": capsule_id,
        "observed_facts": dpio_read["observed_facts"],
        "creator_reported_context": dpio_read["creator_reported_context"],
        "supported_inferences": dpio_read["supported_inferences"],
        "unresolved_causes": dpio_read["unresolved_causes"],
        "competing_hypotheses": dpio_read["competing_hypotheses"],
        "source_return_requests": dpio_read["source_return_requests"],
        "claims_not_to_make": CLAIMS_NOT_TO_MAKE,
        "loop_state": loop_state,
        "human_review_required": True,
    }

    capsule_errors = validate_capsule(capsule, schema_dir)
    causal_errors = validate_causal_arc(causal_arc, schema_dir)
    all_errors = capsule_errors + causal_errors
    if all_errors:
        shutil.rmtree(packet_dir, ignore_errors=True)
        raise ValueError("Packet validation failed:\n- " + "\n- ".join(all_errors))

    gates = gate_receipt(consent_state, consent_scope)
    run_manifest = {
        "prototype": "SUBSTRATE_CREATOR_DISTRIBUTION_DPIO_HF_SPACE_PROTOTYPE",
        "version": "v0.2.0",
        "capsule_id": capsule_id,
        "generated_at": created_at,
        "runtime_posture": {
            "session_local": True,
            "external_calls": False,
            "database_writes": False,
            "automatic_publication": False,
            "uploaded_execution": False,
            "human_review_required": True,
        },
        "gate_receipt": gates,
        "sensitive_data_warnings": sensitive_warnings,
        "schema_validation": "PASS",
        "semantic_validation": "PASS",
        "loop_state": loop_state,
        "known_limitations": [
            "The prototype does not scrape platforms or verify creator-entered metrics against platform APIs.",
            "The prototype preserves platform-displayed receipts but does not infer unique-human counts.",
            "The DPIO read is deterministic and rule-bound; it is not an adjudication or legal conclusion.",
            "Cross-account comparison is a separate v0.2.0 route requiring at least two eligible packets and explicit human review approval.",
            "Image and video uploads are not OCR-scanned for sensitive data; creator redaction remains required.",
        ],
    }

    write_text_lf(packet_dir / "00_READ_ME_FIRST.md", "\n".join([
        "# Creator Evidence Packet",
        "",
        f"Capsule: `{capsule_id}`",
        "",
        "This packet preserves creator-provided sources, temporal order, declared uncertainty, and consent boundaries.",
        "It does not independently prove suppression, targeting, theft, motive, intent, or executive direction.",
        "Human review is required before comparison, publication, or causal attribution.",
        "",
        f"Current loop state: `{loop_state}`",
    ]))
    write_canonical_json(packet_dir / "creator_trace_capsule.json", capsule)
    write_text_lf(packet_dir / "creator_self_description.md", "# Creator Self-Description\n\n" + exact_description)
    write_canonical_json(packet_dir / "evidence_asset_manifest.json", assets)
    _write_csv(packet_dir / "evidence_asset_manifest.csv", assets, [
        "asset_id", "original_filename", "sha256", "byte_size", "media_type", "source_surface",
        "source_url", "creator_description", "date_represented", "uploaded_at", "redaction_state",
        "public_status", "linked_event_ids", "execution_blocked",
    ])
    write_canonical_json(packet_dir / "metric_receipts.json", metric_receipts)
    _write_csv(packet_dir / "metric_receipts.csv", metric_receipts, METRIC_HEADERS + ["truth_surface", "epistemic_class", "established_fact_boundary"])
    write_canonical_json(packet_dir / "distribution_transition_report.json", distribution_report)
    write_text_lf(packet_dir / "distribution_transition_report.md", distribution_report_markdown(distribution_report))
    write_canonical_json(packet_dir / "truth_surface_ledger.json", truth_surface_ledger)
    write_canonical_json(packet_dir / "holographic_precondition_receipt.json", holographic_precondition)
    write_canonical_json(packet_dir / "source_lineage_receipt.json", {
        "prototype_version": "v0.2.0",
        "governing_lineage_registry": "lineage/DONOR_REGISTRY.json",
        "authority_law": "Authority(State) <= Support(Lineage)",
        "receipt_law": "Receipt presence is not semantic validation; source bytes, causal order, and claim boundaries remain separately gated.",
    })
    write_canonical_json(packet_dir / "temporal_chain.json", events)
    _write_csv(packet_dir / "temporal_chain.csv", events, [
        "event_id", "sequence_index", "event_type", "event_time", "observed_time", "discovered_time",
        "recorded_time", "description", "evidence_basis", "predecessor_event_ids", "corrects_event_id",
        "state", "uncertainty",
    ])
    write_canonical_json(packet_dir / "causal_arc_packet.json", causal_arc)
    write_text_lf(packet_dir / "causal_arc_packet.md", _causal_arc_markdown(causal_arc))
    write_canonical_json(packet_dir / "dpio_procedural_read.json", dpio_read)
    write_text_lf(packet_dir / "dpio_procedural_read.md", _dpio_markdown(dpio_read))
    write_canonical_json(packet_dir / "dpio_execution_order_receipt.json", dpio_read["execution_order_receipt"])
    write_canonical_json(packet_dir / "causal_families.json", dpio_read["causal_families"])
    write_canonical_json(packet_dir / "frozen_discriminator_predictions.json", dpio_read["frozen_discriminator_predictions"])
    write_canonical_json(packet_dir / "pressure_test_results.json", dpio_read["pressure_test_results"])
    write_canonical_json(packet_dir / "minimum_cut_candidates.json", dpio_read["minimum_cut_candidates"])
    write_text_lf(packet_dir / "source_return_requests.md", "# Source-Return Requests\n\n" + _md_list(source_returns))
    write_text_lf(packet_dir / "claim_boundary.md", "# Claim Boundary\n\n" + _md_list(CLAIMS_NOT_TO_MAKE) + "\n\nL5 attributed cause and L6 intent/motive remain blocked in this prototype.")
    write_text_lf(packet_dir / "human_review_checklist.md", "\n".join([
        "# Human Review Checklist",
        "",
        "- [ ] Verify every observed fact against its cited source.",
        "- [ ] Confirm the creator's exact description was preserved without ideological relabeling.",
        "- [ ] Check temporal order, correction ancestry, and missing timestamps.",
        "- [ ] Test competing hypotheses and their falsifiers.",
        "- [ ] Review redaction and third-party privacy.",
        "- [ ] Confirm consent scope before any comparison.",
        "- [ ] Obtain separate approval before any public exhibit.",
        "- [ ] Keep L5/L6 blocked absent supporting sources and explicit human authority.",
    ]))
    write_canonical_json(packet_dir / "consent_receipt.json", consent)
    write_canonical_json(packet_dir / "run_manifest.json", run_manifest)
    write_text_lf(packet_dir / "validation_receipt.txt", "\n".join([
        "SCHEMA_VALIDATION: PASS",
        "SEMANTIC_VALIDATION: PASS",
        "SOURCE_IMMUTABILITY: PASS",
        "TEMPORAL_RATCHET: PASS",
        "METRIC_RECEIPT_BOUNDARY: PASS",
        "HOLOGRAPHIC_PRECONDITION: HELD_FOR_HUMAN_REVIEW",
        "CONSENT_GATE: PASS",
        "CLAIM_LADDER: PASS",
        "FALSE_CLOSURE_GATE: PASS",
        f"LOOP_STATE: {loop_state}",
        "OVERALL: PASS",
    ]))

    # Hash every packet member except the checksum ledger itself.
    checksums: list[str] = []
    for member in sorted(p for p in packet_dir.rglob("*") if p.is_file() and p.name != "SHA256SUMS.txt"):
        checksums.append(f"{sha256_file(member)}  {member.relative_to(packet_dir).as_posix()}")
    write_text_lf(packet_dir / "SHA256SUMS.txt", "\n".join(checksums))

    zip_path = workspace_root / f"CREATOR_PACKET_{safe_slug(capsule_id)}.zip"
    build_deterministic_zip(packet_dir, zip_path)
    zip_sha = sha256_file(zip_path)
    sidecar_path = workspace_root / f"{zip_path.name}.sha256"
    write_text_lf(sidecar_path, f"{zip_sha}  {zip_path.name}")

    return {
        "capsule": capsule,
        "causal_arc": causal_arc,
        "dpio_read": dpio_read,
        "run_manifest": run_manifest,
        "packet_dir": str(packet_dir),
        "zip_path": str(zip_path),
        "sidecar_path": str(sidecar_path),
        "zip_sha256": zip_sha,
    }