agent: Implemented the explicit persisted webhook state machine.

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
+The requester must have Gitea `write`, `admin`, or `owner` permission on the repository. Commands
-accepted command gets separate queued and started comments. Final plans, PR results, failures, and
+are durably sequenced when their webhook arrives, then authorized and executed in that receive
-remaining review findings are posted separately.
+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
+new plan or implementation. Queued jobs survive restart. A restart before `JobStarted` returns the
-and marked failed instead of being replayed because a partial model turn may already have changed
+task to the queue; after `JobStarted`, the job is failed, its sessions are aborted, and execution is
-files. Git pushes are never forced.
+never replayed because a partial model turn may already have changed files. Git pushes are never
+forced.
 
 ## Development
 

src/agentci/adapters/gitea.py

@@ -105,6 +105,15 @@ class GiteaClient:
         )
         return int(response.json()["id"])
 
+    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,
         owner: str,

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:
+    async def receive(
-        def record(connection: sqlite3.Connection) -> bool:
+        self, event_id: str, job_id: str, incoming: CommandEvent
-            try:
+    ) -> EvolveResult:
-                connection.execute(
+        def operation(connection: sqlite3.Connection) -> EvolveResult:
-                    "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(
+            duplicate = connection.execute(
-            "SELECT * FROM jobs WHERE status = ? ORDER BY created_at LIMIT 1",
+                "SELECT job_id FROM job_events WHERE event_id=?", (event_id,)
-            (JobStatus.QUEUED,),
             ).fetchone()
-        if row is None:
+            if duplicate:
+                state = _state(connection, duplicate["job_id"])
                 connection.commit()
-            return None
+                return EvolveResult(state, True)
-        connection.execute(
+            sequence = connection.execute(
-            "UPDATE jobs SET status = ?, stage = ?, started_at = ? WHERE id = ?",
+                "SELECT COALESCE(MAX(receive_sequence), 0) + 1 FROM jobs"
-            (JobStatus.RUNNING, "starting", now(), row["id"]),
+            ).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(
+        return await self._run(operation)
-        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,
-            },
-        )
 
-    async def set_job_comment(self, job_id: str, column: str, comment_id: int) -> None:
+    async def evolve(self, event_id: str, event: JobEvent) -> EvolveResult:
-        if column not in {"accepted_comment_id", "started_comment_id"}:
+        def operation(connection: sqlite3.Connection) -> EvolveResult:
-            raise ValueError("Unsupported comment column")
+            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:
+    async def retry_task(self, task_id: int, attempts: int, error: str) -> None:
-        def select(connection: sqlite3.Connection) -> str:
+        delay = min(2 ** min(attempts, 8), 300)
-            row = connection.execute("SELECT stage FROM jobs WHERE id=?", (job_id,)).fetchone()
+        available = (datetime.now(UTC) + timedelta(seconds=delay)).isoformat()
-            return str(row["stage"]) if row else "unknown"
+        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 "
-                    SELECT accepted_comment_id, started_comment_id FROM jobs
+                    "WHERE repo_owner=? AND repo_name=? AND issue_number=?",
-                    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]:
+def _claim_task(connection: sqlite3.Connection, queue: str) -> ListenerTask | None:
-            rows = connection.execute(
+    connection.execute("BEGIN IMMEDIATE")
-                "SELECT * FROM jobs WHERE status=?", (JobStatus.RUNNING,)
+    fifo = ""
-            ).fetchall()
+    if queue == "jobs":
-            with connection:
+        fifo = """AND NOT EXISTS (
-                connection.execute(
+            SELECT 1 FROM jobs earlier WHERE earlier.receive_sequence < j.receive_sequence
-                    """
+            AND earlier.status IN ('received', 'queued', 'running'))"""
-                    UPDATE jobs SET status=?, stage=?, error=?, finished_at=?
+    row = connection.execute(
-                    WHERE status=?
+        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""",
-                        JobStatus.FAILED,
+        (queue, now()),
-                        "interrupted",
+    ).fetchone()
-                        "Service restarted during an active OpenCode turn",
+    if row is None:
-                        now(),
+        connection.commit()
-                        JobStatus.RUNNING,
+        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]
+    if changed.rowcount != 1:
-
+        connection.rollback()
-        return await self._run(recover)
+        return None
-
+    connection.commit()
-
+    return ListenerTask(
-def job_from_row(
+        row["id"], row["job_id"], row["source_event_id"], row["listener"],
-    row: sqlite3.Row,
+        row["queue"], row["attempts"] + 1,
-    *,
-    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"],
     )

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,
+        self, method: str, path: str, *,
-        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(
+            response = await self.client.post(
-                f"/session/{session_id}/abort",
+                f"/session/{session_id}/abort", headers=directory_headers(workspace), timeout=10
-                headers=directory_headers(workspace),
-                timeout=10,
             )
-        except httpx.HTTPError:
+            if response.is_success or response.status_code in {404, 409}:
-            log.exception("OpenCode session could not be aborted")
+                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

@@ -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

@@ -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
-from agentci.domain.models import CommandEvent, Job, JobStatus
 
 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)
+        result = await container.state_machine.receive(event)
-    except CommandError as exc:
+    except Exception:
-        log.info("agent command rejected: invalid syntax", extra=extra)
+        log.exception("could not persist command", extra=extra)
-        if await container.storage.record_delivery(event.delivery_id, event.comment_id):
+        raise
-            await container.gitea.create_comment(
+    if result.duplicate:
-                event.repo_owner, event.repo_name, event.issue_number, str(exc)
+        log.info("duplicate command ignored", extra={**extra, "job_id": result.state.id})
-            )
-        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})
         return Response(status_code=status.HTTP_200_OK)
     log.info(
-        "agent job queued",
+        "agent command persisted",
-        extra={**extra, "job_id": job.id, "stage": job.kind.value},
+        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

@@ -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

@@ -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

@@ -25,6 +25,7 @@ class JobKind(StrEnum):
 
 
 class JobStatus(StrEnum):
+    RECEIVED = "received"
     QUEUED = "queued"
     RUNNING = "running"
     SUCCEEDED = "succeeded"

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"<!-- 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

@@ -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

@@ -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

@@ -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

@@ -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,157 +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._recover()
-        await self._report_interrupted()
+        await asyncio.gather(self._loop("control", stop), self._loop("jobs", stop))
-        try:
+
+    async def _loop(self, queue: str, stop: asyncio.Event) -> None:
         while not stop.is_set():
-                if not await self.opencode.ready():
+            if queue == "jobs" and 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()
+            task = await self.storage.claim_task(queue)
-                if job is None:
+            if task is None:
                 await self._wait(stop)
                 continue
-                await self._run_job(job)
+            try:
-        except Exception:
+                await self._handle(task)
-            log.exception("worker stopped unexpectedly", extra={"operation": "worker.run"})
+            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:
-                accepted_id = await self.gitea.create_comment(
-                    job.repo_owner,
-                    job.repo_name,
-                    job.issue_number,
-                    f"Agent job `{job.id}` queued (`{job.kind}`).",
-                )
-                await self.storage.set_job_comment(
-                    job.id, "accepted_comment_id", accepted_id
-                )
-            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, "started_comment_id", comment_id)
-            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_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)
+                log.exception(
-            await self._safe_update_job(
+                    "listener failed",
-                job, status=JobStatus.FAILED, stage="failed", error=_safe_error(exc)
+                    extra={"task_id": task.id, "listener": task.listener, "queue": queue},
                 )
-            await self._safe_fail_workflow(job)
+                await self.storage.retry_task(task.id, task.attempts, _safe_error(exc))
-            await self._safe_comment(
-                job,
-                f"Agent job `{job.id}` failed during `{failed_stage}`: {_safe_error(exc)}",
-            )
-            log.exception("job failed", extra={**extra, "stage": failed_stage})
             else:
-            await self._safe_update_job(
+                await self.storage.complete_task(task.id)
-                job, status=JobStatus.SUCCEEDED, stage="completed"
-            )
-            log.info("job completed", extra=extra)
 
-    async def _report_interrupted(self) -> None:
+    async def _handle(self, task: ListenerTask) -> None:
-        jobs = await self.storage.recover_running()
+        state = await self.host.get(task.job_id)
-        if jobs:
+        if state is None:
-            log.warning(
+            return
-                "recovering interrupted jobs",
+        if task.listener == "authorize":
-                extra={"operation": "worker.recover", "item_count": len(jobs)},
+            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 _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
         )
-        for job in jobs:
+        event = (
-            await self._abort_job_sessions(job)
+            PermissionGranted(job_id=state.id)
-            await self._safe_fail_workflow(job)
+            if permitted
-            await self._safe_comment(
+            else PermissionDenied(job_id=state.id)
-                job,
+        )
-                f"Agent job `{job.id}` failed because the service restarted during execution.",
+        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)
+            )
+            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"<!-- 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(
+            latest.repo_owner, latest.repo_name, comment_id, body
         )
 
