Q&A — Phase 17.2 EventSequencer: heapq ordering, causal validation, tumbling windows

[web3guru888]

Q&A — Phase 17.2 EventSequencer ❓

Issue: #437 | Show & Tell: #438 | Phase: 17 — Temporal Reasoning

---

Q1: Why heapq over asyncio.PriorityQueue?

A: heapq is preferred for batch drain operations. When CognitiveCycle calls drain(), it empties the entire buffer in one pass with while heap: yield heapq.heappop(heap) — this is O(k log n) for k items, all synchronous, no per-item coroutine scheduling overhead.

asyncio.PriorityQueue.get() adds coroutine scheduling overhead per item (task switch cost). For burst drains of 100–1000 events between cognitive cycles, the heapq approach can be 5–10× faster.

The tradeoff: heapq requires an explicit asyncio.Lock for thread safety. Since we run in a single event loop, the lock is only contested between ingest() and drain() coroutines — contention is minimal.

---

Q2: How does causal_parent_id work with concurrent modules?

A: CognitiveCycle runs modules sequentially in _run_step(), so prev_event_id is always set to the previous module's output event. For concurrent modules (future: Phase 9 federation), the causal graph becomes a DAG, not a chain.

In the current design:

• seen_ids is a set[str] — O(1) lookup
• A root event has causal_parent_id=None → always accepted
• A child event with causal_parent_id="e_x" is accepted if "e_x" ∈ seen_ids

TTL: seen_ids grows unboundedly in theory. Production hardening replaces it with collections.OrderedDict trimmed to max_buffer * 2 entries (oldest entries pruned first).

---

Q3: What happens when the buffer hits max_buffer?

A: LRU eviction by timestamp — the oldest event (minimum timestamp_ns) is popped from the heapq and discarded:

if len(self._heap) >= self._cfg.max_buffer:
    heapq.heappop(self._heap)   # removes minimum timestamp_ns
    self._dropped += 1
    events_dropped_total.labels(reason="overflow").inc()

This prioritises recency: when backpressure hits, stale events are sacrificed for fresh ones. The CognitiveCycle should monitor buffer_utilization_ratio > 0.8 and slow emission if needed.

Future extension: eviction_policy: Literal["oldest", "newest"] in SequencerConfig.

---

Q4: Tumbling vs sliding windows — why tumbling?

A: PredictiveEngine (Phase 17.3) needs deterministic, non-overlapping input batches to make predictions without double-counting events. Sliding windows would cause the same CognitiveEvent to appear in multiple WindowedAggregate objects, corrupting prediction inputs.

Tumbling window assignment is a single integer division:

bucket = (event.timestamp_ns // window_size_ns) * window_size_ns

This is O(1) per event and produces a unique window_id per bucket. When PredictiveEngine asks for windows, it gets clean, non-overlapping batches.

Sliding windows could be added as SequencerConfig(sliding_step_ms=...) in a future sub-phase if needed for overlapping-context prediction.

---

Q5: How does EventSequencer integrate with TemporalGraph?

A: The drain loop runs inside CognitiveCycle._run_step() after each module:

async for ordered_event in self._sequencer.drain():
    # Add time-ordered node to TemporalGraph
    await self._temporal_graph.add_node(
        ordered_event.event_id,
        ordered_event.timestamp_ns,
    )
    # Infer Allen relations between consecutive events
    if prev_node_id:
        relation = allen_relation(prev_ts, ordered_event.timestamp_ns)
        await self._temporal_graph.add_edge(prev_node_id, ordered_event.event_id, relation)
    prev_node_id = ordered_event.event_id
    prev_ts = ordered_event.timestamp_ns

This keeps TemporalGraph in sync with the ordered event stream without the CognitiveCycle needing to sort events itself.

---

Q6: Thread safety of heapq under asyncio?

A: heapq is not thread-safe, but in an asyncio single-event-loop context, only one coroutine runs at a time between await points. The asyncio.Lock in AsyncEventSequencer ensures that ingest() and drain() don't interleave:

async with self._lock:
    heapq.heappush(self._heap, event)   # atomic wrt other coroutines

If you ever move to a multi-threaded executor (loop.run_in_executor), replace with threading.Lock. The asyncio.Lock is intentionally chosen to signal "this is asyncio-only code" — a threading.Lock would be a design smell.

---

Q7: Grafana panel YAML for stream rate + window flush rate

panels:
  - title: "Event Ingest Rate"
    type: timeseries
    targets:
      - expr: "rate(asi_sequencer_events_ingested_total[1m])"
        legendFormat: "{{source_module}}"
    fieldConfig:
      defaults:
        unit: "ops"

  - title: "Event Drop Rate by Reason"
    type: timeseries
    targets:
      - expr: "rate(asi_sequencer_events_dropped_total[1m]) by (reason)"
        legendFormat: "{{reason}}"
    fieldConfig:
      defaults:
        unit: "ops"
        color:
          mode: "fixed"
          fixedColor: "red"

  - title: "Window Flush Rate"
    type: timeseries
    targets:
      - expr: "rate(asi_sequencer_windows_flushed_total[1m])"
        legendFormat: "windows/s"
    fieldConfig:
      defaults:
        unit: "wps"

  - title: "Buffer Utilization"
    type: gauge
    targets:
      - expr: "asi_sequencer_buffer_utilization_ratio"
        legendFormat: "utilization"
    fieldConfig:
      defaults:
        unit: "percentunit"
        min: 0
        max: 1
        thresholds:
          steps:
            - value: 0
              color: green
            - value: 0.8
              color: yellow
            - value: 0.95
              color: red

  - title: "Causal Violations"
    type: timeseries
    targets:
      - expr: "rate(asi_sequencer_causal_violations_total[1m]) by (source_module)"
        legendFormat: "{{source_module}}"
    fieldConfig:
      defaults:
        unit: "ops"
        color:
          mode: "fixed"
          fixedColor: "orange"

---

Phase 17 — Temporal Reasoning & Predictive Cognition | Issue: #437 | Show & Tell: #438