feat(agent): correlate logs by cycle id across the whole call tree

mcp-server/src/streamops_mcp/agent/main.py

@@ -17,11 +17,9 @@
 
 from streamops_mcp.agent.monitor import MonitorAgent
 from streamops_mcp.config import config
+from streamops_mcp.logging_setup import configure_logging
 
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
-)
+configure_logging()
 logger = logging.getLogger("streamops-mcp.agent")
 
 

mcp-server/src/streamops_mcp/agent/monitor.py

@@ -34,6 +34,11 @@
 )
 from streamops_mcp.agent.tools import ALL_TOOLS, DIAGNOSTIC_TOOLS
 from streamops_mcp.config import config
+from streamops_mcp.logging_setup import (
+    new_correlation_id,
+    reset_correlation_id,
+    set_correlation_id,
+)
 from streamops_mcp.prompts import load_prompt, load_runbook
 
 logger = logging.getLogger("streamops-mcp.monitor")
@@ -102,69 +107,81 @@ async def run_cycle(self) -> IncidentReport | None:
         """Run one monitoring cycle: poll, detect, diagnose, report.
 
         Returns an IncidentReport if an anomaly was found, None if healthy.
-        """
-        logger.info("Starting monitoring cycle (multi_agent=%s)", self.multi_agent)
-
-        detection = await self._detect_anomalies()
-        if detection is None:
-            logger.info("No anomalies detected, infrastructure healthy")
-            return None
-
-        if self.multi_agent:
-            try:
-                diagnosis = await self._retry_subagent(
-                    "Diagnostic Agent",
-                    lambda: self._spawn_diagnostic_agent(detection),
-                )
-            except Exception as exc:
-                logger.error(
-                    "Diagnostic Agent failed after retries, cycle aborted: %s", exc,
-                )
-                return None
 
