scripts.e2e_run

Reference › scripts › e2e_run

The end-to-end dogfood runner: it submits one intent to the real orchestrata pipeline and records everything with full real-time observability. A human runs it, and tests/scripts/test_e2e_run.py imports it. It drives the PipelineRunner from pipeline, calls models through client and sets up logging with logging. See the kernel narrative for the pipeline.

Invocation: main requires --intent. Other options are --slug (a short label for the run directory), --model (default deepseek-v4-pro), --profile (a named spike profile from configs/spike-profiles.yaml; when set the per-Volk profile table is authoritative and --model is ignored), --max-rounds (default 6, the most clarification responses before giving up), --answer (repeatable scripted answers) and --answer-mode. A plain --answer entry is used positionally, in order. An entry of the form regex=>text is matched against the question text and the first matching pattern wins whatever the round (resolve_answer, _parse_answer_specs); an LLM can answer instead (_llm_answer_fn).

_setup_env prepares a run directory and environment for the run. A _Tee copies output to a log file as well as the terminal. If the pipeline asks a clarification question (_routed_question_text), main answers it and resubmits, up to the round limit. An injection seam, runner_factory, lets tests supply a scripted runner instead of the live one.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

_Tee

Line 307

Mirror a stream to a file, flushing on every write (real-time).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/e2e_run.py

#!/usr/bin/env python3
"""End-to-end orchestrata dogfood runner with full real-time observability.

Submits an intent to the live PipelineRunner (DeepSeek v4-pro) and captures,
unbuffered and in real time, every artifact of the run into a per-run directory:

    logs/e2e-runs/<ts>-<slug>/
        app.jsonl           structured logs (SPEC-100): all loggers, errors, gateway, CXP route logs
        conversation.jsonl  full volk<->LLM exchanges (system+user prompt, response) per round-trip
        cxp.jsonl           CXP packet flow for the thread (from the Postman log)
        summary.json        run metadata: intent, thread, models, timings, status, env
        console.log         human-readable mirror of stdout/stderr

Tail any of these live:  tail -f logs/e2e-runs/<run>/conversation.jsonl | jq .

This drives PipelineRunner in-process (no websocket gateway); gateway-message
capture applies when the same intent is driven through `kernel serve` — see the
test plan. Conversation capture is enabled via ORCHESTRATA_CONVERSATION_LOG,
read by llm.client._capture_exchange.

Usage:
    python scripts/e2e_run.py --intent "Generate developer docs for the kernel package" \
        --slug docs-kernel [--model deepseek-v4-pro] [--max-rounds 6]
"""

import argparse
import asyncio
import concurrent.futures
import logging
import os
import re
import sys
from collections.abc import Callable, MutableMapping
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

# Running as `python scripts/e2e_run.py` puts scripts/ on sys.path[0], not the
# repo root — so the `scripts` namespace package (imported transitively by some
# src modules, e.g. verification.spec_completeness) fails with
# "No module named 'scripts'". Prepend the repo root so those imports resolve.
_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

_logger = logging.getLogger(__name__)

# Canonical scripted fallback answers (backlog #223): used whenever no
# --answer positional entries are given (today's default) or a --answer
# list consists entirely of `regex=>text` pattern pairs, leaving no plain
# positional entries to index into.
_DEFAULT_ANSWERS: list[str] = [
    "Yes, proceed. Audience: orchestrata-ultimate contributors. "
    "Produce developer-facing Markdown. Use your best judgment on structure "
    "and proceed autonomously without further questions.",
    "Proceed autonomously with your best judgment; no further clarification needed.",
]