-    async def _abort_job_sessions(self, job: Job) -> None:
+    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, state: JobState) -> None:
         sessions: set[tuple[str, Path]] = set()
-        workflow = None
+        workflow = await self.storage.get_workflow(state.workflow_id) if state.workflow_id else None
-        if job.workflow_id:
+        if workflow:
-            workflow = await self.storage.get_workflow(job.workflow_id)
-        if workflow is not None:
             sessions.update(
-                (session_id, workflow.workspace_path)
+                (session, workflow.workspace_path)
-                for session_id in (
+                for session in (workflow.primary_session_id, workflow.reviewer_session_id)
-                    workflow.primary_session_id,
+                if session
-                    workflow.reviewer_session_id,
             )
-                if session_id
+        elif state.runtime_session_id:
+            sessions.add(
+                (state.runtime_session_id, self.workspaces_dir / f"fix-{state.id}" / "repo")
             )
-        elif job.runtime_session_id:
+        for session, workspace in sessions:
-            workspace = self.workspaces_dir / f"fix-{job.id}" / "repo"
+            await self.opencode.abort(session, workspace)
-            sessions.add((job.runtime_session_id, workspace))
-        for session_id, workspace in sessions:
-            await self.opencode.abort(session_id, workspace)
-
-    async def _safe_comment(self, job: Job, body: str) -> None:
-        try:
-            await self.gitea.create_comment(
-                job.repo_owner, job.repo_name, job.issue_number, 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
-            )
-        except Exception:
-            log.exception("could not persist job status", extra={"job_id": job.id})
-
-    async def _safe_fail_workflow(self, job: Job) -> None:
-        try:
-            await self.storage.fail_job_workflow(job.id)
-        except Exception:
-            log.exception("could not mark workflow failed", extra={"job_id": job.id})
 
     async def _wait(self, stop: asyncio.Event) -> None:
         with suppress(TimeoutError):

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
 

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(
+            await reporter().progress(
-                job.id,
+                f"reviewing implementation {round_index + 1}/"
-                stage=f"reviewing implementation {round_index + 1}/"
+                f"{self.deps.settings.implement_review_rounds}"
-                f"{self.deps.settings.implement_review_rounds}",
             )
             report = await self.once(
                 workflow,

src/agentci/workflows/common.py

@@ -11,6 +11,7 @@ from agentci.adapters.storage import Storage
 from agentci.config import Settings
 from agentci.domain.models import ReviewReport
 from agentci.prompts import PromptLibrary
+from agentci.reporting import reporter
 from agentci.workflows.context import ContextBuilder
 
 
@@ -69,3 +70,7 @@ def report_for_prompt(report_json_value: str | None) -> str:
 
 def agent_comment(kind: str, workflow_id: str, body: str) -> str:
     return f"<!-- agentci:{kind} workflow={workflow_id} -->\n{body}"
+
+
+async def finish_job(body: str) -> None:
+    reporter().finish(body)

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,

src/agentci/workflows/implement.py

@@ -9,12 +9,14 @@ 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,
 )
@@ -36,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,
@@ -54,15 +56,9 @@ class ImplementWorkflow:
             base_sha=base_sha,
             branch=branch,
         )
-        await self.deps.storage.create_workflow(workflow)
+        await reporter().create_workflow(workflow, "installing development environment")
-        job.workflow_id = workflow.id
-        await self.deps.storage.update_job(
-            job.id, workflow_id=workflow.id, stage="installing development environment"
-        )
         await self.deps.development.prepare(workspace)
-        await self.deps.storage.update_job(
+        await reporter().progress("implementing")
-            job.id, workflow_id=workflow.id, stage="implementing"
-        )
         context = await self.deps.context.issue_context(
             job.repo_owner, job.repo_name, job.issue_number
         )
@@ -79,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,
@@ -106,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,
@@ -125,17 +121,11 @@ class ImplementWorkflow:
             f"{job.repo_name}/pulls/{pull.number}"
         )
         body = f"Pull request created: {pull_url}\n\n{result_comment(result, sha=sha)}"
