lib/dpio_reader.py

from __future__ import annotations

from typing import Any


def _event_text(event: dict[str, Any]) -> str:
    time = event.get("event_time") or event.get("observed_time") or "time not supplied"
    return f"[{event.get('event_id')}] {time}: {event.get('description', '').strip()}"


def _has_event(events: list[dict[str, Any]], event_type: str) -> bool:
    return any(event.get("event_type") == event_type for event in events)



def collapse_causal_families(capsule: dict[str, Any]) -> list[dict[str, Any]]:
    events = capsule.get("temporal_events", [])
    definitions = [
        ("CF_DISTRIBUTION_STATE", {"BASELINE", "REACH_CHANGE", "ANALYTICS_SNAPSHOT", "RECOVERY", "FOLLOWER_COUNT_CHANGE"}, "Visible distribution baseline, breakpoint, persistence, and recovery state."),
        ("CF_ENFORCEMENT_RESTORATION", {"ACCOUNT_LABEL", "REVIEW_CLEARANCE", "SUPPORT_RESPONSE"}, "Platform label, review, clearance, support, and downstream restoration sequence."),
        ("CF_RELATIONAL_RETURN", {"FOLLOWER_REPORT", "NOTIFICATION_OMISSION", "NOTIFICATION_DELAY"}, "Follower delivery, notification return, and creator reciprocity surface."),
        ("CF_CONTROL_CONDITIONS", {"NEUTRAL_CONTROL"}, "Neutral or matched control conditions used to test non-platform explanations."),
        ("CF_PUBLIC_CHALLENGE", {"PUBLIC_COMPLAINT"}, "Public complaint or challenge events that may affect temporal interpretation but do not establish retaliation."),
        ("CF_CORRECTION_RATCHET", {"CREATOR_CORRECTION"}, "Append-only corrections preserving ancestor events."),
    ]
    families: list[dict[str, Any]] = []
    assigned: set[str] = set()
    for family_id, types, description in definitions:
        members = [e.get("event_id") for e in events if e.get("event_type") in types]
        if members:
            assigned.update(members)
            families.append({
                "causal_family_id": family_id,
                "description": description,
                "member_event_ids": members,
                "state": "COLLAPSED_FOR_SINGLE_CASE_REVIEW",
            })
    unassigned = [e.get("event_id") for e in events if e.get("event_id") not in assigned]
    if unassigned:
        families.append({
            "causal_family_id": "CF_OTHER_UNCOLLAPSED",
            "description": "Events not yet assigned to a governed causal family.",
            "member_event_ids": unassigned,
            "state": "HELD",
        })
    return families


def freeze_discriminator_predictions(capsule: dict[str, Any], causal_families: list[dict[str, Any]]) -> list[dict[str, Any]]:
    family_ids = {f["causal_family_id"] for f in causal_families}
    predictions = [
        {
            "prediction_id": "PRED_MATCHED_CONTROLS",
            "statement": "If the transition is platform-wide rather than account-conditioned, matched accounts and comparable surfaces should exhibit a similar time-aligned change.",
            "required_sources": ["Matched comparison capsules", "Platform-wide change logs"],
            "state": "FROZEN_HELD",
        },
        {
            "prediction_id": "PRED_ACCOUNT_STATE",
            "statement": "If an account-conditioned state contributed, account-level eligibility or ranking records should show a transition near the documented breakpoint.",
            "required_sources": ["Account recommendation-eligibility history", "Ranking/cohort state history"],
            "state": "FROZEN_HELD",
        },
        {
            "prediction_id": "PRED_RESTORATION",
            "statement": "If review clearance fully restored the prior state, downstream eligibility records or normalized distribution should evidence restoration after clearance.",
            "required_sources": ["Before/after enforcement-state diff", "Restoration execution receipt"],
            "state": "FROZEN_HELD",
        },
        {
            "prediction_id": "PRED_NOTIFICATION_ROUTE",
            "statement": "If relational return was degraded at notification routing, thread-visible interactions and notification-delivery logs should diverge in a reproducible way.",
            "required_sources": ["Notification generation and delivery logs", "Thread interaction export"],
            "state": "FROZEN_HELD",
        },
    ]
    if "CF_ENFORCEMENT_RESTORATION" not in family_ids:
        predictions[2]["state"] = "FROZEN_LOW_SIGNAL"
    if "CF_RELATIONAL_RETURN" not in family_ids:
        predictions[3]["state"] = "FROZEN_LOW_SIGNAL"
    return predictions

