make concurrent

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

.env.example

@@ -18,6 +18,7 @@ AGENTCI_CONTEXT7_API_KEY=
 AGENTCI_PLAN_REVIEW_ROUNDS=4
 AGENTCI_IMPLEMENT_REVIEW_ROUNDS=3
 AGENTCI_TURN_TIMEOUT_SECONDS=3600
+AGENTCI_MAX_CONCURRENT_JOBS=2
 # Comma-delimited built-in or custom script names, for example: python,dotnet,company-tools
 AGENTCI_INSTALL_SCRIPTS=
 AGENTCI_INSTALL_SCRIPT_TIMEOUT_SECONDS=900

README.md

@@ -16,9 +16,11 @@ private OpenCode server on the same Docker network as Gitea.
 | PR | `/agent fix [message]` | Start a fresh one-shot fix session and push one commit. |
 
 The requester must have Gitea `write`, `admin`, or `owner` permission on the repository. Commands
-are durably sequenced when their webhook arrives, then authorized and executed in that receive
+are durably sequenced when their webhook arrives, then authorized and executed in receive order per
-order. Each command gets one Gitea comment, which is reconciled asynchronously through queued,
+issue or pull request. Up to `AGENTCI_MAX_CONCURRENT_JOBS` unrelated targets execute concurrently;
-running, and terminal states. Deleted comments are rediscovered by their hidden marker or recreated.
+the default is two. Each command gets one Gitea comment, which is reconciled asynchronously through
+queued, running, and terminal states. Deleted comments are rediscovered by their hidden marker or
+recreated.
 
 ## Deploy
 
@@ -131,7 +133,9 @@ sessions. Tea's Gitea token configuration is regenerated in an ephemeral tmpfs a
 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
 than reducer state. Control effects retry with bounded backoff. A delayed authorization blocks later
-workflow execution but not later control work.
+workflow execution for the same issue or pull request, but not other targets or later control work.
+Development environment installers remain serialized because they share the persistent tools
+directory.
 
 The OpenCode migration tags existing workflows as Codex-owned and preserves their session IDs for
 rollback, but OpenCode refuses to resume them. Follow-up commands against those workflows ask for a

compose.yaml

@@ -29,6 +29,7 @@ services:
       AGENTCI_PLAN_REVIEW_ROUNDS: ${AGENTCI_PLAN_REVIEW_ROUNDS:-4}
       AGENTCI_IMPLEMENT_REVIEW_ROUNDS: ${AGENTCI_IMPLEMENT_REVIEW_ROUNDS:-3}
       AGENTCI_TURN_TIMEOUT_SECONDS: ${AGENTCI_TURN_TIMEOUT_SECONDS:-3600}
+      AGENTCI_MAX_CONCURRENT_JOBS: ${AGENTCI_MAX_CONCURRENT_JOBS:-2}
       AGENTCI_INSTALL_SCRIPT_TIMEOUT_SECONDS: ${AGENTCI_INSTALL_SCRIPT_TIMEOUT_SECONDS:-900}
       AGENTCI_INSTALL_SCRIPTS: ${AGENTCI_INSTALL_SCRIPTS:-}
       AGENTCI_PYTHON_VERSION: ${AGENTCI_PYTHON_VERSION:-3.13}

src/agentci/adapters/development.py

@@ -32,6 +32,7 @@ class DevelopmentEnvironment:
         self.timeout_seconds = timeout_seconds
         self.python_version = python_version
         self.dotnet_channel = dotnet_channel
+        self._prepare_lock = asyncio.Lock()
 
     @property
     def description(self) -> str:
@@ -40,6 +41,7 @@ class DevelopmentEnvironment:
     async def prepare(self, workspace: Path) -> None:
         if not self.scripts:
             return
+        async with self._prepare_lock:
             self.tools_dir.mkdir(parents=True, exist_ok=True)
             (self.tools_dir / "bin").mkdir(exist_ok=True)
             for name in self.scripts:

src/agentci/adapters/job_store.py

