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

"""Command Intensity Core for TinyMind.

This core converts raw user instructions into intent, constraints, evidence
requirements, and an execution plan. It is deliberately anti-template: vague or
repetitive prompts are not answered with canned text; they are converted into a
clarify-then-execute plan.
"""

from __future__ import annotations

from collections import Counter
from dataclasses import dataclass
import re
from typing import Any


@dataclass(frozen=True)
class CommandSignal:
    keyword: str
    intent: str
    constraint: str | None = None
    priority: int = 1


SIGNALS = (
    CommandSignal("โมเดล", "model_improvement", priority=4),
    CommandSignal("พัฒนา", "model_improvement", priority=3),
    CommandSignal("train", "training"),
    CommandSignal("เทรน", "training"),
    CommandSignal("วัดผล", "evaluation", "real_eval_required"),
    CommandSignal("หลักฐาน", "evaluation", "evidence_required"),
    CommandSignal("log", "evaluation", "evidence_required"),
    CommandSignal("ไม่ตอบฟิก", "instruction_following", "no_fixed_answer"),
    CommandSignal("ฟิก", "instruction_following", "no_fixed_answer"),
    CommandSignal("ภาษาไทย", "language_quality", "thai_naturalness"),
    CommandSignal("โค้ด", "coding", "code_correctness"),
    CommandSignal("ค้นหา", "retrieval", "retrieve_before_claim"),
    CommandSignal("เครื่องมือ", "tool_grounding", "tool_observation_required"),
)


class CommandIntensityCore:
    def analyze(self, command: str) -> dict[str, Any]:
        text = " ".join(command.strip().split())
        lower = text.lower()
        matched = [signal for signal in SIGNALS if signal.keyword.lower() in lower]
        intent = self._intent(matched)
        constraints = sorted({signal.constraint for signal in matched if signal.constraint})
        constraints.extend(self._implicit_constraints(lower))
        constraints = sorted(set(constraints))
        anti_template = self._anti_template_gate(text)
        evidence_required = self._evidence_required(constraints)
        execution_mode = "execute_with_evidence" if anti_template["passed"] and len(text) >= 12 else "clarify_then_execute"
        return {
            "input": command,
            "normalized": text,
            "intent": intent,
            "constraints": constraints,
            "evidence_required": evidence_required,
            "execution_mode": execution_mode,
            "anti_template_gate": anti_template,
            "efficiency_plan": self._efficiency_plan(intent, constraints, evidence_required),
            "claim_boundary": {
                "perfect_instruction_following_claim_allowed": False,
                "requires_task_result_before_success_claim": True,
                "no_fixed_answer_policy": "generate from parsed intent and evidence, not memorized phrasing",
            },
        }

    @staticmethod
    def _intent(signals: list[CommandSignal]) -> str:
        if not signals:
            return "general_task"
        counts: Counter[str] = Counter()
        for signal in signals:
            counts[signal.intent] += signal.priority
        return counts.most_common(1)[0][0]

    @staticmethod
    def _implicit_constraints(lower: str) -> list[str]:
        constraints = []
        if "จริง" in lower or "ไม่ปลอม" in lower:
            constraints.append("real_eval_required")
        if "ห้าม" in lower:
            constraints.append("explicit_negative_constraints")
        if "ประสิทธิภาพ" in lower or "ทรัพยากร" in lower:
            constraints.append("resource_efficiency")
        if "บริสุทธิ" in lower or "เพียว" in lower:
            constraints.append("purity_required")
        if "ทุกคำสั่ง" in lower or "รับฟัง" in lower:
            constraints.append("instruction_obedience")
        return constraints

    @staticmethod
    def _evidence_required(constraints: list[str]) -> list[str]:
        evidence = ["action_trace"]
        if "real_eval_required" in constraints:
            evidence.append("eval_json")
        if "evidence_required" in constraints:
            evidence.append("evidence_log")
        if "retrieve_before_claim" in constraints:
            evidence.append("source_url_or_local_manifest")
        if "tool_observation_required" in constraints:
            evidence.append("tool_observation")
        if "resource_efficiency" in constraints:
            evidence.append("runtime_or_vram_metric")
        return sorted(set(evidence))

    @staticmethod
    def _anti_template_gate(text: str) -> dict[str, Any]:
        tokens = re.findall(r"\S+", text.lower())
        thai_like = bool(re.search(r"[\u0E00-\u0E7F]", text))
        reasons = []
        if len(tokens) < 4 and not (thai_like and len(text) >= 24):
            reasons.append("too_short")
        if len(tokens) > 1 and len(set(tokens)) == 1:
            reasons.append("repetition_or_low_specificity")
        if tokens and not (thai_like and len(tokens) <= 3 and len(text) >= 24):
            most_common_count = Counter(tokens).most_common(1)[0][1]
            if most_common_count / max(len(tokens), 1) >= 0.45:
                reasons.append("repetition_or_low_specificity")
        if "lorem ipsum" in text.lower():
            reasons.append("template_artifact")
        return {
            "passed": not reasons,
            "reasons": reasons,
            "specificity_score": min(1.0, len(set(tokens)) / max(len(tokens), 1) + min(len(tokens), 20) / 80.0),
        }

    @staticmethod
    def _efficiency_plan(intent: str, constraints: list[str], evidence_required: list[str]) -> list[dict[str, str]]:
        plan = [
            {
                "step": "1",
                "action": "extract_user_success_criteria",
                "why": "avoid wasted work and fixed-answer behavior",
            },
            {
                "step": "2",
                "action": f"route_to_{intent}",
                "why": "use the smallest capable subsystem for the command",
            },
            {
                "step": "3",
                "action": "collect_required_evidence",
                "why": ", ".join(evidence_required),
            },
            {
                "step": "4",
                "action": "answer_naturally_with_result_and_limits",
                "why": "high obedience without unsupported claims",
            },
        ]
        if "resource_efficiency" in constraints:
            plan.insert(
                2,
                {
                    "step": "2b",
                    "action": "prefer_low_cost_path",
                    "why": "preserve power while reducing unnecessary work",
                },
            )
        return plan