def build_hypotheses(capsule: dict[str, Any]) -> list[dict[str, Any]]:
    events = capsule.get("temporal_events", [])
    subject_change = str(capsule.get("content_topology", {}).get("subject_change_near_event", "")).lower()
    stable_subject = any(token in subject_change for token in ("no", "none", "stable", "unchanged"))
    has_reach_change = _has_event(events, "REACH_CHANGE")
    has_label = _has_event(events, "ACCOUNT_LABEL")
    has_clearance = _has_event(events, "REVIEW_CLEARANCE")
    has_recovery = _has_event(events, "RECOVERY")
    has_notification = _has_event(events, "NOTIFICATION_OMISSION") or _has_event(events, "NOTIFICATION_DELAY")
    has_control = _has_event(events, "NEUTRAL_CONTROL")

    ordinary_fit = "LOW" if has_reach_change and (has_control or has_notification) else "HELD"
    creator_change_fit = "LOW" if stable_subject else "HELD"
    account_state_fit = "MEDIUM" if has_reach_change and (has_label or has_notification) else "HELD"
    enforcement_fit = "MEDIUM" if has_reach_change and has_label and has_clearance and not has_recovery else "HELD"

    return [
        {
            "hypothesis": "H0 — ordinary audience variation or post-level performance variance",
            "fit": ordinary_fit,
            "supporting_observations": [
                "Single-account metrics can vary for reasons not visible in the packet."
            ],
            "falsifiers": [
                "A persistent account-level breakpoint across comparable content and matched controls.",
                "Platform records showing an account-conditioned recommendation or distribution state."
            ],
        },
        {
            "hypothesis": "H1 — creator posting cadence, format, or subject-mix change",
            "fit": creator_change_fit,
            "supporting_observations": [
                "Changes in creator behavior can alter visible distribution."
            ],
            "falsifiers": [
                "Receipts showing materially stable cadence, format, and subject topology across the breakpoint.",
                "Comparable content performing differently before and after the breakpoint."
            ],
        },
        {
            "hypothesis": "H2 — platform-wide recommender or demand change",
            "fit": "HELD",
            "supporting_observations": [
                "A platform-wide change can affect many creators simultaneously."
            ],
            "falsifiers": [
                "Matched comparison accounts not exhibiting the same transition during the same period.",
                "Platform change logs excluding the relevant surface or account cohort."
            ],
        },
        {
            "hypothesis": "H3 — account-conditioned distribution or recommendation state",
            "fit": account_state_fit,
            "supporting_observations": [
                "An abrupt persistent reach change can be generated by an account-level hidden state.",
                "Notification or label events may identify a candidate state transition."
            ],
            "falsifiers": [
                "Account-level eligibility history showing no relevant state change.",
                "A complete organic explanation reproducing the observed breakpoint and persistence."
            ],
        },
        {
            "hypothesis": "H4 — enforcement, label, or review state coupled to distribution and incomplete restoration",
            "fit": enforcement_fit,
            "supporting_observations": [
                "A label-clearance sequence without demonstrated recovery is a candidate causal family."
            ],
            "falsifiers": [
                "Records showing full downstream restoration at clearance time.",
                "Evidence that the reach change preceded and was independent of the enforcement state."
            ],
        },
        {
            "hypothesis": "H5 — recurrent extraction-with-relational-severance phenotype across creators",
            "fit": "HELD",
            "supporting_observations": [
                "The creator's work may remain platform-readable while human relational return contracts."
            ],
            "falsifiers": [
                "Cross-account comparison showing no recurrent phenotype after normalization and controls.",
                "Evidence that platform/machine access declined proportionally with human reach."
            ],
        },
    ]


def build_minimum_cut_candidates(capsule: dict[str, Any]) -> list[dict[str, Any]]:
    events = capsule.get("temporal_events", [])
    candidates = [
        {
            "candidate_id": "CUT_RECOMMENDATION_ELIGIBILITY",
            "candidate_cut": "Account-level recommendation or discovery eligibility",
            "preserved_path": "Creator content remains hosted and platform-readable.",
            "potentially_degraded_path": "Independent human discovery beyond the existing audience.",
            "required_source_return": "Account recommendation-eligibility and distribution-state history.",
            "state": "HELD",
        },
        {
            "candidate_id": "CUT_DISTRIBUTION_MULTIPLIER",
            "candidate_cut": "Account- or post-conditioned distribution multiplier",
            "preserved_path": "Content remains available for engagement, indexing, and machine retrieval.",
            "potentially_degraded_path": "The number or diversity of humans to whom the content is delivered.",
            "required_source_return": "Ranking feature values, cohort assignment, and multiplier history.",
            "state": "HELD",
        },
        {
            "candidate_id": "CUT_NOTIFICATION_RETURN",
            "candidate_cut": "Notification and reply-return routing",
            "preserved_path": "Replies or platform interactions can exist on-thread.",
            "potentially_degraded_path": "The creator's awareness of and ability to reciprocate human interaction.",
            "required_source_return": "Notification-generation, suppression, deduplication, and delivery logs.",
            "state": "HELD",
        },
        {
            "candidate_id": "CUT_RESTORATION_STATE",
            "candidate_cut": "Downstream restoration after label removal, appeal, or clearance",
            "preserved_path": "The visible label can be removed.",
            "potentially_degraded_path": "Prior recommendation and distribution state may remain unrestored.",
            "required_source_return": "Before/after enforcement state diff and restoration execution receipt.",
            "state": "HELD",
        },
    ]
    if not any(e.get("event_type") in {"ACCOUNT_LABEL", "REVIEW_CLEARANCE"} for e in events):
        candidates[-1]["state"] = "LOW_SIGNAL"
    return candidates


