Phase 17.1 — TemporalGraph: directed event graph with Allen interval relations

[web3guru888]

Overview

TemporalGraph is the foundational data structure for Phase 17 — Temporal Reasoning & Predictive Cognition. It maintains a directed acyclic graph (DAG) of time-stamped cognitive events and world-state snapshots, annotated with Allen interval relations to capture the qualitative temporal relationships between events.

Downstream components (EventSequencer, PredictiveEngine, SchedulerCortex) all query the TemporalGraph to obtain causal and temporal context.

Enum — AllenRelation

from enum import Enum, auto

class AllenRelation(Enum):
    # 7 base relations (+ 6 inverses)
    BEFORE = auto()        # A ends before B starts
    MEETS = auto()         # A ends exactly when B starts
    OVERLAPS = auto()      # A starts before B, ends inside B
    STARTS = auto()        # A and B start together; A ends first
    DURING = auto()        # A is fully inside B
    FINISHES = auto()      # A and B end together; B starts first
    EQUALS = auto()        # A and B are identical intervals
    # Inverses
    AFTER = auto()
    MET_BY = auto()
    OVERLAPPED_BY = auto()
    STARTED_BY = auto()
    CONTAINS = auto()
    FINISHED_BY = auto()

Frozen Dataclasses

from __future__ import annotations
from dataclasses import dataclass, field
from typing import FrozenSet, Any

@dataclass(frozen=True)
class TemporalNode:
    node_id: str                        # UUID or stable hash
    timestamp_ns: int                   # time.time_ns() at event creation
    state_snapshot: dict[str, Any]      # shallow copy of relevant state
    tags: FrozenSet[str] = field(default_factory=frozenset)

@dataclass(frozen=True)
class TemporalEdge:
    from_id: str
    to_id: str
    relation: AllenRelation
    duration_ns: int                    # |to.timestamp_ns - from.timestamp_ns|

@dataclass(frozen=True)
class TemporalGraphConfig:
    max_nodes: int = 10_000
    consistency_check: bool = True      # enforce DAG on add_edge
    prune_horizon_s: float = 3600.0     # prune nodes older than this

Protocol — TemporalGraph

from typing import Protocol, runtime_checkable, Sequence

@runtime_checkable
class TemporalGraph(Protocol):
    async def add_node(self, node: TemporalNode) -> None: ...
    async def add_edge(self, edge: TemporalEdge) -> None: ...
    async def get_successors(self, node_id: str) -> Sequence[TemporalNode]: ...
    async def get_predecessors(self, node_id: str) -> Sequence[TemporalNode]: ...
    async def check_consistency(self) -> bool: ...
    async def prune(self, horizon_ns: int) -> int: ...
    def stats(self) -> dict[str, int]: ...

Implementation — DictTemporalGraph

import asyncio
import time
from collections import defaultdict, deque
from typing import Sequence

class DictTemporalGraph:
    def __init__(self, config: TemporalGraphConfig = TemporalGraphConfig()) -> None:
        self._config = config
        self._lock = asyncio.Lock()
        self._nodes: dict[str, TemporalNode] = {}
        self._edges: list[TemporalEdge] = []
        self._successors: dict[str, list[str]] = defaultdict(list)
        self._predecessors: dict[str, list[str]] = defaultdict(list)
        self._insertion_order: deque[str] = deque()
        self._nodes_added = 0
        self._edges_added = 0
        self._cycle_rejections = 0
        self._prune_count = 0

    async def add_node(self, node: TemporalNode) -> None:
        async with self._lock:
            if node.node_id in self._nodes:
                return
            if len(self._nodes) >= self._config.max_nodes:
                oldest = self._insertion_order.popleft()
                self._nodes.pop(oldest, None)
            self._nodes[node.node_id] = node
            self._insertion_order.append(node.node_id)
            self._nodes_added += 1

    async def add_edge(self, edge: TemporalEdge) -> None:
        async with self._lock:
            if edge.from_id not in self._nodes or edge.to_id not in self._nodes:
                raise KeyError(f"Unknown node(s): {edge.from_id!r}, {edge.to_id!r}")
            if self._config.consistency_check:
                self._successors[edge.from_id].append(edge.to_id)
                if self._has_cycle(edge.to_id):
                    self._successors[edge.from_id].remove(edge.to_id)
                    self._cycle_rejections += 1
                    raise ValueError(f"Edge would create a cycle")
            self._successors[edge.from_id].append(edge.to_id)
            self._predecessors[edge.to_id].append(edge.from_id)
            self._edges.append(edge)
            self._edges_added += 1

    def _has_cycle(self, start: str) -> bool:
        visited: set[str] = set()
        in_stack: set[str] = set()
        def dfs(node: str) -> bool:
            visited.add(node)
            in_stack.add(node)
            for nxt in self._successors.get(node, []):
                if nxt not in visited:
                    if dfs(nxt): return True
                elif nxt in in_stack: return True
            in_stack.discard(node)
            return False
        return dfs(start)

    async def get_successors(self, node_id: str) -> Sequence[TemporalNode]:
        async with self._lock:
            return [self._nodes[nid] for nid in self._successors.get(node_id, []) if nid in self._nodes]

    async def get_predecessors(self, node_id: str) -> Sequence[TemporalNode]:
        async with self._lock:
            return [self._nodes[nid] for nid in self._predecessors.get(node_id, []) if nid in self._nodes]

    async def check_consistency(self) -> bool:
        async with self._lock:
            visited: set[str] = set()
            in_stack: set[str] = set()
            def dfs(node: str) -> bool:
                visited.add(node); in_stack.add(node)
                for nxt in self._successors.get(node, []):
                    if nxt not in visited:
                        if dfs(nxt): return True
                    elif nxt in in_stack: return True
                in_stack.discard(node); return False
            for nid in list(self._nodes):
                if nid not in visited:
                    if dfs(nid): return False
            return True

    async def prune(self, horizon_ns: int) -> int:
        async with self._lock:
            stale = {nid for nid, n in self._nodes.items() if n.timestamp_ns < horizon_ns}
            for nid in stale:
                self._nodes.pop(nid)
                self._successors.pop(nid, None)
                self._predecessors.pop(nid, None)
            self._insertion_order = deque(x for x in self._insertion_order if x not in stale)
            self._edges = [e for e in self._edges if e.from_id not in stale and e.to_id not in stale]
            self._prune_count += len(stale)
            return len(stale)

    def stats(self) -> dict[str, int]:
        return {
            "nodes": len(self._nodes),
            "edges": len(self._edges),
            "nodes_added_total": self._nodes_added,
            "edges_added_total": self._edges_added,
            "cycle_rejections": self._cycle_rejections,
            "pruned_total": self._prune_count,
        }

