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

"""Omni action and file perception registry for TinyMind.

This layer does not pretend that raw bytes are magically understood. It maps
files/actions into semantic frames, required parsers, evidence boundaries, and
tool chains so the model can respond to tools and multimodal artifacts without
falling back to fixed text.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass
import json
from pathlib import Path
from typing import Any


@dataclass(frozen=True)
class ModalitySpec:
    family: str
    extensions: tuple[str, ...]
    tool_chain: tuple[str, ...]
    evidence_required: tuple[str, ...]
    risk_notes: tuple[str, ...] = ()


MODALITY_REGISTRY: tuple[ModalitySpec, ...] = (
    ModalitySpec("image", (".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tiff"), ("image.decode", "vision.embed", "ocr.extract", "caption.verify"), ("decoded_pixels", "metadata")),
    ModalitySpec("video", (".mp4", ".mkv", ".mov", ".webm", ".avi"), ("video.demux", "frame.sample", "audio.extract", "vision.timeline", "scene.summarize"), ("container_metadata", "sampled_frames")),
    ModalitySpec("audio", (".wav", ".mp3", ".flac", ".m4a", ".ogg", ".aac"), ("audio.decode", "asr.transcribe", "speaker.segment", "acoustic.event.detect"), ("duration", "transcript_or_features")),
    ModalitySpec("document", (".pdf", ".docx", ".doc", ".pptx", ".xlsx", ".csv", ".txt", ".md", ".html"), ("document.parse", "layout.extract", "table.extract", "citation.trace"), ("text_or_layout", "source_pages")),
    ModalitySpec("archive", (".zip", ".7z", ".rar", ".tar", ".gz", ".zst", ".bz2"), ("archive.list", "policy.scan", "safe.extract", "manifest.hash"), ("file_list", "hash_manifest"), ("extract only in sandbox",)),
    ModalitySpec("code", (".py", ".js", ".ts", ".tsx", ".go", ".rs", ".c", ".cpp", ".h", ".java", ".kt", ".lua", ".sql", ".sh", ".ps1"), ("code.parse", "dependency.scan", "test.plan", "sandbox.run"), ("source_text", "language_id")),
    ModalitySpec("binary", (".bin", ".exe", ".dll", ".so", ".apk", ".ipa", ".wasm", ".class", ".dex"), ("binary.identify", "hash.compute", "strings.extract", "sandbox.static_analyze"), ("hashes", "format_signature"), ("never execute without explicit sandbox policy",)),
    ModalitySpec("network", (".har", ".pcap", ".pcapng", ".curl"), ("network.parse", "endpoint.extract", "privacy.scan", "timeline.build"), ("packets_or_requests", "endpoint_manifest"), ("redact secrets before training",)),
    ModalitySpec("system_event", (".evtx", ".etl", ".log", ".journal"), ("event.parse", "timestamp.normalize", "actor.action.extract", "causal.chain"), ("event_records", "time_range")),
)


ACTION_INTENTS = {
    "sandbox.run_code": "execute_code",
    "shell.powershell": "run_shell_command",
    "web.fetch_official_doc": "retrieve_evidence",
    "browser.open": "navigate_browser",
    "file.read": "inspect_file",
    "file.write": "modify_file",
}


class OmniActionPerception:
    def __init__(self, registry: tuple[ModalitySpec, ...] = MODALITY_REGISTRY):
        self.registry = registry

    def plan_input(self, path_or_name: str) -> dict[str, Any]:
        suffix = Path(path_or_name).suffix.lower()
        for spec in self.registry:
            if suffix in spec.extensions:
                return self._plan_for_spec(path_or_name, spec, confidence=0.92)
        return {
            "input": path_or_name,
            "modality": "unknown",
            "confidence": 0.15,
            "tool_chain": ("file.signature", "mime.detect", "safe.preview"),
            "evidence_required": ("magic_bytes", "mime_type"),
            "risk_notes": ("unknown format; do not train or execute until identified",),
            "semantic_frame": {
                "object": "file",
                "action": "identify_before_use",
                "parser_required": True,
            },
            "claim_boundary": self._claim_boundary(),
        }

    def plan_action(self, event: dict[str, Any]) -> dict[str, Any]:
        tool = str(event.get("tool") or event.get("name") or "unknown")
        observation = event.get("observation")
        arguments = event.get("arguments") if isinstance(event.get("arguments"), dict) else {}
        phase = "observation" if observation is not None else "request"
        intent = ACTION_INTENTS.get(tool, "unknown_tool_action")
        return {
            "kind": "tool_action",
            "tool": tool,
            "phase": phase,
            "semantic_frame": {
                "intent": intent,
                "arguments_shape": sorted(arguments.keys()),
                "has_observation": observation is not None,
                "parser_required": intent == "unknown_tool_action",
            },
            "tool_chain": ("policy.check", "execute_or_parse", "observe", "critic.verify"),
            "claim_boundary": {
                **self._claim_boundary(),
                "observation_required_before_claim": True,
            },
        }

    def write_manifest(self, out_dir: str | Path) -> dict[str, Any]:
        out = Path(out_dir)
        out.mkdir(parents=True, exist_ok=True)
        manifest_path = out / "omni_action_perception_manifest.json"
        report = {
            "schema_version": "tinymind-omni-action-perception-v1",
            "coverage": {
                "family_count": len(self.registry),
                "families": [spec.family for spec in self.registry],
                "extension_count": sum(len(spec.extensions) for spec in self.registry),
            },
            "registry": [asdict(spec) for spec in self.registry],
            "action_intents": dict(sorted(ACTION_INTENTS.items())),
            "claim_gate": {
                "extensible_omni_perception_ready": True,
                "supports_all_world_formats_claim_allowed": False,
                "raw_bytes_understood_without_parser": False,
                "tool_observation_required_before_claim": True,
                "multimodal_native_claim_allowed": False,
            },
            "manifest_path": str(manifest_path),
        }
        manifest_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
        return report

    @staticmethod
    def _plan_for_spec(path_or_name: str, spec: ModalitySpec, confidence: float) -> dict[str, Any]:
        return {
            "input": path_or_name,
            "modality": spec.family,
            "confidence": confidence,
            "tool_chain": spec.tool_chain,
            "evidence_required": spec.evidence_required,
            "risk_notes": spec.risk_notes,
            "semantic_frame": {
                "object": "file",
                "action": "parse_then_ground",
                "parser_required": True,
            },
            "claim_boundary": OmniActionPerception._claim_boundary(),
        }

    @staticmethod
    def _claim_boundary() -> dict[str, bool]:
        return {
            "raw_bytes_understood_without_parser": False,
            "answer_requires_extracted_evidence": True,
            "safe_sandbox_required_for_active_content": True,
            "world_all_formats_claim_allowed": False,
        }