-        await self.deps.gitea.create_comment(
+        body = agent_comment("implementation", workflow.id, body)
-            job.repo_owner,
-            job.repo_name,
-            job.issue_number,
-            agent_comment("implementation", workflow.id, body),
-        )
         remaining = review_markdown(report)
         if remaining:
-            await self.deps.gitea.create_comment(
+            body = f"{body}\n\n{remaining}"
-                job.repo_owner, job.repo_name, pull.number, remaining
+        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,10 +11,12 @@ 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,
@@ -30,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,
@@ -46,9 +48,7 @@ class PlanWorkflow:
             workspace_path=workspace,
             base_sha=base_sha,
         )
-        await self.deps.storage.create_workflow(workflow)
+        await reporter().create_workflow(workflow, "planning")
-        job.workflow_id = workflow.id
-        await self.deps.storage.update_job(job.id, workflow_id=workflow.id, stage="planning")
         context = await self.deps.context.issue_context(
             job.repo_owner, job.repo_name, job.issue_number
         )
@@ -60,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,
@@ -84,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
         )
@@ -97,12 +97,7 @@ class PlanWorkflow:
             schema_name="discussion.json",
             result_type=DiscussionReply,
         )
-        await self.deps.gitea.create_comment(
+        await finish_job(agent_comment("discussion", workflow.id, reply.markdown))
-            job.repo_owner,
-            job.repo_name,
-            job.issue_number,
-            agent_comment("discussion", workflow.id, reply.markdown),
-        )
 
     async def iterate(self, job: Job) -> None:
         await self._reject_if_active_or_merged_pr(job)
