core/packs/common/runtime_rules.py

"""Deterministic runtime helpers for local language-pack rules."""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

from core.physics.energy import EnergyClass
from core.physics.valence import lift_valence
from core_ingest.types import (
    CandidateGeometricPressure,
    DeterminismClass,
    FrontendTrace,
    Modality,
    ReviewLevel,
    SourceSpan,
)


def read_jsonl(path: Path) -> list[dict[str, Any]]:
    return [
        json.loads(line)
        for line in path.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]


def analysis_payload(analysis: object) -> dict[str, Any]:
    if isinstance(analysis, dict):
        payload = dict(analysis.get("input", analysis))
    else:
        payload = dict(getattr(analysis, "__dict__", {}))
    if not payload:
        raise ValueError("analysis must expose lemma_id or sense_id fields")
    return payload


def lift_from_pack(pack_dir: Path, analysis: object, *, language: str) -> list[CandidateGeometricPressure]:
    payload = analysis_payload(analysis)
    senses = {record["sense_id"]: record for record in read_jsonl(pack_dir / "senses.jsonl")}
    lemmas = {record["lemma_id"]: record for record in read_jsonl(pack_dir / "lemmas.jsonl")}
    sense = senses.get(str(payload.get("sense_id", "")))
    lemma_id = str(payload.get("lemma_id") or (sense or {}).get("lemma_id") or "")
    lemma = lemmas.get(lemma_id)
    if lemma is None:
        raise KeyError(f"unknown lemma_id: {lemma_id}")
    field_target = str((sense or {}).get("field_target") or lemma["field_hooks"][0])
    pressure_kind = str(payload.get("pressure_kind", "semantic"))
    features = {
        "morph_class": lemma.get("morph_class", ""),
        "semantic_family": lemma.get("semantic_family", ""),
    }
    valence = lift_valence(
        lemma=str(lemma["script_form"]),
        language=language,
        features=features,
    ).to_payload()
    packet_payload = {
        "field_target": field_target,
        "pressure_kind": pressure_kind,
        "energy_class_hint": EnergyClass.E2.value,
        "valence": valence,
        "source": {
            "lemma_id": lemma_id,
            "sense_id": payload.get("sense_id"),
            "frame_id": payload.get("frame_id"),
        },
    }
    canonical = json.dumps(packet_payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    span = SourceSpan(byte_start=0, byte_end=max(1, len(canonical.encode("utf-8"))), source_sha256=digest)
    packet = CandidateGeometricPressure(
        kind=pressure_kind,
        modality=Modality.SCRIPTURE if language in {"he", "el", "grc"} else Modality.TEXT,
        provenance=(span,),
        frontend=FrontendTrace(
            instrument_id=f"{language}.lift_rules",
            determinism=DeterminismClass.D0,
            version="1.0.0",
        ),
        review_level=ReviewLevel.AUTO_ACCEPT_ELIGIBLE,
        confidence=1.0,
        uncertainty=0.0,
        lemma=str(lemma["script_form"]),
        payload_json=canonical,
    )
    return [packet]