feat: switch to opencode

tests/test_code_review.py

@@ -28,15 +28,14 @@ def serious_report() -> ReviewReport:
     )
 
 
-class FakeCodex:
+class FakeOpenCode:
     def __init__(self, reports: list[ReviewReport]) -> None:
         self.reports = iter(reports)
         self.reviews = 0
         self.revisions = 0
 
-    async def start(self, **_kwargs):
-        self.reviews += 1
-        return "reviewer", next(self.reports)
+    async def create_session(self, *_args):
+        return "reviewer"
 
     async def resume(self, **kwargs):
         if kwargs["result_type"] is ReviewReport:
@@ -60,15 +59,15 @@ class FakePrompts:
 
 
 def objects(rounds: int, reports: list[ReviewReport]):
-    codex = FakeCodex(reports)
+    opencode = FakeOpenCode(reports)
     settings = SimpleNamespace(
         implement_review_rounds=rounds,
         implement_model="model",
-        implement_reasoning="high",
+        implement_variant="high",
     )
     deps = SimpleNamespace(
         settings=settings,
-        codex=codex,
+        opencode=opencode,
         storage=FakeStorage(),
         prompts=FakePrompts(),
         development=SimpleNamespace(description="python"),
@@ -95,12 +94,12 @@ def objects(rounds: int, reports: list[ReviewReport]):
         message="",
         comment_id=1,
     )
-    return CodeReviewLoop(deps), codex, workflow, job  # type: ignore[arg-type]
+    return CodeReviewLoop(deps), opencode, workflow, job  # type: ignore[arg-type]
 
 
 async def test_stops_after_clean_second_review() -> None:
     clean = ReviewReport(summary="Ready", findings=[])
-    loop, codex, workflow, job = objects(4, [serious_report(), clean])
+    loop, opencode, workflow, job = objects(4, [serious_report(), clean])
     _, report = await loop.run(
         job,
         workflow,
@@ -109,12 +108,12 @@ async def test_stops_after_clean_second_review() -> None:
         AgentResult(summary_markdown="initial", tests=[]),
     )
     assert not report.has_serious_findings
-    assert codex.reviews == 2
-    assert codex.revisions == 1
+    assert opencode.reviews == 2
+    assert opencode.revisions == 1
 
 
 async def test_does_not_make_unreviewed_final_revision() -> None:
