feat(scripts): add symbolic forward-chaining rule engine

scripts/symbolic_reasoner.py

@@ -0,0 +1,276 @@
+#!/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()

scripts/temporal_reasoner.py

@@ -1,307 +0,0 @@
-#!/usr/bin/env python3
-"""temporal_reasoner.py - GOFAI temporal reasoning engine for the Timmy Foundation fleet.
-
-A symbolic temporal constraint network (TCN) for scheduling and ordering events.
-Models Allen's interval algebra relations (before, after, meets, overlaps, etc.)
-and propagates temporal constraints via path-consistency to detect conflicts.
-No ML, no embeddings - just constraint propagation over a temporal graph.
-
-Core concepts:
-    TimePoint:  A named instant on a symbolic timeline.
-    Interval:   A pair of time-points (start, end) with start < end.
-    Constraint: A relation between two time-points or intervals
-                (e.g. A.before(B), A.meets(B)).
-
-Usage (Python API):
-    from temporal_reasoner import TemporalNetwork, Interval
-    tn = TemporalNetwork()
-    deploy = tn.add_interval('deploy', duration=(10, 30))
-    test   = tn.add_interval('test',   duration=(5, 15))
-    tn.add_constraint(deploy, 'before', test)
-    consistent = tn.propagate()
-
-CLI:
-    python temporal_reasoner.py --demo
-"""
-
-from __future__ import annotations
-
-import argparse
-import sys
-from dataclasses import dataclass, field
-from enum import Enum
-from typing import Dict, List, Optional, Set, Tuple
-
-INF = float('inf')
-
-# ---------------------------------------------------------------------------
-# Data model
-# ---------------------------------------------------------------------------
-
-
-@dataclass(frozen=True)
-class TimePoint:
-    """A named instant on the timeline."""
-    name: str
-    id: int = field(default=0)
-
-    def __str__(self) -> str:
-        return self.name
-
-
-@dataclass
-class Interval:
-    """A named interval bounded by two time-points."""
-    name: str
-    start: int  # index into the distance matrix
-    end: int    # index into the distance matrix
-
-    def __str__(self) -> str:
-        return self.name
-
-
-class Relation(Enum):
-    """Allen's interval algebra relations (simplified subset)."""
-    BEFORE = 'before'
-    AFTER = 'after'
-    MEETS = 'meets'
-    MET_BY = 'met_by'
-    OVERLAPS = 'overlaps'
-    DURING = 'during'
-    EQUALS = 'equals'
-
-
-# ---------------------------------------------------------------------------
-# Simple Temporal Network (STN) via distance matrix
-# ---------------------------------------------------------------------------
-
-
-class TemporalNetwork:
-    """Simple Temporal Network with Floyd-Warshall propagation.
-
-    Internally maintains a distance matrix D where D[i][j] is the
-    maximum allowed distance from time-point i to time-point j.
-    Negative cycles indicate inconsistency.
-    """
-
-    def __init__(self) -> None:
-        self._n = 0
-        self._names: List[str] = []
-        self._dist: List[List[float]] = []
-        self._intervals: Dict[str, Interval] = {}
-        self._origin_idx: int = -1
-        self._add_point('origin')
-        self._origin_idx = 0
-
-    # ------------------------------------------------------------------
-    # Point management
-    # ------------------------------------------------------------------
-
-    def _add_point(self, name: str) -> int:
-        """Add a time-point and return its index."""
-        idx = self._n
-        self._n += 1
-        self._names.append(name)
-        # Extend distance matrix
-        for row in self._dist:
-            row.append(INF)
-        self._dist.append([INF] * self._n)
-        self._dist[idx][idx] = 0.0
-        return idx
-
-    # ------------------------------------------------------------------
-    # Interval management
-    # ------------------------------------------------------------------
-
-    def add_interval(
-        self,
-        name: str,
-        duration: Optional[Tuple[float, float]] = None,
-    ) -> Interval:
-        """Add a named interval with optional duration bounds [lo, hi].
-
-        Returns the Interval object with start/end indices.
-        """
-        s = self._add_point(f"{name}.start")
-        e = self._add_point(f"{name}.end")
-        # start < end  (at least 1 time unit)
-        self._dist[s][e] = min(self._dist[s][e], duration[1] if duration else INF)
-        self._dist[e][s] = min(self._dist[e][s], -(duration[0] if duration else 1))
-        interval = Interval(name=name, start=s, end=e)
-        self._intervals[name] = interval
-        return interval
-
-    # ------------------------------------------------------------------
-    # Constraint management
-    # ------------------------------------------------------------------
-
-    def add_distance_constraint(
-        self, i: int, j: int, lo: float, hi: float
-    ) -> None:
-        """Add constraint: lo <= t_j - t_i <= hi."""
-        self._dist[i][j] = min(self._dist[i][j], hi)
-        self._dist[j][i] = min(self._dist[j][i], -lo)
-
-    def add_constraint(
-        self, a: Interval, relation: str, b: Interval, gap: Tuple[float, float] = (0, INF)
-    ) -> None:
-        """Add an Allen-style relation between two intervals.
-
-        Supported relations: before, after, meets, met_by, equals.
-        """
-        rel = relation.lower()
-        if rel == 'before':
-            # a.end + gap <= b.start
-            self.add_distance_constraint(a.end, b.start, gap[0], gap[1])
-        elif rel == 'after':
-            self.add_distance_constraint(b.end, a.start, gap[0], gap[1])
-        elif rel == 'meets':
-            # a.end == b.start
-            self.add_distance_constraint(a.end, b.start, 0, 0)
-        elif rel == 'met_by':
-            self.add_distance_constraint(b.end, a.start, 0, 0)
-        elif rel == 'equals':
-            self.add_distance_constraint(a.start, b.start, 0, 0)
-            self.add_distance_constraint(a.end, b.end, 0, 0)
-        else:
-            raise ValueError(f"Unsupported relation: {relation}")
-
-    # ------------------------------------------------------------------
-    # Propagation (Floyd-Warshall)
-    # ------------------------------------------------------------------
-
-    def propagate(self) -> bool:
-        """Run Floyd-Warshall to propagate all constraints.
-
-        Returns True if the network is consistent (no negative cycles).
-        """
-        n = self._n
-        d = self._dist
-        for k in range(n):
-            for i in range(n):
-                for j in range(n):
-                    if d[i][k] + d[k][j] < d[i][j]:
-                        d[i][j] = d[i][k] + d[k][j]
-        # Check for negative cycles
-        for i in range(n):
-            if d[i][i] < 0:
-                return False
-        return True
-
-    def is_consistent(self) -> bool:
-        """Check consistency without mutating (copies matrix first)."""
-        import copy
-        saved = copy.deepcopy(self._dist)
-        result = self.propagate()
-        self._dist = saved
-        return result
-
-    # ------------------------------------------------------------------
-    # Query
-    # ------------------------------------------------------------------
-
-    def earliest(self, point_idx: int) -> float:
-        """Earliest possible time for a point (relative to origin)."""
-        return -self._dist[point_idx][self._origin_idx]
-
-    def latest(self, point_idx: int) -> float:
-        """Latest possible time for a point (relative to origin)."""
-        return self._dist[self._origin_idx][point_idx]
-
-    def interval_bounds(self, interval: Interval) -> Dict[str, Tuple[float, float]]:
-        """Return earliest/latest start and end for an interval."""
-        return {
-            'start': (self.earliest(interval.start), self.latest(interval.start)),
-            'end': (self.earliest(interval.end), self.latest(interval.end)),
-        }
-
-    # ------------------------------------------------------------------
-    # Display
-    # ------------------------------------------------------------------
-
-    def dump(self) -> None:
-        """Print the current distance matrix and interval bounds."""
-        print(f"Temporal Network — {self._n} time-points, {len(self._intervals)} intervals")
-        print()
-        for name, interval in self._intervals.items():
-            bounds = self.interval_bounds(interval)
-            s_lo, s_hi = bounds['start']
-            e_lo, e_hi = bounds['end']
-            print(f"  {name}:")
-            print(f"    start: [{s_lo:.1f}, {s_hi:.1f}]")
-            print(f"    end:   [{e_lo:.1f}, {e_hi:.1f}]")
-
-
-# ---------------------------------------------------------------------------
-# Demo: Timmy fleet deployment pipeline
-# ---------------------------------------------------------------------------
-
-
-def run_demo() -> None:
-    """Run a demo temporal reasoning scenario for the Timmy fleet."""
-    print("=" * 60)
-    print("Temporal Reasoner Demo - Fleet Deployment Pipeline")
-    print("=" * 60)
-    print()
-
-    tn = TemporalNetwork()
-
-    # Define pipeline stages with duration bounds [min, max]
-    build   = tn.add_interval('build',   duration=(5, 15))
-    test    = tn.add_interval('test',    duration=(10, 30))
-    review  = tn.add_interval('review',  duration=(2, 10))
-    deploy  = tn.add_interval('deploy',  duration=(1, 5))
-    monitor = tn.add_interval('monitor', duration=(20, 60))
-
-    # Temporal constraints
-    tn.add_constraint(build, 'meets', test)        # test starts when build ends
-    tn.add_constraint(test, 'before', review, gap=(0, 5))  # review within 5 of test
-    tn.add_constraint(review, 'meets', deploy)     # deploy immediately after review
-    tn.add_constraint(deploy, 'before', monitor, gap=(0, 2))  # monitor within 2 of deploy
-
-    # Global deadline: everything done within 120 time units
-    tn.add_distance_constraint(tn._origin_idx, monitor.end, 0, 120)
-
-    # Build must start within first 10 units
-    tn.add_distance_constraint(tn._origin_idx, build.start, 0, 10)
-
-    print("Constraints added. Propagating...")
-    consistent = tn.propagate()
-    print(f"Network consistent: {consistent}")
-    print()
-
-    if consistent:
-        tn.dump()
-        print()
-
-        # Now add a conflicting constraint to show inconsistency detection
-        print("--- Adding conflicting constraint: monitor.before(build) ---")
-        tn.add_constraint(monitor, 'before', build)
-        consistent2 = tn.propagate()
-        print(f"Network consistent after conflict: {consistent2}")
-
-
-# ---------------------------------------------------------------------------
-# CLI
-# ---------------------------------------------------------------------------
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(
-        description="GOFAI temporal reasoning engine"
-    )
-    parser.add_argument(
-        "--demo",
-        action="store_true",
-        help="Run the fleet deployment pipeline demo",
-    )
-    args = parser.parse_args()
-
-    if args.demo or not any(vars(args).values()):
-        run_demo()
-    else:
-        parser.print_help()
-
-
-if __name__ == "__main__":
-    main()