timmy-config/scripts/symbolic_reasoner.py

#!/usr/bin/env python3
"""symbolic_reasoner.py — Forward-chaining rule engine for the Timmy Foundation fleet.

A classical GOFAI approach: declarative IF-THEN rules evaluated over a
working-memory of facts.  Rules fire until quiescence (no new facts) or
a configurable cycle limit.  Designed to sit *beside* the LLM layer so
that hard policy constraints never depend on probabilistic inference.

Usage:
    python symbolic_reasoner.py --rules rules.yaml --facts facts.yaml
    python symbolic_reasoner.py --self-test
"""
from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Dict, FrozenSet, List, Optional, Set, Tuple

try:
    import yaml
except ImportError:
    yaml = None  # graceful fallback — JSON-only mode


# ---------------------------------------------------------------------------
# Domain types
# ---------------------------------------------------------------------------
Fact = Tuple[str, ...]  # e.g. ("agent", "timmy", "role", "infrastructure")


@dataclass(frozen=True)
class Rule:
    """A single IF-THEN production rule."""
    name: str
    conditions: FrozenSet[Fact]       # all must be present
    negations: FrozenSet[Fact]        # none may be present
    conclusions: FrozenSet[Fact]      # added when rule fires
    priority: int = 0                 # higher fires first

    def satisfied(self, wm: Set[Fact]) -> bool:
        return self.conditions.issubset(wm) and self.negations.isdisjoint(wm)


# ---------------------------------------------------------------------------
# Engine
# ---------------------------------------------------------------------------
class SymbolicReasoner:
    """Forward-chaining production system."""

    def __init__(self, rules: List[Rule], *, cycle_limit: int = 200):
        self._rules = sorted(rules, key=lambda r: -r.priority)
        self._cycle_limit = cycle_limit
        self._trace: List[str] = []

    # -- public API ---------------------------------------------------------

    def infer(self, initial_facts: Set[Fact]) -> Set[Fact]:
        """Run to quiescence and return the final working-memory."""
        wm = set(initial_facts)
        fired: Set[str] = set()
        for cycle in range(self._cycle_limit):
            progress = False
            for rule in self._rules:
                if rule.name in fired:
                    continue
                if rule.satisfied(wm):
                    new = rule.conclusions - wm
                    if new:
                        wm |= new
                        fired.add(rule.name)
                        self._trace.append(
                            f"cycle {cycle}: {rule.name} => {_fmt_facts(new)}"
                        )
                        progress = True
                        break  # restart from highest-priority rule
            if not progress:
                break
        return wm

    def query(self, wm: Set[Fact], pattern: Tuple[Optional[str], ...]) -> List[Fact]:
        """Return facts matching *pattern* (None = wildcard)."""
        return [
            f for f in wm
            if len(f) == len(pattern)
            and all(p is None or p == v for p, v in zip(pattern, f))
        ]

    @property
    def trace(self) -> List[str]:
        return list(self._trace)

    # -- serialisation helpers -----------------------------------------------

    @classmethod
    def from_dicts(cls, raw_rules: List[Dict], **kw) -> "SymbolicReasoner":
        rules = [_parse_rule(r) for r in raw_rules]
        return cls(rules, **kw)

    @classmethod
    def from_file(cls, path: Path, **kw) -> "SymbolicReasoner":
        text = path.read_text()
        if path.suffix in (".yaml", ".yml"):
            if yaml is None:
                raise RuntimeError("PyYAML required for .yaml rules")
            data = yaml.safe_load(text)
        else:
            data = json.loads(text)
        return cls.from_dicts(data["rules"], **kw)


# ---------------------------------------------------------------------------
# Parsing helpers
# ---------------------------------------------------------------------------
def _parse_fact(raw: list | str) -> Fact:
    if isinstance(raw, str):
        return tuple(raw.split())
    return tuple(str(x) for x in raw)


def _parse_rule(d: Dict) -> Rule:
    return Rule(
        name=d["name"],
        conditions=frozenset(_parse_fact(c) for c in d.get("if", [])),
        negations=frozenset(_parse_fact(c) for c in d.get("unless", [])),
        conclusions=frozenset(_parse_fact(c) for c in d.get("then", [])),
        priority=d.get("priority", 0),
    )


def _fmt_facts(facts: Set[Fact]) -> str:
    return ", ".join(" ".join(f) for f in sorted(facts))