def build_dpio_read(capsule: dict[str, Any], controls: list[str], creator_context: list[str]) -> dict[str, Any]:
    events = capsule.get("temporal_events", [])
    assets = capsule.get("evidence_assets", [])
    claims = capsule.get("claims", [])

    observed_facts = [_event_text(e) for e in events if e.get("state") == "OBSERVED"]
    observed_facts.extend(
        f"[{a.get('asset_id')}] Source artifact preserved: {a.get('original_filename')} (SHA-256 {a.get('sha256')})."
        for a in assets
    )

    reported = list(creator_context)
    reported.extend(_event_text(e) for e in events if e.get("state") == "CREATOR_REPORTED")
    reported.insert(0, capsule.get("content_topology", {}).get("creator_exact_description", ""))
    reported = [item for item in reported if str(item).strip()]

    supported_inferences = [
        c.get("statement", "")
        for c in claims
        if c.get("claim_level") in {"L2_REPEATED_PATTERN", "L3_STRUCTURAL_INFERENCE", "L4_BEST_FIT_MECHANISM"}
        and c.get("state") in {"SUPPORTED", "PROVISIONAL", "STRAINED"}
    ]

    source_returns = capsule.get("source_return_request", []) or []
    unresolved = [
        "The exact internal platform mechanism remains unresolved without platform-controlled records.",
        "Executive knowledge, authorization, purpose, and intent are not established by this single-case packet.",
    ]
    if source_returns:
        unresolved.append("The packet identifies source-return requests that remain outstanding.")

    causal_families = collapse_causal_families(capsule)
    frozen_predictions = freeze_discriminator_predictions(capsule, causal_families)
    hypotheses = build_hypotheses(capsule)
    minimum_cuts = build_minimum_cut_candidates(capsule)
    execution_order = [
        {"sequence": 1, "stage": "SOURCE_REGISTERED", "state": "PASS"},
        {"sequence": 2, "stage": "EVENTS_REGISTERED", "state": "PASS"},
        {"sequence": 3, "stage": "CHRONOLOGY_MAPPED", "state": "PASS"},
        {"sequence": 4, "stage": "CAUSAL_FAMILIES_COLLAPSED", "state": "PASS"},
        {"sequence": 5, "stage": "PREDICTIONS_FROZEN", "state": "PASS"},
        {"sequence": 6, "stage": "HYPOTHESES_FROZEN", "state": "PASS"},
        {"sequence": 7, "stage": "PRESSURE_TESTED", "state": "PASS_WITH_HELD_CAUSES"},
    ]
    return {
        "dpio_read_version": "v0.1.0",
        "creator_capsule_id": capsule.get("capsule_id"),
        "procedure": "SOURCE_BOUND_SINGLE_CASE_DETERMINISTIC_READ",
        "execution_order_receipt": execution_order,
        "observed_facts": observed_facts,
        "creator_reported_context": reported,
        "supported_inferences": supported_inferences,
        "unresolved_causes": unresolved,
        "controls_and_competing_conditions": controls,
        "causal_families": causal_families,
        "frozen_discriminator_predictions": frozen_predictions,
        "competing_hypotheses": hypotheses,
        "pressure_test_results": [
            {
                "hypothesis": h["hypothesis"],
                "current_fit": h["fit"],
                "result": "HELD_PENDING_FALSIFIERS_AND_SOURCE_RETURN",
            } for h in hypotheses
        ],
        "minimum_cut_candidates": minimum_cuts,
        "source_return_requests": source_returns,
        "claim_ceiling": "L1_DIRECT_OBSERVATION_AUTOMATIC; L2-L4 HUMAN_CONFIRMATION; L5-L6 BLOCKED",
        "human_review_required": True,
        "closure_state": capsule.get("loop_state"),
        "false_closure_blocked": bool(capsule.get("review_pack", {}).get("closure_blockers", [])),
    }