diff --git a/docs/architecture/orchestrator.md b/docs/architecture/orchestrator.md index 0ffd396601..b576872822 100644 --- a/docs/architecture/orchestrator.md +++ b/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//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. diff --git a/docs/reference/mcp-deployment-tools.md b/docs/reference/mcp-deployment-tools.md index e66833fe7a..d8ec94276e 100644 --- a/docs/reference/mcp-deployment-tools.md +++ b/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): diff --git a/orchestrator/agent_log_store.py b/orchestrator/agent_log_store.py new file mode 100644 index 0000000000..dd73640c0c --- /dev/null +++ b/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 diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index b1ae243dc7..e63e92f748 100644 --- a/orchestrator/concurrent_executor.py +++ b/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": {}} diff --git a/orchestrator/kubernetes_client.py b/orchestrator/kubernetes_client.py index a1cf584cd3..8bc9080af7 100644 --- a/orchestrator/kubernetes_client.py +++ b/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) diff --git a/orchestrator/kubernetes_spawner/__init__.py b/orchestrator/kubernetes_spawner/__init__.py index c5bf099e40..e24c1b5ecb 100644 --- a/orchestrator/kubernetes_spawner/__init__.py +++ b/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 diff --git a/orchestrator/kubernetes_spawner/_jobs.py b/orchestrator/kubernetes_spawner/_jobs.py index a7ae3fa4f9..52a990255c 100644 --- a/orchestrator/kubernetes_spawner/_jobs.py +++ b/orchestrator/kubernetes_spawner/_jobs.py @@ -59,6 +59,43 @@ def stop_agent_job( raise +def _persist_job_logs_best_effort(self, job_name: str) -> None: + """Snapshot the Job's pod logs into the agent-log store before removal (#3547). + + One-shot event Jobs are reaped moments after their exit is observed, so + this pre-removal capture is the only durable copy of the agent's stdout; + ``get_container_logs`` falls back to it once the pod is gone. Strictly + best-effort: any failure (pod already GC'd, store unavailable, stub k8s + client in tests) logs and returns so removal is never blocked. + """ + try: + snapshot = self.k8s.read_job_log_snapshot(job_name) + if not snapshot: + return + pipeline_id = snapshot.get("pipeline_id") + logs = snapshot.get("logs") + if not pipeline_id or logs is None: + # Without a pipeline label there is no operator-facing key to + # file the capture under; without logs there is nothing to keep. + return + from agent_log_store import get_agent_log_store + + get_agent_log_store().put( + pipeline_id, + snapshot.get("job_name") or job_name, + logs=logs, + agent_role=snapshot.get("agent_role"), + slice_id=snapshot.get("slice_id"), + exit_code=snapshot.get("exit_code"), + ) + except Exception as exc: # noqa: BLE001; capture must never block removal + logger.warning( + "Failed to persist agent logs before Job removal", + job_name=job_name, + error=str(exc), + ) + + def remove_agent_job( self, job_name: str, @@ -72,6 +109,7 @@ def remove_agent_job( force: Force removal (foreground propagation) cleanup_session: Whether to delete gateway session """ + _pkg._persist_job_logs_best_effort(self, job_name) try: self.k8s.remove_container(job_name, force=force) finally: diff --git a/orchestrator/log_filter.py b/orchestrator/log_filter.py index fa1443f2dc..4ec9edd890 100644 --- a/orchestrator/log_filter.py +++ b/orchestrator/log_filter.py @@ -1,22 +1,32 @@ -"""Server-side filtering for the ``get_service_logs`` tool (#3032). - -The gateway/orchestrator pods emit one structured JSON log object per line -(see ``shared/egg_logging/formatters.py``): severity lives in the ``severity`` -field (GCP strings ``DEBUG``/``INFO``/``WARNING``/``ERROR``/``CRITICAL``) and -the pipeline id is carried in one of three places depending on how the call -site spelled it. The ``JsonFormatter`` only allowlists ``task_id`` / -``repository`` / ``pr_number`` into the nested ``context`` block; any other -kwarg — including ``pipeline_id``, which is the spelling 25+ orchestrator -call sites actually use — lands in the ``extra`` block instead. The -pipeline-id filter therefore checks ``context.task_id`` **and** -``extra.pipeline_id`` **and** ``extra.task_id`` (in that order); matching -any one of them keeps the line. - -``filter_log_lines`` lets an operator scope a noisy multi-pipeline pod tail to -the lines they actually want ("WARNING+ for pipeline X in the last 5 min") -*before* the response is truncated to the MCP token cap, instead of fetching a -raw tail that is mostly health-check noise and watching the one line they need -scroll out of the window they can afford to fetch. +"""Server-side filtering for the ``get_service_logs`` tool (#3032, #3547). + +The gateway/orchestrator pods emit one log record per *event*, in one of the +two formats ``shared/egg_logging/formatters.py`` can produce: + +* **JSON** (``JsonFormatter``, only when the environment detects as GCP): + one JSON object per line; severity in the ``severity`` field and the + pipeline id in ``context.task_id`` / ``extra.pipeline_id`` / + ``extra.task_id`` (the formatter's context allowlist only includes + ``task_id``; every other kwarg; including ``pipeline_id``, the spelling + 25+ orchestrator call sites use; lands in ``extra``). +* **Console** (``ConsoleFormatter``, everywhere else; including the k8s + pods this endpoint actually tails, which detect as ``container``): + ``YYYY-MM-DD HH:MM:SS [LEVEL ] service: message key=value ...`` with + structured kwargs rendered inline as ``key=value`` pairs, and exception + tracebacks appended as real newlines below the record line. + +Filtering therefore works on **logical records, not physical lines**: a new +record starts at a line that looks like a record head (JSON object or the +console timestamp prefix); anything else; traceback frames, multi-line +payloads; is a continuation attached to the preceding record. A ``pattern`` +that matches an exception message returns the whole traceback with it +instead of one orphaned frame (#3547). + +``filter_log_lines`` lets an operator scope a noisy multi-pipeline pod tail +to the records they actually want ("WARNING+ for pipeline X in the last +5 min") *before* the response is truncated to the MCP token cap, instead of +fetching a raw tail that is mostly health-check noise and watching the one +line they need scroll out of the window they can afford to fetch. """ from __future__ import annotations @@ -34,6 +44,31 @@ "CRITICAL": 50, } +# Head of a ConsoleFormatter record: ``YYYY-MM-DD HH:MM:SS [LEVEL``. The level +# group feeds the min_level filter; padding inside the brackets is optional so +# both the padded (``[INFO ]``) and unpadded forms match. Head/field matching +# runs on an ANSI-stripped copy (see ``_strip_ansi``), so the bare form here also +# covers the colorized ``[\x1b[32mINFO \x1b[0m]`` the formatter emits on a TTY. +_CONSOLE_HEAD_RE = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \[([A-Za-z]+)\s*\]") + +# Inline ``pipeline_id=...`` / ``task_id=...`` pair as ConsoleFormatter renders +# it. Ids are slugs (``issue-3523``), never quoted, so a bare token capture is +# exact; the lookbehind stops ``sub_task_id=`` from matching as ``task_id=``. +_CONSOLE_ID_RE = re.compile(r"(? str: + """Remove ANSI SGR escapes so head/field regexes see the plain record text.""" + return _ANSI_SGR_RE.sub("", text) + def severity_rank(name: str | None) -> int | None: """Return the numeric rank for a GCP severity string, or ``None`` if unknown. @@ -63,30 +98,69 @@ def _parse_json_line(line: str) -> dict | None: return obj if isinstance(obj, dict) else None -def _extract_pipeline_id(obj: dict | None) -> str: - """Pull a pipeline/task id from a parsed log object. +def _extract_pipeline_id(obj: dict | None, head: str) -> str: + """Pull a pipeline/task id from a record's parsed head or its raw text. - Checks the three places the JSON formatter can land an id (the formatter's - context allowlist only includes ``task_id``; everything else falls into - ``extra``). Returns ``""`` when none of them are set so the caller can - treat ``""`` as "no determinable id". + JSON records: checks the three places the JSON formatter can land an id + (the formatter's context allowlist only includes ``task_id``; everything + else falls into ``extra``). Console records (``obj is None``): matches the + inline ``pipeline_id=`` / ``task_id=`` pair the ConsoleFormatter renders + (#3547; the k8s pods emit console format, so the structured lookup alone + dropped every production line). Returns ``""`` when no id is determinable. """ - if obj is None: + if obj is not None: + ctx = obj.get("context") + if isinstance(ctx, dict): + task_id = ctx.get("task_id") + if task_id: + return str(task_id) + extra = obj.get("extra") + if isinstance(extra, dict): + # Production call sites use ``pipeline_id=...``; ``task_id`` is the + # defensive companion for any caller that picked the other spelling. + for key in ("pipeline_id", "task_id"): + value = extra.get(key) + if value: + return str(value) return "" - ctx = obj.get("context") - if isinstance(ctx, dict): - task_id = ctx.get("task_id") - if task_id: - return str(task_id) - extra = obj.get("extra") - if isinstance(extra, dict): - # Production call sites use ``pipeline_id=...``; ``task_id`` is the - # defensive companion for any caller that picked the other spelling. - for key in ("pipeline_id", "task_id"): - value = extra.get(key) - if value: - return str(value) - return "" + # Structured kwargs are appended at the END of a console line, so the + # record's own id is the LAST ``pipeline_id=``/``task_id=`` token. A + # leftmost match (``re.search``) could otherwise pick up an id embedded in + # the message body — e.g. a logged URL/command containing ``?pipeline_id=`` + # — and silently surface or hide the wrong records (#3566 review). + matches = _CONSOLE_ID_RE.findall(_strip_ansi(head)) + return matches[-1] if matches else "" + + +def _extract_severity(obj: dict | None, head: str) -> str | None: + """Pull the severity string from a record's parsed head or its raw text.""" + if obj is not None: + severity = obj.get("severity") + return severity if isinstance(severity, str) else None + match = _CONSOLE_HEAD_RE.match(_strip_ansi(head)) + return match.group(1) if match else None + + +def _group_records(lines: list[str]) -> list[list[str]]: + """Group physical lines into logical records. + + A record starts at a JSON object line or a console-format head + (timestamp + level). Every other line; traceback frames, wrapped + payloads; is a continuation of the preceding record. A leading run of + continuation lines (tail cut mid-record) forms its own headless record so + no input is silently dropped. + """ + records: list[list[str]] = [] + for line in lines: + stripped = _strip_ansi(line.strip()) + is_head = bool(stripped) and ( + stripped[0] == "{" or _CONSOLE_HEAD_RE.match(stripped) is not None + ) + if is_head or not records: + records.append([line]) + else: + records[-1].append(line) + return records def filter_log_lines( @@ -97,27 +171,33 @@ def filter_log_lines( pattern: re.Pattern[str] | None = None, limit: int | None = None, ) -> str: - """Filter a pod's raw log tail to the lines an operator asked for. - - Each line is parsed as a structured JSON log object. Filters are ANDed: - - * ``pipeline_id`` — keep lines whose pipeline/task id matches. A line's id - is read from ``context.task_id``, ``extra.pipeline_id`` or - ``extra.task_id`` (the three places the JSON formatter can land it); a - line with none of them set fails this filter (dropped). - * ``min_level`` — keep lines whose ``severity`` rank is ``>=`` the floor. A - line with no determinable severity fails this filter (dropped). An + """Filter a pod's raw log tail to the records an operator asked for. + + Physical lines are grouped into logical records first (a traceback stays + attached to the log line that raised it; see :func:`_group_records`), and + filters are ANDed per record: + + * ``pipeline_id``; keep records whose pipeline/task id matches. For a + JSON record the id is read from ``context.task_id``, + ``extra.pipeline_id`` or ``extra.task_id``; for a console record it is + matched from the inline ``pipeline_id=`` / ``task_id=`` pair. A record + with no determinable id fails this filter (dropped). + * ``min_level``; keep records whose severity rank is ``>=`` the floor + (JSON ``severity`` field, or the console ``[LEVEL]`` bracket). A record + with no determinable severity fails this filter (dropped). An unrecognised ``min_level`` raises ``ValueError`` — silently dropping the filter on a deliberately-set parameter is the footgun this guard avoids; callers should still validate up front to surface a useful error. - * ``pattern`` — keep lines the compiled regex finds (``re.search``) anywhere - in the raw line text. - - When at least one filter is active, only the last ``limit`` *matching* lines - are returned (the most recent), preserving order. With no active filter the - input is returned unchanged aside from the ``limit`` tail. ``limit <= 0`` - returns the empty string (Python's ``lines[-0:]`` gotcha would otherwise - return everything). + * ``pattern``; keep records the compiled regex finds (``re.search``) + anywhere in the record's raw text, continuation lines included, so a + match inside a traceback returns the whole traceback. + + When at least one filter is active, only the last ``limit`` *matching* + records are returned (the most recent), preserving order. With no active + filter the input is returned unchanged aside from the ``limit`` tail + (counted in physical lines, matching the raw-tail semantics callers + expect). ``limit <= 0`` returns the empty string (Python's ``lines[-0:]`` + gotcha would otherwise return everything). """ if min_level is not None: min_rank = severity_rank(min_level) @@ -127,7 +207,7 @@ def filter_log_lines( min_rank = None have_filter = bool(pipeline_id) or min_rank is not None or pattern is not None - def _tail(items: list[str]) -> list[str]: + def _tail(items: list) -> list: if limit is None: return items if limit <= 0: @@ -141,17 +221,19 @@ def _tail(items: list[str]) -> list[str]: return "\n".join(_tail(lines)) kept: list[str] = [] - for line in lines: - if pattern is not None and not pattern.search(line): + for record in _group_records(lines): + head = record[0] + text = "\n".join(record) + if pattern is not None and not pattern.search(text): continue if pipeline_id or min_rank is not None: - obj = _parse_json_line(line) - if pipeline_id and _extract_pipeline_id(obj) != pipeline_id: + obj = _parse_json_line(head) + if pipeline_id and _extract_pipeline_id(obj, head) != pipeline_id: continue if min_rank is not None: - rank = severity_rank(obj.get("severity") if obj else None) + rank = severity_rank(_extract_severity(obj, head)) if rank is None or rank < min_rank: continue - kept.append(line) + kept.append(text) return "\n".join(_tail(kept)) diff --git a/orchestrator/mcp_tools/__init__.py b/orchestrator/mcp_tools/__init__.py index dbd98507f1..67ff95a8eb 100644 --- a/orchestrator/mcp_tools/__init__.py +++ b/orchestrator/mcp_tools/__init__.py @@ -67,6 +67,10 @@ def cap_result_dict(result, **_kwargs): # type: ignore[no-redef] "get_container_logs": ( "lower `lines` or set a smaller `since_seconds` window, or target a single container" ), + "get_agent_transcript": ( + "lower `lines` (the transcript tail is returned newest-last), or omit " + "`agent_role` to list available transcripts without their bodies" + ), "list_containers": ( "request a single pipeline/phase if the API supports it; otherwise " "this result is proportional to the running container count" @@ -162,6 +166,8 @@ def __init__( _handle_check_health = _health._handle_check_health _handle_list_containers = _health._handle_list_containers _handle_get_container_logs = _health._handle_get_container_logs + _persisted_agent_logs_fallback = _health._persisted_agent_logs_fallback + _handle_get_agent_transcript = _health._handle_get_agent_transcript _handle_send_message = _health._handle_send_message # _consensus _handle_get_consensus_status = _consensus._handle_get_consensus_status diff --git a/orchestrator/mcp_tools/_dispatch.py b/orchestrator/mcp_tools/_dispatch.py index 33dc13fb27..848112d9bc 100644 --- a/orchestrator/mcp_tools/_dispatch.py +++ b/orchestrator/mcp_tools/_dispatch.py @@ -39,6 +39,7 @@ def handle_tool_call(self, tool_name: str, arguments: dict[str, Any]) -> dict[st "check_health": self._handle_check_health, "list_containers": self._handle_list_containers, "get_container_logs": self._handle_get_container_logs, + "get_agent_transcript": self._handle_get_agent_transcript, "send_message": self._handle_send_message, "get_consensus_status": self._handle_get_consensus_status, "get_phase": self._handle_get_phase, diff --git a/orchestrator/mcp_tools/_health.py b/orchestrator/mcp_tools/_health.py index 9cd72e2add..306da20c71 100644 --- a/orchestrator/mcp_tools/_health.py +++ b/orchestrator/mcp_tools/_health.py @@ -92,7 +92,13 @@ def _handle_list_containers(self, args: dict[str, Any]) -> dict[str, Any]: def _handle_get_container_logs(self, args: dict[str, Any]) -> dict[str, Any]: - """Get container logs, with auto-selection if container_id not specified.""" + """Get container logs, with auto-selection if container_id not specified. + + One-shot agent pods are reaped within minutes of exit, so when no live + container matches (or the live fetch 404s) this falls back to the + post-reap captures persisted by ``remove_agent_job`` (#3547); fallback + results carry ``"source": "persisted"``. + """ task_id = quote(args["task_id"], safe="") container_id = args.get("container_id") agent_role = args.get("agent_role") @@ -103,14 +109,22 @@ def _handle_get_container_logs(self, args: dict[str, Any]) -> dict[str, Any]: # Auto-select: list containers, filter by role, pick best match containers_result = self._make_request(f"/api/v1/pipelines/{task_id}/containers?all=true") containers = containers_result.get("data", {}).get("containers", []) - if not containers: - return {"error": "No containers found for this pipeline"} # Filter by agent_role if specified if agent_role: filtered = [c for c in containers if c.get("agent_role") == agent_role] if filtered: containers = filtered + elif containers: + # No live container for this role; the pod was likely + # already reaped; go straight to the persisted captures. + containers = [] + + if not containers: + fallback = self._persisted_agent_logs_fallback(task_id, None, agent_role) + if fallback is not None: + return fallback + return {"error": "No containers found for this pipeline"} # Prefer running containers, then most recently started running = [c for c in containers if c.get("status") == "running"] @@ -123,15 +137,147 @@ def _handle_get_container_logs(self, args: dict[str, Any]) -> dict[str, Any]: container_id = selected.get("container_id", "") cid = quote(container_id, safe="") - logs_result = self._make_request( - f"/api/v1/pipelines/{task_id}/containers/{cid}/logs?tail={lines}" - ) + try: + logs_result = self._make_request( + f"/api/v1/pipelines/{task_id}/containers/{cid}/logs?tail={lines}" + ) + except Exception: + # Live pod gone between listing and fetch (or an explicit + # container_id for an already-reaped pod): try the captures. + # When auto-select picked the container, forward its known role so + # role re-narrowing can recover the capture even if the exact + # container_id (a pod UID) misses the job-name-keyed lookup; the + # miss-on-ambiguity guard would otherwise give up in that race + # window (#3566 review). + fallback = self._persisted_agent_logs_fallback( + task_id, container_id, agent_role or selected.get("agent_role") + ) + if fallback is not None: + return fallback + raise - return { + data = logs_result.get("data", {}) + result = { "container_id": container_id, - "agent_role": agent_role or selected.get("agent_role") or None, + "agent_role": agent_role or selected.get("agent_role") or data.get("agent_role") or None, "status": selected.get("status") or None, - "logs": logs_result.get("data", {}).get("logs", ""), + "logs": data.get("logs", ""), + } + if data.get("source") == "persisted": + # The route itself served a post-reap capture; surface its metadata. + result["source"] = "persisted" + result["captured_at"] = data.get("captured_at") + result["exit_code"] = data.get("exit_code") + return result + + +def _persisted_agent_logs_fallback( + self, + task_id: str, + container_id: str | None, + agent_role: str | None, +) -> dict[str, Any] | None: + """Serve logs from the post-reap agent-log captures, or ``None`` on miss (#3547). + + Prefers an exact ``container_id``/job-name match, then the newest capture + for ``agent_role``, then the newest capture overall. ``task_id`` arrives + already URL-quoted by the caller. + """ + record: dict[str, Any] = {} + if container_id: + # Exact match first; but captures are keyed by Job name while an + # auto-selected container_id may be a pod UID, so a miss here falls + # through to the role/index lookup rather than giving up. + try: + cid = quote(container_id, safe="") + record_result = self._make_request(f"/api/v1/pipelines/{task_id}/agent-logs/{cid}") + record = record_result.get("data") or {} + except Exception: + record = {} + if not record and container_id and not agent_role: + # An explicit container_id that missed the exact lookup must not be + # silently substituted with an unrelated capture. Only a role filter is + # a legitimate re-narrowing (below); without one, treat it as a miss + # rather than returning "newest capture overall" labelled as a different + # job — that would hand the operator job-B when they asked for job-A + # (#3566 review). Newest-overall stays reserved for the no-container_id + # auto-select path. + return None + if not record: + try: + index_result = self._make_request(f"/api/v1/pipelines/{task_id}/agent-logs") + records = index_result.get("data", {}).get("records", []) + if agent_role: + records = [r for r in records if r.get("agent_role") == agent_role] + if not records: + return None + # Newest first per the route's ordering; fetch the full body. + job = quote(records[0].get("job_name", ""), safe="") + record_result = self._make_request(f"/api/v1/pipelines/{task_id}/agent-logs/{job}") + record = record_result.get("data") or {} + except Exception: + return None + if not record: + return None + return { + "container_id": record.get("job_name") or container_id, + "agent_role": record.get("agent_role") or agent_role, + "status": "reaped", + "source": "persisted", + "captured_at": record.get("captured_at"), + "exit_code": record.get("exit_code"), + "slice_id": record.get("slice_id"), + "truncated": record.get("truncated", False), + "logs": record.get("logs", ""), + } + + +def _handle_get_agent_transcript(self, args: dict[str, Any]) -> dict[str, Any]: + """Read a role's session transcript from the session-state store (#3547). + + Transcripts are pushed on every event-pod exit (``session-state push``) + and are the one artifact that always survives a one-shot run; this is the + operator-facing read path. On a miss (or when ``agent_role`` is omitted) + returns the store's index so the caller can pick a valid + ``(agent_role, slice_id)`` pair and retry. + """ + task_id = quote(args["task_id"], safe="") + agent_role = args.get("agent_role") + slice_id = args.get("slice_id") + lines = args.get("lines", 200) + + if agent_role: + query = f"role={quote(agent_role, safe='')}" + if slice_id: + query += f"&slice_id={quote(slice_id, safe='')}" + result = self._make_request(f"/api/v1/pipelines/{task_id}/session-state?{query}") + if result.get("found"): + data = result.get("data", {}) + transcript = data.get("transcript") or "" + all_lines = transcript.splitlines() + tail = all_lines[-lines:] if isinstance(lines, int) and lines > 0 else [] + return { + "found": True, + "agent_role": agent_role, + "slice_id": slice_id, + "session_id": data.get("session_id"), + "window_occupancy": data.get("window_occupancy"), + "total_transcript_lines": len(all_lines), + "lines_returned": len(tail), + "transcript_tail": "\n".join(tail), + } + + index = self._make_request(f"/api/v1/pipelines/{task_id}/session-state/index") + return { + "found": False, + "agent_role": agent_role, + "slice_id": slice_id, + "available_transcripts": index.get("records", []), + "hint": ( + "No stored transcript for that (agent_role, slice_id); retry with a " + "pair from available_transcripts. Records expire 6h after the " + "agent's last push." + ), } diff --git a/orchestrator/mcp_tools/_tool_defs.py b/orchestrator/mcp_tools/_tool_defs.py index 6ecbd0bbdc..5a5fa691a5 100644 --- a/orchestrator/mcp_tools/_tool_defs.py +++ b/orchestrator/mcp_tools/_tool_defs.py @@ -271,7 +271,10 @@ "description": ( "Get logs from a pipeline container. If container_id is omitted, " "auto-selects the best container (filtered by agent_role if given, " - "preferring running containers)." + "preferring running containers). One-shot agent pods are reaped " + "within minutes of exit; when the live pod is gone this falls back " + "to the post-reap log capture (result carries `source: persisted` " + "plus `captured_at`/`exit_code`), so exited agents stay diagnosable." ), "inputSchema": { "type": "object", @@ -297,6 +300,46 @@ "required": ["task_id"], }, }, + { + "name": "get_agent_transcript", + "description": ( + "Read an agent's Claude Code session transcript from the " + "session-state store. Agents push their transcript on every " + "event-pod exit, so this survives pod reaping (records expire ~6h " + "after the last push); use it to diagnose WHY an agent exited " + "when its pod (and pod logs) are already gone. Returns the last " + "`lines` JSONL entries. Omit agent_role to list the available " + "(agent_role, slice_id) transcripts for the pipeline." + ), + "inputSchema": { + "type": "object", + "properties": { + "task_id": { + "type": "string", + "description": "Pipeline/task ID", + }, + "agent_role": { + "type": "string", + "description": ( + "Agent role (e.g. 'coder', 'reviewer_code'). Omit to " + "list available transcripts instead." + ), + }, + "slice_id": { + "type": "string", + "description": ( + "Slice scope (e.g. 'slice-3'); omit for pipeline-level (non-sliced) agents." + ), + }, + "lines": { + "type": "integer", + "description": "Number of transcript JSONL lines to return from the tail", + "default": 200, + }, + }, + "required": ["task_id"], + }, + }, { "name": "send_message", "description": "Send a message to an agent in a pipeline. Sent as the 'overseer' role.", @@ -1055,12 +1098,12 @@ "pipeline_id": { "type": "string", "description": ( - "Keep only lines emitted for this pipeline/task id. " - "Matched against the log's `context.task_id`, with " - "`extra.pipeline_id` and `extra.task_id` as fallbacks " - "— production call sites use `pipeline_id=...`, which " - "the JsonFormatter lands in `extra` rather than the " - "context-allowlisted `task_id` slot." + "Keep only records emitted for this pipeline/task id. " + "Works for both log formats: JSON records are matched " + "on `context.task_id` / `extra.pipeline_id` / " + "`extra.task_id`; console-formatted records (what the " + "k8s pods emit) are matched on the inline " + "`pipeline_id=` / `task_id=` pair." ), }, "level": { @@ -1073,8 +1116,11 @@ "pattern": { "type": "string", "description": ( - "Python regular expression; keep only lines it matches " - "(`re.search`). A plain substring is a valid pattern." + "Python regular expression; keep only records it matches " + "(`re.search`). A plain substring is a valid pattern. " + "Multi-line tracebacks stay attached to the log line that " + "raised them, so matching the exception message returns " + "the whole stack." ), }, }, diff --git a/orchestrator/routes/containers.py b/orchestrator/routes/containers.py index a0097b2c6c..7f8515bc6c 100644 --- a/orchestrator/routes/containers.py +++ b/orchestrator/routes/containers.py @@ -31,6 +31,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] import os +from agent_log_store import get_agent_log_store from container_monitor import get_container_monitor from docker_client import ( ContainerNotFoundError, @@ -482,6 +483,11 @@ def get_container_logs(pipeline_id: str, container_id: str) -> tuple[Response, i "logs": "..." } } + + When the live container is gone (one-shot agent pods are reaped within + minutes of exit), falls back to the post-reap capture in the agent-log + store (#3547); the fallback payload carries ``"source": "persisted"`` + plus the capture metadata. """ try: tail = int(request.args.get("tail", request.args.get("lines", 100))) @@ -503,6 +509,20 @@ def get_container_logs(pipeline_id: str, container_id: str) -> tuple[Response, i status_code=400, ) except ContainerNotFoundError, PodNotFoundError: + record = get_agent_log_store().get(pipeline_id, container_id) + if record is not None: + return make_success_response( + "Logs retrieved from post-reap capture", + data={ + "logs": record.get("logs", ""), + "source": "persisted", + "captured_at": record.get("captured_at"), + "exit_code": record.get("exit_code"), + "agent_role": record.get("agent_role"), + "slice_id": record.get("slice_id"), + "truncated": record.get("truncated", False), + }, + ) return make_error_response( f"Container {container_id} not found", status_code=404, @@ -511,6 +531,34 @@ def get_container_logs(pipeline_id: str, container_id: str) -> tuple[Response, i return make_error_response(f"Backend error: {e}", status_code=500) +@containers_bp.route("//agent-logs", methods=["GET"]) +def list_agent_logs(pipeline_id: str) -> tuple[Response, int]: + """List the pipeline's post-reap agent log captures, newest first (#3547). + + Metadata only (``log_bytes`` instead of the log body); fetch a specific + capture via ``GET .../agent-logs/``. Captures are written by + ``remove_agent_job`` just before Job deletion and expire after + ``AGENT_LOG_TTL_SECONDS``. + """ + records = get_agent_log_store().list_records(pipeline_id) + return make_success_response( + "Persisted agent logs", + data={"records": records}, + ) + + +@containers_bp.route("//agent-logs/", methods=["GET"]) +def get_agent_log(pipeline_id: str, job_name: str) -> tuple[Response, int]: + """Return one post-reap agent log capture, including the log body (#3547).""" + record = get_agent_log_store().get(pipeline_id, job_name) + if record is None: + return make_error_response( + f"No persisted logs for {job_name}", + status_code=404, + ) + return make_success_response("Persisted agent logs", data=record) + + @containers_bp.route("//containers//health", methods=["GET"]) def check_container_health(pipeline_id: str, container_id: str) -> tuple[Response, int]: """ diff --git a/orchestrator/routes/deployment/_service_logs.py b/orchestrator/routes/deployment/_service_logs.py index 00538f192f..eca5986e09 100644 --- a/orchestrator/routes/deployment/_service_logs.py +++ b/orchestrator/routes/deployment/_service_logs.py @@ -41,15 +41,20 @@ def get_service_logs() -> tuple[Response, int]: per-pod scan window — the backing fetch is widened to 10 000 lines (``_MAX_LOG_LINES``) so the filter has material to match. - pipeline_id: keep only lines whose pipeline/task id matches; checks - ``context.task_id`` and the ``extra.pipeline_id`` / - ``extra.task_id`` fallbacks the JsonFormatter lands kwargs in - (#3032). + pipeline_id: keep only records whose pipeline/task id matches; for + JSON-formatted lines this checks ``context.task_id`` and the + ``extra.pipeline_id`` / ``extra.task_id`` fallbacks the + JsonFormatter lands kwargs in (#3032); for console-formatted + lines (what the k8s pods actually emit) it matches the inline + ``pipeline_id=`` / ``task_id=`` pair (#3547). level: minimum severity (case-sensitive; ``DEBUG`` … ``CRITICAL``); drops lower-severity and unstructured lines. The MCP ``level`` enum is the source of truth — the HTTP route rejects lowercase for parity (#3032). - pattern: Python regex; keep only lines it finds via ``re.search``. + pattern: Python regex; keep only records it finds via ``re.search``. + Filters operate on logical records; a multi-line traceback stays + attached to the log line that raised it, so a pattern matching + the exception message returns the whole stack (#3547). Compiled with no complexity guardrail — pathological patterns (catastrophic backtracking) can spin a request thread per pod line. This endpoint is gated behind ``require_lifecycle_secret`` diff --git a/orchestrator/routes/session_state.py b/orchestrator/routes/session_state.py index 9e4b776e27..4c01e1c018 100644 --- a/orchestrator/routes/session_state.py +++ b/orchestrator/routes/session_state.py @@ -114,6 +114,20 @@ def push_session_state(pipeline_id: str) -> tuple[Response, int]: return jsonify({"success": True, "stored": stored}), 200 +@session_state_bp.route("//session-state/index", methods=["GET"]) +def list_session_state(pipeline_id: str) -> tuple[Response, int]: + """Return the operator-facing index of stored records (#3547). + + Transcripts are the one artifact that always survives a one-shot agent + run, but until this route the only reader was the next event pod's + ``session-state pull``. This index lets an operator (via the + ``get_agent_transcript`` MCP tool) discover which ``(slice, role)`` + transcripts are currently readable; metadata only, no transcript bodies. + """ + records = get_session_state_store().list_records(pipeline_id) + return jsonify({"success": True, "records": records}), 200 + + @session_state_bp.route("//session-state", methods=["GET"]) def pull_session_state(pipeline_id: str) -> tuple[Response, int]: """Return a role's warm-resume record, or ``found: false`` on any miss. diff --git a/orchestrator/session_state_store.py b/orchestrator/session_state_store.py index b60c3d6dc5..e5e692815d 100644 --- a/orchestrator/session_state_store.py +++ b/orchestrator/session_state_store.py @@ -208,6 +208,48 @@ def get(self, pipeline_id: str, slice_id: str | None, role: str) -> SessionState transcript, ) + def list_records(self, pipeline_id: str) -> list[dict[str, Any]]: + """Enumerate the pipeline's stored records as an operator-facing index (#3547). + + Returns one metadata entry per live ``(slice, role)`` record - + ``slice_id`` (``None`` for pipeline-level), ``role``, ``session_id``, + ``window_occupancy`` and ``transcript_bytes``; without the transcript + bodies, so an operator can discover what is readable before pulling a + specific transcript. Best-effort like every other method: any Redis or + parse failure degrades to omitting the entry (or returning ``[]``), + never raising. + """ + prefix = f"{_KEY_PREFIX}:{pipeline_id}:" + try: + keys = sorted(self._redis.scan_iter(match=f"{prefix}*".encode())) + except Exception as exc: # noqa: BLE001; best-effort index + logger.warning("Failed to scan session-state 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 + remainder = key_str[len(prefix) :] + slice_segment, sep, role = remainder.partition(":") + if not sep or not role: + continue + slice_id = None if slice_segment == "none" else slice_segment + record = self.get(pipeline_id, slice_id, role) + if record is None: + continue + transcript_bytes = ( + len(record.transcript.encode("utf-8")) if record.transcript is not None else 0 + ) + entries.append( + { + "slice_id": slice_id, + "role": role, + "session_id": record.session_id, + "window_occupancy": record.window_occupancy, + "transcript_bytes": transcript_bytes, + } + ) + return entries + _store: SessionStateStore | None = None diff --git a/orchestrator/tests/test_agent_log_store.py b/orchestrator/tests/test_agent_log_store.py new file mode 100644 index 0000000000..4bd6d1b08c --- /dev/null +++ b/orchestrator/tests/test_agent_log_store.py @@ -0,0 +1,120 @@ +"""Tests for the post-reap agent-log store (#3547).""" + +import sys +from pathlib import Path + +import fakeredis +import pytest + +_orchestrator_path = Path(__file__).parent.parent +if str(_orchestrator_path) not in sys.path: + sys.path.insert(0, str(_orchestrator_path)) + +from agent_log_store import ( + AGENT_LOG_TTL_SECONDS, + MAX_LOG_BYTES, + AgentLogStore, +) + + +@pytest.fixture +def store(): + return AgentLogStore(fakeredis.FakeRedis()) + + +class TestRoundTrip: + def test_put_then_get(self, store): + assert ( + store.put( + "issue-1", + "egg-agent-issue-1-coder-abc", + logs="hello\nworld\n", + agent_role="coder", + slice_id="slice-3", + exit_code=137, + ) + is True + ) + rec = store.get("issue-1", "egg-agent-issue-1-coder-abc") + assert rec["logs"] == "hello\nworld\n" + assert rec["agent_role"] == "coder" + assert rec["slice_id"] == "slice-3" + assert rec["exit_code"] == 137 + assert rec["truncated"] is False + assert rec["captured_at"] + + def test_miss_returns_none(self, store): + assert store.get("issue-1", "nope") is None + + def test_requires_pipeline_and_job(self, store): + assert store.put("", "job", logs="x") is False + assert store.put("issue-1", "", logs="x") is False + + def test_ttl_applied(self, store): + store.put("issue-1", "job-1", logs="x") + ttl = store._redis.ttl(AgentLogStore._key("issue-1", "job-1")) + assert 0 < ttl <= AGENT_LOG_TTL_SECONDS + + def test_oversized_log_tail_truncated(self, store): + logs = "a" * MAX_LOG_BYTES + "TAIL" + store.put("issue-1", "job-1", logs=logs) + rec = store.get("issue-1", "job-1") + assert rec["truncated"] is True + assert rec["logs"].endswith("TAIL") + assert len(rec["logs"].encode()) == MAX_LOG_BYTES + + +class TestListRecords: + def test_index_newest_first_without_bodies(self, store): + store.put( + "issue-1", + "job-old", + logs="old", + agent_role="coder", + captured_at="2026-07-07T01:00:00+00:00", + ) + store.put( + "issue-1", + "job-new", + logs="newer!", + agent_role="reviewer_code", + captured_at="2026-07-07T02:00:00+00:00", + ) + store.put("issue-2", "job-other", logs="other") + + records = store.list_records("issue-1") + assert [r["job_name"] for r in records] == ["job-new", "job-old"] + assert all("logs" not in r for r in records) + assert records[0]["log_bytes"] == len("newer!") + + def test_include_logs_keeps_bodies(self, store): + store.put("issue-1", "job-1", logs="body") + records = store.list_records("issue-1", include_logs=True) + assert records[0]["logs"] == "body" + + def test_scan_failure_returns_empty(self): + class _Boom: + def scan_iter(self, *_a, **_k): + raise RuntimeError("redis down") + + assert AgentLogStore(_Boom()).list_records("issue-1") == [] + + +class TestDefensiveContract: + def test_write_failure_returns_false(self): + class _Boom: + def setex(self, *_a, **_k): + raise RuntimeError("redis down") + + assert AgentLogStore(_Boom()).put("issue-1", "job", logs="x") is False + + def test_read_failure_returns_none(self): + class _Boom: + def get(self, *_a, **_k): + raise RuntimeError("redis down") + + assert AgentLogStore(_Boom()).get("issue-1", "job") is None + + def test_malformed_payload_returns_none(self, store): + store._redis.set(AgentLogStore._key("issue-1", "job"), b"not json") + assert store.get("issue-1", "job") is None diff --git a/orchestrator/tests/test_consensus_log_dedup.py b/orchestrator/tests/test_consensus_log_dedup.py new file mode 100644 index 0000000000..02e6123323 --- /dev/null +++ b/orchestrator/tests/test_consensus_log_dedup.py @@ -0,0 +1,143 @@ +"""Regression for #3547 pain point 3; consensus poll-tick log spam. + +The run loop calls ``check_consensus`` every ~5 seconds per active slice, and +pre-fix each incomplete tick logged "Consensus incomplete — checking fallbacks" +and "Skipping pipeline-wide message-bus fallback ..." at INFO with +identical content. Two lines per tick per slice defined the INFO noise floor +for the whole service: with ``get_service_logs``'s 10,000-line scan budget the +effective window shrank to ~half an hour, and role-name pattern filters +matched the spam itself (roles appear in ``blocking_agents``). + +Post-fix the observations log at INFO only when the incomplete state +(confirmed count, blocking set, unresolved-NACK flag) changes, and at DEBUG +otherwise; reaching consensus resets the memo so the next round's first +observation is INFO again. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +# sys.path setup; orchestrator + shared. +_orchestrator_path = Path(__file__).parent.parent +_shared_path = _orchestrator_path.parent / "shared" +for _p in (_orchestrator_path, _shared_path): + if _p.exists() and str(_p) not in sys.path: + sys.path.insert(0, str(_p)) + +from concurrent_executor import ConcurrentPhaseExecutor # noqa: E402 +from models import ( # noqa: E402 + Pipeline, + PipelineConfig, + PipelinePhase, + PipelineStatus, +) +from peer_consensus import ( # noqa: E402 + create_peer_consensus_tracker, + remove_peer_consensus_tracker, +) +from review_graph import get_review_graph_for_phase # noqa: E402 + + +def _make_pipeline(pipeline_id: str) -> Pipeline: + config = PipelineConfig(concurrent_execution=True) + return Pipeline( + id=pipeline_id, + repo="test/repo", + issue_number=3547, + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + config=config, + ) + + +def _messages_logged(mock_logger, level: str) -> list[str]: + return [call.args[0] for call in getattr(mock_logger, level).call_args_list] + + +class TestIncompleteConsensusLogDedup: + def _run_ticks(self, ticks: int) -> MagicMock: + """Run ``ticks`` unchanged incomplete polls on one slice executor.""" + pipeline_id = "issue-3547-log-dedup" + pipeline = _make_pipeline(pipeline_id) + graph = get_review_graph_for_phase("implement", repo="test/repo") + create_peer_consensus_tracker(pipeline_id, graph, slice_id="slice-1") + try: + executor = ConcurrentPhaseExecutor( + pipeline, + spawn_fn=MagicMock(), + review_graph=graph, + slice_id="slice-1", + ) + with patch("concurrent_executor.logger") as mock_logger: + for _ in range(ticks): + result = executor.check_consensus() + assert result["is_complete"] is False + finally: + remove_peer_consensus_tracker(pipeline_id, "slice-1") + return mock_logger + + def test_first_tick_logs_info(self): + mock_logger = self._run_ticks(1) + infos = _messages_logged(mock_logger, "info") + assert "Consensus incomplete — checking fallbacks" in infos + assert any("Skipping pipeline-wide message-bus fallback" in m for m in infos) + + def test_unchanged_ticks_drop_to_debug(self): + mock_logger = self._run_ticks(5) + infos = _messages_logged(mock_logger, "info") + debugs = _messages_logged(mock_logger, "debug") + # One INFO per line for the first observation only... + assert infos.count("Consensus incomplete — checking fallbacks") == 1 + assert sum("Skipping pipeline-wide" in m for m in infos) == 1 + # ...and the four unchanged repeats land at DEBUG. + assert debugs.count("Consensus incomplete — checking fallbacks") == 4 + assert sum("Skipping pipeline-wide" in m for m in debugs) == 4 + + def test_state_change_logs_info_again(self): + pipeline_id = "issue-3547-log-dedup-change" + pipeline = _make_pipeline(pipeline_id) + graph = get_review_graph_for_phase("implement", repo="test/repo") + create_peer_consensus_tracker(pipeline_id, graph, slice_id="slice-1") + try: + executor = ConcurrentPhaseExecutor( + pipeline, + spawn_fn=MagicMock(), + review_graph=graph, + slice_id="slice-1", + ) + with patch("concurrent_executor.logger") as mock_logger: + executor.check_consensus() + # Simulate a prior differing observation (e.g. an agent just + # confirmed): the next tick must log at INFO again. + executor._last_incomplete_consensus_log = ("different",) + executor.check_consensus() + finally: + remove_peer_consensus_tracker(pipeline_id, "slice-1") + infos = _messages_logged(mock_logger, "info") + assert infos.count("Consensus incomplete — checking fallbacks") == 2 + + def test_memo_resets_when_consensus_completes(self): + """A completed round clears the memo so the next round starts at INFO.""" + pipeline_id = "issue-3547-log-dedup-reset" + pipeline = _make_pipeline(pipeline_id) + graph = get_review_graph_for_phase("implement", repo="test/repo") + create_peer_consensus_tracker(pipeline_id, graph, slice_id="slice-1") + try: + executor = ConcurrentPhaseExecutor( + pipeline, + spawn_fn=MagicMock(), + review_graph=graph, + slice_id="slice-1", + ) + executor._last_incomplete_consensus_log = ("stale",) + tracker = MagicMock() + tracker.evaluate.return_value = {"is_complete": True} + with patch("concurrent_executor.get_peer_consensus_tracker", return_value=tracker): + result = executor.check_consensus() + assert result["is_complete"] is True + assert executor._last_incomplete_consensus_log is None + finally: + remove_peer_consensus_tracker(pipeline_id, "slice-1") diff --git a/orchestrator/tests/test_containers_routes.py b/orchestrator/tests/test_containers_routes.py index f86f3cd91c..9c31a245cb 100644 --- a/orchestrator/tests/test_containers_routes.py +++ b/orchestrator/tests/test_containers_routes.py @@ -207,3 +207,83 @@ def test_join_helper_degrades_to_empty_on_failure(self): with patch("routes.get_repo_path", side_effect=RuntimeError("no repo path")): assert _resolved_models_by_container("issue-77") == {} + + +class TestPersistedAgentLogFallback: + """Post-reap log capture read paths (#3547). + + One-shot agent pods are reaped minutes after exit; ``remove_agent_job`` + snapshots their logs into the agent-log store, and the logs route falls + back to that capture instead of returning 404. + """ + + @pytest.fixture(autouse=True) + def _fakeredis_store(self): + import agent_log_store + import fakeredis + from agent_log_store import AgentLogStore + + agent_log_store.set_agent_log_store(AgentLogStore(fakeredis.FakeRedis())) + yield + agent_log_store.reset_agent_log_store() + + def _persist(self, pipeline_id="p-1", job_name="egg-agent-p-1-coder-abc", **kwargs): + from agent_log_store import get_agent_log_store + + defaults = {"logs": "agent stdout tail", "agent_role": "coder", "exit_code": 1} + defaults.update(kwargs) + get_agent_log_store().put(pipeline_id, job_name, **defaults) + + def test_logs_route_falls_back_to_capture(self, client): + from kubernetes_client import PodNotFoundError + + self._persist() + with patch("routes.containers._get_backend") as mock_get_backend: + mock_get_backend.return_value.get_container_logs.side_effect = PodNotFoundError( + "pod gone" + ) + response = client.get("/api/v1/pipelines/p-1/containers/egg-agent-p-1-coder-abc/logs") + assert response.status_code == 200 + data = response.get_json()["data"] + assert data["logs"] == "agent stdout tail" + assert data["source"] == "persisted" + assert data["exit_code"] == 1 + assert data["agent_role"] == "coder" + + def test_logs_route_404_when_no_capture(self, client): + from kubernetes_client import PodNotFoundError + + with patch("routes.containers._get_backend") as mock_get_backend: + mock_get_backend.return_value.get_container_logs.side_effect = PodNotFoundError( + "pod gone" + ) + response = client.get("/api/v1/pipelines/p-1/containers/unknown/logs") + assert response.status_code == 404 + + def test_live_logs_do_not_touch_store(self, client): + self._persist(logs="stale capture") + with patch("routes.containers._get_backend") as mock_get_backend: + mock_get_backend.return_value.get_container_logs.return_value = "live logs" + response = client.get("/api/v1/pipelines/p-1/containers/egg-agent-p-1-coder-abc/logs") + data = response.get_json()["data"] + assert data["logs"] == "live logs" + assert "source" not in data + + def test_agent_logs_index(self, client): + self._persist(job_name="job-a", captured_at="2026-07-07T01:00:00+00:00") + self._persist(job_name="job-b", captured_at="2026-07-07T02:00:00+00:00") + response = client.get("/api/v1/pipelines/p-1/agent-logs") + assert response.status_code == 200 + records = response.get_json()["data"]["records"] + assert [r["job_name"] for r in records] == ["job-b", "job-a"] + assert all("logs" not in r for r in records) + + def test_agent_logs_get_by_job_name(self, client): + self._persist(job_name="job-a") + response = client.get("/api/v1/pipelines/p-1/agent-logs/job-a") + assert response.status_code == 200 + assert response.get_json()["data"]["logs"] == "agent stdout tail" + + def test_agent_logs_get_miss_404(self, client): + response = client.get("/api/v1/pipelines/p-1/agent-logs/unknown") + assert response.status_code == 404 diff --git a/orchestrator/tests/test_kubernetes_spawner.py b/orchestrator/tests/test_kubernetes_spawner.py index b6abe44f24..1502da6a5d 100644 --- a/orchestrator/tests/test_kubernetes_spawner.py +++ b/orchestrator/tests/test_kubernetes_spawner.py @@ -1300,6 +1300,64 @@ def test_remove_cleans_session_on_k8s_error(self, spawner, mock_k8s_client, mock mock_gateway.delete_session_by_container.assert_called_once_with("job-name") +class TestRemoveAgentJobLogCapture: + """Pre-removal log capture into the agent-log store (#3547).""" + + @pytest.fixture(autouse=True) + def _fakeredis_store(self): + import agent_log_store + import fakeredis + from agent_log_store import AgentLogStore + + agent_log_store.set_agent_log_store(AgentLogStore(fakeredis.FakeRedis())) + yield + agent_log_store.reset_agent_log_store() + + def test_remove_persists_snapshot_before_deletion(self, spawner, mock_k8s_client): + from agent_log_store import get_agent_log_store + + mock_k8s_client.read_job_log_snapshot.return_value = { + "job_name": "job-name", + "pod_name": "job-name-xyz", + "pipeline_id": "issue-1", + "agent_role": "coder", + "slice_id": "slice-3", + "exit_code": 137, + "logs": "agent stdout\n", + } + spawner.remove_agent_job("job-name") + mock_k8s_client.remove_container.assert_called_once_with("job-name", force=False) + rec = get_agent_log_store().get("issue-1", "job-name") + assert rec["logs"] == "agent stdout\n" + assert rec["agent_role"] == "coder" + assert rec["slice_id"] == "slice-3" + assert rec["exit_code"] == 137 + + def test_remove_proceeds_when_snapshot_unavailable(self, spawner, mock_k8s_client): + """Pod already GC'd (snapshot None); removal is not blocked.""" + mock_k8s_client.read_job_log_snapshot.return_value = None + spawner.remove_agent_job("job-name") + mock_k8s_client.remove_container.assert_called_once_with("job-name", force=False) + + def test_remove_proceeds_when_snapshot_raises(self, spawner, mock_k8s_client): + mock_k8s_client.read_job_log_snapshot.side_effect = RuntimeError("api down") + spawner.remove_agent_job("job-name") + mock_k8s_client.remove_container.assert_called_once_with("job-name", force=False) + + def test_no_capture_without_pipeline_label(self, spawner, mock_k8s_client): + """A pod with no pipeline label has no operator-facing key; skip.""" + from agent_log_store import get_agent_log_store + + mock_k8s_client.read_job_log_snapshot.return_value = { + "job_name": "job-name", + "pipeline_id": None, + "logs": "orphan logs", + } + spawner.remove_agent_job("job-name") + assert get_agent_log_store().list_records("issue-1") == [] + mock_k8s_client.remove_container.assert_called_once() + + # --------------------------------------------------------------------------- # TestListPipelineJobs # --------------------------------------------------------------------------- diff --git a/orchestrator/tests/test_log_filter.py b/orchestrator/tests/test_log_filter.py index e454c32d95..4392f57a62 100644 --- a/orchestrator/tests/test_log_filter.py +++ b/orchestrator/tests/test_log_filter.py @@ -1,4 +1,4 @@ -"""Unit tests for the ``get_service_logs`` server-side filter (#3032).""" +"""Unit tests for the ``get_service_logs`` server-side filter (#3032, #3547).""" from __future__ import annotations @@ -17,6 +17,12 @@ def _line(severity: str, message: str, task_id: str | None = None) -> str: return json.dumps(obj) +def _console_line(severity: str, message: str, pipeline_id: str | None = None) -> str: + """A ConsoleFormatter-shaped line; what the k8s pods actually emit (#3547).""" + suffix = f" pipeline_id={pipeline_id}" if pipeline_id is not None else "" + return f"2026-07-07 21:30:00 [{severity:<8}] orchestrator: {message}{suffix}" + + def _line_extra_pipeline(severity: str, message: str, pipeline_id: str) -> str: """Production-shape line: the kwarg landed in ``extra`` (not ``context``). @@ -58,9 +64,11 @@ def test_level_floor_drops_lower_and_unstructured(self): raw = "\n".join( [ _line("INFO", "info"), + # Attached to the INFO record as a continuation line (#3547), + # so the level floor drops it along with its record. + "plain non-json line", _line("WARNING", "warn"), _line("ERROR", "err"), - "plain non-json line", ] ) out = filter_log_lines(raw, min_level="WARNING") @@ -69,6 +77,18 @@ def test_level_floor_drops_lower_and_unstructured(self): assert "info" not in out assert "plain non-json" not in out + def test_continuation_lines_survive_with_their_record(self): + """A continuation line rides along when its record passes the floor (#3547).""" + raw = "\n".join( + [ + _line("ERROR", "boom"), + " raw stderr detail", + ] + ) + out = filter_log_lines(raw, min_level="ERROR") + assert "boom" in out + assert "raw stderr detail" in out + def test_pipeline_id_scopes_to_task(self): raw = "\n".join( [ @@ -224,3 +244,222 @@ def test_limit_zero_returns_empty(self): def test_limit_negative_returns_empty(self): raw = "\n".join(_line("INFO", f"m{i}") for i in range(3)) assert filter_log_lines(raw, limit=-5) == "" + + +class TestConsoleFormatLines: + """Console-formatted (non-JSON) lines; what production pods emit (#3547). + + ``EggLogger`` only installs ``JsonFormatter`` when the environment detects + as GCP (``K_SERVICE`` set); the k8s pods detect as ``container`` and emit + ``ConsoleFormatter`` text with ``pipeline_id=...`` inline. Pre-fix the + ``pipeline_id`` filter parsed only JSON and dropped every production line. + """ + + def test_pipeline_id_matches_inline_pair(self): + raw = "\n".join( + [ + _console_line("INFO", "mine", pipeline_id="issue-3523"), + _console_line("INFO", "theirs", pipeline_id="issue-9999"), + _console_line("INFO", "untagged"), + ] + ) + out = filter_log_lines(raw, pipeline_id="issue-3523") + assert "mine" in out + assert "theirs" not in out + assert "untagged" not in out + + def test_pipeline_id_matches_inline_task_id_spelling(self): + raw = "2026-07-07 21:30:00 [INFO ] orchestrator: msg task_id=issue-3523" + assert "msg" in filter_log_lines(raw, pipeline_id="issue-3523") + + def test_pipeline_id_prefix_does_not_partial_match(self): + """``issue-35`` must not match ``issue-3523`` and vice versa.""" + raw = _console_line("INFO", "mine", pipeline_id="issue-3523") + assert filter_log_lines(raw, pipeline_id="issue-35") == "" + # ...and a longer requested id doesn't match a shorter logged one. + raw_short = _console_line("INFO", "short", pipeline_id="issue-35") + assert filter_log_lines(raw_short, pipeline_id="issue-3523") == "" + + def test_sub_task_id_key_does_not_match_task_id(self): + raw = "2026-07-07 21:30:00 [INFO ] orchestrator: msg sub_task_id=p-1" + assert filter_log_lines(raw, pipeline_id="p-1") == "" + + def test_min_level_reads_console_bracket(self): + raw = "\n".join( + [ + _console_line("INFO", "quiet"), + _console_line("WARNING", "warn"), + _console_line("ERROR", "err"), + ] + ) + out = filter_log_lines(raw, min_level="WARNING") + assert "warn" in out + assert "err" in out + assert "quiet" not in out + + def test_end_to_end_via_console_formatter(self): + """Producer/consumer parity with the real ``ConsoleFormatter``.""" + from egg_logging.formatters import ConsoleFormatter + + formatter = ConsoleFormatter(service="orchestrator", use_colors=False) + + def _emit(level: int, message: str, exc_info=None, **kwargs: object) -> str: + record = logging.LogRecord( + name="orchestrator.executor", + level=level, + pathname=__file__, + lineno=1, + msg=message, + args=(), + exc_info=exc_info, + ) + for key, value in kwargs.items(): + setattr(record, key, value) + return formatter.format(record) + + raw = "\n".join( + [ + _emit(logging.WARNING, "spawn failed", pipeline_id="issue-3523"), + _emit(logging.INFO, "routine poll", pipeline_id="issue-3523"), + _emit(logging.ERROR, "boom", pipeline_id="other-1"), + ] + ) + + out = filter_log_lines(raw, pipeline_id="issue-3523", min_level="WARNING") + assert "spawn failed" in out + assert "routine poll" not in out + assert "boom" not in out + + out = filter_log_lines(raw, pipeline_id="issue-3523") + assert "spawn failed" in out + assert "routine poll" in out + + def test_pipeline_id_prefers_trailing_kwarg_over_message_body(self): + """The authoritative id is the LAST token; a body id must not win (#3566). + + A logged URL/command in the message body can carry a ``pipeline_id=`` + token, but the record's own id is the structured kwarg appended at the + end. A leftmost match would surface the wrong record (false positive) + and hide the right one (false negative). + """ + # Record belongs to issue-3523 but its message body mentions other-1. + raw = ( + "2026-07-07 21:30:00 [INFO ] orchestrator: " + "GET /api/v1/pipelines?pipeline_id=other-1 pipeline_id=issue-3523" + ) + # Filtering for the real (trailing) id keeps it... + assert "GET /api/v1/pipelines" in filter_log_lines(raw, pipeline_id="issue-3523") + # ...and filtering for the body-embedded id does not. + assert filter_log_lines(raw, pipeline_id="other-1") == "" + + def test_colorized_console_lines_parity(self): + """A colorized (``use_colors=True``) capture still filters correctly (#3566). + + Production pods are non-TTY (colors off), but ANSI escapes wrapping the + level bracket / inline kwargs must not defeat head detection or the id + lookbehind if a colorized source is ever filtered. + """ + from egg_logging.formatters import ConsoleFormatter + + formatter = ConsoleFormatter(service="orchestrator", use_colors=True) + + def _emit(level: int, message: str, **kwargs: object) -> str: + record = logging.LogRecord( + name="orchestrator.executor", + level=level, + pathname=__file__, + lineno=1, + msg=message, + args=(), + exc_info=None, + ) + for key, value in kwargs.items(): + setattr(record, key, value) + return formatter.format(record) + + raw = "\n".join( + [ + _emit(logging.WARNING, "spawn failed", pipeline_id="issue-3523"), + _emit(logging.INFO, "routine poll", pipeline_id="issue-3523"), + _emit(logging.ERROR, "boom", pipeline_id="other-1"), + ] + ) + # Sanity: the fixture really carries ANSI escapes. + assert "\x1b[" in raw + + out = filter_log_lines(raw, pipeline_id="issue-3523", min_level="WARNING") + assert "spawn failed" in out + assert "routine poll" not in out + assert "boom" not in out + + +class TestTracebackGrouping: + """Multi-line tracebacks stay attached to their record (#3547).""" + + def _raw_with_traceback(self) -> str: + from egg_logging.formatters import ConsoleFormatter + + formatter = ConsoleFormatter(service="orchestrator", use_colors=False) + try: + raise RuntimeError("kaboom") + except RuntimeError: + import sys + + exc_info = sys.exc_info() + record = logging.LogRecord( + name="orchestrator.executor", + level=logging.ERROR, + pathname=__file__, + lineno=1, + msg="unhandled exception in poll loop", + args=(), + exc_info=exc_info, + ) + record.pipeline_id = "issue-3523" + with_tb = formatter.format(record) + after = _console_line("INFO", "next tick", pipeline_id="issue-3523") + return f"{_console_line('INFO', 'before')}\n{with_tb}\n{after}" + + def test_pattern_on_message_returns_whole_traceback(self): + out = filter_log_lines( + self._raw_with_traceback(), pattern=re.compile("unhandled exception") + ) + assert "Traceback (most recent call last):" in out + assert "RuntimeError: kaboom" in out + assert "next tick" not in out + + def test_pattern_on_frame_returns_whole_record(self): + """Matching inside the stack returns the record head too; no more + orphaned ``File "..."`` lines (#3547 pain point 6).""" + out = filter_log_lines(self._raw_with_traceback(), pattern=re.compile("kaboom")) + assert "unhandled exception in poll loop" in out + assert "Traceback (most recent call last):" in out + + def test_min_level_keeps_traceback_with_error_record(self): + out = filter_log_lines(self._raw_with_traceback(), min_level="ERROR") + assert "RuntimeError: kaboom" in out + assert "before" not in out + assert "next tick" not in out + + def test_pipeline_filter_keeps_traceback(self): + out = filter_log_lines(self._raw_with_traceback(), pipeline_id="issue-3523") + assert "RuntimeError: kaboom" in out + assert "next tick" in out + assert "before" not in out + + def test_limit_counts_records_not_lines(self): + out = filter_log_lines( + self._raw_with_traceback(), + pipeline_id="issue-3523", + limit=1, + ) + # The newest matching record is the single-line "next tick" one. + assert out == _console_line("INFO", "next tick", pipeline_id="issue-3523") + + def test_leading_headless_lines_form_their_own_record(self): + """A tail cut mid-record can start with continuation lines; they must + not crash the grouper and stay matchable by pattern.""" + raw = ' File "x.py", line 1, in f\nRuntimeError: cut\n' + _console_line("INFO", "head") + out = filter_log_lines(raw, pattern=re.compile("cut")) + assert "RuntimeError: cut" in out + assert "head" not in out diff --git a/orchestrator/tests/test_mcp_tools.py b/orchestrator/tests/test_mcp_tools.py index 4264528717..fc79dbcb39 100644 --- a/orchestrator/tests/test_mcp_tools.py +++ b/orchestrator/tests/test_mcp_tools.py @@ -246,6 +246,245 @@ def test_auto_select_most_recent_stopped(self, handler): assert result["container_id"] == "new" +class TestGetContainerLogsPersistedFallback: + """Post-reap capture fallback in get_container_logs (#3547).""" + + _RECORD = { + "job_name": "egg-agent-issue-42-coder-abc", + "agent_role": "coder", + "slice_id": "slice-3", + "exit_code": 137, + "captured_at": "2026-07-07T21:00:00+00:00", + "truncated": False, + "logs": "captured stdout", + } + + def test_no_live_container_for_role_uses_captures(self, handler): + """Live list has other roles but not the requested one; the reaped + role's capture is served instead of another role's live logs.""" + containers_response = { + "data": { + "containers": [ + { + "container_id": "c-tester", + "status": "running", + "agent_role": "tester", + "started_at": "2026-01-01T00:00:00Z", + } + ] + } + } + index_response = {"data": {"records": [dict(self._RECORD, logs=None, log_bytes=15)]}} + record_response = {"data": dict(self._RECORD)} + + with patch.object(handler, "_make_request") as mock_req: + mock_req.side_effect = [containers_response, index_response, record_response] + result = handler.handle_tool_call( + "get_container_logs", {"task_id": "issue-42", "agent_role": "coder"} + ) + + assert result["source"] == "persisted" + assert result["logs"] == "captured stdout" + assert result["container_id"] == "egg-agent-issue-42-coder-abc" + assert result["agent_role"] == "coder" + assert result["exit_code"] == 137 + assert result["status"] == "reaped" + + def test_live_fetch_404_falls_back_to_exact_capture(self, handler): + from urllib.error import HTTPError + + record_response = {"data": dict(self._RECORD)} + + def _sequence(endpoint, **kwargs): + if endpoint.endswith("/logs?tail=100"): + raise HTTPError(endpoint, 404, "not found", None, None) + return record_response + + with patch.object(handler, "_make_request", side_effect=_sequence): + result = handler.handle_tool_call( + "get_container_logs", + {"task_id": "issue-42", "container_id": "egg-agent-issue-42-coder-abc"}, + ) + + assert result["source"] == "persisted" + assert result["logs"] == "captured stdout" + + def test_no_containers_and_no_captures_errors(self, handler): + with patch.object(handler, "_make_request") as mock_req: + mock_req.return_value = {"data": {"containers": [], "records": []}} + result = handler.handle_tool_call("get_container_logs", {"task_id": "issue-42"}) + + assert "error" in result + + def test_explicit_container_miss_no_role_does_not_serve_unrelated(self, handler): + """An explicit container_id that misses must not return newest-overall (#3566). + + With no agent_role to re-narrow, guessing the newest capture would hand + the operator a *different* job's logs than the one they named. The index + endpoint must not even be consulted; the original fetch error surfaces. + """ + from urllib.error import HTTPError + + index_calls: list[str] = [] + + def _sequence(endpoint, **kwargs): + if endpoint.endswith("/logs?tail=100"): + raise HTTPError(endpoint, 404, "not found", None, None) + if "/agent-logs/" in endpoint: + return {"data": {}} # exact-match miss for the requested id + if endpoint.endswith("/agent-logs"): + index_calls.append(endpoint) + # A sibling capture exists; it must NOT be substituted. + return {"data": {"records": [dict(self._RECORD)]}} + return {"data": {}} + + with patch.object(handler, "_make_request", side_effect=_sequence): + result = handler.handle_tool_call( + "get_container_logs", + {"task_id": "issue-42", "container_id": "pod-uid-does-not-match"}, + ) + + assert index_calls == [] # newest-overall lookup never attempted + assert result.get("logs") != "captured stdout" + assert "error" in result + + def test_explicit_container_miss_with_role_falls_through_to_role_capture(self, handler): + """A role IS a legitimate re-narrowing: container_id miss + role serves + the newest capture for that role (#3566).""" + from urllib.error import HTTPError + + def _sequence(endpoint, **kwargs): + if endpoint.endswith("/logs?tail=100"): + raise HTTPError(endpoint, 404, "not found", None, None) + if endpoint.endswith("/agent-logs/pod-uid-does-not-match"): + return {"data": {}} # exact-match miss + if endpoint.endswith("/agent-logs"): + return {"data": {"records": [dict(self._RECORD, logs=None, log_bytes=15)]}} + return {"data": dict(self._RECORD)} # full body for the role's job + + with patch.object(handler, "_make_request", side_effect=_sequence): + result = handler.handle_tool_call( + "get_container_logs", + { + "task_id": "issue-42", + "container_id": "pod-uid-does-not-match", + "agent_role": "coder", + }, + ) + + assert result["source"] == "persisted" + assert result["logs"] == "captured stdout" + assert result["agent_role"] == "coder" + + def test_auto_selected_container_forwards_role_to_fallback(self, handler): + """Auto-select picks a container, the live fetch loses the race, and the + selected UID misses the job-name-keyed exact lookup. The selected + container's role must be forwarded so role re-narrowing recovers the + capture instead of the miss-on-ambiguity guard giving up (#3566).""" + from urllib.error import HTTPError + + containers_response = { + "data": { + "containers": [ + { + "container_id": "pod-uid-not-a-job-name", + "status": "running", + "agent_role": "coder", + "started_at": "2026-01-01T00:00:00Z", + } + ] + } + } + + def _sequence(endpoint, **kwargs): + if endpoint.endswith("/logs?tail=100"): + raise HTTPError(endpoint, 404, "not found", None, None) + if "/containers?all=true" in endpoint: + return containers_response + if endpoint.endswith("/agent-logs/pod-uid-not-a-job-name"): + return {"data": {}} # exact-match miss on the pod UID + if endpoint.endswith("/agent-logs"): + return {"data": {"records": [dict(self._RECORD, logs=None, log_bytes=15)]}} + return {"data": dict(self._RECORD)} # full body for the role's job + + with patch.object(handler, "_make_request", side_effect=_sequence): + # No container_id and no agent_role: pure auto-select path. + result = handler.handle_tool_call("get_container_logs", {"task_id": "issue-42"}) + + assert result["source"] == "persisted" + assert result["logs"] == "captured stdout" + assert result["agent_role"] == "coder" + + +class TestGetAgentTranscript: + """Operator read path for session-state transcripts (#3547).""" + + def test_returns_transcript_tail(self, handler): + transcript = "\n".join(f'{{"line": {i}}}' for i in range(10)) + response = { + "success": True, + "found": True, + "data": { + "session_id": "sid-1", + "window_occupancy": 12345, + "transcript": transcript, + }, + } + with patch.object(handler, "_make_request") as mock_req: + mock_req.return_value = response + result = handler.handle_tool_call( + "get_agent_transcript", + { + "task_id": "issue-42", + "agent_role": "coder", + "slice_id": "slice-3", + "lines": 3, + }, + ) + + mock_req.assert_called_once_with( + "/api/v1/pipelines/issue-42/session-state?role=coder&slice_id=slice-3" + ) + assert result["found"] is True + assert result["session_id"] == "sid-1" + assert result["window_occupancy"] == 12345 + assert result["total_transcript_lines"] == 10 + assert result["lines_returned"] == 3 + assert result["transcript_tail"].splitlines() == [ + '{"line": 7}', + '{"line": 8}', + '{"line": 9}', + ] + + def test_miss_returns_index(self, handler): + miss = {"success": True, "found": False} + index = { + "success": True, + "records": [ + {"slice_id": "slice-3", "role": "coder", "session_id": "s", "transcript_bytes": 9} + ], + } + with patch.object(handler, "_make_request") as mock_req: + mock_req.side_effect = [miss, index] + result = handler.handle_tool_call( + "get_agent_transcript", {"task_id": "issue-42", "agent_role": "documenter"} + ) + + assert result["found"] is False + assert result["available_transcripts"] == index["records"] + assert "hint" in result + + def test_no_role_lists_available(self, handler): + index = {"success": True, "records": []} + with patch.object(handler, "_make_request") as mock_req: + mock_req.return_value = index + result = handler.handle_tool_call("get_agent_transcript", {"task_id": "issue-42"}) + + mock_req.assert_called_once_with("/api/v1/pipelines/issue-42/session-state/index") + assert result["found"] is False + assert result["available_transcripts"] == [] + + class TestSendMessage: def test_send_basic(self, handler): with patch.object(handler, "_make_request") as mock_req: @@ -1097,6 +1336,8 @@ def test_all_tools_registered(self, handler): "check_health", "list_containers", "get_container_logs", + # Session-transcript read path (#3547) + "get_agent_transcript", "send_message", "get_consensus_status", "get_phase", diff --git a/orchestrator/tests/test_session_state_routes.py b/orchestrator/tests/test_session_state_routes.py index f6df956cde..f79dd34b0e 100644 --- a/orchestrator/tests/test_session_state_routes.py +++ b/orchestrator/tests/test_session_state_routes.py @@ -85,6 +85,48 @@ def test_pipeline_level_pull_omits_slice(self, client): assert body["data"]["session_id"] == "pl" +class TestIndex: + """Operator-facing index of stored records (#3547).""" + + _INDEX_URL = f"{_URL}/index" + + def test_empty_index(self, client): + r = client.get(self._INDEX_URL) + assert r.status_code == 200 + assert r.get_json() == {"success": True, "records": []} + + def test_index_lists_metadata_without_transcripts(self, client): + client.post( + _URL, + json={ + "role": "coder", + "slice_id": "slice-3", + "session_id": "sid-a", + "window_occupancy": 42, + "transcript": '{"l": 1}\n', + }, + ) + client.post(_URL, json={"role": "reviewer_code", "session_id": "sid-b"}) + + records = client.get(self._INDEX_URL).get_json()["records"] + assert len(records) == 2 + by_role = {r["role"]: r for r in records} + coder = by_role["coder"] + assert coder["slice_id"] == "slice-3" + assert coder["session_id"] == "sid-a" + assert coder["window_occupancy"] == 42 + assert coder["transcript_bytes"] == len('{"l": 1}\n') + assert "transcript" not in coder + reviewer = by_role["reviewer_code"] + assert reviewer["slice_id"] is None + assert reviewer["transcript_bytes"] == 0 + + def test_index_scoped_to_pipeline(self, client): + client.post(_URL, json={"role": "coder", "session_id": "a"}) + other = client.get("/api/v1/pipelines/issue-2/session-state/index").get_json() + assert other["records"] == [] + + class TestValidation: def test_push_requires_role(self, client): r = client.post(_URL, json={"session_id": "a"}) diff --git a/orchestrator/tests/test_session_state_store.py b/orchestrator/tests/test_session_state_store.py index 469f73ce75..8a416a8d3c 100644 --- a/orchestrator/tests/test_session_state_store.py +++ b/orchestrator/tests/test_session_state_store.py @@ -114,3 +114,33 @@ def setex(self, *_a, **_k): store = SessionStateStore(_Boom()) assert store.put("issue-1", "slice-3", "coder", session_id="a") is False + + +class TestListRecords: + """Operator-facing index over the pipeline's stored records (#3547).""" + + def test_lists_all_records_for_pipeline(self, store): + store.put("issue-1", "slice-3", "coder", session_id="a", transcript='{"l":1}\n') + store.put("issue-1", None, "architect", session_id="b", window_occupancy=7) + store.put("issue-2", "slice-1", "coder", session_id="c") + + records = store.list_records("issue-1") + assert len(records) == 2 + by_role = {r["role"]: r for r in records} + assert by_role["coder"]["slice_id"] == "slice-3" + assert by_role["coder"]["transcript_bytes"] == len('{"l":1}\n') + assert by_role["architect"]["slice_id"] is None + assert by_role["architect"]["window_occupancy"] == 7 + + def test_scan_failure_returns_empty(self): + class _Boom: + def scan_iter(self, *_a, **_k): + raise RuntimeError("redis down") + + assert SessionStateStore(_Boom()).list_records("issue-1") == [] + + def test_malformed_record_omitted(self, store): + store.put("issue-1", "slice-3", "coder", session_id="a") + store._redis.set(SessionStateStore._key("issue-1", "slice-4", "coder"), b"not json") + records = store.list_records("issue-1") + assert [r["slice_id"] for r in records] == ["slice-3"]