From 7527831af6f0ed73d60f7aa5da5b24525dad8c39 Mon Sep 17 00:00:00 2001 From: StanPonomarev Date: Wed, 22 Jul 2026 17:37:14 +0200 Subject: [PATCH] make concurrent --- .env.example | 1 + README.md | 12 ++++++--- compose.yaml | 1 + src/agentci/adapters/development.py | 10 ++++--- src/agentci/adapters/job_store.py | 1 + src/agentci/config.py | 1 + src/agentci/container.py | 1 + src/agentci/worker.py | 7 ++++- tests/test_config.py | 11 ++++++++ tests/test_development.py | 28 ++++++++++++++++++++ tests/test_storage.py | 41 ++++++++++++++++++++++++++--- tests/test_worker.py | 21 +++++++++++++++ 12 files changed, 122 insertions(+), 13 deletions(-) diff --git a/.env.example b/.env.example index 66c8692..88c6b67 100644 --- a/.env.example +++ b/.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 diff --git a/README.md b/README.md index ebfadf8..babdeed 100644 --- a/README.md +++ b/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 -order. 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. +are durably sequenced when their webhook arrives, then authorized and executed in receive order per +issue or pull request. Up to `AGENTCI_MAX_CONCURRENT_JOBS` unrelated targets execute concurrently; +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 diff --git a/compose.yaml b/compose.yaml index e45682a..2762d9c 100644 --- a/compose.yaml +++ b/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} diff --git a/src/agentci/adapters/development.py b/src/agentci/adapters/development.py index d30faec..1c51b6e 100644 --- a/src/agentci/adapters/development.py +++ b/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,10 +41,11 @@ class DevelopmentEnvironment: async def prepare(self, workspace: Path) -> None: if not self.scripts: return - self.tools_dir.mkdir(parents=True, exist_ok=True) - (self.tools_dir / "bin").mkdir(exist_ok=True) - for name in self.scripts: - await self._run(name, self._resolve(name), workspace) + 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: + await self._run(name, self._resolve(name), workspace) def _resolve(self, name: str) -> Path: path = self.scripts_dir / name diff --git a/src/agentci/adapters/job_store.py b/src/agentci/adapters/job_store.py index 77d9e4e..8f38e26 100644 --- a/src/agentci/adapters/job_store.py +++ b/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 diff --git a/src/agentci/config.py b/src/agentci/config.py index f982343..f324599 100644 --- a/src/agentci/config.py +++ b/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" diff --git a/src/agentci/container.py b/src/agentci/container.py index 8076c22..09f502d 100644 --- a/src/agentci/container.py +++ b/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, ) diff --git a/src/agentci/worker.py b/src/agentci/worker.py index da30a30..b8d1be9 100644 --- a/src/agentci/worker.py +++ b/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(): diff --git a/tests/test_config.py b/tests/test_config.py index 23d649f..91c7107 100644 --- a/tests/test_config.py +++ b/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") diff --git a/tests/test_development.py b/tests/test_development.py index 3f78d0c..d6b665b 100644 --- a/tests/test_development.py +++ b/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() diff --git a/tests/test_storage.py b/tests/test_storage.py index 78a1b5f..15b9a8f 100644 --- a/tests/test_storage.py +++ b/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 diff --git a/tests/test_worker.py b/tests/test_worker.py index a0a9184..5bf2446 100644 --- a/tests/test_worker.py +++ b/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