diff --git a/scripts/temporal_reasoner.py b/scripts/temporal_reasoner.py new file mode 100644 index 00000000..58808910 --- /dev/null +++ b/scripts/temporal_reasoner.py @@ -0,0 +1,307 @@ +#!/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() \ No newline at end of file