diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 05ec9437..11bf9d51 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -137,3 +137,14 @@ jobs:
             exit 1
           fi
           grep -q "1 passed" /tmp/drift.txt
+
+  opencollab:
+    name: OpenCollab integration
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - run: pip install -e ".[opencollab,dev]"
+      - run: python -m pytest tests/worker/agents/opencollab -q -m "not slow"
diff --git a/README.md b/README.md
index d25eea47..8f208c73 100644
--- a/README.md
+++ b/README.md
@@ -132,6 +132,8 @@ worker of step 3. The worker grades
 [SWE-rebench V2](https://github.com/SWE-rebench/SWE-rebench-V2) tasks with a port of its official
 harness.
 
+OpenCollab Single2 is an optional agent slot; see [OpenCollab setup and RL integration](docs/OPENCOLLAB.md).
+
 ## Learn more
 
 | Guide | What it covers |
diff --git a/docs/OPENCOLLAB.md b/docs/OPENCOLLAB.md
new file mode 100644
index 00000000..ce448be1
--- /dev/null
+++ b/docs/OPENCOLLAB.md
@@ -0,0 +1,134 @@
+# OpenCollab
+
+Sprout can run OpenCollab's Single2 agent through the same worker, sandbox,
+checkpoint, search and training-export interfaces as mini-swe-agent. The default
+slot remains `mini-swe-agent`. Select `opencollab-base` to run Single2.
+
+## Install
+
+```bash
+pip install -e ".[opencollab,harbor,runstore,rl,dev]"
+```
+
+The optional extra pins the public OpenCollab commit
+`c69a2076f4b54cc97891df432bf81273e3c5ebf1` (SDK version 0.9.3).
+It needs Git and network access at installation. The adapter does not modify the
+SDK. Without this extra, mini, oracle, nop and the service clients still import
+and run normally; SDK-dependent tests skip. CI installs the extra in a separate job.
+
+Use the normal host setup and services from [README](../README.md). Branching
+requires a snapshot backend, such as rootless Podman on a btrfs store.
+
+## Run and evaluate
+
+A direct run uses the worker's existing entry point:
+
+```bash
+export OPENAI_BASE_URL=http://127.0.0.1:8000/v1
+export OPENAI_API_KEY=local
+python -m sprout.worker run --slot opencollab-base --model MODEL \
+  --sandbox-image docker.io/library/python:3.11-slim --cwd /tmp \
+  --runtime-bin runtime/sprout-runtime --journal runs/opencollab.jsonl \
+  "Write and test fizzbuzz.py"
+```
+
+Evaluation goes through the Run Store and a running worker, as in
+[EVALUATION](EVALUATION.md). Add `--slot opencollab-base` to its existing command.
+Use the current upstream's fully prepared task images. For older Harbor bundles,
+prepare their declared environment setup and standard logging/artifact directories
+in the image before scheduling jobs; this adapter does not restore the removed
+Harbor trial runner.
+
+The task's workdir is resolved inside the sandbox before the agent starts, and
+recorded with its repository root. A branch restores that physical directory,
+not the host directory containing its logs.
+
+Single2 receives its native tools (`bash`, `file_read`, `file_write`, `apply_patch`,
+`git_diff`, `grep`). These are SDK tools backed by Sprout's `ToolAccess`; the Go
+runtime toolset is unchanged. Every environment operation runs in the owned
+sandbox, as the image's configured user. File writes use bounded, verified chunks.
+
+## Checkpoints and branches
+
+A recovery point pairs a completed runtime turn's disk snapshot with a
+checksummed native transcript prefix and the corresponding SDK session state.
+The worker restores the prefix and SDK state before executing the continuation.
+A different slot, protocol, checksum or workspace is rejected.
+
+The SDK uses non-streaming Chat Completions through Sprout's metered gateway.
+Native context shaping is recorded as the actual replacement context. Recovery
+point indexing only admits closed tool-call groups with matching snapshots.
+Cancellation waits for an owned runtime command to settle; unknown execution or
+checkpoint failure stops the run and remains an infrastructure failure.
+
+SPROUT still chooses branch points and guidance through its existing search and
+review jobs. Flat reviewer commands are serialized as the selected parent's
+actual command tool. Nested reviewer calls are validated against that parent's
+recorded schemas, with unique ids and no reviewer-authored submission.
+Both ordinary-content and reasoning-content continuations are supported.
+
+## RL messages
+
+In the RL driver's existing `miles.run_defaults`, select:
+
+```json
+{
+  "slot": "opencollab-base",
+  "extra": {
+    "mini": {"environment": {"cwd": "/workspace"}}
+  }
+}
+```
+
+`mini.environment.cwd` is the existing worker workspace configuration key,
+shared by these slots. The rest of the driver, worker profiles and task grader
+configuration follows [RL_DRIVER](RL_DRIVER.md). Set resources, endpoints and
+paths for your host. Search uses the current upstream request's `search` field
+and its documented variable-size group semantics; this adapter does not pad
+successful searches or change their stopping policy.
+
+The worker exports the SDK's actual messages and tool schemas, plus provenance:
+restored prefix messages, reviewer-authored assistant turns, and SDK-generated
+submission receipts. The existing driver excludes these from assistant loss.
+The sampled submission action remains a policy target. Real tool replies stay
+paired with their calls. Original task verifiers provide rewards; grader or
+transport failure is never fabricated as reward zero.
+
+Miles owns tokenization, assistant token masks, advantages, loss, optimization,
+checkpoints and serving-weight synchronization. No trainer is included here.
+Use a Miles client compatible with the current upstream driver's search request
+and result contract; older fixed-size Ash clients need their upstream update.
+
+## Historical tool compatibility
+
+For external trajectories recorded with `run_command`, `read_file`, `edit_file`,
+`write_file`, `get_status` and `submit_patch`, explicitly select
+`opencollab_tool_protocol: review50-v1` in the slot's `extra`. Evaluation exposes
+`--opencollab-tool-protocol review50-v1`. This compatibility mode still runs the
+Single2 session loop. It does not require or distribute the review50 dataset.
+
+Configure the actual checkout as the task workdir or `mini.environment.cwd`.
+Compatibility mode exposes that checkout at `/workspace` before the first model
+action; it refuses an existing alias pointing elsewhere.
+`submit_patch` produces a real diff receipt and records the SDK submission fact;
+it terminates through native Single2. The optional `/opt/review50-baseline`
+marker names the task's prepared Git baseline; otherwise the diff uses `HEAD`.
+Official grading remains separate. `get_status` reports the actual working-tree
+state, deadline and enforced model-turn count. Tool calls are counted and have
+no numeric budget limit; an absent limit is `null`.
+
+## Validation
+
+```bash
+python -m pytest tests/worker/agents/opencollab -q
+python -m pytest -q -m "not slow"
+./gate/run.sh
+python ops/contracts/ci_check.py
+```
+
+The integration tests exercise real Single2 sessions against a scripted model
+and an isolated shell fixture: multi-tool turns, snapshots, native restoration,
+reviewer injection, context shaping, submission, training provenance, large
+writes, cancellation and infrastructure failures. They do not claim policy
+learning or benchmark performance. GPU, task images and snapshot storage are
+external prerequisites for live RL; none is required by ordinary unit tests.
diff --git a/docs/RL_DRIVER.md b/docs/RL_DRIVER.md
index 017ff897..125b9bcc 100644
--- a/docs/RL_DRIVER.md
+++ b/docs/RL_DRIVER.md
@@ -62,7 +62,7 @@ return 503. These fields need care:
 | --- | --- |
 | `driver_token_env` | The variable of the callers' bearer token. `null` turns authentication off. The example config sets `null`, so use it only for local access. |
 | `ledger` | Required. A relative path starts from the directory of the config file. |
-| `miles.run_defaults` | RunSpec fields for each root rollout. `slot` must be `mini-swe-agent`. `model` applies when the request names none. |
+| `miles.run_defaults` | RunSpec fields for each root rollout. `slot` is `mini-swe-agent` or `opencollab-base` ([OpenCollab](OPENCOLLAB.md)). `model` applies when the request names none. |
 | `miles.image_resources` | Required: the `cpu` and the `memory_mb` of each sandbox. |
 | `miles.tasks` | For each `task_id`, a `grade` job template without `snapshot_id`. The adapter refuses a request for a task without one. |
 | `miles.api_key_env` | The variable in the worker's environment that holds the key of the model endpoint. The worker profile's `worker_env` must set it. |
diff --git a/pyproject.toml b/pyproject.toml
index 2bb557a8..fc70adc2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -18,6 +18,7 @@ dependencies = [
 [project.optional-dependencies]
 # The agent and the trial runner the evaluations pin.
 mini = ["mini-swe-agent==2.4.6"]
+opencollab = ["opencollab @ git+https://github.com/RISE-X-Lab/OpenCollab.git@c69a2076f4b54cc97891df432bf81273e3c5ebf1"]
 harbor = ["harbor==0.22.0"]
 # The Run Store queue and the RL driver.
 runstore = ["psycopg2-binary>=2.9,<3", "fastapi>=0.110,<1", "uvicorn>=0.29,<1"]
@@ -29,6 +30,9 @@ docs = ["fonttools>=4", "uharfbuzz>=0.39"]
 [project.urls]
 Homepage = "https://github.com/dreamyang-liu/sprout"
 
+[tool.hatch.metadata]
+allow-direct-references = true
+
 [tool.hatch.build.targets.wheel]
 packages = ["src/sprout", "src/sprout_sandbox"]
 
@@ -61,6 +65,10 @@ warn_unreachable = true
 module = ["harbor.*"]
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = ["opencollab.*"]
+ignore_missing_imports = true
+
 [tool.pytest.ini_options]
 pythonpath = ["src", "."]
 testpaths = ["tests"]
diff --git a/resources/opencollab/review50_protocol.json b/resources/opencollab/review50_protocol.json
new file mode 100644
index 00000000..0ec423ad
--- /dev/null
+++ b/resources/opencollab/review50_protocol.json
@@ -0,0 +1,122 @@
+{
+  "name": "review50-v1",
+  "tools": [
+    {
+      "type": "function",
+      "function": {
+        "name": "run_command",
+        "description": "Execute a shell command inside the repository sandbox.\n\nYou can run any shell command (Python scripts, tests, git commands, etc.).\nThe command runs in the repository workspace. No network access.\n\nArgs:\n    command: Shell command to execute (e.g., \"pytest tests/\", \"python3 script.py\").",
+        "parameters": {
+          "type": "object",
+          "properties": {
+            "command": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "command"
+          ]
+        }
+      }
+    },
+    {
+      "type": "function",
+      "function": {
+        "name": "read_file",
+        "description": "Read the contents of a file from the repository workspace.\n\nSupports reading specific line ranges using start_line and end_line (1-indexed).\nTo avoid overwhelming the context window, large files are automatically truncated\nto a maximum number of lines (default: 150 lines).\n\nArgs:\n    filepath: Relative path within the workspace directory (e.g., \"fastapi/main.py\", \"pyproject.toml\").\n    start_line: Optional starting line number (1-indexed, inclusive).\n    end_line: Optional ending line number (1-indexed, inclusive).\n\nReturns:\n    Dictionary containing file content, line range, and truncation status.",
+        "parameters": {
+          "type": "object",
+          "properties": {
+            "filepath": {
+              "type": "string"
+            },
+            "start_line": {
+              "type": "integer"
+            },
+            "end_line": {
+              "type": "integer"
+            }
+          },
+          "required": [
+            "filepath"
+          ]
+        }
+      }
+    },
+    {
+      "type": "function",
+      "function": {
+        "name": "edit_file",
+        "description": "Replace a contiguous block of text in an existing file.\n\nMore token-efficient than write_file for targeted changes. Provide 2-4 lines of unique\nsurrounding context in old_string to ensure an unambiguous match.\n\nArgs:\n    filepath: Relative path to the target file.\n    old_string: Text block to replace.\n    new_string: Replacement text block.\n    allow_multiple: Replace all occurrences of old_string if True.\n\nReturns:\n    Dictionary containing file status, occurrences count, match strategy, and unified diff.",
+        "parameters": {
+          "type": "object",
+          "properties": {
+            "filepath": {
+              "type": "string"
+            },
+            "old_string": {
+              "type": "string"
+            },
+            "new_string": {
+              "type": "string"
+            },
+            "allow_multiple": {
+              "type": "boolean",
+              "default": false
+            }
+          },
+          "required": [
+            "filepath",
+            "old_string",
+            "new_string"
+          ]
+        }
+      }
+    },
+    {
+      "type": "function",
+      "function": {
+        "name": "write_file",
+        "description": "Create or overwrite a file in the workspace.\n\nArgs:\n    filepath: Relative path to the file to create or overwrite.\n    content: Full content of the file.",
+        "parameters": {
+          "type": "object",
+          "properties": {
+            "filepath": {
+              "type": "string"
+            },
+            "content": {
+              "type": "string"
+            }
+          },
+          "required": [
+            "filepath",
+            "content"
+          ]
+        }
+      }
+    },
+    {
+      "type": "function",
+      "function": {
+        "name": "get_status",
+        "description": "Check active budget details (time, tool calls) and active patch details.",
+        "parameters": {
+          "type": "object",
+          "properties": {}
+        }
+      }
+    },
+    {
+      "type": "function",
+      "function": {
+        "name": "submit_patch",
+        "description": "Capture the current working tree, including new files, as the final submission and end this agent session. The runtime returns a real acknowledgement. Official grading runs separately; submission does not assert that the fix passes.",
+        "parameters": {
+          "type": "object",
+          "properties": {}
+        }
+      }
+    }
+  ],
+  "system_prompt": "You are an expert software engineer working autonomously in a sandboxed Python repository at `/workspace`. Your goal is to resolve the issue in the problem statement with a minimal, correct change to the source code, verify it, and submit it.\n\n<ENVIRONMENT>\n* `/workspace` is a git checkout. Its current state is the baseline your patch is measured against.\n* The sandbox is offline and all dependencies are already installed. Do not install packages.\n* `run_command` runs each command in a fresh `/bin/bash -c` inside `/workspace`. The current directory, environment variables and activated environments do not carry over between calls, so start a command with `cd <dir> && ...` when it needs a specific directory, and chain dependent steps with `&&`.\n* A command is stopped after 300 seconds. Do not start servers or interactive programs that wait for input.\n* Only the first 5000 characters of a command's output are returned. When output may be long, narrow it with `grep`, `head`, `tail`, or quieter test flags such as `-q` and `--tb=short`.\n* `read_file` returns at most 150 lines or 10000 characters per call, along with `total_lines` and `is_truncated`. Use `start_line` and `end_line` to read the part you need.\n* `edit_file` replaces `old_string` with `new_string`. `old_string` must match the file and be unique unless `allow_multiple` is true, so include enough surrounding lines to identify the location.\n* `write_file` creates or overwrites a whole file. Use `edit_file` to change existing files.\n* File tool paths are relative to `/workspace`.\n</ENVIRONMENT>\n\n<RULES>\n* Do not modify the existing tests or the test configuration (`tests/`, `test_*.py`, `*_test.py`, `conftest.py`, `pytest.ini`, `setup.cfg`, `pyproject.toml`, `tox.ini`). Graders reset these files and run their own tests, so changes to them are discarded.\n* Make the smallest change to non-test source files that satisfies the problem statement. Keep existing behavior and public interfaces unless the issue asks for a change.\n* Base every conclusion on output you have actually seen. If an output was truncated, look at the missing part before relying on it.\n* When the fix is complete and verified, call `submit_patch`. It records the `git diff` of `/workspace`, including new files.\n</RULES>\n\n<WORKFLOW>\nWork through these phases, keeping each one as short as the task allows:\nPhase 1. READING: Restate the problem in clear terms. Note the error messages, names, file paths and expected behavior it mentions, and how to reproduce it.\nPhase 2. RUNNING: Find out how the project's tests are run (README, existing test files) and run the relevant ones.\nPhase 3. EXPLORATION: Locate the code involved with `grep` and `find`. List the candidate files and functions, and choose the most likely place to fix.\nPhase 4. TEST CREATION: Write a minimal script that reproduces the issue, and run it to confirm the failure.\nPhase 5. FIX ANALYSIS: State what is wrong, where it is, and how you will fix it.\nPhase 6. FIX IMPLEMENTATION: Make a minimal, focused change.\nPhase 7. VERIFICATION: Rerun the reproduction script, add edge cases, and run the existing tests for the code you changed.\nPhase 8. FINAL REVIEW: Re-read the problem statement, confirm every requirement is met and the related tests pass, then submit.\n</WORKFLOW>\n\n<PRACTICES>\n* Understand the relevant code before changing it, and fix the root cause rather than the symptom.\n* Combine related shell steps into one command to save tool calls, and prefer targeted searches (`grep -n`, filtered `find`) over reading whole directories.\n* Edit files in place instead of creating copies under new names.\n* Match the surrounding code style and avoid redundant comments.\n* If repeated attempts fail, step back, list the plausible causes, and test the most likely one first.\n</PRACTICES>\n\n<RUNTIME SUBMISSION CONTRACT>\nThe runtime provides the six tools listed in this conversation. A successful submit_patch call captures the current working tree and ends the session after a real tool acknowledgement. Do not commit or reset the working tree. Submission is separate from official grading and does not imply success. If you end with a final response without tools, the caller still grades the current working tree.\n</RUNTIME SUBMISSION CONTRACT>"
+}
diff --git a/src/sprout/contract/agents/__init__.py b/src/sprout/contract/agents/__init__.py
new file mode 100644
index 00000000..d050d557
--- /dev/null
+++ b/src/sprout/contract/agents/__init__.py
@@ -0,0 +1 @@
+"""Native agent conversation formats shared by clients and workers."""
diff --git a/src/sprout/contract/agents/history.py b/src/sprout/contract/agents/history.py
new file mode 100644
index 00000000..03c3a5f7
--- /dev/null
+++ b/src/sprout/contract/agents/history.py
@@ -0,0 +1,68 @@
+"""Dispatch native history operations without importing either agent SDK."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Any
+
+from sprout.contract import mini_history
+from sprout.contract.agents import opencollab
+
+
+def load_prefix(reference: Mapping[str, Any]) -> list[dict[str, Any]]:
+    """Read a checked prefix, using its recorded native format."""
+    with Path(reference["path"]).open("rb") as stream:
+        header = json.loads(stream.readline())
+    is_oc = header.get("format") == opencollab.FORMAT
+    if reference.get("slot") is not None and ((reference["slot"] == opencollab.SLOT) != is_oc):
+        raise ValueError("Native prefix slot differs from its recorded format")
+    if is_oc:
+        return opencollab.load_prefix(reference)
+    return mini_history.load_prefix(reference)
+
+
+def training_messages(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """The conversation represented by a native transcript, including SDK context shaping."""
+    if entries and entries[0].get("format") == opencollab.FORMAT:
+        return opencollab.training_messages(entries)
+    return mini_history.training_messages(entries)
+
+
+def materialize(
+    reference: Mapping[str, Any], cwd: Path, destination: Path, native_home: Path
+) -> dict[str, Any]:
+    """Copy a checked native prefix into an attempt-owned directory."""
+    if reference["slot"] != opencollab.SLOT:
+        return mini_history.materialize(reference, cwd, destination, native_home)
+    data = mini_history.read_prefix(reference)
+    load_prefix(reference)
+    destination.mkdir(parents=True, exist_ok=False)
+    path = destination / "prefix.jsonl"
+    path.write_bytes(data)
+    return {
+        "native_prefix": {"path": str(path.resolve()), "sha256": reference["sha256"], "slot": opencollab.SLOT}
+    }
+
+
+def validate_assistant_turn(
+    message: object,
+    *,
+    history: Sequence[Mapping[str, Any]] = (),
+    tools: list[dict[str, Any]] | None = None,
+    slot: str = "mini-swe-agent",
+    allow_reasoning: bool = False,
+) -> dict[str, Any]:
+    """Validate a reviewer turn against the selected actor's tool contract."""
+    if slot == opencollab.SLOT:
+        if tools is None:
+            raise ValueError("OpenCollab continuation requires recorded tool schemas")
+        return opencollab.validate_assistant_turn(
+            message, history=history, tools=tools, allow_reasoning=allow_reasoning
+        )
+    if slot != "mini-swe-agent":
+        raise ValueError("Unsupported native history slot")
+    return mini_history.validate_assistant_turn(
+        message, history=history, tools=tools, allow_reasoning=allow_reasoning
+    )
diff --git a/src/sprout/contract/agents/opencollab.py b/src/sprout/contract/agents/opencollab.py
new file mode 100644
index 00000000..7728612c
--- /dev/null
+++ b/src/sprout/contract/agents/opencollab.py
@@ -0,0 +1,96 @@
+"""OpenCollab native prefixes and reviewer turns, without importing the SDK."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from copy import deepcopy
+from pathlib import Path
+from typing import Any, cast
+
+SLOT = "opencollab-base"
+FORMAT = "sprout-opencollab-v1"
+
+
+def training_messages(rows: Any) -> list[dict[str, Any]]:
+    messages = []
+    for row in rows:
+        if row.get("type") == "oc.context":
+            messages = deepcopy(row["messages"])
+        elif row.get("type") == "oc.message":
+            messages.append(deepcopy(row["message"]))
+    return messages
+
+
+def load_prefix(reference: Any) -> list[dict[str, Any]]:
+    with Path(reference["path"]).open("rb") as stream:
+        data = stream.read(reference.get("byte_length", -1))
+    if not data.endswith(b"\n") or hashlib.sha256(data).hexdigest() != reference["sha256"]:
+        raise ValueError("OpenCollab native prefix checksum changed")
+    rows = [json.loads(line) for line in data.splitlines() if line.strip()]
+    if (
+        not rows
+        or rows[0].get("format") != FORMAT
+        or rows[0].get("profile") != "single2"
+        or rows[-1].get("type") != "oc.turn"
+    ):
+        raise ValueError("OpenCollab restore requires a closed Single2 turn")
+    return rows
+
+
+def validate_assistant_turn(
+    message: object, *, history: Any = (), tools: list[dict[str, Any]], allow_reasoning: bool = False
+) -> dict[str, Any]:
+    """Validate new reviewer calls against the recorded schemas, preserving mini's separate contract."""
+    from jsonschema import Draft202012Validator, ValidationError
+
+    fields = {"role", "content", "tool_calls"}
+    if not isinstance(message, dict) or set(message) not in (
+        fields,
+        fields | {"reasoning_content"} if allow_reasoning else fields,
+    ):
+        raise ValueError("assistant_turn requires content and tool calls")
+    content = message["content"]
+    reasoning = message.get("reasoning_content")
+    if message["role"] != "assistant" or not isinstance(content, str):
+        raise ValueError("assistant_turn requires assistant role and text content")
+    if reasoning is not None and (
+        not allow_reasoning or not isinstance(reasoning, str) or not reasoning.strip() or "\n" in content
+    ):
+        raise ValueError("Invalid reviewer reasoning content")
+    if not content.strip() and not reasoning:
+        raise ValueError("assistant_turn requires non-empty content or reasoning")
+    calls = message["tool_calls"]
+    if not isinstance(calls, list) or not calls:
+        raise ValueError("assistant_turn requires tool calls")
+    schemas = {tool["function"]["name"]: tool["function"]["parameters"] for tool in tools}
+    used = {call["id"] for row in history for call in row.get("tool_calls") or []}
+    for call in calls:
+        if (
+            not isinstance(call, dict)
+            or set(call) != {"id", "type", "function"}
+            or call["type"] != "function"
+        ):
+            raise ValueError("Invalid reviewer tool call")
+        if not isinstance(call["id"], str) or not call["id"].strip() or call["id"] in used:
+            raise ValueError("Reviewer tool call ids must be unique across history")
+        used.add(call["id"])
+        function = call["function"]
+        if (
+            not isinstance(function, dict)
+            or set(function) != {"name", "arguments"}
+            or function["name"] not in schemas
+        ):
+            raise ValueError("Reviewer call names an unavailable Single2 tool")
+        if not isinstance(function["arguments"], str):
+            raise ValueError("Reviewer tool arguments must be a JSON string")
+        arguments = json.loads(function["arguments"])
+        schema = deepcopy(schemas[function["name"]])
+        schema["additionalProperties"] = False
+        try:
+            Draft202012Validator(schema).validate(arguments)
+        except ValidationError as error:
+            raise ValueError(
+                "Reviewer arguments violate the recorded tool schema: " + error.message
+            ) from error
+    return cast(dict[str, Any], deepcopy(message))
diff --git a/src/sprout/contract/continuation.py b/src/sprout/contract/continuation.py
index 2ca31423..8be6a979 100644
--- a/src/sprout/contract/continuation.py
+++ b/src/sprout/contract/continuation.py
@@ -26,7 +26,7 @@
 from dataclasses import dataclass
 from typing import Any
 
-from sprout.contract.mini_history import validate_assistant_turn
+from sprout.contract.agents.history import validate_assistant_turn
 
 #: The values of ``Continuation.guidance``.
 GUIDANCE = ("assistant-turn", "user-hint", "none")
@@ -154,8 +154,17 @@ def prepare(
     """
     # Continuation ties each field to its guidance: assistant_turn to assistant-turn, hint to user-hint.
     if continuation.assistant_turn is not None:
-        history = [entry["message"] for entry in prefix if entry.get("type") == "mini.message"]
-        turn = validate_assistant_turn(continuation.assistant_turn, history=history, tools=tools)
+        from sprout.contract.agents.history import training_messages
+
+        history = training_messages([dict(entry) for entry in prefix])
+        if prefix and prefix[0].get("type") == "oc.session":
+            tools = tools or prefix[0]["tools"]
+        turn = validate_assistant_turn(
+            continuation.assistant_turn,
+            history=history,
+            tools=tools,
+            slot="opencollab-base" if prefix and prefix[0].get("type") == "oc.session" else "mini-swe-agent",
+        )
         if continuation.assistant_turn_rendering == "reasoning":
             turn = render_reasoning(turn)
         return "", {"assistant_turn": turn}
diff --git a/src/sprout/contract/run_spec.py b/src/sprout/contract/run_spec.py
index 5f689071..3c0d9553 100644
--- a/src/sprout/contract/run_spec.py
+++ b/src/sprout/contract/run_spec.py
@@ -15,8 +15,8 @@ class RunSpec:
     """Everything one run needs. ``python -m sprout.worker run`` sets the fields that a person types."""
 
     prompt: str
-    #: The harness that a ``mini`` agent runs in. ``mini-swe-agent`` is the only one. The runner
-    #: refuses any other name before it makes a sandbox.
+    #: The model agent implementation: ``mini-swe-agent`` or ``opencollab-base``.
+    #: The runner refuses other names before it makes a sandbox.
     slot: str = "mini-swe-agent"
     #: What the run does in its sandbox: ``mini`` runs the model in ``slot``,
     #: ``oracle`` runs the ``solution/solve.sh`` of ``harbor_task``, ``nop`` does nothing.
diff --git a/src/sprout/eval/evaluate.py b/src/sprout/eval/evaluate.py
index 79e992d1..9c1be3a5 100644
--- a/src/sprout/eval/evaluate.py
+++ b/src/sprout/eval/evaluate.py
@@ -327,6 +327,8 @@ def _leak_judge(args: argparse.Namespace) -> dict[str, Any]:
 
 def _actor(args: argparse.Namespace) -> Actor:
     return Actor(
+        slot=getattr(args, "slot", "mini-swe-agent"),
+        tool_protocol=getattr(args, "opencollab_tool_protocol", "single2"),
         model=args.model,
         model_endpoint=args.model_endpoint,
         api_key_env=args.model_key_env,
@@ -709,6 +711,8 @@ def _add_runstore_options(parser: argparse.ArgumentParser) -> None:
 
 
 def _add_actor_options(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("--slot", choices=("mini-swe-agent", "opencollab-base"), default="mini-swe-agent")
+    parser.add_argument("--opencollab-tool-protocol", choices=("single2", "review50-v1"), default="single2")
     parser.add_argument("--model", required=True)
     parser.add_argument("--model-endpoint", required=True)
     parser.add_argument(
diff --git a/src/sprout/eval/jobs.py b/src/sprout/eval/jobs.py
index 55884c7e..ca92ad40 100644
--- a/src/sprout/eval/jobs.py
+++ b/src/sprout/eval/jobs.py
@@ -44,6 +44,8 @@ class Actor:
     max_turns: int
     command_timeout_s: int
     reasoning_effort: str | None = None
+    slot: str = "mini-swe-agent"
+    tool_protocol: str = "single2"
 
 
 @dataclass(frozen=True)
@@ -155,9 +157,20 @@ def mini_job(
     job = rollout_job(task, agent="mini", profile=profile, checkpoint=checkpoint, image=image)
     spec = job["spec"]
     spec["model"] = actor.model
+    spec["slot"] = actor.slot
     spec["extra"] = {
         "rollout_contract": _contract(actor, deadline_at=now + spec["timeout_s"] + QUEUE_ALLOWANCE_S),
-        "mini": {"environment": {"timeout": actor.command_timeout_s, **_workdir(task)}},
+        "mini": {
+            "environment": {
+                "timeout": actor.command_timeout_s,
+                **(
+                    {"cwd": task.workdir or "/workspace"}
+                    if actor.slot == "opencollab-base" and actor.tool_protocol == "review50-v1"
+                    else _workdir(task)
+                ),
+            }
+        },
+        **({"opencollab_tool_protocol": actor.tool_protocol} if actor.slot == "opencollab-base" else {}),
     }
     return job
 
diff --git a/src/sprout/rl_driver/messages.py b/src/sprout/rl_driver/messages.py
index 1bb35f36..18cd74de 100644
--- a/src/sprout/rl_driver/messages.py
+++ b/src/sprout/rl_driver/messages.py
@@ -69,12 +69,12 @@ def validate_endpoint(value: str) -> None:
 def _check_run_defaults(defaults: object) -> None:
     """Raise ValueError unless ``defaults``, the ``run_defaults`` of a Miles section, can go into a rollout.
 
-    It must be an object whose ``slot`` is ``mini-swe-agent``, and whose ``extra`` is an object without
+    Its ``slot`` is ``mini-swe-agent`` or ``opencollab-base``, and its ``extra`` is an object without
     ``rollout_contract``. Either key can be absent."""
     if not isinstance(defaults, dict):
         raise ValueError("run_defaults must be a JSON object")
-    if defaults.get("slot", "mini-swe-agent") != "mini-swe-agent":
-        raise ValueError("The Miles adapter supports mini-swe-agent only")
+    if defaults.get("slot", "mini-swe-agent") not in {"mini-swe-agent", "opencollab-base"}:
+        raise ValueError("The Miles adapter requires mini-swe-agent or opencollab-base")
     extra = defaults.get("extra", {})
     if not isinstance(extra, dict):
         raise ValueError("run_defaults.extra must be a JSON object")
diff --git a/src/sprout/runstore/api.py b/src/sprout/runstore/api.py
index 88be7631..d5791044 100644
--- a/src/sprout/runstore/api.py
+++ b/src/sprout/runstore/api.py
@@ -10,8 +10,8 @@
 from dataclasses import replace
 from typing import TYPE_CHECKING, Any
 
+from sprout.contract.agents.history import load_prefix
 from sprout.contract.continuation import Continuation, prepare
-from sprout.contract.mini_history import load_prefix
 from sprout.runstore.config import merge
 from sprout.runstore.index import Index
 from sprout.runstore.specs import (
diff --git a/src/sprout/runstore/specs.py b/src/sprout/runstore/specs.py
index 2086f28d..00dd9940 100644
--- a/src/sprout/runstore/specs.py
+++ b/src/sprout/runstore/specs.py
@@ -182,8 +182,8 @@ def _validate_run(spec: dict[str, Any]) -> None:
         raise ValueError("Unknown RunSpec fields")
     if not isinstance(spec.get("prompt"), str) or not spec["prompt"].strip():
         raise ValueError("RunSpec requires a nonempty prompt")
-    if spec.get("slot", "mini-swe-agent") != "mini-swe-agent":
-        raise ValueError("The only queued-run slot is mini-swe-agent")
+    if spec.get("slot", "mini-swe-agent") not in {"mini-swe-agent", "opencollab-base"}:
+        raise ValueError("Queued-run slot must be mini-swe-agent or opencollab-base")
     agent = spec.get("agent", "mini")
     if agent not in _RUN_AGENTS:
         raise ValueError("RunSpec agent must be mini, oracle or nop")
diff --git a/src/sprout/worker/agents/history.py b/src/sprout/worker/agents/history.py
new file mode 100644
index 00000000..d7d245de
--- /dev/null
+++ b/src/sprout/worker/agents/history.py
@@ -0,0 +1,90 @@
+"""Recovery points and tool schemas for the selected native agent slot."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from pathlib import Path
+from typing import Any
+
+from sprout.contract.mini_history import NativePoint
+from sprout.worker.agents.mini_swe import history as mini
+from sprout.worker.checkpoint.rollback import Checkpoint, branch_checkpoints
+from sprout.worker.run.journal import journal_events, read_journal
+
+
+def _slot(events: list[dict[str, Any]]) -> str | None:
+    slots = {e["slot"] for e in events if e.get("type") == "run.started" and e.get("slot")}
+    return next(iter(slots)) if len(slots) == 1 else None
+
+
+def index_native(
+    journal: str | Path,
+    transcript: str | Path,
+    slot: str,
+    session_id: str,
+    *,
+    events: list[dict[str, Any]] | None = None,
+    inherited_native: dict[str, Any] | None = None,
+) -> list[NativePoint]:
+    """Index only closed native turns that have matching disk checkpoints."""
+    if slot == "opencollab-base":
+        from sprout.worker.agents.opencollab.history import index_native as oc_index
+
+        return oc_index(journal, transcript, session_id, events=events, inherited_native=inherited_native)
+    return mini.index_native(
+        journal, transcript, slot, session_id, events=events, inherited_native=inherited_native
+    )
+
+
+def reference_at(journal: str | Path, step: int, session_id: str) -> dict[str, Any] | None:
+    """The exact native prefix of the checkpoint at tool depth ``step``."""
+    if _slot(read_journal(journal)) == "opencollab-base":
+        from sprout.worker.agents.opencollab.history import reference_at as oc_reference
+
+        return oc_reference(journal, step, session_id)
+    return mini.reference_at(journal, step, session_id)
+
+
+def conversation_restore(journal: str | Path, step: int, session_id: str) -> str | None:
+    """The native turn id at a supported actor's exact checkpoint."""
+    if _slot(read_journal(journal)) != "opencollab-base":
+        return mini.conversation_restore(journal, step, session_id)
+    reference = reference_at(journal, step, session_id)
+    return str(reference["cut"]) if reference else None
+
+
+def available_branch_points(journal: str | Path) -> dict[int, Checkpoint]:
+    """Only disk checkpoints whose complete conversation and SDK state can be restored."""
+    events = journal_events(journal)
+    if _slot(events) != "opencollab-base":
+        return mini.available_branch_points(journal)
+    references = {
+        e["native_session_id"]: e
+        for e in events
+        if e.get("type") == "session.ref" and e.get("transcript_path")
+    }
+    points = branch_checkpoints(journal, events=events)
+    valid: dict[int, Checkpoint] = {}
+    for session_id, ref in references.items():
+        for point in index_native(
+            journal, ref["transcript_path"], "opencollab-base", session_id, events=events
+        ):
+            if point.tool_depth in points:
+                valid[point.tool_depth] = points[point.tool_depth]
+    return valid
+
+
+def actor_tools_at(journal: str | Path, step: int) -> list[dict[str, Any]]:
+    """The actor's actual tool schemas at the selected checkpoint."""
+    events = read_journal(journal)
+    if _slot(events) != "opencollab-base":
+        return mini.actor_tools_at(journal, step)
+    tools = None
+    for event in events:
+        if event.get("type") == "rollout.model_tools" and event.get("shape") == "chat/completions":
+            tools = event.get("tools")
+        if event.get("type") == "model.turn.completed" and event.get("step") == step:
+            if not isinstance(tools, list) or not tools:
+                raise ValueError("No recorded actor tool schema at the selected checkpoint")
+            return deepcopy(tools)
+    raise ValueError("No completed turn for the selected actor tool schema")
diff --git a/src/sprout/worker/agents/mini_swe/sandbox.py b/src/sprout/worker/agents/mini_swe/sandbox.py
index 8057fdbb..5b86c014 100644
--- a/src/sprout/worker/agents/mini_swe/sandbox.py
+++ b/src/sprout/worker/agents/mini_swe/sandbox.py
@@ -13,7 +13,7 @@
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any, TypedDict
 
-from sprout.contract.mini_history import load_prefix
+from sprout.contract.agents.history import load_prefix
 from sprout.worker.agents.mini_swe.tools import DEFAULT_COMMAND_TIMEOUT_S
 from sprout.worker.run.control import RunControl
 from sprout.worker.run.journal import JournalWriter
@@ -60,10 +60,16 @@ def resolve_workspace(
     inherited: dict[str, Any] | None = None
     if extra.get("native_prefix"):
         entries = load_prefix(extra["native_prefix"])
-        metadata = next(entry for entry in entries if entry.get("type") == "mini.session")
+        metadata = next(entry for entry in entries if entry.get("type") in {"mini.session", "oc.session"})
         inherited = metadata.get("workspace")
     if inherited and configured not in (None, inherited["cwd"]):
-        raise ValueError("mini branch working directory differs from its inherited workspace")
+        if metadata.get("type") != "oc.session":
+            raise ValueError("mini branch working directory differs from its inherited workspace")
+        alias = session.execute(
+            "shell", {"command": "pwd -P", "working_dir": configured, "timeout": 10}, timeout=20
+        )
+        if _stdout(alias, "inherited working-directory alias") != inherited["cwd"]:
+            raise ValueError("OpenCollab branch working directory differs from its inherited workspace")
     requested = inherited["cwd"] if inherited else configured
     if requested is not None and (not isinstance(requested, str) or not requested.startswith("/")):
         raise ValueError("mini working directory must be an absolute path or omitted for discovery")
diff --git a/src/sprout/worker/agents/opencollab/__init__.py b/src/sprout/worker/agents/opencollab/__init__.py
new file mode 100644
index 00000000..ead44e91
--- /dev/null
+++ b/src/sprout/worker/agents/opencollab/__init__.py
@@ -0,0 +1 @@
+"""OpenCollab Base (Single2) on Sprout's owned sandbox and checkpoint path."""
diff --git a/src/sprout/worker/agents/opencollab/environment.py b/src/sprout/worker/agents/opencollab/environment.py
new file mode 100644
index 00000000..d0f262e3
--- /dev/null
+++ b/src/sprout/worker/agents/opencollab/environment.py
@@ -0,0 +1,175 @@
+"""OpenCollab's environment port over Sprout ToolAccess, never host execution."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import math
+import posixpath
+import shlex
+import time
+from contextlib import suppress
+from typing import Any
+from uuid import uuid4
+
+from opencollab.adapters._env_base import ENV_FILE_WRITE_LIMIT_BYTES
+from opencollab.adapters._env_docker import DockerEnvironment
+from opencollab.adapters.env import Environment, ExecResult
+
+from sprout.worker.run.control import RunAborted
+
+
+class SproutEnvironment(Environment):  # type: ignore[misc]  # SDK base has no published typing.
+    process_isolated = True
+    # These implementations use exec_cmd only, and retain native output limits.
+    read_text_range = DockerEnvironment.read_text_range
+    write_temp_file = DockerEnvironment.write_temp_file
+    remove_file = DockerEnvironment.remove_file
+
+    def __init__(
+        self, tools: Any, workspace: Any, control: Any, deadline: Any, *, image_shell: Any = None
+    ) -> None:
+        super().__init__()
+        self.tools, self.workspace = tools, workspace
+        self.control, self.deadline = control, deadline
+        self.image_shell = image_shell
+        self.turn_id: str | None = None
+        self.call_ids: list[str] = []
+        self.failure: BaseException | None = None
+        self._temporary_files: set[str] = set()
+
+    async def exec_cmd(self, cmd: str, timeout: float = 120.0) -> ExecResult:
+        self._ensure_active()
+        self.control.raise_if_stopped()
+        if not self.turn_id:
+            raise RuntimeError("OpenCollab execution requires an active model turn")
+        if not math.isfinite(timeout) or timeout <= 0:
+            raise ValueError("command timeout must be positive and finite")
+        remaining = self.deadline - time.monotonic()
+        if remaining <= 0:
+            self.control.request_stop("OpenCollab deadline exhausted", stop_reason="timeout")
+            self.control.raise_if_stopped()
+        identifier = uuid4().hex
+        arguments = {
+            "command": "bash -c " + shlex.quote(cmd),
+            "working_dir": self.workspace,
+            "timeout": max(1, math.ceil(min(timeout, remaining))),
+        }
+        if self.image_shell is not None:
+            arguments["command"] = self.image_shell.command(arguments["command"], working_dir=self.workspace)
+        worker = asyncio.create_task(
+            asyncio.to_thread(
+                self.tools.call,
+                "shell",
+                arguments,
+                call_id=identifier,
+                turn_id=self.turn_id,
+            )
+        )
+        try:
+            # Shield the serialized runtime operation until its own timeout settles it.
+            result = await asyncio.shield(worker)
+            self.call_ids.append(identifier)
+            outcome = result.outcome
+            if outcome is None or type(outcome.exit_code) is not int or outcome.running:
+                raise RuntimeError("OpenCollab command did not return a settled outcome")
+            self.control.raise_if_stopped()
+            return ExecResult(
+                outcome.exit_code,
+                outcome.stdout,
+                outcome.stderr,
+                stdout_truncated=outcome.stdout_truncated,
+                stderr_truncated=outcome.stderr_truncated,
+                timed_out=outcome.timed_out,
+            )
+        except asyncio.CancelledError:
+            # Drain a cancelled operation before another model turn may begin.
+            while not worker.done():
+                try:
+                    await asyncio.shield(worker)
+                except asyncio.CancelledError:
+                    continue
+                except Exception:
+                    break
+            try:
+                worker.result()
+            except Exception as error:  # the cancelled call really lost transport
+                self.failure = error
+            else:
+                self.call_ids.append(identifier)
+            raise
+        except RunAborted:
+            # A rollout deadline or controller stop that arrives after the
+            # command settled does not make the sandbox transport uncertain.
+            raise
+        except BaseException as error:
+            # OpenCollab's tools can convert exceptions to feedback. Retain the
+            # transport failure so that cannot turn it into a benchmark failure.
+            self.failure = error
+            raise
+
+    async def write_file(self, path: str, content: str) -> None:
+        raw = content.encode("utf-8")
+        if len(raw) > ENV_FILE_WRITE_LIMIT_BYTES:
+            raise ValueError("file exceeds the native OpenCollab write limit")
+        target = posixpath.normpath(path)
+        if not target.startswith("/") or target == "/":
+            raise ValueError("file path must be absolute and non-root")
+        parent, quoted = shlex.quote(posixpath.dirname(target)), shlex.quote(target)
+        temporary = shlex.quote(
+            posixpath.join(
+                posixpath.dirname(target),
+                ".sprout-write-" + uuid4().hex,
+            )
+        )
+
+        async def checked(command: str, phase: str) -> None:
+            result = await self.exec_cmd(command)
+            if result.returncode != 0:
+                raise OSError(
+                    f"sandbox file write failed ({phase}, exit={result.returncode}): " + result.stderr
+                )
+
+        # Bound each shell argument below Linux execve limits, even for large files.
+        owned = False
+        try:
+            await checked(
+                f"mkdir -p -- {parent} && (umask 077; set -C; : > {temporary})",
+                "create stage",
+            )
+            owned = True
+            for offset in range(0, len(raw), 32 * 1024):
+                payload = base64.b64encode(raw[offset : offset + 32 * 1024]).decode("ascii")
+                await checked(
+                    f"printf '%s' {shlex.quote(payload)} | base64 -d >> {temporary}",
+                    "append chunk",
+                )
+            digest = hashlib.sha256(raw).hexdigest()
+            await checked(
+                f'test "$(wc -c < {temporary})" -eq {len(raw)} && '
+                f"test \"$(sha256sum -- {temporary} | cut -d ' ' -f 1)\" = {digest} && "
+                f"if [ -f {quoted} ]; then chmod --reference={quoted} {temporary}; fi && "
+                f"mv -f -- {temporary} {quoted}",
+                "verify and commit",
+            )
+            owned = False
+        finally:
+            # Never mask cancellation or a transport failure; never bypass a
+            # stopped rollout to clean up. Harbor ultimately owns the sandbox.
+            if owned and self.failure is None:
+                with suppress(Exception, asyncio.CancelledError):
+                    await self.exec_cmd(f"rm -f -- {temporary}")
+
+    async def read_file(self, path: str) -> str:
+        """Read through ToolAccess without borrowing private Docker transport."""
+        result = await self.exec_cmd(f"cat -- {shlex.quote(path)}")
+        if result.returncode != 0:
+            raise FileNotFoundError(result.stderr)
+        if result.stdout_truncated:
+            raise OSError(f"sandbox read exceeded capture limit for {path}")
+        return str(result.stdout)
+
+    async def cleanup(self) -> None:
+        # Harbor owns the VM. Never delete it, or a restore point, here.
+        pass
diff --git a/src/sprout/worker/agents/opencollab/history.py b/src/sprout/worker/agents/opencollab/history.py
new file mode 100644
index 00000000..82f72574
--- /dev/null
+++ b/src/sprout/worker/agents/opencollab/history.py
@@ -0,0 +1,175 @@
+"""Exact Single2 session states paired with closed Sprout runtime turns."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from copy import deepcopy
+from pathlib import Path
+from typing import Any
+
+from sprout.contract.agents.opencollab import (
+    FORMAT,
+    SLOT,
+    training_messages,
+)
+from sprout.contract.mini_history import NativePoint
+from sprout.worker.checkpoint.rollback import branch_checkpoints, closed_turns
+from sprout.worker.run.journal import read_journal
+
+
+class History:
+    def __init__(
+        self,
+        path: Any,
+        session_id: Any,
+        workspace: Any,
+        prefix: Any = (),
+        *,
+        tool_protocol: Any = "single2",
+        tools: Any = (),
+    ) -> None:
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.stream = self.path.open("x", encoding="utf-8")
+        self.append(
+            {
+                "type": "oc.session",
+                "format": FORMAT,
+                "profile": "single2",
+                "session_id": session_id,
+                "workspace": workspace,
+                "tool_protocol": tool_protocol,
+                "tools": list(tools),
+            }
+        )
+        for row in prefix:
+            if row["type"] != "oc.session":
+                self.append(row)
+        self.messages = training_messages(prefix)
+
+    def append(self, row: Any) -> None:
+        self.stream.write(json.dumps(row, ensure_ascii=False) + "\n")
+        self.stream.flush()
+        os.fsync(self.stream.fileno())
+
+    def persist_messages(self, messages: Any) -> None:
+        if messages[: len(self.messages)] == self.messages:
+            for message in messages[len(self.messages) :]:
+                self.append({"type": "oc.message", "message": deepcopy(message)})
+        else:
+            # Native context shaping may rewrite observations or remove old
+            # turns. Persist that state, not a stale append-only conversation.
+            self.append({"type": "oc.context", "messages": deepcopy(messages)})
+        self.messages = deepcopy(messages)
+
+    def close(self) -> None:
+        self.stream.close()
+
+
+def index_native(
+    journal: Any, transcript: Any, session_id: Any, *, events: Any = None, inherited_native: Any = None
+) -> list[NativePoint]:
+    events = list(read_journal(journal)) if events is None else events
+    calls = [row["call_id"] for row in events if row.get("type") == "tool.started"]
+    positions = {identifier: index for index, identifier in enumerate(calls, 1)}
+    checkpoints = branch_checkpoints(journal, events=events)
+    closed = closed_turns(events)
+    points: list[NativePoint] = []
+    seen: set[str] = set()
+    checksum, offset, message_step = hashlib.sha256(), 0, 0
+    pending: set[str] = set()
+    used: set[str] = set()
+    for line in Path(transcript).read_bytes().splitlines(keepends=True):
+        if not line.endswith(b"\n"):
+            break
+        checksum.update(line)
+        offset += len(line)
+        row = json.loads(line)
+        if row.get("type") == "oc.session":
+            if row.get("format") != FORMAT:
+                raise ValueError("Unsupported OpenCollab native format")
+        elif row.get("type") == "oc.context":
+            pending, used = set(), set()
+            for message in row["messages"]:
+                identifiers = [call["id"] for call in message.get("tool_calls") or []]
+                if identifiers:
+                    if pending or len(set(identifiers)) != len(identifiers) or set(identifiers) & used:
+                        raise ValueError("Invalid shaped Single2 tool-call history")
+                    pending.update(identifiers)
+                    used.update(identifiers)
+                if message.get("role") == "tool":
+                    identifier = message.get("tool_call_id")
+                    if identifier not in pending:
+                        raise ValueError("Unpaired shaped Single2 tool result")
+                    pending.remove(identifier)
+        elif row.get("type") == "oc.message":
+            message = row["message"]
+            identifiers = [call["id"] for call in message.get("tool_calls") or []]
+            if identifiers:
+                if pending or len(set(identifiers)) != len(identifiers) or set(identifiers) & used:
+                    break
+                pending.update(identifiers)
+                used.update(identifiers)
+            if message.get("role") == "tool":
+                identifier = message.get("tool_call_id")
+                if identifier not in pending:
+                    break
+                pending.remove(identifier)
+        elif row.get("type") == "oc.turn":
+            if pending or not isinstance(row.get("session_snapshot"), dict):
+                break
+            message_step += 1
+            runtime_ids = row.get("runtime_call_ids", [])
+            local = [identifier for identifier in runtime_ids if identifier in positions]
+            if not local:
+                continue  # inherited turn or schema rejection: no local disk change
+            if len(local) != len(runtime_ids) or set(local) & seen:
+                break
+            seen.update(local)
+            depth = max(positions[identifier] for identifier in local)
+            turn = closed.get(row["turn_id"], {})
+            point = checkpoints.get(depth)
+            if turn.get("call_ids") != runtime_ids or turn.get("step") != depth or set(calls[:depth]) != seen:
+                break
+            if (
+                point
+                and point.snapshot_id is not None
+                and point.turn_id == row["turn_id"]
+                and point.session_ckpt == session_id
+            ):
+                points.append(
+                    NativePoint(
+                        message_step,
+                        depth,
+                        point.snapshot_id,
+                        {
+                            "slot": SLOT,
+                            "session_id": session_id,
+                            "path": str(Path(transcript).resolve()),
+                            "byte_length": offset,
+                            "sha256": checksum.hexdigest(),
+                            "cut": row["turn_id"],
+                        },
+                    )
+                )
+        else:
+            raise ValueError("Unknown OpenCollab native record")
+    return points
+
+
+def reference_at(journal: Any, step: Any, session_id: Any) -> dict[str, Any] | None:
+    events = list(read_journal(journal))
+    reference = next(
+        (
+            row
+            for row in reversed(events)
+            if row.get("type") == "session.ref" and row.get("native_session_id") == session_id
+        ),
+        None,
+    )
+    if reference is None:
+        return None
+    points = index_native(journal, reference["transcript_path"], session_id, events=events)
+    return next((point.native for point in points if point.tool_depth == step), None)
diff --git a/src/sprout/worker/agents/opencollab/review50.py b/src/sprout/worker/agents/opencollab/review50.py
new file mode 100644
index 00000000..52cbb5e8
--- /dev/null
+++ b/src/sprout/worker/agents/opencollab/review50.py
@@ -0,0 +1,108 @@
+"""Reviewed-data tool compatibility using OpenCollab's native execution loop."""
+
+from __future__ import annotations
+
+import json
+import shlex
+import time
+from copy import deepcopy
+from pathlib import Path
+from typing import Any, cast
+
+from opencollab.adapters.tools.base import Tool
+
+from sprout.resource import resource
+
+
+def definition() -> dict[str, Any]:
+    return cast(dict[str, Any], json.loads(resource("opencollab/review50_protocol.json").read_text()))
+
+
+class ReviewTool(Tool):  # type: ignore[misc]  # SDK base has no published typing.
+    """Keep the recorded tool names and parameter meanings at the SDK boundary."""
+
+    def __init__(self, schema: dict[str, Any], started: float) -> None:
+        function = deepcopy(schema["function"])
+        self.name = function["name"]
+        self.description = function["description"]
+        self.parameters = function["parameters"]
+        self.started = started
+        self.submission_receipt = None
+
+    async def execute_with_runtime(self, params: Any, runtime: Any) -> Any:
+        env = runtime.environment
+        env.agent_tool_calls = getattr(env, "agent_tool_calls", 0) + 1
+        if self.name == "run_command":
+            if runtime.safety_policy is not None:
+                runtime.safety_policy.check_cmd(params["command"])
+            result = await env.exec_cmd(params["command"], timeout=300)
+            runtime.observations.record_execution(exit_code=result.returncode, timed_out=result.timed_out)
+            if result.returncode or result.timed_out:
+                return json.dumps(
+                    {
+                        "status": "error",
+                        "error_type": "CommandError",
+                        "error_message": (result.stdout + result.stderr)[:5000],
+                        "returncode": result.returncode,
+                        "timed_out": result.timed_out,
+                    }
+                )
+            return json.dumps(
+                {
+                    "status": "ok",
+                    "stdout": result.stdout[:5000],
+                    "stderr": result.stderr[:5000],
+                    "is_truncated": result.stdout_truncated
+                    or result.stderr_truncated
+                    or len(result.stdout) > 5000
+                    or len(result.stderr) > 5000,
+                }
+            )
+        source = Path(__file__).with_name("review50_sandbox.py").read_text()
+        command = (
+            "python3 -c "
+            + shlex.quote(source)
+            + " "
+            + shlex.quote(self.name)
+            + " "
+            + shlex.quote(json.dumps(params))
+        )
+        result = await env.exec_cmd(command, timeout=300)
+        if result.returncode or result.timed_out or result.stdout_truncated:
+            raise RuntimeError("Review50 sandbox helper failed: " + result.stderr[:1000])
+        answer = json.loads(result.stdout)
+        if self.name == "get_status":
+            controls = getattr(env, "rollout_controls", None)
+            budget = (
+                controls.status()
+                if controls is not None
+                else {
+                    "model_calls": None,
+                    "max_turns": None,
+                    "remaining_turns": None,
+                    "runtime_tool_calls": None,
+                    "max_tool_calls": None,
+                    "remaining_seconds": max(0, env.deadline - time.monotonic()),
+                }
+            )
+            answer.update(
+                elapsed_seconds=time.monotonic() - self.started,
+                remaining_seconds=budget["remaining_seconds"],
+                budget={**budget, "agent_tool_calls": env.agent_tool_calls},
+                runtime_calls_in_turn=len(env.call_ids),
+                grading="pending official verifier",
+            )
+        if self.name in {"write_file", "edit_file"}:
+            runtime.observations.record_write(
+                completed=answer["status"] == "ok", changed=None, path=params["filepath"]
+            )
+        if self.name == "submit_patch" and answer["status"] == "submitted":
+            # This is OpenCollab's own submission fact, not a replacement stop loop.
+            runtime.observations.record_submission(json.dumps(answer))
+            self.submission_receipt = answer
+        return json.dumps(answer, ensure_ascii=False)
+
+
+def build_tools() -> tuple[Tool, ...]:
+    started = time.monotonic()
+    return tuple(ReviewTool(schema, started) for schema in definition()["tools"])
diff --git a/src/sprout/worker/agents/opencollab/review50_sandbox.py b/src/sprout/worker/agents/opencollab/review50_sandbox.py
new file mode 100644
index 00000000..27aaea82
--- /dev/null
+++ b/src/sprout/worker/agents/opencollab/review50_sandbox.py
@@ -0,0 +1,101 @@
+"""Review50 file operations executed inside the sandbox, using only stdlib."""
+
+from __future__ import annotations
+
+import difflib
+import hashlib
+import json
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+
+def operate(name: str, args: dict[str, Any]) -> dict[str, Any]:
+    root = Path.cwd().resolve()
+    if name in {"submit_patch", "get_status"}:
+        marker = Path("/opt/review50-baseline")
+        baseline = marker.read_text().strip() if marker.is_file() else "HEAD"
+
+        def git(*parts: Any) -> Any:
+            return subprocess.run(["git", *parts], check=True, capture_output=True, text=True).stdout
+
+        if name == "get_status":
+            status = git("status", "--porcelain=v1", "--untracked-files=all")
+            diff = git("diff", "--stat", baseline)
+            return {
+                "status": "ok",
+                "patch": {
+                    "baseline": baseline,
+                    "working_tree_status": status[:10000],
+                    "tracked_diff_stat": diff[:10000],
+                    "is_truncated": len(status) > 10000 or len(diff) > 10000,
+                },
+            }
+        # Include newly created files by the same mechanism as verifier.collect.
+        git("add", "-N", "--", ".")
+        patch = git("diff", "--binary", baseline)
+        return {
+            "status": "submitted",
+            "patch_sha256": hashlib.sha256(patch.encode()).hexdigest(),
+            "patch_chars": len(patch),
+            "baseline": baseline,
+            "grading": "pending official verifier",
+        }
+    path = (root / args["filepath"]).resolve()
+    try:
+        path.relative_to(root)
+    except ValueError:
+        raise ValueError("File paths must remain within the repository workspace") from None
+    result = {"status": "ok", "filepath": args["filepath"]}
+    if name == "read_file":
+        lines = path.read_text().splitlines(keepends=True)
+        start = args.get("start_line", 1)
+        end = args.get("end_line", start + 149)
+        if start < 1 or end < start:
+            raise ValueError("Line bounds must be positive, ordered and inclusive")
+        stop = min(end, start + 149, len(lines))
+        text = "".join(lines[start - 1 : stop])
+        result.update(
+            content=text[:10000],
+            start_line=start,
+            end_line=stop,
+            total_lines=len(lines),
+            is_truncated=stop < len(lines) or len(text) > 10000,
+        )
+    elif name == "write_file":
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(args["content"])
+        result["size"] = len(args["content"].encode())
+    elif name == "edit_file":
+        before = path.read_text()
+        old, new = args["old_string"], args["new_string"]
+        if not old:
+            raise ValueError("old_string must not be empty")
+        occurrences = before.count(old)
+        if occurrences == 0 or occurrences > 1 and not args.get("allow_multiple", False):
+            raise ValueError(f"old_string matches {occurrences} times; require one or allow_multiple")
+        after = before.replace(old, new)
+        path.write_text(after)
+        diff = "".join(
+            difflib.unified_diff(
+                before.splitlines(True),
+                after.splitlines(True),
+                fromfile="a/" + args["filepath"],
+                tofile="b/" + args["filepath"],
+            )
+        )
+        result.update(
+            occurrences=occurrences, strategy="exact", diff=diff[:10000], is_truncated=len(diff) > 10000
+        )
+    else:
+        raise ValueError(f"Unknown review50 operation: {name}")
+    return result
+
+
+if __name__ == "__main__":
+    try:
+        answer = operate(sys.argv[1], json.loads(sys.argv[2]))
+    except (OSError, ValueError, subprocess.CalledProcessError) as error:
+        answer = {"status": "error", "error_type": type(error).__name__, "error_message": str(error)}
+    print(json.dumps(answer, ensure_ascii=False))
diff --git a/src/sprout/worker/agents/opencollab/slot.py b/src/sprout/worker/agents/opencollab/slot.py
new file mode 100644
index 00000000..d54e8ca8
--- /dev/null
+++ b/src/sprout/worker/agents/opencollab/slot.py
@@ -0,0 +1,412 @@
+"""Drive the real Single2 session loop through Sprout-owned collaborators."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import threading
+import time
+from dataclasses import replace
+from importlib.metadata import version
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from opencollab.adapters.llm import LLMClient
+from opencollab.adapters.llm.types import LLMResponse
+from opencollab.adapters.llm.types import Usage as OCUsage
+from opencollab.application.session_run import GenerationTimeoutError
+from opencollab.bootstrap.agent_profiles import resolve_agent_profile
+from opencollab.bootstrap.runtime_context import build_workspace_safety_policy
+from opencollab.bootstrap.session_factory import build_session
+from opencollab.domain.agent import Agent
+
+from sprout.contract.agents.opencollab import load_prefix, training_messages, validate_assistant_turn
+from sprout.worker.agents.opencollab.environment import SproutEnvironment
+from sprout.worker.agents.opencollab.history import History
+from sprout.worker.agents.slot import AgentSlot, SlotResult, SlotStatus, TaskSpec
+from sprout.worker.gateway.server import _absorb_usage
+from sprout.worker.run.control import RunAborted, RunControl
+from sprout.worker.run.events import Usage
+from sprout.worker.run.journal import JournalWriter
+from sprout.worker.tools.access import ToolAccess
+
+
+class RecordedModel:
+    """Preserve native provider adaptation, recording usage and branch injection."""
+
+    def __init__(
+        self, client: Any, journal: Any, control: Any, pending: Any = None, client_factory: Any = None
+    ) -> None:
+        self.client, self.journal, self.control = client, journal, control
+        self.owner_loop = asyncio.get_running_loop()
+        self.client_factory = client_factory
+        self.pending = pending
+        self.usage = Usage()
+        self.last_usage: dict[str, Any] = {}
+        self.source = "actor"
+
+    def __getattr__(self, name: Any) -> Any:
+        return getattr(self.client, name)
+
+    async def complete(self, messages: Any, **kwargs: Any) -> Any:
+        self.control.raise_if_stopped()
+        self.source = "actor"
+        if self.pending is not None and kwargs.get("tools"):
+            turn, self.pending = self.pending, None
+            self.source = "reviewer"
+            self.last_usage = {}
+            self.journal.emit("branch.assistant_turn", source="reviewer", message=turn)
+            return LLMResponse(
+                content=turn["content"],
+                tool_calls=turn["tool_calls"],
+                usage=OCUsage(),
+                reasoning=turn.get("reasoning_content"),
+                finish_reason="tool_calls",
+            )
+        # Native read-time compaction runs in its own short-lived event loop.
+        # HTTP connection pools must be created and closed on that same loop.
+        if self.client_factory and asyncio.get_running_loop() is not self.owner_loop:
+            async with self.client_factory() as client:
+                response = await client.complete(messages, **kwargs)
+        else:
+            response = await self.client.complete(messages, **kwargs)
+        # Persist token consumption before the loop validates or executes it.
+        raw = response.usage.raw_usage or {
+            "prompt_tokens": response.usage.input_tokens,
+            "completion_tokens": response.usage.output_tokens,
+            "total_tokens": response.usage.total_tokens,
+        }
+        usage = Usage()
+        _absorb_usage(raw, usage)
+        self.usage.add(usage)
+        self.last_usage = usage.as_dict()
+        self.journal.emit(
+            "raw.opencollab-base",
+            response={
+                "usage": raw,
+                "choices": [
+                    {
+                        "message": {
+                            "role": "assistant",
+                            "content": response.content,
+                            "tool_calls": response.tool_calls,
+                        },
+                        "finish_reason": response.finish_reason,
+                    }
+                ],
+            },
+        )
+        self.journal.emit("usage.updated", usage=self.last_usage)
+        if response.content:
+            self.journal.emit("agent.message", text=response.content)
+        if response.reasoning:
+            self.journal.emit("agent.thinking", text=response.reasoning)
+        self.control.raise_if_stopped()
+        return response
+
+
+class TurnBridge:
+    def __init__(self, history: Any, tools: Any, env: Any, model: Any, journal: Any, control: Any) -> None:
+        self.history, self.tools, self.env, self.model = history, tools, env, model
+        self.journal, self.control = journal, control
+        self.session: Any = None
+        self.error: Exception | None = None
+
+    async def emit(self, event: Any) -> Any:
+        try:
+            self.journal.emit("opencollab.event", native_event_type=event.type, data=event.data)
+            if event.type == "step_start":
+                self.env.turn_id = uuid4().hex
+                self.env.call_ids = []
+                self.journal.emit("turn.started", turn=event.data["step"])
+            elif event.type == "tool_start":
+                # Native session has appended the complete assistant response;
+                # make that durable before any sandbox side effect begins.
+                self.history.persist_messages(self.session.messages)
+            elif event.type == "step_end":
+                if self.env.failure:
+                    raise RuntimeError("Single2 sandbox transport failed") from self.env.failure
+                self.history.persist_messages(self.session.messages)
+                _, metadata = self.session._snapshot_for_save()
+                metadata["session_state"]["phase"] = "precheck"
+                metadata["session_state"]["pending_events"] = []
+                metadata["session_state"]["pending_step_latency"] = None
+                self.history.append(
+                    {
+                        "type": "oc.turn",
+                        "turn_id": self.env.turn_id,
+                        "runtime_call_ids": list(self.env.call_ids),
+                        "source": self.model.source,
+                        "session_snapshot": metadata,
+                    }
+                )
+                if self.env.call_ids:
+                    self.journal.emit(
+                        "model.turn.completed",
+                        turn_id=self.env.turn_id,
+                        call_ids=list(self.env.call_ids),
+                        source=self.model.source,
+                        step=self.journal.tool_calls()[-1]["step"],
+                    )
+                    self.tools.close_turn(self.env.turn_id)
+                self.journal.emit(
+                    "turn.completed",
+                    turn=event.data["step"],
+                    source=self.model.source,
+                    usage=self.model.last_usage,
+                )
+        except Exception as error:  # noqa: BLE001 - checkpoint failure aborts its owner
+            # EventBus isolates subscriber failures. Fail the owning rollout
+            # explicitly so a lost history/checkpoint is never silently accepted.
+            self.error = error
+            cause = error.__cause__
+            fields = {"error_type": type(error).__name__, "error": str(error)}
+            if cause is not None:
+                fields.update(cause_type=type(cause).__name__, cause=str(cause))
+            self.journal.emit("opencollab.bridge.failed", **fields)
+            if self.env.failure is not None:
+                detail = "%s: %s" % (type(self.env.failure).__name__, self.env.failure)
+                self.control.request_stop(
+                    "Single2 sandbox transport failed: " + detail, stop_reason="infrastructure_error"
+                )
+            else:
+                self.control.request_stop(
+                    "Single2 checkpoint bridge failed: " + str(error), stop_reason="infrastructure_error"
+                )
+
+
+class OpenCollabSlot(AgentSlot):
+    name = "opencollab-base"
+
+    def version(self) -> str:
+        return version("opencollab")
+
+    def run(self, task: TaskSpec, journal: JournalWriter, tools: ToolAccess) -> SlotResult:
+        if tools is None or not task.model:
+            raise ValueError("OpenCollab requires ToolAccess and an explicit model")
+        return asyncio.run(self._run(task, journal, tools))
+
+    async def _run(self, task: Any, journal: JournalWriter, tools: ToolAccess) -> SlotResult:
+        control = task.control or RunControl()
+        started_at = time.monotonic()
+        deadline = started_at + task.timeout_s
+        profile = resolve_agent_profile("base")
+        if profile.name != "single2":
+            raise ValueError("This experiment pins Base to the Single2 implementation")
+        workspace = task.extra["mini_workspace"]
+        prefix = load_prefix(task.extra["native_prefix"]) if task.extra.get("native_prefix") else []
+        protocol = task.extra.get("opencollab_tool_protocol", "single2")
+        if protocol not in {"single2", "review50-v1"}:
+            raise ValueError("Unknown OpenCollab tool protocol")
+        if prefix and prefix[0].get("tool_protocol", "single2") != protocol:
+            raise ValueError("OpenCollab branch tool protocol differs from its parent")
+        if prefix and prefix[0]["workspace"] != workspace:
+            raise ValueError("Single2 branch workspace differs from the parent's")
+        inherited_messages = training_messages(prefix)
+        if protocol == "review50-v1":
+            from sprout.worker.agents.opencollab.review50 import build_tools, definition
+
+            profile = replace(profile, system_prompt=definition()["system_prompt"])
+            native_tools = build_tools()
+        else:
+            native_tools = profile.resolve_tools("coding")
+        # Use the runtime deadline; SDK timers can cancel a healthy queued command.
+        for tool in native_tools:
+            tool.disable_outer_timeout = True
+        schemas = [tool.to_openai_schema() for tool in native_tools]
+        pending = task.extra.get("assistant_turn")
+        if pending is not None:
+            if not prefix or task.extra.get("resume_without_hint"):
+                raise ValueError("Reviewer continuation requires an exact prefix")
+            pending = validate_assistant_turn(
+                pending, history=inherited_messages, tools=schemas, allow_reasoning=True
+            )
+        if task.extra.get("resume_without_hint") and not prefix:
+            raise ValueError("Point continuation requires an exact prefix")
+        session_id = uuid4().hex
+        home = Path(task.extra.get("native_home") or journal.path.parent / "opencollab-native")
+        home.mkdir(parents=True, exist_ok=True)
+        history = History(
+            home / f"{session_id}.jsonl", session_id, workspace, prefix, tool_protocol=protocol, tools=schemas
+        )
+        settings = dict(task.extra.get("mini", {}).get("model", {}).get("model_kwargs", {}))
+        wire_env = {**os.environ, **task.env}
+
+        def client_factory() -> Any:
+            return LLMClient(
+                model=task.model,
+                provider="openai",
+                wire_protocol="chat_completions",
+                base_url=wire_env.get("OPENAI_BASE_URL"),
+                api_key=wire_env.get("OPENAI_API_KEY"),
+                max_retries=5,
+                request_timeout=task.timeout_s,
+                stream_chat=False,
+                first_event_timeout=None,
+                stream_idle_timeout=None,
+                provider_error_time_budget=min(300, task.timeout_s),
+            )
+
+        injected = task.extra.get("opencollab_llm")
+        client = injected or client_factory()
+        model = RecordedModel(
+            client, journal, control, pending, client_factory=None if injected else client_factory
+        )
+        env = SproutEnvironment(tools, workspace["cwd"], control, deadline, image_shell=task.image_shell)
+        env.rollout_controls = task.extra.get("rollout_controls")
+        bridge = TurnBridge(history, tools, env, model, journal, control)
+        agent = Agent(
+            name="single2",
+            system_prompt=profile.system_prompt,
+            tools=list(native_tools),
+            model=task.model,
+            provider="openai",
+            max_tokens_per_step=settings.get("max_tokens", 64000),
+            reasoning_effort=settings.get("reasoning_effort", "high"),
+            temperature=settings.get("temperature", 0.0),
+            top_p=settings.get("top_p"),
+        )
+        session = build_session(
+            agent=agent,
+            env=env,
+            llm=model,
+            event_sink=bridge,
+            max_budget_tokens=None,
+            max_steps=None,
+            agent_profile=profile,
+            llm_timeout=None,
+            safety_policy=build_workspace_safety_policy(env),
+        )
+        bridge.session = session
+        journal.emit(
+            "run.started",
+            slot=self.name,
+            slot_version=self.version(),
+            agent_profile=profile.name,
+            tool_protocol=protocol,
+            model=task.model,
+            task_prompt=task.prompt,
+            cwd=task.cwd,
+            actor_timeout_s=task.timeout_s,
+            generation_timeout_s=None,
+        )
+        journal.emit("session.ref", native_session_id=session_id, transcript_path=str(history.path.resolve()))
+        journal.emit("rollout.model_tools", shape="chat/completions", tools=schemas)
+        timer = threading.Timer(
+            task.timeout_s,
+            lambda: control.request_stop("OpenCollab wall-time exhausted", stop_reason="timeout"),
+        )
+        timer.daemon = True
+        timer.start()
+        status: SlotStatus = "error"
+        error: str | None = None
+        text = ""
+        wall_timeout = asyncio.timeout_at(deadline)
+        try:
+            if prefix:
+                restored = home / f"{session_id}.restore.json"
+                session.store.save(str(restored), inherited_messages, meta=prefix[-1]["session_snapshot"])
+                session.restore(str(restored))
+                # Keep inherited state/history, grant only NEW turns to this run.
+                journal.emit(
+                    "opencollab.restored",
+                    inherited_native=task.extra["native_prefix"],
+                    inherited_steps=session.step_count,
+                    inherited_tokens=session.used_tokens,
+                )
+                if not pending and not task.extra.get("resume_without_hint") and task.prompt:
+                    await session.add_user_message(task.prompt)
+            else:
+                await session.add_user_message(task.prompt)
+            # The outer context identifies its own deadline; an inner provider
+            # TimeoutError must never masquerade as the whole rollout deadline.
+            async with wall_timeout:
+                text = await session.run_loop()
+            if env.failure:
+                raise RuntimeError("Single2 sandbox transport failed") from env.failure
+            if bridge.error:
+                raise bridge.error
+            control.raise_if_stopped()
+            if session.phase.value == "error":
+                raise RuntimeError(session.state.terminal_reason or "Single2 session failed")
+            status = "completed"
+        except RunAborted as exc:
+            if env.failure or bridge.error:
+                status, error = "error", str(env.failure or bridge.error)
+            else:
+                status = "timeout" if control.stop_reason in {"timeout", "max_turns_reached"} else "killed"
+                error = str(exc) or control.reason
+        except GenerationTimeoutError as exc:
+            error = f"{type(exc).__name__}: {exc}"
+            control.request_stop(error, stop_reason="infrastructure_error")
+            status = "error"
+        except TimeoutError as exc:
+            if wall_timeout.expired():
+                control.request_stop("OpenCollab wall-time exhausted", stop_reason="timeout")
+                status, error = "timeout", control.reason
+            else:
+                error = f"{type(exc).__name__}: {exc}"
+                control.request_stop(error, stop_reason="infrastructure_error")
+                status = "error"
+        except Exception as exc:  # noqa: BLE001 - the slot returns a durable outcome
+            if (
+                control.stop_reason in {"timeout", "max_turns_reached"}
+                and not env.failure
+                and not bridge.error
+            ):
+                status, error = "timeout", control.reason
+            else:
+                status, error = "error", f"{type(exc).__name__}: {exc}"
+        finally:
+            timer.cancel()
+            history.persist_messages(session.messages)
+            session.save(str(home / f"{session_id}.session.json"))
+            history.close()
+            await session.aclose()
+            close = getattr(client, "close", None) or getattr(client, "aclose", None)
+            if close:
+                result = close()
+                if asyncio.iscoroutine(result):
+                    await result
+        usage = model.usage.as_dict()
+        journal.emit("run.result", text=text)
+        submission = next(
+            (tool.submission_receipt for tool in native_tools if getattr(tool, "submission_receipt", None)),
+            None,
+        )
+        journal.emit(
+            "opencollab.termination",
+            tool_protocol=protocol,
+            submitted=submission is not None,
+            submission_receipt=submission,
+            kind=(
+                "explicit_submission"
+                if submission
+                else "final_response"
+                if status == "completed"
+                else control.stop_reason or status
+            ),
+        )
+        termination = (
+            "completed"
+            if status == "completed"
+            else "model_request_budget"
+            if status == "timeout" and control.stop_reason == "max_turns_reached"
+            else "actor_deadline"
+            if status == "timeout"
+            else "infrastructure_error"
+        )
+        journal.emit(
+            "run.finished",
+            status=status,
+            error=error,
+            usage=usage,
+            termination_kind=termination,
+            stop_reason=("infrastructure_error" if env.failure or bridge.error else control.stop_reason),
+            elapsed_s=time.monotonic() - started_at,
+            actor_timeout_s=task.timeout_s,
+        )
+        return SlotResult(
+            status=status, final_text=text, usage=dict(usage), native_session_id=session_id, error=error
+        )
diff --git a/src/sprout/worker/agents/opencollab/workspace.py b/src/sprout/worker/agents/opencollab/workspace.py
new file mode 100644
index 00000000..56b8688d
--- /dev/null
+++ b/src/sprout/worker/agents/opencollab/workspace.py
@@ -0,0 +1,32 @@
+"""Resolve the sandbox checkout and the historical protocol's workspace alias."""
+
+from __future__ import annotations
+
+import shlex
+from typing import Any
+
+from sprout.worker.agents.mini_swe.sandbox import resolve_workspace as resolve_native_workspace
+from sprout.worker.run.journal import JournalWriter
+from sprout.worker.sandbox.session import SandboxSession
+
+
+def resolve_workspace(
+    session: SandboxSession, extra: dict[str, Any], journal: JournalWriter
+) -> dict[str, Any]:
+    """Resolve a checkout; compatibility tools additionally expose it at ``/workspace``."""
+    resolved = resolve_native_workspace(session, extra, journal)
+    if resolved.get("opencollab_tool_protocol", "single2") != "review50-v1":
+        return resolved
+    target = resolved["mini_workspace"]["cwd"]
+    command = (
+        "if [ -e /workspace ] || [ -L /workspace ]; then "
+        f'test "$(cd /workspace && pwd -P)" = {shlex.quote(target)}; '
+        f"else ln -s -- {shlex.quote(target)} /workspace; fi"
+    )
+    result = session.execute("shell", {"command": command, "working_dir": target, "timeout": 10}, timeout=20)
+    if not result.success:
+        raise ValueError(
+            "OpenCollab /workspace alias is unavailable or points outside the configured checkout"
+        )
+    journal.emit("opencollab.workspace_alias", alias="/workspace", cwd=target)
+    return resolved
diff --git a/src/sprout/worker/attempts.py b/src/sprout/worker/attempts.py
index aeb9f054..42f7723b 100644
--- a/src/sprout/worker/attempts.py
+++ b/src/sprout/worker/attempts.py
@@ -28,7 +28,7 @@
 from sprout.runstore.payload import PayloadUnavailable, encode_payload, send_payload
 from sprout.runstore.specs import defaults_key, digest, validate_continuation
 from sprout.runstore.store import Store
-from sprout.worker.agents.mini_swe.history import index_native
+from sprout.worker.agents.history import index_native
 from sprout.worker.cleanup import LEFTOVERS_NAME, Leftovers, forget, log_section, with_log
 from sprout.worker.files import read_json, write_json
 from sprout.worker.kept import LEDGER_NAME, KeptSandboxes
diff --git a/src/sprout/worker/cli.py b/src/sprout/worker/cli.py
index f797ca77..fb207326 100644
--- a/src/sprout/worker/cli.py
+++ b/src/sprout/worker/cli.py
@@ -308,9 +308,9 @@ def _add_run_options(run: argparse.ArgumentParser) -> None:
     run.add_argument("prompt")
     run.add_argument(
         "--slot",
-        choices=("mini-swe-agent",),
+        choices=("mini-swe-agent", "opencollab-base"),
         default="mini-swe-agent",
-        help="the agent; mini-swe-agent is the only one",
+        help="the agent implementation",
     )
     run.add_argument("--cwd", default=".")
     run.add_argument("--model")
diff --git a/src/sprout/worker/jobs/export.py b/src/sprout/worker/jobs/export.py
index f90f0f97..03cfb21f 100644
--- a/src/sprout/worker/jobs/export.py
+++ b/src/sprout/worker/jobs/export.py
@@ -12,6 +12,7 @@
 from pathlib import Path
 from typing import Any
 
+from sprout.contract.agents.history import training_messages
 from sprout.contract.messages import (
     LIMIT_STOP_REASONS,
     branch_origin,
@@ -31,7 +32,7 @@ def export_messages(
     Raises ValueError without a session id, for a slot other than ``mini-swe-agent``, or
     unless exactly one history exists. ``allow_incomplete_tail`` drops an unfinished last record.
     """
-    if not isinstance(session_id, str) or not session_id or slot != "mini-swe-agent":
+    if not isinstance(session_id, str) or not session_id or slot not in {"mini-swe-agent", "opencollab-base"}:
         raise ValueError("Missing or unsupported native session identity")
     files = list((directory / "native-home").glob(f"**/*{session_id}.jsonl"))
     if len(files) != 1:
@@ -42,7 +43,6 @@ def export_messages(
             raise ValueError("Native history has an incomplete final record")
         data = data[: data.rfind(b"\n") + 1]
     entries = [json.loads(line) for line in data.splitlines() if line.strip()]
-    from sprout.contract.mini_history import training_messages
 
     # mini writes its formatted observation before it closes the turn. The runtime's own
     # presentation in ``tool.finished`` differs, and the export does not use it.
@@ -60,7 +60,7 @@ def provenance(
     ``prefix_messages`` counts the leading messages restored from ``recovery``. ``inserted_messages``
     indexes the assistant messages that a reviewer wrote, in this job or in an ancestor that it continues.
     """
-    from sprout.contract.mini_history import read_entries, read_prefix, training_messages
+    from sprout.contract.mini_history import read_entries, read_prefix
 
     files = list((directory / "native-home").glob(f"**/*{session_id}.jsonl"))
     if len(files) != 1:
@@ -72,11 +72,31 @@ def provenance(
         and (entry["message"].get("extra") or {}).get("source") == "reviewer"
         for call in entry["message"].get("tool_calls") or []
     }
+    oc_inserted = set()
+    records = read_entries(files[0].read_bytes())
+    if records and records[0].get("type") == "oc.session":
+        events = journal_events(directory / "trajectory.jsonl")
+        sampled = [
+            choice["message"]
+            for e in events
+            if e.get("type") == "raw.opencollab-base"
+            for choice in e.get("response", {}).get("choices", [])
+        ]
+        for index, message in enumerate(messages):
+            if message["role"] == "assistant" and not any(
+                (message.get("content") or "") == (candidate.get("content") or "")
+                and (message.get("tool_calls") or []) == (candidate.get("tool_calls") or [])
+                for candidate in sampled
+            ):
+                oc_inserted.add(index)
     inserted = [
         index
         for index, message in enumerate(messages)
-        if message["role"] == "assistant"
-        and {call["id"] for call in message.get("tool_calls") or []} & inserted_calls
+        if index in oc_inserted
+        or (
+            message["role"] == "assistant"
+            and {call["id"] for call in message.get("tool_calls") or []} & inserted_calls
+        )
     ]
     restored = 0
     if recovery:
@@ -98,7 +118,7 @@ def salvage_interrupted(
     The attempt started from the point ``recovery``, if any. The salvaged result is ``truncated``, with
     ``stop_reason: "interrupted"`` and the failure under ``interrupted``. None if ``point`` cannot be rebuilt.
     """
-    from sprout.contract.mini_history import read_entries, read_prefix, training_messages
+    from sprout.contract.mini_history import read_entries, read_prefix
 
     try:
         events = journal_events(directory / "trajectory.jsonl")
@@ -186,7 +206,7 @@ def complete_message_result(
         if slot == "mini-swe-agent" and is_truncated_result(result) and not unsafe:
             # After a timeout between two actions of one response, half the response does not
             # match the disk. Fall back to the last pair of messages and snapshot that is proven.
-            from sprout.contract.mini_history import read_entries, read_prefix, training_messages
+            from sprout.contract.mini_history import read_entries, read_prefix
             from sprout.worker.agents.mini_swe.history import index_mini
 
             files = list((directory / "native-home").glob(f"**/*{session_id}.jsonl"))
diff --git a/src/sprout/worker/jobs/review/branching.py b/src/sprout/worker/jobs/review/branching.py
index 29a4ff9f..ea5966e9 100644
--- a/src/sprout/worker/jobs/review/branching.py
+++ b/src/sprout/worker/jobs/review/branching.py
@@ -18,8 +18,8 @@
 from pathlib import Path
 from typing import Any, Protocol
 
-from sprout.contract.mini_history import load_prefix, training_messages, validate_assistant_turn
-from sprout.worker.agents.mini_swe.history import actor_tools_at, conversation_restore, reference_at
+from sprout.contract.agents.history import load_prefix, training_messages, validate_assistant_turn
+from sprout.worker.agents.history import actor_tools_at, conversation_restore, reference_at
 from sprout.worker.checkpoint.rollback import Checkpoint, load_checkpoints
 from sprout.worker.jobs.review import structured_review
 from sprout.worker.jobs.review.branch_plan import (
@@ -83,7 +83,7 @@ def json_overhead(self, schema: dict[str, Any], name: str) -> int: ...
 def require_mini_parent(journal: Path) -> None:
     """Raises ValueError unless ``journal`` has runs, and each of them is a mini-swe-agent run."""
     slots = {e["slot"] for e in read_journal(journal) if e.get("type") == "run.started" and e.get("slot")}
-    if slots != {"mini-swe-agent"}:
+    if slots not in ({"mini-swe-agent"}, {"opencollab-base"}):
         raise ValueError("This branch mode accepts only mini-swe-agent parent trajectories")
 
 
@@ -110,15 +110,19 @@ def reviewer_context(journal: Path, checkpoints: Mapping[int, Checkpoint]) -> di
     }
     count, cuts = 0, {}
     for entry in prefix:
-        if entry["type"] == "mini.message" and entry["message"].get("role") != "exit":
+        if entry["type"] in {"mini.message", "oc.message"} and entry["message"].get("role") != "exit":
             count += 1
-        elif entry["type"] == "mini.turn":
+        elif entry["type"] == "oc.context":
+            count = len(entry["messages"])
+        elif entry["type"] in {"mini.turn", "oc.turn"}:
             step = closed.get(entry["turn_id"])
             if step in checkpoints:
                 cuts[str(step)] = count
     if set(cuts) != {str(step) for step in checkpoints}:
         raise ValueError("Reviewer mini history does not cover every eligible checkpoint")
-    workspace = next((e.get("workspace") for e in prefix if e["type"] == "mini.session"), None)
+    workspace = next(
+        (e.get("workspace") for e in prefix if e["type"] in {"mini.session", "oc.session"}), None
+    )
     return {
         "messages": training_messages(prefix),
         "prefix_message_counts": cuts,
@@ -330,13 +334,17 @@ def prepare_branches(
             try:
                 assistant_turn = validate_assistant_turn(
                     proposed,
-                    history=[e["message"] for e in native if e["type"] == "mini.message"],
+                    history=training_messages(native),
                     tools=tools,
+                    slot="opencollab-base" if native[0].get("type") == "oc.session" else "mini-swe-agent",
                 )
             except ValueError as error:
                 raise ReviewerPlanError(f"branch {index} ({base_name}@{step}): {error}") from error
             for call_no, call in enumerate(assistant_turn["tool_calls"], 1):
-                if SUBMISSION_MARKER in call["function"]["arguments"]:
+                if (
+                    call["function"]["name"] == "submit_patch"
+                    or SUBMISSION_MARKER in call["function"]["arguments"]
+                ):
                     raise ReviewerPlanError(
                         f"branch {index} ({base_name}@{step}): a reviewer-authored turn never submits; "
                         f"command {call_no} contains {SUBMISSION_MARKER}"
@@ -610,7 +618,12 @@ def review(
     result: tuple[dict[str, Any], list[BranchChoice]] | None = review_with_feedback(
         _reviewer_request(transport, model=model, schema=schema),
         fit(),
-        _reply_parser(guidance_mode, round_no=round_no, structured=schema is not None),
+        _reply_parser(
+            guidance_mode,
+            round_no=round_no,
+            structured=schema is not None,
+            tools_by_base={row["name"]: row.get("native_history", {}).get("tools", []) for row in reports},
+        ),
         validate,
         max_attempts=reviewer_max_attempts,
         record=record,
@@ -633,7 +646,7 @@ def _eligible_points(
     eligible: Mapping[str, Collection[int]] | None,
 ) -> dict[str, dict[int, Checkpoint]]:
     """Each attempt's branch points up to the last step of its report, narrowed by ``eligible``."""
-    from sprout.worker.agents.mini_swe.history import available_branch_points
+    from sprout.worker.agents.history import available_branch_points
 
     available = {a.name: available_branch_points(a.outcome.journal_path) for a in attempts}
     return {
@@ -795,13 +808,22 @@ def reduce() -> str | None:
     return fit, reduce
 
 
-def _reply_parser(guidance_mode: str, *, round_no: int, structured: bool) -> Callable[[str], Any]:
+def _reply_parser(
+    guidance_mode: str,
+    *,
+    round_no: int,
+    structured: bool,
+    tools_by_base: dict[str, list[dict[str, Any]]] | None = None,
+) -> Callable[[str], Any]:
     """The parser that reads the branch plan from a reviewer reply in ``guidance_mode``."""
     parsers: dict[str, Callable[[str], Any]] = {
         "user-hint": extract_json,
         "none": extract_branch_plan,
         "assistant-turn": partial(
-            structured_review.extract_assistant_turn_plan, round_no=round_no, json_only=structured
+            structured_review.extract_assistant_turn_plan,
+            round_no=round_no,
+            json_only=structured,
+            tools_by_base=tools_by_base,
         ),
     }
     return parsers[guidance_mode]
diff --git a/src/sprout/worker/jobs/review/job.py b/src/sprout/worker/jobs/review/job.py
index 0cfdb1c9..edd3c402 100644
--- a/src/sprout/worker/jobs/review/job.py
+++ b/src/sprout/worker/jobs/review/job.py
@@ -32,10 +32,10 @@
 from pathlib import Path
 from typing import Any
 
+from sprout.contract.agents.history import training_messages
 from sprout.contract.messages import clean_messages
-from sprout.contract.mini_history import training_messages
 from sprout.runstore.specs import ReviewSpec
-from sprout.worker.agents.mini_swe.history import available_branch_points
+from sprout.worker.agents.history import available_branch_points
 from sprout.worker.jobs.review.branching import (
     Attempt,
     BranchChoice,
diff --git a/src/sprout/worker/jobs/review/structured_review.py b/src/sprout/worker/jobs/review/structured_review.py
index 38f522a2..8a9f4b25 100644
--- a/src/sprout/worker/jobs/review/structured_review.py
+++ b/src/sprout/worker/jobs/review/structured_review.py
@@ -125,7 +125,9 @@ def _validate(schema: dict[str, Any], value: object, where: str) -> None:
         ) from error
 
 
-def _assistant_turn(branch: dict[str, Any], *, round_no: int, index: int) -> dict[str, Any]:
+def _assistant_turn(
+    branch: dict[str, Any], *, round_no: int, index: int, command_tool: str = "bash"
+) -> dict[str, Any]:
     """The wire message for a flat branch: its text, and one bash call per command."""
     calls = []
     for call, command in enumerate(branch["commands"], 1):
@@ -138,7 +140,7 @@ def _assistant_turn(branch: dict[str, Any], *, round_no: int, index: int) -> dic
             {
                 "id": CALL_ID.format(round_no=round_no, branch=index + 1, call=call),
                 "type": "function",
-                "function": {"name": "bash", "arguments": {"command": command}},
+                "function": {"name": command_tool, "arguments": {"command": command}},
             }
         )
     if not branch["content"].strip():
@@ -146,7 +148,13 @@ def _assistant_turn(branch: dict[str, Any], *, round_no: int, index: int) -> dic
     return {"role": "assistant", "content": branch["content"], "tool_calls": calls}
 
 
-def extract_assistant_turn_plan(text: str, *, round_no: int = 0, json_only: bool = False) -> dict[str, Any]:
+def extract_assistant_turn_plan(
+    text: str,
+    *,
+    round_no: int = 0,
+    json_only: bool = False,
+    tools_by_base: dict[str, list[dict[str, Any]]] | None = None,
+) -> dict[str, Any]:
     """The plan in ``text`` with each branch in the nested form, and each call's arguments as a JSON string.
 
     A branch gives ``content`` and ``commands``, or a whole ``assistant_turn``. The calls built from
@@ -157,6 +165,9 @@ def extract_assistant_turn_plan(text: str, *, round_no: int = 0, json_only: bool
     result: dict[str, Any] = deepcopy(plan)
     for index, branch in enumerate(result["branches"]):
         where = "branches.%d" % index
+        tools = (tools_by_base or {}).get(branch.get("base"), [])
+        names = {tool["function"]["name"] for tool in tools}
+        command_tool = "run_command" if "run_command" in names else "bash"
         flat = {"content", "commands"} & set(branch)
         if flat and "assistant_turn" in branch:
             raise ValueError(
@@ -165,12 +176,19 @@ def extract_assistant_turn_plan(text: str, *, round_no: int = 0, json_only: bool
             )
         if flat:
             _validate(FLAT_BRANCH, branch, where)
-            turn = _assistant_turn(branch, round_no=round_no, index=index)
+            turn = _assistant_turn(branch, round_no=round_no, index=index, command_tool=command_tool)
             for key in ("content", "commands"):
                 del branch[key]
             branch["assistant_turn"] = turn
         elif "assistant_turn" in branch:
-            _validate(NESTED_BRANCH, branch, where)
+            nested = deepcopy(NESTED_BRANCH)
+            if names:
+                function = nested["properties"]["assistant_turn"]["properties"]["tool_calls"]["items"][
+                    "properties"
+                ]["function"]["properties"]
+                function["name"]["enum"] = sorted(names)
+                function["arguments"] = {"type": "object"}
+            _validate(nested, branch, where)
         else:
             raise ValueError("Structured branch-plan field %s: a branch needs content and commands" % where)
         for number, call in enumerate(branch["assistant_turn"]["tool_calls"]):
diff --git a/src/sprout/worker/jobs/rollout.py b/src/sprout/worker/jobs/rollout.py
index c804c0d6..4856fad5 100644
--- a/src/sprout/worker/jobs/rollout.py
+++ b/src/sprout/worker/jobs/rollout.py
@@ -24,8 +24,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from sprout.contract.agents.history import load_prefix, materialize
 from sprout.contract.continuation import Continuation, prepare
-from sprout.contract.mini_history import load_prefix, materialize
 from sprout.contract.run_spec import KEEP_SANDBOX_S, RunSpec
 from sprout.runstore.config import referenced_env, resolve
 from sprout.runstore.payload import receive_payload
diff --git a/src/sprout/worker/run/rollout.py b/src/sprout/worker/run/rollout.py
index 9730c158..008b928a 100644
--- a/src/sprout/worker/run/rollout.py
+++ b/src/sprout/worker/run/rollout.py
@@ -121,6 +121,18 @@ def before(self, ctx: CallContext) -> Verdict:
         except ValueError as error:
             return Reject(str(error))
 
+    def status(self) -> dict[str, int | float | None]:
+        """The enforced limits and measured calls of this run; tool calls have no count limit."""
+        with self.lock:
+            return {
+                "model_calls": self.model_calls,
+                "max_turns": self.max_turns,
+                "remaining_turns": max(0, self.max_turns - self.model_calls),
+                "runtime_tool_calls": self.tool_calls,
+                "max_tool_calls": None,
+                "remaining_seconds": max(0.0, self.deadline - time.time()),
+            }
+
     def finish(self) -> None:
         """Disarm the deadline, and journal the call counts as ``rollout.usage``."""
         if self.timer:
diff --git a/src/sprout/worker/run/runner.py b/src/sprout/worker/run/runner.py
index 20952175..7c6034dd 100644
--- a/src/sprout/worker/run/runner.py
+++ b/src/sprout/worker/run/runner.py
@@ -147,11 +147,15 @@ def run(self, spec: RunSpec) -> RunOutcome:
                 detach = self._follow_reboards(spec, provisioned, claim)
                 _emit_sandbox_created(spec=spec, journal=journal, sandbox_id=provisioned.sandbox_id)
                 if spec.agent == "mini":
-                    from sprout.worker.agents.mini_swe.sandbox import resolve_workspace
+                    if spec.slot == "opencollab-base":
+                        from sprout.worker.agents.opencollab.workspace import resolve_workspace
+                    else:
+                        from sprout.worker.agents.mini_swe.sandbox import resolve_workspace
 
                     extra = resolve_workspace(provisioned.session, extra, journal)
                 tools = self._tool_access(spec, journal, provisioned, control, rollout_controls)
                 task = _task_spec(spec=spec, extra=extra, control=control, session=provisioned.session)
+                task.extra["rollout_controls"] = rollout_controls
                 gateway = self._wire_gateway(spec, journal, task, run_id, controls=rollout_controls)
                 result = slot.run(task, journal, tools)
                 error = _stop_error(control=control, result=result)
@@ -475,9 +479,13 @@ def _image_reference(spec: RunSpec) -> str | None:
 
 def _agent_slot(spec: RunSpec) -> AgentSlot:
     """The slot that runs ``spec.agent``. Raises ValueError for a slot or an agent that no slot runs."""
-    if spec.slot != "mini-swe-agent":
+    if spec.slot not in {"mini-swe-agent", "opencollab-base"}:
         raise ValueError("unknown slot %r; the only slot is mini-swe-agent" % spec.slot)
     if spec.agent == "mini":
+        if spec.slot == "opencollab-base":
+            from sprout.worker.agents.opencollab.slot import OpenCollabSlot
+
+            return OpenCollabSlot()
         return MiniSweSlot()
     if spec.agent == "nop":
         return NopSlot()
diff --git a/tests/rl_driver/test_miles_requests.py b/tests/rl_driver/test_miles_requests.py
index ee666bbf..06fabf18 100644
--- a/tests/rl_driver/test_miles_requests.py
+++ b/tests/rl_driver/test_miles_requests.py
@@ -310,7 +310,7 @@ def test_a_search_request_one_slot_short_of_its_branches_is_refused(fields, slot
         pytest.param(adapter_config(workers=4), "Invalid Miles adapter config", id="unknown-field"),
         pytest.param(
             adapter_config(run_defaults={"slot": "codex"}),
-            "The Miles adapter supports mini-swe-agent only",
+            "The Miles adapter requires mini-swe-agent or opencollab-base",
             id="other-slot",
         ),
         pytest.param(
diff --git a/tests/runstore/test_specs.py b/tests/runstore/test_specs.py
index 7673c9aa..043b8f3f 100644
--- a/tests/runstore/test_specs.py
+++ b/tests/runstore/test_specs.py
@@ -329,8 +329,9 @@ def test_a_continuation_that_changes_anything_else_is_refused(child):
         rollout_spec(agent="nop"),
         rollout_spec(agent="mini", harbor_task=TASK_DIR, keep_sandbox=True),
         rollout_spec(slot="mini-swe-agent", cwd=".", run_id=None),
+        rollout_spec(slot="opencollab-base"),
     ],
-    ids=["kept-sandbox", "oracle", "nop", "mini-on-a-harbor-task", "the-defaults-spelled-out"],
+    ids=["kept-sandbox", "oracle", "nop", "mini-on-a-harbor-task", "the-defaults-spelled-out", "opencollab"],
 )
 def test_a_queued_rollout_may_keep_its_sandbox_and_run_each_agent(spec):
     assert JobSpec("rollout", spec).validate()["spec"] == spec
@@ -342,7 +343,7 @@ def test_a_queued_rollout_may_keep_its_sandbox_and_run_each_agent(spec):
         ({"prompt": "  "}, "RunSpec requires a nonempty prompt"),
         ({"model": "policy"}, "RunSpec requires a nonempty prompt"),
         (rollout_spec(sandbox="podman"), "Unknown RunSpec fields"),
-        (rollout_spec(slot="codex-cli"), "The only queued-run slot is mini-swe-agent"),
+        (rollout_spec(slot="codex-cli"), "Queued-run slot must be mini-swe-agent or opencollab-base"),
         (rollout_spec(agent="codex"), "RunSpec agent must be mini, oracle or nop"),
         (rollout_spec(agent="nop", harbor_task=" "), "RunSpec harbor_task must name a Harbor task directory"),
         (rollout_spec(agent="oracle"), "An oracle run needs the harbor_task whose solution it runs"),
diff --git a/tests/worker/agents/opencollab/__init__.py b/tests/worker/agents/opencollab/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/worker/agents/opencollab/test_integration.py b/tests/worker/agents/opencollab/test_integration.py
new file mode 100644
index 00000000..26117a48
--- /dev/null
+++ b/tests/worker/agents/opencollab/test_integration.py
@@ -0,0 +1,196 @@
+# ruff: noqa: E402
+"""The shared worker, branch restoration, reviewer and training-export boundaries."""
+
+import json
+from dataclasses import asdict
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("opencollab")
+
+from sprout.contract.run_spec import RunSpec
+from sprout.worker.agents.history import available_branch_points, reference_at
+from sprout.worker.jobs.export import complete_message_result
+from sprout.worker.jobs.review.structured_review import extract_assistant_turn_plan
+from sprout.worker.jobs.rollout import _from_point
+from sprout.worker.run.runner import _agent_slot
+
+from .test_opencollab import Model, call, response, run
+from .test_review50_protocol import repository
+
+
+def test_runtime_submission_receipt_is_context_without_becoming_policy_output(tmp_path):
+    root = repository(tmp_path / "repo")
+    result, _, journal = run(
+        root, tmp_path / "run", Model([response(call("submit", "submit_patch"))]), protocol="review50-v1"
+    )
+    assert result.status == "completed", result.error
+    exported = complete_message_result(asdict(result), journal.parent, "opencollab-base")
+    assert "training_export_error" not in exported
+    messages = exported["training_messages"]
+    receipt = len(messages) - 1
+    action = next(i for i, m in enumerate(messages) if m.get("tool_calls"))
+    assert receipt in exported["training_provenance"]["inserted_messages"]
+    assert action not in exported["training_provenance"]["inserted_messages"]
+    assert exported["training_provenance"]["prefix_messages"] == 0
+    assert "patch_sha256" in messages[receipt]["content"]
+
+
+@pytest.mark.parametrize("rendering", ["content", "reasoning"])
+def test_queued_branch_materializes_exact_sdk_state_and_masks_reviewer(tmp_path, rendering):
+    root = repository(tmp_path / "repo")
+    parent, sandbox, journal = run(
+        root,
+        tmp_path / "parent",
+        Model([response(call("read", "read_file", filepath="source.txt")), response(content="Parent done.")]),
+        protocol="review50-v1",
+    )
+    assert parent.status == "completed", parent.error
+    step, checkpoint = min(available_branch_points(journal).items())
+    native = reference_at(journal, step, parent.native_session_id)
+    point = {
+        "id": "point",
+        "snapshot_id": checkpoint.snapshot_id,
+        "tool_depth": step,
+        "message_step": 1,
+        "native": native,
+    }
+    turn = {
+        "role": "assistant",
+        "content": "Check the retained source before editing.",
+        "tool_calls": [call("review", "run_command", command="git diff --stat")],
+    }
+    destination = tmp_path / "queued"
+    fields = _from_point(
+        {"prompt": "Fix the fixture", "extra": {}},
+        point,
+        continuation={
+            "guidance": "assistant-turn",
+            "assistant_turn": turn,
+            "assistant_turn_rendering": rendering,
+        },
+        directory=destination,
+        slot="opencollab-base",
+    )
+    assert fields["sandbox_image"] == checkpoint.snapshot_id
+    assert Path(fields["extra"]["native_prefix"]["path"]).is_file()
+    result, _, child_journal = run(
+        root,
+        tmp_path / "child",
+        Model([response(content="Child done.")]),
+        prefix=fields["extra"]["native_prefix"],
+        reviewer=fields["extra"]["assistant_turn"],
+        sandbox=sandbox,
+        protocol="review50-v1",
+    )
+    assert result.status == "completed", result.error
+    exported = complete_message_result(
+        asdict(result), child_journal.parent, "opencollab-base", recovery=point
+    )
+    assert "training_export_error" not in exported
+    provenance = exported["training_provenance"]
+    prefix = provenance["prefix_messages"]
+    assert prefix > 0 and prefix in provenance["inserted_messages"]
+    assert len(exported["training_messages"]) - 1 not in provenance["inserted_messages"]
+
+
+def test_flat_review_commands_use_the_selected_parents_actual_tool():
+    from sprout.worker.agents.opencollab.review50 import definition
+
+    branch = {
+        "name": "repair",
+        "base": "parent",
+        "branch_step": 1,
+        "why": "inspect",
+        "content": "Inspect the source first.",
+        "commands": ["git diff"],
+    }
+    plan = extract_assistant_turn_plan(
+        json.dumps({"synthesis": "Inspect first", "branches": [branch]}),
+        json_only=True,
+        tools_by_base={"parent": definition()["tools"]},
+    )
+    tool = plan["branches"][0]["assistant_turn"]["tool_calls"][0]
+    assert tool["function"]["name"] == "run_command"
+    assert json.loads(tool["function"]["arguments"]) == {"command": "git diff"}
+    mini = extract_assistant_turn_plan(
+        json.dumps({"synthesis": "Inspect first", "branches": [branch]}), json_only=True
+    )
+    assert mini["branches"][0]["assistant_turn"]["tool_calls"][0]["function"]["name"] == "bash"
+
+
+def test_slot_is_selected_by_the_existing_worker_runner():
+    assert _agent_slot(RunSpec(prompt="fix", slot="opencollab-base")).name == "opencollab-base"
+    assert _agent_slot(RunSpec(prompt="fix")).name == "mini-swe-agent"
+
+
+@pytest.mark.parametrize(
+    "protocol,success", [("single2", True), ("review50-v1", True), ("review50-v1", False)]
+)
+def test_historical_workspace_alias_is_prepared_without_overwriting_another_checkout(
+    tmp_path, monkeypatch, protocol, success
+):
+    from types import SimpleNamespace
+
+    from sprout.worker.agents.opencollab import workspace
+    from sprout.worker.run.journal import JournalWriter
+    from sprout.worker.tools.result import ToolResult
+
+    resolved = {
+        "mini_workspace": {"cwd": "/testbed", "repository_dir": "/testbed"},
+        "opencollab_tool_protocol": protocol,
+    }
+    monkeypatch.setattr(workspace, "resolve_native_workspace", lambda *_: resolved)
+    calls = []
+
+    def execute(name, arguments, **kwargs):
+        calls.append((name, arguments))
+        return ToolResult(success, "")
+
+    session = SimpleNamespace(execute=execute)
+    with JournalWriter(tmp_path / "workspace.jsonl") as journal:
+        if not success:
+            with pytest.raises(ValueError, match="alias"):
+                workspace.resolve_workspace(session, resolved, journal)
+        else:
+            assert workspace.resolve_workspace(session, resolved, journal) == resolved
+    if protocol == "single2":
+        assert not calls
+    else:
+        assert len(calls) == 1 and calls[0][0] == "shell"
+        assert 'test "$(cd /workspace && pwd -P)" = /testbed' in calls[0][1]["command"]
+        assert "ln -s -- /testbed /workspace" in calls[0][1]["command"]
+
+
+def test_native_sdk_uses_nonstreaming_requests_through_the_real_gateway(tmp_path, monkeypatch):
+    from sprout.worker.gateway import GatewayServer, RoutingTable
+    from sprout.worker.gateway.routing import ModelRoute
+    from sprout.worker.run.journal import JournalWriter
+    from tests.support.http import chat_server
+
+    completion = {
+        "id": "response",
+        "object": "chat.completion",
+        "created": 0,
+        "model": "fixture-model",
+        "choices": [
+            {"index": 0, "message": {"role": "assistant", "content": "Done."}, "finish_reason": "stop"}
+        ],
+        "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+    }
+    with chat_server([completion]) as provider, JournalWriter(tmp_path / "gateway.jsonl") as journal:
+        table = RoutingTable({"default": ModelRoute(base_url=provider.url)})
+        gateway = GatewayServer(table, journal=journal).start()
+        token = table.mint("agent")
+        try:
+            for key, value in gateway.env_for(token).items():
+                monkeypatch.setenv(key, value)
+            checkout = tmp_path / "repo"
+            checkout.mkdir()
+            result, _, _ = run(checkout, tmp_path / "run", None)
+        finally:
+            gateway.stop()
+    assert result.status == "completed", result.error
+    assert result.final_text == "Done."
+    assert len(provider.requests) == 1 and not provider.requests[0].get("stream", False)
diff --git a/tests/worker/agents/opencollab/test_opencollab.py b/tests/worker/agents/opencollab/test_opencollab.py
new file mode 100644
index 00000000..c45b74e2
--- /dev/null
+++ b/tests/worker/agents/opencollab/test_opencollab.py
@@ -0,0 +1,464 @@
+# ruff: noqa: E402
+"""Single2 execution, exact rollback, and accounting across native branches."""
+
+import asyncio
+import json
+import subprocess
+import threading
+import time
+from uuid import uuid4
+
+import pytest
+
+pytest.importorskip("opencollab")
+from opencollab.adapters.llm.types import LLMResponse, Usage
+from opencollab.application.session_run import GenerationTimeoutError
+
+from sprout.contract.agents.history import load_prefix, training_messages, validate_assistant_turn
+from sprout.worker.agents.history import actor_tools_at, available_branch_points, reference_at
+from sprout.worker.agents.opencollab.environment import SproutEnvironment
+from sprout.worker.agents.opencollab.slot import OpenCollabSlot
+from sprout.worker.agents.slot import TaskSpec
+from sprout.worker.checkpoint.checkpointer import Checkpointer
+from sprout.worker.run.control import RunControl
+from sprout.worker.run.journal import JournalWriter, read_journal
+from sprout.worker.tools.access import ToolAccess
+from sprout.worker.tools.interceptors import MutationTracker
+from sprout.worker.tools.pipeline import ToolPipeline
+from sprout.worker.tools.result import CommandOutcome, ToolResult
+from sprout_sandbox.pool import Snapshot
+
+
+def call(identifier, name, **args):
+    return {"id": identifier, "type": "function", "function": {"name": name, "arguments": json.dumps(args)}}
+
+
+class Model:
+    def context_window(self):
+        return 128000
+
+    def __init__(self, responses):
+        self.responses = iter(responses)
+        self.requests = []
+        self.closed = False
+
+    async def complete(self, messages, **kwargs):
+        self.requests.append((json.loads(json.dumps(messages)), kwargs))
+        return next(self.responses)
+
+    async def close(self):
+        self.closed = True
+
+
+def response(*calls, content="Testing a concrete change."):
+    return LLMResponse(
+        content=content,
+        tool_calls=list(calls),
+        finish_reason="tool_calls" if calls else "stop",
+        usage=Usage(
+            input_tokens=10, output_tokens=5, raw_usage={"prompt_tokens": 10, "completion_tokens": 5}
+        ),
+    )
+
+
+class Sandbox:
+    """Only a controlled test fixture; does not claim live VM verification."""
+
+    def __init__(self, root):
+        self.root = root
+        self.sandbox_id = uuid4().hex
+        self.snapshots = {}
+        self.failure = False
+
+    def supports_snapshot(self):
+        return True
+
+    def call(self, name, args, timeout):
+        if self.failure:
+            raise ConnectionError("synthetic lost execution transport")
+        result = subprocess.run(
+            args["command"],
+            shell=True,
+            cwd=args["working_dir"],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+        return ToolResult(
+            result.returncode == 0,
+            result.stdout + result.stderr,
+            outcome=CommandOutcome(exit_code=result.returncode, stdout=result.stdout, stderr=result.stderr),
+        )
+
+    def snapshot(self, **kwargs):
+        identifier = str(uuid4())
+        self.snapshots[identifier] = {
+            path.name: path.read_bytes() for path in self.root.iterdir() if path.is_file()
+        }
+        return Snapshot(identifier)
+
+
+def run(
+    root,
+    logdir,
+    model,
+    *,
+    prefix=None,
+    reviewer=None,
+    point_only=False,
+    sandbox=None,
+    timeout=30,
+    protocol="single2",
+):
+    logdir.mkdir()
+    sandbox = sandbox or Sandbox(root)
+    with JournalWriter(logdir / "trajectory.jsonl") as journal:
+        control, tracker = RunControl(), MutationTracker()
+        tools = ToolAccess(
+            sandbox,
+            journal,
+            control=control,
+            pipeline=ToolPipeline([tracker]),
+            checkpointer=Checkpointer(session=sandbox, tracker=tracker),
+        )
+        extra = {
+            "mini_workspace": {"cwd": str(root), "repository_dir": str(root)},
+            "native_home": str(logdir / "native-home"),
+            "opencollab_llm": model,
+            "opencollab_tool_protocol": protocol,
+        }
+        if prefix:
+            extra["native_prefix"] = prefix
+        if reviewer:
+            extra["assistant_turn"] = reviewer
+        if point_only:
+            extra["resume_without_hint"] = True
+        task = TaskSpec(
+            "Fix the fixture.",
+            str(logdir),
+            model="fixture-model",
+            timeout_s=timeout,
+            control=control,
+            extra=extra,
+        )
+        result = OpenCollabSlot().run(task, journal, tools)
+        tools.finalize()
+    return result, sandbox, logdir / "trajectory.jsonl"
+
+
+def test_native_six_tools_closed_turns_and_reviewer_branch(tmp_path):
+    root = tmp_path / "repo"
+    root.mkdir()
+    model = Model(
+        [
+            response(
+                call("a", "file_write", path="fixture.txt", content="parent", mode="create"),
+                call("b", "file_read", path="fixture.txt"),
+            ),
+            response(call("c", "bash", command="printf later > fixture.txt")),
+            response(content="Done."),
+        ]
+    )
+    result, sandbox, journal = run(root, tmp_path / "parent", model)
+    assert result.status == "completed", result.error
+    assert result.usage["input_tokens"] == 30
+    assert model.closed
+    assert [schema["function"]["name"] for schema in model.requests[0][1]["tools"]] == [
+        "bash",
+        "file_read",
+        "file_write",
+        "apply_patch",
+        "git_diff",
+        "grep",
+    ]
+    assert model.requests[0][1]["reasoning_effort"] == "high"
+    points = available_branch_points(journal)
+    assert len(points) == 2
+    step, point = min(points.items())
+    prefix = reference_at(journal, step, point.session_ckpt)
+    entries = load_prefix(prefix)
+    retained = training_messages(entries)
+    assert retained[-1]["role"] == "tool"
+    assert "parent" in retained[-1]["content"]
+    assert not any("printf later" in str(message) for message in retained)
+    # Restore this test fixture's disk BEFORE continuing from the exact state.
+    for path in root.iterdir():
+        if path.is_file():
+            path.unlink()
+    for name, contents in sandbox.snapshots[point.snapshot_id].items():
+        (root / name).write_bytes(contents)
+    branch_model = Model([response(content="Branch done.")])
+    reviewer = {
+        "role": "assistant",
+        "content": "I will verify a revised fix.",
+        "tool_calls": [call("review", "file_write", path="fixture.txt", content="branch", mode="create")],
+    }
+    branch, _, child_journal = run(root, tmp_path / "child", branch_model, prefix=prefix, reviewer=reviewer)
+    assert branch.status == "completed", branch.error
+    assert (root / "fixture.txt").read_text() == "branch"
+    assert branch.usage["input_tokens"] == 10  # inherited and reviewer tokens are not recharged
+    child_messages = branch_model.requests[0][0]
+    assert child_messages[: len(retained)] == retained
+    # Single2 adds its own native progress steering; branching must not add
+    # another task prompt or reviewer instruction as a synthetic user message.
+    new_users = [message for message in child_messages[len(retained) :] if message["role"] == "user"]
+    assert all(message["content"].startswith("[Progress:") for message in new_users)
+    assert len(available_branch_points(child_journal)) == 1
+    validate_assistant_turn(
+        reviewer, history=retained, tools=actor_tools_at(journal, step), slot="opencollab-base"
+    )
+
+
+def test_transport_failure_cannot_be_reported_completed(tmp_path):
+    root = tmp_path / "repo"
+    root.mkdir()
+    sandbox = Sandbox(root)
+    sandbox.failure = True
+    result, _, journal = run(
+        root,
+        tmp_path / "broken",
+        Model([response(call("a", "bash", command="true")), response(content="Done.")]),
+        sandbox=sandbox,
+    )
+    assert result.status == "error"
+    assert not available_branch_points(journal)
+    rows = read_journal(journal)
+    assert any(row["type"] == "execution.uncertain" for row in rows)
+    bridge = next(row for row in rows if row["type"] == "opencollab.bridge.failed")
+    assert bridge["cause_type"] == "ToolInfraError"
+    assert "did not settle" in bridge["cause"]
+    uncertain = next(row for row in rows if row["type"] == "execution.uncertain")
+    assert uncertain["exception"] == "ConnectionError"
+    assert uncertain["message"] == "synthetic lost execution transport"
+    finished = next(row for row in rows if row["type"] == "run.finished")
+    assert finished["stop_reason"] == "infrastructure_error"
+
+
+@pytest.mark.parametrize(
+    "returncode,stdout,truncated,error",
+    [
+        (0, "Unicode: 中文\n", False, None),
+        (1, "", False, FileNotFoundError),
+        (0, "partial", True, OSError),
+    ],
+)
+def test_full_file_read_uses_owned_transport(tmp_path, returncode, stdout, truncated, error):
+    """New Docker private methods must not bypass the SPROUT environment port."""
+    from opencollab.adapters.env import ExecResult
+
+    env = SproutEnvironment(None, str(tmp_path), RunControl(), time.monotonic() + 10)
+    calls = []
+
+    async def execute(command, timeout=120):
+        calls.append(command)
+        return ExecResult(returncode, stdout, "missing", stdout_truncated=truncated)
+
+    env.exec_cmd = execute
+    path = "/workspace/a'b.txt"
+    if error is None:
+        assert asyncio.run(env.read_file(path)) == stdout
+    else:
+        with pytest.raises(error):
+            asyncio.run(env.read_file(path))
+    import shlex
+
+    assert calls == [f"cat -- {shlex.quote(path)}"]
+
+
+def test_cancelled_command_is_settled_without_becoming_transport_failure(tmp_path):
+    """OpenCollab may cancel a tool deadline while its synchronous worker is
+    queued.  The adapter must let that owned call settle before returning the
+    cancellation, otherwise it runs behind a later checkpoint."""
+    settled = threading.Event()
+
+    class SlowTools:
+        def call(self, _name, _args, *, call_id, turn_id):
+            assert call_id and turn_id
+            time.sleep(0.05)
+            settled.set()
+            return ToolResult(True, "done", outcome=CommandOutcome(exit_code=0, stdout="done"))
+
+    async def scenario():
+        env = SproutEnvironment(SlowTools(), str(tmp_path), RunControl(), time.monotonic() + 10)
+        env.turn_id = "turn"
+        task = asyncio.create_task(env.exec_cmd("true"))
+        await asyncio.sleep(0.005)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        return env
+
+    env = asyncio.run(scenario())
+    assert settled.is_set()
+    assert env.failure is None
+    assert len(env.call_ids) == 1
+
+
+def test_controller_stop_after_settled_command_is_not_transport_failure(tmp_path):
+    control = RunControl()
+
+    class StoppingTools:
+        def call(self, _name, _args, *, call_id, turn_id):
+            assert call_id and turn_id
+            control.request_stop("deadline", stop_reason="timeout")
+            return ToolResult(True, "done", outcome=CommandOutcome(exit_code=0, stdout="done"))
+
+    async def scenario():
+        env = SproutEnvironment(StoppingTools(), str(tmp_path), control, time.monotonic() + 10)
+        env.turn_id = "turn"
+        with pytest.raises(Exception, match="deadline"):
+            await env.exec_cmd("true")
+        return env
+
+    env = asyncio.run(scenario())
+    assert env.failure is None
+    assert len(env.call_ids) == 1
+
+
+def test_prefix_checksum_and_invalid_reviewer_arguments(tmp_path):
+    import hashlib
+
+    from opencollab.bootstrap.agent_profiles import resolve_agent_profile
+
+    from sprout.worker.agents.opencollab.history import History
+
+    path = tmp_path / "native.jsonl"
+    history = History(path, "session", {"cwd": "/fixture"})
+    history.append({"type": "oc.turn", "session_snapshot": {}, "runtime_call_ids": []})
+    history.close()
+    reference = {"path": str(path), "sha256": hashlib.sha256(path.read_bytes()).hexdigest()}
+    assert load_prefix(reference)
+    with path.open("ab") as stream:
+        stream.write(b" \n")
+    with pytest.raises(ValueError, match="checksum"):
+        load_prefix(reference)
+    schemas = [tool.to_openai_schema() for tool in resolve_agent_profile("base").resolve_tools("coding")]
+    invalid = {
+        "role": "assistant",
+        "content": "Check a hypothesis.",
+        "tool_calls": [call("a", "bash", command="true", invented=True)],
+    }
+    with pytest.raises(ValueError, match="recorded tool schema"):
+        validate_assistant_turn(invalid, tools=schemas, slot="opencollab-base")
+
+
+def test_native_context_shaping_preserves_actual_model_context(tmp_path):
+    from sprout.worker.agents.opencollab.history import History
+
+    path = tmp_path / "shaped.jsonl"
+    history = History(path, "native", {"cwd": "/fixture"})
+    initial = [{"role": "system", "content": "System."}, {"role": "user", "content": "Original task."}]
+    history.persist_messages(initial)
+    shaped = [{"role": "system", "content": "System."}, {"role": "user", "content": "Shaped native task."}]
+    history.persist_messages(shaped)
+    history.persist_messages(shaped[:1])
+    history.close()
+    rows = [json.loads(line) for line in path.read_text().splitlines()]
+    assert training_messages(rows) == shaped[:1]
+
+
+def test_slow_generation_has_no_native_per_call_stop(tmp_path, monkeypatch):
+    import asyncio
+
+    from sprout.worker.agents.opencollab import slot
+
+    build = slot.build_session
+    sessions = []
+    agents = []
+
+    def capture(**kwargs):
+        agents.append(kwargs["agent"])
+        session = build(**kwargs)
+        sessions.append(session)
+        return session
+
+    class SlowModel(Model):
+        async def complete(self, messages, **kwargs):
+            await asyncio.sleep(0.08)
+            return response(content="Done.")
+
+    monkeypatch.setattr(slot, "build_session", capture)
+    root = tmp_path / "repo"
+    root.mkdir()
+    result, _, _ = run(root, tmp_path / "slow", SlowModel([]), timeout=1)
+    assert sessions[0].runner._per_call_timeout is None
+    assert all(tool.disable_outer_timeout for tool in agents[0].tools)
+    assert result.status == "completed"
+
+
+@pytest.mark.parametrize(
+    "exception", [TimeoutError("provider timed out"), GenerationTimeoutError("synthetic generation deadline")]
+)
+def test_inner_timeout_is_infrastructure_error_not_actor_deadline(tmp_path, exception):
+    class FailedModel(Model):
+        async def complete(self, messages, **kwargs):
+            raise exception
+
+    root = tmp_path / "repo"
+    root.mkdir()
+    result, _, journal = run(root, tmp_path / "failed", FailedModel([]))
+    assert result.status == "error"
+    assert type(exception).__name__ in result.error
+    assert "wall-time exhausted" not in result.error
+    finished = next(r for r in read_journal(journal) if r["type"] == "run.finished")
+    assert finished["termination_kind"] == "infrastructure_error"
+    assert finished["elapsed_s"] < finished["actor_timeout_s"]
+
+
+def test_real_actor_deadline_still_stops_a_pending_generation(tmp_path):
+    import asyncio
+
+    class HungModel(Model):
+        async def complete(self, messages, **kwargs):
+            await asyncio.Event().wait()
+
+    root = tmp_path / "repo"
+    root.mkdir()
+    result, _, journal = run(root, tmp_path / "deadline", HungModel([]), timeout=0.08)
+    assert result.status == "timeout"
+    finished = next(r for r in read_journal(journal) if r["type"] == "run.finished")
+    assert finished["termination_kind"] == "actor_deadline"
+    assert finished["elapsed_s"] >= 0.08
+
+
+def test_native_compaction_uses_loop_owned_provider_clients(tmp_path):
+    import asyncio
+
+    from opencollab.application.compaction_summary import run_coro_blocking
+
+    from sprout.worker.agents.opencollab.slot import RecordedModel
+
+    created = []
+
+    class LoopOwnedModel(Model):
+        def __init__(self):
+            super().__init__([])
+            self.loop = asyncio.get_running_loop()
+            created.append(self)
+
+        async def complete(self, messages, **kwargs):
+            assert asyncio.get_running_loop() is self.loop
+            return response(content="<summary>Retained progress.</summary>")
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            assert asyncio.get_running_loop() is self.loop
+            await self.close()
+
+    async def exercise(journal):
+        primary = LoopOwnedModel()
+        model = RecordedModel(primary, journal, RunControl(), client_factory=LoopOwnedModel)
+        await model.complete([])
+        for _ in range(2):
+            run_coro_blocking(lambda: model.complete([]))
+        await model.complete([])
+        assert model.usage.input_tokens == 40
+        assert not primary.closed
+        await primary.close()
+
+    with JournalWriter(tmp_path / "compaction.jsonl") as journal:
+        asyncio.run(exercise(journal))
+    assert len(created) == 3 and all(client.closed for client in created)
diff --git a/tests/worker/agents/opencollab/test_opencollab_large_write.py b/tests/worker/agents/opencollab/test_opencollab_large_write.py
new file mode 100644
index 00000000..9b916998
--- /dev/null
+++ b/tests/worker/agents/opencollab/test_opencollab_large_write.py
@@ -0,0 +1,202 @@
+# ruff: noqa: E402
+"""Large writes remain bounded, atomic, and on the owned ToolAccess path."""
+
+import asyncio
+import subprocess
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+pytest.importorskip("opencollab")
+from opencollab.adapters.env import ExecResult
+
+from sprout.worker.agents.opencollab import environment
+from sprout.worker.run.control import RunControl
+from sprout.worker.run.journal import JournalWriter, read_journal
+from sprout.worker.tools.access import ToolAccess
+
+from .test_opencollab import Model, Sandbox, call, response, run
+
+
+@pytest.fixture
+def port(tmp_path):
+    sandbox = Sandbox(tmp_path)
+    with JournalWriter(tmp_path / "calls.jsonl") as journal:
+        control = RunControl()
+        tools = ToolAccess(sandbox, journal, control=control)
+        env = environment.SproutEnvironment(
+            tools,
+            str(tmp_path),
+            control,
+            time.monotonic() + 120,
+        )
+        env.turn_id = "large-write"
+        yield env
+
+
+@pytest.mark.parametrize("size", [0, 32769, 300000, 4 * 1024 * 1024])
+def test_large_utf8_roundtrip_bounded_owned_calls(tmp_path, port, size):
+    target = tmp_path / "quoted ' 中文" / "file.txt"
+    content = "中" * (size // 3) + "x" * (size % 3)
+    asyncio.run(port.write_file(str(target), content))
+    assert target.read_bytes() == content.encode()
+    calls = [row for row in read_journal(tmp_path / "calls.jsonl") if row["type"] == "tool.started"]
+    assert calls
+    assert max(len(row["args"]["command"].encode()) for row in calls) < 65536
+    assert len(calls) == 2 + (size + 32767) // 32768
+    assert not list(target.parent.glob(".sprout-write-*"))
+
+
+def test_existing_executable_permissions_are_preserved(tmp_path, port):
+    target = tmp_path / "script"
+    target.write_text("original")
+    target.chmod(0o751)
+    asyncio.run(port.write_file(str(target), "new" * 100000))
+    assert target.stat().st_mode & 0o777 == 0o751
+
+
+@pytest.mark.parametrize("failure", ["chunk", "checksum", "cancel"])
+def test_failure_keeps_original_and_cleans_owned_stage(tmp_path, port, failure):
+    target = tmp_path / "target"
+    target.write_text("original")
+    execute = port.exec_cmd
+    chunks = 0
+
+    async def inject(command, timeout=120):
+        nonlocal chunks
+        if "base64 -d >>" in command:
+            chunks += 1
+            if chunks == 2:
+                if failure == "chunk":
+                    return ExecResult(1, "", "injected append failure")
+                if failure == "cancel":
+                    raise asyncio.CancelledError()
+                command = command.replace("base64 -d >>", "head -c 1 >>")
+        return await execute(command, timeout)
+
+    port.exec_cmd = inject
+    error = asyncio.CancelledError if failure == "cancel" else OSError
+    with pytest.raises(error):
+        asyncio.run(port.write_file(str(target), "new" * 100000))
+    assert target.read_text() == "original"
+    assert not list(tmp_path.glob(".sprout-write-*"))
+
+
+def test_collision_does_not_overwrite_or_delete_foreign_stage(tmp_path, port, monkeypatch):
+    monkeypatch.setattr(environment, "uuid4", lambda: SimpleNamespace(hex="collision"))
+    stage = tmp_path / ".sprout-write-collision"
+    stage.write_text("not ours")
+    with pytest.raises(OSError, match="create stage"):
+        asyncio.run(port.write_file(str(tmp_path / "target"), "new"))
+    assert stage.read_text() == "not ours"
+    assert not (tmp_path / "target").exists()
+
+
+@pytest.mark.parametrize(
+    "path,content",
+    [
+        ("relative", "value"),
+        ("/", "value"),
+        ("/target", "x" * (4 * 1024 * 1024 + 1)),
+    ],
+)
+def test_invalid_write_has_no_sandbox_effect(tmp_path, port, path, content):
+    with pytest.raises(ValueError):
+        asyncio.run(port.write_file(path, content))
+    assert not port.call_ids
+
+
+def test_native_small_str_replace_in_large_existing_file(tmp_path):
+    root = tmp_path / "repo"
+    root.mkdir()
+    target = root / "large.py"
+    original = "# padding\n" * 30000 + "old_unique_value = 1\n"
+    target.write_text(original)
+    result, _, journal = run(
+        root,
+        tmp_path / "native",
+        Model(
+            [
+                response(
+                    call(
+                        "edit",
+                        "file_write",
+                        path="large.py",
+                        mode="str_replace",
+                        old_str="old_unique_value = 1",
+                        new_str="new_unique_value = 2",
+                    )
+                ),
+                response(content="Done."),
+            ]
+        ),
+    )
+    assert result.status == "completed", result.error
+    assert target.read_text() == original.replace("old_unique_value = 1", "new_unique_value = 2")
+    calls = [row for row in read_journal(journal) if row["type"] == "tool.started"]
+    assert len(calls) > 5
+    assert max(len(row["args"]["command"].encode()) for row in calls) < 65536
+
+
+def test_same_runtime_binary_accepts_large_write(tmp_path, port):
+    """No Podman/global cleanup fixtures: only a private local runtime process."""
+    import httpx
+
+    from sprout.worker.tools.result import CommandOutcome, ToolResult
+    from tests.support.paths import RUNTIME_BIN
+
+    binary = RUNTIME_BIN
+    if not binary.is_file():
+        pytest.skip("prebuilt runtime unavailable")
+    # Unix socket path length is limited, independent of shell argument size.
+    import tempfile
+
+    with tempfile.TemporaryDirectory(prefix="sprout-write-") as temporary:
+        socket = Path(temporary) / "r.sock"
+        process = subprocess.Popen(
+            [str(binary), "--listen", f"unix:{socket}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
+        )
+        try:
+            for _ in range(100):
+                if socket.exists():
+                    break
+                assert process.poll() is None
+                time.sleep(0.02)
+            with httpx.Client(
+                transport=httpx.HTTPTransport(uds=str(socket)), base_url="http://runtime", timeout=10
+            ) as client:
+
+                def call_runtime(name, args, timeout):
+                    import json
+
+                    reply = client.post(
+                        "/",
+                        json={
+                            "jsonrpc": "2.0",
+                            "id": 1,
+                            "method": "tools/call",
+                            "params": {"name": name, "arguments": args},
+                        },
+                    )
+                    reply.raise_for_status()
+                    result = reply.json()["result"]
+                    output = result["content"][0]["text"]
+                    if result.get("isError"):
+                        value = json.loads(output)
+                        outcome = CommandOutcome(
+                            exit_code=value["exit_code"], stdout=value["stdout"], stderr=value["stderr"]
+                        )
+                    else:
+                        outcome = CommandOutcome(exit_code=0, stdout=output)
+                    return ToolResult(not result.get("isError"), output, outcome=outcome)
+
+                port.tools.session.call = call_runtime
+                target = tmp_path / "runtime-large.txt"
+                content = "runtime 中文\n" * 30000
+                asyncio.run(port.write_file(str(target), content))
+                assert target.read_bytes() == content.encode()
+        finally:
+            process.terminate()
+            process.wait(timeout=8)
diff --git a/tests/worker/agents/opencollab/test_review50_protocol.py b/tests/worker/agents/opencollab/test_review50_protocol.py
new file mode 100644
index 00000000..d12e8fef
--- /dev/null
+++ b/tests/worker/agents/opencollab/test_review50_protocol.py
@@ -0,0 +1,200 @@
+# ruff: noqa: E402
+"""The reviewed tool interface must execute and finish through native Single2."""
+
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("opencollab")
+
+from sprout.contract.agents.history import load_prefix, training_messages
+from sprout.contract.messages import clean_messages, export_tools
+from sprout.worker.agents.history import available_branch_points, reference_at
+from sprout.worker.agents.opencollab.review50 import build_tools, definition
+from sprout.worker.run.journal import JournalWriter, read_journal
+from tests.worker.agents.opencollab.test_opencollab import Model, call, response, run
+
+
+def repository(root: Path) -> Path:
+    root.mkdir()
+    subprocess.run(["git", "init", "-q", str(root)], check=True)
+    (root / "source.txt").write_text("original\n")
+    subprocess.run(["git", "-C", str(root), "add", "."], check=True)
+    # This is isolated test-fixture history, never a benchmark's history.
+    subprocess.run(
+        [
+            "git",
+            "-C",
+            str(root),
+            "-c",
+            "user.name=Fixture",
+            "-c",
+            "user.email=fixture@example.invalid",
+            "commit",
+            "-qm",
+            "fixture",
+        ],
+        check=True,
+    )
+    return root
+
+
+def exported(journal):
+    events = read_journal(journal)
+    ref = next(e for e in events if e["type"] == "session.ref")
+    rows = [json.loads(line) for line in Path(ref["transcript_path"]).read_text().splitlines()]
+    return clean_messages(training_messages(rows)), events
+
+
+def test_review50_file_tools_submission_and_closed_history(tmp_path):
+    root = repository(tmp_path / "repo")
+    model = Model(
+        [
+            response(call("read", "read_file", filepath="source.txt", start_line=1, end_line=1)),
+            response(
+                call("edit", "edit_file", filepath="source.txt", old_string="original", new_string="fixed")
+            ),
+            response(call("write", "write_file", filepath="new.txt", content="new file\n")),
+            response(call("status", "get_status")),
+            response(call("command", "run_command", command="printf verified")),
+            response(call("submit", "submit_patch")),
+        ]
+    )
+    result, _, journal = run(root, tmp_path / "run", model, protocol="review50-v1")
+    assert result.status == "completed", result.error
+    assert len(model.requests) == 6  # no extra model request needed to end
+    messages, events = exported(journal)
+    results = {m["tool_call_id"]: json.loads(m["content"]) for m in messages if m["role"] == "tool"}
+    assert results["read"]["content"] == "original\n"
+    assert results["edit"]["occurrences"] == 1
+    assert results["write"]["size"] == 9
+    assert results["status"]["grading"] == "pending official verifier"
+    assert results["status"]["budget"]["agent_tool_calls"] == 4
+    assert results["status"]["budget"]["max_tool_calls"] is None
+    assert "new.txt" in results["status"]["patch"]["working_tree_status"]
+    assert "source.txt" in results["status"]["patch"]["tracked_diff_stat"]
+    assert results["command"]["stdout"] == "verified"
+    assert results["submit"]["status"] == "submitted"
+    assert results["submit"]["patch_chars"] > 0
+    patch = subprocess.run(
+        ["git", "-C", str(root), "diff", "HEAD"], check=True, text=True, capture_output=True
+    ).stdout
+    assert "new file" in patch and "+fixed" in patch
+    assert export_tools(events) == definition()["tools"]
+    assert messages[0]["content"] == definition()["system_prompt"]
+    assert next(e for e in events if e["type"] == "opencollab.termination")["kind"] == "explicit_submission"
+    assert available_branch_points(journal)
+
+
+def test_review50_submission_rejects_later_calls_but_keeps_pairs(tmp_path):
+    root = repository(tmp_path / "repo")
+    model = Model(
+        [
+            response(
+                call("s", "submit_patch"),
+                call("later", "write_file", filepath="forbidden.txt", content="not executed"),
+            )
+        ]
+    )
+    result, _, journal = run(root, tmp_path / "run", model, protocol="review50-v1")
+    assert result.status == "completed", result.error
+    assert not (root / "forbidden.txt").exists()
+    messages, _ = exported(journal)
+    assert {m["tool_call_id"] for m in messages if m["role"] == "tool"} == {"s", "later"}
+
+
+def test_failed_submission_is_not_a_successful_finish(tmp_path):
+    root = tmp_path / "not-a-repo"
+    root.mkdir()
+    model = Model([response(call("s", "submit_patch")), response(content="No patch available.")])
+    result, _, journal = run(root, tmp_path / "run", model, protocol="review50-v1")
+    assert result.status == "completed", result.error
+    messages, events = exported(journal)
+    assert json.loads(next(m["content"] for m in messages if m["role"] == "tool"))["status"] == "error"
+    assert next(e for e in events if e["type"] == "opencollab.termination")["kind"] == "final_response"
+    assert len(model.requests) == 2
+
+
+def test_review50_branch_inherits_protocol_and_can_submit(tmp_path):
+    root = repository(tmp_path / "repo")
+    result, _, journal = run(
+        root,
+        tmp_path / "parent",
+        Model([response(call("r", "read_file", filepath="source.txt")), response(content="Done.")]),
+        protocol="review50-v1",
+    )
+    assert result.status == "completed", result.error
+    step, point = min(available_branch_points(journal).items())
+    prefix = reference_at(journal, step, point.session_ckpt)
+    assert load_prefix(prefix)[0]["tool_protocol"] == "review50-v1"
+    reviewer = {
+        "role": "assistant",
+        "content": "I will make a focused correction.",
+        "tool_calls": [
+            call("review", "edit_file", filepath="source.txt", old_string="original", new_string="branch")
+        ],
+    }
+    model = Model([response(call("s", "submit_patch"))])
+    child, _, child_journal = run(
+        root, tmp_path / "child", model, prefix=prefix, reviewer=reviewer, protocol="review50-v1"
+    )
+    assert child.status == "completed", child.error
+    assert (root / "source.txt").read_text() == "branch\n"
+    exported(child_journal)
+    with pytest.raises(ValueError, match="protocol differs"):
+        run(root, tmp_path / "wrong", Model([]), prefix=prefix)
+
+
+def test_file_errors_and_repeated_edits_do_not_mutate_ambiguously(tmp_path):
+    root = repository(tmp_path / "repo")
+    (root / "source.txt").write_text("same same\n")
+    result, _, journal = run(
+        root,
+        tmp_path / "run",
+        Model(
+            [
+                response(
+                    call(
+                        "ambiguous",
+                        "edit_file",
+                        filepath="source.txt",
+                        old_string="same",
+                        new_string="different",
+                    )
+                ),
+                response(call("outside", "read_file", filepath="../outside.txt")),
+                response(content="Stopped."),
+            ]
+        ),
+        protocol="review50-v1",
+    )
+    assert result.status == "completed", result.error
+    messages, _ = exported(journal)
+    assert all(json.loads(m["content"])["status"] == "error" for m in messages if m["role"] == "tool")
+    assert (root / "source.txt").read_text() == "same same\n"
+
+
+def test_compatibility_tools_have_independent_state():
+    first, second = build_tools(), build_tools()
+    assert [t.to_openai_schema() for t in first] == definition()["tools"]
+    assert all(a is not b for a, b in zip(first, second, strict=True))
+
+
+def test_status_uses_enforced_rollout_counts_and_does_not_invent_a_tool_limit(tmp_path):
+    import time
+
+    from sprout.worker.run.control import RunControl
+    from sprout.worker.run.rollout import RolloutControls
+
+    with JournalWriter(tmp_path / "controls.jsonl") as journal:
+        controls = RolloutControls(
+            {"model_endpoint": "http://127.0.0.1:1234", "deadline_at": time.time() + 30, "max_turns": 8},
+            journal,
+            RunControl(),
+        )
+        controls.prepare_model_request({})
+        result = controls.status()
+        assert result["model_calls"] == 1 and result["remaining_turns"] == 7
+        assert result["runtime_tool_calls"] == 0 and result["max_tool_calls"] is None