StanPonomarev/agentci
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add structured step logging and error reporting

Browse Source
This commit is contained in:
StanPonomarev
2026-07-19 12:50:09 +02:00
parent e9b895366d
commit 276b5590b5
11 changed files with 289 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agentci/adapters/codex.py
+28 -1
View File
@@ -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)
src/agentci/adapters/database.py
+10
View File
@@ -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)
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)
src/agentci/adapters/git.py
+26
View File
@@ -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,6 +115,7 @@ class GitClient:
"AGENTCI_GIT_PASSWORD": self.token,
}
)
try:
process = await asyncio.create_subprocess_exec(
"git",
*args,
@@ -117,7 +125,25 @@ class GitClient:
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)
src/agentci/adapters/gitea.py
+29
View File
@@ -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)
src/agentci/adapters/job_store.py
+13
View File
@@ -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"}:
src/agentci/api/webhook.py
+26
View File
@@ -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,
src/agentci/app.py
+31 -2
View File
@@ -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()
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()
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()
src/agentci/container.py
+8 -1
View File
@@ -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
src/agentci/logging.py
+22 -3
View File
@@ -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,11 +29,16 @@ 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)
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)
@@ -29,4 +49,3 @@ def configure_logging() -> None:
root.handlers.clear()
root.addHandler(handler)
root.setLevel(logging.INFO)
src/agentci/worker.py
+56 -14
View File
@@ -31,9 +31,15 @@ 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()
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()
@@ -41,6 +47,11 @@ class Worker:
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)
src/agentci/workflows/dispatcher.py
+16 -1
View File
@@ -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:
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)