diff --git a/src/agentci/adapters/codex.py b/src/agentci/adapters/codex.py index 7d3b28b..5e8e3ac 100644 --- a/src/agentci/adapters/codex.py +++ b/src/agentci/adapters/codex.py @@ -2,14 +2,17 @@ from __future__ import annotations import asyncio import json +import logging import os import tempfile from pathlib import Path +from time import monotonic from typing import TypeVar from pydantic import BaseModel T = TypeVar("T", bound=BaseModel) +log = logging.getLogger(__name__) class CodexError(RuntimeError): @@ -113,6 +116,9 @@ class CodexClient: async def _invoke( self, args: list[str], prompt: str, result_type: type[T] ) -> tuple[str | None, T]: + started = monotonic() + operation = "codex.resume" if "resume" in args else "codex.start" + log.info("Codex turn started", extra={"operation": operation}) file_descriptor, output_name = tempfile.mkstemp( prefix="agentci-codex-", suffix=".json" ) @@ -134,14 +140,31 @@ class CodexClient: except TimeoutError as exc: process.terminate() await process.wait() + log.exception( + "Codex turn timed out", + extra={"operation": operation, "duration_ms": _elapsed_ms(started)}, + ) raise CodexError(f"Codex turn exceeded {self.timeout_seconds} seconds") from exc if process.returncode: detail = stderr.decode(errors="replace").strip() + log.error( + "Codex turn failed", + extra={"operation": operation, "duration_ms": _elapsed_ms(started)}, + ) raise CodexError(f"Codex exited with {process.returncode}: {detail[-2000:]}") session_id = _session_id(stdout.decode(errors="replace")) text = output_path.read_text() # noqa: ASYNC240 - tiny host-owned result file - return session_id, result_type.model_validate_json(text) + result = result_type.model_validate_json(text) + log.info( + "Codex turn completed", + extra={"operation": operation, "duration_ms": _elapsed_ms(started)}, + ) + return session_id, result except (OSError, ValueError) as exc: + log.exception( + "Codex result could not be processed", + extra={"operation": operation, "duration_ms": _elapsed_ms(started)}, + ) raise CodexError(f"Invalid Codex result: {exc}") from exc finally: output_path.unlink(missing_ok=True) # noqa: ASYNC240 @@ -162,3 +185,7 @@ def _session_id(output: str) -> str | None: if event.get("type") == "thread.started": return str(event["thread_id"]) return None + + +def _elapsed_ms(started: float) -> int: + return round((monotonic() - started) * 1000) diff --git a/src/agentci/adapters/database.py b/src/agentci/adapters/database.py index fc76f4d..4c858d8 100644 --- a/src/agentci/adapters/database.py +++ b/src/agentci/adapters/database.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import sqlite3 from collections.abc import Callable from datetime import UTC, datetime @@ -7,6 +8,7 @@ from pathlib import Path from typing import TypeVar T = TypeVar("T") +log = logging.getLogger(__name__) def now() -> str: @@ -19,8 +21,16 @@ class Database: self.migrations_dir = migrations_dir async def initialize(self) -> None: + log.info("database initialization started", extra={"operation": "database.initialize"}) self.database_path.parent.mkdir(parents=True, exist_ok=True) - await self._run(self._initialize_sync) + try: + await self._run(self._initialize_sync) + except Exception: + log.exception( + "database initialization failed", extra={"operation": "database.initialize"} + ) + raise + log.info("database initialization completed", extra={"operation": "database.initialize"}) def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self.database_path, timeout=30) diff --git a/src/agentci/adapters/git.py b/src/agentci/adapters/git.py index 628d523..8bd442d 100644 --- a/src/agentci/adapters/git.py +++ b/src/agentci/adapters/git.py @@ -1,8 +1,12 @@ from __future__ import annotations import asyncio +import logging import os from pathlib import Path +from time import monotonic + +log = logging.getLogger(__name__) class GitError(RuntimeError): @@ -98,6 +102,9 @@ class GitClient: cwd: Path, authenticated: bool = False, ) -> str: + operation = f"git.{args[0]}" + started = monotonic() + log.info("git step started", extra={"operation": operation}) environment = os.environ.copy() if authenticated: environment.update( @@ -108,16 +115,35 @@ class GitClient: "AGENTCI_GIT_PASSWORD": self.token, } ) - process = await asyncio.create_subprocess_exec( - "git", - *args, - cwd=cwd, - env=environment, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - stdout, stderr = await process.communicate() + try: + process = await asyncio.create_subprocess_exec( + "git", + *args, + cwd=cwd, + env=environment, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + except OSError as exc: + log.exception( + "git step could not start", + extra={"operation": operation, "duration_ms": _elapsed_ms(started)}, + ) + raise GitError(f"Could not run git {args[0]}: {exc}") from exc if process.returncode: detail = stderr.decode(errors="replace").strip() + log.error( + "git step failed", + extra={"operation": operation, "duration_ms": _elapsed_ms(started)}, + ) raise GitError(f"git {args[0]} failed: {detail[-1000:]}") + log.info( + "git step completed", + extra={"operation": operation, "duration_ms": _elapsed_ms(started)}, + ) return stdout.decode(errors="replace") + + +def _elapsed_ms(started: float) -> int: + return round((monotonic() - started) * 1000) diff --git a/src/agentci/adapters/gitea.py b/src/agentci/adapters/gitea.py index 7375e3f..2e20e22 100644 --- a/src/agentci/adapters/gitea.py +++ b/src/agentci/adapters/gitea.py @@ -1,6 +1,8 @@ from __future__ import annotations import asyncio +import logging +from time import monotonic from typing import Any import httpx @@ -12,6 +14,8 @@ from agentci.adapters.gitea_models import ( RepositoryInfo, ) +log = logging.getLogger(__name__) + class GiteaError(RuntimeError): pass @@ -138,23 +142,45 @@ class GiteaClient: **kwargs: Any, ) -> httpx.Response: for attempt in range(self.retries): + started = monotonic() + extra = { + "operation": "gitea.request", + "method": method, + "path": path, + "attempt": attempt + 1, + } + log.info("Gitea request started", extra=extra) try: response = await self.client.request(method, path, **kwargs) except httpx.RequestError as exc: + log.warning( + "Gitea request transport failure", + extra={**extra, "duration_ms": _elapsed_ms(started)}, + exc_info=exc, + ) if attempt == self.retries - 1: raise GiteaError(f"Gitea request failed: {method} {path}") from exc await asyncio.sleep(2**attempt) continue + response_extra = { + **extra, + "status_code": response.status_code, + "duration_ms": _elapsed_ms(started), + } if response.status_code == 404 and allow_not_found: + log.info("Gitea request completed", extra=response_extra) return response if response.status_code not in {429, 500, 502, 503, 504}: try: response.raise_for_status() except httpx.HTTPStatusError as exc: + log.error("Gitea request rejected", extra=response_extra) raise GiteaError( f"Gitea returned {response.status_code} for {method} {path}" ) from exc + log.info("Gitea request completed", extra=response_extra) return response + log.warning("Gitea request will be retried", extra=response_extra) if attempt < self.retries - 1: await asyncio.sleep(2**attempt) raise GiteaError(f"Gitea remained unavailable for {method} {path}") @@ -168,3 +194,6 @@ def _comment(data: dict[str, Any]) -> CommentInfo: created_at=data.get("created_at") or "", ) + +def _elapsed_ms(started: float) -> int: + return round((monotonic() - started) * 1000) diff --git a/src/agentci/adapters/job_store.py b/src/agentci/adapters/job_store.py index dce2566..3600306 100644 --- a/src/agentci/adapters/job_store.py +++ b/src/agentci/adapters/job_store.py @@ -1,10 +1,13 @@ from __future__ import annotations +import logging import sqlite3 from agentci.adapters.database import Database, now from agentci.domain.models import Job, JobKind, JobStatus +log = logging.getLogger(__name__) + class JobStore(Database): async def record_delivery(self, delivery_id: str, comment_id: int) -> bool: @@ -101,6 +104,16 @@ class JobStore(Database): if workflow_id is not None: updates["workflow_id"] = workflow_id await self._update("jobs", job_id, updates) + log.info( + "job state updated", + extra={ + "operation": "job.update", + "job_id": job_id, + "workflow_id": workflow_id, + "stage": stage, + "status_code": status.value if status is not None else None, + }, + ) async def set_job_comment(self, job_id: str, column: str, comment_id: int) -> None: if column not in {"accepted_comment_id", "started_comment_id"}: diff --git a/src/agentci/api/webhook.py b/src/agentci/api/webhook.py index 9a902d6..8e3d762 100644 --- a/src/agentci/api/webhook.py +++ b/src/agentci/api/webhook.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import hmac import json +import logging from typing import Any from uuid import uuid4 @@ -12,6 +13,7 @@ from agentci.domain.commands import CommandError, parse_command, resolve_job_kin from agentci.domain.models import CommandEvent, Job, JobStatus router = APIRouter() +log = logging.getLogger(__name__) SUPPORTED_EVENTS = { "issue_comment", "pull_request_comment", @@ -25,11 +27,19 @@ async def webhook(request: Request) -> Response: body = await request.body() signature = request.headers.get("X-Gitea-Signature", "") if not valid_signature(container.settings.webhook_secret, body, signature): + log.warning( + "webhook signature rejected", + extra={"operation": "webhook.verify", "path": request.url.path}, + ) raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid webhook signature") event_name = request.headers.get("X-Gitea-Event-Type") or request.headers.get( "X-Gitea-Event", "" ) if event_name not in SUPPORTED_EVENTS: + log.info( + "unsupported webhook ignored", + extra={"operation": "webhook.filter", "stage": event_name or "missing"}, + ) return Response(status_code=status.HTTP_204_NO_CONTENT) try: payload = json.loads(body) @@ -38,8 +48,14 @@ async def webhook(request: Request) -> Response: payload, ) except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + log.warning( + "webhook payload rejected", + extra={"operation": "webhook.parse", "stage": event_name}, + exc_info=exc, + ) raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid webhook payload") from exc if event is None or event.requester == container.settings.bot_username: + log.info("webhook ignored", extra={"operation": "webhook.filter", "stage": event_name}) return Response(status_code=status.HTTP_204_NO_CONTENT) return await _handle_command(container, event) @@ -50,12 +66,15 @@ def valid_signature(secret: bytes, body: bytes, signature: str) -> bool: async def _handle_command(container: Any, event: CommandEvent) -> Response: + extra = {"operation": "command.handle", "target": event.target_key} if not event.body.strip().startswith("/agent"): return Response(status_code=status.HTTP_204_NO_CONTENT) + log.info("agent command received", extra=extra) permitted = await container.gitea.has_write_permission( event.repo_owner, event.repo_name, event.requester ) if not permitted: + log.warning("agent command rejected: insufficient permission", extra=extra) if await container.storage.record_delivery(event.delivery_id, event.comment_id): await container.gitea.create_comment( event.repo_owner, @@ -67,6 +86,7 @@ async def _handle_command(container: Any, event: CommandEvent) -> Response: try: command = parse_command(event.body) except CommandError as exc: + log.info("agent command rejected: invalid syntax", extra=extra) if await container.storage.record_delivery(event.delivery_id, event.comment_id): await container.gitea.create_comment( event.repo_owner, event.repo_name, event.issue_number, str(exc) @@ -77,6 +97,7 @@ async def _handle_command(container: Any, event: CommandEvent) -> Response: try: kind = resolve_job_kind(command, is_pull_request=event.is_pull_request) except CommandError as exc: + log.info("agent command rejected: invalid target", extra=extra) if await container.storage.record_delivery(event.delivery_id, event.comment_id): await container.gitea.create_comment( event.repo_owner, event.repo_name, event.issue_number, str(exc) @@ -97,7 +118,12 @@ async def _handle_command(container: Any, event: CommandEvent) -> Response: stage="queued", ) if not await container.storage.enqueue(event.delivery_id, job): + log.info("duplicate command ignored", extra={**extra, "job_id": job.id}) return Response(status_code=status.HTTP_200_OK) + log.info( + "agent job queued", + extra={**extra, "job_id": job.id, "stage": job.kind.value}, + ) comment_id = await container.gitea.create_comment( event.repo_owner, event.repo_name, diff --git a/src/agentci/app.py b/src/agentci/app.py index 3c3e8bd..ab3871f 100644 --- a/src/agentci/app.py +++ b/src/agentci/app.py @@ -1,10 +1,12 @@ from __future__ import annotations import asyncio +import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from fastapi import FastAPI +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse from agentci.api.health import router as health_router from agentci.api.webhook import router as webhook_router @@ -12,6 +14,7 @@ from agentci.config import Settings from agentci.container import build_container from agentci.logging import configure_logging +log = logging.getLogger(__name__) def create_app(settings: Settings | None = None) -> FastAPI: selected_settings = settings or Settings() @@ -19,22 +22,48 @@ def create_app(settings: Settings | None = None) -> FastAPI: @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: configure_logging() - container = await build_container(selected_settings) + log.info("service startup started", extra={"operation": "service.startup"}) + try: + container = await build_container(selected_settings) + except Exception: + log.exception("service startup failed", extra={"operation": "service.startup"}) + raise app.state.container = container stop = asyncio.Event() worker_task = asyncio.create_task(container.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() - await worker_task - await container.close() + try: + await worker_task + finally: + await container.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/container.py b/src/agentci/container.py index 341761f..087e902 100644 --- a/src/agentci/container.py +++ b/src/agentci/container.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging from dataclasses import dataclass from pathlib import Path @@ -14,6 +15,7 @@ from agentci.workflows.common import Dependencies from agentci.workflows.context import ContextBuilder from agentci.workflows.dispatcher import Dispatcher +log = logging.getLogger(__name__) @dataclass(frozen=True) class Container: @@ -25,10 +27,13 @@ class Container: worker: Worker async def close(self) -> None: + log.info("container shutdown started", extra={"operation": "container.close"}) await self.gitea.close() + log.info("container shutdown completed", extra={"operation": "container.close"}) async def build_container(settings: Settings) -> Container: + log.info("container initialization started", extra={"operation": "container.build"}) package_dir = Path(__file__).parent settings.data_dir.mkdir(parents=True, exist_ok=True) settings.workspaces_dir.mkdir(parents=True, exist_ok=True) @@ -68,4 +73,6 @@ async def build_container(settings: Settings) -> Container: dispatcher=dispatcher, poll_seconds=settings.worker_poll_seconds, ) - return Container(settings, storage, gitea, git, codex, worker) + container = Container(settings, storage, gitea, git, codex, worker) + log.info("container initialization completed", extra={"operation": "container.build"}) + return container diff --git a/src/agentci/logging.py b/src/agentci/logging.py index 258ddc1..d32f9e6 100644 --- a/src/agentci/logging.py +++ b/src/agentci/logging.py @@ -2,9 +2,24 @@ from __future__ import annotations import json import logging +import traceback from datetime import UTC, datetime from typing import Any +LOG_FIELDS = ( + "operation", + "job_id", + "workflow_id", + "stage", + "target", + "method", + "path", + "status_code", + "attempt", + "item_count", + "duration_ms", +) + class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: @@ -14,12 +29,17 @@ class JsonFormatter(logging.Formatter): "logger": record.name, "message": record.getMessage(), } - for key in ("job_id", "stage", "target"): - if value := getattr(record, key, None): + for key in LOG_FIELDS: + if (value := getattr(record, key, None)) is not None: payload[key] = value if record.exc_info: payload["exception"] = self.formatException(record.exc_info) - return json.dumps(payload, ensure_ascii=False) + try: + return json.dumps(payload, ensure_ascii=False) + except (TypeError, ValueError): + payload["message"] = "Log record could not be serialized" + payload["exception"] = traceback.format_exc() + return json.dumps(payload, ensure_ascii=False) def configure_logging() -> None: @@ -29,4 +49,3 @@ def configure_logging() -> None: root.handlers.clear() root.addHandler(handler) root.setLevel(logging.INFO) - diff --git a/src/agentci/worker.py b/src/agentci/worker.py index 2044b00..1c68343 100644 --- a/src/agentci/worker.py +++ b/src/agentci/worker.py @@ -31,16 +31,27 @@ class Worker: self.poll_seconds = poll_seconds async def run(self, stop: asyncio.Event) -> None: + log.info("worker started", extra={"operation": "worker.run"}) await self._report_interrupted() - while not stop.is_set(): - if not await self.codex.login_ready(): - await self._wait(stop) - continue - job = await self.storage.claim_next() - if job is None: - await self._wait(stop) - continue - await self._run_job(job) + try: + while not stop.is_set(): + if not await self.codex.login_ready(): + log.warning( + "worker waiting for Codex authentication", + extra={"operation": "worker.poll"}, + ) + await self._wait(stop) + continue + job = await self.storage.claim_next() + if job is None: + await self._wait(stop) + continue + await self._run_job(job) + except Exception: + log.exception("worker stopped unexpectedly", extra={"operation": "worker.run"}) + raise + finally: + log.info("worker stopped", extra={"operation": "worker.run"}) async def _run_job(self, job: Job) -> None: extra = {"job_id": job.id, "target": job.target_key} @@ -65,34 +76,37 @@ class Worker: await self.storage.set_job_comment(job.id, "started_comment_id", comment_id) await self.dispatcher.dispatch(job) except JobRejected as exc: - await self.storage.update_job( - job.id, status=JobStatus.REJECTED, stage="rejected", error=str(exc) + await self._safe_update_job( + job, status=JobStatus.REJECTED, stage="rejected", error=str(exc) ) await self._safe_comment(job, f"Agent job `{job.id}` was rejected: {exc}") - log.info("job rejected", extra=extra) + log.info("job rejected", extra={**extra, "stage": "rejected"}) except Exception as exc: - failed_stage = await self.storage.job_stage(job.id) - await self.storage.update_job( - job.id, - status=JobStatus.FAILED, - stage="failed", - error=_safe_error(exc), + failed_stage = await self._safe_job_stage(job) + await self._safe_update_job( + job, status=JobStatus.FAILED, stage="failed", error=_safe_error(exc) ) - await self.storage.fail_job_workflow(job.id) + await self._safe_fail_workflow(job) await self._safe_comment( job, f"Agent job `{job.id}` failed during `{failed_stage}`: {_safe_error(exc)}", ) log.exception("job failed", extra={**extra, "stage": failed_stage}) else: - await self.storage.update_job( - job.id, status=JobStatus.SUCCEEDED, stage="completed" + await self._safe_update_job( + job, status=JobStatus.SUCCEEDED, stage="completed" ) log.info("job completed", extra=extra) async def _report_interrupted(self) -> None: - for job in await self.storage.recover_running(): - await self.storage.fail_job_workflow(job.id) + jobs = await self.storage.recover_running() + if jobs: + log.warning( + "recovering interrupted jobs", + extra={"operation": "worker.recover", "item_count": len(jobs)}, + ) + for job in jobs: + await self._safe_fail_workflow(job) await self._safe_comment( job, f"Agent job `{job.id}` failed because the service restarted during execution.", @@ -106,6 +120,34 @@ class Worker: except Exception: log.exception("could not publish job status", extra={"job_id": job.id}) + async def _safe_job_stage(self, job: Job) -> str: + try: + return await self.storage.job_stage(job.id) + except Exception: + log.exception("could not read failed job stage", extra={"job_id": job.id}) + return job.stage or "unknown" + + async def _safe_update_job( + self, + job: Job, + *, + status: JobStatus, + stage: str, + error: str | None = None, + ) -> None: + try: + await self.storage.update_job( + job.id, status=status, stage=stage, error=error + ) + except Exception: + log.exception("could not persist job status", extra={"job_id": job.id}) + + async def _safe_fail_workflow(self, job: Job) -> None: + try: + await self.storage.fail_job_workflow(job.id) + except Exception: + log.exception("could not mark workflow failed", extra={"job_id": job.id}) + 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/workflows/dispatcher.py b/src/agentci/workflows/dispatcher.py index 1d3b9f5..de93dd2 100644 --- a/src/agentci/workflows/dispatcher.py +++ b/src/agentci/workflows/dispatcher.py @@ -1,11 +1,15 @@ from __future__ import annotations +import logging + from agentci.domain.models import Job, JobKind from agentci.workflows.common import Dependencies from agentci.workflows.implement import ImplementWorkflow from agentci.workflows.plan import PlanWorkflow from agentci.workflows.pull_request import PullRequestWorkflow +log = logging.getLogger(__name__) + class Dispatcher: def __init__(self, dependencies: Dependencies) -> None: @@ -21,5 +25,16 @@ class Dispatcher: } async def dispatch(self, job: Job) -> None: - await self.handlers[job.kind](job) - + extra = { + "operation": "workflow.dispatch", + "job_id": job.id, + "target": job.target_key, + "stage": job.kind.value, + } + log.info("workflow dispatch started", extra=extra) + try: + await self.handlers[job.kind](job) + except Exception: + log.exception("workflow dispatch failed", extra=extra) + raise + log.info("workflow dispatch completed", extra=extra)