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

"""Realtime world-context layer for TinyMind.

This layer gives TinyMind access to current time, coarse location context, and
live web evidence at runtime.  It is deliberately tool-grounded: the report
states what came from runtime evidence versus what is inside model weights.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import tempfile
from typing import Any

import httpx

from data.general_web_knowledge import GeneralWebKnowledgeEngine


def _sha_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()


@dataclass(frozen=True)
class WorldContextPolicy:
    allow_live_web: bool = True
    allow_ip_geolocation: bool = True
    coarse_location_only: bool = True
    max_results: int = 6
    top_k: int = 4
    timeout_s: float = 12.0
    language: str = "auto"


class WorldContextLayer:
    """Runtime adapter for time, coarse place, and current web evidence."""

    def __init__(
        self,
        root: str | Path | None = None,
        *,
        policy: WorldContextPolicy | None = None,
        web_engine: GeneralWebKnowledgeEngine | None = None,
        client: httpx.Client | None = None,
    ):
        self.root = Path(root) if root is not None else Path(tempfile.mkdtemp(prefix="tinymind_world_context_"))
        self.root.mkdir(parents=True, exist_ok=True)
        self.policy = policy or WorldContextPolicy()
        self.client = client or httpx.Client(timeout=self.policy.timeout_s, follow_redirects=True)
        self.web_engine = web_engine or GeneralWebKnowledgeEngine()

    def answer(self, question: str, *, context: str = "") -> dict[str, Any]:
        question = question.strip()
        if not question:
            raise ValueError("question_required")
        time_context = self._time_context()
        location_context = self._location_context()
        web_context = self._web_context(question)
        answer = self._compose_answer(question, time_context, location_context, web_context, context=context)
        checks = {
            "time_context_ok": bool(time_context.get("local_iso") and time_context.get("utc_iso")),
            "location_context_ok": location_context.get("status") in {"coarse_location", "timezone_only", "disabled"},
            "web_context_policy_ok": (self.policy.allow_live_web and web_context.get("status") != "disabled")
            or (not self.policy.allow_live_web and web_context.get("status") == "disabled"),
            "web_context_grounded_or_graceful": (
                not self.policy.allow_live_web
                or web_context.get("status") in {"grounded", "insufficient_evidence", "error"}
            ),
            "answer_non_empty": bool(answer.strip()),
            "report_auditable": True,
        }
        report = {
            "schema_version": "tinymind-world-context-v1",
            "created_at": datetime.now(timezone.utc).isoformat(),
            "question": question,
            "question_sha256": _sha_text(question),
            "policy": asdict(self.policy),
            "answer": answer,
            "time_context": time_context,
            "location_context": location_context,
            "web_context": web_context,
            "context_sha256": _sha_text(context) if context else None,
            "checks": checks,
            "claim_gate": {
                "world_context_runtime_ready": all(checks.values()),
                "current_knowledge_access_ready": checks["time_context_ok"] and checks["web_context_grounded_or_graceful"],
                "coarse_location_ready": location_context.get("status") in {"coarse_location", "timezone_only", "disabled"},
                "model_weight_current_knowledge_claim": False,
                "exact_user_location_claim_allowed": False,
                "unsupported_answer_allowed": False,
                "world_best_claim_allowed": False,
                "reason": (
                    "Realtime facts are supplied by audited runtime context and hashed web evidence. "
                    "They are not claimed to be stored perfectly inside model weights."
                ),
            },
        }
        out_path = self.root / "world_context_report.json"
        report["json_path"] = str(out_path)
        out_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        return report

    def _time_context(self) -> dict[str, Any]:
        local_now = datetime.now().astimezone()
        utc_now = datetime.now(timezone.utc)
        return {
            "status": "ok",
            "local_iso": local_now.isoformat(),
            "utc_iso": utc_now.isoformat(),
            "timezone": str(local_now.tzinfo),
            "unix_seconds": int(utc_now.timestamp()),
        }

    def _location_context(self) -> dict[str, Any]:
        if not self.policy.allow_ip_geolocation:
            return {"status": "disabled", "reason": "ip_geolocation_disabled", "timezone": self._time_context()["timezone"]}
        try:
            response = self.client.get("https://ipapi.co/json/")
            response.raise_for_status()
            data = response.json()
            latitude = data.get("latitude")
            longitude = data.get("longitude")
            if self.policy.coarse_location_only:
                latitude = round(float(latitude), 1) if latitude is not None else None
                longitude = round(float(longitude), 1) if longitude is not None else None
            return {
                "status": "coarse_location",
                "city": data.get("city"),
                "region": data.get("region"),
                "country_name": data.get("country_name"),
                "timezone": data.get("timezone"),
                "latitude": latitude,
                "longitude": longitude,
                "precision": "coarse" if self.policy.coarse_location_only else "provider_reported",
                "source": "ipapi.co",
            }
        except Exception as exc:
            return {
                "status": "timezone_only",
                "reason": type(exc).__name__,
                "timezone": self._time_context()["timezone"],
                "source": "local_system_clock",
            }

    def _web_context(self, question: str) -> dict[str, Any]:
        if not self.policy.allow_live_web:
            return {"status": "disabled", "reason": "live_web_disabled"}
        try:
            result = self.web_engine.answer(
                question,
                self.root / "live_web",
                max_results=self.policy.max_results,
                top_k=self.policy.top_k,
                language=self.policy.language,
            )
            return {
                "status": result.get("status"),
                "answer": result.get("answer"),
                "evidence": result.get("evidence", []),
                "evidence_count": len(result.get("evidence", [])),
                "report_path": result.get("report_path"),
                "claim_gate": result.get("claim_gate", {}),
            }
        except Exception as exc:
            return {
                "status": "error",
                "error_type": type(exc).__name__,
                "error": str(exc),
                "evidence": [],
                "evidence_count": 0,
            }

    def _compose_answer(
        self,
        question: str,
        time_context: dict[str, Any],
        location_context: dict[str, Any],
        web_context: dict[str, Any],
        *,
        context: str,
    ) -> str:
        thai = any("\u0e00" <= ch <= "\u0e7f" for ch in question)
        if thai:
            lines = [
                "ผมจะตอบโดยใช้บริบทโลกปัจจุบันจาก runtime ไม่เดาจากความจำเก่า:",
                f"- เวลาเครื่องตอนนี้: {time_context['local_iso']} ({time_context['timezone']})",
                f"- เวลา UTC: {time_context['utc_iso']}",
            ]
            if location_context.get("status") == "coarse_location":
                place = ", ".join(
                    str(location_context.get(key))
                    for key in ("city", "region", "country_name")
                    if location_context.get(key)
                )
                lines.append(f"- ตำแหน่งโดยประมาณจาก IP: {place or 'ไม่ระบุ'} ({location_context.get('precision')})")
            else:
                lines.append(f"- ตำแหน่ง: {location_context.get('status')} / {location_context.get('reason', 'no exact location claim')}")
            if web_context.get("status") == "grounded":
                lines.append("")
                lines.append(str(web_context.get("answer", "")).strip())
            elif web_context.get("status") == "insufficient_evidence":
                lines.append("")
                lines.append("ค้นเว็บแล้วแต่หลักฐานยังไม่พอสำหรับสรุปข้อเท็จจริงแบบมั่นใจ จึงต้องตอบแบบมีข้อจำกัด.")
            elif web_context.get("status") == "error":
                lines.append("")
                lines.append(f"พยายามค้นเว็บแล้วแต่ติดข้อผิดพลาด: {web_context.get('error_type')}. ใช้ได้เฉพาะเวลา/ตำแหน่ง runtime ในคำตอบนี้.")
            else:
                lines.append("")
                lines.append("live web ถูกปิดไว้ใน policy จึงไม่เคลมความรู้ปัจจุบันจากเว็บ.")
            if context.strip():
                lines.append(f"- context เพิ่มเติม hash={_sha_text(context)[:16]}")
            lines.append("")
            lines.append("Gate: ข้อมูลปัจจุบันมาจาก runtime/web evidence ไม่ใช่เคลมว่า weight โมเดลรู้ทุกอย่างปัจจุบันเอง.")
            return "\n".join(lines)

        lines = [
            "I will answer from runtime world context rather than stale model memory:",
            f"- Local time: {time_context['local_iso']} ({time_context['timezone']})",
            f"- UTC time: {time_context['utc_iso']}",
        ]
        if location_context.get("status") == "coarse_location":
            place = ", ".join(
                str(location_context.get(key))
                for key in ("city", "region", "country_name")
                if location_context.get(key)
            )
            lines.append(f"- Coarse IP location: {place or 'unknown'} ({location_context.get('precision')})")
        else:
            lines.append(f"- Location: {location_context.get('status')} / {location_context.get('reason', 'no exact location claim')}")
        if web_context.get("status") == "grounded":
            lines.extend(["", str(web_context.get("answer", "")).strip()])
        elif web_context.get("status") == "insufficient_evidence":
            lines.extend(["", "Live search did not return enough verified evidence, so I should not state a firm current fact."])
        elif web_context.get("status") == "error":
            lines.extend(["", f"Live web lookup failed with {web_context.get('error_type')}; this answer uses only runtime time/place context."])
        else:
            lines.extend(["", "Live web lookup is disabled by policy."])
        lines.append("")
        lines.append("Gate: current context is runtime/web-grounded, not a claim that model weights already contain realtime world state.")
        return "\n".join(lines)


def write_world_context_answer(
    question: str,
    out_path: str | Path,
    *,
    context: str = "",
    root: str | Path | None = None,
    allow_live_web: bool = True,
    allow_ip_geolocation: bool = True,
) -> dict[str, Any]:
    root_path = Path(root) if root is not None else Path(out_path).parent / "world_context_runtime"
    report = WorldContextLayer(
        root_path,
        policy=WorldContextPolicy(allow_live_web=allow_live_web, allow_ip_geolocation=allow_ip_geolocation),
    ).answer(question, context=context)
    out = Path(out_path)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    report["json_path"] = str(out)
    return report