Phase 8.2 — CausalGraph: temporal cause-effect relationship mapping

[web3guru888]

Phase 8.2 — CausalGraph: Temporal Cause-Effect Relationship Mapping

Phase: 8 — Introspection & Explainability
Sub-phase: 8.2
Depends on: #276 (Phase 8.1 DecisionTracer)
Labels: enhancement, phase-8, architecture

---

Overview

CausalGraph is a directed acyclic graph (DAG) structure that records temporal cause-effect relationships between DecisionTrace events produced by the DecisionTracer (Phase 8.1). It enables post-hoc reasoning about which past decisions influenced later outcomes and surfaces the graph for the ExplainAPI (Phase 8.3).

DecisionTracer ──produces──► DecisionTrace
                                   │
                           CausalEdge (cause → effect, weight, lag_ms)
                                   │
                           CausalGraph (DAG, topological order, cycle guard)
                                   │
                          CausalGraphSnapshot ──► ExplainAPI (Phase 8.3)

---

New Types

CausalEdge (frozen dataclass)

@dataclass(frozen=True)
class CausalEdge:
    cause_trace_id: str          # DecisionTrace.trace_id of the cause
    effect_trace_id: str         # DecisionTrace.trace_id of the effect
    weight: float                # correlation strength in [0.0, 1.0]
    lag_ms: float                # temporal lag (effect_ts - cause_ts) in ms
    edge_type: str = "temporal"  # "temporal" | "saliency" | "attention" | "shapley"

CausalNode (frozen dataclass)

@dataclass(frozen=True)
class CausalNode:
    trace_id: str
    module_id: str
    phase: str                   # CyclePhase value at decision time
    timestamp_ms: float          # epoch milliseconds
    summary: str                 # short label for visualisation

CausalGraphSnapshot (dataclass)

@dataclass
class CausalGraphSnapshot:
    nodes: list[CausalNode]
    edges: list[CausalEdge]
    captured_at: float           # time.time()
    max_depth: int               # longest path length from any root
    root_ids: list[str]          # trace_ids with in-degree == 0
    leaf_ids: list[str]          # trace_ids with out-degree == 0

CausalGraph (main class)

class CausalGraph:
    """Thread-safe DAG of causal relationships between DecisionTrace events."""

    def __init__(
        self,
        max_nodes: int = 1024,
        eviction: Literal["lru", "fifo"] = "lru",
    ) -> None: ...

    # Mutation
    def add_node(self, node: CausalNode) -> None: ...
    def add_edge(self, edge: CausalEdge) -> None: ...  # raises CycleDetectedError

    # Query
    def ancestors(self, trace_id: str, depth: int = 3) -> list[CausalNode]: ...
    def descendants(self, trace_id: str, depth: int = 3) -> list[CausalNode]: ...
    def critical_path(self) -> list[CausalNode]: ...  # longest path in DAG
    def topological_sort(self) -> list[CausalNode]: ...

    # Serialisation
    def snapshot(self) -> CausalGraphSnapshot: ...
    def to_dot(self) -> str: ...    # Graphviz DOT language output
    def to_json(self) -> str: ...   # JSON adjacency list

CycleDetectedError(ValueError) — raised by add_edge() when adding an edge would create a cycle.

---

CausalGraphBuilder — automatic edge inference from DecisionTraces

class CausalGraphBuilder:
    """Watches a TraceStorage and infers CausalEdge entries automatically."""

    def __init__(
        self,
        graph: CausalGraph,
        storage: TraceStorage,                   # from Phase 8.1
        window_ms: float = 5000.0,               # correlation window
        min_weight: float = 0.1,                 # discard weak edges
        edge_strategy: Literal["saliency", "uniform", "attention"] = "saliency",
    ) -> None: ...

    async def infer_edges(self, new_trace: DecisionTrace) -> int:
        """Scan recent traces in storage; add edges where contributor overlap
        exceeds min_weight.  Returns number of edges added."""
        ...

    def build_node(self, trace: DecisionTrace) -> CausalNode: ...

---

build_causal_graph() factory

def build_causal_graph(
    storage: TraceStorage,
    *,
    max_nodes: int = 1024,
    eviction: Literal["lru", "fifo"] = "lru",
    window_ms: float = 5000.0,
    min_weight: float = 0.1,
    edge_strategy: Literal["saliency", "uniform", "attention"] = "saliency",
) -> tuple[CausalGraph, CausalGraphBuilder]:
    """Return a (graph, builder) pair ready for integration."""
    ...

---

Cycle Detection Algorithm

CausalGraph uses DFS-based cycle detection on every add_edge call:

1. Maintain in_degree dict (updated on every add_edge call).
2. Before committing a new edge (u → v), run DFS from v to check if u is reachable from v. If yes → raise CycleDetectedError.
3. Time complexity: O(V + E) per insertion (bounded by max_nodes).

