diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..b58b603 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,5 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Editor-based HTTP Client requests +/httpRequests/ diff --git a/.idea/agentci.iml b/.idea/agentci.iml new file mode 100644 index 0000000..34264c8 --- /dev/null +++ b/.idea/agentci.iml @@ -0,0 +1,11 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/encodings.xml b/.idea/encodings.xml new file mode 100644 index 0000000..df87cf9 --- /dev/null +++ b/.idea/encodings.xml @@ -0,0 +1,4 @@ + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 0000000..105ce2d --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..f6b3aba --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/pyLspTools.xml b/.idea/pyLspTools.xml new file mode 100644 index 0000000..28adef1 --- /dev/null +++ b/.idea/pyLspTools.xml @@ -0,0 +1,23 @@ + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..35eb1dd --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md index 93ea428..c1371d6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,12 +14,12 @@ - `src/agentci/engine/`: immutable domain models and events, the pure reducer, SQLite persistence, task claiming, and the `JobRun` event interface. -- `src/agentci/worker.py`: durable control/job queues, authorization, execution, recovery, and - comment reconciliation. +- `src/agentci/application/`: runtime composition, durable control/job queues, task handlers, + recovery, and comment reconciliation. - `src/agentci/workflows/`: planning, implementation, review, and pull-request orchestration. -- `src/agentci/{gitea,git,opencode,codegraph,development}.py`: external-effect boundaries. -- `src/agentci/{app,runtime,config,webhook,health}.py`: application lifecycle, configuration, and - HTTP entry points. +- `src/agentci/integrations/`: Gitea, Git, OpenCode, CodeGraph, and development external effects. +- `src/agentci/api/`: FastAPI construction, lifespan, dependencies, errors, and HTTP routes. +- `src/agentci/config/` and `src/agentci/observability/`: settings and structured logging. - `src/agentci/prompts/` and `src/agentci/prompts/schemas/`: model prompts and structured-output contracts; keep these concerns outside Python orchestration. - `src/agentci/migrations/`: ordered SQLite migrations. diff --git a/README.md b/README.md index 176e7ab..d4517de 100644 --- a/README.md +++ b/README.md @@ -13,9 +13,9 @@ webhook -> repository -> reducer -> durable task -> worker -> workflow -> integr ``` `engine/reducer.py` is the pure job state machine. `engine/repository.py` applies its transitions -atomically to SQLite and persists the resulting tasks. `worker.py` executes those tasks and passes -an explicit `JobRun` into the functions under `workflows/`. The top-level Gitea, Git, OpenCode, -development, and CodeGraph modules own external effects. +atomically to SQLite and persists the resulting tasks. `application/worker/` executes those tasks +and passes an explicit `JobRun` into the functions under `workflows/`. Modules under +`integrations/` own external effects, while `api/` contains the FastAPI host and routes. Jobs and workflows are immutable snapshots. Workflows return their final comment body directly; progress and resource links are emitted as state-machine events through `JobRun`. The `jobs` table diff --git a/src/agentci/__main__.py b/src/agentci/__main__.py index 5945a05..03fc141 100644 --- a/src/agentci/__main__.py +++ b/src/agentci/__main__.py @@ -2,8 +2,8 @@ from __future__ import annotations import uvicorn -from agentci.app import create_app -from agentci.config import Settings +from agentci.api.app import create_app +from agentci.config.settings import Settings def main() -> None: @@ -19,4 +19,3 @@ def main() -> None: if __name__ == "__main__": main() - diff --git a/src/agentci/api/__init__.py b/src/agentci/api/__init__.py new file mode 100644 index 0000000..f6c85c6 --- /dev/null +++ b/src/agentci/api/__init__.py @@ -0,0 +1 @@ +"""HTTP application and routes.""" diff --git a/src/agentci/api/app.py b/src/agentci/api/app.py new file mode 100644 index 0000000..a2cad02 --- /dev/null +++ b/src/agentci/api/app.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from fastapi import FastAPI + +from agentci.api.errors import register_error_handlers +from agentci.api.lifespan import create_lifespan +from agentci.api.routes.health import router as health_router +from agentci.api.routes.webhook import router as webhook_router +from agentci.config.settings import Settings + + +def create_app(settings: Settings | None = None) -> FastAPI: + selected_settings = settings or Settings() + app = FastAPI( + title="Agent CI", + version="0.1.0", + lifespan=create_lifespan(selected_settings), + ) + app.include_router(health_router) + app.include_router(webhook_router) + register_error_handlers(app) + return app + + +app = create_app() diff --git a/src/agentci/api/dependencies.py b/src/agentci/api/dependencies.py new file mode 100644 index 0000000..875e883 --- /dev/null +++ b/src/agentci/api/dependencies.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from typing import Annotated + +from fastapi import Depends, Request + +from agentci.application.runtime import Runtime + + +def get_runtime(request: Request) -> Runtime: + return request.app.state.runtime + + +type RuntimeDependency = Annotated[Runtime, Depends(get_runtime)] diff --git a/src/agentci/api/errors.py b/src/agentci/api/errors.py new file mode 100644 index 0000000..189456a --- /dev/null +++ b/src/agentci/api/errors.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import logging + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +log = logging.getLogger(__name__) + + +async def unhandled_error(request: Request, exc: Exception) -> JSONResponse: + log.exception( + "unhandled HTTP request failure", + extra={ + "operation": "http.request", + "method": request.method, + "path": request.url.path, + "error_message": str(exc), + "status_code": 500, + }, + ) + return JSONResponse( + status_code=500, + content={"detail": "Internal server error. See service logs for diagnostics."}, + ) + + +def register_error_handlers(app: FastAPI) -> None: + app.exception_handler(Exception)(unhandled_error) diff --git a/src/agentci/api/lifespan.py b/src/agentci/api/lifespan.py new file mode 100644 index 0000000..f7e066e --- /dev/null +++ b/src/agentci/api/lifespan.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncGenerator, Callable +from contextlib import AbstractAsyncContextManager, asynccontextmanager, suppress + +from fastapi import FastAPI + +from agentci.application.runtime import build_runtime +from agentci.config.settings import Settings +from agentci.observability.logging import configure_logging + +log = logging.getLogger(__name__) +Lifespan = Callable[[FastAPI], AbstractAsyncContextManager[None]] + + +def create_lifespan(settings: Settings) -> Lifespan: + @asynccontextmanager + async def lifespan(app: FastAPI) -> AsyncGenerator[None]: + configure_logging() + log.info("service startup started", extra={"operation": "service.startup"}) + try: + runtime = await build_runtime(settings) + except Exception: + log.exception("service startup failed", extra={"operation": "service.startup"}) + raise + app.state.runtime = runtime + stop = asyncio.Event() + worker_task = asyncio.create_task(runtime.worker.run(stop), name="agentci-worker") + try: + log.info("service startup completed", extra={"operation": "service.startup"}) + yield + finally: + log.info("service shutdown started", extra={"operation": "service.shutdown"}) + stop.set() + worker_task.cancel() + try: + with suppress(asyncio.CancelledError): + await worker_task + finally: + await runtime.close() + log.info("service shutdown completed", extra={"operation": "service.shutdown"}) + + return lifespan diff --git a/src/agentci/api/routes/__init__.py b/src/agentci/api/routes/__init__.py new file mode 100644 index 0000000..7595942 --- /dev/null +++ b/src/agentci/api/routes/__init__.py @@ -0,0 +1 @@ +"""FastAPI route modules.""" diff --git a/src/agentci/health.py b/src/agentci/api/routes/health.py similarity index 65% rename from src/agentci/health.py rename to src/agentci/api/routes/health.py index 4cea842..5b5eb0d 100644 --- a/src/agentci/health.py +++ b/src/agentci/api/routes/health.py @@ -1,8 +1,8 @@ from __future__ import annotations -from fastapi import APIRouter, Request, Response, status +from fastapi import APIRouter, Response, status -from agentci.runtime import Runtime +from agentci.api.dependencies import RuntimeDependency router = APIRouter() @@ -13,8 +13,7 @@ async def live() -> dict[str, str]: @router.get("/health/ready") -async def ready(request: Request, response: Response) -> dict[str, str]: - runtime: Runtime = request.app.state.runtime +async def ready(runtime: RuntimeDependency, response: Response) -> dict[str, str]: if not await runtime.opencode.ready(): response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE return {"status": "not-ready", "reason": "opencode provider is not connected"} diff --git a/src/agentci/webhook.py b/src/agentci/api/routes/webhook.py similarity index 66% rename from src/agentci/webhook.py rename to src/agentci/api/routes/webhook.py index c2ef276..d08cbbf 100644 --- a/src/agentci/webhook.py +++ b/src/agentci/api/routes/webhook.py @@ -1,28 +1,25 @@ from __future__ import annotations -import hashlib -import hmac import json import logging -from typing import Any from fastapi import APIRouter, HTTPException, Request, Response, status +from agentci.api.dependencies import RuntimeDependency +from agentci.application.runtime import Runtime from agentci.engine.model import IncomingCommand -from agentci.runtime import Runtime +from agentci.integrations.gitea.webhooks import ( + SUPPORTED_EVENTS, + incoming_command_from_payload, + valid_signature, +) router = APIRouter() log = logging.getLogger(__name__) -SUPPORTED_EVENTS = { - "issue_comment", - "pull_request_comment", - "pull_request_review_comment", -} @router.post("/webhooks/gitea") -async def webhook(request: Request) -> Response: - runtime: Runtime = request.app.state.runtime +async def webhook(request: Request, runtime: RuntimeDependency) -> Response: body = await request.body() signature = request.headers.get("X-Gitea-Signature", "") if not valid_signature(runtime.settings.webhook_secret, body, signature): @@ -44,7 +41,7 @@ async def webhook(request: Request) -> Response: payload = json.loads(body) if not isinstance(payload, dict): raise ValueError("Webhook payload must be a JSON object") - event = _event_from_payload( + event = incoming_command_from_payload( request.headers.get("X-Gitea-Delivery", ""), payload, ) @@ -61,11 +58,6 @@ async def webhook(request: Request) -> Response: return await _handle_command(runtime, event) -def valid_signature(secret: bytes, body: bytes, signature: str) -> bool: - expected = hmac.new(secret, body, hashlib.sha256).hexdigest() - return bool(signature) and hmac.compare_digest(expected, signature) - - async def _handle_command(runtime: Runtime, event: IncomingCommand) -> Response: extra = {"operation": "command.handle", "target": event.target_key} if not event.body.strip().startswith("/agent"): @@ -90,29 +82,3 @@ async def _handle_command(runtime: Runtime, event: IncomingCommand) -> Response: }, ) return Response(status_code=status.HTTP_202_ACCEPTED) - - -def _event_from_payload(delivery_id: str, payload: dict[str, Any]) -> IncomingCommand | None: - if payload.get("action") != "created": - return None - comment = payload["comment"] - repository = payload["repository"] - owner = repository["owner"] - owner_name = owner.get("login") or owner.get("username") or owner["name"] - pull = payload.get("pull_request") - is_pull = bool(payload.get("is_pull") or pull) - issue = payload.get("issue") - target = pull or issue - if target is None: - raise ValueError("Comment payload has no issue or pull request") - number = int(target["number"]) - return IncomingCommand( - delivery_id=delivery_id, - comment_id=int(comment["id"]), - repo_owner=owner_name, - repo_name=repository["name"], - issue_number=number, - pr_number=number if is_pull else None, - requester=comment["user"]["login"], - body=comment.get("body") or "", - ) diff --git a/src/agentci/app.py b/src/agentci/app.py deleted file mode 100644 index a67e439..0000000 --- a/src/agentci/app.py +++ /dev/null @@ -1,71 +0,0 @@ -from __future__ import annotations - -import asyncio -import logging -from collections.abc import AsyncIterator -from contextlib import asynccontextmanager, suppress - -from fastapi import FastAPI, Request -from fastapi.responses import JSONResponse - -from agentci.config import Settings -from agentci.health import router as health_router -from agentci.logging import configure_logging -from agentci.runtime import build_runtime -from agentci.webhook import router as webhook_router - -log = logging.getLogger(__name__) - -def create_app(settings: Settings | None = None) -> FastAPI: - selected_settings = settings or Settings() - - @asynccontextmanager - async def lifespan(app: FastAPI) -> AsyncIterator[None]: - configure_logging() - log.info("service startup started", extra={"operation": "service.startup"}) - try: - runtime = await build_runtime(selected_settings) - except Exception: - log.exception("service startup failed", extra={"operation": "service.startup"}) - raise - app.state.runtime = runtime - stop = asyncio.Event() - worker_task = asyncio.create_task(runtime.worker.run(stop), name="agentci-worker") - try: - log.info("service startup completed", extra={"operation": "service.startup"}) - yield - finally: - log.info("service shutdown started", extra={"operation": "service.shutdown"}) - stop.set() - worker_task.cancel() - try: - with suppress(asyncio.CancelledError): - await worker_task - finally: - await runtime.close() - log.info("service shutdown completed", extra={"operation": "service.shutdown"}) - - app = FastAPI(title="Agent CI", version="0.1.0", lifespan=lifespan) - app.include_router(health_router) - app.include_router(webhook_router) - - @app.exception_handler(Exception) - async def unhandled_error(request: Request, exc: Exception) -> JSONResponse: - log.exception( - "unhandled HTTP request failure", - extra={ - "operation": "http.request", - "method": request.method, - "path": request.url.path, - "status_code": 500, - }, - ) - return JSONResponse( - status_code=500, - content={"detail": "Internal server error. See service logs for diagnostics."}, - ) - - return app - - -app = create_app() diff --git a/src/agentci/application/__init__.py b/src/agentci/application/__init__.py new file mode 100644 index 0000000..858c2a1 --- /dev/null +++ b/src/agentci/application/__init__.py @@ -0,0 +1 @@ +"""Application composition and durable orchestration.""" diff --git a/src/agentci/runtime.py b/src/agentci/application/runtime.py similarity index 85% rename from src/agentci/runtime.py rename to src/agentci/application/runtime.py index 2e241ce..f2f29af 100644 --- a/src/agentci/runtime.py +++ b/src/agentci/application/runtime.py @@ -2,16 +2,15 @@ from __future__ import annotations import logging from dataclasses import dataclass -from pathlib import Path -from agentci.config import Settings -from agentci.development import DevelopmentEnvironment +from agentci.application.worker.runner import Worker +from agentci.config.settings import Settings from agentci.engine.repository import Repository -from agentci.git import Git -from agentci.gitea import Gitea -from agentci.opencode import OpenCode -from agentci.prompts import PromptLibrary -from agentci.worker import Worker +from agentci.integrations.development import DevelopmentEnvironment +from agentci.integrations.git import Git +from agentci.integrations.gitea.client import Gitea +from agentci.integrations.opencode.client import OpenCode +from agentci.prompts.library import PromptLibrary from agentci.workflows.services import WorkflowServices log = logging.getLogger(__name__) @@ -36,10 +35,9 @@ class Runtime: async def build_runtime(settings: Settings) -> Runtime: log.info("runtime initialization started", extra={"operation": "runtime.build"}) - package_dir = Path(__file__).parent settings.data_dir.mkdir(parents=True, exist_ok=True) settings.workspaces_dir.mkdir(parents=True, exist_ok=True) - repository = Repository(settings.database_path, package_dir / "migrations") + repository = Repository(settings.database_path) await repository.initialize() gitea = Gitea(settings.gitea_url, settings.gitea_token) git = Git( @@ -50,11 +48,12 @@ async def build_runtime(settings: Settings) -> Runtime: commit_name=settings.bot_name, commit_email=settings.bot_email, ) + prompts = PromptLibrary() opencode = OpenCode( base_url=settings.opencode_url, username=settings.opencode_server_username, password=settings.opencode_server_password, - schemas_dir=package_dir / "prompts" / "schemas", + schemas_dir=prompts.schemas_dir, health_directory=settings.workspaces_dir, required_models=( (settings.plan_model, settings.plan_variant), @@ -78,7 +77,7 @@ async def build_runtime(settings: Settings) -> Runtime: gitea=gitea, git=git, opencode=opencode, - prompts=PromptLibrary(), + prompts=prompts, development=development, ) worker = Worker( diff --git a/src/agentci/application/worker/__init__.py b/src/agentci/application/worker/__init__.py new file mode 100644 index 0000000..36a41e6 --- /dev/null +++ b/src/agentci/application/worker/__init__.py @@ -0,0 +1 @@ +"""Durable task runner and task handlers.""" diff --git a/src/agentci/application/worker/authorization.py b/src/agentci/application/worker/authorization.py new file mode 100644 index 0000000..4ab08fd --- /dev/null +++ b/src/agentci/application/worker/authorization.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from agentci.engine.events import PermissionDenied, PermissionGranted +from agentci.engine.model import Job, JobStatus, Task +from agentci.engine.repository import Repository +from agentci.integrations.gitea.client import Gitea + + +async def authorize_job( + *, task: Task, job: Job, repository: Repository, gitea: Gitea +) -> None: + if job.status is not JobStatus.RECEIVED: + return + permitted = await gitea.has_write_permission(job.repo_owner, job.repo_name, job.requester) + event = PermissionGranted(job_id=job.id) if permitted else PermissionDenied(job_id=job.id) + outcome = "permission-granted" if permitted else "permission-denied" + await repository.apply(f"task:{task.id}:{outcome}", event) diff --git a/src/agentci/application/worker/comments.py b/src/agentci/application/worker/comments.py new file mode 100644 index 0000000..c8ed505 --- /dev/null +++ b/src/agentci/application/worker/comments.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from agentci.engine.events import CommentLinked +from agentci.engine.model import Job, Task +from agentci.engine.reducer import render_job_comment +from agentci.engine.repository import Repository +from agentci.integrations.gitea.client import Gitea + + +async def reconcile_comment( + *, + task: Task, + job: Job, + repository: Repository, + gitea: Gitea, + bot_username: str, +) -> None: + latest = await repository.get_job(job.id) + if latest is None: + return + body = render_job_comment(latest) + comment_id = latest.accepted_comment_id + if comment_id is not None and await gitea.update_comment( + latest.repo_owner, latest.repo_name, comment_id, body + ): + return + marker = f"" + matches = sorted( + comment.id + for comment in await gitea.issue_comments( + latest.repo_owner, latest.repo_name, latest.issue_number + ) + if comment.body.startswith(marker) + and comment.author.casefold() == bot_username.casefold() + ) + if matches: + comment_id = matches[0] + else: + comment_id = await gitea.create_comment( + latest.repo_owner, latest.repo_name, latest.issue_number, body + ) + await repository.apply( + f"task:{task.id}:comment:{comment_id}", + CommentLinked(job_id=latest.id, comment_id=comment_id), + ) + await gitea.update_comment( + latest.repo_owner, latest.repo_name, comment_id, body + ) diff --git a/src/agentci/application/worker/errors.py b/src/agentci/application/worker/errors.py new file mode 100644 index 0000000..e8d236b --- /dev/null +++ b/src/agentci/application/worker/errors.py @@ -0,0 +1,6 @@ +from __future__ import annotations + + +def safe_error(error: Exception) -> str: + message = " ".join(str(error).split()) + return f"{type(error).__name__}: {message}"[:1000] diff --git a/src/agentci/application/worker/execution.py b/src/agentci/application/worker/execution.py new file mode 100644 index 0000000..b0e73b7 --- /dev/null +++ b/src/agentci/application/worker/execution.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from agentci.application.worker.errors import safe_error +from agentci.engine.events import JobCompleted, JobFailed, JobStarted, ServiceRestarted +from agentci.engine.events import JobRejected as RejectedEvent +from agentci.engine.model import Job, JobStatus, Task +from agentci.engine.repository import Repository +from agentci.engine.run import JobRun +from agentci.workflows.dispatch import dispatch +from agentci.workflows.render import JobRejected +from agentci.workflows.services import WorkflowServices + + +async def execute_job( + *, task: Task, job: Job, repository: Repository, services: WorkflowServices +) -> None: + if job.status is JobStatus.RUNNING: + await repository.apply( + f"task:{task.id}:interrupted", ServiceRestarted(job_id=job.id) + ) + return + if job.status is not JobStatus.QUEUED: + return + result = await repository.apply(f"task:{task.id}:started", JobStarted(job_id=job.id)) + running = result.job + run = JobRun(repository, job.id, task.id) + try: + body = await dispatch(running, run, services) + except JobRejected as exc: + await repository.apply( + f"task:{task.id}:rejected", RejectedEvent(job_id=job.id, reason=str(exc)) + ) + except Exception as exc: + latest = await repository.get_job(job.id) + stage = latest.stage if latest else running.stage + await repository.apply( + f"task:{task.id}:failed", + JobFailed(job_id=job.id, error=safe_error(exc), stage=stage), + ) + else: + await repository.apply( + f"task:{task.id}:completed", + JobCompleted( + job_id=job.id, + comment_body=body or "Agent job completed.", + ), + ) diff --git a/src/agentci/application/worker/recovery.py b/src/agentci/application/worker/recovery.py new file mode 100644 index 0000000..7c7d32b --- /dev/null +++ b/src/agentci/application/worker/recovery.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +from agentci.engine.events import ServiceRestarted +from agentci.engine.repository import Repository + + +async def recover_jobs(repository: Repository) -> None: + await repository.recover_tasks() + for job in await repository.running_jobs(): + await repository.apply( + f"recovery:{job.id}:service-restarted", + ServiceRestarted(job_id=job.id), + ) diff --git a/src/agentci/application/worker/runner.py b/src/agentci/application/worker/runner.py new file mode 100644 index 0000000..73d2e2e --- /dev/null +++ b/src/agentci/application/worker/runner.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import asyncio +import logging +from contextlib import suppress +from pathlib import Path + +from agentci.application.worker.authorization import authorize_job +from agentci.application.worker.comments import reconcile_comment +from agentci.application.worker.errors import safe_error +from agentci.application.worker.execution import execute_job +from agentci.application.worker.recovery import recover_jobs +from agentci.application.worker.sessions import abort_job_sessions +from agentci.engine.model import Job, QueueName, Task, TaskKind +from agentci.engine.repository import Repository +from agentci.integrations.gitea.client import Gitea +from agentci.integrations.opencode.client import OpenCode +from agentci.workflows.services import WorkflowServices + +log = logging.getLogger(__name__) + + +class Worker: + def __init__( + self, + *, + repository: Repository, + gitea: Gitea, + opencode: OpenCode, + services: WorkflowServices, + poll_seconds: float, + max_concurrent_jobs: int, + workspaces_dir: Path, + bot_username: str, + ) -> None: + self.repository = repository + self.gitea = gitea + self.opencode = opencode + self.services = services + 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(QueueName.CONTROL, stop), + *(self._loop(QueueName.JOBS, stop) for _ in range(self.max_concurrent_jobs)), + ) + + async def _loop(self, queue: QueueName, stop: asyncio.Event) -> None: + while not stop.is_set(): + if queue is QueueName.JOBS and not await self.opencode.ready(): + await self._wait(stop) + continue + task = await self.repository.claim_task(queue) + if task is None: + await self._wait(stop) + continue + try: + await self._handle(task) + except asyncio.CancelledError: + raise + except Exception as exc: + log.exception( + "listener failed", + extra={ + "task_id": task.id, + "listener": task.kind.value, + "queue": queue.value, + }, + ) + await self.repository.retry_task(task.id, task.attempts, safe_error(exc)) + else: + await self.repository.complete_task(task.id) + + async def _handle(self, task: Task) -> None: + job = await self.repository.get_job(task.job_id) + if job is None: + return + match task.kind: + case TaskKind.AUTHORIZE: + await self._authorize(task, job) + case TaskKind.EXECUTE: + await self._execute(task, job) + case TaskKind.RECONCILE_COMMENT: + await self._reconcile(task, job) + case TaskKind.FAIL_WORKFLOW: + await self.repository.fail_job_workflow(job.id) + case TaskKind.ABORT_SESSIONS: + await self._abort_job_sessions(job) + case _: + raise RuntimeError(f"Unknown task kind {task.kind}") + + async def _authorize(self, task: Task, job: Job) -> None: + await authorize_job( + task=task, + job=job, + repository=self.repository, + gitea=self.gitea, + ) + + async def _execute(self, task: Task, job: Job) -> None: + await execute_job( + task=task, + job=job, + repository=self.repository, + services=self.services, + ) + + async def _reconcile(self, task: Task, job: Job) -> None: + await reconcile_comment( + task=task, + job=job, + repository=self.repository, + gitea=self.gitea, + bot_username=self.bot_username, + ) + + async def _recover(self) -> None: + await recover_jobs(self.repository) + + async def _abort_job_sessions(self, job: Job) -> None: + await abort_job_sessions( + job=job, + repository=self.repository, + opencode=self.opencode, + workspaces_dir=self.workspaces_dir, + ) + + async def _wait(self, stop: asyncio.Event) -> None: + with suppress(TimeoutError): + await asyncio.wait_for(stop.wait(), timeout=self.poll_seconds) diff --git a/src/agentci/application/worker/sessions.py b/src/agentci/application/worker/sessions.py new file mode 100644 index 0000000..47f367c --- /dev/null +++ b/src/agentci/application/worker/sessions.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from pathlib import Path + +from agentci.engine.model import Job +from agentci.engine.repository import Repository +from agentci.integrations.opencode.client import OpenCode + + +async def abort_job_sessions( + *, + job: Job, + repository: Repository, + opencode: OpenCode, + workspaces_dir: Path, +) -> None: + sessions: set[tuple[str, Path]] = set() + workflow = await repository.get_workflow(job.workflow_id) if job.workflow_id else None + if workflow: + sessions.update( + (session, workflow.workspace_path) + for session in (workflow.primary_session_id, workflow.reviewer_session_id) + if session + ) + elif job.runtime_session_id: + sessions.add( + (job.runtime_session_id, workspaces_dir / f"fix-{job.id}" / "repo") + ) + for session, workspace in sessions: + await opencode.abort(session, workspace) diff --git a/src/agentci/config/__init__.py b/src/agentci/config/__init__.py new file mode 100644 index 0000000..30025d0 --- /dev/null +++ b/src/agentci/config/__init__.py @@ -0,0 +1 @@ +"""Service configuration.""" diff --git a/src/agentci/config.py b/src/agentci/config/settings.py similarity index 100% rename from src/agentci/config.py rename to src/agentci/config/settings.py diff --git a/src/agentci/integrations/__init__.py b/src/agentci/integrations/__init__.py new file mode 100644 index 0000000..8761bc1 --- /dev/null +++ b/src/agentci/integrations/__init__.py @@ -0,0 +1 @@ +"""External service and process integrations.""" diff --git a/src/agentci/codegraph.py b/src/agentci/integrations/codegraph.py similarity index 100% rename from src/agentci/codegraph.py rename to src/agentci/integrations/codegraph.py diff --git a/src/agentci/development.py b/src/agentci/integrations/development.py similarity index 100% rename from src/agentci/development.py rename to src/agentci/integrations/development.py diff --git a/src/agentci/git.py b/src/agentci/integrations/git.py similarity index 100% rename from src/agentci/git.py rename to src/agentci/integrations/git.py diff --git a/src/agentci/integrations/gitea/__init__.py b/src/agentci/integrations/gitea/__init__.py new file mode 100644 index 0000000..4ce406f --- /dev/null +++ b/src/agentci/integrations/gitea/__init__.py @@ -0,0 +1 @@ +"""Gitea API and webhook integration.""" diff --git a/src/agentci/gitea.py b/src/agentci/integrations/gitea/client.py similarity index 92% rename from src/agentci/gitea.py rename to src/agentci/integrations/gitea/client.py index 6b489d8..001922c 100644 --- a/src/agentci/gitea.py +++ b/src/agentci/integrations/gitea/client.py @@ -2,49 +2,16 @@ from __future__ import annotations import asyncio import logging -from dataclasses import dataclass from time import monotonic from typing import Any import httpx +from agentci.integrations.gitea.models import CommentInfo, IssueInfo, PullRequestInfo + log = logging.getLogger(__name__) -@dataclass(frozen=True) -class IssueInfo: - number: int - title: str - body: str - state: str - - -@dataclass(frozen=True) -class CommentInfo: - id: int - author: str - body: str - created_at: str - - -@dataclass(frozen=True) -class PullRequestInfo: - number: int - title: str - body: str - state: str - merged: bool - base_branch: str - head_branch: str - head_sha: str - head_owner: str - head_repo: str - - @property - def is_open(self) -> bool: - return self.state == "open" and not self.merged - - class GiteaError(RuntimeError): pass diff --git a/src/agentci/integrations/gitea/models.py b/src/agentci/integrations/gitea/models.py new file mode 100644 index 0000000..2b4b9b6 --- /dev/null +++ b/src/agentci/integrations/gitea/models.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class IssueInfo: + number: int + title: str + body: str + state: str + + +@dataclass(frozen=True) +class CommentInfo: + id: int + author: str + body: str + created_at: str + + +@dataclass(frozen=True) +class PullRequestInfo: + number: int + title: str + body: str + state: str + merged: bool + base_branch: str + head_branch: str + head_sha: str + head_owner: str + head_repo: str + + @property + def is_open(self) -> bool: + return self.state == "open" and not self.merged diff --git a/src/agentci/integrations/gitea/webhooks.py b/src/agentci/integrations/gitea/webhooks.py new file mode 100644 index 0000000..4fee9a4 --- /dev/null +++ b/src/agentci/integrations/gitea/webhooks.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import hashlib +import hmac +from typing import Any + +from agentci.engine.model import IncomingCommand + +SUPPORTED_EVENTS = { + "issue_comment", + "pull_request_comment", + "pull_request_review_comment", +} + + +def valid_signature(secret: bytes, body: bytes, signature: str) -> bool: + expected = hmac.new(secret, body, hashlib.sha256).hexdigest() + return bool(signature) and hmac.compare_digest(expected, signature) + + +def incoming_command_from_payload( + delivery_id: str, payload: dict[str, Any] +) -> IncomingCommand | None: + if payload.get("action") != "created": + return None + comment = payload["comment"] + repository = payload["repository"] + owner = repository["owner"] + owner_name = owner.get("login") or owner.get("username") or owner["name"] + pull = payload.get("pull_request") + is_pull = bool(payload.get("is_pull") or pull) + issue = payload.get("issue") + target = pull or issue + if target is None: + raise ValueError("Comment payload has no issue or pull request") + number = int(target["number"]) + return IncomingCommand( + delivery_id=delivery_id, + comment_id=int(comment["id"]), + repo_owner=owner_name, + repo_name=repository["name"], + issue_number=number, + pr_number=number if is_pull else None, + requester=comment["user"]["login"], + body=comment.get("body") or "", + ) diff --git a/src/agentci/integrations/opencode/__init__.py b/src/agentci/integrations/opencode/__init__.py new file mode 100644 index 0000000..7334d1d --- /dev/null +++ b/src/agentci/integrations/opencode/__init__.py @@ -0,0 +1 @@ +"""OpenCode API integration.""" diff --git a/src/agentci/opencode.py b/src/agentci/integrations/opencode/client.py similarity index 78% rename from src/agentci/opencode.py rename to src/agentci/integrations/opencode/client.py index 3f6333e..4ff1a0b 100644 --- a/src/agentci/opencode.py +++ b/src/agentci/integrations/opencode/client.py @@ -1,7 +1,6 @@ from __future__ import annotations import asyncio -import json import logging from contextlib import suppress from pathlib import Path @@ -11,7 +10,9 @@ from typing import Any, TypeVar import httpx from pydantic import BaseModel, ValidationError -from agentci.codegraph import CodeGraph +from agentci.integrations.codegraph import CodeGraph +from agentci.integrations.opencode.readiness import api_contract_ready, models_ready +from agentci.integrations.opencode.schemas import load_schema T = TypeVar("T", bound=BaseModel) log = logging.getLogger(__name__) @@ -265,67 +266,3 @@ def elapsed_ms(started: float) -> int: def directory_headers(workspace: Path) -> dict[str, str]: return {"X-Opencode-Directory": str(workspace.resolve())} - -def load_schema(schemas_dir: Path, name: str) -> dict[str, Any]: - try: - value = json.loads((schemas_dir / name).read_text()) - except (OSError, ValueError) as exc: - raise ValueError(f"Cannot load result schema {name}: {exc}") from exc - if not isinstance(value, dict): - raise ValueError(f"Result schema {name} is not a JSON object") - return value - - -def api_contract_ready(document: object) -> bool: - if not isinstance(document, dict) or not isinstance(document.get("paths"), dict): - return False - paths = document["paths"] - fixed = {"/global/health": "get", "/provider": "get", "/session": "post"} - for path, method in fixed.items(): - operations = paths.get(path) - if not isinstance(operations, dict) or method not in operations: - return False - session_paths = [ - path - for path, operations in paths.items() - if isinstance(path, str) and isinstance(operations, dict) and path.startswith("/session/{") - ] - has_message = any(path.endswith("/message") and "post" in paths[path] for path in session_paths) - has_abort = any(path.endswith("/abort") and "post" in paths[path] for path in session_paths) - return has_message and has_abort - - -def models_ready(payload: object, requirements: set[tuple[str, str, str | None]]) -> bool: - if not isinstance(payload, dict): - return False - connected_value = payload.get("connected") - provider_values = payload.get("all") - if not isinstance(connected_value, list) or not all( - isinstance(item, str) for item in connected_value - ): - return False - if not isinstance(provider_values, list): - return False - connected = set(connected_value) - providers: dict[str, dict[str, Any]] = {} - for item in provider_values: - if not isinstance(item, dict): - continue - provider_id = item.get("id") - if isinstance(provider_id, str) and isinstance(item.get("models"), dict): - providers[provider_id] = item - for provider_id, model_id, variant in requirements: - provider = providers.get(provider_id) - if provider_id not in connected or not isinstance(provider, dict): - return False - model: Any = provider["models"].get(model_id) - if not isinstance(model, dict) or model.get("status") == "deprecated": - return False - capabilities = model.get("capabilities") - if not isinstance(capabilities, dict) or capabilities.get("toolcall") is not True: - return False - if variant: - variants = model.get("variants") - if not isinstance(variants, dict) or variant not in variants: - return False - return True diff --git a/src/agentci/integrations/opencode/readiness.py b/src/agentci/integrations/opencode/readiness.py new file mode 100644 index 0000000..4f63a59 --- /dev/null +++ b/src/agentci/integrations/opencode/readiness.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +from typing import Any + + +def api_contract_ready(document: object) -> bool: + if not isinstance(document, dict) or not isinstance(document.get("paths"), dict): + return False + paths = document["paths"] + fixed = {"/global/health": "get", "/provider": "get", "/session": "post"} + for path, method in fixed.items(): + operations = paths.get(path) + if not isinstance(operations, dict) or method not in operations: + return False + session_paths = [ + path + for path, operations in paths.items() + if isinstance(path, str) and isinstance(operations, dict) and path.startswith("/session/{") + ] + has_message = any(path.endswith("/message") and "post" in paths[path] for path in session_paths) + has_abort = any(path.endswith("/abort") and "post" in paths[path] for path in session_paths) + return has_message and has_abort + + +def models_ready(payload: object, requirements: set[tuple[str, str, str | None]]) -> bool: + if not isinstance(payload, dict): + return False + connected_value = payload.get("connected") + provider_values = payload.get("all") + if not isinstance(connected_value, list) or not all( + isinstance(item, str) for item in connected_value + ): + return False + if not isinstance(provider_values, list): + return False + connected = set(connected_value) + providers: dict[str, dict[str, Any]] = {} + for item in provider_values: + if not isinstance(item, dict): + continue + provider_id = item.get("id") + if isinstance(provider_id, str) and isinstance(item.get("models"), dict): + providers[provider_id] = item + for provider_id, model_id, variant in requirements: + provider = providers.get(provider_id) + if provider_id not in connected or not isinstance(provider, dict): + return False + model: Any = provider["models"].get(model_id) + if not isinstance(model, dict) or model.get("status") == "deprecated": + return False + capabilities = model.get("capabilities") + if not isinstance(capabilities, dict) or capabilities.get("toolcall") is not True: + return False + if variant: + variants = model.get("variants") + if not isinstance(variants, dict) or variant not in variants: + return False + return True diff --git a/src/agentci/integrations/opencode/schemas.py b/src/agentci/integrations/opencode/schemas.py new file mode 100644 index 0000000..6995cb5 --- /dev/null +++ b/src/agentci/integrations/opencode/schemas.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + + +def load_schema(schemas_dir: Path, name: str) -> dict[str, Any]: + try: + value = json.loads((schemas_dir / name).read_text()) + except (OSError, ValueError) as exc: + raise ValueError(f"Cannot load result schema {name}: {exc}") from exc + if not isinstance(value, dict): + raise ValueError(f"Result schema {name} is not a JSON object") + return value diff --git a/src/agentci/observability/__init__.py b/src/agentci/observability/__init__.py new file mode 100644 index 0000000..f5912d1 --- /dev/null +++ b/src/agentci/observability/__init__.py @@ -0,0 +1 @@ +"""Service observability configuration.""" diff --git a/src/agentci/logging.py b/src/agentci/observability/logging.py similarity index 98% rename from src/agentci/logging.py rename to src/agentci/observability/logging.py index 98f5a90..c1dd836 100644 --- a/src/agentci/logging.py +++ b/src/agentci/observability/logging.py @@ -15,6 +15,7 @@ LOG_FIELDS = ( "method", "path", "status_code", + "error_message", "attempt", "item_count", "duration_ms", diff --git a/src/agentci/prompts/__init__.py b/src/agentci/prompts/__init__.py index 83d790f..077a3fe 100644 --- a/src/agentci/prompts/__init__.py +++ b/src/agentci/prompts/__init__.py @@ -1,14 +1 @@ -from __future__ import annotations - -from pathlib import Path -from string import Template - - -class PromptLibrary: - def __init__(self, directory: Path | None = None) -> None: - self.directory = directory or Path(__file__).parent - - def render(self, name: str, **values: str) -> str: - template = Template((self.directory / f"{name}.md").read_text()) - return template.substitute(values) - +"""Prompt templates and structured-output contracts.""" diff --git a/src/agentci/prompts/library.py b/src/agentci/prompts/library.py new file mode 100644 index 0000000..63143e6 --- /dev/null +++ b/src/agentci/prompts/library.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from pathlib import Path +from string import Template + + +class PromptLibrary: + def __init__(self, directory: Path | None = None) -> None: + self.directory = directory or Path(__file__).parent + + def render(self, name: str, **values: str) -> str: + template = Template((self.directory / f"{name}.md").read_text()) + return template.substitute(values) + + @property + def schemas_dir(self) -> Path: + return self.directory / "schemas" diff --git a/src/agentci/worker.py b/src/agentci/worker.py deleted file mode 100644 index a9bb845..0000000 --- a/src/agentci/worker.py +++ /dev/null @@ -1,221 +0,0 @@ -from __future__ import annotations - -import asyncio -import logging -from contextlib import suppress -from pathlib import Path - -from agentci.engine.events import ( - CommentLinked, - JobCompleted, - JobFailed, - JobStarted, - PermissionDenied, - PermissionGranted, - ServiceRestarted, -) -from agentci.engine.events import ( - JobRejected as RejectedEvent, -) -from agentci.engine.model import Job, JobStatus, QueueName, Task, TaskKind -from agentci.engine.reducer import render_job_comment -from agentci.engine.repository import Repository -from agentci.engine.run import JobRun -from agentci.gitea import Gitea -from agentci.opencode import OpenCode -from agentci.workflows.dispatch import dispatch -from agentci.workflows.render import JobRejected -from agentci.workflows.services import WorkflowServices - -log = logging.getLogger(__name__) - - -class Worker: - def __init__( - self, - *, - repository: Repository, - gitea: Gitea, - opencode: OpenCode, - services: WorkflowServices, - poll_seconds: float, - max_concurrent_jobs: int, - workspaces_dir: Path, - bot_username: str, - ) -> None: - self.repository = repository - self.gitea = gitea - self.opencode = opencode - self.services = services - 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(QueueName.CONTROL, stop), - *(self._loop(QueueName.JOBS, stop) for _ in range(self.max_concurrent_jobs)), - ) - - async def _loop(self, queue: QueueName, stop: asyncio.Event) -> None: - while not stop.is_set(): - if queue is QueueName.JOBS and not await self.opencode.ready(): - await self._wait(stop) - continue - task = await self.repository.claim_task(queue) - if task is None: - await self._wait(stop) - continue - try: - await self._handle(task) - except asyncio.CancelledError: - raise - except Exception as exc: - log.exception( - "listener failed", - extra={ - "task_id": task.id, - "listener": task.kind.value, - "queue": queue.value, - }, - ) - await self.repository.retry_task(task.id, task.attempts, _safe_error(exc)) - else: - await self.repository.complete_task(task.id) - - async def _handle(self, task: Task) -> None: - job = await self.repository.get_job(task.job_id) - if job is None: - return - match task.kind: - case TaskKind.AUTHORIZE: - await self._authorize(task, job) - case TaskKind.EXECUTE: - await self._execute(task, job) - case TaskKind.RECONCILE_COMMENT: - await self._reconcile(task, job) - case TaskKind.FAIL_WORKFLOW: - await self.repository.fail_job_workflow(job.id) - case TaskKind.ABORT_SESSIONS: - await self._abort_job_sessions(job) - case _: - raise RuntimeError(f"Unknown task kind {task.kind}") - - async def _authorize(self, task: Task, job: Job) -> None: - if job.status is not JobStatus.RECEIVED: - return - permitted = await self.gitea.has_write_permission( - job.repo_owner, job.repo_name, job.requester - ) - event = ( - PermissionGranted(job_id=job.id) - if permitted - else PermissionDenied(job_id=job.id) - ) - outcome = "permission-granted" if permitted else "permission-denied" - await self.repository.apply(f"task:{task.id}:{outcome}", event) - - async def _execute(self, task: Task, job: Job) -> None: - if job.status is JobStatus.RUNNING: - await self.repository.apply( - f"task:{task.id}:interrupted", ServiceRestarted(job_id=job.id) - ) - return - if job.status is not JobStatus.QUEUED: - return - result = await self.repository.apply( - f"task:{task.id}:started", JobStarted(job_id=job.id) - ) - running = result.job - run = JobRun(self.repository, job.id, task.id) - try: - body = await dispatch(running, run, self.services) - except JobRejected as exc: - await self.repository.apply( - f"task:{task.id}:rejected", RejectedEvent(job_id=job.id, reason=str(exc)) - ) - except Exception as exc: - latest = await self.repository.get_job(job.id) - stage = latest.stage if latest else running.stage - await self.repository.apply( - f"task:{task.id}:failed", - JobFailed(job_id=job.id, error=_safe_error(exc), stage=stage), - ) - else: - await self.repository.apply( - f"task:{task.id}:completed", - JobCompleted( - job_id=job.id, - comment_body=body or "Agent job completed.", - ), - ) - - async def _reconcile(self, task: Task, job: Job) -> None: - latest = await self.repository.get_job(job.id) - if latest is None: - return - body = render_job_comment(latest) - comment_id = latest.accepted_comment_id - if comment_id is not None and await self.gitea.update_comment( - latest.repo_owner, latest.repo_name, comment_id, body - ): - return - marker = f"" - matches = sorted( - comment.id - for comment in await self.gitea.issue_comments( - latest.repo_owner, latest.repo_name, latest.issue_number - ) - if comment.body.startswith(marker) - and comment.author.casefold() == self.bot_username.casefold() - ) - if matches: - comment_id = matches[0] - else: - comment_id = await self.gitea.create_comment( - latest.repo_owner, latest.repo_name, latest.issue_number, body - ) - await self.repository.apply( - f"task:{task.id}:comment:{comment_id}", - CommentLinked(job_id=latest.id, comment_id=comment_id), - ) - await self.gitea.update_comment( - latest.repo_owner, latest.repo_name, comment_id, body - ) - - async def _recover(self) -> None: - await self.repository.recover_tasks() - for job in await self.repository.running_jobs(): - await self.repository.apply( - f"recovery:{job.id}:service-restarted", - ServiceRestarted(job_id=job.id), - ) - - async def _abort_job_sessions(self, job: Job) -> None: - sessions: set[tuple[str, Path]] = set() - workflow = ( - await self.repository.get_workflow(job.workflow_id) if job.workflow_id else None - ) - if workflow: - sessions.update( - (session, workflow.workspace_path) - for session in (workflow.primary_session_id, workflow.reviewer_session_id) - if session - ) - elif job.runtime_session_id: - sessions.add( - (job.runtime_session_id, self.workspaces_dir / f"fix-{job.id}" / "repo") - ) - for session, workspace in sessions: - await self.opencode.abort(session, workspace) - - async def _wait(self, stop: asyncio.Event) -> None: - with suppress(TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=self.poll_seconds) - - -def _safe_error(error: Exception) -> str: - message = " ".join(str(error).split()) - return f"{type(error).__name__}: {message}"[:1000] diff --git a/src/agentci/workflows/context.py b/src/agentci/workflows/context.py index b1e8f02..8b8f571 100644 --- a/src/agentci/workflows/context.py +++ b/src/agentci/workflows/context.py @@ -4,7 +4,8 @@ import asyncio from typing import Any from agentci.engine.repository import Repository -from agentci.gitea import CommentInfo, Gitea, PullRequestInfo +from agentci.integrations.gitea.client import Gitea +from agentci.integrations.gitea.models import CommentInfo, PullRequestInfo async def build_issue_context( diff --git a/src/agentci/workflows/services.py b/src/agentci/workflows/services.py index 7e19ffb..f73f2b8 100644 --- a/src/agentci/workflows/services.py +++ b/src/agentci/workflows/services.py @@ -2,13 +2,13 @@ from __future__ import annotations from dataclasses import dataclass -from agentci.config import Settings -from agentci.development import DevelopmentEnvironment +from agentci.config.settings import Settings from agentci.engine.repository import Repository -from agentci.git import Git -from agentci.gitea import Gitea -from agentci.opencode import OpenCode -from agentci.prompts import PromptLibrary +from agentci.integrations.development import DevelopmentEnvironment +from agentci.integrations.git import Git +from agentci.integrations.gitea.client import Gitea +from agentci.integrations.opencode.client import OpenCode +from agentci.prompts.library import PromptLibrary @dataclass(frozen=True) diff --git a/tests/test_app.py b/tests/test_app.py index 066a0f6..50ca4e0 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -4,7 +4,8 @@ from types import SimpleNamespace import pytest from httpx import ASGITransport, AsyncClient -import agentci.app as app_module +import agentci.api.app as app_module +import agentci.api.lifespan as lifespan_module class BlockingWorker: @@ -56,8 +57,8 @@ async def test_lifespan_starts_worker_cancels_it_and_closes_runtime( built_with.append(settings) return runtime - monkeypatch.setattr(app_module, "build_runtime", build) - monkeypatch.setattr(app_module, "configure_logging", lambda: configured.append(True)) + monkeypatch.setattr(lifespan_module, "build_runtime", build) + monkeypatch.setattr(lifespan_module, "configure_logging", lambda: configured.append(True)) application = app_module.create_app(selected_settings) # type: ignore[arg-type] async with application.router.lifespan_context(application): @@ -80,8 +81,8 @@ async def test_lifespan_closes_runtime_when_worker_task_fails( async def build(_settings: object) -> FakeRuntime: return runtime - monkeypatch.setattr(app_module, "build_runtime", build) - monkeypatch.setattr(app_module, "configure_logging", lambda: None) + monkeypatch.setattr(lifespan_module, "build_runtime", build) + monkeypatch.setattr(lifespan_module, "configure_logging", lambda: None) application = app_module.create_app(SimpleNamespace()) # type: ignore[arg-type] with pytest.raises(RuntimeError, match="worker failed"): @@ -100,8 +101,8 @@ async def test_lifespan_propagates_runtime_startup_failure_without_starting_work async def fail_build(_settings: object) -> None: raise RuntimeError("database unavailable") - monkeypatch.setattr(app_module, "build_runtime", fail_build) - monkeypatch.setattr(app_module, "configure_logging", lambda: configured.append(True)) + monkeypatch.setattr(lifespan_module, "build_runtime", fail_build) + monkeypatch.setattr(lifespan_module, "configure_logging", lambda: configured.append(True)) application = app_module.create_app(SimpleNamespace()) # type: ignore[arg-type] with pytest.raises(RuntimeError, match="database unavailable"): diff --git a/tests/test_codegraph.py b/tests/test_codegraph.py index b30fced..7d866b7 100644 --- a/tests/test_codegraph.py +++ b/tests/test_codegraph.py @@ -2,7 +2,7 @@ from pathlib import Path import pytest -from agentci.codegraph import CodeGraph, CodeGraphError +from agentci.integrations.codegraph import CodeGraph, CodeGraphError class FakeProcess: @@ -27,7 +27,7 @@ async def test_initializes_incomplete_index_and_excludes_it_from_git( return FakeProcess() monkeypatch.setattr( - "agentci.codegraph.asyncio.create_subprocess_exec", + "agentci.integrations.codegraph.asyncio.create_subprocess_exec", create_subprocess_exec, ) @@ -53,7 +53,7 @@ async def test_syncs_an_existing_index_without_duplicating_exclude( return FakeProcess() monkeypatch.setattr( - "agentci.codegraph.asyncio.create_subprocess_exec", + "agentci.integrations.codegraph.asyncio.create_subprocess_exec", create_subprocess_exec, ) @@ -86,7 +86,7 @@ async def test_handles_exclude_file_boundaries( return FakeProcess() monkeypatch.setattr( - "agentci.codegraph.asyncio.create_subprocess_exec", + "agentci.integrations.codegraph.asyncio.create_subprocess_exec", create_subprocess_exec, ) @@ -103,7 +103,7 @@ async def test_reports_missing_executable(tmp_path: Path, monkeypatch: pytest.Mo raise FileNotFoundError(2, "No such file or directory", "codegraph") monkeypatch.setattr( - "agentci.codegraph.asyncio.create_subprocess_exec", + "agentci.integrations.codegraph.asyncio.create_subprocess_exec", create_subprocess_exec, ) @@ -125,7 +125,7 @@ async def test_reports_nonzero_exit_with_bounded_non_utf8_stderr( return FakeProcess(returncode=7, stderr=stderr) monkeypatch.setattr( - "agentci.codegraph.asyncio.create_subprocess_exec", + "agentci.integrations.codegraph.asyncio.create_subprocess_exec", create_subprocess_exec, ) diff --git a/tests/test_config.py b/tests/test_config.py index dddacef..9ba8381 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -4,7 +4,7 @@ from pathlib import Path import pytest from pydantic import ValidationError -from agentci.config import Settings +from agentci.config.settings import Settings @pytest.fixture(autouse=True) diff --git a/tests/test_context.py b/tests/test_context.py index 6e0fdb0..f308952 100644 --- a/tests/test_context.py +++ b/tests/test_context.py @@ -1,7 +1,8 @@ from typing import cast from agentci.engine.repository import Repository -from agentci.gitea import CommentInfo, Gitea, IssueInfo, PullRequestInfo +from agentci.integrations.gitea.client import Gitea +from agentci.integrations.gitea.models import CommentInfo, IssueInfo, PullRequestInfo from agentci.workflows.context import build_issue_context, build_pull_request_context diff --git a/tests/test_development.py b/tests/test_development.py index 65260c8..0ca47ea 100644 --- a/tests/test_development.py +++ b/tests/test_development.py @@ -3,7 +3,7 @@ from pathlib import Path import pytest -from agentci.development import ( +from agentci.integrations.development import ( DevelopmentEnvironment, DevelopmentEnvironmentError, ) @@ -169,7 +169,7 @@ async def test_wraps_subprocess_start_error( raise OSError("exec unavailable") monkeypatch.setattr( - "agentci.development.asyncio.create_subprocess_exec", + "agentci.integrations.development.asyncio.create_subprocess_exec", create_subprocess_exec, ) diff --git a/tests/test_git.py b/tests/test_git.py index 142983e..63354b5 100644 --- a/tests/test_git.py +++ b/tests/test_git.py @@ -5,7 +5,7 @@ from typing import Any import pytest -from agentci.git import Git, GitError +from agentci.integrations.git import Git, GitError class FakeProcess: diff --git a/tests/test_gitea.py b/tests/test_gitea.py index aede42b..15488a4 100644 --- a/tests/test_gitea.py +++ b/tests/test_gitea.py @@ -1,17 +1,18 @@ import json -from collections.abc import AsyncIterator, Callable, Coroutine +from collections.abc import AsyncGenerator, Callable, Coroutine from contextlib import asynccontextmanager import httpx import pytest -from agentci.gitea import CommentInfo, Gitea, GiteaError, IssueInfo, PullRequestInfo +from agentci.integrations.gitea.client import Gitea, GiteaError +from agentci.integrations.gitea.models import CommentInfo, IssueInfo, PullRequestInfo Handler = Callable[[httpx.Request], Coroutine[None, None, httpx.Response]] @asynccontextmanager -async def gitea_client(handler: Handler, *, retries: int = 3) -> AsyncIterator[Gitea]: +async def gitea_client(handler: Handler, *, retries: int = 3) -> AsyncGenerator[Gitea]: client = Gitea( "https://gitea.example/", "secret", @@ -232,7 +233,7 @@ async def test_nonretryable_status_fails_once(status: int, monkeypatch: pytest.M async def sleep(delay: int) -> None: sleeps.append(delay) - monkeypatch.setattr("agentci.gitea.asyncio.sleep", sleep) + monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep) async with gitea_client(handler) as client: with pytest.raises( GiteaError, @@ -261,7 +262,7 @@ async def test_retryable_status_recovers_after_backoff( async def sleep(delay: int) -> None: sleeps.append(delay) - monkeypatch.setattr("agentci.gitea.asyncio.sleep", sleep) + monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep) async with gitea_client(handler) as client: assert await client.default_branch("org", "repo") == "main" @@ -283,7 +284,7 @@ async def test_retryable_status_exhaustion_uses_exponential_backoff( async def sleep(delay: int) -> None: sleeps.append(delay) - monkeypatch.setattr("agentci.gitea.asyncio.sleep", sleep) + monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep) async with gitea_client(handler) as client: with pytest.raises( GiteaError, @@ -309,7 +310,7 @@ async def test_transport_failure_retries_and_recovers(monkeypatch: pytest.Monkey async def sleep(delay: int) -> None: sleeps.append(delay) - monkeypatch.setattr("agentci.gitea.asyncio.sleep", sleep) + monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep) async with gitea_client(handler) as client: assert await client.default_branch("org", "repo") == "main" @@ -331,7 +332,7 @@ async def test_transport_failure_exhaustion_preserves_cause( async def sleep(delay: int) -> None: sleeps.append(delay) - monkeypatch.setattr("agentci.gitea.asyncio.sleep", sleep) + monkeypatch.setattr("agentci.integrations.gitea.client.asyncio.sleep", sleep) async with gitea_client(handler, retries=2) as client: with pytest.raises( GiteaError, diff --git a/tests/test_health.py b/tests/test_health.py index 8e3d763..82e2c60 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -4,7 +4,7 @@ import pytest from fastapi import FastAPI from httpx import ASGITransport, AsyncClient, Response -from agentci.health import router +from agentci.api.routes.health import router class Provider: diff --git a/tests/test_logging.py b/tests/test_logging.py index a7c38b6..3101391 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -6,7 +6,7 @@ from datetime import UTC, datetime import pytest -from agentci.logging import JsonFormatter, configure_logging +from agentci.observability.logging import JsonFormatter, configure_logging @pytest.fixture(autouse=True) diff --git a/tests/test_opencode.py b/tests/test_opencode.py index 1939533..7370ce8 100644 --- a/tests/test_opencode.py +++ b/tests/test_opencode.py @@ -5,7 +5,7 @@ from pathlib import Path import httpx import pytest -from agentci.opencode import OpenCode, OpenCodeError +from agentci.integrations.opencode.client import OpenCode, OpenCodeError from agentci.workflows.model import AgentResult API_DOCUMENT = { diff --git a/tests/test_opencode_abort.py b/tests/test_opencode_abort.py index 92ac9bb..c35106c 100644 --- a/tests/test_opencode_abort.py +++ b/tests/test_opencode_abort.py @@ -3,7 +3,7 @@ from pathlib import Path import httpx import pytest -from agentci.opencode import OpenCode, OpenCodeError +from agentci.integrations.opencode.client import OpenCode, OpenCodeError def client(tmp_path: Path, status: int) -> OpenCode: diff --git a/tests/test_opencode_support.py b/tests/test_opencode_support.py index e5f5129..e5f05ae 100644 --- a/tests/test_opencode_support.py +++ b/tests/test_opencode_support.py @@ -1,6 +1,6 @@ import pytest -from agentci.opencode import api_contract_ready, models_ready +from agentci.integrations.opencode.readiness import api_contract_ready, models_ready def test_api_contract_requires_session_message_and_abort_routes() -> None: diff --git a/tests/test_prompts.py b/tests/test_prompts.py index 96f35d9..56884e3 100644 --- a/tests/test_prompts.py +++ b/tests/test_prompts.py @@ -5,8 +5,8 @@ import pytest from pydantic import BaseModel import agentci.prompts -from agentci.opencode import load_schema -from agentci.prompts import PromptLibrary +from agentci.integrations.opencode.schemas import load_schema +from agentci.prompts.library import PromptLibrary from agentci.workflows.model import ( AgentResult, DiscussionReply, diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 3fc5dc6..d6dcfc3 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -4,9 +4,9 @@ from unittest.mock import AsyncMock, Mock import pytest -import agentci.runtime as runtime_module -from agentci.config import Settings -from agentci.runtime import Runtime +import agentci.application.runtime as runtime_module +from agentci.application.runtime import Runtime +from agentci.config.settings import Settings class ClosingClient: @@ -90,7 +90,7 @@ async def test_build_runtime_wires_components_without_starting_real_clients( git = SimpleNamespace() opencode = SimpleNamespace() development = SimpleNamespace() - prompts = SimpleNamespace() + prompts = SimpleNamespace(schemas_dir=tmp_path / "schemas") services = SimpleNamespace() worker = SimpleNamespace() repository_constructor = Mock(return_value=repository) @@ -112,11 +112,7 @@ async def test_build_runtime_wires_components_without_starting_real_clients( built = await runtime_module.build_runtime(settings) - assert runtime_module.__file__ is not None - package_dir = Path(runtime_module.__file__).parent - repository_constructor.assert_called_once_with( - settings.database_path, package_dir / "migrations" - ) + repository_constructor.assert_called_once_with(settings.database_path) initialize.assert_awaited_once_with() gitea_constructor.assert_called_once_with("https://gitea.example", "token") git_constructor.assert_called_once_with( @@ -131,7 +127,7 @@ async def test_build_runtime_wires_components_without_starting_real_clients( base_url="https://opencode.example", username=settings.opencode_server_username, password="password", - schemas_dir=package_dir / "prompts" / "schemas", + schemas_dir=prompts.schemas_dir, health_directory=settings.workspaces_dir, required_models=( (settings.plan_model, settings.plan_variant), diff --git a/tests/test_webhook.py b/tests/test_webhook.py index 722e96f..aede8f3 100644 --- a/tests/test_webhook.py +++ b/tests/test_webhook.py @@ -7,8 +7,12 @@ import pytest from fastapi import FastAPI from httpx import ASGITransport, AsyncClient, Response +from agentci.api.routes.webhook import router from agentci.engine.model import IncomingCommand -from agentci.webhook import _event_from_payload, router, valid_signature +from agentci.integrations.gitea.webhooks import ( + incoming_command_from_payload, + valid_signature, +) class FakeRepository: @@ -167,7 +171,7 @@ def test_payload_parser_supports_owner_username_and_pull_request() -> None: value = payload("/agent plan", is_pull=True) value["repository"]["owner"] = {"username": "fallback-owner"} - event = _event_from_payload("delivery", value) + event = incoming_command_from_payload("delivery", value) assert event is not None assert event.repo_owner == "fallback-owner" diff --git a/tests/test_worker.py b/tests/test_worker.py index bc97fdb..7b59346 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -5,6 +5,8 @@ from types import SimpleNamespace import pytest +from agentci.application.worker.errors import safe_error +from agentci.application.worker.runner import Worker from agentci.engine.events import ( CommentLinked, JobCompleted, @@ -30,8 +32,7 @@ from agentci.engine.model import ( ) from agentci.engine.reducer import render_job_comment from agentci.engine.repository import Repository -from agentci.gitea import CommentInfo -from agentci.worker import Worker, _safe_error +from agentci.integrations.gitea.models import CommentInfo from agentci.workflows.render import JobRejected MIGRATIONS = Path(__file__).parents[1] / "src" / "agentci" / "migrations" @@ -408,7 +409,7 @@ async def test_execute_completes_with_workflow_comment( async def dispatch(_job: Job, _run: object, _services: object) -> str: return "final workflow body" - monkeypatch.setattr("agentci.worker.dispatch", dispatch) + monkeypatch.setattr("agentci.application.worker.execution.dispatch", dispatch) execute = task() await value._execute(execute, current) @@ -432,7 +433,7 @@ async def test_execute_records_expected_rejection( async def dispatch(_job: Job, _run: object, _services: object) -> str: raise JobRejected("pull request is closed") - monkeypatch.setattr("agentci.worker.dispatch", dispatch) + monkeypatch.setattr("agentci.application.worker.execution.dispatch", dispatch) await make_worker(tmp_path, repository)._execute(task(), current) @@ -452,7 +453,7 @@ async def test_execute_records_failure_at_latest_persisted_stage( repository.job = replace(repository.job, stage="cloning") raise RuntimeError("provider\nfailed") - monkeypatch.setattr("agentci.worker.dispatch", dispatch) + monkeypatch.setattr("agentci.application.worker.execution.dispatch", dispatch) await make_worker(tmp_path, repository)._execute(task(), current) @@ -471,7 +472,7 @@ async def test_execute_marks_running_job_interrupted_after_restart( async def unexpected_dispatch(_job: Job, _run: object, _services: object) -> str: pytest.fail("running jobs must not be dispatched again") - monkeypatch.setattr("agentci.worker.dispatch", unexpected_dispatch) + monkeypatch.setattr("agentci.application.worker.execution.dispatch", unexpected_dispatch) execute = task() await make_worker(tmp_path, repository)._execute(execute, current) @@ -642,7 +643,7 @@ async def test_abort_without_persisted_sessions_is_noop(tmp_path: Path) -> None: def test_safe_error_is_single_line_and_bounded() -> None: - value = _safe_error(RuntimeError("bad\n" + "x" * 2000)) + value = safe_error(RuntimeError("bad\n" + "x" * 2000)) assert "\n" not in value assert len(value) == 1000 diff --git a/tests/test_workflow_implementation.py b/tests/test_workflow_implementation.py index 1d66552..2c82cde 100644 --- a/tests/test_workflow_implementation.py +++ b/tests/test_workflow_implementation.py @@ -7,7 +7,7 @@ import pytest from agentci.engine.model import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus from agentci.engine.run import JobRun -from agentci.gitea import CommentInfo, IssueInfo, PullRequestInfo +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 diff --git a/tests/test_workflow_plan.py b/tests/test_workflow_plan.py index 9646037..e0cbc79 100644 --- a/tests/test_workflow_plan.py +++ b/tests/test_workflow_plan.py @@ -8,7 +8,7 @@ from pydantic import BaseModel from agentci.engine.model import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus from agentci.engine.run import JobRun -from agentci.gitea import CommentInfo, IssueInfo, PullRequestInfo +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 diff --git a/tests/test_workflow_pull_request.py b/tests/test_workflow_pull_request.py index 1aa00ce..8f34e51 100644 --- a/tests/test_workflow_pull_request.py +++ b/tests/test_workflow_pull_request.py @@ -8,7 +8,7 @@ from pydantic import BaseModel from agentci.engine.model import Job, JobKind, Workflow, WorkflowKind, WorkflowStatus from agentci.engine.run import JobRun -from agentci.gitea import CommentInfo, IssueInfo, PullRequestInfo +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