# ---------------------------------------------------------------------------
# Built-in fleet rules (loaded when no --rules file is given)
# ---------------------------------------------------------------------------
DEFAULT_FLEET_RULES: List[Dict] = [
    {
        "name": "route-ci-to-timmy",
        "if": [["task", "category", "ci"]],
        "then": [["assign", "timmy"], ["reason", "timmy", "best-ci-merge-rate"]],
        "priority": 10,
    },
    {
        "name": "route-security-to-timmy",
        "if": [["task", "category", "security"]],
        "then": [["assign", "timmy"], ["reason", "timmy", "security-specialist"]],
        "priority": 10,
    },
    {
        "name": "route-architecture-to-gemini",
        "if": [["task", "category", "architecture"]],
        "unless": [["assign", "timmy"]],
        "then": [["assign", "gemini"], ["reason", "gemini", "architecture-strength"]],
        "priority": 8,
    },
    {
        "name": "route-review-to-allegro",
        "if": [["task", "category", "review"]],
        "then": [["assign", "allegro"], ["reason", "allegro", "highest-quality-per-pr"]],
        "priority": 9,
    },
    {
        "name": "route-frontend-to-claude",
        "if": [["task", "category", "frontend"]],
        "unless": [["task", "repo", "fleet-ops"]],
        "then": [["assign", "claude"], ["reason", "claude", "high-volume-frontend"]],
        "priority": 5,
    },
    {
        "name": "block-merge-without-review",
        "if": [["pr", "status", "open"], ["pr", "reviews", "0"]],
        "then": [["pr", "action", "block-merge"], ["reason", "policy", "no-unreviewed-merges"]],
        "priority": 20,
    },
    {
        "name": "block-merge-ci-failing",
        "if": [["pr", "status", "open"], ["pr", "ci", "failing"]],
        "then": [["pr", "action", "block-merge"], ["reason", "policy", "ci-must-pass"]],
        "priority": 20,
    },
    {
        "name": "auto-label-hotfix",
        "if": [["pr", "title-prefix", "hotfix"]],
        "then": [["pr", "label", "hotfix"], ["pr", "priority", "urgent"]],
        "priority": 15,
    },
]


# ---------------------------------------------------------------------------
# Self-test
# ---------------------------------------------------------------------------
def _self_test() -> bool:
    """Verify core behaviour; returns True on success."""
    engine = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)

    # Scenario 1: CI task should route to Timmy
    wm = engine.infer({("task", "category", "ci")})
    assert ("assign", "timmy") in wm, f"expected timmy assignment, got {wm}"

    # Scenario 2: architecture task routes to gemini (not timmy)
    engine2 = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
    wm2 = engine2.infer({("task", "category", "architecture")})
    assert ("assign", "gemini") in wm2, f"expected gemini assignment, got {wm2}"

    # Scenario 3: open PR with no reviews should block merge
    engine3 = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
    wm3 = engine3.infer({("pr", "status", "open"), ("pr", "reviews", "0")})
    assert ("pr", "action", "block-merge") in wm3

    # Scenario 4: negation — frontend + fleet-ops should NOT assign claude
    engine4 = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)
    wm4 = engine4.infer({("task", "category", "frontend"), ("task", "repo", "fleet-ops")})
    assert ("assign", "claude") not in wm4

    # Scenario 5: query with wildcards
    results = engine.query(wm, ("reason", None, None))
    assert len(results) > 0

    print("All 5 self-test scenarios passed.")
    for line in engine.trace:
        print(f"  {line}")
    return True


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--rules", type=Path, help="YAML/JSON rule file")
    ap.add_argument("--facts", type=Path, help="YAML/JSON initial facts")
    ap.add_argument("--self-test", action="store_true")
    ap.add_argument("--json", action="store_true", help="output as JSON")
    args = ap.parse_args()

    if args.self_test:
        ok = _self_test()
        sys.exit(0 if ok else 1)

    if args.rules:
        engine = SymbolicReasoner.from_file(args.rules)
    else:
        engine = SymbolicReasoner.from_dicts(DEFAULT_FLEET_RULES)

    if args.facts:
        text = args.facts.read_text()
        if args.facts.suffix in (".yaml", ".yml"):
            raw = yaml.safe_load(text)
        else:
            raw = json.loads(text)
        initial = {_parse_fact(f) for f in raw.get("facts", [])}
    else:
        initial = set()
        print("No --facts provided; running with empty working memory.")

    wm = engine.infer(initial)

    if args.json:
        print(json.dumps({"facts": [list(f) for f in sorted(wm)], "trace": engine.trace}, indent=2))
    else:
        print(f"Final working memory ({len(wm)} facts):")
        for f in sorted(wm):
            print(f"  {' '.join(f)}")
        if engine.trace:
            print(f"\nInference trace ({len(engine.trace)} firings):")
            for line in engine.trace:
                print(f"  {line}")


if __name__ == "__main__":
    main()