@@ -186,6 +186,7 @@ def _claim_task(connection: sqlite3.Connection, queue: str) -> ListenerTask | No
     if queue == "jobs":
         fifo = """AND NOT EXISTS (
             SELECT 1 FROM jobs earlier WHERE earlier.receive_sequence < j.receive_sequence
+            AND earlier.target_key = j.target_key
             AND earlier.status IN ('received', 'queued', 'running'))"""
     row = connection.execute(
         f"""SELECT t.* FROM listener_tasks t JOIN jobs j ON j.id=t.job_id

src/agentci/config.py

@@ -45,6 +45,7 @@ class Settings(BaseSettings):
     turn_timeout_seconds: int = Field(default=3600, ge=60)
     install_script_timeout_seconds: int = Field(default=900, ge=1)
     worker_poll_seconds: float = Field(default=1.0, ge=0.1)
+    max_concurrent_jobs: int = Field(default=2, ge=1, le=32)
     install_scripts: Annotated[list[str], NoDecode] = Field(default_factory=list)
     install_scripts_dir: Path = Path("/etc/agentci/install-scripts")
     python_version: str = "3.13"

src/agentci/container.py

@@ -95,6 +95,7 @@ async def build_container(settings: Settings) -> Container:
         opencode=opencode,
         dispatcher=dispatcher,
         poll_seconds=settings.worker_poll_seconds,
+        max_concurrent_jobs=settings.max_concurrent_jobs,
         workspaces_dir=settings.workspaces_dir,
         bot_username=settings.bot_username,
     )

src/agentci/worker.py

@@ -41,6 +41,7 @@ class Worker:
         opencode: OpenCodeClient,
         dispatcher: Dispatcher,
         poll_seconds: float,
+        max_concurrent_jobs: int,
         workspaces_dir: Path,
         bot_username: str,
     ) -> None:
@@ -50,12 +51,16 @@ class Worker:
         self.opencode = opencode
         self.dispatcher = dispatcher
         self.poll_seconds = poll_seconds
+        self.max_concurrent_jobs = max_concurrent_jobs
         self.workspaces_dir = workspaces_dir
         self.bot_username = bot_username
 
     async def run(self, stop: asyncio.Event) -> None:
         await self._recover()
-        await asyncio.gather(self._loop("control", stop), self._loop("jobs", stop))
+        await asyncio.gather(
+            self._loop("control", stop),
+            *(self._loop("jobs", stop) for _ in range(self.max_concurrent_jobs)),
+        )
 
     async def _loop(self, queue: str, stop: asyncio.Event) -> None:
         while not stop.is_set():

tests/test_config.py

@@ -39,6 +39,17 @@ def test_defaults_explore_agent_to_luna_low() -> None:
     assert settings.explore_variant == "low"
 
 
+def test_defaults_to_two_concurrent_jobs() -> None:
+    settings = Settings(_env_file=None)  # type: ignore[call-arg]
+    assert settings.max_concurrent_jobs == 2
+
+
+@pytest.mark.parametrize("value", [0, 33])
+def test_rejects_unsafe_job_concurrency(value: int) -> None:
+    with pytest.raises(ValidationError):
+        Settings(_env_file=None, max_concurrent_jobs=value)  # type: ignore[call-arg]
+
+
 def test_reads_comma_delimited_install_scripts_from_environment(monkeypatch) -> None:
     monkeypatch.setenv("AGENTCI_INSTALL_SCRIPTS", "python,dotnet")
 

tests/test_development.py

@@ -1,3 +1,4 @@
+import asyncio
 from pathlib import Path
 
 import pytest
@@ -74,6 +75,33 @@ async def test_runs_non_executable_shell_script_from_bind_mount(tmp_path) -> Non
     assert (workspace / "selected").read_text() == "mounted\n"
 
 
+async def test_serializes_concurrent_preparation(tmp_path, monkeypatch) -> None:
+    development = environment(tmp_path, ["shared"])
+    script(development.scripts_dir / "shared", "true")
+    started = asyncio.Event()
+    release = asyncio.Event()
+    active = 0
+    maximum_active = 0
+
+    async def run(*_args) -> None:
+        nonlocal active, maximum_active
+        active += 1
+        maximum_active = max(maximum_active, active)
+        started.set()
+        await release.wait()
+        active -= 1
+
+    monkeypatch.setattr(development, "_run", run)
+    first = asyncio.create_task(development.prepare(tmp_path / "first"))
+    await started.wait()
+    second = asyncio.create_task(development.prepare(tmp_path / "second"))
+    await asyncio.sleep(0)
+    release.set()
+    await asyncio.gather(first, second)
+
+    assert maximum_active == 1
+
+
 async def test_reports_script_failure_output(tmp_path) -> None:
     workspace = tmp_path / "workspace"
     workspace.mkdir()

tests/test_storage.py

@@ -22,14 +22,20 @@ async def storage(tmp_path: Path) -> Storage:
     return value
 
 
-def command(delivery: str, body: str = "/agent plan") -> CommandEvent:
+def command(
+    delivery: str,
+    body: str = "/agent plan",
+    *,
+    issue: int = 3,
+    pr: int | None = None,
+) -> CommandEvent:
     return CommandEvent(
         delivery_id=delivery,
         comment_id=int(delivery.rsplit("-", 1)[-1]),
         repo_owner="alice",
         repo_name="repo",
-        issue_number=3,
+        issue_number=issue,
-        pr_number=None,
+        pr_number=pr,
         requester="alice",
         body=body,
     )
@@ -47,7 +53,9 @@ async def test_receive_is_idempotent_without_consuming_sequence(storage: Storage
     assert second.state.receive_sequence == first.state.receive_sequence + 1
 
 
-async def test_received_job_blocks_later_execute_task(storage: Storage) -> None:
+async def test_received_job_blocks_later_execute_task_for_same_target(
+    storage: Storage,
+) -> None:
     host = StateMachine(storage)
     first = (await host.receive(command("delivery-1"))).state
     second = (await host.receive(command("delivery-2"))).state
@@ -61,6 +69,31 @@ async def test_received_job_blocks_later_execute_task(storage: Storage) -> None:
     assert task.job_id == second.id
 
 
+async def test_received_job_does_not_block_a_different_target(storage: Storage) -> None:
+    host = StateMachine(storage)
+    await host.receive(command("delivery-1", issue=3))
+    second = (await host.receive(command("delivery-2", issue=4))).state
+    await host.evolve("grant-2", PermissionGranted(job_id=second.id))
+
+    task = await storage.claim_task("jobs")
+
+    assert task is not None
+    assert task.job_id == second.id
+
+
+async def test_claims_multiple_eligible_targets_without_duplicates(storage: Storage) -> None:
+    host = StateMachine(storage)
+    first = (await host.receive(command("delivery-1", issue=3))).state
+    second = (await host.receive(command("delivery-2", issue=4))).state
+    await host.evolve("grant-1", PermissionGranted(job_id=first.id))
+    await host.evolve("grant-2", PermissionGranted(job_id=second.id))
+
+    claimed = [await storage.claim_task("jobs"), await storage.claim_task("jobs")]
+
+    assert [task.job_id for task in claimed if task is not None] == [first.id, second.id]
+    assert await storage.claim_task("jobs") is None
+
+
 async def test_started_and_finished_timestamps_are_owned_by_store(storage: Storage) -> None:
     host = StateMachine(storage)
     state = (await host.receive(command("delivery-1"))).state

tests/test_worker.py

@@ -1,3 +1,4 @@
+import asyncio
 from pathlib import Path
 from types import SimpleNamespace
 from typing import cast
@@ -31,6 +32,7 @@ def worker(tmp_path: Path, storage: FakeStorage, opencode: FakeOpenCode) -> Work
         opencode=opencode,  # type: ignore[arg-type]
         dispatcher=SimpleNamespace(),  # type: ignore[arg-type]
         poll_seconds=1,
+        max_concurrent_jobs=2,
         workspaces_dir=tmp_path,
         bot_username="agentci",
     )
@@ -66,6 +68,25 @@ async def test_abort_uses_one_shot_fix_workspace(tmp_path: Path) -> None:
     assert opencode.aborted == {("session", tmp_path / "fix-job" / "repo")}
 
 
+async def test_run_starts_configured_job_consumers(tmp_path: Path, monkeypatch) -> None:
+    value = worker(tmp_path, FakeStorage(), FakeOpenCode())
+    queues = []
+
+    async def recover() -> None:
+        pass
+
+    async def loop(queue, _stop) -> None:
+        queues.append(queue)
+
+    monkeypatch.setattr(value, "_recover", recover)
+    monkeypatch.setattr(value, "_loop", loop)
+
+    await value.run(asyncio.Event())
+
+    assert queues.count("control") == 1
+    assert queues.count("jobs") == 2
+
+
 def test_safe_error_is_single_line_and_bounded() -> None:
     value = _safe_error(RuntimeError("bad\n" + "x" * 2000))
     assert "\n" not in value