tinymind-native-8b-remote-handoff/bundle/model/self_compression.py

"""Self-compression governor for TinyMind artifacts.

No hard size cap is imposed.  The rule is conservative: accept a smaller or
faster candidate only when quality and purity remain within declared gates.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass


@dataclass(frozen=True)
class CompressionCandidate:
    name: str
    quality: float
    purity: float
    size_mb: float
    latency_ms: float

    def score(self) -> float:
        size_penalty = self.size_mb / 1024.0
        latency_penalty = self.latency_ms / 1000.0
        return (2.0 * self.quality) + self.purity - (0.15 * size_penalty) - (0.10 * latency_penalty)


class SelfCompressionGovernor:
    policy = {
        "no_hard_size_limit": True,
        "always_try_smaller_when_quality_preserved": True,
        "rollback_on_quality_or_purity_drop": True,
        "claim_world_best_requires_external_evidence": True,
    }

    def __init__(self, max_quality_drop: float = 0.01, max_purity_drop: float = 0.01):
        self.max_quality_drop = float(max_quality_drop)
        self.max_purity_drop = float(max_purity_drop)

    def _block_reason(self, base: CompressionCandidate, candidate: CompressionCandidate) -> str | None:
        if base.quality - candidate.quality > self.max_quality_drop:
            return f"quality_drop:{base.quality - candidate.quality:.6f}"
        if base.purity - candidate.purity > self.max_purity_drop:
            return f"purity_drop:{base.purity - candidate.purity:.6f}"
        return None

    def choose(self, base: CompressionCandidate, candidates: list[CompressionCandidate]) -> dict:
        accepted: list[CompressionCandidate] = []
        blocked: list[str] = []
        for candidate in candidates:
            reason = self._block_reason(base, candidate)
            if reason:
                blocked.append(f"{candidate.name}:{reason}")
            else:
                accepted.append(candidate)

        selected = base
        accepted_flag = False
        reason = "rollback_to_base"
        if accepted:
            pool = [base] + accepted
            selected = max(pool, key=lambda item: item.score())
            accepted_flag = selected.name != base.name
            reason = "quality_preserved_smaller_or_faster" if accepted_flag else "base_score_remains_best"

        return {
            "schema_version": "tinymind-self-compression-governor-v1",
            "policy": dict(self.policy),
            "base": asdict(base),
            "candidates": [asdict(candidate) for candidate in candidates],
            "selected": selected.name,
            "accepted": accepted_flag,
            "reason": reason,
            "blocked": blocked,
            "selected_score": selected.score(),
        }