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", [])), }