timmy-config/scripts/temporal_reasoner.py

#!/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()