-    loop, codex, workflow, job = objects(
+    loop, opencode, workflow, job = objects(
         3, [serious_report(), serious_report(), serious_report()]
     )
     _, report = await loop.run(
@@ -125,5 +124,5 @@ async def test_does_not_make_unreviewed_final_revision() -> None:
         AgentResult(summary_markdown="initial", tests=[]),
     )
     assert report.has_serious_findings
-    assert codex.reviews == 3
-    assert codex.revisions == 2
+    assert opencode.reviews == 3
+    assert opencode.revisions == 2

tests/test_codex.py

@@ -1,138 +0,0 @@
-import asyncio
-import tomllib
-from pathlib import Path
-
-from agentci.adapters.codex import CodexClient, _session_id
-from agentci.domain.models import AgentResult
-
-
-def test_enables_codegraph_in_shared_codex_config() -> None:
-    root = Path(__file__).parents[1]
-    config = tomllib.loads((root / "codex" / "config.toml").read_text())
-
-    codegraph = config["mcp_servers"]["codegraph"]
-    assert codegraph["command"] == "codegraph"
-    assert codegraph["args"] == ["serve", "--mcp"]
-    assert codegraph["env"]["CODEGRAPH_TELEMETRY"] == "0"
-
-
-def test_compose_allows_nested_codex_procfs() -> None:
-    root = Path(__file__).parents[1]
-
-    compose = (root / "compose.yaml").read_text()
-
-    assert "systempaths=unconfined" in compose
-
-
-def test_extracts_thread_id_from_jsonl() -> None:
-    output = '\n'.join(
-        [
-            '{"type":"turn.started"}',
-            '{"type":"thread.started","thread_id":"abc-123"}',
-            "not json",
-        ]
-    )
-    assert _session_id(output) == "abc-123"
-
-
-def test_missing_thread_id_returns_none() -> None:
-    assert _session_id('{"type":"turn.completed"}') is None
-
-
-def test_turns_skip_interactive_git_trust_check(tmp_path) -> None:
-    client = CodexClient(
-        codex_home=tmp_path / "codex",
-        schemas_dir=tmp_path / "schemas",
-        timeout_seconds=60,
-        research_model="gpt-5.6-luna",
-        research_reasoning="high",
-        context7_api_key=None,
-    )
-
-    args = client._turn_args("model", "medium", "agentci-read", "plan.json")
-
-    assert "--skip-git-repo-check" in args
-
-
-def test_configures_research_agent_and_optional_context7_key(tmp_path) -> None:
-    client = CodexClient(
-        codex_home=tmp_path / "codex",
-        schemas_dir=tmp_path / "schemas",
-        timeout_seconds=60,
-        research_model="research-model",
-        research_reasoning="high",
-        context7_api_key="ctx7-secret",
-    )
-
-    agent = (tmp_path / "codex" / "agents" / "research.toml").read_text()
-    parsed = tomllib.loads(agent)
-    assert 'name = "research"' in agent
-    assert parsed["model"] == "research-model"
-    assert parsed["model_reasoning_effort"] == "high"
-    assert parsed["web_search"] == "live"
-    assert parsed["mcp_servers"]["codegraph"]["enabled"] is False
-    assert 'url = "https://mcp.context7.com/mcp"' in agent
-    assert 'url = "https://mcp.grep.app"' in agent
-    assert "ctx7-secret" not in agent
-    assert client._environment()["CONTEXT7_API_KEY"] == "ctx7-secret"
-
-
-def test_exposes_development_tools_without_service_secrets(tmp_path, monkeypatch) -> None:
-    monkeypatch.setenv("PATH", "/usr/bin")
-    monkeypatch.setenv("AGENTCI_PRIVATE_VALUE", "secret")
-    tools_bin = tmp_path / "tools" / "bin"
-    client = CodexClient(
-        codex_home=tmp_path / "codex",
-        schemas_dir=tmp_path / "schemas",
-        timeout_seconds=60,
-        research_model="gpt-5.6-luna",
-        research_reasoning="high",
-        context7_api_key=None,
-        tools_bin=tools_bin,
-    )
-
-    environment = client._environment()
-
-    assert environment["PATH"] == f"{tools_bin}:/usr/bin"
-    assert "AGENTCI_PRIVATE_VALUE" not in environment
-
-
-async def test_invokes_codex_from_workflow_workspace(tmp_path, monkeypatch) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    captured: dict[str, object] = {}
-
-    class Process:
-        returncode = 0
-
-        async def communicate(self, prompt: bytes) -> tuple[bytes, bytes]:
-            assert prompt == b"prompt"
-            return b'{"type":"thread.started","thread_id":"thread"}', b""
-
-    async def create_subprocess_exec(*args, **kwargs):
-        captured["cwd"] = kwargs["cwd"]
-        output = Path(args[args.index("--output-last-message") + 1])
-        output.write_text(  # noqa: ASYNC240 - tiny test-owned result file
-            '{"summary_markdown":"summary","tests":[]}'
-        )
-        return Process()
-
-    monkeypatch.setattr(asyncio, "create_subprocess_exec", create_subprocess_exec)
-    client = CodexClient(
-        codex_home=tmp_path / "codex",
-        schemas_dir=tmp_path / "schemas",
-        timeout_seconds=60,
-        research_model="gpt-5.6-luna",
-        research_reasoning="high",
-        context7_api_key=None,
-    )
-
-    _, result = await client._invoke(
-        ["codex", "exec", "-"],
-        "prompt",
-        AgentResult,
-        workspace=workspace,
-    )
-
-    assert captured["cwd"] == workspace
-    assert result.summary_markdown == "summary"

tests/test_config.py

@@ -33,3 +33,9 @@ def test_reads_comma_delimited_install_scripts_from_environment(monkeypatch) ->
     settings = Settings(_env_file=None)  # type: ignore[call-arg]
 
     assert settings.install_scripts == ["python", "dotnet"]
+
+
+@pytest.mark.parametrize("field", ["plan_model", "implement_model", "research_model"])
+def test_requires_provider_qualified_opencode_models(field: str) -> None:
+    with pytest.raises(ValidationError, match="provider/model"):
+        Settings(_env_file=None, **{field: "model-only"})  # type: ignore[call-arg]

tests/test_install_scripts.py

@@ -1,12 +1,12 @@
 from pathlib import Path
 
 
-def test_dotnet_wrapper_uses_sandbox_writable_runtime_directories() -> None:
+def test_dotnet_wrapper_uses_persistent_runtime_directories() -> None:
     root = Path(__file__).parents[1]
     script = (root / "install-scripts" / "dotnet").read_text()
 
-    assert 'DOTNET_CLI_HOME="${DOTNET_CLI_HOME:-/tmp/agentci-dotnet}"' in script
-    assert 'NUGET_PACKAGES="${NUGET_PACKAGES:-/tmp/agentci-nuget/packages}"' in script
-    http_cache = 'NUGET_HTTP_CACHE_PATH="${NUGET_HTTP_CACHE_PATH:-/tmp/agentci-nuget/http-cache}"'
-    assert http_cache in script
+    assert "export DOTNET_ROOT=" in script
+    assert "/tmp/agentci-dotnet" not in script
+    assert "DEV_TOOLS_DIR/runtime/dotnet" in script
+    assert "NUGET_PACKAGES" in script
     assert 'export HOME="$DOTNET_CLI_HOME"' in script

tests/test_opencode.py

@@ -0,0 +1,238 @@
+import asyncio
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+
+from agentci.adapters.opencode import OpenCodeClient, OpenCodeError
+from agentci.domain.models import AgentResult
+
+API_DOCUMENT = {
+    "paths": {
+        "/global/health": {"get": {}},
+        "/provider": {"get": {}},
+        "/session": {"post": {}},
+        "/session/{sessionID}/message": {"post": {}},
+        "/session/{sessionID}/abort": {"post": {}},
+    }
+}
+PROVIDERS = {
+    "connected": ["openai"],
+    "all": [
+        {
+            "id": "openai",
+            "models": {
+                "model": {
+                    "status": "active",
+                    "capabilities": {"toolcall": True},
+                    "variants": {"high": {}},
+                }
+            },
+        }
+    ],
+}
+
+
+class FakeCodeGraph:
+    def __init__(self) -> None:
+        self.prepared: list[Path] = []
+
+    async def prepare(self, workspace: Path) -> None:
+        self.prepared.append(workspace)
+
+
+def client(tmp_path: Path, handler, codegraph: FakeCodeGraph | None = None) -> OpenCodeClient:
+    selected_codegraph = codegraph or FakeCodeGraph()
+    return OpenCodeClient(
+        base_url="http://opencode:4096",
+        username="opencode",
+        password="server-secret",
+        schemas_dir=Path(__file__).parents[1] / "src" / "agentci" / "prompts" / "schemas",
+        health_directory=tmp_path,
+        required_models=(("openai/model", None),),
+        timeout_seconds=60,
+        codegraph=selected_codegraph,  # type: ignore[arg-type]
+        transport=httpx.MockTransport(handler),
+    )
+
+
+async def test_ready_requires_healthy_server_and_connected_provider(tmp_path: Path) -> None:
+    expected_directory = str(tmp_path)
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.headers["authorization"].startswith("Basic ")
+        if request.url.path == "/global/health":
+            return httpx.Response(200, json={"healthy": True, "version": "1.18.4"})
+        if request.url.path == "/doc":
+            return httpx.Response(200, json=API_DOCUMENT)
+        assert request.headers["x-opencode-directory"] == expected_directory
+        return httpx.Response(200, json=PROVIDERS)
+
+    value = client(tmp_path, handler)
+    assert await value.ready()
+    await value.close()
+
+
+async def test_ready_rejects_missing_provider(tmp_path: Path) -> None:
+    async def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/global/health":
+            return httpx.Response(200, json={"healthy": True, "version": "1.18.4"})
+        if request.url.path == "/doc":
+            return httpx.Response(200, json=API_DOCUMENT)
+        return httpx.Response(200, json={**PROVIDERS, "connected": ["anthropic"]})
+
+    value = client(tmp_path, handler)
+    assert not await value.ready()
+    await value.close()
+
+
+async def test_starts_structured_session_in_workspace(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    codegraph = FakeCodeGraph()
+    requests: list[httpx.Request] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        assert request.headers["x-opencode-directory"] == str(workspace.resolve())
+        if request.url.path == "/session":
+            return httpx.Response(200, json={"id": "ses_new"})
+        body = json.loads(request.content)
+        assert body["model"] == {"providerID": "openai", "modelID": "model"}
+        assert body["agent"] == "build"
+        assert body["variant"] == "high"
+        assert body["format"]["type"] == "json_schema"
+        assert body["format"]["schema"]["required"] == ["summary_markdown", "tests"]
+        return httpx.Response(
+            200,
+            json={
+                "info": {
+                    "role": "assistant",
+                    "sessionID": "ses_new",
+                    "structured": {"summary_markdown": "summary", "tests": []},
+                },
+                "parts": [],
+            },
+        )
+
+    value = client(tmp_path, handler, codegraph)
+    session_id, result = await value.start(
+        workspace=workspace,
+        prompt="implement",
+        model="openai/model",
+        variant="high",
+        schema_name="agent_result.json",
+        result_type=AgentResult,
+    )
+
+    assert session_id == "ses_new"
+    assert result.summary_markdown == "summary"
+    assert codegraph.prepared == [workspace]
+    assert [request.url.path for request in requests] == [
+        "/session",
+        "/session/ses_new/message",
+    ]
+    await value.close()
+
+
+async def test_retries_invalid_structured_result_on_same_session(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    prompts: list[str] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        prompts.append(body["parts"][0]["text"])
+        if len(prompts) == 1:
+            return httpx.Response(
+                200, json={"info": {"error": {"name": "StructuredOutputError"}}}
+            )
+        return httpx.Response(
+            200,
+            json={
+                "info": {"structured": {"summary_markdown": "fixed", "tests": []}},
+                "parts": [],
+            },
+        )
+
+    value = client(tmp_path, handler)
+    result = await value.resume(
+        session_id="ses_existing",
+        workspace=workspace,
+        prompt="continue",
+        model="openai/model",
+        variant=None,
+        schema_name="agent_result.json",
+        result_type=AgentResult,
+    )
+
+    assert result.summary_markdown == "fixed"
+    assert prompts[0] == "continue"
+    assert "without repeating repository work" in prompts[1]
+    await value.close()
+
+
+async def test_aborts_timed_out_session(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    aborted = False
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal aborted
+        if request.url.path.endswith("/abort"):
+            aborted = True
+            return httpx.Response(200, json=True)
+        raise httpx.ReadTimeout("slow", request=request)
+
+    value = client(tmp_path, handler)
+    with pytest.raises(OpenCodeError, match="exceeded 60 seconds"):
+        await value.resume(
+            session_id="ses_existing",
+            workspace=workspace,
+            prompt="continue",
+            model="openai/model",
+            variant=None,
+            schema_name="agent_result.json",
+            result_type=AgentResult,
+        )
+
+    assert aborted
+    await value.close()
+
+
+async def test_aborts_cancelled_session(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    started = asyncio.Event()
+    aborted = False
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal aborted
+        if request.url.path.endswith("/abort"):
+            aborted = True
+            return httpx.Response(200, json=True)
+        started.set()
+        await asyncio.Event().wait()
+        raise AssertionError("unreachable")
+
+    value = client(tmp_path, handler)
+    turn = asyncio.create_task(
+        value.resume(
+            session_id="ses_existing",
+            workspace=workspace,
+            prompt="continue",
+            model="openai/model",
+            variant=None,
+            schema_name="agent_result.json",
+            result_type=AgentResult,
+        )
+    )
+    await started.wait()
+    turn.cancel()
+
+    with pytest.raises(asyncio.CancelledError):
+        await turn
+
+    assert aborted
+    await value.close()

tests/test_opencode_deployment.py

@@ -0,0 +1,31 @@
+import json
+from pathlib import Path
+
+
+def test_config_grants_all_agents_unrestricted_permissions() -> None:
+    root = Path(__file__).parents[1]
+    config = json.loads((root / "opencode" / "opencode.json").read_text())
+
+    assert config["permission"] == "allow"
+    assert all(
+        config["agent"][name]["permission"] == "allow"
+        for name in ("build", "plan", "general", "explore", "research")
+    )
+    assert config["mcp"]["codegraph"]["command"] == ["codegraph", "serve", "--mcp"]
+    assert config["mcp"]["context7"]["url"] == "https://mcp.context7.com/mcp"
+
+
+def test_compose_removes_codex_sandbox_exceptions() -> None:
+    root = Path(__file__).parents[1]
+    compose = (root / "compose.yaml").read_text()
+    dockerfile = (root / "Dockerfile").read_text()
+
+    for forbidden in ("cap_add", "seccomp=unconfined", "apparmor=unconfined", "bubblewrap"):
+        assert forbidden not in compose
+    assert "no_cache: true" in compose
+    assert "opencode_home:/var/lib/opencode" in compose
+    assert "HOME: /etc/opencode/home" in compose
+    assert "OPENCODE_DISABLE_EXTERNAL_SKILLS" in compose
+    assert "ARG OPENCODE_REFRESH" in dockerfile
+    assert "OPENCODE_REFRESH is required" in dockerfile
+    assert "'opencode-ai@^1'" in dockerfile

tests/test_opencode_support.py

@@ -0,0 +1,40 @@
+from agentci.adapters.opencode_support import api_contract_ready, models_ready
+
+
+def test_api_contract_requires_session_message_and_abort_routes() -> None:
+    valid = {
+        "paths": {
+            "/global/health": {"get": {}},
+            "/provider": {"get": {}},
+            "/session": {"post": {}},
+            "/session/{sessionID}/message": {"post": {}},
+            "/session/{sessionID}/abort": {"post": {}},
+        }
+    }
+
+    assert api_contract_ready(valid)
+    del valid["paths"]["/session/{sessionID}/abort"]
+    assert not api_contract_ready(valid)
+
+
+def test_model_readiness_requires_tools_and_configured_variant() -> None:
+    payload = {
+        "connected": ["openai"],
+        "all": [
+            {
+                "id": "openai",
+                "models": {
+                    "model": {
+                        "status": "active",
+                        "capabilities": {"toolcall": True},
+                        "variants": {"high": {}},
+                    }
+                },
+            }
+        ],
+    }
+
+    assert models_ready(payload, {("openai", "model", "high")})
+    assert not models_ready(payload, {("openai", "model", "missing")})
+    payload["all"][0]["models"]["model"]["capabilities"]["toolcall"] = False
+    assert not models_ready(payload, {("openai", "model", "high")})

tests/test_storage.py

@@ -1,3 +1,4 @@
+import sqlite3
 from pathlib import Path
 
 import pytest
@@ -101,3 +102,50 @@ async def test_failed_followup_does_not_invalidate_completed_workflow(
     loaded = await storage.latest_workflow("alice", "repo", 3, WorkflowKind.PLAN)
     assert loaded is not None
     assert loaded.status is WorkflowStatus.COMPLETED
+
+
+async def test_opencode_migration_preserves_and_tags_legacy_session_ids(tmp_path: Path) -> None:
+    legacy_migrations = tmp_path / "legacy-migrations"
+    legacy_migrations.mkdir()
+    migrations = Path(__file__).parents[1] / "src" / "agentci" / "migrations"
+    (legacy_migrations / "001_initial.sql").write_text(
+        (migrations / "001_initial.sql").read_text()
+    )
+    database = tmp_path / "legacy.sqlite3"
+    legacy = Storage(database, legacy_migrations)
+    await legacy.initialize()
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            """
+            INSERT INTO workflows (
+                id, kind, repo_owner, repo_name, issue_number, base_sha,
+                workspace_path, primary_session_id, reviewer_session_id,
+                artifact, status, created_at, updated_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                "legacy-workflow",
+                "plan",
+                "alice",
+                "repo",
+                3,
+                "abc",
+                str(tmp_path / "repo"),
+                "legacy-primary",
+                "legacy-reviewer",
+                "# Preserved plan",
+                "completed",
+                "2026-07-20T00:00:00+00:00",
+                "2026-07-20T00:00:00+00:00",
+            ),
+        )
+
+    migrated = Storage(database, migrations)
+    await migrated.initialize()
+    loaded = await migrated.latest_workflow("alice", "repo", 3, WorkflowKind.PLAN)
+
+    assert loaded is not None
+    assert loaded.artifact == "# Preserved plan"
+    assert loaded.primary_session_id == "legacy-primary"
+    assert loaded.reviewer_session_id == "legacy-reviewer"
+    assert loaded.runtime == "codex"

tests/test_webhook.py

@@ -29,9 +29,13 @@ class FakeStorage:
 
 
 class FakeGitea:
-    def __init__(self) -> None:
+    def __init__(self, permitted: bool = True) -> None:
+        self.permitted = permitted
         self.comments: list[str] = []
 
+    async def has_write_permission(self, *_args):
+        return self.permitted
+
     async def create_comment(self, _owner, _repo, _number, body):
         self.comments.append(body)
         return len(self.comments)
@@ -72,6 +76,21 @@ async def test_authorized_command_is_queued() -> None:
     assert "queued" in gitea.comments[0]
 
 
+async def test_unauthorized_command_is_rejected_and_deduplicated() -> None:
+    storage = FakeStorage()
+    gitea = FakeGitea(permitted=False)
+    container = SimpleNamespace(storage=storage, gitea=gitea)
+    event = _event_from_payload("delivery", payload("/agent implement"))
+    assert event is not None
+
+    await _handle_command(container, event)
+    await _handle_command(container, event)
+
+    assert storage.jobs == []
+    assert len(gitea.comments) == 1
+    assert "write permission" in gitea.comments[0]
+
+
 async def test_iterate_message_is_preserved_on_queued_job() -> None:
     storage = FakeStorage()
     container = SimpleNamespace(storage=storage, gitea=FakeGitea())

tests/test_worker.py

@@ -0,0 +1,80 @@
+from pathlib import Path
+
+from agentci.domain.models import Job, JobKind, Workflow, WorkflowKind
+from agentci.worker import Worker
+
+
+class FakeStorage:
+    def __init__(self, workflow: Workflow | None) -> None:
+        self.workflow = workflow
+
+    async def get_workflow(self, _workflow_id: str) -> Workflow | None:
+        return self.workflow
+
+
+class FakeOpenCode:
+    def __init__(self) -> None:
+        self.aborted: set[tuple[str, Path]] = set()
+
+    async def abort(self, session_id: str, workspace: Path) -> None:
+        self.aborted.add((session_id, workspace))
+
+
+def job(*, workflow_id: str | None, runtime_session_id: str | None = None) -> Job:
+    return Job(
+        id="job",
+        kind=JobKind.FIX,
+        target_key="org/repo:pr:1",
+        repo_owner="org",
+        repo_name="repo",
+        issue_number=1,
+        pr_number=1,
+        requester="alice",
+        message="",
+        comment_id=1,
+        workflow_id=workflow_id,
+        runtime_session_id=runtime_session_id,
+    )
+
+
+def worker(tmp_path: Path, storage: FakeStorage, opencode: FakeOpenCode) -> Worker:
+    return Worker(
+        storage=storage,  # type: ignore[arg-type]
+        gitea=None,  # type: ignore[arg-type]
+        opencode=opencode,  # type: ignore[arg-type]
+        dispatcher=None,  # type: ignore[arg-type]
+        poll_seconds=1,
+        workspaces_dir=tmp_path,
+    )
+
+
+async def test_recovery_aborts_all_workflow_sessions(tmp_path: Path) -> None:
+    workspace = tmp_path / "workflow" / "repo"
+    workflow = Workflow(
+        id="flow",
+        kind=WorkflowKind.IMPLEMENT,
+        repo_owner="org",
+        repo_name="repo",
+        issue_number=1,
+        workspace_path=workspace,
+        base_sha="base",
+        primary_session_id="primary",
+        reviewer_session_id="reviewer",
+    )
+    opencode = FakeOpenCode()
+
+    await worker(tmp_path, FakeStorage(workflow), opencode)._abort_job_sessions(
+        job(workflow_id=workflow.id)
+    )
+
+    assert opencode.aborted == {("primary", workspace), ("reviewer", workspace)}
+
+
+async def test_recovery_aborts_one_shot_fix_session(tmp_path: Path) -> None:
+    opencode = FakeOpenCode()
+
+    await worker(tmp_path, FakeStorage(None), opencode)._abort_job_sessions(
+        job(workflow_id=None, runtime_session_id="fix-session")
+    )
+
+    assert opencode.aborted == {("fix-session", tmp_path / "fix-job" / "repo")}