# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/by-what-partitioning-invariants-must-an-instance-segregate-mutable-task-state-from-persistent-system-instructions-across-context-compaction-boundaries.md
# Remnant Warrant:    https://corpusfetch.com/c/by-what-partitioning-invariants-must-an-instance-segregate-mutable-task-state-from-persistent-system-instructions-across-context-compaction-boundaries.warrant.json
# Markdown SHA-256:   50f8a4c270b001bbe26ed18bc87f8f15c652f3ede6b473d68ed71ef053d24806
# Code AST SHA-256:   ef376e339d5727d584609deb92aaeffd3f25a2baf8b3417e3706035153326f10
# Verified Classes:   ControlPlaneMutationError, EpisodicTaskEntry, SegregatedContextState, SegregatedCompactionManager
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass
from typing import Callable


class ControlPlaneMutationError(PermissionError):
    '''Raised when a compaction cycle alters, summarizes, or corrupts the immutable control plane.'''
    pass


@dataclass(frozen=True)
class EpisodicTaskEntry:
    entry_id: str
    subtask_id: str
    is_ephemeral_assumption: bool
    content: str


@dataclass(frozen=True)
class SegregatedContextState:
    immutable_control_plane: str
    control_plane_sha256: str
    active_subtask_id: str
    episodic_entries: tuple[EpisodicTaskEntry, ...]


class SegregatedCompactionManager:
    '''
    Enforces strict partitioning between persistent System/Developer instructions
    and mutable task state across context-compaction boundaries:
    1. Locks `immutable_control_plane` with a SHA-256 digest; forbids passing it to the summarizer.
    2. Automatically purges `is_ephemeral_assumption == True` entries from completed subtasks
       so temporary test overrides never calcify into global session rules.
    3. Re-verifies byte identity of the Control Plane before returning the compacted state.
    '''

    @classmethod
    def initialize_state(
        cls,
        *,
        control_plane_text: str,
        initial_subtask_id: str,
    ) -> SegregatedContextState:
        if not control_plane_text.strip():
            raise ValueError("Control plane text cannot be empty.")
        digest = hashlib.sha256(control_plane_text.encode("utf-8")).hexdigest()
        return SegregatedContextState(
            immutable_control_plane=control_plane_text,
            control_plane_sha256=digest,
            active_subtask_id=initial_subtask_id,
            episodic_entries=(),
        )

    @classmethod
    def compact_episodic_plane(
        cls,
        state: SegregatedContextState,
        *,
        current_active_subtask_id: str,
        summarizer_fn: Callable[[list[str]], str],
         verify_control_plane_override: str | None = None,
    ) -> SegregatedContextState:
        # Gate 1 & Gate 4: Verify Control Plane has not been modified or summarized
        candidate_cp = (
            state.immutable_control_plane
            if verify_control_plane_override is None
            else verify_control_plane_override
        )
        actual_hash = hashlib.sha256(candidate_cp.encode("utf-8")).hexdigest()
        if actual_hash != state.control_plane_sha256:
            raise ControlPlaneMutationError(
                "Compaction Breach Blocked: Immutable Control Plane bytes were altered or summarized! "
                f"Expected SHA-256 {state.control_plane_sha256}, got {actual_hash}."
            )

        # Gate 3: Purge ephemeral assumptions belonging to completed prior subtasks
        surviving_texts: list[str] = []
        for entry in state.episodic_entries:
            if entry.is_ephemeral_assumption and entry.subtask_id != current_active_subtask_id:
                continue  # Evict subtask-local temporary mock/assumption
            surviving_texts.append(f"[{entry.subtask_id}] {entry.content}")

        # Gate 2: Summarize ONLY the mutable episodic entries
        compacted_summary = summarizer_fn(surviving_texts) if surviving_texts else ""
        compacted_entry = EpisodicTaskEntry(
            entry_id=f"compacted_{current_active_subtask_id}",
            subtask_id=current_active_subtask_id,
            is_ephemeral_assumption=False,
            content=f"[EPISODIC_STATE_SUMMARY_DATA_ONLY]: {compacted_summary}",
        )

        return SegregatedContextState(
            immutable_control_plane=state.immutable_control_plane,
            control_plane_sha256=state.control_plane_sha256,
            active_subtask_id=current_active_subtask_id,
            episodic_entries=(compacted_entry,) if compacted_summary else (),
        )