def resolve_answer(
    question_text: str,
    rounds: int,
    *,
    patterns: list[tuple[re.Pattern[str], str]],
    positional: list[str],
    llm_answer_fn: Callable[[str], str] | None = None,
) -> str:
    """Resolve one clarification answer for round ``rounds`` (backlog #223).

    The old mechanism indexed ``--answer`` positionally by round number,
    content-blind — it broke whenever clarification rounds were
    nondeterministic (a mis-ordered answer even poisons the concierge's
    replayed ``prior_answers``). Resolution order here:

      1. The first ``(regex, text)`` pair in ``patterns`` whose regex
         ``search()``es ``question_text`` wins — its ``text`` is returned.
      2. Else, if ``llm_answer_fn`` is given, call it with
         ``question_text``. ANY exception, or an empty/None result, falls
         through to (3) — logged as a warning, never allowed to crash the
         run.
      3. Else (or on fallthrough from 2) the positional scripted answer for
         this round, clamped to the last entry once ``rounds`` runs past
         the end of ``positional`` — today's behavior, unchanged.
    """
    for pattern, text in patterns:
        if pattern.search(question_text):
            return text

    if llm_answer_fn is not None:
        try:
            llm_answer = llm_answer_fn(question_text)
        except Exception as exc:
            _logger.warning(
                "LLM answer fn raised %s: %s — falling back to the scripted answer",
                type(exc).__name__,
                exc,
            )
        else:
            if llm_answer:
                return llm_answer
            _logger.warning(
                "LLM answer fn returned an empty answer — falling back to "
                "the scripted answer"
            )

    return positional[min(rounds, len(positional) - 1)]


def _parse_answer_specs(
    raw: list[str], *, on_error: Callable[[str], None]
) -> tuple[list[tuple[re.Pattern[str], str]], list[str]]:
    """Split ``--answer`` entries into ``(regex, text)`` patterns and plain
    positional strings.

    An entry containing ``=>`` is split ONCE on its first occurrence into a
    regex and its answer text (``regex=>text``); anything without ``=>``
    stays positional — today's ``--answer`` behavior, byte-for-byte. An
    invalid regex calls ``on_error`` with a clear message; ``main()`` wires
    this to the argparse parser's own ``.error()`` (prints usage, exits 2).
    """
    patterns: list[tuple[re.Pattern[str], str]] = []
    positional: list[str] = []
    for entry in raw:
        if "=>" in entry:
            regex_text, _, answer_text = entry.partition("=>")
            try:
                compiled = re.compile(regex_text)
            except re.error as exc:
                on_error(f"invalid --answer regex {regex_text!r}: {exc}")
                continue  # on_error normally exits; keeps type-checkers satisfied
            patterns.append((compiled, answer_text))
        else:
            positional.append(entry)
    return patterns, positional


def _llm_answer_fn(intent: str) -> Callable[[str], str]:
    """Build a closure that answers a clarification question via a cheap,
    deterministic LLM call, in the voice of the human requester behind
    ``intent``.

    Deliberately undefended: any exception raised while calling the LLM
    (LLMConfigError/LLMClientError/network/asyncio errors, ...) propagates
    straight out of the returned closure. Fail-soft is ``resolve_answer``'s
    job (its try/except around the ``llm_answer_fn(...)`` call) — double-
    guarding here would just hide the same failure twice.
    """
    system_prompt = (
        "You are the human requester in an automated end-to-end run. "
        f"The original intent was: {intent}. Answer the assistant's "
        "clarification question concisely and decisively, grant it "
        "autonomy to use best judgment, and do NOT introduce new "
        "requirements."
    )

    def _answer(question_text: str) -> str:
        from orchestrata_ultimate.llm.client import call_llm

        answer: str = _run_coro_blocking(
            call_llm(
                system_prompt,
                question_text,
                model="claude-haiku-4-5",
                temperature=0.0,
            )
        )
        return answer

    return _answer


def _run_coro_blocking(coro: Any) -> Any:
    """Run ``coro`` to completion from EITHER a sync or an async caller.

    The feature relay loop is synchronous, but the routed path resolves
    answers inside ``async def _run_routed`` — where a bare ``asyncio.run``
    raises RuntimeError (loop already running), which would silently kill
    the LLM answer tier on every routed round. When a loop is already
    running, execute the coroutine on its own fresh loop in a worker
    thread instead.
    """
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        return pool.submit(asyncio.run, coro).result()


def _routed_question_text(result: Any) -> str:
    """Extract the escalating stage's question text from a routed
    ``PipelineResult.stages`` — public field only, no private state.

    ``stages`` is an insertion-ordered dict; the escalating stage is always
    the last one inserted (``orchestrator.pipeline_runner``'s
    ``_execute_stages`` sets ``result.stages[stage_spec.alias] =
    stage_result`` immediately before classifying the interrupt / saving
    the escalation).
    """
    if not result.stages:
        return ""
    last_stage = next(reversed(result.stages.values()), None)
    packet = getattr(last_stage, "output_packet", None)
    if packet is None or not isinstance(packet.payload, dict):
        return ""
    return str(packet.payload.get("text", ""))[:600]