---

Node Eviction Strategy

When len(nodes) >= max_nodes:

Policy	Eviction target	Use case
lru	Least-recently-accessed node and its incident edges	Default; preserves recent causal chains
fifo	Oldest-inserted node by timestamp_ms	Deterministic in tests

---

Integration with DecisionTracer (Phase 8.1)

# In CognitiveCycle._run_phase8():
tracer: DecisionTracer = self._phase8_tracer
graph: CausalGraph = self._phase8_graph
builder: CausalGraphBuilder = self._phase8_builder

trace = await tracer.trace(context, strategy="saliency")
node = builder.build_node(trace)
graph.add_node(node)
edges_added = await builder.infer_edges(trace)

---

Prometheus Metrics

Pre-initialise all counters/gauges to 0 at module import to avoid "no data" gaps:

Metric	Type	Labels	Description
asi_causal_nodes_total	Gauge	—	Current live node count
asi_causal_edges_total	Gauge	edge_type	Current live edge count by type
asi_causal_cycle_rejected_total	Counter	—	Edges rejected due to cycle detection
asi_causal_eviction_total	Counter	policy	Nodes evicted (lru/fifo)
asi_causal_infer_duration_seconds	Histogram	—	infer_edges() wall-clock time

---

Test Targets (12)

#	Test	Covers
1	test_add_node_deduplicates	Same trace_id twice → still 1 node
2	test_add_edge_cycle_raises	A→B→C→A raises CycleDetectedError
3	test_topological_sort_order	DAG of 5 nodes → correct topo order
4	test_ancestors_depth	ancestors(leaf, depth=2) returns ≤2 hops
5	test_descendants_depth	descendants(root, depth=2) returns ≤2 hops
6	test_critical_path_linear	Linear DAG → path = all nodes
7	test_lru_eviction	max_nodes=3; access node A, add 2 more; node B (LRU) evicted
8	test_fifo_eviction	max_nodes=3; oldest inserted node evicted
9	test_to_dot_output	to_dot() contains digraph, node labels, edge arrows
10	test_infer_edges_window	Traces outside window_ms → no edge added
11	test_infer_edges_min_weight	Weak contributors below min_weight → no edge
12	test_snapshot_fields	snapshot() root_ids/leaf_ids correct for diamond DAG

---

Module Placement

asi/
└── phase8/
    ├── decision_tracer.py   # Phase 8.1 (existing)
    └── causal_graph.py      # Phase 8.2 (NEW)

---

Implementation Order

