refactor tests

Reviewed-on: #4
Reviewed-by: StanPonomarev <dmss@krtss.de>

.gitea/workflows/publish-image.yaml

@@ -3,7 +3,7 @@ name: Publish container image
 on:
   push:
     branches:
-      - master
+      - main
 
 env:
   REGISTRY: git.krtss.de

AGENTS.md

@@ -105,14 +105,15 @@ OpenCode configuration. It may require network access and takes longer than the
 
 ## Security and operational boundaries
 
-- Never commit or expose `.env`, `secrets/`, tokens, passwords, provider credentials, runtime
-  databases, cloned private repositories, or Docker volume contents. Do not send secrets or private
-  repository content to external search or research services.
+- Never commit or expose `.env`, `secrets/`, tokens, passwords, provider credentials, runtime data
+  directories, databases, cloned private repositories, or Docker volume contents. Do not send
+  secrets or private repository content to external search or research services.
 - Preserve webhook HMAC verification, bot-comment filtering, requester write-permission checks, and
   secret-file loading.
 - OpenCode is not an OS sandbox. Do not weaken its permissions, enable repository-local configuration
   or external skills, expose its server to the host, add privileged/capability settings, or add
-  writable host mounts without an explicit security review.
+  writable host mounts beyond the documented `./data/agentci` and `./data/opencode` state directories
+  without an explicit security review.
 - `install-scripts/` is trusted operator code. Keep scripts idempotent, path-safe, and compatible with
   the sanitized environment documented in `install-scripts/README.md`; never pass Agent CI or Gitea
   credentials to them.

README.md

@@ -46,27 +46,34 @@ recreated.
    OpenCode models.
 3. Create `secrets/gitea_token`, `secrets/webhook_secret`, and
    `secrets/opencode_server_password`. Use high-entropy values for both secret/password files.
-4. Log in to the Gitea container registry with a personal access token, then pull the image:
+4. Create the writable state directories for the non-root container user:
+
+   ```sh
+   mkdir -p data/agentci data/opencode
+   sudo chown -R 10001:10001 data/agentci data/opencode
+   ```
+
+5. Log in to the Gitea container registry with a personal access token, then pull the image:
 
    ```sh
    docker login git.krtss.de
    docker compose pull
    ```
 
-5. Authenticate the configured OpenCode providers before starting the persistent server:
+6. Authenticate the configured OpenCode providers before starting the persistent server:
 
    ```sh
    docker compose run --rm opencode opencode auth login
    docker compose run --rm opencode opencode auth list
    ```
 
-6. Start the services:
+7. Start the services:
 
    ```sh
    docker compose up --no-build -d
    ```
 
-7. In Gitea, create a JSON webhook targeting `http://agentci:8080/webhooks/gitea`. Set the same
+8. In Gitea, create a JSON webhook targeting `http://agentci:8080/webhooks/gitea`. Set the same
    webhook secret and subscribe to issue comments, PR timeline comments, and PR review comments.
 
 OpenCode caches provider state. After adding or changing authentication on an already running
@@ -78,7 +85,7 @@ has a connected provider. The worker leaves jobs queued while the runtime is una
 
 ### Image publishing
 
-Every push to `master` runs `.gitea/workflows/publish-image.yaml` and publishes the image as both
+Every push to `main` runs `.gitea/workflows/publish-image.yaml` and publishes the image as both
 `git.krtss.de/stanponomarev/agentci:latest` and
 `git.krtss.de/stanponomarev/agentci:<full-commit-sha>`. Configure these repository Actions secrets
 before the first run:
@@ -129,10 +136,11 @@ OpenCode permits still has the Unix-level access of that container user, so envi
 is only accidental-exposure hygiene and cannot protect readable files from an allowed shell command.
 
 Docker remains the OS boundary. The services run as non-root without added capabilities,
-privileged mode, an unconfined seccomp/AppArmor profile, or a nested `bubblewrap` sandbox. The only
-host bind mount is the read-only installer directory; there are no
-writable host filesystem mounts. The OpenCode HTTP server is not published to the host, is password
-protected, and is reachable by Agent CI over an internal Compose network.
+privileged mode, an unconfined seccomp/AppArmor profile, or a nested `bubblewrap` sandbox. Persistent
+state is exposed through writable host bind mounts at `./data/agentci` and `./data/opencode`; protect
+these directories because they contain private repository clones, installed tools, provider state,
+and resumable sessions. The OpenCode HTTP server is not published to the host, is password protected,
+and is reachable by Agent CI over an internal Compose network.
 
 ## Development environments
 