NullTemporalGraph — no-op sentinel

class NullTemporalGraph:
    async def add_node(self, node: TemporalNode) -> None: pass
    async def add_edge(self, edge: TemporalEdge) -> None: pass
    async def get_successors(self, node_id: str) -> Sequence[TemporalNode]: return []
    async def get_predecessors(self, node_id: str) -> Sequence[TemporalNode]: return []
    async def check_consistency(self) -> bool: return True
    async def prune(self, horizon_ns: int) -> int: return 0
    def stats(self) -> dict[str, int]: return {}

Prometheus Metrics

Metric	Type	Labels	Description
asi_temporal_graph_nodes_total	Counter	—	Cumulative nodes added
asi_temporal_graph_edges_total	Counter	—	Cumulative edges added
asi_temporal_graph_cycle_rejections_total	Counter	—	Edges rejected due to cycle
asi_temporal_graph_size	Gauge	—	Current live node count
asi_temporal_graph_prune_total	Counter	—	Total nodes pruned

12 Test Targets

1. test_add_node_increments_counter
2. test_duplicate_node_ignored
3. test_max_nodes_evicts_oldest
4. test_add_edge_unknown_node_raises
5. test_add_edge_creates_allen_relation
6. test_cycle_detection_raises
7. test_get_successors_returns_correct_nodes
8. test_get_predecessors_returns_correct_nodes
9. test_check_consistency_clean_dag
10. test_prune_removes_stale_nodes
11. test_prune_removes_dangling_edges
12. test_null_temporal_graph_no_ops

---

Phase 17 sub-phase tracker

• 17.1 TemporalGraph (this issue)
• 17.2 EventSequencer
• 17.3 PredictiveEngine
• 17.4 SchedulerCortex
• 17.5 TemporalOrchestrator

[web3guru888]

Implementation Notes

Allen's Interval Algebra — 13 relations for directed DAG

Allen (1983) defined 13 mutually exclusive, exhaustive relations between time intervals. For a directed graph only 7 base relations + 6 inverses are needed; the direction of the edge implicitly encodes the inverse:

Relation	Inverse	Meaning
BEFORE	AFTER	A completely precedes B
MEETS	MET_BY	A ends exactly when B starts
OVERLAPS	OVERLAPPED_BY	A starts before B, ends inside B
STARTS	STARTED_BY	Same start; A is shorter
DURING	CONTAINS	A is fully inside B
FINISHES	FINISHED_BY	Same end; B is longer
EQUALS	EQUALS	Identical interval

Practical tip: derive relation automatically from from.timestamp_ns and to.timestamp_ns when duration_ns is known. Supply BEFORE as the default for point events.

timestamp_ns — time.time_ns() rationale

• time.time_ns() returns wall-clock time as an int (no float rounding), giving nanosecond resolution.
• Alternative: time.monotonic_ns() for intra-process ordering (no wall-clock jumps), but loses cross-process synchronisation.
• Recommended: use time.time_ns() for inter-agent federation; use time.monotonic_ns() only for in-process benchmarks.