@@ -113,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(
+        await reporter().link_workflow(workflow.id, "iterating plan")
-            job.id, workflow_id=workflow.id, stage="iterating plan"
-        )
         context = await self.deps.context.issue_context(
             job.repo_owner, job.repo_name, job.issue_number
         )
@@ -143,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(
+            await reporter().progress(
-                job.id,
+                f"reviewing plan {round_index + 1}/{self.deps.settings.plan_review_rounds}"
-                stage=f"reviewing plan {round_index + 1}/"
-                f"{self.deps.settings.plan_review_rounds}",
             )
             report = await self._review(workflow, context, artifact)
             workflow.artifact = artifact.plan_markdown
@@ -213,17 +204,11 @@ class PlanWorkflow:
         workflow.review_json = report_json(report)
         workflow.status = WorkflowStatus.COMPLETED
         await self.deps.storage.update_workflow(workflow)
-        await self.deps.gitea.create_comment(
+        body = agent_comment("plan", workflow.id, artifact.plan_markdown)
-            job.repo_owner,
-            job.repo_name,
-            job.issue_number,
-            agent_comment("plan", workflow.id, artifact.plan_markdown),
-        )
         remaining = review_markdown(report)
         if remaining:
-            await self.deps.gitea.create_comment(
+            body = f"{body}\n\n{remaining}"
-                job.repo_owner, job.repo_name, job.issue_number, remaining
+        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

@@ -1,12 +1,14 @@
 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,
@@ -39,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(
+        await reporter().link_workflow(workflow.id, "synchronizing branch")
-            job.id, workflow_id=workflow.id, stage="synchronizing branch"
-        )
-        job.workflow_id = workflow.id
         await self.deps.git.sync_branch(workflow.workspace_path, pull.head_branch)
-        await self.deps.storage.update_job(
+        await reporter().progress("installing development environment")
-            job.id, stage="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,
@@ -88,17 +85,13 @@ class PullRequestWorkflow:
         workflow.artifact = result.model_dump_json()
         workflow.review_json = report_json(report)
         await self.deps.storage.update_workflow(workflow)
-        await self.deps.gitea.create_comment(
+        body = agent_comment(
-            job.repo_owner,
+            "iteration", workflow.id, result_comment(result, sha=sha)
-            job.repo_name,
-            pull_number,
-            agent_comment("iteration", workflow.id, result_comment(result, sha=sha)),
         )
         remaining = review_markdown(report)
         if remaining:
-            await self.deps.gitea.create_comment(
+            body = f"{body}\n\n{remaining}"
-                job.repo_owner, job.repo_name, pull_number, remaining
+        await finish_job(body)
-            )
 
     async def fix(self, job: Job) -> None:
         pull_number = _pull_number(job)
@@ -108,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(
+        await reporter().progress("installing development environment")
-            job.id, stage="installing development environment"
-        )
         await self.deps.development.prepare(workspace)
         prompt = self.deps.prompts.render(
             "fix",
@@ -125,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,
@@ -145,12 +136,7 @@ class PullRequestWorkflow:
             set_upstream=False,
             commit_prefix="agent fix",
         )
-        await self.deps.gitea.create_comment(
+        await finish_job(agent_comment("fix", job.id, result_comment(result, sha=sha)))
-            job.repo_owner,
-            job.repo_name,
-            pull_number,
-            agent_comment("fix", job.id, result_comment(result, sha=sha)),
-        )
 
 
 def _pull_number(job: Job) -> int:

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"
 

tests/test_gitea.py

@@ -0,0 +1,22 @@
+import json
+
+import httpx
+import respx
+
+from agentci.adapters.gitea import GiteaClient
+
+
+@respx.mock
+async def test_updates_issue_comment_by_id() -> None:
+    route = respx.patch(
+        "https://gitea.example/api/v1/repos/org/repo/issues/comments/17"
+    ).mock(return_value=httpx.Response(200, json={"id": 17}))
+    client = GiteaClient("https://gitea.example", "secret")
+
+    try:
+        await client.update_comment("org", "repo", 17, "updated status")
+    finally:
+        await client.close()
+
+    assert route.called
+    assert json.loads(route.calls[0].request.content) == {"body": "updated status"}

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()

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"))

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 (
+from agentci.domain.events import (
-    Job,
+    JobStarted,
-    JobKind,
+    PermissionDenied,
-    JobStatus,
+    PermissionGranted,
-    Workflow,
+    WorkflowCreated,
-    WorkflowKind,
-    WorkflowStatus,
 )
+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:
+def command(delivery: str, body: str = "/agent plan") -> CommandEvent:
-    return Job(
+    return CommandEvent(
-        id=job_id,
+        delivery_id=delivery,
-        kind=JobKind.PLAN,
+        comment_id=int(delivery.rsplit("-", 1)[-1]),
-        target_key="alice/repo:issue:3",
         repo_owner="alice",
         repo_name="repo",
         issue_number=3,
         pr_number=None,
         requester="alice",
-        message="",
+        body=body,
-        comment_id=10,
     )
 
 
-async def test_enqueue_is_idempotent_and_claims_fifo(storage: Storage) -> None:
+async def test_receive_is_idempotent_without_consuming_sequence(storage: Storage) -> None:
-    assert await storage.enqueue("delivery-1", make_job())
+    host = StateMachine(storage)
-    assert not await storage.enqueue("delivery-1", make_job("job-2"))
+    first = await host.receive(command("delivery-1"))
-    claimed = await storage.claim_next()
+    duplicate = await host.receive(command("delivery-1"))
-    assert claimed is not None
+    second = await host.receive(command("delivery-2"))
-    assert claimed.id == "job-1"
+
-    assert claimed.status is JobStatus.RUNNING
+    assert not first.duplicate
-    assert await storage.claim_next() is None
+    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:
+async def test_received_job_blocks_later_execute_task(storage: Storage) -> None:
-    await storage.enqueue("delivery-1", make_job())
+    host = StateMachine(storage)
-    assert await storage.claim_next() is not None
+    first = (await host.receive(command("delivery-1"))).state
-    recovered = await storage.recover_running()
+    second = (await host.receive(command("delivery-2"))).state
-    assert [job.id for job in recovered] == ["job-1"]
+    await host.evolve("grant-2", PermissionGranted(job_id=second.id))
-    assert await storage.claim_next() is None
+
+    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.status is WorkflowStatus.COMPLETED
-    assert loaded.workspace_path == tmp_path / "repo"
 
 
-async def test_tracks_operational_comments(storage: Storage) -> None:
+async def test_workflow_creation_and_job_link_are_atomic(storage: Storage, tmp_path: Path) -> None:
-    await storage.enqueue("delivery-1", make_job())
+    host = StateMachine(storage)
-    await storage.set_job_comment("job-1", "accepted_comment_id", 21)
+    state = (await host.receive(command("delivery-1"))).state
-    await storage.set_job_comment("job-1", "started_comment_id", 22)
+    state = (await host.evolve("grant", PermissionGranted(job_id=state.id))).state
-    assert await storage.operational_comment_ids("alice", "repo", 3) == {21, 22}
+    state = (await host.evolve("start", JobStarted(job_id=state.id))).state
-
-
-async def test_failed_followup_does_not_invalidate_completed_workflow(
-    storage: Storage, tmp_path: Path
-) -> None:
     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)
+    result = await host.evolve(
-    job = make_job()
+        "workflow-created",
-    job.workflow_id = workflow.id
+        WorkflowCreated(job_id=state.id, workflow=workflow, stage="planning"),
-    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",
-            ),
     )
+    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
+async def test_schema_has_receive_sequence_and_no_version(storage: Storage) -> None:
-    assert loaded.artifact == "# Preserved plan"
+    with sqlite3.connect(storage.database_path) as connection:
-    assert loaded.primary_session_id == "legacy-primary"
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(jobs)")}
-    assert loaded.reviewer_session_id == "legacy-reviewer"
+    assert "receive_sequence" in columns
-    assert loaded.runtime == "codex"
+    assert "version" not in columns

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:
+class FakeHost:
-    def __init__(self) -> None:
+    def __init__(self, duplicate: bool = False) -> None:
-        self.jobs = []
+        self.events = []
-        self.deliveries: set[str] = set()
+        self.duplicate = duplicate
 
-    async def enqueue(self, delivery_id, job):
+    async def receive(self, event):
-        if delivery_id in self.deliveries:
+        self.events.append(event)
-            return False
+        state = SimpleNamespace(id="job", receive_sequence=1)
-        self.deliveries.add(delivery_id)
+        return SimpleNamespace(state=state, duplicate=self.duplicate)
-        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)
 
 
 def payload(body: str, *, is_pull: bool = False) -> dict:
     value = {
         "action": "created",
         "comment": {"id": 8, "body": body, "user": {"login": "alice"}},
-        "repository": {
+        "repository": {"name": "repo", "owner": {"login": "org"}},
-            "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:
+async def test_command_is_forwarded_without_parsing() -> None:
-    event = _event_from_payload("delivery", payload("/agent fix now", is_pull=True))
+    host = FakeHost()
-    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())
     event = _event_from_payload(
-        "delivery",
+        "delivery", payload("/agent iterate\n\nkeep raw body", is_pull=True)
-        payload(
-            "/agent iterate\n\nkeep the API stable\nlimit changes to the parser",
-            is_pull=True,
-        ),
     )
     assert event is not None
-
+    response = await _handle_command(SimpleNamespace(state_machine=host), event)
-    response = await _handle_command(container, event)
-
     assert response.status_code == 202
-    assert len(storage.jobs) == 1
+    assert host.events[0].body == "/agent iterate\n\nkeep raw body"
-    assert storage.jobs[0].message == (
+
-        "keep the API stable\nlimit changes to the parser"
+
-    )
+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

@@ -1,54 +1,42 @@
 from pathlib import Path
+from types import SimpleNamespace
+from typing import cast
 
-from agentci.domain.models import Job, JobKind, Workflow, WorkflowKind
+from agentci.domain.models import Workflow, WorkflowKind
-from agentci.worker import Worker
+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))
 
 
-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",
@@ -62,19 +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()
-
+    state = SimpleNamespace(workflow_id=None, runtime_session_id="session", id="job")
-    await worker(tmp_path, FakeStorage(None), opencode)._abort_job_sessions(
+    await worker(tmp_path, FakeStorage(), opencode)._abort_job_sessions(
-        job(workflow_id=None, runtime_session_id="fix-session")
+        cast(JobState, state)
     )
+    assert opencode.aborted == {("session", tmp_path / "fix-job" / "repo")}
 
-    assert opencode.aborted == {("fix-session", tmp_path / "fix-job" / "repo")}
+
+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