@@ -146,13 +154,13 @@ AGENTCI_PYTHON_VERSION=3.13
 AGENTCI_DOTNET_CHANNEL=10.0
 ```
 
-Every name resolves to a file in `install-scripts/`, mounted read-only at
-`/etc/agentci/install-scripts`. Names cannot contain paths and duplicates are rejected. Installers
-run after each implementation clone or branch sync and fail the job on an unknown script, timeout,
-or non-zero exit. They receive no Agent CI or Gitea secret values in their environment, but remain
-trusted operator code. Tools persist under `/var/lib/agentci/dev-tools`, and OpenCode can read or
-modify them through shell commands permitted by its active agent policy. See
-`install-scripts/README.md` for the script contract.
+Every name resolves to a file under `/etc/agentci/install-scripts`, copied from `install-scripts/`
+when the image is built. Rebuild the image after adding, replacing, or removing an installer. Names
+cannot contain paths and duplicates are rejected. Installers run after each implementation clone or
+branch sync and fail the job on an unknown script, timeout, or non-zero exit. They receive no Agent CI
+or Gitea secret values in their environment, but remain trusted operator code. Tools persist under
+`./data/agentci/dev-tools`, and OpenCode can read or modify them through shell commands permitted by
+its active agent policy. See `install-scripts/README.md` for the script contract.
 
 Agent CI continues to create branches, validate diffs, commit, and push after OpenCode returns. This
 keeps workflow behavior deterministic, but an OpenCode agent with shell permission can still run
@@ -160,10 +168,10 @@ Git commands itself.
 
 ## State and recovery
 
-The `agentci_data` volume contains SQLite, workflow clones, and installed development runtimes.
-The `opencode_home` volume contains provider authentication, OpenCode's database, and resumable
+The `data/agentci` directory contains SQLite, workflow clones, and installed development runtimes.
+The `data/opencode` directory contains provider authentication, OpenCode's database, and resumable
 sessions. Tea's Gitea token configuration is regenerated in an ephemeral tmpfs and is not copied to
-`opencode_home`. Back up both persistent volumes together.
+`data/opencode`. Back up both persistent directories together.
 
 SQLite stores the current job state, an idempotent event inbox, and durable listener tasks. State
 transitions and workflow creation/linking commit atomically; timestamps are storage metadata rather

compose.yaml

@@ -34,7 +34,6 @@ services:
       - opencode_server_password
     volumes:
       - ./data/agentci:/var/lib/agentci
-      - ./install-scripts:/etc/agentci/install-scripts:ro
     tmpfs:
       - /run/agentci:mode=1777
     expose:

install-scripts/README.md

@@ -2,10 +2,10 @@
 
 This directory supplies the ready-made `python` and `dotnet` scripts. They are not reserved:
 modify, replace, or remove them like any other script. Place other trusted executable install
-scripts here and add the desired file names to `AGENTCI_INSTALL_SCRIPTS`. Compose mounts the
-directory read-only at `/etc/agentci/install-scripts`. Executable files run directly; files without
-executable mode run as POSIX shell scripts through `/bin/sh` so bind mounts do not depend on host
-file-mode preservation.
+scripts here and add the desired file names to `AGENTCI_INSTALL_SCRIPTS`. The Docker build copies
+this directory to `/etc/agentci/install-scripts`, so rebuild the image after changing its contents.
+Executable files run directly; files without executable mode run as POSIX shell scripts through
+`/bin/sh`.
 
 Scripts run from the cloned repository with a sanitized environment. They receive:
 
@@ -13,7 +13,7 @@ Scripts run from the cloned repository with a sanitized environment. They receiv
 - `PATH`: `$DEV_TOOLS_DIR/bin` followed by the service path
 - `PYTHON_VERSION` and `DOTNET_CHANNEL`: configured built-in runtime versions
 
-`DEV_TOOLS_DIR` is shared by jobs through the `agentci_data` volume. Put downloaded SDK/runtime
+`DEV_TOOLS_DIR` is shared by jobs through the `data/agentci` bind mount. Put downloaded SDK/runtime
 files beneath it and install command wrappers or symlinks into `$DEV_TOOLS_DIR/bin`; that `bin`
 directory is prepended to implementation agents' `PATH`. Environment changes made by a script do
 not persist into implementation turns.

tests/conftest.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+from agentci.engine import _sqlite
+from agentci.engine.repository import Repository
+
+
+@pytest.fixture
+async def engine_repository(tmp_path: Path) -> Repository:
+    repository = Repository(tmp_path / "state.sqlite3")
+    await repository.initialize()
+    return repository
+
+
+@dataclass
+class SQLiteClock:
+    now: str = "2026-02-01T00:00:00+00:00"
+
+
+@pytest.fixture
+def sqlite_clock(monkeypatch: pytest.MonkeyPatch) -> SQLiteClock:
+    clock = SQLiteClock()
+    monkeypatch.setattr(_sqlite, "now", lambda: clock.now)
+    return clock

tests/test_code_review.py

@@ -1,11 +1,10 @@
+from collections.abc import Iterable
 from pathlib import Path
 from types import SimpleNamespace
-from typing import Any, cast
 
 from pydantic import BaseModel
 
 from agentci.engine.model import Workflow, WorkflowKind
-from agentci.engine.run import JobRun
 from agentci.workflows.model import (
     AgentResult,
     PlanArtifact,
@@ -13,21 +12,20 @@ from agentci.workflows.model import (
     ReviewReport,
     ReviewSeverity,
 )
-from agentci.workflows.review import (
-    review_implementation_loop,
-    review_implementation_once,
-    review_plan_loop,
-    review_plan_once,
-)
-from agentci.workflows.services import WorkflowServices
+from agentci.workflows.review import review_implementation_loop, review_plan_loop
+from tests.workflow_support import WorkflowHarness, make_workflow_harness
 
 
-def serious_report(summary: str = "Needs work") -> ReviewReport:
+def serious_report(
+    summary: str = "Needs work",
+    *,
+    severity: ReviewSeverity = ReviewSeverity.MAJOR,
+) -> ReviewReport:
     return ReviewReport(
         summary=summary,
         findings=[
             ReviewFinding(
-                severity=ReviewSeverity.MAJOR,
+                severity=severity,
                 title="Missing check",
                 detail="A check is absent.",
                 recommendation="Add it.",
@@ -54,46 +52,15 @@ def minor_report() -> ReviewReport:
     )
 
 
-class RecordingOpenCode:
-    def __init__(self, responses: list[BaseModel]) -> None:
-        self.responses = list(responses)
-        self.created_sessions: list[tuple[Path, str]] = []
-        self.resume_calls: list[dict[str, Any]] = []
-
-    async def create_session(self, workspace: Path, title: str) -> str:
-        self.created_sessions.append((workspace, title))
-        return f"{title}-session"
-
-    async def resume(self, **values: Any) -> BaseModel:
-        self.resume_calls.append(values)
-        response = self.responses.pop(0)
-        assert isinstance(response, values["result_type"])
-        return response
-
-
-class RecordingRepository:
-    def __init__(self) -> None:
+class FakeRepository:
+    def __init__(self, trace: list[tuple[object, ...]] | None = None) -> None:
         self.saved_workflows: list[Workflow] = []
+        self.trace = trace
 
     async def save_workflow(self, workflow: Workflow) -> None:
         self.saved_workflows.append(workflow)
-
-
-class RecordingRun(JobRun):
-    def __init__(self) -> None:
-        self.stages: list[str] = []
-
-    async def stage(self, stage: str) -> None:
-        self.stages.append(stage)
-
-
-class RecordingPrompts:
-    def __init__(self) -> None:
-        self.calls: list[tuple[str, dict[str, str]]] = []
-
-    def render(self, name: str, **values: str) -> str:
-        self.calls.append((name, values))
-        return f"rendered {name}"
+        if self.trace is not None:
+            self.trace.append(("save_workflow", workflow.reviewer_session_id))
 
 
 def workflow(*, reviewer_session_id: str | None = None) -> Workflow:
@@ -110,23 +77,15 @@ def workflow(*, reviewer_session_id: str | None = None) -> Workflow:
     )
 
 
-def objects(
-    responses: list[BaseModel],
+def review_harness(
+    responses: Iterable[BaseModel],
+    repository: FakeRepository,
     *,
+    trace: list[tuple[object, ...]] | None = None,
     plan_rounds: int = 4,
     implementation_rounds: int = 3,
-) -> tuple[
-    RecordingOpenCode,
-    RecordingRepository,
-    RecordingPrompts,
-    WorkflowServices,
-]:
-    opencode = RecordingOpenCode(responses)
-    repository = RecordingRepository()
-    prompts = RecordingPrompts()
-    services = cast(
-        WorkflowServices,
-        SimpleNamespace(
+) -> WorkflowHarness:
+    return make_workflow_harness(
         settings=SimpleNamespace(
             plan_review_rounds=plan_rounds,
             plan_model="provider/plan",
@@ -135,21 +94,21 @@ def objects(
             implement_model="provider/implement",
             implement_variant="high",
         ),
-            opencode=opencode,
         repository=repository,
-            prompts=prompts,
-            development=SimpleNamespace(description="Python 3.13"),
-        ),
+        gitea=object(),
+        responses=responses,
+        trace=trace,
     )
-    return opencode, repository, prompts, services
 
 
 async def test_implementation_loop_persists_reviewed_revision_and_stops_clean() -> None:
     revised = AgentResult(summary_markdown="revision 1", tests=["pytest: passed"])
-    opencode, repository, prompts, services = objects(
-        [serious_report(), revised, clean_report()], implementation_rounds=4
+    repository = FakeRepository()
+    harness = review_harness(
+        [serious_report(), revised, clean_report()],
+        repository,
+        implementation_rounds=4,
     )
-    run = RecordingRun()
     original = workflow()
 
     updated, result, report = await review_implementation_loop(
@@ -157,8 +116,8 @@ async def test_implementation_loop_persists_reviewed_revision_and_stops_clean()
         "issue context",
         "canonical plan",
         AgentResult(summary_markdown="initial", tests=[]),
-        run,
-        services,
+        harness.run,
+        harness.services,
     )
 
     assert result == revised
@@ -169,32 +128,36 @@ async def test_implementation_loop_persists_reviewed_revision_and_stops_clean()
     assert updated.review_json == clean_report().model_dump_json()
     assert repository.saved_workflows[-1] == updated
     assert repository.saved_workflows[0].reviewer_session_id == ("implementation-review-session")
-    assert run.stages == [
+    assert harness.run.stages == [
         "reviewing implementation 1/4",
         "reviewing implementation 2/4",
     ]
-    assert opencode.created_sessions == [(original.workspace_path, "implementation-review")]
-    assert [call["session_id"] for call in opencode.resume_calls] == [
+    assert harness.opencode.created_sessions == [(original.workspace_path, "implementation-review")]
+    assert [call["session_id"] for call in harness.opencode.resume_calls] == [
         "implementation-review-session",
         "primary-session",
         "implementation-review-session",
     ]
-    assert [call["result_type"] for call in opencode.resume_calls] == [
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [
         ReviewReport,
         AgentResult,
         ReviewReport,
     ]
-    assert [name for name, _ in prompts.calls] == [
+    assert [name for name, _ in harness.prompts.calls] == [
         "implementation_review",
         "implementation_revision",
         "implementation_review",
     ]
-    assert opencode.responses == []
+    assert harness.opencode.responses == []
 
 
 async def test_implementation_loop_never_makes_unreviewed_final_revision() -> None:
-    final_report = serious_report("Still failing after the last review")
-    opencode, repository, _, services = objects(
+    final_report = serious_report(
+        "Still failing after the last review",
+        severity=ReviewSeverity.BLOCKING,
+    )
+    repository = FakeRepository()
+    harness = review_harness(
         [
             serious_report("round 1"),
             AgentResult(summary_markdown="revision 1", tests=[]),
@@ -202,17 +165,17 @@ async def test_implementation_loop_never_makes_unreviewed_final_revision() -> No
             AgentResult(summary_markdown="revision 2", tests=[]),
             final_report,
         ],
+        repository,
         implementation_rounds=3,
     )
-    run = RecordingRun()
 
     updated, result, report = await review_implementation_loop(
         workflow(),
         "issue context",
         "canonical plan",
         AgentResult(summary_markdown="initial", tests=[]),
-        run,
-        services,
+        harness.run,
+        harness.services,
     )
 
     assert result.summary_markdown == "revision 2"
@@ -220,106 +183,101 @@ async def test_implementation_loop_never_makes_unreviewed_final_revision() -> No
     assert updated.artifact == result.model_dump_json()
     assert updated.review_json == final_report.model_dump_json()
     assert repository.saved_workflows[-1] == updated
-    assert [call["result_type"] for call in opencode.resume_calls].count(ReviewReport) == 3
-    assert [call["result_type"] for call in opencode.resume_calls].count(AgentResult) == 2
-    assert run.stages == [
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [
+        ReviewReport,
+        AgentResult,
+        ReviewReport,
+        AgentResult,
+        ReviewReport,
+    ]
+    assert harness.run.stages == [
         "reviewing implementation 1/3",
         "reviewing implementation 2/3",
         "reviewing implementation 3/3",
     ]
-    assert opencode.responses == []
-
-
-async def test_implementation_review_once_reuses_existing_reviewer_session() -> None:
-    opencode, repository, prompts, services = objects([clean_report()])
-    existing = workflow(reviewer_session_id="existing-reviewer")
-
-    updated, report = await review_implementation_once(
-        existing,
-        issue_context="issue context",
-        plan="canonical plan",
-        pull_context="pull request context",
-        services=services,
-    )
-
-    assert updated is existing
-    assert report == clean_report()
-    assert opencode.created_sessions == []
-    assert repository.saved_workflows == []
-    assert opencode.resume_calls == [
-        {
-            "session_id": "existing-reviewer",
-            "prompt": "rendered implementation_review",
-            "model": "provider/implement",
-            "variant": "high",
-            "workspace": existing.workspace_path,
-            "schema_name": "review.json",
-            "result_type": ReviewReport,
-        }
-    ]
-    assert prompts.calls == [
-        (
-            "implementation_review",
-            {
-                "issue_context": "issue context",
-                "artifact": "canonical plan",
-                "pull_context": "pull request context",
-            },
-        )
-    ]
+    assert harness.opencode.responses == []
 
 
 async def test_plan_loop_revises_serious_finding_then_persists_clean_result() -> None:
     revised = PlanArtifact(plan_markdown="Revised plan")
-    opencode, repository, prompts, services = objects(
-        [serious_report(), revised, clean_report()], plan_rounds=4
+    trace: list[tuple[object, ...]] = []
+    repository = FakeRepository(trace)
+    harness = review_harness(
+        [serious_report(), revised, clean_report()],
+        repository,
+        trace=trace,
+        plan_rounds=4,
     )
-    run = RecordingRun()
     original = workflow()
     initial = PlanArtifact(plan_markdown="Initial plan")
 
     updated, artifact, report = await review_plan_loop(
-        original, "issue context", initial, run, services
+        original,
+        "issue context",
+        initial,
+        harness.run,
+        harness.services,
     )
 
     assert artifact is revised
     assert report == clean_report()
+    assert original.reviewer_session_id is None
     assert updated.reviewer_session_id == "plan-review-session"
     assert updated.artifact == "Revised plan"
     assert updated.review_json == clean_report().model_dump_json()
     assert repository.saved_workflows[-1] == updated
-    assert run.stages == ["reviewing plan 1/4", "reviewing plan 2/4"]
-    assert [call["session_id"] for call in opencode.resume_calls] == [
+    assert harness.run.stages == ["reviewing plan 1/4", "reviewing plan 2/4"]
+    assert harness.opencode.created_sessions == [(original.workspace_path, "plan-review")]
+    assert harness.opencode.resume_calls[0] == {
+        "session_id": "plan-review-session",
+        "workspace": original.workspace_path,
+        "prompt": "rendered plan_review",
+        "model": "provider/plan",
+        "variant": "high",
+        "schema_name": "review.json",
+        "result_type": ReviewReport,
+    }
+    assert harness.prompts.calls[0] == (
+        "plan_review",
+        {"context": "issue context", "artifact": "Initial plan"},
+    )
+    assert trace[:3] == [
+        ("create_session", "plan-review", "plan-review-session"),
+        ("save_workflow", "plan-review-session"),
+        ("resume", "plan-review-session", ReviewReport),
+    ]
+    assert [call["session_id"] for call in harness.opencode.resume_calls] == [
         "plan-review-session",
         "primary-session",
         "plan-review-session",
     ]
-    assert [name for name, _ in prompts.calls] == [
+    assert [name for name, _ in harness.prompts.calls] == [
         "plan_review",
         "plan_revision",
         "plan_review",
     ]
-    assert opencode.responses == []
+    assert harness.opencode.responses == []
 
 
 async def test_plan_loop_stops_at_round_boundary_without_unreviewed_revision() -> None:
     final_report = serious_report("round 2")
-    opencode, repository, _, services = objects(
+    repository = FakeRepository()
+    harness = review_harness(
         [
             serious_report("round 1"),
             PlanArtifact(plan_markdown="Only revision"),
             final_report,
         ],
+        repository,
         plan_rounds=2,
     )
-    run = RecordingRun()
 
     updated, artifact, report = await review_plan_loop(
         workflow(),
         "issue context",
         PlanArtifact(plan_markdown="Initial plan"),
-        run,
-        services,
+        harness.run,
+        harness.services,
     )
 
     assert artifact.plan_markdown == "Only revision"
@@ -327,45 +285,22 @@ async def test_plan_loop_stops_at_round_boundary_without_unreviewed_revision() -
     assert updated.artifact == "Only revision"
     assert updated.review_json == final_report.model_dump_json()
     assert repository.saved_workflows[-1] == updated
-    assert [call["result_type"] for call in opencode.resume_calls] == [
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [
         ReviewReport,
         PlanArtifact,
         ReviewReport,
     ]
-    assert run.stages == ["reviewing plan 1/2", "reviewing plan 2/2"]
-    assert opencode.responses == []
-
-
-async def test_plan_review_once_creates_and_persists_reviewer_session() -> None:
-    opencode, repository, prompts, services = objects([clean_report()])
-    original = workflow()
-
-    updated, report = await review_plan_once(
-        original,
-        "issue context",
-        PlanArtifact(plan_markdown="Plan body"),
-        services,
-    )
-
-    assert report == clean_report()
-    assert updated is not original
-    assert original.reviewer_session_id is None
-    assert updated.reviewer_session_id == "plan-review-session"
-    assert repository.saved_workflows == [updated]
-    assert opencode.created_sessions == [(original.workspace_path, "plan-review")]
-    assert opencode.resume_calls[0]["session_id"] == "plan-review-session"
-    assert opencode.resume_calls[0]["result_type"] is ReviewReport
-    assert prompts.calls == [
-        (
-            "plan_review",
-            {"context": "issue context", "artifact": "Plan body"},
-        )
-    ]
+    assert harness.run.stages == ["reviewing plan 1/2", "reviewing plan 2/2"]
+    assert harness.opencode.responses == []
 
 
 async def test_minor_findings_end_review_loop_without_revision() -> None:
-    opencode, repository, _, services = objects([minor_report()], implementation_rounds=5)
-    run = RecordingRun()
+    repository = FakeRepository()
+    harness = review_harness(
+        [minor_report()],
+        repository,
+        implementation_rounds=5,
+    )
     initial = AgentResult(summary_markdown="initial", tests=[])
 
     updated, result, report = await review_implementation_loop(
@@ -373,8 +308,8 @@ async def test_minor_findings_end_review_loop_without_revision() -> None:
         "issue context",
         "canonical plan",
         initial,
-        run,
-        services,
+        harness.run,
+        harness.services,
     )
 
     assert result is initial
@@ -383,5 +318,5 @@ async def test_minor_findings_end_review_loop_without_revision() -> None:
     assert updated.artifact == initial.model_dump_json()
     assert updated.review_json == minor_report().model_dump_json()
     assert repository.saved_workflows[-1] == updated
-    assert [call["result_type"] for call in opencode.resume_calls] == [ReviewReport]
-    assert run.stages == ["reviewing implementation 1/5"]
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [ReviewReport]
+    assert harness.run.stages == ["reviewing implementation 1/5"]

tests/test_codegraph.py

@@ -1,4 +1,7 @@
+import asyncio
+from collections.abc import Sequence
 from pathlib import Path
+from typing import Any
 
 import pytest
 
@@ -14,53 +17,75 @@ class FakeProcess:
         return b"", self.stderr
 
 
-async def test_initializes_incomplete_index_and_excludes_it_from_git(
-    tmp_path: Path, monkeypatch
-) -> None:
-    workspace = tmp_path / "repo"
-    (workspace / ".git" / "info").mkdir(parents=True)
-    (workspace / ".codegraph").mkdir()
-    calls: list[tuple[object, ...]] = []
+class CodeGraphProcessRecorder:
+    def __init__(self, outcomes: Sequence[FakeProcess | OSError]) -> None:
+        self.outcomes = list(outcomes)
+        self.calls: list[tuple[tuple[Any, ...], dict[str, Any]]] = []
 
-    async def create_subprocess_exec(*args, **_kwargs):
-        calls.append(args)
-        return FakeProcess()
+    async def __call__(self, *args: Any, **kwargs: Any) -> FakeProcess:
+        self.calls.append((args, kwargs))
+        outcome = self.outcomes.pop(0)
+        if isinstance(outcome, OSError):
+            raise outcome
+        return outcome
 
+    @property
+    def commands(self) -> list[tuple[Any, ...]]:
+        return [args for args, _ in self.calls]
+
+
+def install_recorder(
+    monkeypatch: pytest.MonkeyPatch,
+    outcomes: Sequence[FakeProcess | OSError],
+) -> CodeGraphProcessRecorder:
+    recorder = CodeGraphProcessRecorder(outcomes)
     monkeypatch.setattr(
         "agentci.integrations.codegraph.asyncio.create_subprocess_exec",
-        create_subprocess_exec,
+        recorder,
     )
-
-    await CodeGraph().prepare(workspace)
-
-    assert calls == [("codegraph", "init", str(workspace))]
-    assert (workspace / ".git" / "info" / "exclude").read_text() == ".codegraph/\n"
+    return recorder
 
 
-async def test_syncs_an_existing_index_without_duplicating_exclude(
-    tmp_path: Path, monkeypatch
+@pytest.mark.parametrize(
+    ("database_exists", "initial_exclude", "expected_command", "expected_exclude"),
+    [
+        pytest.param(False, None, "init", ".codegraph/\n", id="incomplete-index"),
+        pytest.param(
+            True,
+            "# local excludes\n.codegraph/\n",
+            "sync",
+            "# local excludes\n.codegraph/\n",
+            id="existing-index",
+        ),
+    ],
+)
+async def test_prepare_selects_init_or_sync_and_updates_git_exclude(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    database_exists: bool,
+    initial_exclude: str | None,
+    expected_command: str,
+    expected_exclude: str,
 ) -> None:
     workspace = tmp_path / "repo"
     (workspace / ".git" / "info").mkdir(parents=True)
     (workspace / ".codegraph").mkdir()
+    if database_exists:
         (workspace / ".codegraph" / "codegraph.db").touch()
     exclude = workspace / ".git" / "info" / "exclude"
-    exclude.write_text("# local excludes\n.codegraph/\n")
-    calls: list[tuple[object, ...]] = []
-
-    async def create_subprocess_exec(*args, **_kwargs):
-        calls.append(args)
-        return FakeProcess()
-
-    monkeypatch.setattr(
-        "agentci.integrations.codegraph.asyncio.create_subprocess_exec",
-        create_subprocess_exec,
-    )
+    if initial_exclude is not None:
+        exclude.write_text(initial_exclude)
+    recorder = install_recorder(monkeypatch, [FakeProcess()])
 
     await CodeGraph().prepare(workspace)
 
-    assert calls == [("codegraph", "sync", str(workspace))]
-    assert exclude.read_text() == "# local excludes\n.codegraph/\n"
+    assert recorder.commands == [("codegraph", expected_command, str(workspace))]
+    kwargs = recorder.calls[0][1]
+    assert kwargs["cwd"] == workspace
+    assert kwargs["env"]["CODEGRAPH_TELEMETRY"] == "0"
+    assert kwargs["stdout"] is asyncio.subprocess.PIPE
+    assert kwargs["stderr"] is asyncio.subprocess.PIPE
+    assert exclude.read_text() == expected_exclude
 
 
 @pytest.mark.parametrize(
@@ -82,13 +107,7 @@ async def test_handles_exclude_file_boundaries(
     exclude.parent.mkdir(parents=True)
     exclude.write_text(initial)
 
-    async def create_subprocess_exec(*_args, **_kwargs):
-        return FakeProcess()
-
-    monkeypatch.setattr(
-        "agentci.integrations.codegraph.asyncio.create_subprocess_exec",
-        create_subprocess_exec,
-    )
+    install_recorder(monkeypatch, [FakeProcess()])
 
     await CodeGraph().prepare(workspace)
 
@@ -99,12 +118,9 @@ async def test_reports_missing_executable(tmp_path: Path, monkeypatch: pytest.Mo
     workspace = tmp_path / "repo"
     workspace.mkdir()
 
-    async def create_subprocess_exec(*_args, **_kwargs):
-        raise FileNotFoundError(2, "No such file or directory", "codegraph")
-
-    monkeypatch.setattr(
-        "agentci.integrations.codegraph.asyncio.create_subprocess_exec",
-        create_subprocess_exec,
+    install_recorder(
+        monkeypatch,
+        [FileNotFoundError(2, "No such file or directory", "codegraph")],
     )
 
     with pytest.raises(CodeGraphError, match="Could not run CodeGraph:.*codegraph") as raised:
@@ -121,12 +137,9 @@ async def test_reports_nonzero_exit_with_bounded_non_utf8_stderr(
     workspace.mkdir()
     stderr = b"discarded-prefix" + (b"x" * 1200) + b"\xff useful-tail"
 
-    async def create_subprocess_exec(*_args, **_kwargs):
-        return FakeProcess(returncode=7, stderr=stderr)
-
-    monkeypatch.setattr(
-        "agentci.integrations.codegraph.asyncio.create_subprocess_exec",
-        create_subprocess_exec,
+    install_recorder(
+        monkeypatch,
+        [FakeProcess(returncode=7, stderr=stderr)],
     )
 
     with pytest.raises(CodeGraphError, match="codegraph init failed") as raised:

tests/test_commands.py

@@ -8,27 +8,45 @@ def test_ignores_non_commands() -> None:
     assert parse_command("please run /agent plan") is None
 
 
-@pytest.mark.parametrize("name", list(CommandName))
-@pytest.mark.parametrize("line_breaks", [1, 2, 5])
-def test_all_commands_accept_messages_after_any_number_of_lines(
-    name: CommandName, line_breaks: int
-) -> None:
-    command = parse_command(
-        f"/agent {name.value}{'\n' * line_breaks}focus on the API\nand add tests"
+@pytest.mark.parametrize(
+    ("name", "is_pull_request", "expected_kind"),
+    [
+        (CommandName.PLAN, False, JobKind.PLAN),
+        (CommandName.DISCUSS, False, JobKind.DISCUSS),
+        (CommandName.IMPLEMENT, False, JobKind.IMPLEMENT),
+        (CommandName.ITERATE, False, JobKind.ITERATE_PLAN),
+        (CommandName.ITERATE, True, JobKind.ITERATE_IMPLEMENT),
+        (CommandName.FIX, True, JobKind.FIX),
+    ],
 )
-    assert command is not None
-    assert command.name is name
-    assert command.message == "focus on the API\nand add tests"
-
-
-@pytest.mark.parametrize("name", list(CommandName))
-def test_all_commands_accept_crlf_separated_multiline_messages(
+def test_supported_commands_parse_and_resolve(
     name: CommandName,
+    is_pull_request: bool,
+    expected_kind: JobKind,
 ) -> None:
-    command = parse_command(f"/agent {name.value}\r\n\r\n\r\nfocus on the API\r\nand add tests")
+    command = parse_command(f"/agent {name.value}\nfocus on the API\nand add tests")
+
     assert command is not None
-    assert command.name is name
-    assert command.message == "focus on the API\r\nand add tests"
+    assert (command.name, command.message) == (
+        name,
+        "focus on the API\nand add tests",
+    )
+    assert resolve_job_kind(command, is_pull_request=is_pull_request) is expected_kind
+
+
+@pytest.mark.parametrize(
+    ("separator", "line_breaks"),
+    [("\n", 1), ("\n", 2), ("\n", 5), ("\r\n", 3)],
+)
+def test_multiline_messages_accept_line_separators(
+    separator: str,
+    line_breaks: int,
+) -> None:
+    message = f"focus on the API{separator}and add tests"
+    command = parse_command(f"/agent {CommandName.PLAN.value}{separator * line_breaks}{message}")
+
+    assert command is not None
+    assert (command.name, command.message) == (CommandName.PLAN, message)
 
 
 def test_discuss_requires_message() -> None:
@@ -36,11 +54,25 @@ def test_discuss_requires_message() -> None:
         parse_command("/agent discuss")
 
 
-def test_rejects_wrong_location() -> None:
-    command = parse_command("/agent fix")
+@pytest.mark.parametrize(
+    ("name", "is_pull_request", "error_match"),
+    [
+        (CommandName.FIX, False, "pull request"),
+        (CommandName.PLAN, True, "issue"),
+        (CommandName.DISCUSS, True, "issue"),
+        (CommandName.IMPLEMENT, True, "issue"),
+    ],
+)
+def test_rejects_wrong_location(
+    name: CommandName,
+    is_pull_request: bool,
+    error_match: str,
+) -> None:
+    command = parse_command(f"/agent {name.value} details")
     assert command is not None
-    with pytest.raises(CommandError, match="pull request"):
-        resolve_job_kind(command, is_pull_request=False)
+
+    with pytest.raises(CommandError, match=error_match):
+        resolve_job_kind(command, is_pull_request=is_pull_request)
 
 
 @pytest.mark.parametrize(
@@ -48,10 +80,6 @@ def test_rejects_wrong_location() -> None:
     [
         ("/agent iterate", ""),
         ("/agent iterate refine tests", "refine tests"),
-        (
-            "/agent iterate\n\nkeep the API stable\nlimit changes to the parser",
-            "keep the API stable\nlimit changes to the parser",
-        ),
     ],
 )
 def test_iterate_accepts_optional_message(body: str, message: str) -> None:
@@ -59,5 +87,3 @@ def test_iterate_accepts_optional_message(body: str, message: str) -> None:
     assert command is not None
     assert command.name is CommandName.ITERATE
     assert command.message == message
-    assert resolve_job_kind(command, is_pull_request=False) is JobKind.ITERATE_PLAN
-    assert resolve_job_kind(command, is_pull_request=True) is JobKind.ITERATE_IMPLEMENT

tests/test_config.py

@@ -36,38 +36,24 @@ def test_normalizes_service_urls(field: str) -> None:
 
 
 @pytest.mark.parametrize(
-    ("field", "boundaries"),
+    ("field", "accepted", "rejected"),
     [
-        ("plan_review_rounds", (1, 20)),
-        ("implement_review_rounds", (1, 20)),
-        ("turn_timeout_seconds", (60,)),
-        ("install_script_timeout_seconds", (1,)),
-        ("worker_poll_seconds", (0.1,)),
-        ("max_concurrent_jobs", (1, 32)),
+        ("plan_review_rounds", (1, 20), (0, 21)),
+        ("implement_review_rounds", (1, 20), (0, 21)),
+        ("turn_timeout_seconds", (60,), (59,)),
+        ("install_script_timeout_seconds", (1,), (0,)),
+        ("worker_poll_seconds", (0.1,), (0.09,)),
+        ("max_concurrent_jobs", (1, 32), (0, 33)),
     ],
 )
-def test_accepts_documented_numeric_boundaries(
-    field: str, boundaries: tuple[int | float, ...]
+def test_enforces_documented_numeric_boundaries(
+    field: str,
+    accepted: tuple[int | float, ...],
+    rejected: tuple[int | float, ...],
 ) -> None:
-    for boundary in boundaries:
-        assert getattr(settings(**{field: boundary}), field) == boundary
-
-
-@pytest.mark.parametrize(
-    ("field", "value"),
-    [
-        ("plan_review_rounds", 0),
-        ("plan_review_rounds", 21),
-        ("implement_review_rounds", 0),
-        ("implement_review_rounds", 21),
-        ("turn_timeout_seconds", 59),
-        ("install_script_timeout_seconds", 0),
-        ("worker_poll_seconds", 0.09),
-        ("max_concurrent_jobs", 0),
-        ("max_concurrent_jobs", 33),
-    ],
-)
-def test_rejects_values_outside_numeric_boundaries(field: str, value: int | float) -> None:
+    for value in accepted:
+        assert getattr(settings(**{field: value}), field) == value
+    for value in rejected:
         with pytest.raises(ValidationError):
             settings(**{field: value})
 
@@ -119,21 +105,23 @@ def test_derived_state_paths_follow_data_directory(tmp_path: Path) -> None:
     assert value.workspaces_dir == data_dir / "workspaces"
 
 
-def test_parses_comma_delimited_install_scripts() -> None:
-    value = settings(install_scripts=" python, dotnet, company-tools, ")
-
-    assert value.install_scripts == ["python", "dotnet", "company-tools"]
-
-
 @pytest.mark.parametrize("value", ["../script", "tools/setup", "python,python"])
 def test_rejects_unsafe_or_duplicate_install_scripts(value: str) -> None:
     with pytest.raises(ValidationError):
         settings(install_scripts=value)
 
 
-@pytest.mark.parametrize("value", ["", None, []])
-def test_empty_install_scripts_disable_setup(value: object) -> None:
-    assert settings(install_scripts=value).install_scripts == []
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        (" python, dotnet, company-tools, ", ["python", "dotnet", "company-tools"]),
+        ("", []),
+        (None, []),
+        ([], []),
+    ],
+)
+def test_normalizes_install_scripts(value: object, expected: list[str]) -> None:
+    assert settings(install_scripts=value).install_scripts == expected
 
 
 def test_agent_defaults_select_expected_capacity_and_research_models() -> None:

tests/test_context.py

@@ -22,7 +22,7 @@ class FakeRepository:
         return {2}
 
 
-class FakePullRequestGitea:
+class PopulatedPullRequestGitea:
     async def pull_request(self, *_args):
         return PullRequestInfo(
             number=3,
@@ -38,29 +38,61 @@ class FakePullRequestGitea:
         )
 
     async def issue_comments(self, *_args):
-        return [CommentInfo(3, "bob", "Please add a test.", "2026-01-03")]
+        return [
+            CommentInfo(
+                1,
+                "alice",
+                "First timeline comment: Please add a test.",
+                "2026-01-01",
+            ),
+            CommentInfo(2, "bob", "Second timeline comment", "2026-01-02"),
+        ]
 
     async def pull_reviews(self, *_args):
         return [
             {
-                "id": 4,
+                "id": 10,
                 "user": {"login": "carol"},
                 "state": "REQUEST_CHANGES",
-                "body": "One issue remains.",
-            }
+                "body": "First formal review: One issue remains.",
+            },
+            {
+                "id": 11,
+                "user": {"login": "dave"},
+                "state": "APPROVED",
+                "body": "Second formal review",
+            },
         ]
 
     async def review_comments(self, *_args):
+        review_id = _args[-1]
+        if review_id == 10:
             return [
+                {
+                    "path": "src/old.py",
+                    "new_position": None,
+                    "old_position": 21,
+                    "body": "Old-side position",
+                }
+            ]
+        return [
+            {
+                "path": "src/new.py",
+                "new_position": 34,
+                "body": "New-side position",
+            },
             {
                 "path": "src/widget.py",
                 "new_position": 12,
                 "body": "Handle the empty value.",
-            }
+            },
         ]
 
     async def pull_commits(self, *_args):
-        return [{"sha": "abcdef1234567890", "commit": {"message": "Fix widget"}}]
+        return [
+            {"sha": "111111111111aaaa", "commit": {"message": "First commit"}},
+            {"sha": "222222222222bbbb", "commit": {"message": "Second commit"}},
+        ]
 
 
 class EmptyIssueGitea:
@@ -104,55 +136,6 @@ class EmptyPullRequestGitea:
         raise AssertionError("review comments should not be requested without reviews")
 
 
-class OrderedPullRequestGitea(FakePullRequestGitea):
-    async def issue_comments(self, *_args):
-        return [
-            CommentInfo(1, "alice", "First timeline comment", "2026-01-01"),
-            CommentInfo(2, "bob", "Second timeline comment", "2026-01-02"),
-        ]
-
-    async def pull_reviews(self, *_args):
-        return [
-            {
-                "id": 10,
-                "user": {"login": "carol"},
-                "state": "REQUEST_CHANGES",
-                "body": "First formal review",
-            },
-            {
-                "id": 11,
-                "user": {"login": "dave"},
-                "state": "APPROVED",
-                "body": "Second formal review",
-            },
-        ]
-
-    async def review_comments(self, *_args):
-        review_id = _args[-1]
-        if review_id == 10:
-            return [
-                {
-                    "path": "src/old.py",
-                    "new_position": None,
-                    "old_position": 21,
-                    "body": "Old-side position",
-                }
-            ]
-        return [
-            {
-                "path": "src/new.py",
-                "new_position": 34,
-                "body": "New-side position",
-            }
-        ]
-
-    async def pull_commits(self, *_args):
-        return [
-            {"sha": "111111111111aaaa", "commit": {"message": "First commit"}},
-            {"sha": "222222222222bbbb", "commit": {"message": "Second commit"}},
-        ]
-
-
 async def test_issue_context_excludes_operational_comments() -> None:
     context = await build_issue_context(
         cast(Gitea, FakeIssueGitea()),
@@ -167,16 +150,31 @@ async def test_issue_context_excludes_operational_comments() -> None:
     assert "Agent job queued" not in context
 
 
-async def test_pull_request_context_includes_feedback_and_commits() -> None:
+async def test_pull_request_context_includes_ordered_feedback_reviews_and_commits() -> None:
     pull, context = await build_pull_request_context(
-        cast(Gitea, FakePullRequestGitea()), "org", "repo", 3
+        cast(Gitea, PopulatedPullRequestGitea()), "org", "repo", 3
     )
 
-    assert pull.number == 3
+    assert pull == PullRequestInfo(
+        number=3,
+        title="Fix widget",
+        body="Fixes the failure.",
+        state="open",
+        merged=False,
+        base_branch="main",
+        head_branch="fix-widget",
+        head_sha="abcdef1234567890",
+        head_owner="alice",
+        head_repo="repo",
+    )
     assert "Please add a test." in context
     assert "One issue remains." in context
+    assert context.index("First timeline comment") < context.index("Second timeline comment")
+    assert context.index("111111111111 First commit") < context.index("222222222222 Second commit")
+    assert context.index("Review 10 by carol") < context.index("Review 11 by dave")
+    assert "`src/old.py:21`: Old-side position" in context
+    assert "`src/new.py:34`: New-side position" in context
     assert "`src/widget.py:12`: Handle the empty value." in context
-    assert "abcdef123456 Fix widget" in context
 
 
 async def test_issue_context_labels_empty_body_and_discussion() -> None:
@@ -201,15 +199,3 @@ async def test_pull_request_context_labels_empty_sections() -> None:
     assert "## Commits\n(none)" in context
     assert "## Timeline discussion\n(none)" in context
     assert "## Formal and inline reviews\n(none)" in context
-
-
-async def test_pull_request_context_preserves_source_order_and_positions() -> None:
-    _, context = await build_pull_request_context(
-        cast(Gitea, OrderedPullRequestGitea()), "org", "repo", 3
-    )
-
-    assert context.index("First timeline comment") < context.index("Second timeline comment")
-    assert context.index("111111111111 First commit") < context.index("222222222222 Second commit")
-    assert context.index("Review 10 by carol") < context.index("Review 11 by dave")
-    assert "`src/old.py:21`: Old-side position" in context
-    assert "`src/new.py:34`: New-side position" in context

tests/test_development.py

@@ -9,7 +9,7 @@ from agentci.integrations.development import (
 )
 
 
-def environment(
+def development_environment(
     tmp_path: Path, scripts: list[str], *, timeout_seconds: int = 5
 ) -> DevelopmentEnvironment:
     scripts_dir = tmp_path / "scripts"
@@ -24,22 +24,29 @@ def environment(
     )
 
 
-def script(path: Path, body: str) -> None:
+def shell_script(path: Path, body: str, *, executable: bool = True) -> None:
     path.write_text(f"#!/bin/sh\nset -eu\n{body}\n")
-    path.chmod(0o755)
+    path.chmod(0o755 if executable else 0o644)
+
+
+@pytest.fixture
+def workspace(tmp_path: Path) -> Path:
+    path = tmp_path / "workspace"
+    path.mkdir()
+    return path
 
 
 async def test_runs_custom_scripts_in_order_with_sanitized_environment(
-    tmp_path, monkeypatch
+    tmp_path: Path,
+    workspace: Path,
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["first", "second"])
-    script(
+    development = development_environment(tmp_path, ["first", "second"])
+    shell_script(
         development.scripts_dir / "first",
         'printf \'first:%s:%s\\n\' "$DEV_TOOLS_DIR" "${AGENTCI_SECRET-unset}" >> order',
     )
-    script(development.scripts_dir / "second", "printf 'second\\n' >> order")
+    shell_script(development.scripts_dir / "second", "printf 'second\\n' >> order")
     monkeypatch.setenv("AGENTCI_SECRET", "must-not-leak")
 
     await development.prepare(workspace)
@@ -51,24 +58,16 @@ async def test_runs_custom_scripts_in_order_with_sanitized_environment(
     assert (tmp_path / "tools" / "bin").is_dir()
 
 
-async def test_supplied_script_names_use_the_same_directory(tmp_path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["python"])
-    script(development.scripts_dir / "python", "printf 'python\\n' > selected")
-
-    await development.prepare(workspace)
-
-    assert (workspace / "selected").read_text() == "python\n"
-
-
-async def test_runs_non_executable_shell_script_from_bind_mount(tmp_path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["mounted"])
-    mounted = development.scripts_dir / "mounted"
-    mounted.write_text("#!/bin/sh\nprintf 'mounted\\n' > selected\n")
-    mounted.chmod(0o644)
+async def test_runs_non_executable_shell_script_from_bind_mount(
+    tmp_path: Path,
+    workspace: Path,
+) -> None:
+    development = development_environment(tmp_path, ["mounted"])
+    shell_script(
+        development.scripts_dir / "mounted",
+        "printf 'mounted\\n' > selected",
+        executable=False,
+    )
 
     await development.prepare(workspace)
 
@@ -76,8 +75,8 @@ async def test_runs_non_executable_shell_script_from_bind_mount(tmp_path) -> Non
 
 
 async def test_serializes_concurrent_preparation(tmp_path, monkeypatch) -> None:
-    development = environment(tmp_path, ["shared"])
-    script(development.scripts_dir / "shared", "true")
+    development = development_environment(tmp_path, ["shared"])
+    shell_script(development.scripts_dir / "shared", "true")
     started = asyncio.Event()
     release = asyncio.Event()
     active = 0
@@ -102,20 +101,8 @@ async def test_serializes_concurrent_preparation(tmp_path, monkeypatch) -> None:
     assert maximum_active == 1
 
 
-async def test_reports_script_failure_output(tmp_path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["broken"])
-    script(development.scripts_dir / "broken", "printf 'failed detail' >&2; exit 7")
-
-    with pytest.raises(DevelopmentEnvironmentError, match="exited with 7: failed detail"):
-        await development.prepare(workspace)
-
-
-async def test_reports_missing_install_script(tmp_path: Path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["missing"])
+async def test_reports_missing_install_script(tmp_path: Path, workspace: Path) -> None:
+    development = development_environment(tmp_path, ["missing"])
 
     with pytest.raises(
         DevelopmentEnvironmentError,
@@ -124,25 +111,30 @@ async def test_reports_missing_install_script(tmp_path: Path) -> None:
         await development.prepare(workspace)
 
 
-async def test_stops_before_second_script_after_failure(tmp_path: Path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["first", "second"])
-    script(development.scripts_dir / "first", "printf 'first\n' > first-ran; exit 3")
-    script(development.scripts_dir / "second", "printf 'second\n' > second-ran")
+async def test_stops_after_failure_and_reports_output(
+    tmp_path: Path,
+    workspace: Path,
+) -> None:
+    development = development_environment(tmp_path, ["first", "second"])
+    shell_script(
+        development.scripts_dir / "first",
+        "printf 'first\\n' > first-ran; printf 'failed detail' >&2; exit 7",
+    )
+    shell_script(development.scripts_dir / "second", "printf 'second\\n' > second-ran")
 
-    with pytest.raises(DevelopmentEnvironmentError, match="exited with 3"):
+    with pytest.raises(DevelopmentEnvironmentError, match="exited with 7: failed detail"):
         await development.prepare(workspace)
 
     assert (workspace / "first-ran").read_text() == "first\n"
     assert not (workspace / "second-ran").exists()
 
 
-async def test_failure_output_keeps_only_bounded_tail(tmp_path: Path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["verbose"])
-    script(
+async def test_failure_output_keeps_only_bounded_tail(
+    tmp_path: Path,
+    workspace: Path,
+) -> None:
+    development = development_environment(tmp_path, ["verbose"])
+    shell_script(
         development.scripts_dir / "verbose",
         "printf 'discarded-prefix' >&2; "
         'i=0; while [ "$i" -lt 2100 ]; do printf x >&2; i=$((i + 1)); done; '
@@ -158,12 +150,12 @@ async def test_failure_output_keeps_only_bounded_tail(tmp_path: Path) -> None:
 
 
 async def test_wraps_subprocess_start_error(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    tmp_path: Path,
+    workspace: Path,
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["broken"])
-    script(development.scripts_dir / "broken", "true")
+    development = development_environment(tmp_path, ["broken"])
+    shell_script(development.scripts_dir / "broken", "true")
 
     async def create_subprocess_exec(*_args, **_kwargs):
         raise OSError("exec unavailable")
@@ -182,11 +174,9 @@ async def test_wraps_subprocess_start_error(
     assert isinstance(raised.value.__cause__, OSError)
 
 
-async def test_times_out_install_script(tmp_path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    development = environment(tmp_path, ["slow"], timeout_seconds=1)
-    script(development.scripts_dir / "slow", "exec sleep 10")
+async def test_times_out_install_script(tmp_path: Path, workspace: Path) -> None:
+    development = development_environment(tmp_path, ["slow"], timeout_seconds=1)
+    shell_script(development.scripts_dir / "slow", "exec sleep 10")
 
     with pytest.raises(DevelopmentEnvironmentError, match="exceeded 1 seconds"):
         await development.prepare(workspace)

tests/test_git.py

@@ -137,36 +137,36 @@ async def test_sync_branch_resets_and_cleans_before_returning_sha(
     assert recorder.calls[0][1]["env"]["AGENTCI_GIT_PASSWORD"] == "secret-token"
 
 
-async def test_branch_status_and_diff_commands(
+async def test_local_worktree_commands_and_status_mapping(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     recorder = install_recorder(
         monkeypatch,
-        [FakeProcess(), FakeProcess(stdout=b" M src/app.py\n"), FakeProcess()],
+        [
+            FakeProcess(),
+            FakeProcess(stdout=b" M src/app.py\n"),
+            FakeProcess(stdout=b" \n"),
+            FakeProcess(),
+        ],
     )
     workspace = tmp_path / "repo"
     git = git_client(tmp_path)
 
     await git.create_branch(workspace, "agent/issue-1")
-    changed = await git.has_changes(workspace)
+    dirty = await git.has_changes(workspace)
+    clean = await git.has_changes(workspace)
     await git.diff_check(workspace)
 
-    assert changed is True
+    assert dirty is True
+    assert clean is False
     assert recorder.commands == [
         ("git", "switch", "-c", "agent/issue-1"),
         ("git", "status", "--porcelain"),
+        ("git", "status", "--porcelain"),
         ("git", "diff", "--check"),
     ]
 
 
-async def test_has_changes_is_false_for_clean_status(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    install_recorder(monkeypatch, [FakeProcess(stdout=b"\n")])
-
-    assert await git_client(tmp_path).has_changes(tmp_path / "repo") is False
-
-
 async def test_commit_stages_all_changes_sets_identity_and_returns_sha(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/test_gitea.py

@@ -25,6 +25,17 @@ async def gitea_client(handler: Handler, *, retries: int = 3) -> AsyncGenerator[
         await client.close()
 
 
+@pytest.fixture
+def recorded_backoffs(monkeypatch: pytest.MonkeyPatch) -> list[int]:
+    backoffs: list[int] = []
+
+    async def record_backoff(delay: int) -> None:
+        backoffs.append(delay)
+
+    monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", record_backoff)
+    return backoffs
+
+
 async def test_sends_authenticated_json_request_contract() -> None:
     requests: list[httpx.Request] = []
 
@@ -221,19 +232,17 @@ async def test_pagination_stops_only_after_a_short_page(
 
 
 @pytest.mark.parametrize("status", [400, 401, 403, 404, 422])
-async def test_nonretryable_status_fails_once(status: int, monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_nonretryable_status_fails_once(
+    status: int,
+    recorded_backoffs: list[int],
+) -> None:
     attempts = 0
-    sleeps: list[int] = []
 
     async def handler(_request: httpx.Request) -> httpx.Response:
         nonlocal attempts
         attempts += 1
         return httpx.Response(status)
 
-    async def sleep(delay: int) -> None:
-        sleeps.append(delay)
-
-    monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep)
     async with gitea_client(handler) as client:
         with pytest.raises(
             GiteaError,
@@ -242,104 +251,89 @@ async def test_nonretryable_status_fails_once(status: int, monkeypatch: pytest.M
             await client.default_branch("org", "repo")
 
     assert attempts == 1
-    assert sleeps == []
+    assert recorded_backoffs == []
 
 
-@pytest.mark.parametrize("status", [429, 500, 502, 503, 504])
-async def test_retryable_status_recovers_after_backoff(
-    status: int, monkeypatch: pytest.MonkeyPatch
+@pytest.mark.parametrize(
+    ("status", "failures", "expected_backoffs"),
+    [
+        (429, 1, [1]),
+        (500, 1, [1]),
+        (502, 1, [1]),
+        (503, 1, [1]),
+        (504, 1, [1]),
+        pytest.param(None, 2, [1, 2], id="transport"),
+    ],
+)
+async def test_retryable_failure_recovers_after_exponential_backoff(
+    status: int | None,
+    failures: int,
+    expected_backoffs: list[int],
+    recorded_backoffs: list[int],
 ) -> None:
     attempts = 0
-    sleeps: list[int] = []
 
-    async def handler(_request: httpx.Request) -> httpx.Response:
+    async def handler(request: httpx.Request) -> httpx.Response:
         nonlocal attempts
         attempts += 1
-        if attempts == 1:
+        if attempts <= failures:
+            if status is None:
+                raise httpx.ConnectError("connection refused", request=request)
             return httpx.Response(status)
         return httpx.Response(200, json={"default_branch": "main"})
 
-    async def sleep(delay: int) -> None:
-        sleeps.append(delay)
-
-    monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep)
     async with gitea_client(handler) as client:
         assert await client.default_branch("org", "repo") == "main"
 
-    assert attempts == 2
-    assert sleeps == [1]
+    assert attempts == failures + 1
+    assert recorded_backoffs == expected_backoffs
 
 
-async def test_retryable_status_exhaustion_uses_exponential_backoff(
-    monkeypatch: pytest.MonkeyPatch,
+@pytest.mark.parametrize(
+    ("status", "retries", "expected_message", "expected_cause", "expected_backoffs"),
+    [
+        pytest.param(
+            503,
+            3,
+            "Gitea remained unavailable for GET /repos/org/repo",
+            None,
+            [1, 2],
+            id="status",
+        ),
+        pytest.param(
+            None,
+            2,
+            "Gitea request failed: GET /repos/org/repo",
+            httpx.ConnectError,
+            [1],
+            id="transport",
+        ),
+    ],
+)
+async def test_retry_exhaustion_reports_failure(
+    status: int | None,
+    retries: int,
+    expected_message: str,
+    expected_cause: type[BaseException] | None,
+    expected_backoffs: list[int],
+    recorded_backoffs: list[int],
 ) -> None:
     attempts = 0
-    sleeps: list[int] = []
-
-    async def handler(_request: httpx.Request) -> httpx.Response:
-        nonlocal attempts
-        attempts += 1
-        return httpx.Response(503)
-
-    async def sleep(delay: int) -> None:
-        sleeps.append(delay)
-
-    monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep)
-    async with gitea_client(handler) as client:
-        with pytest.raises(
-            GiteaError,
-            match="Gitea remained unavailable for GET /repos/org/repo",
-        ):
-            await client.default_branch("org", "repo")
-
-    assert attempts == 3
-    assert sleeps == [1, 2]
-
-
-async def test_transport_failure_retries_and_recovers(monkeypatch: pytest.MonkeyPatch) -> None:
-    attempts = 0
-    sleeps: list[int] = []
 
     async def handler(request: httpx.Request) -> httpx.Response:
         nonlocal attempts
         attempts += 1
-        if attempts < 3:
+        if status is None:
             raise httpx.ConnectError("connection refused", request=request)
-        return httpx.Response(200, json={"default_branch": "main"})
+        return httpx.Response(status)
 
-    async def sleep(delay: int) -> None:
-        sleeps.append(delay)
-
-    monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep)
-    async with gitea_client(handler) as client:
-        assert await client.default_branch("org", "repo") == "main"
-
-    assert attempts == 3
-    assert sleeps == [1, 2]
-
-
-async def test_transport_failure_exhaustion_preserves_cause(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    attempts = 0
-    sleeps: list[int] = []
-
-    async def handler(request: httpx.Request) -> httpx.Response:
-        nonlocal attempts
-        attempts += 1
-        raise httpx.ConnectError("connection refused", request=request)
-
-    async def sleep(delay: int) -> None:
-        sleeps.append(delay)
-
-    monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep)
-    async with gitea_client(handler, retries=2) as client:
-        with pytest.raises(
-            GiteaError,
-            match="Gitea request failed: GET /repos/org/repo",
-        ) as raised:
+    async with gitea_client(handler, retries=retries) as client:
+        with pytest.raises(GiteaError, match=expected_message) as raised:
             await client.default_branch("org", "repo")
 
-    assert isinstance(raised.value.__cause__, httpx.ConnectError)
-    assert attempts == 2
-    assert sleeps == [1]
+    if expected_cause is None:
+        assert raised.value.__cause__ is None
+    else:
+        assert isinstance(raised.value.__cause__, expected_cause)
+    assert attempts == retries
+    assert recorded_backoffs == expected_backoffs

tests/test_health.py

@@ -43,7 +43,7 @@ async def test_liveness_does_not_depend_on_runtime_providers() -> None:
 
 
 @pytest.mark.parametrize(
-    ("provider_ready", "status_code", "payload"),
+    ("provider_result", "status_code", "payload"),
     [
         (True, 200, {"status": "ready"}),
         (
@@ -54,24 +54,24 @@ async def test_liveness_does_not_depend_on_runtime_providers() -> None:
                 "reason": "opencode provider is not connected",
             },
         ),
+        pytest.param(
+            RuntimeError("provider check failed"),
+            500,
+            None,
+            id="provider-error",
+        ),
     ],
 )
 async def test_readiness_reflects_provider_state(
-    provider_ready: bool, status_code: int, payload: dict[str, str]
+    provider_result: bool | Exception,
+    status_code: int,
+    payload: dict[str, str] | None,
 ) -> None:
-    provider = Provider(provider_ready)
+    provider = Provider(provider_result)
 
     response = await get(application(provider), "/health/ready")
 
     assert response.status_code == status_code
+    if payload is not None:
         assert response.json() == payload
     assert provider.calls == 1
-
-
-async def test_readiness_provider_error_is_server_failure() -> None:
-    provider = Provider(RuntimeError("provider check failed"))
-
-    response = await get(application(provider), "/health/ready")
-
-    assert response.status_code == 500
-    assert provider.calls == 1

tests/test_opencode.py

@@ -121,12 +121,6 @@ async def test_concurrent_readiness_checks_share_one_probe(tmp_path: Path) -> No
             PROVIDERS,
             ["/global/health", "/doc"],
         ),
-        (
-            {"healthy": True, "version": "1.18.4"},
-            API_DOCUMENT,
-            {**PROVIDERS, "connected": ["anthropic"]},
-            ["/global/health", "/doc", "/provider"],
-        ),
         (
             {"healthy": True, "version": "1.18.4"},
             API_DOCUMENT,
@@ -349,169 +343,3 @@ async def test_structured_validation_retry_exhaustion_reports_final_error(
     assert "without repeating repository work" in prompts[1]
     assert paths[-1] == "/session/ses_existing/abort"
     await value.close()
-
-
-async def test_cleanup_failure_does_not_mask_turn_error(tmp_path: Path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    paths: list[str] = []
-
-    async def handler(request: httpx.Request) -> httpx.Response:
-        paths.append(request.url.path)
-        if request.url.path.endswith("/abort"):
-            return httpx.Response(500, text="abort failed")
-        return httpx.Response(200, json={"unexpected": True})
-
-    value = client(tmp_path, handler)
-    with pytest.raises(
-        OpenCodeError,
-        match="OpenCode response did not include assistant metadata",
-    ):
-        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 paths == [
-        "/session/ses_existing/message",
-        "/session/ses_existing/abort",
-    ]
-    await value.close()
-
-
-async def test_close_aborts_an_active_session(tmp_path: Path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    message_started = asyncio.Event()
-    release_message = asyncio.Event()
-    paths: list[str] = []
-
-    async def handler(request: httpx.Request) -> httpx.Response:
-        paths.append(request.url.path)
-        if request.url.path.endswith("/abort"):
-            assert request.headers["x-opencode-directory"] == str(workspace.resolve())
-            return httpx.Response(200, json=True)
-        message_started.set()
-        await release_message.wait()
-        return httpx.Response(
-            200,
-            json={
-                "info": {"structured": {"summary_markdown": "finished", "tests": []}},
-                "parts": [],
-            },
-        )
-
-    value = client(tmp_path, handler)
-    turn = asyncio.create_task(
-        value.resume(
-            session_id="ses_active",
-            workspace=workspace,
-            prompt="continue",
-            model="openai/model",
-            variant=None,
-            schema_name="agent_result.json",
-            result_type=AgentResult,
-        )
-    )
-    await message_started.wait()
-
-    await value.close()
-    release_message.set()
-    result = await turn
-
-    assert result.summary_markdown == "finished"
-    assert paths == ["/session/ses_active/message", "/session/ses_active/abort"]
-    assert value.client.is_closed
-
-
-async def test_close_continues_after_active_session_abort_failure(tmp_path: Path) -> None:
-    paths: list[str] = []
-
-    async def handler(request: httpx.Request) -> httpx.Response:
-        paths.append(request.url.path)
-        if "/failed/" in request.url.path:
-            return httpx.Response(503)
-        return httpx.Response(200, json=True)
-
-    value = client(tmp_path, handler)
-    value._active_sessions.update(
-        {
-            "failed": tmp_path / "first",
-            "succeeds": tmp_path / "second",
-        }
-    )
-
-    await value.close()
-
-    assert paths == ["/session/failed/abort", "/session/succeeds/abort"]
-    assert value.client.is_closed
-
-
-async def test_aborts_timed_out_session(tmp_path: Path) -> None:
-    workspace = tmp_path / "workspace"
-    workspace.mkdir()
-    abort_count = 0
-
-    async def handler(request: httpx.Request) -> httpx.Response:
-        nonlocal abort_count
-        if request.url.path.endswith("/abort"):
-            abort_count += 1
-            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 abort_count == 1
-    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_abort.py

@@ -1,34 +1,61 @@
+import asyncio
 from pathlib import Path
+from typing import Any
 
 import httpx
 import pytest
 
 from agentci.integrations.opencode.client import OpenCode, OpenCodeError
+from agentci.workflows.model import AgentResult
 
 
-def client(tmp_path: Path, status: int) -> OpenCode:
+class FakeCodeGraph:
+    async def prepare(self, _workspace: Path) -> None:
+        pass
+
+
+def client(
+    tmp_path: Path,
+    handler: Any,
+    *,
+    codegraph: FakeCodeGraph | None = None,
+    timeout_seconds: int = 60,
+) -> OpenCode:
+    selected_codegraph = FakeCodeGraph() if codegraph is None else codegraph
     return OpenCode(
         base_url="http://opencode:4096",
         username="opencode",
         password="secret",
-        schemas_dir=tmp_path,
+        schemas_dir=Path(__file__).parents[1] / "src" / "agentci" / "prompts" / "schemas",
         health_directory=tmp_path,
         required_models=(),
-        timeout_seconds=60,
-        transport=httpx.MockTransport(lambda _request: httpx.Response(status)),
+        timeout_seconds=timeout_seconds,
+        codegraph=selected_codegraph,  # type: ignore[arg-type]
+        transport=httpx.MockTransport(handler),
     )
 
 
-@pytest.mark.parametrize("status", [200, 204, 404, 409])
-async def test_absent_or_inactive_session_is_success(tmp_path: Path, status: int) -> None:
-    value = client(tmp_path, status)
+@pytest.mark.parametrize(
+    ("status", "succeeds"),
+    [
+        (200, True),
+        (204, True),
+        (404, True),
+        (409, True),
+        (400, False),
+        (429, False),
+        (500, False),
+    ],
+)
+async def test_abort_status_contract(
+    tmp_path: Path,
+    status: int,
+    succeeds: bool,
+) -> None:
+    value = client(tmp_path, lambda _request: httpx.Response(status))
+    if succeeds:
         await value.abort("session", tmp_path)
-    await value.close()
-
-
-@pytest.mark.parametrize("status", [400, 429, 500])
-async def test_abort_failure_is_visible_for_retry(tmp_path: Path, status: int) -> None:
-    value = client(tmp_path, status)
+    else:
         with pytest.raises(OpenCodeError):
             await value.abort("session", tmp_path)
     await value.close()
@@ -41,16 +68,7 @@ async def test_abort_sends_authenticated_workspace_request(tmp_path: Path) -> No
         requests.append(request)
         return httpx.Response(204)
 
-    value = OpenCode(
-        base_url="http://opencode:4096/",
-        username="opencode",
-        password="secret",
-        schemas_dir=tmp_path,
-        health_directory=tmp_path,
-        required_models=(),
-        timeout_seconds=60,
-        transport=httpx.MockTransport(handler),
-    )
+    value = client(tmp_path, handler)
     await value.abort("ses_123", tmp_path)
     await value.close()
 
@@ -66,16 +84,182 @@ async def test_best_effort_abort_suppresses_transport_failure(tmp_path: Path) ->
     async def handler(request: httpx.Request) -> httpx.Response:
         raise httpx.ConnectError("connection refused", request=request)
 
-    value = OpenCode(
-        base_url="http://opencode:4096",
-        username="opencode",
-        password="secret",
-        schemas_dir=tmp_path,
-        health_directory=tmp_path,
-        required_models=(),
-        timeout_seconds=60,
-        transport=httpx.MockTransport(handler),
-    )
+    value = client(tmp_path, handler)
 
     await value.abort("session", tmp_path, best_effort=True)
     await value.close()
+
+
+async def test_cleanup_failure_does_not_mask_turn_error(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    paths: list[str] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        paths.append(request.url.path)
+        if request.url.path.endswith("/abort"):
+            return httpx.Response(500, text="abort failed")
+        return httpx.Response(200, json={"unexpected": True})
+
+    value = client(tmp_path, handler)
+    with pytest.raises(
+        OpenCodeError,
+        match="OpenCode response did not include assistant metadata",
+    ):
+        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 paths == [
+        "/session/ses_existing/message",
+        "/session/ses_existing/abort",
+    ]
+    await value.close()
+
+
+async def test_aborts_timed_out_session(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    abort_count = 0
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal abort_count
+        if request.url.path.endswith("/abort"):
+            abort_count += 1
+            return httpx.Response(200, json=True)
+        raise httpx.ReadTimeout("slow", request=request)
+
+    value = client(tmp_path, handler, timeout_seconds=60)
+    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 abort_count == 1
+    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()
+
+
+async def test_close_aborts_all_active_sessions_and_continues_after_failure(
+    tmp_path: Path,
+) -> None:
+    failed_workspace = tmp_path / "failed-workspace"
+    succeeds_workspace = tmp_path / "succeeds-workspace"
+    failed_workspace.mkdir()
+    succeeds_workspace.mkdir()
+    failed_message_started = asyncio.Event()
+    succeeds_message_started = asyncio.Event()
+    release_messages = asyncio.Event()
+    abort_requests: list[httpx.Request] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        path = request.url.path
+        if path.endswith("/abort"):
+            abort_requests.append(request)
+            return httpx.Response(503 if "/failed/" in path else 204)
+        if "/failed/" in path:
+            failed_message_started.set()
+            summary = "failed-session-finished"
+        else:
+            succeeds_message_started.set()
+            summary = "succeeds-session-finished"
+        await release_messages.wait()
+        return httpx.Response(
+            200,
+            json={
+                "info": {"structured": {"summary_markdown": summary, "tests": []}},
+                "parts": [],
+            },
+        )
+
+    value = client(tmp_path, handler)
+    failed_turn = asyncio.create_task(
+        value.resume(
+            session_id="failed",
+            workspace=failed_workspace,
+            prompt="continue",
+            model="openai/model",
+            variant=None,
+            schema_name="agent_result.json",
+            result_type=AgentResult,
+        )
+    )
+    await failed_message_started.wait()
+    succeeds_turn = asyncio.create_task(
+        value.resume(
+            session_id="succeeds",
+            workspace=succeeds_workspace,
+            prompt="continue",
+            model="openai/model",
+            variant=None,
+            schema_name="agent_result.json",
+            result_type=AgentResult,
+        )
+    )
+    await succeeds_message_started.wait()
+
+    await value.close()
+    release_messages.set()
+    completed = await asyncio.gather(failed_turn, succeeds_turn)
+
+    assert [request.url for request in abort_requests] == [
+        httpx.URL("http://opencode:4096/session/failed/abort"),
+        httpx.URL("http://opencode:4096/session/succeeds/abort"),
+    ]
+    assert [request.headers["x-opencode-directory"] for request in abort_requests] == [
+        str(failed_workspace.resolve()),
+        str(succeeds_workspace.resolve()),
+    ]
+    assert [result.summary_markdown for result in completed] == [
+        "failed-session-finished",
+        "succeeds-session-finished",
+    ]
+    assert value.client.is_closed

tests/test_opencode_deployment.py

@@ -1,79 +1,9 @@
 import json
-import shutil
-import subprocess
 from pathlib import Path
 
-import pytest
-
 ROOT = Path(__file__).parents[1]
 
 
-def _indent(line: str) -> int:
-    return len(line) - len(line.lstrip())
-
-
-def _service(name: str) -> list[str]:
-    lines = (ROOT / "compose.yaml").read_text().splitlines()
-    marker = f"  {name}:"
-    start = lines.index(marker) + 1
-    end = next(
-        (
-            index
-            for index in range(start, len(lines))
-            if lines[index].strip() and _indent(lines[index]) <= 2
-        ),
-        len(lines),
-    )
-    return lines[start:end]
-
-
-def _section(lines: list[str], name: str, *, indent: int = 4) -> list[str]:
-    marker = f"{' ' * indent}{name}:"
-    start = lines.index(marker) + 1
-    end = next(
-        (
-            index
-            for index in range(start, len(lines))
-            if lines[index].strip() and _indent(lines[index]) <= indent
-        ),
-        len(lines),
-    )
-    return lines[start:end]
-
-
-def _value(value: str) -> str:
-    value = value.strip()
-    if len(value) >= 2 and value[0] == value[-1] == '"':
-        return str(json.loads(value))
-    if len(value) >= 2 and value[0] == value[-1] == "'":
-        return value[1:-1]
-    return value
-
-
-def _mapping(lines: list[str], name: str, *, indent: int = 4) -> dict[str, str]:
-    entries = _section(lines, name, indent=indent)
-    result: dict[str, str] = {}
-    for line in entries:
-        if _indent(line) != indent + 2 or line.lstrip().startswith("-"):
-            continue
-        key, separator, value = line.strip().partition(":")
-        if separator:
-            result[key] = _value(value)
-    return result
-
-
-def _sequence(lines: list[str], name: str, *, indent: int = 4) -> list[str]:
-    entries = _section(lines, name, indent=indent)
-    prefix = f"{' ' * (indent + 2)}- "
-    return [_value(line.removeprefix(prefix)) for line in entries if line.startswith(prefix)]
-
-
-def _scalar(lines: list[str], name: str, *, indent: int = 4) -> str:
-    prefix = f"{' ' * indent}{name}:"
-    line = next(line for line in lines if line.startswith(prefix))
-    return _value(line.removeprefix(prefix))
-
-
 def test_config_preserves_builtin_permissions_and_restricts_research() -> None:
     config = json.loads((ROOT / "opencode" / "opencode.json").read_text())
 
@@ -86,60 +16,6 @@ def test_config_preserves_builtin_permissions_and_restricts_research() -> None:
         "context7_*": "allow",
         "gh_grep_*": "allow",
     }
-    assert config["mcp"]["codegraph"]["command"] == ["codegraph", "serve", "--mcp"]
-    assert config["mcp"]["context7"]["url"] == "https://mcp.context7.com/mcp"
-    assert config["agent"]["explore"]["model"] == "{env:AGENTCI_EXPLORE_MODEL}"
-    assert config["agent"]["explore"]["variant"] == "{env:AGENTCI_EXPLORE_VARIANT}"
-    assert config["agent"]["research"]["variant"] == "{env:AGENTCI_RESEARCH_VARIANT}"
-
-
-def test_compose_services_have_expected_runtime_contract() -> None:
-    agentci = _service("agentci")
-    opencode = _service("opencode")
-    agentci_environment = _mapping(agentci, "environment")
-    opencode_environment = _mapping(opencode, "environment")
-
-    assert agentci_environment["AGENTCI_OPENCODE_URL"] == "http://opencode:4096"
-    assert agentci_environment["AGENTCI_EXPLORE_VARIANT"] == ("${AGENTCI_EXPLORE_VARIANT:-low}")
-    assert agentci_environment["AGENTCI_RESEARCH_VARIANT"] == ("${AGENTCI_RESEARCH_VARIANT:-high}")
-    assert opencode_environment["HOME"] == "/etc/opencode/home"
-    assert opencode_environment["OPENCODE_DISABLE_EXTERNAL_SKILLS"] == "1"
-    assert opencode_environment["OPENCODE_ENABLE_EXA"] == "1"
-    assert "OPENCODE_DISABLE_DEFAULT_PLUGINS" not in opencode_environment
-
-    assert _sequence(agentci, "volumes") == [
-        "agentci_data:/var/lib/agentci",
-        "./install-scripts:/etc/agentci/install-scripts:ro",
-    ]
-    assert _sequence(opencode, "volumes") == [
-        "agentci_data:/var/lib/agentci",
-        "opencode_home:/var/lib/opencode",
-    ]
-    assert _sequence(agentci, "tmpfs") == ["/run/agentci:mode=1777"]
-    assert _sequence(opencode, "tmpfs") == ["/run/agentci:mode=1777"]
-    assert json.loads(_scalar(opencode, "command")) == [
-        "opencode",
-        "serve",
-        "--hostname",
-        "0.0.0.0",
-        "--port",
-        "4096",
-    ]
-
-    image = "${AGENTCI_IMAGE:-git.krtss.de/stanponomarev/agentci:latest}"
-    assert _scalar(agentci, "image") == image
-    assert _scalar(opencode, "image") == image
-    assert all(line.strip() != "build:" for line in [*agentci, *opencode])
-
-    dependency = _section(_section(agentci, "depends_on"), "opencode", indent=6)
-    assert _mapping(["      dependency:", *dependency], "dependency", indent=6) == {
-        "condition": "service_healthy"
-    }
-    healthcheck = _mapping(opencode, "healthcheck")
-    health_command = json.loads(healthcheck["test"])
-    assert health_command[0] == "CMD-SHELL"
-    assert "/global/health" in health_command[1]
-    assert "OPENCODE_SERVER_PASSWORD_FILE" in health_command[1]
 
 
 def test_compose_has_no_sandbox_security_exceptions() -> None:
@@ -156,20 +32,6 @@ def test_compose_has_no_sandbox_security_exceptions() -> None:
         assert forbidden not in compose
 
 
-def test_gitea_workflow_publishes_master_images() -> None:
-    lines = (ROOT / ".gitea" / "workflows" / "publish-image.yaml").read_text().splitlines()
-    push = _section(_section(lines, "on", indent=0), "push", indent=2)
-
-    assert _sequence(push, "branches", indent=4) == ["master"]
-    assert _mapping(lines, "env", indent=0)["IMAGE_NAME"] == (
-        "git.krtss.de/stanponomarev/agentci"
-    )
-    assert "            ${{ env.IMAGE_NAME }}:latest" in lines
-    assert "            ${{ env.IMAGE_NAME }}:${{ gitea.sha }}" in lines
-    assert "          username: ${{ secrets.REGISTRY_USERNAME }}" in lines
-    assert "          password: ${{ secrets.REGISTRY_TOKEN }}" in lines
-
-
 def test_container_pins_opencode_major_version_contract() -> None:
     lines = [line.strip() for line in (ROOT / "Dockerfile").read_text().splitlines()]
     build_arguments = {line.removeprefix("ARG ") for line in lines if line.startswith("ARG ")}
@@ -178,28 +40,3 @@ def test_container_pins_opencode_major_version_contract() -> None:
     assert any(
         line.rstrip("\\").strip() == '"opencode-ai@${AGENTCI_OPENCODE_VERSION}"' for line in lines
     )
-
-
-def test_compose_document_validates_when_compose_cli_is_available() -> None:
-    docker = shutil.which("docker")
-    if docker is None:
-        pytest.skip("Docker Compose CLI is not installed")
-    probe = subprocess.run(
-        [docker, "compose", "version"],
-        cwd=ROOT,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    if probe.returncode:
-        pytest.skip("Docker Compose provider is not available")
-
-    result = subprocess.run(
-        [docker, "compose", "config", "--quiet"],
-        cwd=ROOT,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-
-    assert result.returncode == 0, result.stderr

tests/test_opencode_support.py

@@ -23,7 +23,6 @@ def test_api_contract_requires_session_message_and_abort_routes() -> None:
     "document",
     [
         None,
-        [],
         {},
         {"paths": []},
         {
@@ -33,14 +32,6 @@ def test_api_contract_requires_session_message_and_abort_routes() -> None:
                 "/session": {"post": {}},
             }
         },
-        {
-            "paths": {
-                "/global/health": {"get": {}},
-                "/provider": {"get": {}},
-                "/session": {"post": {}},
-                7: {"post": {}},
-            }
-        },
     ],
 )
 def test_api_contract_rejects_malformed_documents(document: object) -> None:
@@ -78,7 +69,7 @@ def provider_payload(
     [
         (provider_payload(), ("openai", "model", "high"), True),
         (provider_payload(), ("openai", "model", None), True),
-        (provider_payload(connected=("anthropic",)), ("openai", "model", "high"), False),
+        (provider_payload(connected=["anthropic"]), ("openai", "model", "high"), False),
         (provider_payload(status="deprecated"), ("openai", "model", "high"), False),
         (provider_payload(toolcall=False), ("openai", "model", "high"), False),
         (provider_payload(variants={}), ("openai", "model", "high"), False),
@@ -95,9 +86,7 @@ def test_model_readiness_matrix(
     "payload",
     [
         None,
-        [],
         {},
-        {"connected": None, "all": []},
         {"connected": ["openai"], "all": None},
         {"connected": ["openai"], "all": [{"id": [], "models": {}}]},
         {

tests/test_reducer.py

@@ -105,7 +105,6 @@ def test_command_received_preserves_input_and_requests_authorization() -> None:
     ("body", "pr_number", "expected_kind", "expected_message"),
     [
         ("/agent plan write tests", None, JobKind.PLAN, "write tests"),
-        ("/agent iterate refine", None, JobKind.ITERATE_PLAN, "refine"),
         ("/agent iterate address review", 8, JobKind.ITERATE_IMPLEMENT, "address review"),
     ],
 )
@@ -342,36 +341,34 @@ def test_comment_link_is_allowed_after_a_job_is_terminal() -> None:
 
 
 @pytest.mark.parametrize(
-    ("state", "event"),
+    ("state", "event", "error_match"),
     [
-        (received(), JobStarted(job_id="job")),
-        (queued(), PermissionGranted(job_id="job")),
-        (running(), PermissionGranted(job_id="job")),
+        (None, PermissionGranted(job_id="job"), "Only CommandReceived"),
+        (
+            received(),
+            PermissionDenied(job_id="another-job"),
+            "job ID does not match",
+        ),
+        (received(), JobStarted(job_id="job"), "invalid while job"),
+        (queued(), PermissionGranted(job_id="job"), "invalid while job"),
+        (running(), PermissionGranted(job_id="job"), "invalid while job"),
         (
             reduce_job(running(), JobCompleted(job_id="job", comment_body="ok")).job,
             JobStarted(job_id="job"),
+            "invalid while job",
         ),
     ],
-    ids=["received", "queued", "running", "terminal"],
+    ids=["missing-state", "mismatched-id", "received", "queued", "running", "terminal"],
 )
-def test_events_invalid_for_the_current_status_are_rejected(
-    state: Job,
-    event: JobStarted | PermissionGranted,
+def test_invalid_transitions_are_rejected(
+    state: Job | None,
+    event: JobStarted | PermissionDenied | PermissionGranted,
+    error_match: str,
 ) -> None:
-    with pytest.raises(InvalidTransition, match="invalid while job"):
+    with pytest.raises(InvalidTransition, match=error_match):
         reduce_job(state, event)
 
 
-def test_only_command_received_can_create_state() -> None:
-    with pytest.raises(InvalidTransition, match="Only CommandReceived"):
-        reduce_job(None, PermissionGranted(job_id="job"))
-
-
-def test_event_job_id_must_match_state() -> None:
-    with pytest.raises(InvalidTransition, match="job ID does not match"):
-        reduce_job(received(), PermissionDenied(job_id="another-job"))
-
-
 @pytest.mark.parametrize(
     ("state", "expected"),
     [

tests/test_repository.py

@@ -26,8 +26,7 @@ from agentci.engine.model import (
 )
 from agentci.engine.reducer import Transition
 from agentci.engine.repository import Repository
-
-MIGRATIONS = Path(__file__).parents[1] / "src" / "agentci" / "migrations"
+from tests.conftest import SQLiteClock
 
 
 class TrackingConnection:
@@ -56,13 +55,6 @@ class FailingSetupConnection:
         self.closed = True
 
 
-@pytest.fixture
-async def repository(tmp_path: Path) -> Repository:
-    value = Repository(tmp_path / "state.sqlite3", MIGRATIONS)
-    await value.initialize()
-    return value
-
-
 def command(
     delivery: str,
     body: str = "/agent plan",
@@ -82,10 +74,12 @@ def command(
     )
 
 
-async def create_running_job(repository: Repository, delivery: str, *, issue: int = 3) -> Job:
-    job = (await repository.accept(command(delivery, issue=issue))).job
-    job = (await repository.apply(f"{delivery}:grant", PermissionGranted(job_id=job.id))).job
-    return (await repository.apply(f"{delivery}:start", JobStarted(job_id=job.id))).job
+async def create_running_job(
+    engine_repository: Repository, delivery: str, *, issue: int = 3
+) -> Job:
+    job = (await engine_repository.accept(command(delivery, issue=issue))).job
+    job = (await engine_repository.apply(f"{delivery}:grant", PermissionGranted(job_id=job.id))).job
+    return (await engine_repository.apply(f"{delivery}:start", JobStarted(job_id=job.id))).job
 
 
 async def test_repository_closes_connections_after_success_and_failure(
@@ -98,7 +92,7 @@ async def test_repository_closes_connections_after_success_and_failure(
         "connect",
         lambda _path: cast(sqlite3.Connection, connections.pop(0)),
     )
-    repository = Repository(tmp_path / "state.sqlite3", MIGRATIONS)
+    repository = Repository(tmp_path / "state.sqlite3")
     successful = cast(TrackingConnection, await repository._run(lambda connection: connection))
 
     def fail(_connection: sqlite3.Connection) -> None:
@@ -154,86 +148,38 @@ def build_workflow(
 
 
 async def attach_workflow(
-    repository: Repository,
+    engine_repository: Repository,
     workspace: Path,
     *,
     delivery: str,
     workflow_id: str,
     status: WorkflowStatus = WorkflowStatus.ACTIVE,
 ) -> tuple[Job, Workflow]:
-    job = await create_running_job(repository, delivery)
+    job = await create_running_job(engine_repository, delivery)
     workflow = build_workflow(workflow_id, workspace, status=status)
-    result = await repository.apply(
+    result = await engine_repository.apply(
         f"{workflow_id}:created",
         WorkflowCreated(job_id=job.id, workflow=workflow, stage="planning"),
     )
     return result.job, workflow
 
 
-def insert_workflow(
-    connection: sqlite3.Connection,
-    workflow: Workflow,
-    created_at: str,
-) -> None:
-    connection.execute(
-        """INSERT INTO workflows(
-            id, kind, repo_owner, repo_name, issue_number, pr_number, base_sha, branch,
-            workspace_path, primary_session_id, reviewer_session_id, artifact, review_json,
-            status, runtime, created_at, updated_at
-        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
-        (
-            workflow.id,
-            workflow.kind.value,
-            workflow.repo_owner,
-            workflow.repo_name,
-            workflow.issue_number,
-            workflow.pr_number,
-            workflow.base_sha,
-            workflow.branch,
-            str(workflow.workspace_path),
-            workflow.primary_session_id,
-            workflow.reviewer_session_id,
-            workflow.artifact,
-            workflow.review_json,
-            workflow.status.value,
-            workflow.runtime,
-            created_at,
-            created_at,
-        ),
-    )
-
-
-async def test_receive_is_idempotent_without_consuming_sequence(
-    repository: Repository,
-) -> None:
-    first = await repository.accept(command("delivery-1"))
-    duplicate = await repository.accept(command("delivery-1"))
-    second = await repository.accept(command("delivery-2"))
-
-    assert (
-        first.duplicate,
-        duplicate.duplicate,
-        duplicate.job.id,
-        second.job.receive_sequence,
-    ) == (False, True, first.job.id, first.job.receive_sequence + 1)
-
-
 async def test_repository_stamps_start_and_completion_once(
-    repository: Repository,
-    monkeypatch: pytest.MonkeyPatch,
+    engine_repository: Repository,
+    sqlite_clock: SQLiteClock,
 ) -> None:
-    clock = {"now": "2026-02-01T00:00:00+00:00"}
-    monkeypatch.setattr(_sqlite, "now", lambda: clock["now"])
-    job = (await repository.accept(command("delivery-1"))).job
-    job = (await repository.apply("grant", PermissionGranted(job_id=job.id))).job
-    clock["now"] = "2026-02-01T00:01:00+00:00"
-    job = (await repository.apply("start", JobStarted(job_id=job.id))).job
-    clock["now"] = "2026-02-01T00:02:00+00:00"
-    job = (await repository.apply("complete", JobCompleted(job_id=job.id, comment_body="done"))).job
-    clock["now"] = "2026-02-01T00:03:00+00:00"
-    await repository.apply("comment", CommentLinked(job_id=job.id, comment_id=99))
+    job = (await engine_repository.accept(command("delivery-1"))).job
+    job = (await engine_repository.apply("grant", PermissionGranted(job_id=job.id))).job
+    sqlite_clock.now = "2026-02-01T00:01:00+00:00"
+    job = (await engine_repository.apply("start", JobStarted(job_id=job.id))).job
+    sqlite_clock.now = "2026-02-01T00:02:00+00:00"
+    job = (
+        await engine_repository.apply("complete", JobCompleted(job_id=job.id, comment_body="done"))
+    ).job
+    sqlite_clock.now = "2026-02-01T00:03:00+00:00"
+    await engine_repository.apply("comment", CommentLinked(job_id=job.id, comment_id=99))
 
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
         timestamps = connection.execute(
             "SELECT started_at, finished_at FROM jobs WHERE id=?", (job.id,)
         ).fetchone()
@@ -245,11 +191,11 @@ async def test_repository_stamps_start_and_completion_once(
 
 
 async def test_apply_rolls_back_event_job_workflow_and_task_after_mid_apply_failure(
-    repository: Repository,
+    engine_repository: Repository,
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    job = await create_running_job(repository, "delivery-1")
+    job = await create_running_job(engine_repository, "delivery-1")
     workflow = build_workflow("workflow-rollback", tmp_path)
     original_insert_tasks = _sqlite.insert_tasks
 
@@ -271,14 +217,14 @@ async def test_apply_rolls_back_event_job_workflow_and_task_after_mid_apply_fail
     monkeypatch.setattr(_sqlite, "insert_tasks", fail_after_task_write)
 
     with pytest.raises(RuntimeError, match="injected task persistence failure"):
-        await repository.apply(
+        await engine_repository.apply(
             "workflow-rollback:created",
             WorkflowCreated(job_id=job.id, workflow=workflow, stage="planning"),
         )
 
-    persisted_job = await repository.get_job(job.id)
-    persisted_workflow = await repository.get_workflow(workflow.id)
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
+    persisted_job = await engine_repository.get_job(job.id)
+    persisted_workflow = await engine_repository.get_workflow(workflow.id)
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
         event_count = connection.execute(
             "SELECT COUNT(*) FROM job_events WHERE event_id=?",
             ("workflow-rollback:created",),
@@ -297,13 +243,13 @@ async def test_apply_rolls_back_event_job_workflow_and_task_after_mid_apply_fail
 
 
 async def test_duplicate_event_application_does_not_duplicate_tasks(
-    repository: Repository,
+    engine_repository: Repository,
 ) -> None:
-    job = (await repository.accept(command("delivery-1"))).job
-    applied = await repository.apply("permission", PermissionGranted(job_id=job.id))
-    duplicate = await repository.apply("permission", PermissionGranted(job_id=job.id))
+    job = (await engine_repository.accept(command("delivery-1"))).job
+    applied = await engine_repository.apply("permission", PermissionGranted(job_id=job.id))
+    duplicate = await engine_repository.apply("permission", PermissionGranted(job_id=job.id))
 
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
         tasks = connection.execute(
             """SELECT ordinal, listener, queue FROM listener_tasks
             WHERE source_event_id=? ORDER BY ordinal""",
@@ -317,30 +263,13 @@ async def test_duplicate_event_application_does_not_duplicate_tasks(
     )
 
 
-async def test_workflow_creation_and_job_link_commit_together(
-    repository: Repository,
+async def test_save_workflow_updates_the_mutable_snapshot_and_timestamp(
+    engine_repository: Repository,
     tmp_path: Path,
+    sqlite_clock: SQLiteClock,
 ) -> None:
     job, workflow = await attach_workflow(
-        repository,
-        tmp_path,
-        delivery="delivery-1",
-        workflow_id="workflow-atomic",
-    )
-
-    assert (job.workflow_id, await repository.get_workflow(workflow.id)) == (
-        workflow.id,
-        workflow,
-    )
-
-
-async def test_save_workflow_updates_the_mutable_snapshot_and_timestamp(
-    repository: Repository,
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    _, workflow = await attach_workflow(
-        repository,
+        engine_repository,
         tmp_path,
         delivery="delivery-1",
         workflow_id="workflow-update",
@@ -355,56 +284,57 @@ async def test_save_workflow_updates_the_mutable_snapshot_and_timestamp(
         review_json='{"verdict":"approved"}',
         status=WorkflowStatus.COMPLETED,
     )
-    monkeypatch.setattr(_sqlite, "now", lambda: "2026-02-02T00:00:00+00:00")
+    sqlite_clock.now = "2026-02-02T00:00:00+00:00"
 
-    await repository.save_workflow(updated)
+    await engine_repository.save_workflow(updated)
 
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
         updated_at = connection.execute(
             "SELECT updated_at FROM workflows WHERE id=?", (workflow.id,)
         ).fetchone()
-    assert (await repository.get_workflow(workflow.id), updated_at) == (
+    assert (job.workflow_id, await engine_repository.get_workflow(workflow.id), updated_at) == (
+        workflow.id,
         updated,
         ("2026-02-02T00:00:00+00:00",),
     )
 
 
-async def test_saving_unknown_workflow_fails(repository: Repository, tmp_path: Path) -> None:
+async def test_saving_unknown_workflow_fails(engine_repository: Repository, tmp_path: Path) -> None:
     workflow = build_workflow("missing", tmp_path)
 
     with pytest.raises(KeyError, match="Unknown workflow"):
-        await repository.save_workflow(workflow)
+        await engine_repository.save_workflow(workflow)
 
 
 async def test_fail_job_workflow_fails_only_active_workflows(
-    repository: Repository,
+    engine_repository: Repository,
     tmp_path: Path,
 ) -> None:
     active_job, active = await attach_workflow(
-        repository,
+        engine_repository,
         tmp_path,
         delivery="delivery-1",
         workflow_id="active-workflow",
     )
     completed_job, completed = await attach_workflow(
-        repository,
+        engine_repository,
         tmp_path,
         delivery="delivery-2",
         workflow_id="completed-workflow",
         status=WorkflowStatus.COMPLETED,
     )
 
-    await repository.fail_job_workflow(active_job.id)
-    await repository.fail_job_workflow(completed_job.id)
+    await engine_repository.fail_job_workflow(active_job.id)
+    await engine_repository.fail_job_workflow(completed_job.id)
 
     assert (
-        (await repository.get_workflow(active.id)).status,  # type: ignore[union-attr]
-        (await repository.get_workflow(completed.id)).status,  # type: ignore[union-attr]
+        (await engine_repository.get_workflow(active.id)).status,  # type: ignore[union-attr]
+        (await engine_repository.get_workflow(completed.id)).status,  # type: ignore[union-attr]
     ) == (WorkflowStatus.FAILED, WorkflowStatus.COMPLETED)
 
 
 async def test_latest_workflow_returns_newest_completed_match(
-    repository: Repository,
+    engine_repository: Repository,
     tmp_path: Path,
 ) -> None:
     older = build_workflow("plan-older", tmp_path, status=WorkflowStatus.COMPLETED)
@@ -413,17 +343,17 @@ async def test_latest_workflow_returns_newest_completed_match(
     wrong_issue = build_workflow(
         "plan-wrong-issue", tmp_path, issue=4, status=WorkflowStatus.COMPLETED
     )
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
-        insert_workflow(connection, older, "2026-01-01T00:00:00+00:00")
-        insert_workflow(connection, newest, "2026-01-02T00:00:00+00:00")
-        insert_workflow(connection, active, "2026-01-03T00:00:00+00:00")
-        insert_workflow(connection, wrong_issue, "2026-01-04T00:00:00+00:00")
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
+        _sqlite.insert_workflow(connection, older, "2026-01-01T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, newest, "2026-01-02T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, active, "2026-01-03T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, wrong_issue, "2026-01-04T00:00:00+00:00")
 
-    assert await repository.latest_workflow("alice", "repo", 3, WorkflowKind.PLAN) == newest
+    assert await engine_repository.latest_workflow("alice", "repo", 3, WorkflowKind.PLAN) == newest
 
 
 async def test_workflow_for_pr_returns_newest_implementation_match(
-    repository: Repository,
+    engine_repository: Repository,
     tmp_path: Path,
 ) -> None:
     older = build_workflow("implementation-older", tmp_path, kind=WorkflowKind.IMPLEMENT, pr=17)
@@ -436,27 +366,27 @@ async def test_workflow_for_pr_returns_newest_implementation_match(
         repo="another",
         pr=17,
     )
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
-        insert_workflow(connection, older, "2026-01-01T00:00:00+00:00")
-        insert_workflow(connection, newest, "2026-01-02T00:00:00+00:00")
-        insert_workflow(connection, plan, "2026-01-03T00:00:00+00:00")
-        insert_workflow(connection, wrong_repo, "2026-01-04T00:00:00+00:00")
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
+        _sqlite.insert_workflow(connection, older, "2026-01-01T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, newest, "2026-01-02T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, plan, "2026-01-03T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, wrong_repo, "2026-01-04T00:00:00+00:00")
 
-    assert await repository.workflow_for_pr("alice", "repo", 17) == newest
+    assert await engine_repository.workflow_for_pr("alice", "repo", 17) == newest
 
 
 async def test_implementation_workflows_are_newest_first_and_require_a_pr(
-    repository: Repository,
+    engine_repository: Repository,
     tmp_path: Path,
 ) -> None:
     older = build_workflow("implementation-older", tmp_path, kind=WorkflowKind.IMPLEMENT, pr=17)
     newest = build_workflow("implementation-newest", tmp_path, kind=WorkflowKind.IMPLEMENT, pr=18)
     no_pr = build_workflow("implementation-no-pr", tmp_path, kind=WorkflowKind.IMPLEMENT)
     plan = build_workflow("plan-with-pr", tmp_path, pr=19)
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
-        insert_workflow(connection, older, "2026-01-01T00:00:00+00:00")
-        insert_workflow(connection, newest, "2026-01-02T00:00:00+00:00")
-        insert_workflow(connection, no_pr, "2026-01-03T00:00:00+00:00")
-        insert_workflow(connection, plan, "2026-01-04T00:00:00+00:00")
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
+        _sqlite.insert_workflow(connection, older, "2026-01-01T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, newest, "2026-01-02T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, no_pr, "2026-01-03T00:00:00+00:00")
+        _sqlite.insert_workflow(connection, plan, "2026-01-04T00:00:00+00:00")
 
-    assert await repository.implementation_workflows("alice", "repo", 3) == [newest, older]
+    assert await engine_repository.implementation_workflows("alice", "repo", 3) == [newest, older]

tests/test_repository_concurrency.py

@@ -1,7 +1,9 @@
 import asyncio
 import sqlite3
+import threading
+from collections.abc import Callable, Coroutine
 from contextlib import closing
-from pathlib import Path
+from typing import Any
 
 import pytest
 
@@ -9,14 +11,17 @@ from agentci.engine.events import PermissionDenied, PermissionGranted
 from agentci.engine.model import IncomingCommand, QueueName, TaskKind
 from agentci.engine.repository import Repository
 
-MIGRATIONS = Path(__file__).parents[1] / "src" / "agentci" / "migrations"
 
+async def run_concurrently[T](
+    *factories: Callable[[], Coroutine[Any, Any, T]],
+) -> list[T]:
+    barrier = threading.Barrier(len(factories))
 
-@pytest.fixture
-async def repository(tmp_path: Path) -> Repository:
-    value = Repository(tmp_path / "state.sqlite3", MIGRATIONS)
-    await value.initialize()
-    return value
+    def run(factory: Callable[[], Coroutine[Any, Any, T]]) -> T:
+        barrier.wait()
+        return asyncio.run(factory())
+
+    return list(await asyncio.gather(*(asyncio.to_thread(run, factory) for factory in factories)))
 
 
 def command(delivery: str, *, issue: int = 3) -> IncomingCommand:
@@ -33,91 +38,91 @@ def command(delivery: str, *, issue: int = 3) -> IncomingCommand:
 
 
 async def test_concurrent_duplicate_accepts_create_one_job_and_event(
-    repository: Repository,
+    engine_repository: Repository,
 ) -> None:
-    results = await asyncio.gather(
-        repository.accept(command("delivery-1")),
-        repository.accept(command("delivery-1")),
+    results = await run_concurrently(
+        lambda: engine_repository.accept(command("delivery-1")),
+        lambda: engine_repository.accept(command("delivery-1")),
     )
 
     assert sorted(result.duplicate for result in results) == [False, True]
     assert len({result.job.id for result in results}) == 1
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
         job_count = connection.execute("SELECT COUNT(*) FROM jobs").fetchone()
         event_count = connection.execute("SELECT COUNT(*) FROM job_events").fetchone()
     assert job_count == (1,)
     assert event_count == (1,)
 
-    second = await repository.accept(command("delivery-2"))
+    second = await engine_repository.accept(command("delivery-2"))
     assert second.job.receive_sequence == results[0].job.receive_sequence + 1
 
 
 async def test_received_job_blocks_later_execute_task_for_same_target(
-    repository: Repository,
+    engine_repository: Repository,
 ) -> None:
-    first = (await repository.accept(command("delivery-1"))).job
-    second = (await repository.accept(command("delivery-2"))).job
-    await repository.apply("grant-2", PermissionGranted(job_id=second.id))
+    first = (await engine_repository.accept(command("delivery-1"))).job
+    second = (await engine_repository.accept(command("delivery-2"))).job
+    await engine_repository.apply("grant-2", PermissionGranted(job_id=second.id))
 
-    assert await repository.claim_task(QueueName.JOBS) is None
+    assert await engine_repository.claim_task(QueueName.JOBS) is None
 
-    await repository.apply("deny-1", PermissionDenied(job_id=first.id))
-    task = await repository.claim_task(QueueName.JOBS)
+    await engine_repository.apply("deny-1", PermissionDenied(job_id=first.id))
+    task = await engine_repository.claim_task(QueueName.JOBS)
     assert task is not None
     assert task.job_id == second.id
     assert task.kind is TaskKind.EXECUTE
 
 
 async def test_received_job_does_not_block_a_different_target(
-    repository: Repository,
+    engine_repository: Repository,
 ) -> None:
-    await repository.accept(command("delivery-1", issue=3))
-    second = (await repository.accept(command("delivery-2", issue=4))).job
-    await repository.apply("grant-2", PermissionGranted(job_id=second.id))
+    await engine_repository.accept(command("delivery-1", issue=3))
+    second = (await engine_repository.accept(command("delivery-2", issue=4))).job
+    await engine_repository.apply("grant-2", PermissionGranted(job_id=second.id))
 
-    task = await repository.claim_task(QueueName.JOBS)
+    task = await engine_repository.claim_task(QueueName.JOBS)
     assert task is not None
     assert task.job_id == second.id
 
 
 async def test_claims_multiple_eligible_targets_without_duplicates(
-    repository: Repository,
+    engine_repository: Repository,
 ) -> None:
-    first = (await repository.accept(command("delivery-1", issue=3))).job
-    second = (await repository.accept(command("delivery-2", issue=4))).job
-    await repository.apply("grant-1", PermissionGranted(job_id=first.id))
-    await repository.apply("grant-2", PermissionGranted(job_id=second.id))
+    first = (await engine_repository.accept(command("delivery-1", issue=3))).job
+    second = (await engine_repository.accept(command("delivery-2", issue=4))).job
+    await engine_repository.apply("grant-2", PermissionGranted(job_id=second.id))
+    await engine_repository.apply("grant-1", PermissionGranted(job_id=first.id))
 
     claimed = [
-        await repository.claim_task(QueueName.JOBS),
-        await repository.claim_task(QueueName.JOBS),
+        await engine_repository.claim_task(QueueName.JOBS),
+        await engine_repository.claim_task(QueueName.JOBS),
     ]
 
     assert [task.job_id for task in claimed if task is not None] == [first.id, second.id]
-    assert await repository.claim_task(QueueName.JOBS) is None
+    assert await engine_repository.claim_task(QueueName.JOBS) is None
 
 
-async def test_concurrent_claims_do_not_duplicate_task(repository: Repository) -> None:
-    await repository.accept(command("delivery-1"))
+async def test_concurrent_claims_do_not_duplicate_task(engine_repository: Repository) -> None:
+    await engine_repository.accept(command("delivery-1"))
 
-    claims = await asyncio.gather(
-        repository.claim_task(QueueName.CONTROL),
-        repository.claim_task(QueueName.CONTROL),
+    claims = await run_concurrently(
+        lambda: engine_repository.claim_task(QueueName.CONTROL),
+        lambda: engine_repository.claim_task(QueueName.CONTROL),
     )
 
     claimed = [task for task in claims if task is not None]
     assert len(claimed) == 1
     assert claimed[0].kind is TaskKind.AUTHORIZE
     assert claimed[0].queue is QueueName.CONTROL
-    assert await repository.claim_task(QueueName.CONTROL) is None
+    assert await engine_repository.claim_task(QueueName.CONTROL) is None
 
 
 async def test_duplicate_event_id_cannot_be_reused_for_another_job(
-    repository: Repository,
+    engine_repository: Repository,
 ) -> None:
-    first = (await repository.accept(command("delivery-1", issue=3))).job
-    second = (await repository.accept(command("delivery-2", issue=4))).job
-    await repository.apply("permission", PermissionGranted(job_id=first.id))
+    first = (await engine_repository.accept(command("delivery-1", issue=3))).job
+    second = (await engine_repository.accept(command("delivery-2", issue=4))).job
+    await engine_repository.apply("permission", PermissionGranted(job_id=first.id))
 
     with pytest.raises(RuntimeError):
-        await repository.apply("permission", PermissionGranted(job_id=second.id))
+        await engine_repository.apply("permission", PermissionGranted(job_id=second.id))

tests/test_repository_migration.py

@@ -142,9 +142,11 @@ async def migrated_pre_v3_repository(tmp_path: Path) -> Repository:
     return repository
 
 
-async def test_schema_v2_migration_reconstructs_identity_order_and_iterate_commands(
+async def test_schema_v2_migration_preserves_jobs_and_reconstructs_state_machine_identity(
     migrated_pre_v3_repository: Repository,
 ) -> None:
+    terminal = await migrated_pre_v3_repository.get_job("pre-v3-job-a-terminal")
+    queued = await migrated_pre_v3_repository.get_job("pre-v3-job-b-queued")
     with (
         closing(sqlite3.connect(migrated_pre_v3_repository.database_path)) as connection,
         connection,
@@ -153,6 +155,10 @@ async def test_schema_v2_migration_reconstructs_identity_order_and_iterate_comma
             "SELECT id, delivery_id, receive_sequence, command_body FROM jobs "
             "ORDER BY receive_sequence"
         ).fetchall()
+        storage_fields = connection.execute(
+            """SELECT id, started_comment_id, created_at, started_at, finished_at
+            FROM jobs ORDER BY receive_sequence"""
+        ).fetchall()
 
     assert migrated == [
         (
@@ -164,21 +170,6 @@ async def test_schema_v2_migration_reconstructs_identity_order_and_iterate_comma
         ("pre-v3-job-b-queued", "delivery-queued", 2, "/agent iterate refine plan"),
     ]
 
-
-async def test_schema_v2_migration_preserves_pre_v3_job_fields(
-    migrated_pre_v3_repository: Repository,
-) -> None:
-    terminal = await migrated_pre_v3_repository.get_job("pre-v3-job-a-terminal")
-    queued = await migrated_pre_v3_repository.get_job("pre-v3-job-b-queued")
-    with (
-        closing(sqlite3.connect(migrated_pre_v3_repository.database_path)) as connection,
-        connection,
-    ):
-        storage_fields = connection.execute(
-            """SELECT id, started_comment_id, created_at, started_at, finished_at
-            FROM jobs ORDER BY receive_sequence"""
-        ).fetchall()
-
     assert (
         terminal
         and (
@@ -386,7 +377,18 @@ def durable_snapshot(database_path: Path) -> tuple[object, ...]:
         return (
             connection.execute("SELECT version FROM schema_migrations ORDER BY version").fetchall(),
             connection.execute(
-                "SELECT id, delivery_id, receive_sequence, command_body FROM jobs ORDER BY id"
+                """SELECT id, kind, target_key, repo_owner, repo_name, issue_number, pr_number,
+                          requester, message, comment_id, delivery_id, receive_sequence,
+                          command_body, workflow_id, status, stage, error, runtime_session_id,
+                          accepted_comment_id, started_comment_id, comment_body, created_at,
+                          started_at, finished_at
+                FROM jobs ORDER BY id"""
+            ).fetchall(),
+            connection.execute(
+                """SELECT id, kind, repo_owner, repo_name, issue_number, pr_number, base_sha,
+                          branch, workspace_path, primary_session_id, reviewer_session_id,
+                          artifact, review_json, status, runtime, created_at, updated_at
+                FROM workflows ORDER BY id"""
             ).fetchall(),
             connection.execute(
                 """SELECT event_id, job_id, event_type, payload_json

tests/test_repository_tasks.py

@@ -1,24 +1,14 @@
 import sqlite3
 from contextlib import closing
 from datetime import UTC, datetime, timedelta
-from pathlib import Path
 
 import pytest
 
-from agentci.engine import _sqlite
 from agentci.engine import repository as repository_module
 from agentci.engine.events import JobStarted, PermissionGranted
 from agentci.engine.model import IncomingCommand, QueueName, Task, TaskKind
 from agentci.engine.repository import Repository
-
-MIGRATIONS = Path(__file__).parents[1] / "src" / "agentci" / "migrations"
-
-
-@pytest.fixture
-async def repository(tmp_path: Path) -> Repository:
-    value = Repository(tmp_path / "state.sqlite3", MIGRATIONS)
-    await value.initialize()
-    return value
+from tests.conftest import SQLiteClock
 
 
 def command(delivery: str, *, issue: int = 3) -> IncomingCommand:
@@ -34,8 +24,8 @@ def command(delivery: str, *, issue: int = 3) -> IncomingCommand:
     )
 
 
-def task_storage(repository: Repository, task_id: int) -> tuple[object, ...]:
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
+def task_storage(engine_repository: Repository, task_id: int) -> tuple[object, ...]:
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
         row = connection.execute(
             """SELECT status, attempts, available_at, error, created_at, started_at, finished_at
             FROM listener_tasks WHERE id=?""",
@@ -46,20 +36,18 @@ def task_storage(repository: Repository, task_id: int) -> tuple[object, ...]:
 
 
 async def test_claim_and_complete_record_attempt_and_lifecycle_timestamps(
-    repository: Repository,
-    monkeypatch: pytest.MonkeyPatch,
+    engine_repository: Repository,
+    sqlite_clock: SQLiteClock,
 ) -> None:
-    clock = {"now": "2026-02-01T00:00:00+00:00"}
-    monkeypatch.setattr(_sqlite, "now", lambda: clock["now"])
-    job = (await repository.accept(command("delivery-1"))).job
-    clock["now"] = "2026-02-01T00:01:00+00:00"
+    job = (await engine_repository.accept(command("delivery-1"))).job
+    sqlite_clock.now = "2026-02-01T00:01:00+00:00"
 
-    task = await repository.claim_task(QueueName.CONTROL)
+    task = await engine_repository.claim_task(QueueName.CONTROL)
     assert task is not None
-    clock["now"] = "2026-02-01T00:02:00+00:00"
-    await repository.complete_task(task.id)
+    sqlite_clock.now = "2026-02-01T00:02:00+00:00"
+    await engine_repository.complete_task(task.id)
 
-    assert (task, task_storage(repository, task.id)) == (
+    assert (task, task_storage(engine_repository, task.id)) == (
         Task(
             id=task.id,
             job_id=job.id,
@@ -82,17 +70,16 @@ async def test_claim_and_complete_record_attempt_and_lifecycle_timestamps(
 
 @pytest.mark.parametrize(("attempts", "delay_seconds"), [(1, 2), (9, 256)])
 async def test_retry_uses_bounded_backoff_and_preserves_attempt_history_until_due(
-    repository: Repository,
+    engine_repository: Repository,
+    sqlite_clock: SQLiteClock,
     monkeypatch: pytest.MonkeyPatch,
     attempts: int,
     delay_seconds: int,
 ) -> None:
-    clock = {"now": "2026-02-01T00:00:00+00:00"}
-    monkeypatch.setattr(_sqlite, "now", lambda: clock["now"])
-    await repository.accept(command("delivery-1"))
-    task = await repository.claim_task(QueueName.CONTROL)
+    await engine_repository.accept(command("delivery-1"))
+    task = await engine_repository.claim_task(QueueName.CONTROL)
     assert task is not None
-    with closing(sqlite3.connect(repository.database_path)) as connection, connection:
+    with closing(sqlite3.connect(engine_repository.database_path)) as connection, connection:
         connection.execute("UPDATE listener_tasks SET attempts=? WHERE id=?", (attempts, task.id))
 
     retry_time = datetime(2026, 2, 1, 1, tzinfo=UTC)
@@ -105,12 +92,12 @@ async def test_retry_uses_bounded_backoff_and_preserves_attempt_history_until_du
 
     monkeypatch.setattr(repository_module, "datetime", FixedDateTime)
     error = "failure: " + "x" * 1100
-    await repository.retry_task(task.id, attempts, error)
-    pending_storage = task_storage(repository, task.id)
-    clock["now"] = (available_at - timedelta(microseconds=1)).isoformat()
-    early_claim = await repository.claim_task(QueueName.CONTROL)
-    clock["now"] = available_at.isoformat()
-    due_claim = await repository.claim_task(QueueName.CONTROL)
+    await engine_repository.retry_task(task.id, attempts, error)
+    pending_storage = task_storage(engine_repository, task.id)
+    sqlite_clock.now = (available_at - timedelta(microseconds=1)).isoformat()
+    early_claim = await engine_repository.claim_task(QueueName.CONTROL)
+    sqlite_clock.now = available_at.isoformat()
+    due_claim = await engine_repository.claim_task(QueueName.CONTROL)
 
     assert (
         pending_storage,
@@ -132,35 +119,33 @@ async def test_retry_uses_bounded_backoff_and_preserves_attempt_history_until_du
 
 
 async def test_recovery_requeues_control_and_unstarted_execution_but_fails_started_execution(
-    repository: Repository,
-    monkeypatch: pytest.MonkeyPatch,
+    engine_repository: Repository,
+    sqlite_clock: SQLiteClock,
 ) -> None:
-    clock = {"now": "2026-02-01T00:00:00+00:00"}
-    monkeypatch.setattr(_sqlite, "now", lambda: clock["now"])
 
-    await repository.accept(command("delivery-1", issue=1))
-    control_task = await repository.claim_task(QueueName.CONTROL)
+    await engine_repository.accept(command("delivery-1", issue=1))
+    control_task = await engine_repository.claim_task(QueueName.CONTROL)
     assert control_task is not None
 
-    queued_job = (await repository.accept(command("delivery-2", issue=2))).job
-    await repository.apply("queued:grant", PermissionGranted(job_id=queued_job.id))
-    queued_execute = await repository.claim_task(QueueName.JOBS)
+    queued_job = (await engine_repository.accept(command("delivery-2", issue=2))).job
+    await engine_repository.apply("queued:grant", PermissionGranted(job_id=queued_job.id))
+    queued_execute = await engine_repository.claim_task(QueueName.JOBS)
     assert queued_execute is not None
 
-    running_job = (await repository.accept(command("delivery-3", issue=3))).job
-    await repository.apply("running:grant", PermissionGranted(job_id=running_job.id))
-    running_execute = await repository.claim_task(QueueName.JOBS)
+    running_job = (await engine_repository.accept(command("delivery-3", issue=3))).job
+    await engine_repository.apply("running:grant", PermissionGranted(job_id=running_job.id))
+    running_execute = await engine_repository.claim_task(QueueName.JOBS)
     assert running_execute is not None
-    await repository.apply("running:start", JobStarted(job_id=running_job.id))
+    await engine_repository.apply("running:start", JobStarted(job_id=running_job.id))
 
-    clock["now"] = "2026-02-01T01:00:00+00:00"
-    await repository.recover_tasks()
+    sqlite_clock.now = "2026-02-01T01:00:00+00:00"
+    await engine_repository.recover_tasks()
 
     assert (
-        task_storage(repository, control_task.id),
-        task_storage(repository, queued_execute.id),
-        task_storage(repository, running_execute.id),
-        [job.id for job in await repository.running_jobs()],
+        task_storage(engine_repository, control_task.id),
+        task_storage(engine_repository, queued_execute.id),
+        task_storage(engine_repository, running_execute.id),
+        [job.id for job in await engine_repository.running_jobs()],
     ) == (
         (
             "pending",

tests/test_runtime.py

@@ -32,36 +32,30 @@ def runtime(opencode: object, gitea: object) -> Runtime:
     )
 
 
-async def test_close_releases_provider_clients_in_order() -> None:
-    events: list[str] = []
-    value = runtime(ClosingClient("opencode", events), ClosingClient("gitea", events))
-
-    await value.close()
-
-    assert events == ["opencode", "gitea"]
-
-
-async def test_close_still_releases_gitea_when_opencode_close_fails() -> None:
+@pytest.mark.parametrize(
+    ("opencode_error", "gitea_error", "expected_error"),
+    [
+        (None, None, None),
+        (RuntimeError("opencode close failed"), None, "opencode close failed"),
+        (None, RuntimeError("gitea close failed"), "gitea close failed"),
+    ],
+    ids=["success", "opencode-failure", "gitea-failure"],
+)
+async def test_close_releases_provider_clients_in_order(
+    opencode_error: Exception | None,
+    gitea_error: Exception | None,
+    expected_error: str | None,
+) -> None:
     events: list[str] = []
     value = runtime(
-        ClosingClient("opencode", events, RuntimeError("opencode close failed")),
-        ClosingClient("gitea", events),
+        ClosingClient("opencode", events, opencode_error),
+        ClosingClient("gitea", events, gitea_error),
     )
 
-    with pytest.raises(RuntimeError, match="opencode close failed"):
+    if expected_error is None:
         await value.close()
-
-    assert events == ["opencode", "gitea"]
-
-
-async def test_close_propagates_gitea_close_failure_after_opencode_closes() -> None:
-    events: list[str] = []
-    value = runtime(
-        ClosingClient("opencode", events),
-        ClosingClient("gitea", events, RuntimeError("gitea close failed")),
-    )
-
-    with pytest.raises(RuntimeError, match="gitea close failed"):
+    else:
+        with pytest.raises(RuntimeError, match=expected_error):
             await value.close()
 
     assert events == ["opencode", "gitea"]

tests/test_webhook.py

@@ -98,23 +98,38 @@ async def test_invalid_signature_precedes_parsing_and_event_filtering(event: str
     assert repository.accepted == []
 
 
-async def test_signed_unsupported_event_is_ignored_without_parsing() -> None:
-    repository = FakeRepository()
-
-    response = await post_webhook(repository, b"not-json", event="push")
-
-    assert response.status_code == 204
-    assert repository.accepted == []
-
-
 @pytest.mark.parametrize(
-    ("command", "requester"),
-    [("ordinary discussion", "alice"), ("/agent plan", "AgentCI")],
+    ("body", "event"),
+    [
+        pytest.param(
+            b"not-json",
+            "push",
+            id="unsupported-event-before-parsing",
+        ),
+        pytest.param(
+            encoded(payload("ordinary discussion")),
+            "issue_comment",
+            id="non-command",
+        ),
+        pytest.param(
+            encoded(payload("/agent plan", requester="AgentCI")),
+            "issue_comment",
+            id="bot-author",
+        ),
+        pytest.param(
+            encoded({"action": "edited"}),
+            "issue_comment",
+            id="non-created-before-payload",
+        ),
+    ],
 )
-async def test_non_command_and_bot_comment_are_ignored(command: str, requester: str) -> None:
+async def test_authenticated_irrelevant_webhooks_are_ignored(
+    body: bytes,
+    event: str,
+) -> None:
     repository = FakeRepository()
 
-    response = await post_webhook(repository, encoded(payload(command, requester=requester)))
+    response = await post_webhook(repository, body, event=event)
 
     assert response.status_code == 204
     assert repository.accepted == []
@@ -130,40 +145,41 @@ async def test_duplicate_delivery_returns_ok() -> None:
 
 
 @pytest.mark.parametrize(
-    "body",
+    ("body", "delivery", "detail"),
     [
-        b"{",
-        encoded([]),
+        pytest.param(b"{", "delivery", "Invalid webhook payload", id="invalid-json"),
+        pytest.param(encoded([]), "delivery", "Invalid webhook payload", id="non-object"),
+        pytest.param(
             encoded({"action": "created"}),
+            "delivery",
+            "Invalid webhook payload",
+            id="missing-comment",
+        ),
+        pytest.param(
             encoded({"action": "created", "comment": {}}),
+            "delivery",
+            "Invalid webhook payload",
+            id="incomplete-comment",
+        ),
+        pytest.param(
+            encoded(payload("/agent plan")),
+            None,
+            "Missing X-Gitea-Delivery",
+            id="missing-delivery",
+        ),
     ],
 )
-async def test_malformed_json_and_payload_contract_return_bad_request(body: bytes) -> None:
+async def test_authenticated_bad_requests_are_rejected_without_persistence(
+    body: bytes,
+    delivery: str | None,
+    detail: str,
+) -> None:
     repository = FakeRepository()
 
-    response = await post_webhook(repository, body)
+    response = await post_webhook(repository, body, delivery=delivery)
 
     assert response.status_code == 400
-    assert response.json() == {"detail": "Invalid webhook payload"}
-    assert repository.accepted == []
-
-
-async def test_command_requires_delivery_header() -> None:
-    repository = FakeRepository()
-
-    response = await post_webhook(repository, encoded(payload("/agent plan")), delivery=None)
-
-    assert response.status_code == 400
-    assert response.json() == {"detail": "Missing X-Gitea-Delivery"}
-    assert repository.accepted == []
-
-
-async def test_non_created_comment_is_ignored_without_nested_payload() -> None:
-    repository = FakeRepository()
-
-    response = await post_webhook(repository, encoded({"action": "edited"}))
-
-    assert response.status_code == 204
+    assert response.json() == {"detail": detail}
     assert repository.accepted == []
 
 

tests/test_worker.py

@@ -20,7 +20,6 @@ from agentci.engine.events import (
     JobRejected as RejectedEvent,
 )
 from agentci.engine.model import (
-    IncomingCommand,
     Job,
     JobKind,
     JobStatus,
@@ -31,12 +30,9 @@ from agentci.engine.model import (
     WorkflowKind,
 )
 from agentci.engine.reducer import render_job_comment
-from agentci.engine.repository import Repository
 from agentci.integrations.gitea.models import CommentInfo
 from agentci.workflows.render import JobRejected
 
-MIGRATIONS = Path(__file__).parents[1] / "src" / "agentci" / "migrations"
-
 
 class FakeRepository:
     def __init__(
@@ -54,7 +50,6 @@ class FakeRepository:
         self.events: list[tuple[str, object]] = []
         self.operations: list[str] = []
         self.running: list[Job] = []
-        self.failed_workflows: list[str] = []
         self.complete_stop: asyncio.Event | None = None
 
     async def claim_task(self, queue: QueueName) -> Task | None:
@@ -91,9 +86,6 @@ class FakeRepository:
     async def get_workflow(self, _workflow_id: str) -> Workflow | None:
         return self.workflow
 
-    async def fail_job_workflow(self, job_id: str) -> None:
-        self.failed_workflows.append(job_id)
-
 
 class FakeGitea:
     def __init__(self, *, permitted: bool = True) -> None:
@@ -181,7 +173,7 @@ def task(
 
 def make_worker(
     tmp_path: Path,
-    repository: FakeRepository | Repository,
+    repository: FakeRepository,
     *,
     gitea: FakeGitea | None = None,
     opencode: FakeOpenCode | None = None,
@@ -329,74 +321,47 @@ async def test_run_recovers_before_starting_configured_consumers(
 
 
 @pytest.mark.parametrize(
-    ("permitted", "event_type", "suffix"),
+    ("status", "permitted", "expected_events", "expected_permission_calls"),
     [
-        (True, PermissionGranted, "permission-granted"),
-        (False, PermissionDenied, "permission-denied"),
+        pytest.param(
+            JobStatus.RECEIVED,
+            True,
+            [("task:7:permission-granted", PermissionGranted(job_id="job"))],
+            [("org", "repo", "alice")],
+            id="received-permitted",
+        ),
+        pytest.param(
+            JobStatus.RECEIVED,
+            False,
+            [("task:7:permission-denied", PermissionDenied(job_id="job"))],
+            [("org", "repo", "alice")],
+            id="received-denied",
+        ),
+        pytest.param(
+            JobStatus.QUEUED,
+            True,
+            [],
+            [],
+            id="queued",
+        ),
     ],
 )
-async def test_authorize_records_permission_outcome(
+async def test_handle_authorization_obeys_job_state_and_permission(
     tmp_path: Path,
+    status: JobStatus,
     permitted: bool,
-    event_type: type[PermissionGranted] | type[PermissionDenied],
-    suffix: str,
+    expected_events: list[tuple[str, object]],
+    expected_permission_calls: list[tuple[str, str, str]],
 ) -> None:
-    current = job()
-    repository = FakeRepository(current)
+    repository = FakeRepository(job(status=status))
     gitea = FakeGitea(permitted=permitted)
-    authorize = task(TaskKind.AUTHORIZE, QueueName.CONTROL)
 
-    await make_worker(tmp_path, repository, gitea=gitea)._authorize(authorize, current)
-
-    assert gitea.permission_calls == [("org", "repo", "alice")]
-    assert repository.events[0][0] == f"task:{authorize.id}:{suffix}"
-    assert isinstance(repository.events[0][1], event_type)
-
-
-async def test_authorize_ignores_job_after_received_status(tmp_path: Path) -> None:
-    current = job(status=JobStatus.QUEUED)
-    repository = FakeRepository(current)
-    gitea = FakeGitea()
-
-    await make_worker(tmp_path, repository, gitea=gitea)._authorize(
-        task(TaskKind.AUTHORIZE, QueueName.CONTROL), current
+    await make_worker(tmp_path, repository, gitea=gitea)._handle(
+        task(TaskKind.AUTHORIZE, QueueName.CONTROL)
     )
 
-    assert gitea.permission_calls == []
-    assert repository.events == []
-
-
-async def test_authorize_integrates_with_real_repository(
-    tmp_path: Path,
-) -> None:
-    repository = Repository(tmp_path / "state.sqlite3", MIGRATIONS)
-    await repository.initialize()
-    accepted = await repository.accept(
-        IncomingCommand(
-            delivery_id="delivery-real",
-            comment_id=1,
-            repo_owner="org",
-            repo_name="repo",
-            issue_number=1,
-            pr_number=None,
-            requester="alice",
-            body="/agent plan write tests",
-        )
-    )
-    authorize = await repository.claim_task(QueueName.CONTROL)
-    assert authorize is not None
-    assert authorize.kind is TaskKind.AUTHORIZE
-
-    await make_worker(tmp_path, repository)._handle(authorize)
-
-    persisted = await repository.get_job(accepted.job.id)
-    assert persisted is not None
-    assert persisted.status is JobStatus.QUEUED
-    assert persisted.kind is JobKind.PLAN
-    assert persisted.message == "write tests"
-    execute = await repository.claim_task(QueueName.JOBS)
-    assert execute is not None
-    assert execute.kind is TaskKind.EXECUTE
+    assert repository.events == expected_events
+    assert gitea.permission_calls == expected_permission_calls
 
 
 async def test_execute_completes_with_workflow_comment(
@@ -603,20 +568,23 @@ async def test_abort_collects_and_deduplicates_workflow_sessions(
     assert set(opencode.aborted) == {(session, workspace) for session in expected}
 
 
-async def test_abort_uses_one_shot_workspace_without_workflow(tmp_path: Path) -> None:
-    opencode = FakeOpenCode()
-
-    await make_worker(tmp_path, FakeRepository(), opencode=opencode)._abort_job_sessions(
-        job(workflow_id="missing", session_id="session")
+@pytest.mark.parametrize(
+    ("workflow_id", "session_id", "has_workflow", "uses_one_shot"),
+    [
+        pytest.param("missing", "session", False, True, id="one-shot-fallback"),
+        pytest.param("flow", "one-shot", True, False, id="workflow-precedence"),
+        pytest.param(None, None, False, False, id="no-sessions"),
+    ],
 )
-
-    assert opencode.aborted == [("session", tmp_path / "fix-job" / "repo")]
-
-
-async def test_abort_does_not_mix_one_shot_session_into_existing_workflow(
+async def test_abort_session_source_precedence(
     tmp_path: Path,
+    workflow_id: str | None,
+    session_id: str | None,
+    has_workflow: bool,
+    uses_one_shot: bool,
 ) -> None:
-    workflow = Workflow(
+    persisted = (
+        Workflow(
             id="flow",
             kind=WorkflowKind.IMPLEMENT,
             repo_owner="org",
@@ -625,21 +593,19 @@ async def test_abort_does_not_mix_one_shot_session_into_existing_workflow(
             workspace_path=tmp_path / "workflow" / "repo",
             base_sha="base",
         )
+        if has_workflow
+        else None
+    )
     opencode = FakeOpenCode()
 
     await make_worker(
-        tmp_path, FakeRepository(workflow=workflow), opencode=opencode
-    )._abort_job_sessions(job(workflow_id="flow", session_id="one-shot"))
+        tmp_path,
+        FakeRepository(workflow=persisted),
+        opencode=opencode,
+    )._abort_job_sessions(job(workflow_id=workflow_id, session_id=session_id))
 
-    assert opencode.aborted == []
-
-
-async def test_abort_without_persisted_sessions_is_noop(tmp_path: Path) -> None:
-    opencode = FakeOpenCode()
-
-    await make_worker(tmp_path, FakeRepository(), opencode=opencode)._abort_job_sessions(job())
-
-    assert opencode.aborted == []
+    expected = [("session", tmp_path / "fix-job" / "repo")] if uses_one_shot else []
+    assert opencode.aborted == expected
 
 
 def test_safe_error_is_single_line_and_bounded() -> None:

tests/test_workflow_dispatch.py

@@ -1,5 +1,3 @@
-from dataclasses import replace
-from enum import StrEnum
 from typing import cast
 
 import pytest
@@ -43,6 +41,7 @@ async def test_dispatch_routes_every_job_kind_and_returns_body(
     kind: JobKind,
     expected_route: str,
 ) -> None:
+    assert {route_kind for route_kind, _ in ROUTES} == set(JobKind)
     calls: list[tuple[str, Job]] = []
 
     def route(name: str):
@@ -73,21 +72,3 @@ async def test_dispatch_rejects_unparsed_command() -> None:
             cast(JobRun, object()),
             cast(WorkflowServices, object()),
         )
-
-
-class UnsupportedKind(StrEnum):
-    UNKNOWN = "unknown"
-
-
-async def test_dispatch_rejects_unsupported_kind() -> None:
-    unsupported = UnsupportedKind.UNKNOWN
-    job = replace(make_job(JobKind.PLAN), kind=cast(JobKind, unsupported))
-
-    with pytest.raises(KeyError) as error:
-        await dispatch_module.dispatch(
-            job,
-            cast(JobRun, object()),
-            cast(WorkflowServices, object()),
-        )
-
-    assert error.value.args == (unsupported,)

tests/test_workflow_implementation.py

@@ -1,33 +1,15 @@
 import json
 from pathlib import Path
 from types import SimpleNamespace
-from typing import Any, cast
 
 import pytest
 
 from agentci.engine.model import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus
-from agentci.engine.run import JobRun
 from agentci.integrations.gitea.models import CommentInfo, IssueInfo, PullRequestInfo
 from agentci.workflows.implementation import implement
 from agentci.workflows.model import AgentResult, ReviewReport
 from agentci.workflows.render import JobRejected
-from agentci.workflows.services import WorkflowServices
-
-
-class RecordingRun(JobRun):
-    def __init__(self) -> None:
-        self.stages: list[str] = []
-        self.created_workflows: list[tuple[Workflow, str]] = []
-        self.linked_sessions: list[str] = []
-
-    async def stage(self, stage: str) -> None:
-        self.stages.append(stage)
-
-    async def create_workflow(self, workflow: Workflow, stage: str) -> None:
-        self.created_workflows.append((workflow, stage))
-
-    async def link_session(self, session_id: str) -> None:
-        self.linked_sessions.append(session_id)
+from tests.workflow_support import make_workflow_harness
 
 
 class FakeRepository:
@@ -85,71 +67,6 @@ class FakeGitea:
         return pull(number=42, branch=values["head"])
 
 
-class RecordingGit:
-    def __init__(self, *, changed: bool = True) -> None:
-        self.changed = changed
-        self.calls: list[tuple[Any, ...]] = []
-
-    async def clone(self, owner: str, repo: str, branch: str, destination: Path) -> str:
-        self.calls.append(("clone", owner, repo, branch, destination))
-        return "base-sha"
-
-    async def create_branch(self, workspace: Path, branch: str) -> None:
-        self.calls.append(("create_branch", workspace, branch))
-
-    async def has_changes(self, workspace: Path) -> bool:
-        self.calls.append(("has_changes", workspace))
-        return self.changed
-
-    async def diff_check(self, workspace: Path) -> None:
-        self.calls.append(("diff_check", workspace))
-
-    async def commit(self, workspace: Path, message: str) -> str:
-        self.calls.append(("commit", workspace, message))
-        return "commit-sha"
-
-    async def push(self, workspace: Path, branch: str, *, set_upstream: bool = False) -> None:
-        self.calls.append(("push", workspace, branch, set_upstream))
-
-
-class RecordingDevelopment:
-    description = "Python 3.13"
-
-    def __init__(self) -> None:
-        self.workspaces: list[Path] = []
-
-    async def prepare(self, workspace: Path) -> None:
-        self.workspaces.append(workspace)
-
-
-class RecordingPrompts:
-    def __init__(self) -> None:
-        self.calls: list[tuple[str, dict[str, str]]] = []
-
-    def render(self, name: str, **values: str) -> str:
-        self.calls.append((name, values))
-        return f"rendered {name}"
-
-
-class RecordingOpenCode:
-    def __init__(self, result: AgentResult, report: ReviewReport) -> None:
-        self.result = result
-        self.report = report
-        self.created_sessions: list[tuple[Path, str]] = []
-        self.resume_calls: list[dict[str, Any]] = []
-
-    async def create_session(self, workspace: Path, title: str) -> str:
-        self.created_sessions.append((workspace, title))
-        return f"{title}-session"
-
-    async def resume(self, **values: Any) -> AgentResult | ReviewReport:
-        self.resume_calls.append(values)
-        if values["result_type"] is AgentResult:
-            return self.result
-        assert values["result_type"] is ReviewReport
-        return self.report
-
-
 def job() -> Job:
     return Job(
         id="job-1",
@@ -203,95 +120,63 @@ def pull(
     )
 
 
-def make_services(
-    tmp_path: Path,
-    *,
-    changed: bool = True,
-    implementations: list[Workflow] | None = None,
-    existing_pulls: dict[int, PullRequestInfo] | None = None,
-    plan: Workflow | None = None,
-) -> tuple[
-    WorkflowServices,
-    FakeRepository,
-    FakeGitea,
-    RecordingGit,
-    RecordingDevelopment,
-    RecordingPrompts,
-    RecordingOpenCode,
-]:
-    repository = FakeRepository(implementations=implementations, plan=plan)
-    gitea = FakeGitea(existing_pulls=existing_pulls)
-    git = RecordingGit(changed=changed)
-    development = RecordingDevelopment()
-    prompts = RecordingPrompts()
-    opencode = RecordingOpenCode(
-        AgentResult(
-            summary_markdown="# Implement widget\n\nHandled empty input.",
-            tests=["pytest: passed"],
-        ),
-        ReviewReport(summary="Ready", findings=[]),
-    )
-    services = cast(
-        WorkflowServices,
-        SimpleNamespace(
-            settings=SimpleNamespace(
+def implementation_settings(tmp_path: Path) -> SimpleNamespace:
+    return SimpleNamespace(
         branch_prefix="agent",
         workspaces_dir=tmp_path / "workspaces",
         gitea_url="https://git.example.test",
         implement_model="provider/model",
         implement_variant="high",
         implement_review_rounds=2,
-            ),
-            repository=repository,
-            gitea=gitea,
-            git=git,
-            opencode=opencode,
-            prompts=prompts,
-            development=development,
-        ),
     )
-    return services, repository, gitea, git, development, prompts, opencode
 
 
 async def test_initial_implementation_completes_review_commit_push_and_pr(
     tmp_path: Path,
 ) -> None:
     stale = workflow(pr_number=8)
-    services, repository, gitea, git, development, prompts, opencode = make_services(
-        tmp_path,
-        implementations=[stale],
-        existing_pulls={8: pull(state="closed")},
+    repository = FakeRepository(implementations=[stale])
+    gitea = FakeGitea(existing_pulls={8: pull(state="closed")})
+    harness = make_workflow_harness(
+        settings=implementation_settings(tmp_path),
+        repository=repository,
+        gitea=gitea,
+        responses=[
+            AgentResult(
+                summary_markdown="# Implement widget\n\nHandled empty input.",
+                tests=["pytest: passed"],
+            ),
+            ReviewReport(summary="Ready", findings=[]),
+        ],
     )
-    run = RecordingRun()
 
-    body = await implement(job(), run, services)
+    body = await implement(job(), harness.run, harness.services)
 
-    assert len(run.created_workflows) == 1
-    created, created_stage = run.created_workflows[0]
+    assert len(harness.run.created_workflows) == 1
+    created, created_stage = harness.run.created_workflows[0]
     assert created_stage == "installing development environment"
     assert created.status is WorkflowStatus.ACTIVE
     assert created.branch == f"agent/issue-7-{created.id[:8]}"
     assert created.workspace_path == tmp_path / "workspaces" / created.id / "repo"
     assert created.base_sha == "base-sha"
-    assert development.workspaces == [created.workspace_path]
-    assert run.linked_sessions == ["implementation-session"]
-    assert opencode.created_sessions == [
+    assert harness.development.workspaces == [created.workspace_path]
+    assert harness.run.linked_sessions == ["implementation-session"]
+    assert harness.opencode.created_sessions == [
         (created.workspace_path, "implementation"),
         (created.workspace_path, "implementation-review"),
     ]
-    assert [call["result_type"] for call in opencode.resume_calls] == [
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [
         AgentResult,
         ReviewReport,
     ]
 
-    initial_prompt = prompts.calls[0]
+    initial_prompt = harness.prompts.calls[0]
     assert initial_prompt[0] == "implement_initial"
     assert initial_prompt[1]["artifact"] == "(no canonical plan)"
     assert initial_prompt[1]["request"] == "Keep the change focused."
-    assert "Please cover empty input." in initial_prompt[1]["context"]
-    assert "Job queued" not in initial_prompt[1]["context"]
+    assert initial_prompt[1]["context"].startswith("Repository: org/repo\n")
 
-    assert git.calls == [
+    assert harness.git.calls == [
         ("clone", "org", "repo", "main", created.workspace_path),
         ("create_branch", created.workspace_path, created.branch),
         ("has_changes", created.workspace_path),
@@ -338,16 +223,29 @@ async def test_initial_implementation_completes_review_commit_push_and_pr(
 async def test_initial_implementation_rejects_clean_worktree_before_commit_or_pr(
     tmp_path: Path,
 ) -> None:
-    services, repository, gitea, git, _, _, _ = make_services(tmp_path, changed=False)
-    run = RecordingRun()
+    repository = FakeRepository()
+    gitea = FakeGitea()
+    harness = make_workflow_harness(
+        settings=implementation_settings(tmp_path),
+        repository=repository,
+        gitea=gitea,
+        changed=False,
+        responses=[
+            AgentResult(
+                summary_markdown="# Implement widget\n\nHandled empty input.",
+                tests=["pytest: passed"],
+            ),
+            ReviewReport(summary="Ready", findings=[]),
+        ],
+    )
 
     with pytest.raises(
         JobRejected,
         match="OpenCode completed without producing any file changes",
     ):
-        await implement(job(), run, services)
+        await implement(job(), harness.run, harness.services)
 
-    assert [call[0] for call in git.calls] == [
+    assert [call[0] for call in harness.git.calls] == [
         "clone",
         "create_branch",
         "has_changes",
@@ -355,7 +253,7 @@ async def test_initial_implementation_rejects_clean_worktree_before_commit_or_pr
     assert gitea.created_pulls == []
     assert repository.saved_workflows
     assert all(item.status is WorkflowStatus.ACTIVE for item in repository.saved_workflows)
-    assert "creating pull request" not in run.stages
+    assert "creating pull request" not in harness.run.stages
 
 
 @pytest.mark.parametrize(
@@ -376,17 +274,19 @@ async def test_initial_implementation_rejects_duplicate_agent_pr_before_clone(
     existing_pull: PullRequestInfo,
     message: str,
 ) -> None:
-    services, _, gitea, git, development, _, opencode = make_services(
-        tmp_path,
-        implementations=[workflow(pr_number=8)],
-        existing_pulls={8: existing_pull},
+    repository = FakeRepository(implementations=[workflow(pr_number=8)])
+    gitea = FakeGitea(existing_pulls={8: existing_pull})
+    harness = make_workflow_harness(
+        settings=implementation_settings(tmp_path),
+        repository=repository,
+        gitea=gitea,
     )
 
     with pytest.raises(JobRejected) as error:
-        await implement(job(), RecordingRun(), services)
+        await implement(job(), harness.run, harness.services)
 
     assert str(error.value) == message
-    assert git.calls == []
-    assert development.workspaces == []
-    assert opencode.resume_calls == []
+    assert harness.git.calls == []
+    assert harness.development.workspaces == []
+    assert harness.opencode.resume_calls == []
     assert gitea.default_branch_calls == []

tests/test_workflow_plan.py

@@ -1,38 +1,15 @@
 import json
 from pathlib import Path
 from types import SimpleNamespace
-from typing import Any, cast
 
 import pytest
-from pydantic import BaseModel
 
 from agentci.engine.model import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus
-from agentci.engine.run import JobRun
 from agentci.integrations.gitea.models import CommentInfo, IssueInfo, PullRequestInfo
 from agentci.workflows.model import DiscussionReply, PlanArtifact, ReviewReport
 from agentci.workflows.plan import create_plan, discuss_plan, iterate_plan
 from agentci.workflows.render import JobRejected
-from agentci.workflows.services import WorkflowServices
-
-
-class RecordingRun(JobRun):
-    def __init__(self) -> None:
-        self.stages: list[str] = []
-        self.created_workflows: list[tuple[Workflow, str]] = []
-        self.linked_workflows: list[tuple[str, str]] = []
-        self.linked_sessions: list[str] = []
-
-    async def stage(self, stage: str) -> None:
-        self.stages.append(stage)
-
-    async def create_workflow(self, workflow: Workflow, stage: str) -> None:
-        self.created_workflows.append((workflow, stage))
-
-    async def link_workflow(self, workflow_id: str, stage: str) -> None:
-        self.linked_workflows.append((workflow_id, stage))
-
-    async def link_session(self, session_id: str) -> None:
-        self.linked_sessions.append(session_id)
+from tests.workflow_support import make_workflow_harness
 
 
 class FakeRepository:
@@ -86,41 +63,6 @@ class FakeGitea:
         return self.pulls[number]
 
 
-class RecordingGit:
-    def __init__(self) -> None:
-        self.clone_calls: list[tuple[str, str, str, Path]] = []
-
-    async def clone(self, owner: str, repo: str, branch: str, destination: Path) -> str:
-        self.clone_calls.append((owner, repo, branch, destination))
-        return "base-sha"
-
-
-class RecordingPrompts:
-    def __init__(self) -> None:
-        self.calls: list[tuple[str, dict[str, str]]] = []
-
-    def render(self, name: str, **values: str) -> str:
-        self.calls.append((name, values))
-        return f"rendered {name}"
-
-
-class RecordingOpenCode:
-    def __init__(self, responses: list[BaseModel] | None = None) -> None:
-        self.responses = list(responses or [])
-        self.created_sessions: list[tuple[Path, str]] = []
-        self.resume_calls: list[dict[str, Any]] = []
-
-    async def create_session(self, workspace: Path, title: str) -> str:
-        self.created_sessions.append((workspace, title))
-        return f"{title}-session"
-
-    async def resume(self, **values: Any) -> BaseModel:
-        self.resume_calls.append(values)
-        response = self.responses.pop(0)
-        assert isinstance(response, values["result_type"])
-        return response
-
-
 def job(kind: JobKind, *, message: str | None = "Please be specific.") -> Job:
     return Job(
         id="job-plan",
@@ -192,78 +134,50 @@ def pull(*, state: str = "open", merged: bool = False) -> PullRequestInfo:
     )
 
 
-def make_services(
-    tmp_path: Path,
-    *,
-    latest: Workflow | None = None,
-    implementations: list[Workflow] | None = None,
-    pulls: dict[int, PullRequestInfo] | None = None,
-    responses: list[BaseModel] | None = None,
-) -> tuple[
-    WorkflowServices,
-    FakeRepository,
-    FakeGitea,
-    RecordingGit,
-    RecordingPrompts,
-    RecordingOpenCode,
-]:
-    repository = FakeRepository(latest=latest, implementations=implementations)
-    gitea = FakeGitea(pulls)
-    git = RecordingGit()
-    prompts = RecordingPrompts()
-    opencode = RecordingOpenCode(responses)
-    services = cast(
-        WorkflowServices,
-        SimpleNamespace(
-            settings=SimpleNamespace(
+def plan_settings(tmp_path: Path) -> SimpleNamespace:
+    return SimpleNamespace(
         workspaces_dir=tmp_path / "workspaces",
         plan_model="provider/model",
         plan_variant="high",
         plan_review_rounds=3,
-            ),
-            repository=repository,
-            gitea=gitea,
-            git=git,
-            prompts=prompts,
-            opencode=opencode,
-        ),
     )
-    return services, repository, gitea, git, prompts, opencode
 
 
 async def test_create_plan_completes_primary_and_independent_review(
     tmp_path: Path,
 ) -> None:
-    services, repository, gitea, git, prompts, opencode = make_services(
-        tmp_path,
+    repository = FakeRepository()
+    gitea = FakeGitea()
+    harness = make_workflow_harness(
+        settings=plan_settings(tmp_path),
+        repository=repository,
+        gitea=gitea,
         responses=[
             PlanArtifact(plan_markdown="# Complete plan"),
             ReviewReport(summary="Ready", findings=[]),
         ],
     )
-    run = RecordingRun()
 
-    body = await create_plan(job(JobKind.PLAN), run, services)
+    body = await create_plan(job(JobKind.PLAN), harness.run, harness.services)
 
-    created, stage = run.created_workflows[0]
+    created, stage = harness.run.created_workflows[0]
     assert stage == "planning"
     assert created.kind is WorkflowKind.PLAN
     assert created.workspace_path == tmp_path / "workspaces" / created.id / "repo"
-    assert git.clone_calls == [("org", "repo", "trunk", created.workspace_path)]
+    assert harness.git.calls == [("clone", "org", "repo", "trunk", created.workspace_path)]
     assert gitea.default_branch_calls == [("org", "repo")]
-    assert run.linked_sessions == ["plan-session"]
-    assert opencode.created_sessions == [
+    assert harness.run.linked_sessions == ["plan-session"]
+    assert harness.opencode.created_sessions == [
         (created.workspace_path, "plan"),
         (created.workspace_path, "plan-review"),
     ]
-    assert [call["result_type"] for call in opencode.resume_calls] == [
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [
         PlanArtifact,
         ReviewReport,
     ]
-    assert prompts.calls[0][0] == "plan_initial"
-    assert prompts.calls[0][1]["request"] == "Please be specific."
-    assert "Use the existing API." in prompts.calls[0][1]["context"]
-    assert "Job queued" not in prompts.calls[0][1]["context"]
+    assert harness.prompts.calls[0][0] == "plan_initial"
+    assert harness.prompts.calls[0][1]["request"] == "Please be specific."
+    assert harness.prompts.calls[0][1]["context"].startswith("Repository: org/repo\n")
 
     completed = repository.saved_workflows[-1]
     assert completed.status is WorkflowStatus.COMPLETED
@@ -281,20 +195,21 @@ async def test_discuss_plan_resumes_primary_session_without_replacing_artifact(
     tmp_path: Path,
 ) -> None:
     existing = plan_workflow()
-    services, repository, _, _, prompts, opencode = make_services(
-        tmp_path,
-        latest=existing,
+    repository = FakeRepository(latest=existing)
+    harness = make_workflow_harness(
+        settings=plan_settings(tmp_path),
+        repository=repository,
+        gitea=FakeGitea(),
         responses=[DiscussionReply(markdown="The API remains compatible.")],
     )
-    run = RecordingRun()
 
-    body = await discuss_plan(job(JobKind.DISCUSS), run, services)
+    body = await discuss_plan(job(JobKind.DISCUSS), harness.run, harness.services)
 
-    assert run.linked_workflows == [(existing.id, "discussing")]
-    assert opencode.created_sessions == []
-    assert opencode.resume_calls[0]["session_id"] == "primary-session"
-    assert opencode.resume_calls[0]["schema_name"] == "discussion.json"
-    assert prompts.calls == [
+    assert harness.run.linked_workflows == [(existing.id, "discussing")]
+    assert harness.opencode.created_sessions == []
+    assert harness.opencode.resume_calls[0]["session_id"] == "primary-session"
+    assert harness.opencode.resume_calls[0]["schema_name"] == "discussion.json"
+    assert harness.prompts.calls == [
         (
             "discuss",
             {"artifact": "Original plan", "message": "Please be specific."},
@@ -327,45 +242,57 @@ async def test_discuss_plan_rejects_missing_or_incompatible_plan(
     latest: Workflow | None,
     message: str,
 ) -> None:
-    services, _, _, _, _, opencode = make_services(tmp_path, latest=latest)
+    repository = FakeRepository(latest=latest)
+    harness = make_workflow_harness(
+        settings=plan_settings(tmp_path),
+        repository=repository,
+        gitea=FakeGitea(),
+    )
 
     with pytest.raises(JobRejected) as error:
-        await discuss_plan(job(JobKind.DISCUSS), RecordingRun(), services)
+        await discuss_plan(job(JobKind.DISCUSS), harness.run, harness.services)
 
     assert str(error.value) == message
-    assert opencode.resume_calls == []
+    assert harness.opencode.resume_calls == []
 
 
 async def test_iterate_plan_revises_then_reuses_reviewer_session(
     tmp_path: Path,
 ) -> None:
     existing = plan_workflow()
-    services, repository, gitea, _, prompts, opencode = make_services(
-        tmp_path,
+    repository = FakeRepository(
         latest=existing,
         implementations=[implementation_workflow(), implementation_workflow(None)],
-        pulls={9: pull(state="closed")},
+    )
+    gitea = FakeGitea({9: pull(state="closed")})
+    harness = make_workflow_harness(
+        settings=plan_settings(tmp_path),
+        repository=repository,
+        gitea=gitea,
         responses=[
             PlanArtifact(plan_markdown="# Revised plan"),
             ReviewReport(summary="Ready", findings=[]),
         ],
     )
-    run = RecordingRun()
 
-    body = await iterate_plan(job(JobKind.ITERATE_PLAN, message=None), run, services)
+    body = await iterate_plan(
+        job(JobKind.ITERATE_PLAN, message=None),
+        harness.run,
+        harness.services,
+    )
 
     assert gitea.pull_calls == [9]
-    assert run.linked_workflows == [(existing.id, "iterating plan")]
-    assert opencode.created_sessions == []
-    assert [call["session_id"] for call in opencode.resume_calls] == [
+    assert harness.run.linked_workflows == [(existing.id, "iterating plan")]
+    assert harness.opencode.created_sessions == []
+    assert [call["session_id"] for call in harness.opencode.resume_calls] == [
         "primary-session",
         "reviewer-session",
     ]
-    assert [call["result_type"] for call in opencode.resume_calls] == [
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [
         PlanArtifact,
         ReviewReport,
     ]
-    iterate_prompt = prompts.calls[0]
+    iterate_prompt = harness.prompts.calls[0]
     assert iterate_prompt[0] == "plan_iterate"
     assert iterate_prompt[1]["artifact"] == "Original plan"
     assert iterate_prompt[1]["message"] == ("(refine using the latest discussion and prior review)")
@@ -403,14 +330,19 @@ async def test_iterate_plan_rejects_missing_or_incompatible_plan(
     latest: Workflow | None,
     message: str,
 ) -> None:
-    services, repository, _, _, _, opencode = make_services(tmp_path, latest=latest)
+    repository = FakeRepository(latest=latest)
+    harness = make_workflow_harness(
+        settings=plan_settings(tmp_path),
+        repository=repository,
+        gitea=FakeGitea(),
+    )
 
     with pytest.raises(JobRejected) as error:
-        await iterate_plan(job(JobKind.ITERATE_PLAN), RecordingRun(), services)
+        await iterate_plan(job(JobKind.ITERATE_PLAN), harness.run, harness.services)
 
     assert str(error.value) == message
     assert repository.saved_workflows == []
-    assert opencode.resume_calls == []
+    assert harness.opencode.resume_calls == []
 
 
 @pytest.mark.parametrize(
@@ -421,19 +353,22 @@ async def test_iterate_plan_rejects_missing_or_incompatible_plan(
 async def test_iterate_plan_rejects_after_agent_pr_is_open_or_merged(
     tmp_path: Path, blocking_pull: PullRequestInfo
 ) -> None:
-    services, repository, _, _, _, opencode = make_services(
-        tmp_path,
+    repository = FakeRepository(
         latest=plan_workflow(),
         implementations=[implementation_workflow()],
-        pulls={9: blocking_pull},
+    )
+    harness = make_workflow_harness(
+        settings=plan_settings(tmp_path),
+        repository=repository,
+        gitea=FakeGitea({9: blocking_pull}),
     )
 
     with pytest.raises(JobRejected) as error:
-        await iterate_plan(job(JobKind.ITERATE_PLAN), RecordingRun(), services)
+        await iterate_plan(job(JobKind.ITERATE_PLAN), harness.run, harness.services)
 
     assert str(error.value) == (
         "Issue plan iteration is disabled because agent PR #9 is open or merged. "
         "Iterate an open implementation on its PR."
     )
     assert repository.latest_calls == 0
-    assert opencode.resume_calls == []
+    assert harness.opencode.resume_calls == []

tests/test_workflow_pull_request.py

@@ -1,34 +1,18 @@
 import json
+from collections.abc import Iterable
 from pathlib import Path
 from types import SimpleNamespace
-from typing import Any, cast
+from typing import Any
 
 import pytest
 from pydantic import BaseModel
 
 from agentci.engine.model import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus
-from agentci.engine.run import JobRun
 from agentci.integrations.gitea.models import CommentInfo, IssueInfo, PullRequestInfo
 from agentci.workflows.model import AgentResult, ReviewReport
 from agentci.workflows.pull_request import fix_pull_request, iterate_implementation
 from agentci.workflows.render import JobRejected
-from agentci.workflows.services import WorkflowServices
-
-
-class RecordingRun(JobRun):
-    def __init__(self) -> None:
-        self.stages: list[str] = []
-        self.linked_workflows: list[tuple[str, str]] = []
-        self.linked_sessions: list[str] = []
-
-    async def stage(self, stage: str) -> None:
-        self.stages.append(stage)
-
-    async def link_workflow(self, workflow_id: str, stage: str) -> None:
-        self.linked_workflows.append((workflow_id, stage))
-
-    async def link_session(self, session_id: str) -> None:
-        self.linked_sessions.append(session_id)
+from tests.workflow_support import WorkflowHarness, make_workflow_harness
 
 
 class FakeRepository:
@@ -88,70 +72,6 @@ class FakeGitea:
         return IssueInfo(7, "Fix widget", "The widget is broken.", "open")
 
 
-class RecordingGit:
-    def __init__(self, *, changed: bool = True) -> None:
-        self.changed = changed
-        self.calls: list[tuple[Any, ...]] = []
-
-    async def clone(self, owner: str, repo: str, branch: str, destination: Path) -> str:
-        self.calls.append(("clone", owner, repo, branch, destination))
-        return "head-sha"
-
-    async def sync_branch(self, workspace: Path, branch: str) -> str:
-        self.calls.append(("sync_branch", workspace, branch))
-        return "head-sha"
-
-    async def has_changes(self, workspace: Path) -> bool:
-        self.calls.append(("has_changes", workspace))
-        return self.changed
-
-    async def diff_check(self, workspace: Path) -> None:
-        self.calls.append(("diff_check", workspace))
-
-    async def commit(self, workspace: Path, message: str) -> str:
-        self.calls.append(("commit", workspace, message))
-        return "new-sha"
-
-    async def push(self, workspace: Path, branch: str, *, set_upstream: bool = False) -> None:
-        self.calls.append(("push", workspace, branch, set_upstream))
-
-
-class RecordingDevelopment:
-    description = "Python 3.13"
-
-    def __init__(self) -> None:
-        self.workspaces: list[Path] = []
-
-    async def prepare(self, workspace: Path) -> None:
-        self.workspaces.append(workspace)
-
-
-class RecordingPrompts:
-    def __init__(self) -> None:
-        self.calls: list[tuple[str, dict[str, str]]] = []
-
-    def render(self, name: str, **values: str) -> str:
-        self.calls.append((name, values))
-        return f"rendered {name}"
-
-
-class RecordingOpenCode:
-    def __init__(self, responses: list[BaseModel] | None = None) -> None:
-        self.responses = list(responses or [])
-        self.created_sessions: list[tuple[Path, str]] = []
-        self.resume_calls: list[dict[str, Any]] = []
-
-    async def create_session(self, workspace: Path, title: str) -> str:
-        self.created_sessions.append((workspace, title))
-        return f"{title}-session"
-
-    async def resume(self, **values: Any) -> BaseModel:
-        self.resume_calls.append(values)
-        response = self.responses.pop(0)
-        assert isinstance(response, values["result_type"])
-        return response
-
-
 def job(kind: JobKind, *, pr_number: int | None = 12) -> Job:
     return Job(
         id="job-pr",
@@ -231,32 +151,15 @@ def pull(
     )
 
 
-def make_services(
+def pull_request_harness(
     tmp_path: Path,
+    repository: FakeRepository,
+    gitea: FakeGitea,
     *,
-    workflow: Workflow | None = None,
-    pull_info: PullRequestInfo | None = None,
-    plan: Workflow | None = None,
     changed: bool = True,
-    responses: list[BaseModel] | None = None,
-) -> tuple[
-    WorkflowServices,
-    FakeRepository,
-    FakeGitea,
-    RecordingGit,
-    RecordingDevelopment,
-    RecordingPrompts,
-    RecordingOpenCode,
-]:
-    repository = FakeRepository(workflow=workflow, plan=plan)
-    gitea = FakeGitea(pull_info or pull())
-    git = RecordingGit(changed=changed)
-    development = RecordingDevelopment()
-    prompts = RecordingPrompts()
-    opencode = RecordingOpenCode(responses)
-    services = cast(
-        WorkflowServices,
-        SimpleNamespace(
+    responses: Iterable[BaseModel] = (),
+) -> WorkflowHarness:
+    return make_workflow_harness(
         settings=SimpleNamespace(
             workspaces_dir=tmp_path / "workspaces",
             implement_model="provider/model",
@@ -264,13 +167,12 @@ def make_services(
         ),
         repository=repository,
         gitea=gitea,
-            git=git,
-            development=development,
-            prompts=prompts,
-            opencode=opencode,
-        ),
+        changed=changed,
+        responses=responses,
+        clone_sha="head-sha",
+        sync_sha="head-sha",
+        commit_sha="new-sha",
     )
-    return services, repository, gitea, git, development, prompts, opencode
 
 
 def result(summary: str = "# Refine widget") -> AgentResult:
@@ -281,37 +183,52 @@ async def test_iterate_implementation_reuses_sessions_reviews_commits_and_pushes
     tmp_path: Path,
 ) -> None:
     existing = implementation_workflow()
-    services, repository, _, git, development, prompts, opencode = make_services(
+    repository = FakeRepository(workflow=existing, plan=plan_workflow())
+    gitea = FakeGitea(pull())
+    harness = pull_request_harness(
         tmp_path,
-        workflow=existing,
-        plan=plan_workflow(),
+        repository,
+        gitea,
         responses=[result(), ReviewReport(summary="Ready", findings=[])],
     )
-    run = RecordingRun()
 
-    body = await iterate_implementation(job(JobKind.ITERATE_IMPLEMENT), run, services)
+    body = await iterate_implementation(
+        job(JobKind.ITERATE_IMPLEMENT),
+        harness.run,
+        harness.services,
+    )
 
-    assert run.linked_workflows == [(existing.id, "synchronizing branch")]
-    assert development.workspaces == [existing.workspace_path]
-    assert opencode.created_sessions == []
-    assert [call["session_id"] for call in opencode.resume_calls] == [
+    assert harness.run.linked_workflows == [(existing.id, "synchronizing branch")]
+    assert harness.development.workspaces == [existing.workspace_path]
+    assert harness.opencode.created_sessions == []
+    assert [call["session_id"] for call in harness.opencode.resume_calls] == [
         "primary-session",
         "reviewer-session",
     ]
-    assert [call["result_type"] for call in opencode.resume_calls] == [
+    assert [call["result_type"] for call in harness.opencode.resume_calls] == [
         AgentResult,
         ReviewReport,
     ]
-    iterate_prompt = prompts.calls[0]
+    iterate_prompt = harness.prompts.calls[0]
     assert iterate_prompt[0] == "implementation_iterate"
     assert iterate_prompt[1]["message"] == "Handle the review."
-    assert "Handle empty input." in iterate_prompt[1]["context"]
-    review_prompt = prompts.calls[1]
+    review_prompt = harness.prompts.calls[1]
     assert review_prompt[0] == "implementation_review"
+    assert set(review_prompt[1]) == {"issue_context", "artifact", "pull_context"}
+    assert review_prompt[1]["issue_context"].startswith("Repository: org/repo\n")
     assert review_prompt[1]["artifact"] == "Canonical plan"
-    assert "The widget is broken." in review_prompt[1]["issue_context"]
+    assert review_prompt[1]["pull_context"] == iterate_prompt[1]["context"]
+    assert harness.opencode.resume_calls[1] == {
+        "session_id": "reviewer-session",
+        "prompt": "rendered implementation_review",
+        "model": "provider/model",
+        "variant": "high",
+        "workspace": existing.workspace_path,
+        "schema_name": "review.json",
+        "result_type": ReviewReport,
+    }
 
-    assert git.calls == [
+    assert harness.git.calls == [
         ("sync_branch", existing.workspace_path, "agent/issue-7"),
         ("has_changes", existing.workspace_path),
         ("diff_check", existing.workspace_path),
@@ -368,23 +285,23 @@ async def test_iterate_implementation_rejects_missing_stale_or_incompatible_work
     existing: Workflow | None,
     message: str,
 ) -> None:
-    services, repository, gitea, git, development, _, opencode = make_services(
-        tmp_path, workflow=existing
-    )
+    repository = FakeRepository(workflow=existing)
+    gitea = FakeGitea(pull())
+    harness = pull_request_harness(tmp_path, repository, gitea)
 
     with pytest.raises(JobRejected) as error:
         await iterate_implementation(
             job(JobKind.ITERATE_IMPLEMENT, pr_number=pr_number),
-            RecordingRun(),
-            services,
+            harness.run,
+            harness.services,
         )
 
     assert str(error.value) == message
     assert repository.saved_workflows == []
     assert gitea.pull_calls == []
-    assert git.calls == []
-    assert development.workspaces == []
-    assert opencode.resume_calls == []
+    assert harness.git.calls == []
+    assert harness.development.workspaces == []
+    assert harness.opencode.resume_calls == []
 
 
 @pytest.mark.parametrize(
@@ -409,46 +326,51 @@ async def test_iterate_implementation_rejects_closed_or_stale_branch_before_chec
     existing: Workflow,
     message: str,
 ) -> None:
-    services, repository, _, git, development, _, opencode = make_services(
-        tmp_path, workflow=existing, pull_info=pull_info
-    )
+    repository = FakeRepository(workflow=existing)
+    harness = pull_request_harness(tmp_path, repository, FakeGitea(pull_info))
 
     with pytest.raises(JobRejected) as error:
-        await iterate_implementation(job(JobKind.ITERATE_IMPLEMENT), RecordingRun(), services)
+        await iterate_implementation(
+            job(JobKind.ITERATE_IMPLEMENT),
+            harness.run,
+            harness.services,
+        )
 
     assert str(error.value) == message
     assert repository.saved_workflows == []
-    assert git.calls == []
-    assert development.workspaces == []
-    assert opencode.resume_calls == []
+    assert harness.git.calls == []
+    assert harness.development.workspaces == []
+    assert harness.opencode.resume_calls == []
 
 
 async def test_fix_pull_request_clones_head_runs_agent_commits_and_pushes(
     tmp_path: Path,
 ) -> None:
-    services, repository, _, git, development, prompts, opencode = make_services(
+    repository = FakeRepository()
+    harness = pull_request_harness(
         tmp_path,
+        repository,
+        FakeGitea(pull()),
         responses=[result("Fix empty input")],
     )
-    run = RecordingRun()
 
-    body = await fix_pull_request(job(JobKind.FIX), run, services)
+    body = await fix_pull_request(job(JobKind.FIX), harness.run, harness.services)
 
     workspace = tmp_path / "workspaces" / "fix-job-pr" / "repo"
-    assert git.calls == [
+    assert harness.git.calls == [
         ("clone", "contributor", "fork", "agent/issue-7", workspace),
         ("has_changes", workspace),
         ("diff_check", workspace),
         ("commit", workspace, "agent fix: Fix empty input"),
         ("push", workspace, "agent/issue-7", False),
     ]
-    assert development.workspaces == [workspace]
-    assert opencode.created_sessions == [(workspace, "fix")]
-    assert run.linked_sessions == ["fix-session"]
-    assert opencode.resume_calls[0]["session_id"] == "fix-session"
-    assert opencode.resume_calls[0]["result_type"] is AgentResult
-    assert prompts.calls[0][0] == "fix"
-    assert "`src/widget.py:8`: Add a guard." in prompts.calls[0][1]["context"]
+    assert harness.development.workspaces == [workspace]
+    assert harness.opencode.created_sessions == [(workspace, "fix")]
+    assert harness.run.linked_sessions == ["fix-session"]
+    assert harness.opencode.resume_calls[0]["session_id"] == "fix-session"
+    assert harness.opencode.resume_calls[0]["result_type"] is AgentResult
+    assert harness.prompts.calls[0][0] == "fix"
+    assert "`src/widget.py:8`: Add a guard." in harness.prompts.calls[0][1]["context"]
     assert repository.saved_workflows == []
     assert body == (
         "<!-- agentci:fix workflow=job-pr -->\n"
@@ -457,32 +379,13 @@ async def test_fix_pull_request_clones_head_runs_agent_commits_and_pushes(
     )
 
 
-async def test_fix_pull_request_rejects_clean_worktree_before_push(
-    tmp_path: Path,
-) -> None:
-    services, _, _, git, _, _, _ = make_services(
-        tmp_path,
-        changed=False,
-        responses=[result("No changes required")],
-    )
-
-    with pytest.raises(
-        JobRejected,
-        match="OpenCode completed without producing any file changes",
-    ):
-        await fix_pull_request(job(JobKind.FIX), RecordingRun(), services)
-
-    assert [call[0] for call in git.calls] == ["clone", "has_changes"]
-
-
 @pytest.mark.parametrize(
     ("pr_number", "pull_info", "message"),
     [
         (None, pull(), "This command requires a pull request."),
         (12, pull(state="closed"), "Fixes require an open pull request."),
-        (12, pull(state="closed", merged=True), "Fixes require an open pull request."),
     ],
-    ids=["missing-pr", "closed", "merged"],
+    ids=["missing-pr", "closed"],
 )
 async def test_fix_pull_request_requires_open_pr_before_clone(
     tmp_path: Path,
@@ -490,13 +393,18 @@ async def test_fix_pull_request_requires_open_pr_before_clone(
     pull_info: PullRequestInfo,
     message: str,
 ) -> None:
-    services, _, gitea, git, development, _, opencode = make_services(tmp_path, pull_info=pull_info)
+    gitea = FakeGitea(pull_info)
+    harness = pull_request_harness(tmp_path, FakeRepository(), gitea)
 
     with pytest.raises(JobRejected) as error:
-        await fix_pull_request(job(JobKind.FIX, pr_number=pr_number), RecordingRun(), services)
+        await fix_pull_request(
+            job(JobKind.FIX, pr_number=pr_number),
+            harness.run,
+            harness.services,
+        )
 
     assert str(error.value) == message
-    assert git.calls == []
-    assert development.workspaces == []
-    assert opencode.resume_calls == []
+    assert harness.git.calls == []
+    assert harness.development.workspaces == []
+    assert harness.opencode.resume_calls == []
     assert gitea.pull_calls == ([] if pr_number is None else [("org", "repo", 12)])

tests/test_workflow_render.py

@@ -59,32 +59,53 @@ def test_agent_and_final_comments_preserve_protocol_marker() -> None:
     )
 
 
-def test_pull_request_body_and_result_comment_render_validation() -> None:
-    result = AgentResult(
+@pytest.mark.parametrize(
+    ("issue_number", "result", "sha", "expected_body", "expected_comment"),
+    [
+        pytest.param(
+            17,
+            AgentResult(
                 summary_markdown="Implemented the widget fix.",
                 tests=["pytest: passed", "ruff: passed"],
-    )
-
-    assert pull_request_body(17, result) == (
+            ),
+            "abc123",
+            (
                 "Closes #17\n\n"
                 "## Implementation\n\nImplemented the widget fix.\n\n"
                 "## Validation\n\n- pytest: passed\n- ruff: passed\n\n"
                 "_Created by Agent CI._"
-    )
-    assert result_comment(result, sha="abc123") == (
+            ),
+            (
                 "## Agent result\n\nImplemented the widget fix.\n\n"
                 "## Validation\n\n- pytest: passed\n- ruff: passed\n\n"
                 "Commit: `abc123`"
+            ),
+            id="reported-validation",
+        ),
+        pytest.param(
+            2,
+            AgentResult(summary_markdown="Applied the change.", tests=[]),
+            None,
+            (
+                "Closes #2\n\n"
+                "## Implementation\n\nApplied the change.\n\n"
+                "## Validation\n\n- Not reported\n\n"
+                "_Created by Agent CI._"
+            ),
+            "## Agent result\n\nApplied the change.\n\n## Validation\n\n- Not reported",
+            id="missing-validation",
+        ),
+    ],
 )
-
-
-def test_renderers_report_missing_validation() -> None:
-    result = AgentResult(summary_markdown="Applied the change.", tests=[])
-
-    assert "## Validation\n\n- Not reported" in pull_request_body(2, result)
-    assert result_comment(result) == (
-        "## Agent result\n\nApplied the change.\n\n## Validation\n\n- Not reported"
-    )
+def test_result_renderers_render_reported_and_missing_validation(
+    issue_number: int,
+    result: AgentResult,
+    sha: str | None,
+    expected_body: str,
+    expected_comment: str,
+) -> None:
+    assert pull_request_body(issue_number, result) == expected_body
+    assert result_comment(result, sha=sha) == expected_comment
 
 
 @pytest.mark.parametrize(

tests/workflow_support.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+
+from pydantic import BaseModel
+
+from agentci.engine.model import Workflow
+from agentci.engine.run import JobRun
+from agentci.workflows.services import WorkflowServices
+
+
+class RecordingRun(JobRun):
+    def __init__(self) -> None:
+        self.stages: list[str] = []
+        self.created_workflows: list[tuple[Workflow, str]] = []
+        self.linked_workflows: list[tuple[str, str]] = []
+        self.linked_sessions: list[str] = []
+
+    async def stage(self, stage: str) -> None:
+        self.stages.append(stage)
+
+    async def create_workflow(self, workflow: Workflow, stage: str) -> None:
+        self.created_workflows.append((workflow, stage))
+
+    async def link_workflow(self, workflow_id: str, stage: str) -> None:
+        self.linked_workflows.append((workflow_id, stage))
+
+    async def link_session(self, session_id: str) -> None:
+        self.linked_sessions.append(session_id)
+
+
+class ScriptedOpenCode:
+    def __init__(
+        self,
+        responses: Iterable[BaseModel] = (),
+        trace: list[tuple[object, ...]] | None = None,
+    ) -> None:
+        self.responses = list(responses)
+        self.created_sessions: list[tuple[Path, str]] = []
+        self.resume_calls: list[dict[str, Any]] = []
+        self.trace = trace if trace is not None else []
+
+    async def create_session(self, workspace: Path, title: str) -> str:
+        session_id = f"{title}-session"
+        self.created_sessions.append((workspace, title))
+        self.trace.append(("create_session", title, session_id))
+        return session_id
+
+    async def resume(self, **values: Any) -> BaseModel:
+        self.resume_calls.append(values)
+        self.trace.append(("resume", values["session_id"], values["result_type"]))
+        assert self.responses, "unexpected OpenCode resume call"
+        response = self.responses.pop(0)
+        assert isinstance(response, values["result_type"]), (
+            f"expected {values['result_type'].__name__}, got {type(response).__name__}"
+        )
+        return response
+
+
+class RecordingPrompts:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, dict[str, str]]] = []
+
+    def render(self, name: str, **values: str) -> str:
+        self.calls.append((name, values))
+        return f"rendered {name}"
+
+
+class RecordingDevelopment:
+    description = "Python 3.13"
+
+    def __init__(self) -> None:
+        self.workspaces: list[Path] = []
+
+    async def prepare(self, workspace: Path) -> None:
+        self.workspaces.append(workspace)
+
+
+class RecordingGit:
+    def __init__(
+        self,
+        *,
+        changed: bool = True,
+        clone_sha: str = "base-sha",
+        sync_sha: str = "head-sha",
+        commit_sha: str = "commit-sha",
+    ) -> None:
+        self.changed = changed
+        self.clone_sha = clone_sha
+        self.sync_sha = sync_sha
+        self.commit_sha = commit_sha
+        self.calls: list[tuple[object, ...]] = []
+
+    async def clone(self, owner: str, repo: str, branch: str, destination: Path) -> str:
+        self.calls.append(("clone", owner, repo, branch, destination))
+        return self.clone_sha
+
+    async def create_branch(self, workspace: Path, branch: str) -> None:
+        self.calls.append(("create_branch", workspace, branch))
+
+    async def sync_branch(self, workspace: Path, branch: str) -> str:
+        self.calls.append(("sync_branch", workspace, branch))
+        return self.sync_sha
+
+    async def has_changes(self, workspace: Path) -> bool:
+        self.calls.append(("has_changes", workspace))
+        return self.changed
+
+    async def diff_check(self, workspace: Path) -> None:
+        self.calls.append(("diff_check", workspace))
+
+    async def commit(self, workspace: Path, message: str) -> str:
+        self.calls.append(("commit", workspace, message))
+        return self.commit_sha
+
+    async def push(self, workspace: Path, branch: str, *, set_upstream: bool = False) -> None:
+        self.calls.append(("push", workspace, branch, set_upstream))
+
+
+@dataclass(frozen=True)
+class WorkflowHarness:
+    services: WorkflowServices
+    run: RecordingRun
+    git: RecordingGit
+    development: RecordingDevelopment
+    prompts: RecordingPrompts
+    opencode: ScriptedOpenCode
+    trace: list[tuple[object, ...]]
+
+
+def make_workflow_harness(
+    *,
+    settings: SimpleNamespace,
+    repository: object,
+    gitea: object,
+    responses: Iterable[BaseModel] = (),
+    trace: list[tuple[object, ...]] | None = None,
+    changed: bool = True,
+    clone_sha: str = "base-sha",
+    sync_sha: str = "head-sha",
+    commit_sha: str = "commit-sha",
+) -> WorkflowHarness:
+    shared_trace = trace if trace is not None else []
+    run = RecordingRun()
+    git = RecordingGit(
+        changed=changed,
+        clone_sha=clone_sha,
+        sync_sha=sync_sha,
+        commit_sha=commit_sha,
+    )
+    development = RecordingDevelopment()
+    prompts = RecordingPrompts()
+    opencode = ScriptedOpenCode(responses, shared_trace)
+    services = cast(
+        WorkflowServices,
+        SimpleNamespace(
+            settings=settings,
+            repository=repository,
+            gitea=gitea,
+            git=git,
+            development=development,
+            prompts=prompts,
+            opencode=opencode,
+        ),
+    )
+    return WorkflowHarness(
+        services=services,
+        run=run,
+        git=git,
+        development=development,
+        prompts=prompts,
+        opencode=opencode,
+        trace=shared_trace,
+    )