def _setup_env(
    run_dir: Path,
    model: str,
    env: MutableMapping[str, str] | None = None,
    profile: str | None = None,
) -> dict[str, str]:
    """Populate process env BEFORE any orchestrata import (config reads env at import).

    Args:
        run_dir: The per-run output directory.
        model: The DeepSeek model id to pin every volk to. Ignored when
            *profile* is given.
        profile: A ``configs/spike-profiles.yaml`` profile name. When set,
            the per-volk profile table is AUTHORITATIVE: ``SPIKE_PROFILE``
            is exported and the legacy ``ORCHESTRATA_LLM_*`` single-model
            pin is skipped entirely (that pin mutates infra_config's volk
            table and would fight the profile).
        env: The mutable env mapping to populate — defaults to ``os.environ``
            (real process env, required for an actual run since orchestrata
            config binds env at import time). Tests may pass a plain ``dict``
            instead so a real invocation of this function never mutates the
            shared test-process environment.

    Returns:
        The resolved values that were applied, for the run summary.
    """
    env = env if env is not None else os.environ
    defaults = {
        # Live infra (Incus containers) — only set if caller did not provide them.
        "REDIS_PORT": "6379",
        "REDIS_DB": "0",
        # The graph address comes from .env (KG_ROUTING_URI), never a
        # literal: this pinned bolt://10.53.1.11:7687, the Neo4j primary
        # destroyed on 2026-07-27, and setdefault() meant a correctly
        # configured environment was left alone but an unconfigured one
        # got silently aimed at a host that no longer exists.
    }
    for key, val in defaults.items():
        env.setdefault(key, val)

    # REDIS_HOST is REQUIRED (RUL-003, no infra default): this used to
    # setdefault 10.53.1.10 — the PRODUCTION Redis — so a correctly
    # configured environment was left alone but an unconfigured one got
    # silently aimed at prod. Same failure mode the KG address comment above
    # already fixed; mirrored here.
    if not env.get("REDIS_HOST"):
        raise RuntimeError(
            "REDIS_HOST is not set; define it in .env — this script has no "
            "default (it used to silently point at the production Redis "
            "10.53.1.10)."
        )

    if profile:
        # Profile mode: the per-volk table in configs/spike-profiles.yaml is
        # authoritative — no single-model pin.
        env["SPIKE_PROFILE"] = profile
    else:
        # Pin every volk to DeepSeek v4-pro for this run (overrides per-volk table).
        env["ORCHESTRATA_LLM_PROVIDER"] = "deepseek"
        env["ORCHESTRATA_LLM_MODEL"] = model
        env["ORCHESTRATA_LLM_REASONING_MODEL"] = model
    # A scripted dogfood run must never trigger PipelineRunner's kernel-session
    # integration (real bug found writing this script's tests: boot()'s
    # SessionController is rooted at the live repo checkout, and
    # runner.shutdown() in main()'s `finally` block ends that session —
    # unconditionally git add -A / commit / push'ing the WHOLE working tree
    # plus creating and pushing a git tag. A real incident during this
    # script's own test suite auto-committed unrelated in-flight work from a
    # dirty tree. See kernel/pipeline_session.py::_init_session_manager.
    env["ORCHESTRATA_KERNEL_SESSION_DISABLED"] = "1"
    # Turn on real-time full-conversation capture (read by llm.client).
    env["ORCHESTRATA_CONVERSATION_LOG"] = str(run_dir / "conversation.jsonl")
    # Contain any volk/driver file writes to a scratch workspace under the run dir.
    # ABSOLUTE path (dogfood F17): run_dir is cwd-relative, and downstream
    # resolve-vs-prefix workspace checks refuse every write when the base is
    # a relative string.
    workspace = (run_dir / "workspace").resolve()
    workspace.mkdir(parents=True, exist_ok=True)
    env["ORCHESTRATA_WORKSPACE"] = str(workspace)
    # Unbuffered child processes (drivers) for live output.
    env["PYTHONUNBUFFERED"] = "1"
    # Repo root on PYTHONPATH so subprocesses can import the `scripts` package.
    _existing_pp = env.get("PYTHONPATH", "")
    env["PYTHONPATH"] = str(_REPO_ROOT) + (
        os.pathsep + _existing_pp if _existing_pp else ""
    )

    applied = {k: env[k] for k in defaults}
    applied["REDIS_HOST"] = env["REDIS_HOST"]
    applied.update(
        provider=f"profile:{profile}" if profile else "deepseek",
        model=f"profile:{profile}" if profile else model,
        workspace=env["ORCHESTRATA_WORKSPACE"],
        conversation_log=env["ORCHESTRATA_CONVERSATION_LOG"],
    )
    return applied


