fix drain-chain hierarchy for barrier context; convert overlap counter to tracewarning

src/aiu_trace_analyzer/pipeline/barrier.py

@@ -42,10 +42,14 @@ def collection_phase(self) -> bool:
 
     def drain(self) -> list[TraceEvent]:
         if self.phase == self._COLLECTION_PHASE:
+            # first drain call: switch to the application phase. Defer to the application-phase
+            # drain so parent drain is not called twice and cross-phase state in self.queues
+            # survives the transition.
             self.phase = self._APPLICATION_PHASE
+            revents = []
         else:
-            # do nothing if this is the application phase
-            pass
-        # the queues for these contexts don't contain events (events are held in barrier context),
-        # so nothing to drain here
-        return []
+            # application phase (final drain call): the cross-phase state has been consumed, so
+            # it is safe to chain the parent drain, which flushes the (event-less) queues and
+            # emits any additional events if needed.
+            revents = super().drain()
+        return revents

src/aiu_trace_analyzer/pipeline/hashqueue.py

@@ -26,7 +26,8 @@ def drain(self) -> list[TraceEvent]:
             item = self.queues.popitem()
             if isinstance(item, TraceEvent):
                 revents += item
-        return revents
+        # chain to the base drain anything left there is emitted
+        return revents + super().drain()
 
     def insert(self, event: TraceEvent, queue_id=None) -> int:
         '''

src/aiu_trace_analyzer/pipeline/overlap.py

@@ -4,7 +4,7 @@
 
 import aiu_trace_analyzer.logger as aiulog
 from aiu_trace_analyzer.pipeline import AbstractContext, AbstractHashQueueContext, TwoPhaseWithBarrierContext
-from aiu_trace_analyzer.types import TraceEvent, GlobalIngestData
+from aiu_trace_analyzer.types import TraceEvent, GlobalIngestData, TraceWarning
 from aiu_trace_analyzer.pipeline.tools import PipelineContextTool
 
 
@@ -44,21 +44,20 @@ def __init__(self,
                  ts_shift_threshold=0.0,
                  max_tid_streams=5,
                  ) -> None:
-        super().__init__()
+        super().__init__(warnings=[
+            TraceWarning(
+                name="overlaps",
+                text="Partial-overlap slices resolved: {d[count]}",
+                data={"count": 0},
+            )
+        ])
         self.overlap_resolve = overlap_resolve
-        self.resolved = 0
         self.async_id = 0
         self.async_queues = {}
         self.ts_shift_threshold = ts_shift_threshold
         self.tid_space = {}
         self.max_tid_streams = max_tid_streams
 
-    def __del__(self) -> None:
-        if not self.is_enabled():
-            return
-        level = aiulog.WARN if self.resolved else aiulog.INFO
-        aiulog.log(level, "Partial-overlap slices resolved:", self.resolved)
-
     # search for events within the same pid/tid
     # accumulate a queue of events for each pid/tid
     # once the queue is full, run detection and emit events that are fine
@@ -145,11 +144,11 @@ def handle_overlap(self,
                        queue_id: int) -> list[TraceEvent]:
         if self.overlap_resolve == self.OVERLAP_RESOLVE_DROP:
             aiulog.log(aiulog.WARN, "Solving overlap conflict by dropping:", oevent)
-            self.resolved += 1
+            self.issue_warning("overlaps")
             return []
         elif self.overlap_resolve == self.OVERLAP_RESOLVE_WARN:
             aiulog.log(aiulog.WARN, "Detected overlap conflict: ", oevent["name"])
-            self.resolved += 1
+            self.issue_warning("overlaps")
             return [oevent]
         elif self.overlap_resolve == self.OVERLAP_RESOLVE_SHIFT:
             ts_shift = self.get_overlap_time(oevent["ts"], oevent["ts"]+oevent["dur"], self.queues[queue_id])
@@ -174,21 +173,21 @@ def handle_overlap(self,
                            "us: increase threshold or use different overlap res option.")
                 rlist = [oevent]
 
-            self.resolved += 1
+            self.issue_warning("overlaps")
             return rlist
         elif self.overlap_resolve == self.OVERLAP_RESOLVE_TID:
             oevent["tid"] = self.find_next_tid(oevent)
             # feed offending event back into the detector with the new TID to make sure
             # there are no collisions there either
             rlist = self.overlap_detection(oevent)
-            self.resolved += 1
+            self.issue_warning("overlaps")
             return rlist
         elif self.overlap_resolve == self.OVERLAP_RESOLVE_ASYNC:
             oevent["id"] = self.async_id
             end_ts = oevent["ts"] + oevent["dur"]
             oevent.pop("dur")
             self.async_id += 1
-            self.resolved += 1
+            self.issue_warning("overlaps")
 
             e_event = copy.deepcopy(oevent)
             oevent["ph"] = "b"
@@ -264,9 +263,9 @@ def drain(self):
         if self.overlap_resolve == self.OVERLAP_RESOLVE_TID:
             if self.phase == self._COLLECTION_PHASE:
                 self._collect_and_build_tid_space()
-                return super().drain()
-            else:
-                return []
+            # chain to the parent drain in both phases: it advances the two-phase state and
+            # propagates accumulated warnings as trace_issue events
+            return super().drain()
         else:
             revents = []
             # make sure to drain the queue of async 'e' events that might have been hold
@@ -276,7 +275,11 @@ def drain(self):
                 # make sure to keep everything sorted
                 aq.sort(key=lambda e: e['ts'])
                 revents += aq
-            return revents
+            # these modes don't use the two-phase mechanism and are drained only once, so switch
+            # to the application phase before chaining up to make the parent emit the accumulated
+            # warnings as trace_issue events (rather than deferring to a second drain that never comes)
+            self.phase = self._APPLICATION_PHASE
+            return revents + super().drain()
 
 
 def detect_partial_overlap_tids(event: TraceEvent, context: AbstractContext) -> list[TraceEvent]:

tests/aiu_trace_analyzer/pipeline/test_deactivated_stage_output.py

@@ -12,9 +12,6 @@
 
 
 def test_deactivated_stage_contexts_do_not_emit_output(monkeypatch, capsys):
-    log_calls = []
-    monkeypatch.setattr(aiulog, "log", lambda *args: log_calls.append(args))
-
     contexts = []
 
     dma_context = DataTransferExtractionContext()
@@ -29,9 +26,9 @@ def test_deactivated_stage_contexts_do_not_emit_output(monkeypatch, capsys):
     contexts.append(inverse_context)
 
     overlap_context = OverlapDetectionContext()
-    overlap_context.resolved = 5
+    overlap_context.issue_warning("overlaps")
     overlap_context.disable()
-    contexts.append(overlap_context)
+    assert overlap_context.warnings["overlaps"].has_warning() is False
 
     power_context = PowerExtractionContext()
     power_context.bad_events = 1
@@ -52,6 +49,10 @@ def test_deactivated_stage_contexts_do_not_emit_output(monkeypatch, capsys):
     coll_context.disable()
     contexts.append(coll_context)
 
+    # capture output only around teardown: context construction may legitimately emit debug logs
+    log_calls = []
+    monkeypatch.setattr(aiulog, "log", lambda *args: log_calls.append(args))
+
     for context in contexts:
         type(context).__del__(context)