Operator observability: persist reaped agent logs, expose transcripts, fix service-log filters, demote poll spam

docs/architecture/orchestrator.md

@@ -688,7 +688,7 @@ NetworkPolicies (enforced by Cilium CNI):
 - `GET /health` - MCP server health check
 - `POST /mcp` - Streamable HTTP transport endpoint (MCP protocol via JSON-RPC)
 
-Available MCP tools (orchestrator-backed): `submit_task`, `get_status`, `provide_input`, `answer_feedback`, `list_tasks`, `cancel_task`, `check_health`, `list_containers`, `get_container_logs`, `send_message`, `get_consensus_status`, `get_phase`, `get_pipeline_snapshot`, `get_contract`, `validate_config`, `update_pipeline_config`, `restart_agent`, `restart_phase`, `list_agent_local_commits`, `salvage_agent_commits`, `advance_phase`, `start_pipeline`, `start_phase`, `complete_phase`, `populate_contract`, `get_deployment_context`, `validate_deployment_manifests`, `prune_stale_worktrees`, `validate_network_isolation`, `rebuild_and_rollout`, `get_service_logs`
+Available MCP tools (orchestrator-backed): `submit_task`, `get_status`, `provide_input`, `answer_feedback`, `list_tasks`, `cancel_task`, `check_health`, `list_containers`, `get_container_logs`, `get_agent_transcript`, `send_message`, `get_consensus_status`, `get_phase`, `get_pipeline_snapshot`, `get_contract`, `validate_config`, `update_pipeline_config`, `restart_agent`, `restart_phase`, `list_agent_local_commits`, `salvage_agent_commits`, `advance_phase`, `start_pipeline`, `start_phase`, `complete_phase`, `populate_contract`, `get_deployment_context`, `validate_deployment_manifests`, `prune_stale_worktrees`, `validate_network_isolation`, `rebuild_and_rollout`, `get_service_logs`
 
 Blocking host-side waits run via the `egg-orch pipeline wait-status` Bash CLI rather than an MCP tool (issue [#2211](https://github.com/jwbron/egg/issues/2211)). The CLI loops the orchestrator's `/api/v1/pipelines/<id>/status/wait` route server-side and emits one JSON-line per pipeline-relevant event. See [Host-Side Waits](../reference/agent-wait-patterns.md#7-host-side-waits--egg-orch-pipeline-wait-status) for the envelope, exit-code contract, and cursor protocol. The route itself stays — the CLI is a wrapper.
 

docs/reference/mcp-deployment-tools.md

@@ -524,6 +524,12 @@ filtering (`pipeline_id`, `level`, `pattern`) applied **before** truncation,
 so a targeted query returns the relevant lines instead of a raw tail that is
 mostly health-check noise.
 
+[#3547](https://github.com/jwbron/egg/issues/3547) made the filters work on
+the console-formatted lines production pods actually emit (the JSON-only
+`pipeline_id` matching returned empty against real logs) and grouped physical
+lines into logical records so multi-line tracebacks stay attached to the log
+line that raised them.
+
 **HTTP route**: `GET /api/v1/deployment/logs`
 
 **Input schema**:
@@ -552,20 +558,28 @@ mostly health-check noise.
   HH:MM." When filters are active this bounds the per-pod scan window;
   the backing fetch is widened to 10 000 lines so the filter has
   material to match.
-- `pipeline_id` (optional) — keep only lines emitted for this
-  pipeline/task id. Checked against `context.task_id`,
-  `extra.pipeline_id`, and `extra.task_id` (in that order) — production
-  call sites use `pipeline_id=...`, which the `JsonFormatter` lands in
-  `extra` rather than the context-allowlisted `task_id` slot.
+- `pipeline_id` (optional); keep only records emitted for this
+  pipeline/task id. For JSON-formatted records this checks
+  `context.task_id`, `extra.pipeline_id`, and `extra.task_id` (in that
+  order); production call sites use `pipeline_id=...`, which the
+  `JsonFormatter` lands in `extra` rather than the context-allowlisted
+  `task_id` slot. For console-formatted records (what the k8s pods emit -
+  `JsonFormatter` only activates when the environment detects as GCP) it
+  matches the inline `pipeline_id=` / `task_id=` key=value pair (#3547).
 - `level` (optional) — minimum severity (`DEBUG` | `INFO` | `WARNING` |
-  `ERROR` | `CRITICAL`); drops lower-severity and unstructured lines.
+  `ERROR` | `CRITICAL`); reads the JSON `severity` field or the console
+  `[LEVEL]` bracket, and drops records with no determinable severity.
   Case-sensitive on the HTTP route (matches the MCP schema enum).
 - `pattern` (optional) — Python regex applied via `re.search`; a plain
-  substring is a valid pattern. Matches against the raw line, so it can
-  keep non-JSON / unstructured lines too (unlike `level`, which drops
-  them); combine with `level` if you want JSON-only matches.
-
-Filters are ANDed: a line must pass every active filter to be kept.
+  substring is a valid pattern. Matches against the record's full raw
+  text, continuation lines included, so a pattern that matches an
+  exception message returns the whole traceback.
+
+Filters are ANDed and operate on **logical records**, not physical lines
+(#3547): a record starts at a JSON object line or a console timestamp
+head, and everything else (traceback frames, wrapped payloads) attaches
+to the preceding record. With filters active, `lines` caps the matching
+records returned.
 
 **Output shape** (no filters active):
 

orchestrator/agent_log_store.py

@@ -0,0 +1,200 @@
+"""Redis-backed store for one-shot agent pod logs captured at removal (#3547).
+
+In the event-driven model most agent runs last seconds to minutes, and the
+backing Job is reaped moments after the loop observes its exit (the #3181
+observe-once sweep, the #3337 superseded-sibling teardown, pipeline cleanup)
+or by the Job's own ``ttlSecondsAfterFinished``. Once the pod is gone,
+``get_container_logs`` returns 404 and the stdout evidence of *why* an agent
+exited is unrecoverable; during incident response operators had to race a
+respawn to pull logs while the pod was briefly live.
+
+This store is the durable home for that evidence: ``remove_agent_job``
+best-effort snapshots the pod's log tail (plus its identifying labels and
+exit code) here before deleting the Job, and the container-logs read path
+falls back to it when the live pod is gone. Records are keyed
+``agent-logs:{pipeline_id}:{job_name}`` and reaped by TTL.
+
+Mirrors ``session_state_store``: same Redis, same best-effort contract -
+every failure logs and degrades (capture skipped, fallback misses) rather
+than raising into the removal or request path.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import UTC, datetime
+from typing import Any
+
+logger = logging.getLogger("orchestrator.agent_log_store")
+
+__all__ = [
+    "AGENT_LOG_TTL_SECONDS",
+    "MAX_LOG_BYTES",
+    "AgentLogStore",
+    "get_agent_log_store",
+    "reset_agent_log_store",
+    "set_agent_log_store",
+]
+
+_KEY_PREFIX = "agent-logs"
+
+# TTL on each captured log. Sized for incident response; long enough that an
+# operator diagnosing a stall hours later still has the evidence, short enough
+# that Redis never accumulates a pipeline's whole history.
+AGENT_LOG_TTL_SECONDS = 24 * 60 * 60  # 24 hours
+
+# Cap on a stored log. A one-shot agent run's stdout is small (its transcript
+# lives in the session-state store); when a pathological pod exceeds this, the
+# *tail* is kept; the exit evidence is at the end.
+MAX_LOG_BYTES = 1 * 1024 * 1024  # 1 MiB
+
+
+class AgentLogStore:
+    """Redis-backed CRUD over per-(pipeline, job) captured-log records.
+
+    Constructed with an injected redis client (real in production via
+    :func:`get_agent_log_store`, ``fakeredis.FakeRedis()`` in tests). Stores
+    bytes (``decode_responses=False``) and owns its own JSON (de)serialisation.
+    Every method is best-effort: a Redis/serialisation failure logs and returns
+    the miss sentinel (``None`` / ``False`` / ``[]``); it never raises into
+    the caller's removal or request path.
+    """
+
+    def __init__(self, redis_client: Any) -> None:
+        self._redis = redis_client
+
+    @staticmethod
+    def _key(pipeline_id: str, job_name: str) -> str:
+        return f"{_KEY_PREFIX}:{pipeline_id}:{job_name}"
+
+    def put(
+        self,
+        pipeline_id: str,
+        job_name: str,
+        *,
+        logs: str,
+        agent_role: str | None = None,
+        slice_id: str | None = None,
+        exit_code: int | None = None,
+        captured_at: str | None = None,
+    ) -> bool:
+        """Persist (overwrite) a captured log under a fresh TTL; return whether stored.
+
+        An oversized log is tail-truncated to :data:`MAX_LOG_BYTES`; the exit
+        evidence lives at the end; rather than dropped.
+        """
+        if not pipeline_id or not job_name:
+            return False
+        encoded = logs.encode("utf-8")
+        truncated = False
+        if len(encoded) > MAX_LOG_BYTES:
+            logs = encoded[-MAX_LOG_BYTES:].decode("utf-8", errors="replace")
+            truncated = True
+        record = {
+            "job_name": job_name,
+            "agent_role": agent_role,
+            "slice_id": slice_id,
+            "exit_code": exit_code,
+            "captured_at": captured_at or datetime.now(UTC).isoformat(),
+            "truncated": truncated,
+            "logs": logs,
+        }
+        try:
+            payload = json.dumps(record).encode("utf-8")
+            self._redis.setex(self._key(pipeline_id, job_name), AGENT_LOG_TTL_SECONDS, payload)
+            return True
+        except Exception as exc:  # noqa: BLE001; best-effort; never block removal
+            logger.warning(
+                "Failed to persist agent logs (pipeline=%s job=%s): %s",
+                pipeline_id,
+                job_name,
+                exc,
+            )
+            return False
+
+    def get(self, pipeline_id: str, job_name: str) -> dict[str, Any] | None:
+        """Read a captured-log record, or ``None``; never raising."""
+        try:
+            raw = self._redis.get(self._key(pipeline_id, job_name))
+        except Exception as exc:  # noqa: BLE001; best-effort read
+            logger.warning(
+                "Failed to read agent logs (pipeline=%s job=%s): %s",
+                pipeline_id,
+                job_name,
+                exc,
+            )
+            return None
+        if not raw:
+            return None
+        try:
+            data = json.loads(raw)
+        except ValueError, TypeError:
+            logger.warning(
+                "Malformed agent-log payload (pipeline=%s job=%s); ignoring",
+                pipeline_id,
+                job_name,
+            )
+            return None
+        return data if isinstance(data, dict) else None
+
+    def list_records(self, pipeline_id: str, *, include_logs: bool = False) -> list[dict[str, Any]]:
+        """Enumerate the pipeline's captured logs, newest first.
+
+        Metadata only by default (``logs`` replaced with ``log_bytes``) so the
+        index stays cheap to return over MCP; ``include_logs=True`` keeps the
+        bodies for callers that want the newest record in one pass.
+        """
+        try:
+            keys = list(self._redis.scan_iter(match=f"{_KEY_PREFIX}:{pipeline_id}:*".encode()))
+        except Exception as exc:  # noqa: BLE001; best-effort index
+            logger.warning("Failed to scan agent-log keys (pipeline=%s): %s", pipeline_id, exc)
+            return []
+        entries: list[dict[str, Any]] = []
+        for key in keys:
+            key_str = key.decode("utf-8", errors="replace") if isinstance(key, bytes) else key
+            job_name = key_str.rsplit(":", 1)[-1]
+            record = self.get(pipeline_id, job_name)
+            if record is None:
+                continue
+            if not include_logs:
+                logs = record.pop("logs", "") or ""
+                record["log_bytes"] = len(logs.encode("utf-8"))
+            entries.append(record)
+        entries.sort(key=lambda r: r.get("captured_at") or "", reverse=True)
+        return entries
+
+
+_store: AgentLogStore | None = None
+
+
+def get_agent_log_store() -> AgentLogStore:
+    """Return the process-wide store, building a Redis client on first use.
+
+    Reads ``REDIS_HOST`` / ``REDIS_PORT`` / ``REDIS_DB`` exactly like
+    ``session_state_store.get_session_state_store`` so captured logs land on
+    the same Redis the rest of the orchestrator uses.
+    """
+    global _store
+    if _store is None:
+        import redis
+
+        host = os.environ.get("REDIS_HOST", "localhost")
+        port = int(os.environ.get("REDIS_PORT", "6379"))
+        db = int(os.environ.get("REDIS_DB", "0"))
+        client = redis.Redis(host=host, port=port, db=db, decode_responses=False)
+        _store = AgentLogStore(client)
+    return _store
+
+
+def reset_agent_log_store() -> None:
+    """Reset the process-wide store (tests inject a fakeredis-backed store)."""
+    global _store
+    _store = None
+
+
+def set_agent_log_store(store: AgentLogStore | None) -> None:
+    """Install a store instance (tests inject ``AgentLogStore(fakeredis.FakeRedis())``)."""
+    global _store
+    _store = store

orchestrator/concurrent_executor.py

@@ -295,6 +295,12 @@ def __init__(
         # #3064 slice-2: set when orchestrator-ownership mode starts the
         # event loop in ``spawn_all``; ``None`` in pod mode.
         self._event_loop: Any | None = None
+        # #3547: the run loop calls ``check_consensus`` every ~5s, so the
+        # "consensus incomplete" observations repeat verbatim for minutes at
+        # a time and define the INFO noise floor for the whole service. This
+        # holds the last-logged incomplete state; the lines log at INFO only
+        # when it changes and at DEBUG otherwise.
+        self._last_incomplete_consensus_log: tuple[Any, ...] | None = None
 
     def _get_review_graph(self) -> ReviewGraph:
         """Get the review graph, using the override if provided."""
@@ -1424,7 +1430,20 @@ def check_consensus(self) -> dict[str, Any]:
             if not result.get("is_complete"):
                 all_roles = tracker.graph.all_roles()
                 confirmed_in_tracker = len(all_roles) - len(result.get("blocking_agents", []))
-                logger.info(
+                # #3547: this branch runs on every ~5s poll tick, so at INFO
+                # these lines drown the service log (2 lines x N slices every
+                # tick). Log at INFO only when the incomplete state actually
+                # changes; the unchanged repeats drop to DEBUG.
+                incomplete_state = (
+                    confirmed_in_tracker,
+                    len(all_roles),
+                    tuple(sorted(result.get("blocking_agents", []))),
+                    bool(result.get("has_unresolved_nacks", False)),
+                )
+                state_changed = incomplete_state != self._last_incomplete_consensus_log
+                self._last_incomplete_consensus_log = incomplete_state
+                log_incomplete = logger.info if state_changed else logger.debug
+                log_incomplete(
                     "Consensus incomplete — checking fallbacks",
                     pipeline_id=self.pipeline.id,
                     confirmed=confirmed_in_tracker,
@@ -1462,7 +1481,7 @@ def check_consensus(self) -> dict[str, Any]:
                     # an empty fresh tracker correctly returns
                     # is_complete=False here so the pipeline run loop keeps
                     # polling.
-                    logger.info(
+                    log_incomplete(
                         "Skipping pipeline-wide message-bus fallback for slice-scoped tracker",
                         pipeline_id=self.pipeline.id,
                         slice_id=self._slice_id,
@@ -1501,7 +1520,7 @@ def check_consensus(self) -> dict[str, Any]:
                             result["fallback"] = "message_bus"
                         else:
                             missing = all_roles - confirmed_roles if all_roles else set()
-                            logger.info(
+                            log_incomplete(
                                 "Message-bus fallback: not all roles confirmed",
                                 pipeline_id=self.pipeline.id,
                                 confirmed_roles=sorted(confirmed_roles),
@@ -1514,6 +1533,10 @@ def check_consensus(self) -> dict[str, Any]:
                             pipeline_id=self.pipeline.id,
                             error=str(e),
                         )
+            if result.get("is_complete"):
+                # Consensus reached (possibly via a fallback override): reset
+                # so the next incomplete round logs its first tick at INFO.
+                self._last_incomplete_consensus_log = None
             return result
         return {"is_complete": False, "blocking_agents": [], "has_objections": False, "agents": {}}
 

orchestrator/kubernetes_client.py

@@ -452,6 +452,43 @@ def remove_container(
         except Exception as exc:
             raise JobOperationError(f"Failed to remove job {job_name}: {exc}") from exc
 
+    def read_job_log_snapshot(self, container_id: str, tail_lines: int = 2000) -> dict | None:
+        """Best-effort snapshot of a Job's pod logs plus identity labels (#3547).
+
+        Called by ``KubernetesSpawner.remove_agent_job`` immediately before Job
+        deletion so a one-shot agent's stdout survives the reap (persisted via
+        ``agent_log_store``). Returns the log tail together with the pod's
+        pipeline/role/slice labels and terminated exit code, or ``None`` on any
+        failure; capture must never block removal.
+        """
+        try:
+            job_name = self._resolve_job_name(container_id)
+            pod_name = self.get_pod_for_job(job_name, self.namespace)
+            pod = self.core_api.read_namespaced_pod(pod_name, self.namespace)
+            labels = (pod.metadata.labels or {}) if pod.metadata else {}
+            exit_code: int | None = None
+            if pod.status and pod.status.container_statuses:
+                cs = pod.status.container_statuses[0]
+                if cs.state and cs.state.terminated:
+                    exit_code = cs.state.terminated.exit_code
+            logs = self.get_pod_logs(pod_name, self.namespace, tail_lines=tail_lines)
+        except Exception as exc:
+            logger.debug(
+                "Job log snapshot unavailable",
+                container_id=container_id,
+                error=str(exc),
+            )
+            return None
+        return {
+            "job_name": job_name,
+            "pod_name": pod_name,
+            "pipeline_id": labels.get(LABEL_PIPELINE_ID),
+            "agent_role": labels.get(LABEL_AGENT_ROLE),
+            "slice_id": labels.get(LABEL_SLICE_ID),
+            "exit_code": exit_code,
+            "logs": logs,
+        }
+
     def get_container_info(self, container_id: str) -> ContainerInfo:
         """Get information about a Job's pod."""
         job_name = self._resolve_job_name(container_id)

orchestrator/kubernetes_spawner/__init__.py

@@ -522,6 +522,10 @@ def get_kubernetes_spawner(
 KubernetesSpawner.spawn_event_job = _events.spawn_event_job
 KubernetesSpawner.stop_agent_job = _jobs.stop_agent_job
 KubernetesSpawner.remove_agent_job = _jobs.remove_agent_job
+# Module-level (not a class method) so ``remove_agent_job`` reaches it via the
+# barrel; patchable as ``kubernetes_spawner._persist_job_logs_best_effort``
+# (#3547).
+_persist_job_logs_best_effort = _jobs._persist_job_logs_best_effort
 KubernetesSpawner.list_pipeline_jobs = _jobs.list_pipeline_jobs
 KubernetesSpawner.list_slice_jobs = _jobs.list_slice_jobs
 KubernetesSpawner.cleanup_pipeline = _jobs.cleanup_pipeline