opusmanSEEKv4/src/nexus_os/monitoring/token_compactor.py

"""token_compactor.py — Compact agent state for efficient context passing"""

import json
from pathlib import Path
from typing import Dict, Any

STATE_COMPACT_SCHEMA = {
    "version": "1.0.0",
    "last_compacted": "ISO8601",
    "test_count": 0,
    "active_tasks": [],
    "joker_status": {},
    "route_health": {},
    "governance_state": "nominal"
}

class TokenCompactor:
    """Compact agent state to <500 tokens for efficient context passing."""

    def __init__(self, state_file: Path = Path("STATE_COMPACT.json")):
        self.state_file = state_file

    def compact(self, full_state: Dict[str, Any]) -> Dict[str, Any]:
        """Reduce full state to compact representation."""
        return {
            "version": "1.0.0",
            "tests": full_state.get("test_count", 0),
            "tasks": len(full_state.get("active_tasks", [])),
            "jokers": {k: v["status"] for k, v in full_state.get("joker_status", {}).items()},
            "governance": full_state.get("governance_state", "nominal"),
            "compact_at": full_state.get("timestamp")
        }

    def should_compact(self, turn_count: int) -> bool:
        """Compact every 3-5 turns."""
        return turn_count % 4 == 0

    def load_compact(self) -> Dict[str, Any]:
        """Load last compacted state."""
        if self.state_file.exists():
            return json.loads(self.state_file.read_text())
        return STATE_COMPACT_SCHEMA.copy()

    def save_compact(self, compact_state: Dict[str, Any]):
        """Persist compacted state."""
        self.state_file.write_text(json.dumps(compact_state, indent=2))