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

agent: Implemented the explicit persisted webhook state machine.

Browse Source
This commit is contained in:
agentci
2026-07-21 14:13:58 +00:00
parent 73045258fa
commit 378e372a4b
27 changed files with 1295 additions and 762 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+13 -6
View File
@@ -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
src/agentci/adapters/gitea.py
+4 -2
View File
@@ -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,
src/agentci/adapters/job_store.py
+174 -163
View File
@@ -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:
async def receive(
self, event_id: str, job_id: str, incoming: CommandEvent
) -> EvolveResult:
def operation(connection: sqlite3.Connection) -> EvolveResult:
connection.execute("BEGIN IMMEDIATE")
row = connection.execute(
"SELECT * FROM jobs WHERE status = ? ORDER BY created_at LIMIT 1",
(JobStatus.QUEUED,),
duplicate = connection.execute(
"SELECT job_id FROM job_events WHERE event_id=?", (event_id,)
).fetchone()
if row is None:
if duplicate:
state = _state(connection, duplicate["job_id"])
connection.commit()
return None
connection.execute(
"UPDATE jobs SET status = ?, stage = ?, started_at = ? WHERE id = ?",
(JobStatus.RUNNING, "starting", now(), row["id"]),
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 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,
),
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"]),
)
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"],
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,
)
src/agentci/adapters/opencode.py
+13 -12
View File
@@ -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
src/agentci/adapters/state_persistence.py
+115
View File
@@ -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,
),
)
src/agentci/api/webhook.py
+16 -61
View File
@@ -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)
src/agentci/container.py
+6 -1
View File
@@ -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
src/agentci/domain/events.py
+104
View File
@@ -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"),
]
src/agentci/domain/models.py
+1
View File
@@ -25,6 +25,7 @@ class JobKind(StrEnum):
class JobStatus(StrEnum):
RECEIVED = "received"
QUEUED = "queued"
RUNNING = "running"
SUCCEEDED = "succeeded"
src/agentci/domain/state_machine.py
+189
View File
@@ -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"<!-- agentci:job id={state.id} -->"
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}"
src/agentci/migrations/003_state_machine.sql
+96
View File
@@ -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;
src/agentci/reporting.py
+80
View File
@@ -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()
src/agentci/state_machine.py
+33
View File
@@ -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)
src/agentci/worker.py
+148 -123
View File
@@ -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:
await self._recover()
await asyncio.gather(self._loop("control", stop), self._loop("jobs", stop))
async def _loop(self, queue: str, stop: asyncio.Event) -> None:
while not stop.is_set():
if not await self.opencode.ready():
log.warning(
"worker waiting for OpenCode provider authentication",
extra={"operation": "worker.poll"},
)
if queue == "jobs" and not await self.opencode.ready():
await self._wait(stop)
continue
job = await self.storage.claim_next()
if job is None:
task = await self.storage.claim_task(queue)
if task is None:
await self._wait(stop)
continue
await self._run_job(job)
except Exception:
log.exception("worker stopped unexpectedly", extra={"operation": "worker.run"})
try:
await self._handle(task)
except asyncio.CancelledError:
raise
finally:
log.info("worker stopped", extra={"operation": "worker.run"})
async def _run_job(self, job: Job) -> None:
extra = {"job_id": job.id, "target": job.target_key}
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
)
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)
log.exception(
"listener failed",
extra={"task_id": task.id, "listener": task.listener, "queue": queue},
)
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.retry_task(task.id, task.attempts, _safe_error(exc))
else:
await self._safe_update_job(
job, status=JobStatus.SUCCEEDED, stage="completed"
)
log.info("job completed", extra=extra)
await self.storage.complete_task(task.id)
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)},
)
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.",
)
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:
raise RuntimeError(f"Unknown listener {task.listener}")
async def _abort_job_sessions(self, job: Job) -> 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:
sessions.update(
(session_id, workflow.workspace_path)
for session_id in (
workflow.primary_session_id,
workflow.reviewer_session_id,
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
)
if session_id
event = (
PermissionGranted(job_id=state.id)
if permitted
else PermissionDenied(job_id=state.id)
)
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)
outcome = "permission-granted" if permitted else "permission-denied"
await self.host.evolve(f"task:{task.id}:{outcome}", event)
async def _safe_job_comment(self, job: Job, body: str) -> None:
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)
)
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:
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.dispatcher.dispatch(running)
except JobRejected as exc:
await self.host.evolve(
f"task:{task.id}:rejected", RejectedEvent(job_id=state.id, reason=str(exc))
)
await self.storage.set_job_comment(
job.id, "accepted_comment_id", job.accepted_comment_id
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"<!-- agentci:job id={latest.id} -->"
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(
job.repo_owner, job.repo_name, job.accepted_comment_id, body
latest.repo_owner, latest.repo_name, 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
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),
)
except Exception:
log.exception("could not persist job status", extra={"job_id": job.id})
async def _safe_fail_workflow(self, job: Job) -> None:
try:
await self.storage.fail_job_workflow(job.id)
except Exception:
log.exception("could not mark workflow failed", extra={"job_id": job.id})
async def _abort_job_sessions(self, state: JobState) -> None:
sessions: set[tuple[str, Path]] = set()
workflow = await self.storage.get_workflow(state.workflow_id) if state.workflow_id else None
if workflow:
sessions.update(
(session, workflow.workspace_path)
for session in (workflow.primary_session_id, workflow.reviewer_session_id)
if session
)
elif state.runtime_session_id:
sessions.add(
(state.runtime_session_id, self.workspaces_dir / f"fix-{state.id}" / "repo")
)
for session, workspace in sessions:
await self.opencode.abort(session, workspace)
async def _wait(self, stop: asyncio.Event) -> None:
with suppress(TimeoutError):
src/agentci/workflows/change_set.py
+4 -3
View File
@@ -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
src/agentci/workflows/code_review.py
+4 -4
View File
@@ -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,
src/agentci/workflows/common.py
+4 -7
View File
@@ -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"<!-- agentci:{kind} workflow={workflow_id} -->\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)
src/agentci/workflows/dispatcher.py
+7 -3
View File
@@ -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,
src/agentci/workflows/implement.py
+8 -13
View File
@@ -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(
src/agentci/workflows/plan.py
+11 -18
View File
@@ -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(
src/agentci/workflows/pull_request.py
+11 -21
View File
@@ -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:
tests/test_config.py
+2 -1
View File
@@ -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"
tests/test_opencode_abort.py
+34
View File
@@ -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()
tests/test_state_machine.py
+75
View File
@@ -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"))
tests/test_storage.py
+67 -95
View File
@@ -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()
)
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",
),
result = await host.evolve(
"workflow-created",
WorkflowCreated(job_id=state.id, workflow=workflow, stage="planning"),
)
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
tests/test_webhook.py
+36 -95
View File
@@ -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)
tests/test_worker.py
+25 -119
View File
@@ -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