cmetta_space.py

"""Purpose: a space whose matching runs in cmetta, a C MeTTa runtime,
reached as a subprocess: the same SpaceProvider seam that carries SQLite,
DuckDB, Redis and remote engines carries a sibling MeTTa implementation.

The bridge is STORAGE-level on purpose. The atoms live here as MeTTa
atoms; cmetta is consulted per query as a matcher over their text, its
evaluator never runs, and the engine binds every answer through its own
unification. A semantic quirk in cmetta's matcher surfaces as a missing
or extra candidate that the conformance kit's pattern family catches,
not as a silently wrong local answer.

Guarantees:
  - removal takes every stored occurrence cmetta matches for the pattern,
    decided in ONE subprocess run by tagging each stored atom with its
    index and matching (probe $i <pattern>) [tested
    test_cmetta_space.py::test_removal_is_by_unification]
  - an atom whose text spans lines is refused at add, because the
    subprocess crossing is line-shaped
Open Obligations:
  To Do: None
  Hacks: None
  Future Enhancements: None
"""

from __future__ import annotations

import os
import shutil
import subprocess
from collections.abc import Iterator
from typing import Any

from _common import check, done, skip

from metta._declare import declarations as _space_declarations
from metta.foreign import SpaceProvider

_PROBE = "metta-cmetta-probe"


class CMettaSpace(SpaceProvider):
    """Atoms stored beside their text, pattern queries answered by cmetta."""

    def __init__(self, cmetta: str | None = None, timeout: float = 30.0):
        self._cmetta = cmetta or os.environ.get("METTA_CMETTA", "cmetta")
        self._timeout = timeout
        self._atoms: list[Any] = []

    # -- the provider surface -------------------------------------------------

    def add(self, atom) -> None:
        text = str(atom)
        if "\n" in text:
            raise ValueError(f"an atom's text may not span lines: {text!r}")
        self._atoms.append(atom)

    def atoms(self) -> Iterator:
        return iter(list(self._atoms))

    def match(self, pattern) -> Iterator:
        indices = self._matching_indices(pattern)
        return iter([self._atoms[i] for i in indices])

    def remove(self, pattern) -> bool:
        """One occurrence, the oldest: a space is a multiset and removal
        subtracts from it, so two identical atoms need two removals."""
        doomed = self._matching_indices(pattern)
        if not doomed:
            return False
        del self._atoms[min(doomed)]
        return True

    def clear(self) -> None:
        self._atoms.clear()

    # -- the cmetta crossing --------------------------------------------------

    def _matching_indices(self, pattern) -> list[int]:
        """Which stored atoms match, decided by cmetta in one run.

        Every stored atom is asserted as (probe i atom), and the query
        (match &self (probe $i <pattern>) $i) answers exactly the indices
        whose atom cmetta matches against the pattern, with the pattern's
        variables freshened per candidate by match itself.
        """
        if not self._atoms:
            return []
        program = [
            "-e",
            " ".join(f"({_PROBE} {i} {atom})" for i, atom in enumerate(self._atoms)),
            "-e",
            f"!(match &self ({_PROBE} $metta-cmetta-i {pattern}) $metta-cmetta-i)",
        ]
        answer, errors = self._run(program)
        if not answer:
            #cmetta prints nothing at all for an empty answer set, not an
            #empty list, and refuses malformed input with a nonzero exit,
            #so silence here really is "no stored atom matches".
            return []
        return [int(text) for text in _bracket_items(answer, errors) if text]

    def _run(self, arguments: list[str]) -> tuple[str, str]:
        completed = subprocess.run(  # noqa: S603 - the binary is the caller's own configuration
            [self._cmetta, "--quiet", *arguments],
            capture_output=True,
            text=True,
            timeout=self._timeout,
            check=False,
        )
        if completed.returncode != 0:
            raise RuntimeError(
                f"cmetta exited {completed.returncode}: "
                f"{completed.stderr.strip() or completed.stdout.strip()}"
            )
        return completed.stdout.strip(), completed.stderr.strip()


class CMettaMatch:
    """A grounded value whose matching IS cmetta evaluation.

    Held in a MeTTa expression and unified against an operand, it yields
    one answer atom per result cmetta produced for its query, and the
    operand's variables bind by destructuring those answers: arbitrary
    bindings from a foreign evaluator, arriving through ordinary
    unification. This is the matcher tier of the seam, where the value's
    own logic is the authority and nothing re-derives its claims; a
    space is exactly such a value whose matcher is query.

        matcher = CMettaMatch("(sol 2) (sol -2)",
                             "!(match &self (sol $s) (sol $s))", m.parse)
        !(unify <matcher> (sol $x) $x none)   ; answers 2 and -2
    """

    def __init__(self, program: str, query: str, parse, cmetta: str | None = None,
                 timeout: float = 30.0):
        self._program = program
        self._query = query
        self._parse = parse
        self._space = CMettaSpace(cmetta=cmetta, timeout=timeout)

    def match_(self, other):
        answer, _errors = self._space._run(["-e", self._program, "-e", self._query])
        if not answer:
            return
        for text in _bracket_items(answer):
            if text:
                yield self._parse(text)


def _bracket_items(answer: str, errors: str = "") -> list[str]:
    """Split cmetta's `[a, b, c]` answer line at top-level commas."""
    line = answer.splitlines()[-1].strip() if answer else ""
    if not (line.startswith("[") and line.endswith("]")):
        raise RuntimeError(
            f"cmetta answered an unexpected shape: stdout {answer!r}, "
            f"stderr {errors[-500:]!r}"
        )
    inner = line[1:-1].strip()
    if not inner:
        return []
    items, depth, start = [], 0, 0
    for position, character in enumerate(inner):
        if character == "(":
            depth += 1
        elif character == ")":
            depth -= 1
        elif character == "," and depth == 0:
            items.append(inner[start:position].strip())
            start = position + 1
    items.append(inner[start:].strip())
    return items


def demo() -> None:
    """The worked run: this engine's queries answered over atoms cmetta
    matches, and cmetta evaluation results binding variables inside it."""
    cmetta = os.environ.get("METTA_CMETTA") or shutil.which("cmetta")
    if cmetta is None:
        skip("cmetta is not on PATH and METTA_CMETTA does not name it")
    import metta
    from metta import Expression, S, V
    from metta._atoms.factories import Grounded

    m = metta.MeTTa().space()
    space = CMettaSpace(cmetta=cmetta)
    _space_declarations._register_space(m, space, "&cmetta")
    m.run("!(add-atom &cmetta (edge a b))")
    m.run("!(add-atom &cmetta (edge a c))")
    (group,) = m.run("!(collapse (match &cmetta (edge a $x) $x))")
    check("cmetta matches, this engine binds", sorted(str(a) for a in group[0]),
          ["b", "c"])

    matcher = CMettaMatch(
        "(sol 2) (sol -2)",
        "!(match &self (sol $s) (sol $s))",
        m.parse,
        cmetta=cmetta,
    )
    rows = m.eval(Expression(S.unify, Grounded(matcher), Expression(S.sol, V.x), V.x, S.none))
    check("cmetta answers bind inside unify", sorted(str(a) for a in rows),
          ["-2", "2"])
    done("cmetta_space")


if __name__ == "__main__":
    demo()