[RFC 005] 3/4: HarnessEnvironment, subprocess helper, and MCP tool bridge

[splusq]

Stack for RFC 005 — 3 of 4. Depends on #1098.

Targets main because cross-fork PRs cannot chain bases, so the diff includes #1097 and #1098. Review the top commit only: f3d90c2.

1. 1/4 — package split ([RFC 005] 1/4: split openenv.core.harness into a package #1097)
2. 2/4 — foundation types ([RFC 005] 2/4: foundation types for agentic harnesses #1098)
3. 3/4 — this PR: the environment runtime
4. 4/4 — production /harness route + mode wiring

---

What

Makes the types from #1098 runnable: an environment that owns a harness subprocess, hands it the environment's MCP tools, and turns each step() into one conversational turn.

• environment.py — HarnessAction + HarnessEnvironment(MCPEnvironment). reset() stops any live harness, enumerates and conflict-resolves env tools, starts the bridge, injects, then starts the harness (injection strictly before start, per the RFC). step() runs one turn; MCP actions keep their normal routing. Rubrics run after the turn, outside the harness's control loop — RFC 004's reward boundary.
• process.py — HarnessProcess: readiness-gated start, stderr-tail diagnostics, idempotent stop with SIGTERM → SIGKILL escalation over the process group.
• bridge.py — HarnessMCPBridge: serves the env's FastMCP tool surface over loopback HTTP for the harness to consume.

Three decisions I would like challenged

1. HarnessEnvironment subclasses MCPEnvironment, substituting an empty internal FastMCP when mcp=None.
The RFC writes mcp: Optional[FastMCP] = None with a conditional super().__init__(mcp), which cannot work: mcp_server is a required positional, and skipping super().__init__ would skip Environment.__init__ entirely (no rubric, no transform). Substituting an empty server keeps reserved-name validation, _async_handle_list_tools() for injection, and mcp_session() integration for free. Alternative considered: subclass Environment directly and hold a FastMCP — rejected, it means reimplementing all of the above.

2. subprocess.Popen + reader threads, not asyncio.create_subprocess_exec.
Asyncio subprocess transports are bound to their creating loop. This object has to survive three regimes: the server's long-lived loop, the sync facade (run_async_safely spins a fresh loop per call, so a subprocess created in reset()'s loop would be unusable in step()'s), and close() from an executor thread. Popen plus asyncio.to_thread is loop-agnostic. Precedent: envs/julia_env/server/julia_process_pool.py.

3. The bridge is a separate loopback server, not a reuse of the env server's /mcp endpoint.
Reusing /mcp fails on two counts: it lives on the same app as /reset, /step, /state, so pointing the harness at that origin hands the agent the orchestration API — a direct violation of the RFC's security boundary and RFC 001's agents cannot reset; and WS /mcp calls _create_session(), so the harness would talk to a different env instance than the one wrapping it. A separate tool-only server on 127.0.0.1 makes the boundary structural rather than filter-based.

Failure handling

Turn timeouts and harness crashes become terminal observations (done=True, metadata.error_type of turn_timeout / harness_crashed) rather than exceptions, so a training loop scores the episode and moves on instead of unwinding. Deliberate; happy to change if reviewers prefer raising.

Known limitation

HTTP POST /reset and /step create a fresh env per request and close it, which would mean a harness subprocess per HTTP call. Harness envs must be driven over the WebSocket session path. Documented in the class docstring; a route-level guard is a candidate follow-up.

Verification

76 passed

Real subprocesses (no mocks) for the lifecycle edge cases: startup timeout, immediate exit reporting code + stderr tail, crash mid-session, double-stop idempotency, and a child that ignores SIGTERM to prove kill escalation lands within the grace window. Real loopback HTTP for the bridge, connecting an actual fastmcp.Client to list and call a tool. Environment tests cover reset ordering, conflict resolution, rubric ordering, crash/timeout paths, sync-facade parity, and that overrides_method sees the async overrides (which is what makes the server pick the async path). Lint clean.

Subprocess behaviors live in tests/core/scripted_harness.py — a real, lintable module spawned by mode (echo, slow-start, exit-now, crash-after-echo, ignore-sigterm), rather than code embedded in string literals.

---

Note

Medium Risk
New subprocess and loopback MCP surface area affects process lifecycle and tool exposure boundaries; rollout API split is mitigated by re-exports and back-compat tests.

Overview
Splits the monolithic openenv.core.harness package: trainer rollout APIs move to rollout.py with the same symbols re-exported from the package root (collect now imports from .rollout). Adds the RFC 005 turn-based runtime so external agent harnesses can run inside an env container.

HarnessEnvironment wraps an AgenticHarnessAdapter: reset() stops any prior harness, conflict-resolves env MCP tools, starts a loopback HarnessMCPBridge (tool-only HTTP on 127.0.0.1, separate from orchestration /reset//step), injects tools, then starts the harness. Each HarnessAction step() is one conversational turn with events in observation metadata; rubrics run after the turn. Crashes and turn timeouts become terminal observations (done=True, error_type) instead of bubbling exceptions.

HarnessProcess manages long-lived CLI harness subprocesses via Popen + reader threads (loop-agnostic vs asyncio subprocess). build_bridge_server serves renamed tools after resolve_tool_conflicts. Foundation pieces from earlier stack commits (adapter, config, events, tools) are wired here.

Extensive integration tests cover real subprocess lifecycle, loopback MCP clients, reset ordering, and package back-compat.

^{Reviewed by Cursor Bugbot for commit a2bf6f6. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Stale Bugbot comment from a previous run.

[caiotheodoro]

src/openenv/core/harness/process.py:221 — bug: read_line(timeout_s=None) doesn't do what the docstring promises ("None blocks until a line or EOF"). The reader thread's for line in stream: sink(line) (line 253) just returns when the pipe closes — no sentinel goes into the queue — so self._stdout_queue.get(timeout=None) (line 236) blocks forever past real EOF.

Repro: start a process that prints one line and exits, drain that line, then call read_line(timeout_s=None) again.

first read_line -> hello
is_running() -> False
calling read_line(timeout_s=None) on an exited process...
BUG CONFIRMED: read_line(timeout_s=None) hung for 5s past EOF instead of returning None
STILL RUNNING AFTER 12s

Wrapping the call in asyncio.wait_for lets the caller move on, but the to_thread worker stays parked in queue.get() — it isn't one of the daemon _reader_threads that stop() joins, so it outlives stop() and (being a non-daemon default-executor thread) blocks the interpreter from exiting.

Nothing in this PR calls read_line with timeout_s=None yet, but it's public API on a class future harness adapters will drive, and the contract as documented is false. Have the reader push a sentinel (or close the queue) on stream EOF and have read_line return None for it instead of relying on a bare queue.get().

[bot-ci-comment]

The docs for this PR live here. All of your documentation changes will be reflected on that endpoint. The docs are available until 30 days after the last update.

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 4ad1fb2. Configure here.}

[splusq]

@caiotheodoro, your EOF review is addressed by 1a6caa57, which is already included in this PR.

The stdout reader now enqueues an EOF sentinel from its finally block. read_line(timeout_s=None) therefore wakes and returns None once buffered output is drained, instead of leaving the executor worker blocked in queue.get(). The sentinel is reinserted so subsequent or concurrent readers also see EOF. Each process run owns a separate queue, so restarting does not mix old EOF notifications with new output.

Real-subprocess regression tests cover:

• Draining buffered lines, then repeatedly reading EOF with both timeout_s=None and a finite timeout.
• Stdout closing while the subprocess remains alive.
• stop() waking a pending unbounded read, followed by a restart that reads fresh output.

These cases passed in the latest 79-test harness run. The full repository test run also passed: 2,552 passed, 120 skipped.

[splusq]

@caiotheodoro can you please review again?