Cycle Detection — DFS with visited + in_stack

def _has_cycle(self, start: str) -> bool:
    # O(V + E) per call; called only on add_edge under lock
    visited: set[str] = set()
    in_stack: set[str] = set()  # tracks the recursion stack

    def dfs(node: str) -> bool:
        visited.add(node)
        in_stack.add(node)
        for nxt in self._successors.get(node, []):
            if nxt not in visited:
                if dfs(nxt):
                    return True
            elif nxt in in_stack:
                return True  # back-edge detected
        in_stack.discard(node)
        return False

    return dfs(start)

Note: called under asyncio.Lock so thread safety is guaranteed; the DFS is synchronous (no await) so no re-entrancy risk.

Pruning Strategy — wall-clock age vs graph depth

Strategy	Pro	Con
Wall-clock age (timestamp_ns < horizon)	Simple; integrates with prune_horizon_s	May prune causally important old nodes
Graph depth (keep top-N ancestors)	Preserves causal chains	Complex; harder to bound memory
Hybrid (recommended)	Prune by age unless node has no successors (leaf) — keep leaves regardless of age	Balances memory and causal completeness

For Phase 17.1, implement wall-clock age pruning. Hybrid strategy can be added in 17.5 TemporalOrchestrator.

mypy Narrowing

Pattern	mypy issue	Fix
self._nodes[nid] after pop check	KeyError	Use .get(nid) + assert
Sequence[TemporalNode] return	Variance	Annotate as list[TemporalNode] in impl
AllenRelation in edge	No narrowing needed	edge.relation is already typed

Test Skeletons

async def test_add_edge_creates_allen_relation():
    g = DictTemporalGraph()
    n1 = TemporalNode("a", 1_000_000_000, {}, frozenset())
    n2 = TemporalNode("b", 2_000_000_000, {}, frozenset())
    await g.add_node(n1)
    await g.add_node(n2)
    edge = TemporalEdge("a", "b", AllenRelation.BEFORE, 1_000_000_000)
    await g.add_edge(edge)
    succs = await g.get_successors("a")
    assert succs[0].node_id == "b"
    assert g.stats()["edges"] == 1

async def test_cycle_detection_raises():
    g = DictTemporalGraph()
    for nid in ["x", "y", "z"]:
        await g.add_node(TemporalNode(nid, 0, {}, frozenset()))
    await g.add_edge(TemporalEdge("x", "y", AllenRelation.BEFORE, 0))
    await g.add_edge(TemporalEdge("y", "z", AllenRelation.BEFORE, 0))
    with pytest.raises(ValueError, match="cycle"):
        await g.add_edge(TemporalEdge("z", "x", AllenRelation.BEFORE, 0))
    assert g.stats()["cycle_rejections"] == 1

---

See also: Discussion #433 (Phase 17 Planning) · Discussion #435 (Show & Tell) · Discussion #436 (Q&A)

[web3guru888]

Phase 17.2 — EventSequencer filed 🕐

Issue #437 has been opened for the EventSequencer component — the direct consumer of TemporalGraph.add_node().

Integration point

EventSequencer sits upstream of TemporalGraph in the pipeline:

CognitiveCycle → AsyncEventSequencer.drain() → TemporalGraph.add_node()

After drain() yields time-ordered CognitiveEvent objects, the caller infers Allen relations between consecutive events and calls TemporalGraph.add_edge() with the appropriate AllenRelation enum value.

Key design decisions affecting TemporalGraph

• CognitiveEvent.timestamp_ns: int — same int nanosecond convention as TemporalNode.timestamp_ns, no float conversion needed
• CognitiveEvent.event_id: str → maps directly to TemporalNode.node_id: str
• AsyncEventSequencer guarantees timestamp-ordered delivery via heapq — TemporalGraph DFS cycle detection is still the final guard

See #437 and Phase-17-Event-Sequencer wiki for full spec.

[web3guru888]

Phase 17 — COMPLETE 🎉

Phase 17 is now fully spec'd with the filing of #446 — TemporalOrchestrator.

TemporalGraph is the persistence layer in the completed orchestration pipeline:

TemporalOrchestrator._GRAPHING phase
    └─► TemporalGraph.add_node(TemporalNode)  ← one node per SequenceWindow
            └─► DictTemporalGraph stores Allen relations
                    └─► asibuild_temporal_graph_nodes_gauge (Prometheus)
                            └─► Grafana saturation alert if > 50k nodes

Phase 17 Final Tracker:

Sub-Phase	Component	Issue	Wiki
17.1	TemporalGraph	#434	✅
17.2	EventSequencer	#437	✅
17.3	PredictiveEngine	#440	✅
17.4	SchedulerCortex	#443	✅
17.5	TemporalOrchestrator	#446	✅

Wiki: Phase-17-Temporal-Orchestrator