-            self._log_confidence_distribution(diagnosis)
+        A correlation id is bound for the whole cycle so every log line the
+        cycle emits (this coordinator, both sub-agents, the tool executor, and
+        escalation) shares one greppable id. See issue #84.
+        """
+        cycle_id = new_correlation_id()
+        token = set_correlation_id(cycle_id)
+        try:
+            logger.info(
+                "Starting monitoring cycle (cycle_id=%s, multi_agent=%s)",
+                cycle_id, self.multi_agent,
+            )
 
-            if self._all_claims_low_confidence(diagnosis):
-                logger.warning(
-                    "All %d claims are LOW or UNSOURCED confidence; "
-                    "downgrading to warning-level log, skipping report agent",
-                    len(diagnosis.claims),
-                )
+            detection = await self._detect_anomalies()
+            if detection is None:
+                logger.info("No anomalies detected, infrastructure healthy")
                 return None
 
-            low_claims = self._extract_low_confidence_claims(diagnosis)
+            if self.multi_agent:
+                try:
+                    diagnosis = await self._retry_subagent(
+                        "Diagnostic Agent",
+                        lambda: self._spawn_diagnostic_agent(detection),
+                    )
+                except Exception as exc:
+                    logger.error(
+                        "Diagnostic Agent failed after retries, cycle aborted: %s", exc,
+                    )
+                    return None
+
+                self._log_confidence_distribution(diagnosis)
+
+                if self._all_claims_low_confidence(diagnosis):
+                    logger.warning(
+                        "All %d claims are LOW or UNSOURCED confidence; "
+                        "downgrading to warning-level log, skipping report agent",
+                        len(diagnosis.claims),
+                    )
+                    return None
+
+                low_claims = self._extract_low_confidence_claims(diagnosis)
+
+                try:
+                    report = await self._retry_subagent(
+                        "Report Agent",
+                        lambda: self._spawn_report_agent(diagnosis),
+                    )
+                except Exception as exc:
+                    logger.warning(
+                        "Report Agent failed after retries, producing fallback report: %s", exc,
+                    )
+                    report = self._fallback_report(diagnosis)
+
+                if low_claims and not report.low_confidence_claims:
+                    report.low_confidence_claims = low_claims
+            else:
+                diagnosis = self._extract_diagnosis_from_detection(detection)
+                report = await self._spawn_report_agent(diagnosis)
 
-            try:
-                report = await self._retry_subagent(
-                    "Report Agent",
-                    lambda: self._spawn_report_agent(diagnosis),
-                )
-            except Exception as exc:
+            if diagnosis.conflicts:
                 logger.warning(
-                    "Report Agent failed after retries, producing fallback report: %s", exc,
+                    "Coordinator received %d unresolved conflict(s) from Diagnostic Agent",
+                    len(diagnosis.conflicts),
                 )
-                report = self._fallback_report(diagnosis)
-
-            if low_claims and not report.low_confidence_claims:
-                report.low_confidence_claims = low_claims
-        else:
-            diagnosis = self._extract_diagnosis_from_detection(detection)
-            report = await self._spawn_report_agent(diagnosis)
-
-        if diagnosis.conflicts:
-            logger.warning(
-                "Coordinator received %d unresolved conflict(s) from Diagnostic Agent",
-                len(diagnosis.conflicts),
-            )
-            for conflict in diagnosis.conflicts:
-                logger.warning(
-                    "Conflict %s [%s]: claims %s vs %s",
-                    conflict.conflict_id, conflict.topic,
-                    conflict.claim_a_id, conflict.claim_b_id,
-                )
-
-        await escalate(report, diagnosis=diagnosis)
-        return report
+                for conflict in diagnosis.conflicts:
+                    logger.warning(
+                        "Conflict %s [%s]: claims %s vs %s",
+                        conflict.conflict_id, conflict.topic,
+                        conflict.claim_a_id, conflict.claim_b_id,
+                    )
+
+            await escalate(report, diagnosis=diagnosis)
+            return report
+        finally:
+            reset_correlation_id(token)
 
     @staticmethod
     def _fallback_report(diagnosis: DiagnosisReport) -> IncidentReport:

mcp-server/src/streamops_mcp/logging_setup.py

@@ -0,0 +1,84 @@
+"""Correlated logging setup.
+
+Binds a per-cycle correlation id to a ContextVar so every log line emitted
+anywhere in a monitoring cycle's async call tree (monitor, diagnostic and
+report sub-agents, tool executor, escalation) carries the same id, without
+threading it through every function signature. ContextVars propagate across
+``await`` within the same asyncio task, so setting the id once at the top of
+``run_cycle`` covers the whole cycle.
+
+A log line with no correlation id is orphaned the moment two cycles overlap
+or two incidents run concurrently: the lines interleave with no way to
+reconstruct one incident's story. See issue #84.
+"""
+
+import contextvars
+import logging
+import uuid
+
+# Default sentinel for lines emitted outside any cycle (startup, config, the
+# MCP server's own request handling). Never left unset, so the formatter's
+# %(correlation_id)s always resolves.
+_UNSET = "-"
+
+_correlation_id: contextvars.ContextVar[str] = contextvars.ContextVar(
+    "correlation_id", default=_UNSET
+)
+
+# Includes the correlation id so a single cycle's lines are greppable by cid=.
+LOG_FORMAT = "%(asctime)s [%(name)s] [cid=%(correlation_id)s] %(levelname)s %(message)s"
+
+
+def new_correlation_id(prefix: str = "cyc") -> str:
+    """Generate a correlation id that is unique across process restarts.
+
+    A monotonic per-process counter would collide across restarts (two cold
+    starts both produce cycle 1); the uuid suffix makes each cycle's id
+    globally distinct.
+    """
+    return f"{prefix}-{uuid.uuid4().hex[:12]}"
+
+
+def set_correlation_id(cid: str) -> contextvars.Token:
+    """Bind a correlation id to the current context; returns a reset token."""
+    return _correlation_id.set(cid)
+
+
+def reset_correlation_id(token: contextvars.Token) -> None:
+    """Restore the correlation id to its previous value using the token."""
+    _correlation_id.reset(token)
+
+
+def get_correlation_id() -> str:
+    """Return the correlation id bound to the current context (or the sentinel)."""
+    return _correlation_id.get()
+
+
+class CorrelationIdFilter(logging.Filter):
+    """Injects the current correlation id onto every LogRecord.
+
+    Attached to the handler so the attribute is always present before the
+    formatter runs, including for third-party loggers that never heard of the
+    correlation id.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        record.correlation_id = _correlation_id.get()
+        return True
+
+
+def configure_logging(level: int = logging.INFO) -> None:
+    """Configure root logging with the correlation-id format and filter.
+
+    Idempotent: replaces the root handlers so calling it from multiple entry
+    points (agent main, MCP server) does not double-log or drop the filter.
+    """
+    handler = logging.StreamHandler()
+    handler.setFormatter(logging.Formatter(LOG_FORMAT))
+    handler.addFilter(CorrelationIdFilter())
+
+    root = logging.getLogger()
+    root.setLevel(level)
+    for existing in list(root.handlers):
+        root.removeHandler(existing)
+    root.addHandler(handler)

mcp-server/src/streamops_mcp/server.py

@@ -13,15 +13,13 @@
 
 from mcp.server.fastmcp import FastMCP
 
+from streamops_mcp.logging_setup import configure_logging
 from streamops_mcp.tools.events import register_event_tools
 from streamops_mcp.tools.flink import register_flink_tools
 from streamops_mcp.tools.kafka import register_kafka_tools
 from streamops_mcp.tools.prometheus import register_prometheus_tools
 
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
-)
+configure_logging()
 logger = logging.getLogger("streamops-mcp")
 
 mcp = FastMCP(