Pytest Live Observer

The project's in-house pytest plugin: src/orchestrata_ultimate/core/threading/pytest_live_observer.py, with the observer-side CLI scripts/observe_pytest.py. It is mandatory on every test run (RUL-009 "force test observation") and is the single source of truth for per-test timing.

Why it exists

The batched executor (core/threading/test_executor.py) attributes per-module pass/fail from pytest's JUnit XML — which is written only at session end. When a batch hits its external timeout, the whole process group is SIGKILLed: no JUnit, every module in the batch reported "did not run" (even ones that already passed), and the hung module is anonymous. The plugin fixes this by writing one flushed JSON line per pytest event as it happens; flushed bytes survive a SIGKILL (we defend against a killed process, not power loss — no per-line fsync). test_executor.py reads this log only to rescue modules a missing/incomplete JUnit left behind — real pass/fail when the live log shows completion, or the exact in-flight nodeid (a dangling start with no terminal report) naming the hung test.

Three composed responsibilities

Events flow through a SyncObservable[dict] (core.patterns.observer) fanned out to composed observers:

Durability log (_NdjsonWriterObserver) — NDJSON to the path in PYTEST_LIVE_LOG (set by the batched executor). When unset (a bare pytest -p ... run) the plugin mints its own log under the registry dir, so every run is observable either way.
Session broadcast (_SessionRegistryObserver) — a JSON entry per session (pid, live-log path, running counts, status) in a well-known registry directory (session_registry_dir()). All registry I/O is best-effort (log-at-debug, never raise): a broadcast failure must never fail or slow the run it observes.
Duration sink (workflow-improvements Phase 3.3) — terminal-phase durations (call, or setup when it errored first — the same rule read_live_results uses for authoritative outcome) buffered per session and flushed as {test, duration, ts} NDJSON to reports/test-durations.ndjson at sessionfinish. scripts/test_durations.py is a pure reader; never add a second JUnit-parsing timing path. Env: PYTEST_DURATIONS_DISABLED skips the append; PYTEST_DURATIONS_PATH redirects it (tests use this).

Hooks implemented (vs. the pytest plugin API)

Standard pytest_-prefixed hook implementations (see pytest "Writing plugins" / "Writing hook functions" docs — hooks are discovered by name, args pruned to the declared signature, non-runtest hooks must not raise):

Hook	Event line
`pytest_sessionstart`	session start; builds the observable, registers the session
`pytest_runtest_logstart`	`start` (nodeid) — this is what names a hung test
`pytest_runtest_logreport`	`report` per phase (setup/call/teardown) with outcome + duration
`pytest_collectreport`	collection errors
`pytest_sessionfinish`	final counts; flushes durations; marks registry entry finished

Module-level state (_LIVE_LOG_PATH, _observable, _test_durations) is safe because each batch subprocess is a fresh pytest session (one subprocess per batch by design). DI seams (_LiveState, _MissingType sentinel, injectable session_registry_dir_factory) let unit tests exercise the hooks without monkeypatch (RUL-002).

Registration: `-p`, not entry point

pytest discovers plugins via (in order) -p name CLI options, pytest11 packaging entry points, PYTEST_PLUGINS, and conftest pytest_plugins. This plugin is deliberately loaded with the explicit CLI flag:

-p orchestrata_ultimate.core.threading.pytest_live_observer

There is no pytest11 entry point in pyproject.toml — loading is explicit-only, injected automatically by the suite runners (scripts/run_tests.py, run_tests_streaming.py, run_tests_threadpool.py) and demanded by the deny hook for ad-hoc runs. -p loading happens before conftests, which is what lets the plugin observe collection itself.

Enforcement (RUL-009)

.claude/hooks/force-test-observation.sh (PreToolUse deny hook) blocks any Bash pytest invocation that does not include pytest_live_observer, and blocks foreground waiting — runs must be backgrounded and then attached. There is no escape hatch. See RUL-009.

Observing a run

# canonical suite run (plugin auto-injected), backgrounded
python scripts/run_tests.py

# find sessions / attach live
uv run python scripts/observe_pytest.py list
uv run python scripts/observe_pytest.py attach latest          # follow until finish
uv run python scripts/observe_pytest.py attach <session_id> --once
uv run python scripts/observe_pytest.py attach latest --timeout 60

list prints one greppable key=value line per session (session_id, pid, status, counts, started_ago, live_log). attach tails the live NDJSON and renders PASS/FAIL/SKIP/ERROR lines plus a final summary. The observer never talks to the pytest process — files only — so it can attach to any session started anywhere (batched-executor subprocess or bare run) with no coordination.