Preserve OSWorld pre-admission audit reach after extraction

Import the four pre-admission helpers from their defining owners so the
existing one-hop launcher audit can still inspect their bodies. Add a
source-only mutation test proving both the clean path and denied operation
remain visible, without changing the shared auditor.

Update the method location in the benchmark guide and remove duplicate
size thresholds and extraction-campaign residue. Existing behavioral tests
continue to cover claim confinement against the executing checkout.

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>

devtools/benchmarks/osworld/METHODOLOGY.md

@@ -113,7 +113,7 @@ finished "chat-style" left the Ouroboros `loop_outcome.final_answer` empty
 to `not_evaluated`, and several tasks lost reward because the agent answered
 in chat instead of leaving the VM in the evaluator-expected state (e.g.
 `is_expected_active_tab` checks the active URL, `compare_table` checks a saved
-`.xlsx`). Two structural fixes live in `run_step_agent.py`:
+`.xlsx`). Two structural fixes live in `step_agent_policy.py`:
 
 1. **Terminal-message capture.** The per-step JSON schema has a
    `final_answer` field; when the agent emits `done`/`fail` the runner

devtools/benchmarks/osworld/run_cu_bridge_agent.py

@@ -326,8 +326,8 @@ def main() -> int:
 
     # Guards: never drive the live desktop server or publish a bench connection
     # into the owner's live skill state (mirrors run_step_agent.py).
-    from devtools.benchmarks.osworld.run_step_agent import (
-        _is_default_desktop_server,
+    from devtools.benchmarks.osworld.step_agent_env import _is_default_desktop_server
+    from devtools.benchmarks.osworld.step_agent_claims import (
         confined_claims_dir,
         scored_claim_state,
         task_claim_key,

devtools/benchmarks/osworld/step_agent_claims.py

@@ -1,6 +1,6 @@
 """Cross-lane task claims and the scored-claim ledger for OSWorld runs.
 
-Verbatim extraction from ``run_step_agent.py`` (v7 stream W): claim-directory
+Verbatim extraction from ``run_step_agent.py``: claim-directory
 confinement, the claim key, staleness, acquisition and release, plus the
 durable unconfirmed/scored markers that make overlapping lanes, resumes and
 retry passes safe over one shared results tree (METHODOLOGY §7.9).

tests/test_osworld_cu_bridge_extraction.py

@@ -9,8 +9,11 @@ operators reach through `run_cu_bridge_agent.<name>`.
 from __future__ import annotations
 
 import ast
+import copy
 import pathlib
 
+import pytest
+
 from devtools.benchmarks.osworld import (
     cu_bridge_budget,
     cu_bridge_gate,
@@ -18,6 +21,8 @@ from devtools.benchmarks.osworld import (
     cu_bridge_runtime,
     cu_bridge_tool_policy,
     run_cu_bridge_agent as rcb,
+    step_agent_claims,
+    step_agent_env,
 )
 
 
@@ -119,17 +124,26 @@ def test_cu_bridge_launcher_reexports_every_moved_identity():
         assert getattr(rcb, name) is getattr(owner, name), name
 
 
-def test_cu_bridge_extraction_size_bounds_have_meaningful_headroom():
-    counts = {
-        path.name: len(path.read_text(encoding="utf-8").splitlines())
-        for path in (
-            OSWORLD / "run_cu_bridge_agent.py",
-            *(pathlib.Path(module.__file__) for module in _LEAVES),
-        )
-    }
-    assert counts["run_cu_bridge_agent.py"] < 1500
-    assert all(
-        count <= 1000
-        for name, count in counts.items()
-        if name != "run_cu_bridge_agent.py"
-    ), counts
+@pytest.mark.parametrize("helper,owner", [
+    ("_is_default_desktop_server", step_agent_env),
+    ("confined_claims_dir", step_agent_claims),
+    ("task_claim_key", step_agent_claims),
+    ("scored_claim_state", step_agent_claims),
+])
+def test_pre_admission_audit_reaches_step_helpers(helper, owner, monkeypatch):
+    from devtools.benchmarks.common import launcher_audit as audit
+
+    source = (OSWORLD / "run_cu_bridge_agent.py").read_text(encoding="utf-8")
+    assert audit.audit_source(source) == []
+    original_loader = audit._unit_for_module
+    tree = copy.deepcopy(original_loader(owner.__name__).tree)
+    function = next(node for node in tree.body
+                    if isinstance(node, ast.FunctionDef) and node.name == helper)
+    # Mutate only the source AST, never execute the operation or touch a file.
+    function.body.insert(1, ast.parse('write_text("before admission")').body[0])
+    mutant = audit._Unit(tree, owner.__name__)
+    monkeypatch.setattr(audit, "_unit_for_module", lambda name:
+                        mutant if name == owner.__name__ else original_loader(name))
+    violations = audit.audit_source(source)
+    assert any(f"{helper}() runs BEFORE" in item and "write_text" in item
+               for item in violations), violations

tests/test_osworld_step_agent_extraction.py

@@ -73,9 +73,9 @@ _MOVED_OWNERS = {
     "OuroborosStepAgent": step_agent_policy,
 }
 
-# The exact names run_cu_bridge_agent.py imports FROM run_step_agent.py. The
-# split must not force that importer to learn the new owners.
-_CU_BRIDGE_IMPORTS = (
+# Keep the shared runner surface even when consumers import defining owners
+# directly so the launcher audit can reach their bodies.
+_SHARED_HELPERS = (
     "_is_default_desktop_server", "confined_claims_dir", "scored_claim_state",
     "task_claim_key", "amend_task_manifest", "ClaimMarkerNotDurable",
     "acquire_task_claim", "claim_stale_sec", "construct_desktop_env",
@@ -124,8 +124,8 @@ def test_step_agent_launcher_reexports_every_moved_identity():
         assert getattr(rsa, name) is getattr(owner, name), name
 
 
-def test_cu_bridge_still_imports_its_shared_helpers_from_the_launcher():
-    for name in _CU_BRIDGE_IMPORTS:
+def test_step_launcher_preserves_its_shared_helper_surface():
+    for name in _SHARED_HELPERS:
         assert hasattr(rsa, name), name
 
 
@@ -150,14 +150,3 @@ def test_no_step_loop_module_fabricates_bash_history():
         assert "hist.open(" not in text, name
         assert "record_history" not in text, name
         assert ".bash_history'" not in text, name
-
-
-def test_step_agent_extraction_size_bounds_have_meaningful_headroom():
-    counts = {
-        path.name: len(path.read_text(encoding="utf-8").splitlines())
-        for path in (
-            OSWORLD / "run_step_agent.py",
-            *(pathlib.Path(module.__file__) for module in _LEAVES),
-        )
-    }
-    assert all(count <= 1000 for count in counts.values()), counts