From 378e372a4bd842772cff9b7fb69656adea0f01ec Mon Sep 17 00:00:00 2001 From: agentci Date: Tue, 21 Jul 2026 14:13:58 +0000 Subject: [PATCH] agent: Implemented the explicit persisted webhook state machine. --- README.md | 19 +- src/agentci/adapters/gitea.py | 6 +- src/agentci/adapters/job_store.py | 345 ++++++++++--------- src/agentci/adapters/opencode.py | 25 +- src/agentci/adapters/state_persistence.py | 115 +++++++ src/agentci/api/webhook.py | 77 +---- src/agentci/container.py | 7 +- src/agentci/domain/events.py | 104 ++++++ src/agentci/domain/models.py | 1 + src/agentci/domain/state_machine.py | 189 ++++++++++ src/agentci/migrations/003_state_machine.sql | 96 ++++++ src/agentci/reporting.py | 80 +++++ src/agentci/state_machine.py | 33 ++ src/agentci/worker.py | 293 +++++++++------- src/agentci/workflows/change_set.py | 7 +- src/agentci/workflows/code_review.py | 8 +- src/agentci/workflows/common.py | 11 +- src/agentci/workflows/dispatcher.py | 10 +- src/agentci/workflows/implement.py | 21 +- src/agentci/workflows/plan.py | 29 +- src/agentci/workflows/pull_request.py | 32 +- tests/test_config.py | 3 +- tests/test_opencode_abort.py | 34 ++ tests/test_state_machine.py | 75 ++++ tests/test_storage.py | 162 ++++----- tests/test_webhook.py | 131 ++----- tests/test_worker.py | 144 ++------ 27 files changed, 1295 insertions(+), 762 deletions(-) create mode 100644 src/agentci/adapters/state_persistence.py create mode 100644 src/agentci/domain/events.py create mode 100644 src/agentci/domain/state_machine.py create mode 100644 src/agentci/migrations/003_state_machine.sql create mode 100644 src/agentci/reporting.py create mode 100644 src/agentci/state_machine.py create mode 100644 tests/test_opencode_abort.py create mode 100644 tests/test_state_machine.py diff --git a/README.md b/README.md index 56c0e07..ebfadf8 100644 --- a/README.md +++ b/README.md @@ -15,9 +15,10 @@ private OpenCode server on the same Docker network as Gitea. | PR | `/agent iterate [message]` | Resume an agent implementation and its reviewer once. | | PR | `/agent fix [message]` | Start a fresh one-shot fix session and push one commit. | -The requester must have Gitea `write`, `admin`, or `owner` permission on the repository. Each -accepted command gets one Gitea comment, which is updated as the job moves from queued to started -and then to its final result or failure. Remaining review findings are included in the final update. +The requester must have Gitea `write`, `admin`, or `owner` permission on the repository. Commands +are durably sequenced when their webhook arrives, then authorized and executed in that receive +order. Each command gets one Gitea comment, which is reconciled asynchronously through queued, +running, and terminal states. Deleted comments are rediscovered by their hidden marker or recreated. ## Deploy @@ -127,11 +128,17 @@ The `opencode_home` volume contains provider authentication, OpenCode's database sessions. Tea's Gitea token configuration is regenerated in an ephemeral tmpfs and is not copied to `opencode_home`. Back up both persistent volumes together. +SQLite stores the current job state, an idempotent event inbox, and durable listener tasks. State +transitions and workflow creation/linking commit atomically; timestamps are storage metadata rather +than reducer state. Control effects retry with bounded backoff. A delayed authorization blocks later +workflow execution but not later control work. + The OpenCode migration tags existing workflows as Codex-owned and preserves their session IDs for rollback, but OpenCode refuses to resume them. Follow-up commands against those workflows ask for a -new plan or implementation. Queued jobs survive restart. An in-progress job is aborted in OpenCode -and marked failed instead of being replayed because a partial model turn may already have changed -files. Git pushes are never forced. +new plan or implementation. Queued jobs survive restart. A restart before `JobStarted` returns the +task to the queue; after `JobStarted`, the job is failed, its sessions are aborted, and execution is +never replayed because a partial model turn may already have changed files. Git pushes are never +forced. ## Development diff --git a/src/agentci/adapters/gitea.py b/src/agentci/adapters/gitea.py index a766ed9..e2b3572 100644 --- a/src/agentci/adapters/gitea.py +++ b/src/agentci/adapters/gitea.py @@ -105,12 +105,14 @@ class GiteaClient: ) return int(response.json()["id"]) - async def update_comment(self, owner: str, repo: str, comment_id: int, body: str) -> None: - await self._request( + async def update_comment(self, owner: str, repo: str, comment_id: int, body: str) -> bool: + response = await self._request( "PATCH", f"/repos/{owner}/{repo}/issues/comments/{comment_id}", json={"body": body}, + allow_not_found=True, ) + return response.status_code != 404 async def create_pull_request( self, diff --git a/src/agentci/adapters/job_store.py b/src/agentci/adapters/job_store.py index 4d11873..77d9e4e 100644 --- a/src/agentci/adapters/job_store.py +++ b/src/agentci/adapters/job_store.py @@ -1,149 +1,177 @@ from __future__ import annotations -import logging import sqlite3 +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta from agentci.adapters.database import Database, now -from agentci.domain.models import Job, JobKind, JobStatus +from agentci.adapters.state_persistence import ( + insert_event as _insert_event, +) +from agentci.adapters.state_persistence import ( + insert_state as _insert_state, +) +from agentci.adapters.state_persistence import ( + insert_tasks as _insert_tasks, +) +from agentci.adapters.state_persistence import ( + insert_workflow as _insert_workflow, +) +from agentci.adapters.state_persistence import ( + optional_state as _optional_state, +) +from agentci.adapters.state_persistence import ( + replace_state as _replace_state, +) +from agentci.adapters.state_persistence import ( + state as _state, +) +from agentci.adapters.state_persistence import ( + state_from_row as _state_from_row, +) +from agentci.domain.events import CommandReceived, JobEvent, WorkflowCreated +from agentci.domain.models import CommandEvent +from agentci.domain.state_machine import JobState, next_state -log = logging.getLogger(__name__) + +@dataclass(frozen=True) +class EvolveResult: + state: JobState + duplicate: bool + + +@dataclass(frozen=True) +class ListenerTask: + id: int + job_id: str + source_event_id: str + listener: str + queue: str + attempts: int class JobStore(Database): - async def record_delivery(self, delivery_id: str, comment_id: int) -> bool: - def record(connection: sqlite3.Connection) -> bool: - try: - connection.execute( - "INSERT INTO deliveries VALUES (?, ?, ?)", - (delivery_id, comment_id, now()), - ) - except sqlite3.IntegrityError: - return False - return True - - return await self._run(record) - - async def enqueue(self, delivery_id: str, job: Job) -> bool: - return await self._run(lambda connection: self._enqueue(connection, delivery_id, job)) - - @staticmethod - def _enqueue(connection: sqlite3.Connection, delivery_id: str, job: Job) -> bool: - try: - with connection: - connection.execute( - "INSERT INTO deliveries VALUES (?, ?, ?)", - (delivery_id, job.comment_id, now()), - ) - connection.execute( - """ - INSERT INTO jobs ( - id, kind, target_key, repo_owner, repo_name, issue_number, - pr_number, requester, message, comment_id, workflow_id, - status, stage, runtime_session_id, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - job.id, - job.kind, - job.target_key, - job.repo_owner, - job.repo_name, - job.issue_number, - job.pr_number, - job.requester, - job.message, - job.comment_id, - job.workflow_id, - job.status, - job.stage, - job.runtime_session_id, - now(), - ), - ) - except sqlite3.IntegrityError: - return False - return True - - async def claim_next(self) -> Job | None: - return await self._run(self._claim_next) - - @staticmethod - def _claim_next(connection: sqlite3.Connection) -> Job | None: - connection.execute("BEGIN IMMEDIATE") - row = connection.execute( - "SELECT * FROM jobs WHERE status = ? ORDER BY created_at LIMIT 1", - (JobStatus.QUEUED,), - ).fetchone() - if row is None: + async def receive( + self, event_id: str, job_id: str, incoming: CommandEvent + ) -> EvolveResult: + def operation(connection: sqlite3.Connection) -> EvolveResult: + connection.execute("BEGIN IMMEDIATE") + duplicate = connection.execute( + "SELECT job_id FROM job_events WHERE event_id=?", (event_id,) + ).fetchone() + if duplicate: + state = _state(connection, duplicate["job_id"]) + connection.commit() + return EvolveResult(state, True) + sequence = connection.execute( + "SELECT COALESCE(MAX(receive_sequence), 0) + 1 FROM jobs" + ).fetchone()[0] + event = CommandReceived( + job_id=job_id, + delivery_id=incoming.delivery_id, + receive_sequence=sequence, + command_body=incoming.body, + target_key=incoming.target_key, + repo_owner=incoming.repo_owner, + repo_name=incoming.repo_name, + issue_number=incoming.issue_number, + pr_number=incoming.pr_number, + requester=incoming.requester, + comment_id=incoming.comment_id, + ) + transition = next_state(None, event) + timestamp = now() + _insert_event(connection, event_id, event, timestamp) + _insert_state(connection, transition.state, timestamp) + _insert_tasks(connection, event_id, transition, timestamp) connection.commit() - return None - connection.execute( - "UPDATE jobs SET status = ?, stage = ?, started_at = ? WHERE id = ?", - (JobStatus.RUNNING, "starting", now(), row["id"]), - ) - connection.commit() - return job_from_row(row, status=JobStatus.RUNNING, stage="starting") + return EvolveResult(transition.state, False) - async def update_job( - self, - job_id: str, - *, - status: JobStatus | None = None, - stage: str | None = None, - error: str | None = None, - workflow_id: str | None = None, - runtime_session_id: str | None = None, - ) -> None: - updates: dict[str, object] = {} - if status is not None: - updates["status"] = status - if status in {JobStatus.SUCCEEDED, JobStatus.FAILED, JobStatus.REJECTED}: - updates["finished_at"] = now() - if stage is not None: - updates["stage"] = stage - if error is not None: - updates["error"] = error - if workflow_id is not None: - updates["workflow_id"] = workflow_id - if runtime_session_id is not None: - updates["runtime_session_id"] = runtime_session_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, - }, - ) + return await self._run(operation) - 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"}: - raise ValueError("Unsupported comment column") + async def evolve(self, event_id: str, event: JobEvent) -> EvolveResult: + def operation(connection: sqlite3.Connection) -> EvolveResult: + connection.execute("BEGIN IMMEDIATE") + if connection.execute( + "SELECT 1 FROM job_events WHERE event_id=?", (event_id,) + ).fetchone(): + state = _state(connection, event.job_id) + connection.commit() + return EvolveResult(state, True) + current = _state(connection, event.job_id) + transition = next_state(current, event) + timestamp = now() + _insert_event(connection, event_id, event, timestamp) + if isinstance(event, WorkflowCreated): + _insert_workflow(connection, event, timestamp) + _replace_state(connection, transition.state, current, timestamp) + _insert_tasks(connection, event_id, transition, timestamp) + connection.commit() + return EvolveResult(transition.state, False) + + return await self._run(operation) + + async def get_job_state(self, job_id: str) -> JobState | None: + return await self._run(lambda connection: _optional_state(connection, job_id)) + + async def claim_task(self, queue: str) -> ListenerTask | None: + return await self._run(lambda connection: _claim_task(connection, queue)) + + async def complete_task(self, task_id: int) -> None: await self._run( lambda connection: connection.execute( - f"UPDATE jobs SET {column} = ? WHERE id = ?", (comment_id, job_id) + "UPDATE listener_tasks SET status='completed', finished_at=? WHERE id=?", + (now(), task_id), ) ) - async def job_stage(self, job_id: str) -> str: - def select(connection: sqlite3.Connection) -> str: - row = connection.execute("SELECT stage FROM jobs WHERE id=?", (job_id,)).fetchone() - return str(row["stage"]) if row else "unknown" + async def retry_task(self, task_id: int, attempts: int, error: str) -> None: + delay = min(2 ** min(attempts, 8), 300) + available = (datetime.now(UTC) + timedelta(seconds=delay)).isoformat() + await self._run( + lambda connection: connection.execute( + "UPDATE listener_tasks SET status='pending', available_at=?, error=? WHERE id=?", + (available, error[:1000], task_id), + ) + ) - return await self._run(select) + async def running_job_states(self) -> list[JobState]: + return await self._run( + lambda connection: [ + _state_from_row(row) + for row in connection.execute("SELECT * FROM jobs WHERE status='running'") + ] + ) + + async def recover_tasks(self) -> None: + def recover(connection: sqlite3.Connection) -> None: + with connection: + connection.execute( + """UPDATE listener_tasks SET status='pending', started_at=NULL + WHERE status='running' AND queue='control'""" + ) + connection.execute( + """UPDATE listener_tasks SET status='pending', started_at=NULL + WHERE status='running' AND listener='execute' AND job_id IN + (SELECT id FROM jobs WHERE status='queued')""" + ) + connection.execute( + """UPDATE listener_tasks SET status='failed', finished_at=?, + error='Service restarted after execution began' + WHERE status='running' AND listener='execute' AND job_id IN + (SELECT id FROM jobs WHERE status<>'queued')""", + (now(),), + ) + + await self._run(recover) async def operational_comment_ids(self, owner: str, repo: str, issue: int) -> set[int]: return await self._run( lambda connection: { value for row in connection.execute( - """ - SELECT accepted_comment_id, started_comment_id FROM jobs - WHERE repo_owner=? AND repo_name=? AND issue_number=? - """, + "SELECT accepted_comment_id, started_comment_id FROM jobs " + "WHERE repo_owner=? AND repo_name=? AND issue_number=?", (owner, repo, issue), ) for value in row @@ -151,50 +179,33 @@ class JobStore(Database): } ) - async def recover_running(self) -> list[Job]: - def recover(connection: sqlite3.Connection) -> list[Job]: - rows = connection.execute( - "SELECT * FROM jobs WHERE status=?", (JobStatus.RUNNING,) - ).fetchall() - with connection: - connection.execute( - """ - UPDATE jobs SET status=?, stage=?, error=?, finished_at=? - WHERE status=? - """, - ( - JobStatus.FAILED, - "interrupted", - "Service restarted during an active OpenCode turn", - now(), - JobStatus.RUNNING, - ), - ) - return [job_from_row(row) for row in rows] - return await self._run(recover) - - -def job_from_row( - row: sqlite3.Row, - *, - status: JobStatus | None = None, - stage: str | None = None, -) -> Job: - return Job( - id=row["id"], - kind=JobKind(row["kind"]), - target_key=row["target_key"], - repo_owner=row["repo_owner"], - repo_name=row["repo_name"], - issue_number=row["issue_number"], - pr_number=row["pr_number"], - requester=row["requester"], - message=row["message"], - comment_id=row["comment_id"], - workflow_id=row["workflow_id"], - status=status or JobStatus(row["status"]), - stage=stage or row["stage"], - accepted_comment_id=row["accepted_comment_id"], - runtime_session_id=row["runtime_session_id"], +def _claim_task(connection: sqlite3.Connection, queue: str) -> ListenerTask | None: + connection.execute("BEGIN IMMEDIATE") + fifo = "" + if queue == "jobs": + fifo = """AND NOT EXISTS ( + SELECT 1 FROM jobs earlier WHERE earlier.receive_sequence < j.receive_sequence + AND earlier.status IN ('received', 'queued', 'running'))""" + row = connection.execute( + f"""SELECT t.* FROM listener_tasks t JOIN jobs j ON j.id=t.job_id + WHERE t.queue=? AND t.status='pending' AND t.available_at<=? {fifo} + ORDER BY {"j.receive_sequence" if queue == "jobs" else "t.id"} LIMIT 1""", + (queue, now()), + ).fetchone() + if row is None: + connection.commit() + return None + changed = connection.execute( + "UPDATE listener_tasks SET status='running', started_at=?, attempts=attempts+1 " + "WHERE id=? AND status='pending'", + (now(), row["id"]), + ) + if changed.rowcount != 1: + connection.rollback() + return None + connection.commit() + return ListenerTask( + row["id"], row["job_id"], row["source_event_id"], row["listener"], + row["queue"], row["attempts"] + 1, ) diff --git a/src/agentci/adapters/opencode.py b/src/agentci/adapters/opencode.py index fb037c5..98081dd 100644 --- a/src/agentci/adapters/opencode.py +++ b/src/agentci/adapters/opencode.py @@ -60,7 +60,7 @@ class OpenCodeClient: async def close(self) -> None: for session_id, workspace in tuple(self._active_sessions.items()): - await self.abort(session_id, workspace) + await self.abort(session_id, workspace, best_effort=True) await self.client.aclose() async def ready(self) -> bool: @@ -213,10 +213,7 @@ class OpenCodeClient: raise OpenCodeError("OpenCode did not return a valid result") async def _request( - self, - method: str, - path: str, - *, + self, method: str, path: str, *, workspace: Path, json: dict[str, Any] | None = None, ) -> dict[str, Any]: @@ -238,12 +235,16 @@ class OpenCodeClient: raise OpenCodeError(f"OpenCode returned an invalid response for {method} {path}") return payload - async def abort(self, session_id: str, workspace: Path) -> None: + async def abort(self, session_id: str, workspace: Path, *, best_effort: bool = False) -> None: try: - await self.client.post( - f"/session/{session_id}/abort", - headers=directory_headers(workspace), - timeout=10, + response = await self.client.post( + f"/session/{session_id}/abort", headers=directory_headers(workspace), timeout=10 ) - except httpx.HTTPError: - log.exception("OpenCode session could not be aborted") + if response.is_success or response.status_code in {404, 409}: + return + response.raise_for_status() + except httpx.HTTPError as exc: + if best_effort: + log.warning("OpenCode session could not be aborted", exc_info=exc) + return + raise OpenCodeError(f"OpenCode session {session_id} could not be aborted") from exc diff --git a/src/agentci/adapters/state_persistence.py b/src/agentci/adapters/state_persistence.py new file mode 100644 index 0000000..cea99ec --- /dev/null +++ b/src/agentci/adapters/state_persistence.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import sqlite3 + +from agentci.domain.events import JobEvent, WorkflowCreated +from agentci.domain.models import JobKind, JobStatus +from agentci.domain.state_machine import JobState, Transition + + +def insert_event(connection: sqlite3.Connection, event_id: str, event: JobEvent, ts: str) -> None: + connection.execute( + "INSERT INTO job_events VALUES (?, ?, ?, ?, ?)", + (event_id, event.job_id, event.type, event.model_dump_json(), ts), + ) + + +def insert_tasks( + connection: sqlite3.Connection, event_id: str, transition: Transition, timestamp: str +) -> None: + for ordinal, notification in enumerate(transition.notifications): + connection.execute( + "INSERT INTO listener_tasks(job_id, source_event_id, ordinal, listener, queue, " + "available_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + transition.state.id, + event_id, + ordinal, + notification.listener, + notification.queue, + timestamp, + timestamp, + ), + ) + + +def _values(state: JobState) -> tuple[object, ...]: + return ( + state.id, state.kind, state.target_key, state.repo_owner, state.repo_name, + state.issue_number, state.pr_number, state.requester, state.message, state.comment_id, + state.delivery_id, state.receive_sequence, state.command_body, state.workflow_id, + state.status, state.stage, state.error, state.runtime_session_id, + state.accepted_comment_id, state.comment_body, + ) + + +def insert_state(connection: sqlite3.Connection, state: JobState, timestamp: str) -> None: + connection.execute( + """INSERT INTO jobs(id, kind, target_key, repo_owner, repo_name, issue_number, + pr_number, requester, message, comment_id, delivery_id, receive_sequence, command_body, + workflow_id, status, stage, error, runtime_session_id, accepted_comment_id, + comment_body, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + (*_values(state), timestamp), + ) + + +def replace_state( + connection: sqlite3.Connection, state: JobState, previous: JobState, timestamp: str +) -> None: + started = ( + timestamp + if previous.status is JobStatus.QUEUED and state.status is JobStatus.RUNNING + else None + ) + terminal = {JobStatus.SUCCEEDED, JobStatus.REJECTED, JobStatus.FAILED} + finished = timestamp if previous.status not in terminal and state.status in terminal else None + connection.execute( + """UPDATE jobs SET kind=?, target_key=?, repo_owner=?, repo_name=?, issue_number=?, + pr_number=?, requester=?, message=?, comment_id=?, delivery_id=?, receive_sequence=?, + command_body=?, workflow_id=?, status=?, stage=?, error=?, runtime_session_id=?, + accepted_comment_id=?, comment_body=?, started_at=COALESCE(started_at, ?), + finished_at=COALESCE(finished_at, ?) WHERE id=?""", + (*_values(state)[1:], started, finished, state.id), + ) + + +def optional_state(connection: sqlite3.Connection, job_id: str) -> JobState | None: + row = connection.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone() + return state_from_row(row) if row else None + + +def state(connection: sqlite3.Connection, job_id: str) -> JobState: + value = optional_state(connection, job_id) + if value is None: + raise KeyError(f"Unknown job {job_id}") + return value + + +def state_from_row(row: sqlite3.Row) -> JobState: + return JobState( + id=row["id"], kind=JobKind(row["kind"]) if row["kind"] else None, + target_key=row["target_key"], repo_owner=row["repo_owner"], repo_name=row["repo_name"], + issue_number=row["issue_number"], pr_number=row["pr_number"], requester=row["requester"], + message=row["message"], comment_id=row["comment_id"], delivery_id=row["delivery_id"], + receive_sequence=row["receive_sequence"], command_body=row["command_body"], + workflow_id=row["workflow_id"], status=JobStatus(row["status"]), stage=row["stage"], + error=row["error"], runtime_session_id=row["runtime_session_id"], + accepted_comment_id=row["accepted_comment_id"], comment_body=row["comment_body"], + ) + + +def insert_workflow(connection: sqlite3.Connection, event: WorkflowCreated, ts: str) -> None: + workflow = event.workflow + connection.execute( + """INSERT INTO workflows(id, kind, repo_owner, repo_name, issue_number, pr_number, + base_sha, branch, workspace_path, primary_session_id, reviewer_session_id, artifact, + review_json, status, runtime, created_at, updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + workflow.id, workflow.kind, workflow.repo_owner, workflow.repo_name, + workflow.issue_number, workflow.pr_number, workflow.base_sha, workflow.branch, + str(workflow.workspace_path), workflow.primary_session_id, + workflow.reviewer_session_id, workflow.artifact, workflow.review_json, + workflow.status, workflow.runtime, ts, ts, + ), + ) diff --git a/src/agentci/api/webhook.py b/src/agentci/api/webhook.py index 8e3d762..2e7bdbb 100644 --- a/src/agentci/api/webhook.py +++ b/src/agentci/api/webhook.py @@ -5,12 +5,10 @@ import hmac import json import logging from typing import Any -from uuid import uuid4 from fastapi import APIRouter, HTTPException, Request, Response, status -from agentci.domain.commands import CommandError, parse_command, resolve_job_kind -from agentci.domain.models import CommandEvent, Job, JobStatus +from agentci.domain.models import CommandEvent router = APIRouter() log = logging.getLogger(__name__) @@ -54,7 +52,7 @@ async def webhook(request: Request) -> Response: 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: + if event is None or event.requester.casefold() == container.settings.bot_username.casefold(): 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) @@ -69,68 +67,25 @@ 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) + if not event.delivery_id: + raise HTTPException(status.HTTP_400_BAD_REQUEST, "Missing X-Gitea-Delivery") 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, - event.repo_name, - event.issue_number, - "Agent command rejected: repository write permission is required.", - ) - return Response(status_code=status.HTTP_202_ACCEPTED) 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) - ) - return Response(status_code=status.HTTP_202_ACCEPTED) - if command is None: - return Response(status_code=status.HTTP_204_NO_CONTENT) - 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) - ) - return Response(status_code=status.HTTP_202_ACCEPTED) - job = Job( - id=str(uuid4()), - kind=kind, - target_key=event.target_key, - repo_owner=event.repo_owner, - repo_name=event.repo_name, - issue_number=event.issue_number, - pr_number=event.pr_number, - requester=event.requester, - message=command.message, - comment_id=event.comment_id, - status=JobStatus.QUEUED, - stage="queued", - ) - if not await container.storage.enqueue(event.delivery_id, job): - log.info("duplicate command ignored", extra={**extra, "job_id": job.id}) + result = await container.state_machine.receive(event) + except Exception: + log.exception("could not persist command", extra=extra) + raise + if result.duplicate: + log.info("duplicate command ignored", extra={**extra, "job_id": result.state.id}) return Response(status_code=status.HTTP_200_OK) log.info( - "agent job queued", - extra={**extra, "job_id": job.id, "stage": job.kind.value}, + "agent command persisted", + extra={ + **extra, + "job_id": result.state.id, + "receive_sequence": result.state.receive_sequence, + }, ) - comment_id = await container.gitea.create_comment( - event.repo_owner, - event.repo_name, - event.issue_number, - f"Agent job `{job.id}` queued (`{job.kind}`).", - ) - await container.storage.set_job_comment(job.id, "accepted_comment_id", comment_id) return Response(status_code=status.HTTP_202_ACCEPTED) diff --git a/src/agentci/container.py b/src/agentci/container.py index d058de6..8076c22 100644 --- a/src/agentci/container.py +++ b/src/agentci/container.py @@ -11,6 +11,7 @@ from agentci.adapters.opencode import OpenCodeClient from agentci.adapters.storage import Storage from agentci.config import Settings from agentci.prompts import PromptLibrary +from agentci.state_machine import StateMachine from agentci.worker import Worker from agentci.workflows.common import Dependencies from agentci.workflows.context import ContextBuilder @@ -25,6 +26,7 @@ class Container: gitea: GiteaClient git: GitClient opencode: OpenCodeClient + state_machine: StateMachine worker: Worker async def close(self) -> None: @@ -41,6 +43,7 @@ async def build_container(settings: Settings) -> Container: settings.workspaces_dir.mkdir(parents=True, exist_ok=True) storage = Storage(settings.database_path, package_dir / "migrations") await storage.initialize() + state_machine = StateMachine(storage) gitea = GiteaClient(settings.gitea_url, settings.gitea_token) git = GitClient( gitea_url=settings.gitea_url, @@ -87,12 +90,14 @@ async def build_container(settings: Settings) -> Container: dispatcher = Dispatcher(dependencies) worker = Worker( storage=storage, + state_machine=state_machine, gitea=gitea, opencode=opencode, dispatcher=dispatcher, poll_seconds=settings.worker_poll_seconds, workspaces_dir=settings.workspaces_dir, + bot_username=settings.bot_username, ) - container = Container(settings, storage, gitea, git, opencode, worker) + container = Container(settings, storage, gitea, git, opencode, state_machine, worker) log.info("container initialization completed", extra={"operation": "container.build"}) return container diff --git a/src/agentci/domain/events.py b/src/agentci/domain/events.py new file mode 100644 index 0000000..9bd22c1 --- /dev/null +++ b/src/agentci/domain/events.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field + +from agentci.domain.models import Workflow + + +class Event(BaseModel): + model_config = ConfigDict(frozen=True) + + job_id: str + + +class CommandReceived(Event): + type: Literal["command_received"] = "command_received" + delivery_id: str + receive_sequence: int + command_body: str + target_key: str + repo_owner: str + repo_name: str + issue_number: int + pr_number: int | None + requester: str + comment_id: int + + +class PermissionGranted(Event): + type: Literal["permission_granted"] = "permission_granted" + + +class PermissionDenied(Event): + type: Literal["permission_denied"] = "permission_denied" + + +class JobStarted(Event): + type: Literal["job_started"] = "job_started" + + +class JobProgress(Event): + type: Literal["job_progress"] = "job_progress" + stage: str + + +class WorkflowCreated(Event): + type: Literal["workflow_created"] = "workflow_created" + workflow: Workflow + stage: str + + +class WorkflowLinked(Event): + type: Literal["workflow_linked"] = "workflow_linked" + workflow_id: str + stage: str + + +class RuntimeSessionLinked(Event): + type: Literal["runtime_session_linked"] = "runtime_session_linked" + session_id: str + + +class JobCompleted(Event): + type: Literal["job_completed"] = "job_completed" + comment_body: str + + +class JobRejected(Event): + type: Literal["job_rejected"] = "job_rejected" + reason: str + + +class JobFailed(Event): + type: Literal["job_failed"] = "job_failed" + error: str + stage: str + + +class ServiceRestarted(Event): + type: Literal["service_restarted"] = "service_restarted" + + +class CommentLinked(Event): + type: Literal["comment_linked"] = "comment_linked" + comment_id: int + + +JobEvent = Annotated[ + CommandReceived + | PermissionGranted + | PermissionDenied + | JobStarted + | JobProgress + | WorkflowCreated + | WorkflowLinked + | RuntimeSessionLinked + | JobCompleted + | JobRejected + | JobFailed + | ServiceRestarted + | CommentLinked, + Field(discriminator="type"), +] diff --git a/src/agentci/domain/models.py b/src/agentci/domain/models.py index a2a495d..809e387 100644 --- a/src/agentci/domain/models.py +++ b/src/agentci/domain/models.py @@ -25,6 +25,7 @@ class JobKind(StrEnum): class JobStatus(StrEnum): + RECEIVED = "received" QUEUED = "queued" RUNNING = "running" SUCCEEDED = "succeeded" diff --git a/src/agentci/domain/state_machine.py b/src/agentci/domain/state_machine.py new file mode 100644 index 0000000..9c6fd5d --- /dev/null +++ b/src/agentci/domain/state_machine.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +from dataclasses import dataclass, replace + +from agentci.domain.commands import CommandError, parse_command, resolve_job_kind +from agentci.domain.events import ( + CommandReceived, + CommentLinked, + JobCompleted, + JobEvent, + JobFailed, + JobProgress, + JobRejected, + JobStarted, + PermissionDenied, + PermissionGranted, + RuntimeSessionLinked, + ServiceRestarted, + WorkflowCreated, + WorkflowLinked, +) +from agentci.domain.models import JobKind, JobStatus + + +class InvalidTransition(ValueError): + pass + + +@dataclass(frozen=True) +class Notification: + listener: str + queue: str = "control" + + +@dataclass(frozen=True) +class JobState: + id: str + target_key: str + repo_owner: str + repo_name: str + issue_number: int + pr_number: int | None + requester: str + comment_id: int + delivery_id: str + receive_sequence: int + command_body: str + kind: JobKind | None = None + message: str | None = None + status: JobStatus = JobStatus.RECEIVED + stage: str = "received" + error: str | None = None + workflow_id: str | None = None + runtime_session_id: str | None = None + accepted_comment_id: int | None = None + comment_body: str | None = None + + @property + def is_pull_request(self) -> bool: + return self.pr_number is not None + + +@dataclass(frozen=True) +class Transition: + state: JobState + notifications: tuple[Notification, ...] = () + + +RECONCILE = Notification("reconcile_comment") + + +def next_state(state: JobState | None, event: JobEvent) -> Transition: + if state is None: + if not isinstance(event, CommandReceived): + raise InvalidTransition("Only CommandReceived can create a job") + created = JobState( + id=event.job_id, + target_key=event.target_key, + repo_owner=event.repo_owner, + repo_name=event.repo_name, + issue_number=event.issue_number, + pr_number=event.pr_number, + requester=event.requester, + comment_id=event.comment_id, + delivery_id=event.delivery_id, + receive_sequence=event.receive_sequence, + command_body=event.command_body, + ) + return Transition(created, (Notification("authorize"),)) + if event.job_id != state.id: + raise InvalidTransition("Event job ID does not match state") + if isinstance(event, CommentLinked): + return Transition(replace(state, accepted_comment_id=event.comment_id)) + if isinstance(event, ServiceRestarted): + if state.status is not JobStatus.RUNNING: + return Transition(state) + failed = replace( + state, + status=JobStatus.FAILED, + stage="interrupted", + error="Service restarted during an active OpenCode turn", + ) + listeners = [Notification("abort_sessions"), RECONCILE] + if state.workflow_id: + listeners.insert(1, Notification("fail_workflow")) + return Transition(failed, tuple(listeners)) + if state.status is JobStatus.RECEIVED: + return _received(state, event) + if state.status is JobStatus.QUEUED and isinstance(event, JobStarted): + return Transition( + replace(state, status=JobStatus.RUNNING, stage="starting"), (RECONCILE,) + ) + if state.status is JobStatus.RUNNING: + return _running(state, event) + raise InvalidTransition(f"{event.type} is invalid while job is {state.status}") + + +def _received(state: JobState, event: JobEvent) -> Transition: + if isinstance(event, PermissionDenied): + reason = "Agent command rejected: repository write permission is required." + return Transition( + replace(state, status=JobStatus.REJECTED, stage="rejected", error=reason), + (RECONCILE,), + ) + if not isinstance(event, PermissionGranted): + raise InvalidTransition(f"{event.type} is invalid while job is received") + try: + command = parse_command(state.command_body) + if command is None: + raise CommandError("Invalid agent command.") + kind = resolve_job_kind(command, is_pull_request=state.is_pull_request) + except CommandError as exc: + return Transition( + replace(state, status=JobStatus.REJECTED, stage="rejected", error=str(exc)), + (RECONCILE,), + ) + queued = replace( + state, + kind=kind, + message=command.message, + status=JobStatus.QUEUED, + stage="queued", + ) + return Transition(queued, (Notification("execute", "jobs"), RECONCILE)) + + +def _running(state: JobState, event: JobEvent) -> Transition: + if isinstance(event, JobProgress): + return Transition(replace(state, stage=event.stage)) + if isinstance(event, WorkflowCreated): + return Transition(replace(state, workflow_id=event.workflow.id, stage=event.stage)) + if isinstance(event, WorkflowLinked): + return Transition(replace(state, workflow_id=event.workflow_id, stage=event.stage)) + if isinstance(event, RuntimeSessionLinked): + return Transition(replace(state, runtime_session_id=event.session_id)) + if isinstance(event, JobCompleted): + return Transition( + replace( + state, + status=JobStatus.SUCCEEDED, + stage="completed", + comment_body=event.comment_body, + ), + (RECONCILE,), + ) + if isinstance(event, (JobRejected, JobFailed)): + rejected = isinstance(event, JobRejected) + error = event.reason if rejected else event.error + stage = "rejected" if rejected else event.stage + status = JobStatus.REJECTED if rejected else JobStatus.FAILED + listeners = [RECONCILE] + if state.workflow_id: + listeners.append(Notification("fail_workflow")) + return Transition(replace(state, status=status, stage=stage, error=error), tuple(listeners)) + raise InvalidTransition(f"{event.type} is invalid while job is running") + + +def render_job_comment(state: JobState) -> str: + marker = f"" + if state.status is JobStatus.SUCCEEDED and state.comment_body: + body = state.comment_body + elif state.status is JobStatus.REJECTED: + body = f"Agent job `{state.id}` was rejected: {state.error}" + elif state.status is JobStatus.FAILED: + body = f"Agent job `{state.id}` failed during `{state.stage}`: {state.error}" + else: + kind = state.kind.value if state.kind else "command" + body = f"Agent job `{state.id}` {state.status.value} (`{kind}`; stage: `{state.stage}`)." + return f"{marker}\n{body}" diff --git a/src/agentci/migrations/003_state_machine.sql b/src/agentci/migrations/003_state_machine.sql new file mode 100644 index 0000000..184dfb4 --- /dev/null +++ b/src/agentci/migrations/003_state_machine.sql @@ -0,0 +1,96 @@ +PRAGMA foreign_keys=OFF; + +ALTER TABLE jobs RENAME TO jobs_legacy; + +CREATE TABLE jobs ( + id TEXT PRIMARY KEY, + kind TEXT, + target_key TEXT NOT NULL, + repo_owner TEXT NOT NULL, + repo_name TEXT NOT NULL, + issue_number INTEGER NOT NULL, + pr_number INTEGER, + requester TEXT NOT NULL, + message TEXT, + comment_id INTEGER NOT NULL, + delivery_id TEXT NOT NULL UNIQUE, + receive_sequence INTEGER NOT NULL UNIQUE, + command_body TEXT NOT NULL, + workflow_id TEXT REFERENCES workflows(id), + status TEXT NOT NULL, + stage TEXT NOT NULL, + error TEXT, + runtime_session_id TEXT, + accepted_comment_id INTEGER, + started_comment_id INTEGER, + comment_body TEXT, + created_at TEXT NOT NULL, + started_at TEXT, + finished_at TEXT +); + +INSERT INTO jobs ( + id, kind, target_key, repo_owner, repo_name, issue_number, pr_number, + requester, message, comment_id, delivery_id, receive_sequence, command_body, + workflow_id, status, stage, error, runtime_session_id, accepted_comment_id, + started_comment_id, created_at, started_at, finished_at +) +SELECT j.id, j.kind, j.target_key, j.repo_owner, j.repo_name, j.issue_number, + j.pr_number, j.requester, j.message, j.comment_id, + COALESCE(d.delivery_id, 'legacy:' || j.id), + ROW_NUMBER() OVER (ORDER BY j.created_at, j.id), + '/agent ' || CASE j.kind + WHEN 'iterate_plan' THEN 'iterate' + WHEN 'iterate_implement' THEN 'iterate' + ELSE j.kind END || CASE WHEN j.message = '' THEN '' ELSE ' ' || j.message END, + j.workflow_id, j.status, j.stage, j.error, j.runtime_session_id, + j.accepted_comment_id, j.started_comment_id, j.created_at, j.started_at, j.finished_at +FROM jobs_legacy j +LEFT JOIN deliveries d ON d.comment_id = j.comment_id; + +CREATE TABLE job_events ( + event_id TEXT PRIMARY KEY, + job_id TEXT NOT NULL REFERENCES jobs(id) DEFERRABLE INITIALLY DEFERRED, + event_type TEXT NOT NULL, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL +); + +INSERT INTO job_events(event_id, job_id, event_type, payload_json, created_at) +SELECT 'delivery:' || delivery_id, id, 'legacy', '{}', created_at FROM jobs; + +CREATE TABLE listener_tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + job_id TEXT NOT NULL REFERENCES jobs(id), + source_event_id TEXT NOT NULL REFERENCES job_events(event_id), + ordinal INTEGER NOT NULL, + listener TEXT NOT NULL, + queue TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + available_at TEXT NOT NULL, + error TEXT, + created_at TEXT NOT NULL, + started_at TEXT, + finished_at TEXT, + UNIQUE(source_event_id, listener, ordinal) +); + +INSERT INTO listener_tasks( + job_id, source_event_id, ordinal, listener, queue, status, available_at, created_at +) +SELECT id, 'delivery:' || delivery_id, 0, 'execute', 'jobs', 'pending', created_at, created_at +FROM jobs WHERE status = 'queued'; + +INSERT INTO listener_tasks( + job_id, source_event_id, ordinal, listener, queue, status, available_at, created_at +) +SELECT id, 'delivery:' || delivery_id, 1, 'reconcile_comment', 'control', 'pending', + created_at, created_at +FROM jobs WHERE status = 'queued'; + +DROP TABLE jobs_legacy; +CREATE INDEX idx_jobs_queue ON jobs(status, receive_sequence); +CREATE INDEX idx_jobs_target ON jobs(target_key, status); +CREATE INDEX idx_listener_eligible ON listener_tasks(queue, status, available_at, id); +PRAGMA foreign_keys=ON; diff --git a/src/agentci/reporting.py b/src/agentci/reporting.py new file mode 100644 index 0000000..c424b5b --- /dev/null +++ b/src/agentci/reporting.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +from contextvars import ContextVar, Token + +from agentci.domain.events import ( + JobProgress, + RuntimeSessionLinked, + WorkflowCreated, + WorkflowLinked, +) +from agentci.domain.models import Workflow +from agentci.state_machine import StateMachine + + +class NullReporter: + final_body: str | None = None + + async def progress(self, _stage: str) -> None: + pass + + async def create_workflow(self, _workflow: Workflow, _stage: str) -> None: + pass + + async def link_workflow(self, _workflow_id: str, _stage: str) -> None: + pass + + async def link_runtime_session(self, _session_id: str) -> None: + pass + + def finish(self, body: str) -> None: + self.final_body = body + + +_current: ContextVar[JobReporter | NullReporter | None] = ContextVar( + "job_reporter", default=None +) + + +class JobReporter: + def __init__(self, host: StateMachine, job_id: str, task_id: int) -> None: + self.host = host + self.job_id = job_id + self.task_id = task_id + self.sequence = 0 + self.final_body: str | None = None + + async def progress(self, stage: str) -> None: + await self._emit(JobProgress(job_id=self.job_id, stage=stage)) + + async def create_workflow(self, workflow: Workflow, stage: str) -> None: + await self._emit(WorkflowCreated(job_id=self.job_id, workflow=workflow, stage=stage)) + + async def link_workflow(self, workflow_id: str, stage: str) -> None: + await self._emit( + WorkflowLinked(job_id=self.job_id, workflow_id=workflow_id, stage=stage) + ) + + async def link_runtime_session(self, session_id: str) -> None: + await self._emit(RuntimeSessionLinked(job_id=self.job_id, session_id=session_id)) + + def finish(self, body: str) -> None: + self.final_body = body + + async def _emit(self, event) -> None: + self.sequence += 1 + await self.host.evolve(f"task:{self.task_id}:report:{self.sequence}", event) + + +def bind_reporter( + reporter: JobReporter, +) -> Token[JobReporter | NullReporter | None]: + return _current.set(reporter) + + +def reset_reporter(token: Token[JobReporter | NullReporter | None]) -> None: + _current.reset(token) + + +def reporter() -> JobReporter | NullReporter: + return _current.get() or NullReporter() diff --git a/src/agentci/state_machine.py b/src/agentci/state_machine.py new file mode 100644 index 0000000..c6d67d0 --- /dev/null +++ b/src/agentci/state_machine.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from dataclasses import dataclass +from uuid import UUID, uuid5 + +from agentci.adapters.job_store import EvolveResult, JobStore +from agentci.domain.events import JobEvent +from agentci.domain.models import CommandEvent +from agentci.domain.state_machine import JobState + +JOB_NAMESPACE = UUID("59565f0f-f17d-4b80-bfba-7ef1fbfd38eb") + + +@dataclass(frozen=True) +class ReceiveResult: + state: JobState + duplicate: bool + + +class StateMachine: + def __init__(self, store: JobStore) -> None: + self.store = store + + async def receive(self, incoming: CommandEvent) -> ReceiveResult: + job_id = str(uuid5(JOB_NAMESPACE, incoming.delivery_id)) + result = await self.store.receive(f"delivery:{incoming.delivery_id}", job_id, incoming) + return ReceiveResult(result.state, result.duplicate) + + async def evolve(self, event_id: str, event: JobEvent) -> EvolveResult: + return await self.store.evolve(event_id, event) + + async def get(self, job_id: str) -> JobState | None: + return await self.store.get_job_state(job_id) diff --git a/src/agentci/worker.py b/src/agentci/worker.py index 9e1b5b4..da30a30 100644 --- a/src/agentci/worker.py +++ b/src/agentci/worker.py @@ -6,9 +6,25 @@ from contextlib import suppress from pathlib import Path from agentci.adapters.gitea import GiteaClient +from agentci.adapters.job_store import ListenerTask from agentci.adapters.opencode import OpenCodeClient from agentci.adapters.storage import Storage -from agentci.domain.models import Job, JobStatus +from agentci.domain.events import ( + CommentLinked, + JobCompleted, + JobFailed, + JobStarted, + PermissionDenied, + PermissionGranted, + ServiceRestarted, +) +from agentci.domain.events import ( + JobRejected as RejectedEvent, +) +from agentci.domain.models import JobStatus +from agentci.domain.state_machine import JobState, render_job_comment +from agentci.reporting import JobReporter, bind_reporter, reset_reporter +from agentci.state_machine import StateMachine from agentci.workflows.common import JobRejected from agentci.workflows.dispatcher import Dispatcher @@ -20,165 +36,174 @@ class Worker: self, *, storage: Storage, + state_machine: StateMachine, gitea: GiteaClient, opencode: OpenCodeClient, dispatcher: Dispatcher, poll_seconds: float, workspaces_dir: Path, + bot_username: str, ) -> None: self.storage = storage + self.host = state_machine self.gitea = gitea self.opencode = opencode self.dispatcher = dispatcher self.poll_seconds = poll_seconds self.workspaces_dir = workspaces_dir + self.bot_username = bot_username 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.opencode.ready(): - log.warning( - "worker waiting for OpenCode provider 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"}) + await self._recover() + await asyncio.gather(self._loop("control", stop), self._loop("jobs", stop)) - async def _run_job(self, job: Job) -> None: - extra = {"job_id": job.id, "target": job.target_key} - log.info("job started", extra=extra) - try: - if job.accepted_comment_id is None: - job.accepted_comment_id = await self.gitea.create_comment( - job.repo_owner, - job.repo_name, - job.issue_number, - f"Agent job `{job.id}` started (`{job.kind}`).", - ) - await self.storage.set_job_comment( - job.id, "accepted_comment_id", job.accepted_comment_id + async def _loop(self, queue: str, stop: asyncio.Event) -> None: + while not stop.is_set(): + if queue == "jobs" and not await self.opencode.ready(): + await self._wait(stop) + continue + task = await self.storage.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.listener, "queue": queue}, ) + await self.storage.retry_task(task.id, task.attempts, _safe_error(exc)) else: - await self.gitea.update_comment( - job.repo_owner, - job.repo_name, - job.accepted_comment_id, - f"Agent job `{job.id}` started (`{job.kind}`).", - ) - await self.dispatcher.dispatch(job) - except JobRejected as exc: - await self._safe_update_job( - job, status=JobStatus.REJECTED, stage="rejected", error=str(exc) - ) - await self._safe_job_comment(job, f"Agent job `{job.id}` was rejected: {exc}") - log.info("job rejected", extra={**extra, "stage": "rejected"}) - except Exception as 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._safe_fail_workflow(job) - await self._safe_job_comment( - job, - f"Agent job `{job.id}` failed during `{failed_stage}`: {_safe_error(exc)}", - ) - log.exception("job failed", extra={**extra, "stage": failed_stage}) + await self.storage.complete_task(task.id) + + async def _handle(self, task: ListenerTask) -> None: + state = await self.host.get(task.job_id) + if state is None: + return + if task.listener == "authorize": + await self._authorize(task, state) + elif task.listener == "execute": + await self._execute(task, state) + elif task.listener == "reconcile_comment": + await self._reconcile(task, state) + elif task.listener == "fail_workflow": + await self.storage.fail_job_workflow(state.id) + elif task.listener == "abort_sessions": + await self._abort_job_sessions(state) else: - await self._safe_update_job( - job, status=JobStatus.SUCCEEDED, stage="completed" - ) - log.info("job completed", extra=extra) + raise RuntimeError(f"Unknown listener {task.listener}") - async def _report_interrupted(self) -> None: - jobs = await self.storage.recover_running() - if jobs: - log.warning( - "recovering interrupted jobs", - extra={"operation": "worker.recover", "item_count": len(jobs)}, + async def _authorize(self, task: ListenerTask, state: JobState) -> None: + if state.status is not JobStatus.RECEIVED: + return + permitted = await self.gitea.has_write_permission( + state.repo_owner, state.repo_name, state.requester + ) + event = ( + PermissionGranted(job_id=state.id) + if permitted + else PermissionDenied(job_id=state.id) + ) + outcome = "permission-granted" if permitted else "permission-denied" + await self.host.evolve(f"task:{task.id}:{outcome}", event) + + async def _execute(self, task: ListenerTask, state: JobState) -> None: + if state.status is JobStatus.RUNNING: + await self.host.evolve( + f"task:{task.id}:interrupted", ServiceRestarted(job_id=state.id) ) - for job in jobs: - await self._abort_job_sessions(job) - await self._safe_fail_workflow(job) - await self._safe_job_comment( - job, - f"Agent job `{job.id}` failed because the service restarted during execution.", + return + if state.status is not JobStatus.QUEUED: + return + result = await self.host.evolve( + f"task:{task.id}:started", JobStarted(job_id=state.id) + ) + running = result.state + reporter = JobReporter(self.host, state.id, task.id) + token = bind_reporter(reporter) + try: + await self.dispatcher.dispatch(running) + except JobRejected as exc: + await self.host.evolve( + f"task:{task.id}:rejected", RejectedEvent(job_id=state.id, reason=str(exc)) + ) + except Exception as exc: + latest = await self.host.get(state.id) + stage = latest.stage if latest else running.stage + await self.host.evolve( + f"task:{task.id}:failed", + JobFailed(job_id=state.id, error=_safe_error(exc), stage=stage), + ) + else: + await self.host.evolve( + f"task:{task.id}:completed", + JobCompleted( + job_id=state.id, + comment_body=reporter.final_body or "Agent job completed.", + ), + ) + finally: + reset_reporter(token) + + async def _reconcile(self, task: ListenerTask, state: JobState) -> None: + latest = await self.host.get(state.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.host.evolve( + 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.storage.recover_tasks() + for state in await self.storage.running_job_states(): + await self.host.evolve( + f"recovery:{state.id}:service-restarted", + ServiceRestarted(job_id=state.id), ) - async def _abort_job_sessions(self, job: Job) -> None: + async def _abort_job_sessions(self, state: JobState) -> None: sessions: set[tuple[str, Path]] = set() - workflow = None - if job.workflow_id: - workflow = await self.storage.get_workflow(job.workflow_id) - if workflow is not None: + workflow = await self.storage.get_workflow(state.workflow_id) if state.workflow_id else None + if workflow: sessions.update( - (session_id, workflow.workspace_path) - for session_id in ( - workflow.primary_session_id, - workflow.reviewer_session_id, - ) - if session_id + (session, workflow.workspace_path) + for session in (workflow.primary_session_id, workflow.reviewer_session_id) + if session ) - elif job.runtime_session_id: - workspace = self.workspaces_dir / f"fix-{job.id}" / "repo" - sessions.add((job.runtime_session_id, workspace)) - for session_id, workspace in sessions: - await self.opencode.abort(session_id, workspace) - - async def _safe_job_comment(self, job: Job, body: str) -> None: - try: - if job.accepted_comment_id is None: - job.accepted_comment_id = await self.gitea.create_comment( - job.repo_owner, job.repo_name, job.issue_number, body - ) - await self.storage.set_job_comment( - job.id, "accepted_comment_id", job.accepted_comment_id - ) - else: - await self.gitea.update_comment( - job.repo_owner, job.repo_name, job.accepted_comment_id, body - ) - 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 + elif state.runtime_session_id: + sessions.add( + (state.runtime_session_id, self.workspaces_dir / f"fix-{state.id}" / "repo") ) - 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}) + for session, workspace in sessions: + await self.opencode.abort(session, workspace) async def _wait(self, stop: asyncio.Event) -> None: with suppress(TimeoutError): diff --git a/src/agentci/workflows/change_set.py b/src/agentci/workflows/change_set.py index 3d033f4..7f5bd84 100644 --- a/src/agentci/workflows/change_set.py +++ b/src/agentci/workflows/change_set.py @@ -3,6 +3,7 @@ from __future__ import annotations from pathlib import Path from agentci.domain.models import AgentResult, Job +from agentci.reporting import reporter from agentci.workflows.common import Dependencies, JobRejected @@ -20,14 +21,14 @@ class ChangeSet: set_upstream: bool, commit_prefix: str, ) -> str: - await self.deps.storage.update_job(job.id, stage="validating changes") + await reporter().progress("validating changes") if not await self.deps.git.has_changes(workspace): raise JobRejected("OpenCode completed without producing any file changes.") await self.deps.git.diff_check(workspace) title = _commit_title(result.summary_markdown) - await self.deps.storage.update_job(job.id, stage="committing changes") + await reporter().progress("committing changes") sha = await self.deps.git.commit(workspace, f"{commit_prefix}: {title}") - await self.deps.storage.update_job(job.id, stage="pushing changes") + await reporter().progress("pushing changes") await self.deps.git.push(workspace, branch, set_upstream=set_upstream) return sha diff --git a/src/agentci/workflows/code_review.py b/src/agentci/workflows/code_review.py index 659ac96..ef0c736 100644 --- a/src/agentci/workflows/code_review.py +++ b/src/agentci/workflows/code_review.py @@ -1,6 +1,7 @@ from __future__ import annotations from agentci.domain.models import AgentResult, Job, ReviewReport, Workflow +from agentci.reporting import reporter from agentci.workflows.common import ( Dependencies, report_for_prompt, @@ -23,10 +24,9 @@ class CodeReviewLoop: ) -> tuple[AgentResult, ReviewReport]: report = ReviewReport(summary="", findings=[]) for round_index in range(self.deps.settings.implement_review_rounds): - await self.deps.storage.update_job( - job.id, - stage=f"reviewing implementation {round_index + 1}/" - f"{self.deps.settings.implement_review_rounds}", + await reporter().progress( + f"reviewing implementation {round_index + 1}/" + f"{self.deps.settings.implement_review_rounds}" ) report = await self.once( workflow, diff --git a/src/agentci/workflows/common.py b/src/agentci/workflows/common.py index 9afff2e..a8b4e97 100644 --- a/src/agentci/workflows/common.py +++ b/src/agentci/workflows/common.py @@ -9,8 +9,9 @@ from agentci.adapters.gitea import GiteaClient from agentci.adapters.opencode import OpenCodeClient from agentci.adapters.storage import Storage from agentci.config import Settings -from agentci.domain.models import Job, ReviewReport +from agentci.domain.models import ReviewReport from agentci.prompts import PromptLibrary +from agentci.reporting import reporter from agentci.workflows.context import ContextBuilder @@ -71,9 +72,5 @@ def agent_comment(kind: str, workflow_id: str, body: str) -> str: return f"\n{body}" -async def update_job_comment(deps: Dependencies, job: Job, body: str) -> None: - if job.accepted_comment_id is None: - raise RuntimeError("Expected a persisted Gitea job comment ID") - await deps.gitea.update_comment( - job.repo_owner, job.repo_name, job.accepted_comment_id, body - ) +async def finish_job(body: str) -> None: + reporter().finish(body) diff --git a/src/agentci/workflows/dispatcher.py b/src/agentci/workflows/dispatcher.py index de93dd2..5990e91 100644 --- a/src/agentci/workflows/dispatcher.py +++ b/src/agentci/workflows/dispatcher.py @@ -1,8 +1,10 @@ from __future__ import annotations import logging +from typing import Any -from agentci.domain.models import Job, JobKind +from agentci.domain.models import JobKind +from agentci.domain.state_machine import JobState from agentci.workflows.common import Dependencies from agentci.workflows.implement import ImplementWorkflow from agentci.workflows.plan import PlanWorkflow @@ -15,7 +17,7 @@ class Dispatcher: def __init__(self, dependencies: Dependencies) -> None: plan = PlanWorkflow(dependencies) pull_request = PullRequestWorkflow(dependencies) - self.handlers = { + self.handlers: dict[JobKind, Any] = { JobKind.PLAN: plan.plan, JobKind.DISCUSS: plan.discuss, JobKind.ITERATE_PLAN: plan.iterate, @@ -24,7 +26,9 @@ class Dispatcher: JobKind.FIX: pull_request.fix, } - async def dispatch(self, job: Job) -> None: + async def dispatch(self, job: JobState) -> None: + if job.kind is None: + raise RuntimeError("Cannot dispatch an unparsed command") extra = { "operation": "workflow.dispatch", "job_id": job.id, diff --git a/src/agentci/workflows/implement.py b/src/agentci/workflows/implement.py index 661d65e..e5782a3 100644 --- a/src/agentci/workflows/implement.py +++ b/src/agentci/workflows/implement.py @@ -9,15 +9,16 @@ from agentci.domain.models import ( WorkflowKind, WorkflowStatus, ) +from agentci.reporting import reporter from agentci.workflows.change_set import ChangeSet, pull_request_body, result_comment from agentci.workflows.code_review import CodeReviewLoop from agentci.workflows.common import ( Dependencies, JobRejected, agent_comment, + finish_job, report_json, review_markdown, - update_job_comment, ) @@ -37,7 +38,7 @@ class ImplementWorkflow: f"{workflow_id[:8]}" ) workspace = self.deps.settings.workspaces_dir / workflow_id / "repo" - await self.deps.storage.update_job(job.id, stage="cloning") + await reporter().progress("cloning") base_sha = await self.deps.git.clone( job.repo_owner, job.repo_name, @@ -55,15 +56,9 @@ class ImplementWorkflow: base_sha=base_sha, branch=branch, ) - await self.deps.storage.create_workflow(workflow) - job.workflow_id = workflow.id - await self.deps.storage.update_job( - job.id, workflow_id=workflow.id, stage="installing development environment" - ) + await reporter().create_workflow(workflow, "installing development environment") await self.deps.development.prepare(workspace) - await self.deps.storage.update_job( - job.id, workflow_id=workflow.id, stage="implementing" - ) + await reporter().progress("implementing") context = await self.deps.context.issue_context( job.repo_owner, job.repo_name, job.issue_number ) @@ -80,7 +75,7 @@ class ImplementWorkflow: session_id = await self.deps.opencode.create_session(workspace, "implementation") workflow.primary_session_id = session_id await self.deps.storage.update_workflow(workflow) - await self.deps.storage.update_job(job.id, runtime_session_id=session_id) + await reporter().link_runtime_session(session_id) result = await self.deps.opencode.resume( session_id=session_id, workspace=workspace, @@ -107,7 +102,7 @@ class ImplementWorkflow: set_upstream=True, commit_prefix="agent", ) - await self.deps.storage.update_job(job.id, stage="creating pull request") + await reporter().progress("creating pull request") pull = await self.deps.gitea.create_pull_request( job.repo_owner, job.repo_name, @@ -130,7 +125,7 @@ class ImplementWorkflow: remaining = review_markdown(report) if remaining: body = f"{body}\n\n{remaining}" - await update_job_comment(self.deps, job, body) + await finish_job(body) async def _reject_duplicate(self, job: Job) -> None: workflows = await self.deps.storage.implementation_workflows( diff --git a/src/agentci/workflows/plan.py b/src/agentci/workflows/plan.py index 3b727fa..d5a5179 100644 --- a/src/agentci/workflows/plan.py +++ b/src/agentci/workflows/plan.py @@ -11,15 +11,16 @@ from agentci.domain.models import ( WorkflowKind, WorkflowStatus, ) +from agentci.reporting import reporter from agentci.workflows.common import ( Dependencies, JobRejected, agent_comment, + finish_job, report_for_prompt, report_json, required_session, review_markdown, - update_job_comment, ) @@ -31,7 +32,7 @@ class PlanWorkflow: repository = await self.deps.gitea.repository(job.repo_owner, job.repo_name) workflow_id = str(uuid4()) workspace = self.deps.settings.workspaces_dir / workflow_id / "repo" - await self.deps.storage.update_job(job.id, stage="cloning") + await reporter().progress("cloning") base_sha = await self.deps.git.clone( job.repo_owner, job.repo_name, @@ -47,9 +48,7 @@ class PlanWorkflow: workspace_path=workspace, base_sha=base_sha, ) - await self.deps.storage.create_workflow(workflow) - job.workflow_id = workflow.id - await self.deps.storage.update_job(job.id, workflow_id=workflow.id, stage="planning") + await reporter().create_workflow(workflow, "planning") context = await self.deps.context.issue_context( job.repo_owner, job.repo_name, job.issue_number ) @@ -61,7 +60,7 @@ class PlanWorkflow: session_id = await self.deps.opencode.create_session(workspace, "plan") workflow.primary_session_id = session_id await self.deps.storage.update_workflow(workflow) - await self.deps.storage.update_job(job.id, runtime_session_id=session_id) + await reporter().link_runtime_session(session_id) artifact = await self.deps.opencode.resume( session_id=session_id, workspace=workspace, @@ -85,7 +84,7 @@ class PlanWorkflow: ) if not workflow.primary_session_id or not workflow.artifact: raise JobRejected("The latest plan cannot be resumed; start a new `/agent plan`.") - await self.deps.storage.update_job(job.id, workflow_id=workflow.id, stage="discussing") + await reporter().link_workflow(workflow.id, "discussing") prompt = self.deps.prompts.render( "discuss", artifact=workflow.artifact, message=job.message ) @@ -98,9 +97,7 @@ class PlanWorkflow: schema_name="discussion.json", result_type=DiscussionReply, ) - await update_job_comment( - self.deps, job, agent_comment("discussion", workflow.id, reply.markdown) - ) + await finish_job(agent_comment("discussion", workflow.id, reply.markdown)) async def iterate(self, job: Job) -> None: await self._reject_if_active_or_merged_pr(job) @@ -111,9 +108,7 @@ class PlanWorkflow: raise JobRejected("The latest plan is missing resumable sessions; start a new plan.") if not workflow.artifact: raise JobRejected("The latest plan has no saved artifact.") - await self.deps.storage.update_job( - job.id, workflow_id=workflow.id, stage="iterating plan" - ) + await reporter().link_workflow(workflow.id, "iterating plan") context = await self.deps.context.issue_context( job.repo_owner, job.repo_name, job.issue_number ) @@ -141,10 +136,8 @@ class PlanWorkflow: ) -> ReviewReport: report = ReviewReport(summary="", findings=[]) for round_index in range(self.deps.settings.plan_review_rounds): - await self.deps.storage.update_job( - job.id, - stage=f"reviewing plan {round_index + 1}/" - f"{self.deps.settings.plan_review_rounds}", + await reporter().progress( + f"reviewing plan {round_index + 1}/{self.deps.settings.plan_review_rounds}" ) report = await self._review(workflow, context, artifact) workflow.artifact = artifact.plan_markdown @@ -215,7 +208,7 @@ class PlanWorkflow: remaining = review_markdown(report) if remaining: body = f"{body}\n\n{remaining}" - await update_job_comment(self.deps, job, body) + await finish_job(body) async def _latest_plan(self, job: Job) -> Workflow: workflow = await self.deps.storage.latest_workflow( diff --git a/src/agentci/workflows/pull_request.py b/src/agentci/workflows/pull_request.py index f95dbcb..cf65fc8 100644 --- a/src/agentci/workflows/pull_request.py +++ b/src/agentci/workflows/pull_request.py @@ -1,16 +1,17 @@ from __future__ import annotations from agentci.domain.models import AgentResult, Job, WorkflowKind, WorkflowStatus +from agentci.reporting import reporter from agentci.workflows.change_set import ChangeSet, result_comment from agentci.workflows.code_review import CodeReviewLoop from agentci.workflows.common import ( Dependencies, JobRejected, agent_comment, + finish_job, report_for_prompt, report_json, review_markdown, - update_job_comment, ) @@ -40,16 +41,11 @@ class PullRequestWorkflow: raise JobRejected("Implementation iteration requires an open pull request.") if workflow.branch != pull.head_branch: raise JobRejected("The pull request head branch no longer matches its workflow.") - await self.deps.storage.update_job( - job.id, workflow_id=workflow.id, stage="synchronizing branch" - ) - job.workflow_id = workflow.id + await reporter().link_workflow(workflow.id, "synchronizing branch") await self.deps.git.sync_branch(workflow.workspace_path, pull.head_branch) - await self.deps.storage.update_job( - job.id, stage="installing development environment" - ) + await reporter().progress("installing development environment") await self.deps.development.prepare(workflow.workspace_path) - await self.deps.storage.update_job(job.id, stage="implementing iteration") + await reporter().progress("implementing iteration") prompt = self.deps.prompts.render( "implementation_iterate", context=context, @@ -95,7 +91,7 @@ class PullRequestWorkflow: remaining = review_markdown(report) if remaining: body = f"{body}\n\n{remaining}" - await update_job_comment(self.deps, job, body) + await finish_job(body) async def fix(self, job: Job) -> None: pull_number = _pull_number(job) @@ -105,16 +101,14 @@ class PullRequestWorkflow: if not pull.is_open: raise JobRejected("Fixes require an open pull request.") workspace = self.deps.settings.workspaces_dir / f"fix-{job.id}" / "repo" - await self.deps.storage.update_job(job.id, stage="cloning pull request") + await reporter().progress("cloning pull request") await self.deps.git.clone( pull.head_owner, pull.head_repo, pull.head_branch, workspace, ) - await self.deps.storage.update_job( - job.id, stage="installing development environment" - ) + await reporter().progress("installing development environment") await self.deps.development.prepare(workspace) prompt = self.deps.prompts.render( "fix", @@ -122,9 +116,9 @@ class PullRequestWorkflow: message=job.message or "(address the pull request feedback)", development_environment=self.deps.development.description, ) - await self.deps.storage.update_job(job.id, stage="fixing") + await reporter().progress("fixing") session_id = await self.deps.opencode.create_session(workspace, "fix") - await self.deps.storage.update_job(job.id, runtime_session_id=session_id) + await reporter().link_runtime_session(session_id) result = await self.deps.opencode.resume( session_id=session_id, workspace=workspace, @@ -142,11 +136,7 @@ class PullRequestWorkflow: set_upstream=False, commit_prefix="agent fix", ) - await update_job_comment( - self.deps, - job, - agent_comment("fix", job.id, result_comment(result, sha=sha)), - ) + await finish_job(agent_comment("fix", job.id, result_comment(result, sha=sha))) def _pull_number(job: Job) -> int: diff --git a/tests/test_config.py b/tests/test_config.py index 9661146..23d649f 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -27,7 +27,8 @@ def test_empty_install_scripts_disable_setup() -> None: assert settings.install_scripts == [] -def test_defaults_research_variant_to_high() -> None: +def test_defaults_research_variant_to_high(monkeypatch) -> None: + monkeypatch.delenv("AGENTCI_RESEARCH_VARIANT", raising=False) settings = Settings(_env_file=None) # type: ignore[call-arg] assert settings.research_variant == "high" diff --git a/tests/test_opencode_abort.py b/tests/test_opencode_abort.py new file mode 100644 index 0000000..d584216 --- /dev/null +++ b/tests/test_opencode_abort.py @@ -0,0 +1,34 @@ +from pathlib import Path + +import httpx +import pytest + +from agentci.adapters.opencode import OpenCodeClient, OpenCodeError + + +def client(tmp_path: Path, status: int) -> OpenCodeClient: + return OpenCodeClient( + base_url="http://opencode:4096", + username="opencode", + password="secret", + schemas_dir=tmp_path, + health_directory=tmp_path, + required_models=(), + timeout_seconds=60, + transport=httpx.MockTransport(lambda _request: httpx.Response(status)), + ) + + +@pytest.mark.parametrize("status", [200, 204, 404, 409]) +async def test_absent_or_inactive_session_is_success(tmp_path: Path, status: int) -> None: + value = client(tmp_path, status) + await value.abort("session", tmp_path) + await value.close() + + +@pytest.mark.parametrize("status", [400, 429, 500]) +async def test_abort_failure_is_visible_for_retry(tmp_path: Path, status: int) -> None: + value = client(tmp_path, status) + with pytest.raises(OpenCodeError): + await value.abort("session", tmp_path) + await value.close() diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py new file mode 100644 index 0000000..2670a77 --- /dev/null +++ b/tests/test_state_machine.py @@ -0,0 +1,75 @@ +from dataclasses import FrozenInstanceError + +import pytest + +from agentci.domain.events import ( + CommandReceived, + CommentLinked, + JobCompleted, + JobStarted, + PermissionGranted, + ServiceRestarted, +) +from agentci.domain.models import JobKind, JobStatus +from agentci.domain.state_machine import InvalidTransition, next_state, render_job_comment + + +def received(body: str = "/agent plan message"): + return next_state( + None, + CommandReceived( + job_id="job", + delivery_id="delivery", + receive_sequence=1, + command_body=body, + target_key="org/repo:issue:1", + repo_owner="org", + repo_name="repo", + issue_number=1, + pr_number=None, + requester="alice", + comment_id=4, + ), + ) + + +def test_permission_parses_and_queues_execution() -> None: + transition = next_state(received().state, PermissionGranted(job_id="job")) + assert transition.state.status is JobStatus.QUEUED + assert transition.state.kind is JobKind.PLAN + assert transition.state.message == "message" + assert [(item.listener, item.queue) for item in transition.notifications] == [ + ("execute", "jobs"), + ("reconcile_comment", "control"), + ] + + +def test_invalid_syntax_is_rejected_after_permission() -> None: + transition = next_state(received("/agent nonsense").state, PermissionGranted(job_id="job")) + assert transition.state.status is JobStatus.REJECTED + assert "Unknown" in (transition.state.error or "") + + +def test_running_completion_and_restart_are_explicit() -> None: + queued = next_state(received().state, PermissionGranted(job_id="job")).state + running = next_state(queued, JobStarted(job_id="job")).state + completed = next_state(running, JobCompleted(job_id="job", comment_body="# Result")).state + assert completed.status is JobStatus.SUCCEEDED + assert "# Result" in render_job_comment(completed) + assert next_state(completed, ServiceRestarted(job_id="job")).state == completed + + +def test_comment_link_is_allowed_on_terminal_state() -> None: + queued = next_state(received().state, PermissionGranted(job_id="job")).state + running = next_state(queued, JobStarted(job_id="job")).state + completed = next_state(running, JobCompleted(job_id="job", comment_body="ok")).state + linked = next_state(completed, CommentLinked(job_id="job", comment_id=9)).state + assert linked.accepted_comment_id == 9 + + +def test_state_is_immutable_and_invalid_transitions_fail() -> None: + state = received().state + with pytest.raises(FrozenInstanceError): + state.stage = "changed" # type: ignore[misc] + with pytest.raises(InvalidTransition): + next_state(state, JobStarted(job_id="job")) diff --git a/tests/test_storage.py b/tests/test_storage.py index 8b61c62..78a1b5f 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -4,14 +4,14 @@ from pathlib import Path import pytest from agentci.adapters.storage import Storage -from agentci.domain.models import ( - Job, - JobKind, - JobStatus, - Workflow, - WorkflowKind, - WorkflowStatus, +from agentci.domain.events import ( + JobStarted, + PermissionDenied, + PermissionGranted, + WorkflowCreated, ) +from agentci.domain.models import CommandEvent, Workflow, WorkflowKind, WorkflowStatus +from agentci.state_machine import StateMachine @pytest.fixture @@ -22,40 +22,60 @@ async def storage(tmp_path: Path) -> Storage: return value -def make_job(job_id: str = "job-1") -> Job: - return Job( - id=job_id, - kind=JobKind.PLAN, - target_key="alice/repo:issue:3", +def command(delivery: str, body: str = "/agent plan") -> CommandEvent: + return CommandEvent( + delivery_id=delivery, + comment_id=int(delivery.rsplit("-", 1)[-1]), repo_owner="alice", repo_name="repo", issue_number=3, pr_number=None, requester="alice", - message="", - comment_id=10, + body=body, ) -async def test_enqueue_is_idempotent_and_claims_fifo(storage: Storage) -> None: - assert await storage.enqueue("delivery-1", make_job()) - assert not await storage.enqueue("delivery-1", make_job("job-2")) - claimed = await storage.claim_next() - assert claimed is not None - assert claimed.id == "job-1" - assert claimed.status is JobStatus.RUNNING - assert await storage.claim_next() is None +async def test_receive_is_idempotent_without_consuming_sequence(storage: Storage) -> None: + host = StateMachine(storage) + first = await host.receive(command("delivery-1")) + duplicate = await host.receive(command("delivery-1")) + second = await host.receive(command("delivery-2")) + + assert not first.duplicate + assert duplicate.duplicate + assert duplicate.state.id == first.state.id + assert second.state.receive_sequence == first.state.receive_sequence + 1 -async def test_recovers_running_job_as_failed(storage: Storage) -> None: - await storage.enqueue("delivery-1", make_job()) - assert await storage.claim_next() is not None - recovered = await storage.recover_running() - assert [job.id for job in recovered] == ["job-1"] - assert await storage.claim_next() is None +async def test_received_job_blocks_later_execute_task(storage: Storage) -> None: + host = StateMachine(storage) + first = (await host.receive(command("delivery-1"))).state + second = (await host.receive(command("delivery-2"))).state + await host.evolve("grant-2", PermissionGranted(job_id=second.id)) + + assert await storage.claim_task("jobs") is None + + await host.evolve("deny-1", PermissionDenied(job_id=first.id)) + task = await storage.claim_task("jobs") + assert task is not None + assert task.job_id == second.id -async def test_persists_and_finds_workflows(storage: Storage, tmp_path: Path) -> None: +async def test_started_and_finished_timestamps_are_owned_by_store(storage: Storage) -> None: + host = StateMachine(storage) + state = (await host.receive(command("delivery-1"))).state + state = (await host.evolve("grant", PermissionGranted(job_id=state.id))).state + state = (await host.evolve("start", JobStarted(job_id=state.id))).state + with sqlite3.connect(storage.database_path) as connection: + row = connection.execute( + "SELECT started_at, finished_at FROM jobs WHERE id=?", (state.id,) + ).fetchone() + assert row is not None + assert row[0] is not None + assert row[1] is None + + +async def test_workflow_queries_and_completed_protection(storage: Storage, tmp_path: Path) -> None: workflow = Workflow( id="workflow-1", kind=WorkflowKind.PLAN, @@ -68,84 +88,36 @@ async def test_persists_and_finds_workflows(storage: Storage, tmp_path: Path) -> status=WorkflowStatus.COMPLETED, ) await storage.create_workflow(workflow) + await storage.fail_job_workflow("missing-job") loaded = await storage.latest_workflow("alice", "repo", 3, WorkflowKind.PLAN) assert loaded is not None - assert loaded.artifact == "# Plan" - assert loaded.workspace_path == tmp_path / "repo" + assert loaded.status is WorkflowStatus.COMPLETED -async def test_tracks_operational_comments(storage: Storage) -> None: - await storage.enqueue("delivery-1", make_job()) - await storage.set_job_comment("job-1", "accepted_comment_id", 21) - await storage.set_job_comment("job-1", "started_comment_id", 22) - assert await storage.operational_comment_ids("alice", "repo", 3) == {21, 22} - - -async def test_failed_followup_does_not_invalidate_completed_workflow( - storage: Storage, tmp_path: Path -) -> None: +async def test_workflow_creation_and_job_link_are_atomic(storage: Storage, tmp_path: Path) -> None: + host = StateMachine(storage) + state = (await host.receive(command("delivery-1"))).state + state = (await host.evolve("grant", PermissionGranted(job_id=state.id))).state + state = (await host.evolve("start", JobStarted(job_id=state.id))).state workflow = Workflow( - id="workflow-1", + id="workflow-atomic", kind=WorkflowKind.PLAN, repo_owner="alice", repo_name="repo", issue_number=3, workspace_path=tmp_path / "repo", base_sha="abc", - status=WorkflowStatus.COMPLETED, ) - await storage.create_workflow(workflow) - job = make_job() - job.workflow_id = workflow.id - await storage.enqueue("delivery-1", job) - await storage.fail_job_workflow(job.id) - loaded = await storage.latest_workflow("alice", "repo", 3, WorkflowKind.PLAN) - assert loaded is not None - assert loaded.status is WorkflowStatus.COMPLETED - - -async def test_opencode_migration_preserves_and_tags_legacy_session_ids(tmp_path: Path) -> None: - legacy_migrations = tmp_path / "legacy-migrations" - legacy_migrations.mkdir() - migrations = Path(__file__).parents[1] / "src" / "agentci" / "migrations" - (legacy_migrations / "001_initial.sql").write_text( - (migrations / "001_initial.sql").read_text() + result = await host.evolve( + "workflow-created", + WorkflowCreated(job_id=state.id, workflow=workflow, stage="planning"), ) - database = tmp_path / "legacy.sqlite3" - legacy = Storage(database, legacy_migrations) - await legacy.initialize() - with sqlite3.connect(database) as connection: - connection.execute( - """ - INSERT INTO workflows ( - id, kind, repo_owner, repo_name, issue_number, base_sha, - workspace_path, primary_session_id, reviewer_session_id, - artifact, status, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - "legacy-workflow", - "plan", - "alice", - "repo", - 3, - "abc", - str(tmp_path / "repo"), - "legacy-primary", - "legacy-reviewer", - "# Preserved plan", - "completed", - "2026-07-20T00:00:00+00:00", - "2026-07-20T00:00:00+00:00", - ), - ) + assert result.state.workflow_id == workflow.id + assert await storage.get_workflow(workflow.id) is not None - migrated = Storage(database, migrations) - await migrated.initialize() - loaded = await migrated.latest_workflow("alice", "repo", 3, WorkflowKind.PLAN) - assert loaded is not None - assert loaded.artifact == "# Preserved plan" - assert loaded.primary_session_id == "legacy-primary" - assert loaded.reviewer_session_id == "legacy-reviewer" - assert loaded.runtime == "codex" +async def test_schema_has_receive_sequence_and_no_version(storage: Storage) -> None: + with sqlite3.connect(storage.database_path) as connection: + columns = {row[1] for row in connection.execute("PRAGMA table_info(jobs)")} + assert "receive_sequence" in columns + assert "version" not in columns diff --git a/tests/test_webhook.py b/tests/test_webhook.py index 4042c69..4673375 100644 --- a/tests/test_webhook.py +++ b/tests/test_webhook.py @@ -1,54 +1,29 @@ import hashlib import hmac -import json from types import SimpleNamespace +import pytest +from fastapi import HTTPException + from agentci.api.webhook import _event_from_payload, _handle_command, valid_signature -class FakeStorage: - def __init__(self) -> None: - self.jobs = [] - self.deliveries: set[str] = set() +class FakeHost: + def __init__(self, duplicate: bool = False) -> None: + self.events = [] + self.duplicate = duplicate - async def enqueue(self, delivery_id, job): - if delivery_id in self.deliveries: - return False - self.deliveries.add(delivery_id) - self.jobs.append(job) - return True - - async def record_delivery(self, delivery_id, _comment_id): - if delivery_id in self.deliveries: - return False - self.deliveries.add(delivery_id) - return True - - async def set_job_comment(self, *_args): - return None - - -class FakeGitea: - def __init__(self, permitted: bool = True) -> None: - self.permitted = permitted - self.comments: list[str] = [] - - async def has_write_permission(self, *_args): - return self.permitted - - async def create_comment(self, _owner, _repo, _number, body): - self.comments.append(body) - return len(self.comments) + async def receive(self, event): + self.events.append(event) + state = SimpleNamespace(id="job", receive_sequence=1) + return SimpleNamespace(state=state, duplicate=self.duplicate) def payload(body: str, *, is_pull: bool = False) -> dict: value = { "action": "created", "comment": {"id": 8, "body": body, "user": {"login": "alice"}}, - "repository": { - "name": "repo", - "owner": {"login": "org"}, - }, + "repository": {"name": "repo", "owner": {"login": "org"}}, "issue": {"number": 4}, "is_pull": is_pull, } @@ -57,74 +32,40 @@ def payload(body: str, *, is_pull: bool = False) -> dict: return value -def test_extracts_pull_request_event() -> None: - event = _event_from_payload("delivery", payload("/agent fix now", is_pull=True)) - assert event is not None - assert event.pr_number == 4 - assert event.target_key == "org/repo:pr:4" - - -async def test_authorized_command_is_queued() -> None: - storage = FakeStorage() - gitea = FakeGitea() - container = SimpleNamespace(storage=storage, gitea=gitea) - event = _event_from_payload("delivery", payload("/agent plan consider migrations")) - assert event is not None - response = await _handle_command(container, event) - assert response.status_code == 202 - assert len(storage.jobs) == 1 - assert "queued" in gitea.comments[0] - - -async def test_unauthorized_command_is_rejected_and_deduplicated() -> None: - storage = FakeStorage() - gitea = FakeGitea(permitted=False) - container = SimpleNamespace(storage=storage, gitea=gitea) - event = _event_from_payload("delivery", payload("/agent implement")) - assert event is not None - - await _handle_command(container, event) - await _handle_command(container, event) - - assert storage.jobs == [] - assert len(gitea.comments) == 1 - assert "write permission" in gitea.comments[0] - - -async def test_iterate_message_is_preserved_on_queued_job() -> None: - storage = FakeStorage() - container = SimpleNamespace(storage=storage, gitea=FakeGitea()) +async def test_command_is_forwarded_without_parsing() -> None: + host = FakeHost() event = _event_from_payload( - "delivery", - payload( - "/agent iterate\n\nkeep the API stable\nlimit changes to the parser", - is_pull=True, - ), + "delivery", payload("/agent iterate\n\nkeep raw body", is_pull=True) ) assert event is not None - - response = await _handle_command(container, event) - + response = await _handle_command(SimpleNamespace(state_machine=host), event) assert response.status_code == 202 - assert len(storage.jobs) == 1 - assert storage.jobs[0].message == ( - "keep the API stable\nlimit changes to the parser" - ) + assert host.events[0].body == "/agent iterate\n\nkeep raw body" + + +async def test_duplicate_returns_200() -> None: + event = _event_from_payload("delivery", payload("/agent plan")) + assert event is not None + response = await _handle_command(SimpleNamespace(state_machine=FakeHost(True)), event) + assert response.status_code == 200 + + +async def test_missing_delivery_is_rejected() -> None: + event = _event_from_payload("", payload("/agent plan")) + assert event is not None + with pytest.raises(HTTPException) as raised: + await _handle_command(SimpleNamespace(state_machine=FakeHost()), event) + assert raised.value.status_code == 400 async def test_non_command_is_ignored() -> None: - container = SimpleNamespace(storage=FakeStorage(), gitea=FakeGitea()) event = _event_from_payload("delivery", payload("ordinary discussion")) assert event is not None - response = await _handle_command(container, event) + response = await _handle_command(SimpleNamespace(state_machine=FakeHost()), event) assert response.status_code == 204 -def test_rejects_bad_signature() -> None: +def test_signature_validation() -> None: + signature = hmac.new(b"secret", b"{}", hashlib.sha256).hexdigest() + assert valid_signature(b"secret", b"{}", signature) assert not valid_signature(b"secret", b"{}", "bad") - - -def test_accepts_valid_signature() -> None: - body = json.dumps(payload("ordinary comment")).encode() - signature = hmac.new(b"secret", body, hashlib.sha256).hexdigest() - assert valid_signature(b"secret", body, signature) diff --git a/tests/test_worker.py b/tests/test_worker.py index 9686980..a0a9184 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -1,96 +1,42 @@ from pathlib import Path +from types import SimpleNamespace +from typing import cast -from agentci.domain.models import Job, JobKind, JobStatus, Workflow, WorkflowKind -from agentci.worker import Worker -from agentci.workflows.common import JobRejected +from agentci.domain.models import Workflow, WorkflowKind +from agentci.domain.state_machine import JobState +from agentci.worker import Worker, _safe_error class FakeStorage: - def __init__(self, workflow: Workflow | None) -> None: + def __init__(self, workflow=None) -> None: self.workflow = workflow - async def get_workflow(self, _workflow_id: str) -> Workflow | None: + async def get_workflow(self, _workflow_id): return self.workflow class FakeOpenCode: def __init__(self) -> None: - self.aborted: set[tuple[str, Path]] = set() + self.aborted = set() - async def abort(self, session_id: str, workspace: Path) -> None: + async def abort(self, session_id, workspace): self.aborted.add((session_id, workspace)) -class JobStorage(FakeStorage): - def __init__(self) -> None: - super().__init__(None) - self.updates: list[tuple[JobStatus | None, str | None, str | None]] = [] - self.comment_ids: list[int] = [] - - async def update_job(self, _job_id: str, **values) -> None: - self.updates.append( - (values.get("status"), values.get("stage"), values.get("error")) - ) - - async def set_job_comment(self, _job_id: str, _column: str, comment_id: int) -> None: - self.comment_ids.append(comment_id) - - async def job_stage(self, _job_id: str) -> str: - return "working" - - async def fail_job_workflow(self, _job_id: str) -> None: - return None - - -class FakeGitea: - def __init__(self) -> None: - self.created: list[str] = [] - self.updated: list[tuple[int, str]] = [] - - async def create_comment(self, _owner: str, _repo: str, _number: int, body: str) -> int: - self.created.append(body) - return 42 - - async def update_comment( - self, _owner: str, _repo: str, comment_id: int, body: str - ) -> None: - self.updated.append((comment_id, body)) - - -class RejectingDispatcher: - async def dispatch(self, _job: Job) -> None: - raise JobRejected("not applicable") - - -def job(*, workflow_id: str | None, runtime_session_id: str | None = None) -> Job: - return Job( - id="job", - kind=JobKind.FIX, - target_key="org/repo:pr:1", - repo_owner="org", - repo_name="repo", - issue_number=1, - pr_number=1, - requester="alice", - message="", - comment_id=1, - workflow_id=workflow_id, - runtime_session_id=runtime_session_id, - ) - - def worker(tmp_path: Path, storage: FakeStorage, opencode: FakeOpenCode) -> Worker: return Worker( storage=storage, # type: ignore[arg-type] - gitea=None, # type: ignore[arg-type] + state_machine=SimpleNamespace(), # type: ignore[arg-type] + gitea=SimpleNamespace(), # type: ignore[arg-type] opencode=opencode, # type: ignore[arg-type] - dispatcher=None, # type: ignore[arg-type] + dispatcher=SimpleNamespace(), # type: ignore[arg-type] poll_seconds=1, workspaces_dir=tmp_path, + bot_username="agentci", ) -async def test_recovery_aborts_all_workflow_sessions(tmp_path: Path) -> None: +async def test_abort_collects_all_workflow_sessions(tmp_path: Path) -> None: workspace = tmp_path / "workflow" / "repo" workflow = Workflow( id="flow", @@ -104,63 +50,23 @@ async def test_recovery_aborts_all_workflow_sessions(tmp_path: Path) -> None: reviewer_session_id="reviewer", ) opencode = FakeOpenCode() - + state = SimpleNamespace(workflow_id="flow", runtime_session_id=None, id="job") await worker(tmp_path, FakeStorage(workflow), opencode)._abort_job_sessions( - job(workflow_id=workflow.id) + cast(JobState, state) ) - assert opencode.aborted == {("primary", workspace), ("reviewer", workspace)} -async def test_recovery_aborts_one_shot_fix_session(tmp_path: Path) -> None: +async def test_abort_uses_one_shot_fix_workspace(tmp_path: Path) -> None: opencode = FakeOpenCode() - - await worker(tmp_path, FakeStorage(None), opencode)._abort_job_sessions( - job(workflow_id=None, runtime_session_id="fix-session") + state = SimpleNamespace(workflow_id=None, runtime_session_id="session", id="job") + await worker(tmp_path, FakeStorage(), opencode)._abort_job_sessions( + cast(JobState, state) ) - - assert opencode.aborted == {("fix-session", tmp_path / "fix-job" / "repo")} + assert opencode.aborted == {("session", tmp_path / "fix-job" / "repo")} -async def test_job_status_updates_existing_gitea_comment(tmp_path: Path) -> None: - storage = JobStorage() - gitea = FakeGitea() - active_job = job(workflow_id=None) - active_job.accepted_comment_id = 41 - value = Worker( - storage=storage, # type: ignore[arg-type] - gitea=gitea, # type: ignore[arg-type] - opencode=FakeOpenCode(), # type: ignore[arg-type] - dispatcher=RejectingDispatcher(), # type: ignore[arg-type] - poll_seconds=1, - workspaces_dir=tmp_path, - ) - - await value._run_job(active_job) - - assert gitea.created == [] - assert [comment_id for comment_id, _body in gitea.updated] == [41, 41] - assert "started" in gitea.updated[0][1] - assert "rejected" in gitea.updated[1][1] - assert storage.updates[-1][:2] == (JobStatus.REJECTED, "rejected") - - -async def test_worker_creates_only_one_comment_when_queue_comment_is_missing( - tmp_path: Path, -) -> None: - storage = JobStorage() - gitea = FakeGitea() - value = Worker( - storage=storage, # type: ignore[arg-type] - gitea=gitea, # type: ignore[arg-type] - opencode=FakeOpenCode(), # type: ignore[arg-type] - dispatcher=RejectingDispatcher(), # type: ignore[arg-type] - poll_seconds=1, - workspaces_dir=tmp_path, - ) - - await value._run_job(job(workflow_id=None)) - - assert len(gitea.created) == 1 - assert storage.comment_ids == [42] - assert [comment_id for comment_id, _body in gitea.updated] == [42] +def test_safe_error_is_single_line_and_bounded() -> None: + value = _safe_error(RuntimeError("bad\n" + "x" * 2000)) + assert "\n" not in value + assert len(value) == 1000 -- 2.54.0