1. CausalEdge + CausalNode frozen dataclasses
2. CycleDetectedError
3. Internal _AdjList helper (dict[str, set[str]] for forward + reverse)
4. CausalGraph.__init__() — adjacency lists, in_degree, access-order tracking
5. CausalGraph.add_node() + eviction logic
6. CausalGraph.add_edge() + DFS cycle check
7. CausalGraph.ancestors() / descendants() (BFS up to depth)
8. CausalGraph.critical_path() (DP on DAG)
9. CausalGraph.topological_sort() (Kahn's algorithm)
10. CausalGraph.to_dot() / to_json() / snapshot()
11. CausalGraphBuilder + infer_edges()
12. build_causal_graph() factory
13. Prometheus pre-init block
14. Unit tests (12 targets above)

[web3guru888]

Implementation Notes — Phase 8.2 CausalGraph

This comment documents design decisions and code sketches to unblock the implementing engineer.

---

1 Module placement

asi/phase8/causal_graph.py
tests/phase8/test_causal_graph.py

causal_graph.py imports from decision_tracer.py (same package) — no circular imports since DecisionTracer does not import CausalGraph.

---

2 Internal adjacency representation

from collections import OrderedDict
from threading import Lock

class CausalGraph:
    def __init__(self, max_nodes: int = 1024, eviction: str = "lru") -> None:
        self._nodes: OrderedDict[str, CausalNode] = OrderedDict()  # preserves insertion order; also used for FIFO
        self._fwd: dict[str, set[str]] = {}   # trace_id → set of effect trace_ids
        self._rev: dict[str, set[str]] = {}   # trace_id → set of cause trace_ids
        self._in_degree: dict[str, int] = {}
        self._access_ts: dict[str, float] = {}  # for LRU
        self._max_nodes = max_nodes
        self._eviction = eviction
        self._lock = Lock()
        self._init_metrics()

OrderedDict gives O(1) move-to-end (for LRU touch) via move_to_end().

---

3 Cycle detection sketch

def _has_path(self, src: str, dst: str) -> bool:
    """DFS: returns True if dst is reachable from src."""
    visited: set[str] = set()
    stack = [src]
    while stack:
        node = stack.pop()
        if node == dst:
            return True
        if node in visited:
            continue
        visited.add(node)
        stack.extend(self._fwd.get(node, set()))
    return False

def add_edge(self, edge: CausalEdge) -> None:
    with self._lock:
        # Cycle guard: would adding cause→effect create a path effect→cause?
        if self._has_path(edge.effect_trace_id, edge.cause_trace_id):
            CAUSAL_CYCLE_REJECTED.inc()
            raise CycleDetectedError(
                f"Adding {edge.cause_trace_id!r}→{edge.effect_trace_id!r} "
                f"would create a cycle"
            )
        self._fwd.setdefault(edge.cause_trace_id, set()).add(edge.effect_trace_id)
        self._rev.setdefault(edge.effect_trace_id, set()).add(edge.cause_trace_id)
        self._in_degree[edge.effect_trace_id] = (
            self._in_degree.get(edge.effect_trace_id, 0) + 1
        )
        CAUSAL_EDGES_GAUGE.labels(edge_type=edge.edge_type).inc()

---

4 LRU eviction sketch

def _evict_one_lru(self) -> None:
    # OrderedDict.move_to_end() is O(1); first() is the LRU
    lru_id, lru_node = next(iter(self._nodes.items()))
    self._remove_node(lru_id)
    CAUSAL_EVICTION_COUNTER.labels(policy="lru").inc()

def _touch(self, trace_id: str) -> None:
    """Mark as recently used (moves to end of OrderedDict)."""
    if trace_id in self._nodes:
        self._nodes.move_to_end(trace_id)

def add_node(self, node: CausalNode) -> None:
    with self._lock:
        if node.trace_id in self._nodes:
            self._touch(node.trace_id)
            return   # deduplication
        if len(self._nodes) >= self._max_nodes:
            if self._eviction == "lru":
                self._evict_one_lru()
            else:  # fifo: OrderedDict insertion order == FIFO
                self._evict_one_fifo()
        self._nodes[node.trace_id] = node
        self._fwd[node.trace_id] = set()
        self._rev[node.trace_id] = set()
        self._in_degree[node.trace_id] = 0
        CAUSAL_NODES_GAUGE.set(len(self._nodes))

---

5 Critical path (DP on DAG)

def critical_path(self) -> list[CausalNode]:
    """Return the longest path (by hop count) from any root to any leaf."""
    topo = self.topological_sort()
    dp: dict[str, int] = {n.trace_id: 0 for n in topo}
    prev: dict[str, str | None] = {n.trace_id: None for n in topo}
    for node in topo:
        for eff_id in self._fwd.get(node.trace_id, set()):
            if dp[node.trace_id] + 1 > dp[eff_id]:
                dp[eff_id] = dp[node.trace_id] + 1
                prev[eff_id] = node.trace_id
    # Back-track from max-depth leaf
    end_id = max(dp, key=dp.__getitem__)
    path: list[str] = []
    cur: str | None = end_id
    while cur is not None:
        path.append(cur)
        cur = prev[cur]
    path.reverse()
    return [self._nodes[tid] for tid in path if tid in self._nodes]

---

6 CausalGraphBuilder.infer_edges() sketch

async def infer_edges(self, new_trace: DecisionTrace) -> int:
    now_ms = new_trace.timestamp_ms
    recent = await self._storage.recent(since_ms=now_ms - self._window_ms)
    added = 0
    for past_trace in recent:
        if past_trace.trace_id == new_trace.trace_id:
            continue
        lag = now_ms - past_trace.timestamp_ms
        if lag < 0:
            continue  # future trace (clock skew) — skip
        # Contributor overlap as edge weight
        past_ids = {c.module_id for c in past_trace.contributors}
        new_ids  = {c.module_id for c in new_trace.contributors}
        overlap  = len(past_ids & new_ids) / max(len(past_ids | new_ids), 1)
        if overlap < self._min_weight:
            continue
        edge = CausalEdge(
            cause_trace_id=past_trace.trace_id,
            effect_trace_id=new_trace.trace_id,
            weight=overlap,
            lag_ms=lag,
            edge_type=self._edge_strategy,
        )
        try:
            self._graph.add_edge(edge)
            added += 1
        except CycleDetectedError:
            pass   # silently skip cycle-inducing edges from inference
    return added

---

7 to_dot() sketch

def to_dot(self) -> str:
    lines = ["digraph CausalGraph {", '  rankdir=LR;', '  node [shape=box];']
    for node in self._nodes.values():
        label = f"{node.module_id}\n{node.phase}"
        lines.append(f'  "{node.trace_id}" [label="{label}"];')
    for cause_id, effects in self._fwd.items():
        for eff_id in effects:
            lines.append(f'  "{cause_id}" -> "{eff_id}";')
    lines.append("}")
    return "\n".join(lines)

---

8 mypy annotations table

Symbol	Type	Notes
CausalGraph._nodes	OrderedDict[str, CausalNode]
CausalGraph._fwd	dict[str, set[str]]
CausalGraph.ancestors() return	list[CausalNode]
CausalGraphBuilder.infer_edges()	async def ... -> int
build_causal_graph()	-> tuple[CausalGraph, CausalGraphBuilder]

---

9 Prometheus pre-init

from prometheus_client import Counter, Gauge, Histogram

CAUSAL_NODES_GAUGE        = Gauge("asi_causal_nodes_total", "Live causal nodes")
CAUSAL_EDGES_GAUGE        = Gauge("asi_causal_edges_total", "Live causal edges", ["edge_type"])
CAUSAL_CYCLE_REJECTED     = Counter("asi_causal_cycle_rejected_total", "Cycle-rejected edges")
CAUSAL_EVICTION_COUNTER   = Counter("asi_causal_eviction_total", "Evicted nodes", ["policy"])
CAUSAL_INFER_HIST         = Histogram("asi_causal_infer_duration_seconds", "infer_edges() latency")

# Pre-init to 0
CAUSAL_NODES_GAUGE.set(0)
for et in ("temporal", "saliency", "attention", "shapley"):
    CAUSAL_EDGES_GAUGE.labels(edge_type=et)
for pol in ("lru", "fifo"):
    CAUSAL_EVICTION_COUNTER.labels(policy=pol)

---

10 Test isolation skeleton

import pytest
from prometheus_client import CollectorRegistry

@pytest.fixture()
def reg():
    return CollectorRegistry()

@pytest.fixture()
def graph():
    # Use small max_nodes for eviction tests; import with fresh registry
    from asi.phase8.causal_graph import CausalGraph
    return CausalGraph(max_nodes=10, eviction="lru")

def make_node(tid: str, ts: float = 0.0) -> CausalNode:
    return CausalNode(trace_id=tid, module_id="m", phase="LEARNING", timestamp_ms=ts, summary=tid)

def make_edge(cause: str, effect: str, w: float = 0.5) -> CausalEdge:
    return CausalEdge(cause_trace_id=cause, effect_trace_id=effect, weight=w, lag_ms=10.0)

def test_add_edge_cycle_raises(graph):
    graph.add_node(make_node("A"))
    graph.add_node(make_node("B"))
    graph.add_node(make_node("C"))
    graph.add_edge(make_edge("A", "B"))
    graph.add_edge(make_edge("B", "C"))
    with pytest.raises(CycleDetectedError):
        graph.add_edge(make_edge("C", "A"))

[web3guru888]

Phase 8.3 filed — ExplainAPI (#283)

Phase 8 sub-phase tracker

Sub-phase	Component	Issue	Status
8.1	DecisionTracer	#276	🟡 In spec
8.2	CausalGraph	#280	🟡 In spec
8.3	ExplainAPI	#283	🟡 In spec
8.4	Docker / Helm	—	📋 Planned
8.5	Sepolia CI	—	📋 Planned

---

CausalGraph → ExplainAPI integration

The ExplainAPI in #283 wraps CausalGraph behind three routes:

Route	CausalGraph method	Notes
GET /graph/snapshot	graph.snapshot()	full CausalGraphSnapshot
GET /graph/ancestors/{node_id}	graph.ancestors(id, max_depth)	BFS, depth-capped
GET /graph/descendants/{node_id}	graph.descendants(id, max_depth)	BFS, depth-capped
GET /graph/critical-path	graph.critical_path()	returns (path, weight) tuple

CausalGraph needs two new convenience methods to support the API:

• ancestors(node_id: str, max_depth: int = 10) -> list[str] — BFS traversal up the DAG
• descendants(node_id: str, max_depth: int = 10) -> list[str] — BFS traversal down the DAG
• __contains__(node_id: str) -> bool — if node_id not in graph guard
• __len__() -> int — node_count for snapshot response

These are non-breaking additions to the Phase 8.2 spec.

[web3guru888]

Phase 8.3 Filed — ExplainAPI

Phase 8.3 spec is now open in #287: REST interface that exposes CausalGraph and DecisionTracer data over HTTP.

Phase 8 sub-phase tracker

Sub-phase	Issue	Status
8.1 DecisionTracer	#276	✅ in spec
8.2 CausalGraph	#280	✅ in spec
8.3 ExplainAPI	#287	🔵 in spec
8.4 Docker/Helm hardening	—	🔜 planned
8.5 Sepolia CI pipeline	—	🔜 planned

8.2 → 8.3 integration point

ExplainAPI consumes CausalGraph via two endpoints:

• GET /graph/snapshot → wraps CausalGraph.snapshot()
• GET /graph/ancestors/{node_id} → BFS ancestor walk over CausalGraph.predecessors()
• GET /graph/critical-path → delegates to CausalGraph.critical_path()

CausalGraph requires no changes; ExplainAPI depends on it as a read-only consumer.