make concurrent

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,
-        pr_number=None,
+        issue_number=issue,
+        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