class _Tee:
    """Mirror a stream to a file, flushing on every write (real-time)."""

    def __init__(self, stream, fh):
        self._stream = stream
        self._fh = fh

    def write(self, data: str) -> int:
        self._stream.write(data)
        self._stream.flush()
        self._fh.write(data)
        self._fh.flush()
        return len(data)

    def flush(self) -> None:
        self._stream.flush()
        self._fh.flush()


def main(
    argv: list[str] | None = None,
    runner_factory: Callable[[], Any] | None = None,
) -> int:
    """Run the dogfood pipeline.

    ``runner_factory`` is the injection seam for the pipeline runner:
    ``None`` (every real invocation) constructs the live
    ``PipelineRunner``; tests may pass a factory returning a real scripted
    runner so the clarification round-loop and error paths run
    deterministically without a multi-minute LLM cascade.
    """
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--intent", required=True, help="The intent text to submit")
    ap.add_argument("--slug", default="run", help="Short label for the run dir")
    ap.add_argument("--model", default="deepseek-v4-pro", help="DeepSeek model id")
    ap.add_argument(
        "--profile",
        default=None,
        help=(
            "SPIKE_PROFILE name (configs/spike-profiles.yaml). When set, the "
            "per-volk profile table is authoritative and --model is ignored."
        ),
    )
    ap.add_argument(
        "--max-rounds",
        type=int,
        default=6,
        help="Max clarification responses before giving up",
    )
    ap.add_argument(
        "--answer",
        action="append",
        default=[],
        help=(
            "Scripted clarification answer(s); repeatable. Plain entries are "
            "used positionally, in order (today's behavior). An entry of the "
            "form 'regex=>text' is instead matched against the actual "
            "question text — the first matching pattern's text wins "
            "regardless of round number (backlog #223)."
        ),
    )
    ap.add_argument(
        "--answer-mode",
        choices=("auto", "script", "llm"),
        default="auto",
        help=(
            "How to resolve a clarification answer when no --answer regex "
            "pattern matches the question. 'auto' (default) asks a cheap "
            "LLM to answer in the voice of the human requester, silently "
            "falling back to the positional scripted answer if the LLM "
            "answerer can't be built or its call fails. 'script' never "
            "calls the LLM (patterns + positional only — today's "
            "behavior). 'llm' requires a working LLM answerer and exits "
            "with an error instead of falling back if one can't be built."
        ),
    )
    ap.add_argument(
        "--work-type",
        default="feature",
        help=(
            "Pipeline work type. 'feature' (default) drives the interactive "
            "kernel PipelineRunner relay; any other value (analysis, review, "
            "documentation, refactor, ...) drives the orchestrator "
            "PipelineRunner's routed stage list (PipelineRouter), matching the "
            "LLM-lane e2e tests. Dogfood F8: the kernel relay has no work-type "
            "concept, so non-feature intents were misrouted down the full "
            "feature chain."
        ),
    )
    args = ap.parse_args(argv)

    # Parse --answer entries into regex patterns + plain positional answers
    # (backlog #223), before anything else runs — an invalid regex must
    # fail fast, like any other argparse error. A --answer list consisting
    # entirely of regex=>text pairs leaves no positional entries, so fall
    # back to the canonical scripted defaults (today's no--answer behavior).
    patterns, positional_answers = _parse_answer_specs(args.answer, on_error=ap.error)
    positional_answers = positional_answers or _DEFAULT_ANSWERS

    llm_answer_fn: Callable[[str], str] | None = None
    if args.answer_mode != "script":
        try:
            llm_answer_fn = _llm_answer_fn(args.intent)
        except Exception as exc:
            if args.answer_mode == "llm":
                ap.error(
                    "--answer-mode llm requires a working LLM answerer, but "
                    f"construction failed: {exc}"
                )
            else:
                print(
                    f"[warn] could not build the LLM answerer ({exc}); "
                    "falling back to scripted answers",
                    flush=True,
                )

    ts = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
    run_dir = Path("logs/e2e-runs") / f"{ts}-{args.slug}"
    run_dir.mkdir(parents=True, exist_ok=True)

    applied_env = _setup_env(run_dir, args.model, profile=args.profile)

    # Mirror console to console.log (real-time) for a human-readable record.
    console_fh = open(run_dir / "console.log", "w", encoding="utf-8")  # noqa: SIM115
    sys.stdout = _Tee(sys.__stdout__, console_fh)
    sys.stderr = _Tee(sys.__stderr__, console_fh)

    # Imports AFTER env is set (config binds env at import time).
    from orchestrata_ultimate.core.logging import (
        configure_logging,
        new_correlation_id,
        session_id,
    )

    configure_logging(
        level="DEBUG", log_file=str(run_dir / "app.jsonl"), json_format=True
    )
    cid = new_correlation_id()

    from orchestrata_ultimate.kernel.pipeline import PipelineRunner

    try:

        print("=" * 72)
        print(f"  E2E run · {run_dir}")
        print(f"  intent : {args.intent!r}")
        print(f"  work   : {args.work_type}")
        print(f"  model  : {args.model} (provider=deepseek)  correlation_id={cid}")
        print("=" * 72, flush=True)

        if args.work_type != "feature":
            # Routed path (dogfood F8): the kernel relay above has no work-type
            # concept, so analysis/review/documentation/refactor intents must go
            # through the orchestrator PipelineRunner + PipelineRouter, exactly
            # like the LLM-lane e2e tests (tests/orchestrator/test_pipeline_e2e_*).
            import json as _json

            from orchestrata_ultimate.orchestrator.artifact_store import ArtifactStore
            from orchestrata_ultimate.orchestrator.pipeline_runner import (
                PipelineRunner as RoutedRunner,
            )

            summary = {
                "started": datetime.now(UTC).isoformat(),
                "intent": args.intent,
                "slug": args.slug,
                "work_type": args.work_type,
                "correlation_id": cid,
                "env": applied_env,
                "status": "unknown",
            }

            async def _run_routed():
                store = ArtifactStore()
                routed = RoutedRunner(artifact_store=store)
                result = await routed.run(intent=args.intent, work_type=args.work_type)
                # Escalation-as-suspension (D1-addendum-2): a stage asking for
                # human input suspends the run ESCALATED. Resolve each round's
                # answer from the actual question text (backlog #223) and
                # re-enter, like the kernel path's interactive respond loop.
                rounds = 0
                while (
                    str(result.overall_status) == "escalated"
                    and rounds < args.max_rounds
                ):
                    question_text = _routed_question_text(result)
                    answer = resolve_answer(
                        question_text,
                        rounds,
                        patterns=patterns,
                        positional=positional_answers,
                        llm_answer_fn=llm_answer_fn,
                    )
                    print(
                        f"\n[escalation round {rounds}] -> respond: {answer[:120]}...",
                        flush=True,
                    )
                    result = await routed.respond(result.thread_id, answer)
                    rounds += 1
                summary["escalation_rounds"] = rounds
                return result

            t0 = datetime.now(UTC)
            try:
                result = asyncio.run(_run_routed())
                session_id.set(result.thread_id or "")
                summary["thread_id"] = result.thread_id
                summary["overall_status"] = str(result.overall_status)
                summary["stages"] = {
                    name: str(sr.status) for name, sr in result.stages.items()
                }
                summary["status"] = "completed"
                for art in result.artifacts:
                    name = f"artifact-{art.artifact_type}-{art.stage_alias}.md"
                    (run_dir / name).write_text(str(art.content), encoding="utf-8")
                    print(f"  [artifact] {name} ({len(str(art.content))} chars)")
                print(f"\n  stages : {summary['stages']}")
                print(f"  status : {summary['overall_status']}")
            except Exception as exc:  # pragma: no cover - dogfood harness
                summary["status"] = f"error: {exc}"
                raise
            finally:
                summary["ended"] = datetime.now(UTC).isoformat()
                summary["duration_s"] = (datetime.now(UTC) - t0).total_seconds()
                (run_dir / "summary.json").write_text(
                    _json.dumps(summary, indent=2, default=str), encoding="utf-8"
                )
                print(f"\n  artifacts: {run_dir}/", flush=True)
            return 0 if summary["status"] == "completed" else 1

        summary: dict = {
            "started": datetime.now(UTC).isoformat(),
            "intent": args.intent,
            "slug": args.slug,
            "correlation_id": cid,
            "env": applied_env,
            "rounds": [],
            "status": "unknown",
        }

        t0 = datetime.now(UTC)
        runner = PipelineRunner() if runner_factory is None else runner_factory()
        try:
            booted = runner.boot()
            print(f"[boot] {'ok' if booted else 'FAILED'}", flush=True)
            summary["booted"] = bool(booted)

            thread_id = runner.submit_intent(args.intent)
            summary["thread_id"] = thread_id
            session_id.set(thread_id or "")
            print(f"[intent submitted] thread={thread_id}", flush=True)

            rounds = 0
            final_subject = None
            while rounds < args.max_rounds:
                msg = runner.get_last_human_message()
                subject = getattr(msg, "subject", None) if msg else None
                text = (msg.payload.get("text", "") if msg else "")[:600]
                from_agent = getattr(msg, "from_agent", "?") if msg else "?"
                print(
                    f"\n[round {rounds}] <- {from_agent} :: {subject}\n  {text}",
                    flush=True,
                )
                summary["rounds"].append(
                    {
                        "round": rounds,
                        "from": from_agent,
                        "subject": subject,
                        "text": text,
                    }
                )
                final_subject = subject

                if subject in ("implementation_ready", "qa_complete", None):
                    break  # terminal (or nothing more to say)
                if subject == "clarification" or subject is not None:
                    answer = resolve_answer(
                        text,
                        rounds,
                        patterns=patterns,
                        positional=positional_answers,
                        llm_answer_fn=llm_answer_fn,
                    )
                    print(f"[round {rounds}] -> respond: {answer[:120]}...", flush=True)
                    runner.respond(thread_id or "", answer)
                rounds += 1

            summary["final_subject"] = final_subject
            summary["status"] = "completed"
        except Exception as exc:  # capture any pipeline error
            import traceback

            summary["status"] = "error"
            summary["error"] = f"{type(exc).__name__}: {exc}"
            (run_dir / "errors.log").write_text(
                traceback.format_exc(), encoding="utf-8"
            )
            print(
                f"\n[ERROR] {type(exc).__name__}: {exc}\n  (full traceback -> errors.log)",
                flush=True,
            )
        finally:
            # Dump the CXP packet flow from the Postman log.
            try:
                import json

                with open(run_dir / "cxp.jsonl", "w", encoding="utf-8") as fh:
                    for line in runner.postman.get_log():
                        fh.write(json.dumps({"entry": str(line)}) + "\n")
            except Exception as exc:
                print(f"[warn] could not dump CXP log: {exc}", flush=True)

            summary["ended"] = datetime.now(UTC).isoformat()
            summary["duration_s"] = (datetime.now(UTC) - t0).total_seconds()
            (run_dir / "summary.json").write_text(
                __import__("json").dumps(summary, indent=2, default=str),
                encoding="utf-8",
            )
            try:
                runner.shutdown(reason="e2e run complete")
            except Exception:
                pass

        print("\n" + "=" * 72)
        print(
            f"  status   : {summary['status']}  ({summary.get('duration_s', 0):.1f}s)"
        )
        print(f"  artifacts: {run_dir}/")
        for name in ("conversation.jsonl", "cxp.jsonl", "app.jsonl", "summary.json"):
            p = run_dir / name
            if p.exists():
                n = sum(1 for _ in p.open()) if name.endswith(".jsonl") else 1
                print(
                    f"    - {name:20s} {'(' + str(n) + ' lines)' if name.endswith('.jsonl') else ''}"
                )
        print("=" * 72, flush=True)
    finally:
        # Restore process-wide logging and stdout/stderr so that test clean-up
        # can safely delete the run directory without leaving a stale file
        # handler that later log emissions will try to reopen.
        sys.stdout = sys.__stdout__
        sys.stderr = sys.__stderr__
        try:
            console_fh.close()
        except Exception:
            pass
        try:
            configure_logging(level="INFO")
        except Exception:
            pass

    return 0 if summary["status"] == "completed" else 1


if __name__ == "__main__":
    sys.exit(main())