.gitignore

@@ -2,4 +2,3 @@ __pycache__/
 *.py[cod]
 .pytest_cache/
 .release-engine/
-.stackchain-state/

README.md

@@ -23,11 +23,8 @@ Pull-request replies and mobile My Work issue and PR comments use Gitea's
 issue-comment API; mobile issue capture requires issue
 creation and assignment permission. Issue capture and authored mobile actions (issue
 comments, pull-request comments, notification replies, and reviews) persist per-draft
-idempotency keys, so retrying after a timeout, reload, process restart, or handoff to
+idempotency keys, so retrying after a timeout or reload replays a confirmed result instead
-another worker replays a confirmed result instead of posting duplicate content. Results
+of posting duplicate content. Direct API callers should preserve the `Idempotency-Key`
-are coordinated through a bounded SQLite ledger. Set `STACKCHAIN_STATE_DIR` to a
-persistent, writable service directory (or set `STACKCHAIN_IDEMPOTENCY_DB` to an explicit
-SQLite path); the local default is `.stackchain-state/idempotency.sqlite3`. Direct API callers should preserve the `Idempotency-Key`
 header with the unchanged route and payload until a `201` response is confirmed. Closing an assigned issue, native Comment,
 Approve, and Request changes reviews, and assigned-PR merge require repository
 write permission. Native Comment, Approve, and Request changes reviews support

src/idempotency.py

@@ -1,114 +0,0 @@
-import json
-import sqlite3
-import time
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Callable
-
-
-@dataclass(frozen=True)
-class Reservation:
-    state: str
-    response: Any = None
-
-
-class IdempotencyLedger:
-    """Small durable ledger for replaying successful Gitea mutations."""
-
-    def __init__(
-        self,
-        path: str | Path,
-        *,
-        ttl_seconds: float,
-        max_entries: int,
-        clock: Callable[[], float] = time.time,
-    ) -> None:
-        self.path = Path(path)
-        self.ttl_seconds = ttl_seconds
-        self.max_entries = max_entries
-        self.clock = clock
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self._initialize()
-
-    def _connect(self) -> sqlite3.Connection:
-        connection = sqlite3.connect(self.path, timeout=5)
-        connection.execute("PRAGMA busy_timeout = 5000")
-        return connection
-
-    def _initialize(self) -> None:
-        with self._connect() as connection:
-            connection.execute(
-                """
-                CREATE TABLE IF NOT EXISTS idempotency_operations (
-                    key TEXT PRIMARY KEY,
-                    fingerprint TEXT NOT NULL,
-                    status TEXT NOT NULL CHECK(status IN ('pending', 'completed')),
-                    response_json TEXT,
-                    created_at REAL NOT NULL,
-                    completed_at REAL
-                )
-                """
-            )
-            connection.execute(
-                "CREATE INDEX IF NOT EXISTS idempotency_completed_at_idx "
-                "ON idempotency_operations(completed_at)"
-            )
-
-    @staticmethod
-    def _fingerprint(value: tuple[Any, ...]) -> str:
-        return json.dumps(value, separators=(",", ":"), sort_keys=True)
-
-    def reserve(self, key: str, fingerprint: tuple[Any, ...]) -> Reservation:
-        encoded = self._fingerprint(fingerprint)
-        now = self.clock()
-        with self._connect() as connection:
-            connection.execute("BEGIN IMMEDIATE")
-            connection.execute(
-                "DELETE FROM idempotency_operations "
-                "WHERE status = 'completed' AND completed_at <= ?",
-                (now - self.ttl_seconds,),
-            )
-            row = connection.execute(
-                "SELECT fingerprint, status, response_json FROM idempotency_operations "
-                "WHERE key = ?",
-                (key,),
-            ).fetchone()
-            if row is not None:
-                if row[0] != encoded:
-                    return Reservation("conflict")
-                if row[1] == "completed":
-                    return Reservation("completed", json.loads(row[2]))
-                return Reservation("pending")
-            count = connection.execute(
-                "SELECT COUNT(*) FROM idempotency_operations"
-            ).fetchone()[0]
-            if count >= self.max_entries:
-                completed = connection.execute(
-                    "SELECT key FROM idempotency_operations "
-                    "WHERE status = 'completed' ORDER BY completed_at, rowid LIMIT 1"
-                ).fetchone()
-                if completed is None:
-                    return Reservation("busy")
-                connection.execute(
-                    "DELETE FROM idempotency_operations WHERE key = ?",
-                    (completed[0],),
-                )
-            connection.execute(
-                "INSERT INTO idempotency_operations "
-                "(key, fingerprint, status, created_at) VALUES (?, ?, 'pending', ?)",
-                (key, encoded, now),
-            )
-        return Reservation("reserved")
-
-    def clear(self) -> None:
-        with self._connect() as connection:
-            connection.execute("DELETE FROM idempotency_operations")
-
-    def complete(self, key: str, response: Any) -> None:
-        encoded = json.dumps(response, separators=(",", ":"), sort_keys=True)
-        with self._connect() as connection:
-            connection.execute(
-                "UPDATE idempotency_operations SET status = 'completed', "
-                "response_json = ?, completed_at = ? WHERE key = ?",
-                (encoded, self.clock(), key),
-            )

src/main.py

@@ -1,6 +1,5 @@
 import asyncio
 import math
-import os
 import time
 from collections.abc import Awaitable, Coroutine
 from contextlib import asynccontextmanager
@@ -25,7 +24,6 @@ from src.gitea_proxy import (
     pull_review_detail,
     repos,
 )
-from src.idempotency import IdempotencyLedger
 from src.models import Issue, PullRequest, Repo, User
 from src.suggestion_engine import compute
 from src.views import router as frontend_router
@@ -92,18 +90,12 @@ _live_section_retry_at: dict[str, float | None] = {
 }
 _live_snapshot_refreshing_sections: set[str] = set()
 _read_notification_ids: set[int] = set()
+_issue_creation_operations: dict[
+    str, tuple[tuple[Any, ...], asyncio.Task, float]
+] = {}
 _authored_action_operations: dict[
     str, tuple[tuple[Any, ...], asyncio.Task, float]
 ] = {}
-_state_dir = Path(os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state"))
-_idempotency_ledger = IdempotencyLedger(
-    os.getenv("STACKCHAIN_IDEMPOTENCY_DB", str(_state_dir / "idempotency.sqlite3")),
-    ttl_seconds=AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS,
-    max_entries=(
-        AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES
-        + ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES
-    ),
-)
 _available_issue_snapshot_task: asyncio.Task | None = None
 _available_issue_snapshot_value: list[dict] | None = None
 _available_issue_snapshot_created_at: float | None = None
@@ -244,52 +236,39 @@ async def _run_idempotent_authored_action(
     if not idempotency_key:
         return await asyncio.wait_for(operation, timeout=timeout)
 
-    reservation = _idempotency_ledger.reserve(idempotency_key, fingerprint)
+    now = time.monotonic()
-    if reservation.state == "conflict":
+    expired = [
+        key for key, (_, task, created_at) in _authored_action_operations.items()
+        if task.done() and now - created_at >= AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS
+    ]
+    for key in expired:
+        _authored_action_operations.pop(key, None)
+
+    existing = _authored_action_operations.get(idempotency_key)
+    if existing is not None:
         operation.close()
+        if existing[0] != fingerprint:
             raise HTTPException(status_code=409, detail="Idempotency key already used")
-    if reservation.state == "completed":
+        task = existing[1]
-        operation.close()
+    else:
-        return reservation.response
+        while len(_authored_action_operations) >= AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES:
-    if reservation.state == "busy":
+            completed = [
+                key for key, (_, task, _) in _authored_action_operations.items()
+                if task.done()
+            ]
+            if not completed:
                 operation.close()
                 raise HTTPException(
                     status_code=503,
                     detail="Authored action queue is busy; please retry",
                     headers={"Retry-After": "1"},
                 )
-
+            oldest = min(
-    existing = _authored_action_operations.get(idempotency_key)
+                completed, key=lambda key: _authored_action_operations[key][2]
-    if reservation.state == "pending":
-        operation.close()
-        if existing is None or existing[0] != fingerprint:
-            raise HTTPException(
-                status_code=503,
-                detail=(
-                    "This action may still be completing; verify its result before retrying"
-                ),
-                headers={"Retry-After": "5"},
             )
-        task = existing[1]
+            _authored_action_operations.pop(oldest)
-    else:
+        task = asyncio.create_task(operation)
-        async def persist_result():
+        _authored_action_operations[idempotency_key] = (fingerprint, task, now)
-            result = await operation
-            _idempotency_ledger.complete(idempotency_key, result)
-            return result
-
-        task = asyncio.create_task(persist_result())
-        _authored_action_operations[idempotency_key] = (
-            fingerprint,
-            task,
-            time.monotonic(),
-        )
-
-        def discard_completed(completed: asyncio.Task) -> None:
-            existing_operation = _authored_action_operations.get(idempotency_key)
-            if existing_operation is not None and existing_operation[1] is completed:
-                _authored_action_operations.pop(idempotency_key, None)
-
-        task.add_done_callback(discard_completed)
     return await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
 
 
@@ -1290,19 +1269,48 @@ async def create_assigned_issue(
             repository, creation.title, creation.body, login, creation.label_ids
         )
 
-    try:
+    operation = create_issue()
-        result = await _run_idempotent_authored_action(
+    if idempotency_key:
-            create_issue(),
+        now = time.monotonic()
-            idempotency_key=idempotency_key,
+        expired = [
-            fingerprint=(
+            key for key, (_, task, created_at) in _issue_creation_operations.items()
-                "issue-create",
+            if task.done()
-                repository,
+            and now - created_at >= ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS
-                creation.title,
+        ]
-                creation.body,
+        for key in expired:
-                tuple(creation.label_ids),
+            _issue_creation_operations.pop(key, None)
-            ),
+        fingerprint = (
-            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+            repository, creation.title, creation.body, tuple(creation.label_ids)
         )
+        existing = _issue_creation_operations.get(idempotency_key)
+        if existing is not None:
+            operation.close()
+            if existing[0] != fingerprint:
+                raise HTTPException(status_code=409, detail="Idempotency key already used")
+            task = existing[1]
+        else:
+            while len(_issue_creation_operations) >= ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES:
+                completed = [
+                    key for key, (_, task, _) in _issue_creation_operations.items()
+                    if task.done()
+                ]
+                if not completed:
+                    operation.close()
+                    raise HTTPException(
+                        status_code=503,
+                        detail="Issue creation is busy; please retry",
+                        headers={"Retry-After": "1"},
+                    )
+                oldest = min(
+                    completed, key=lambda key: _issue_creation_operations[key][2]
+                )
+                _issue_creation_operations.pop(oldest)
+            task = asyncio.create_task(operation)
+            _issue_creation_operations[idempotency_key] = (fingerprint, task, now)
+        operation = asyncio.shield(task)
+
+    try:
+        result = await asyncio.wait_for(operation, timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
     except HTTPException:
         raise
     except Exception:

tests/test_authored_idempotency.py

@@ -4,16 +4,13 @@ import httpx
 import pytest
 
 from src import main
-from src.idempotency import IdempotencyLedger
 
 
 @pytest.fixture(autouse=True)
 def clear_authored_operations():
     main._authored_action_operations.clear()
-    main._idempotency_ledger.clear()
     yield
     main._authored_action_operations.clear()
-    main._idempotency_ledger.clear()
 
 
 @pytest.mark.anyio
@@ -143,88 +140,3 @@ async def test_review_replays_same_key_and_rejects_changed_head(monkeypatch):
 
     assert [created.status_code, replayed.status_code, conflict.status_code] == [201, 201, 409]
     assert calls == [("abc", "approve", "Good")]
-
-
-@pytest.mark.anyio
-async def test_completed_comment_replays_after_ledger_reconstruction(monkeypatch, tmp_path):
-    calls = 0
-
-    async def assigned(_repository, _number):
-        return True
-
-    async def comment(_repository, _number, body):
-        nonlocal calls
-        calls += 1
-        return {"id": 201, "body": body}
-
-    database = tmp_path / "actions.sqlite3"
-    monkeypatch.setattr(
-        main,
-        "_idempotency_ledger",
-        IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
-        raising=False,
-    )
-    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
-    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
-    transport = httpx.ASGITransport(app=main.app)
-    headers = {"Idempotency-Key": "restart-comment-201"}
-
-    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
-        created = await client.post(
-            "/api/v1/repos/stackchain/api/issues/7/comments",
-            json={"body": "Persist me"},
-            headers=headers,
-        )
-        main._authored_action_operations.clear()
-        monkeypatch.setattr(
-            main,
-            "_idempotency_ledger",
-            IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
-        )
-        replayed = await client.post(
-            "/api/v1/repos/stackchain/api/issues/7/comments",
-            json={"body": "Persist me"},
-            headers=headers,
-        )
-
-    assert [created.status_code, replayed.status_code] == [201, 201]
-    assert replayed.json() == {"id": 201, "body": "Persist me"}
-    assert calls == 1
-
-
-@pytest.mark.anyio
-async def test_orphaned_pending_comment_fails_closed_without_upstream_retry(
-    monkeypatch, tmp_path
-):
-    calls = 0
-
-    async def assigned(_repository, _number):
-        return True
-
-    async def comment(_repository, _number, body):
-        nonlocal calls
-        calls += 1
-        return {"id": 202, "body": body}
-
-    ledger = IdempotencyLedger(
-        tmp_path / "orphan.sqlite3", ttl_seconds=600, max_entries=256
-    )
-    fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first")
-    assert ledger.reserve("orphan-comment-201", fingerprint).state == "reserved"
-    monkeypatch.setattr(main, "_idempotency_ledger", ledger)
-    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
-    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
-    transport = httpx.ASGITransport(app=main.app)
-
-    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
-        response = await client.post(
-            "/api/v1/repos/stackchain/api/issues/7/comments",
-            json={"body": "Verify first"},
-            headers={"Idempotency-Key": "orphan-comment-201"},
-        )
-
-    assert response.status_code == 503
-    assert response.json()["detail"] == (
-        "This action may still be completing; verify its result before retrying"
-    )
-    assert calls == 0

tests/test_idempotency_ledger.py

@@ -1,73 +0,0 @@
-import sqlite3
-from concurrent.futures import ThreadPoolExecutor
-from threading import Barrier
-
-from src.idempotency import IdempotencyLedger
-
-
-def test_completed_result_replays_after_ledger_reconstruction(tmp_path):
-    database = tmp_path / "idempotency.sqlite3"
-    first = IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
-
-    reservation = first.reserve("create-201", ("issue-create", "stackchain/api", "Ship it"))
-    assert reservation.state == "reserved"
-    first.complete("create-201", {"number": 42, "title": "Ship it"})
-
-    reconstructed = IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
-    replay = reconstructed.reserve(
-        "create-201", ("issue-create", "stackchain/api", "Ship it")
-    )
-
-    assert replay.state == "completed"
-    assert replay.response == {"number": 42, "title": "Ship it"}
-    assert sqlite3.connect(database).execute(
-        "SELECT status FROM idempotency_operations WHERE key = ?", ("create-201",)
-    ).fetchone() == ("completed",)
-
-
-def test_reservation_expires_completed_rows_and_enforces_capacity(tmp_path):
-    now = [1_000.0]
-    ledger = IdempotencyLedger(
-        tmp_path / "bounded.sqlite3",
-        ttl_seconds=10,
-        max_entries=2,
-        clock=lambda: now[0],
-    )
-    assert ledger.reserve("old", ("comment", "old")).state == "reserved"
-    ledger.complete("old", {"id": 1})
-    now[0] += 11
-
-    assert ledger.reserve("new-1", ("comment", "new-1")).state == "reserved"
-    ledger.complete("new-1", {"id": 2})
-    assert ledger.reserve("new-2", ("comment", "new-2")).state == "reserved"
-    ledger.complete("new-2", {"id": 3})
-    replacement = ledger.reserve("new-3", ("comment", "new-3"))
-
-    assert replacement.state == "reserved"
-    with sqlite3.connect(ledger.path) as connection:
-        keys = connection.execute(
-            "SELECT key FROM idempotency_operations ORDER BY key"
-        ).fetchall()
-        indexes = connection.execute(
-            "PRAGMA index_list(idempotency_operations)"
-        ).fetchall()
-    assert keys == [("new-2",), ("new-3",)]
-    assert any(index[1] == "idempotency_completed_at_idx" for index in indexes)
-
-
-def test_independent_connections_atomically_reserve_one_operation(tmp_path):
-    database = tmp_path / "workers.sqlite3"
-    ledgers = [
-        IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
-        for _ in range(2)
-    ]
-    barrier = Barrier(2)
-
-    def reserve(ledger):
-        barrier.wait()
-        return ledger.reserve("worker-key-201", ("review", "stackchain/api", 7)).state
-
-    with ThreadPoolExecutor(max_workers=2) as executor:
-        states = list(executor.map(reserve, ledgers))
-
-    assert sorted(states) == ["pending", "reserved"]

tests/test_issue_api.py

@@ -5,7 +5,6 @@ import httpx
 import pytest
 
 from src import gitea_proxy, main
-from src.idempotency import IdempotencyLedger
 
 
 @pytest.mark.anyio
@@ -227,11 +226,9 @@ async def test_edit_assigned_issue_reports_revision_conflict_without_mutation(mo
 
 @pytest.fixture(autouse=True)
 def clear_issue_creation_operations():
-    main._authored_action_operations.clear()
+    main._issue_creation_operations.clear()
-    main._idempotency_ledger.clear()
     yield
-    main._authored_action_operations.clear()
+    main._issue_creation_operations.clear()
-    main._idempotency_ledger.clear()
 
 
 @pytest.mark.anyio
@@ -334,61 +331,6 @@ async def test_create_issue_replays_one_upstream_result_for_concurrent_idempoten
     assert calls == 1
 
 
-@pytest.mark.anyio
-async def test_completed_issue_creation_replays_after_ledger_reconstruction(monkeypatch, tmp_path):
-    calls = 0
-
-    async def user():
-        return {"login": "timmy"}
-
-    async def available_repos():
-        return [{"full_name": "stackchain/api"}]
-
-    async def create(repository, title, body, assignee, label_ids):
-        nonlocal calls
-        calls += 1
-        return {
-            "id": 201, "number": 201, "title": title, "state": "open",
-            "repository": repository, "labels": [], "assignees": [assignee],
-            "updated_at": "2026-08-07T15:00:00Z",
-            "url": "https://forge.example/stackchain/api/issues/201",
-        }
-
-    database = tmp_path / "issues.sqlite3"
-    monkeypatch.setattr(
-        main,
-        "_idempotency_ledger",
-        IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
-    )
-    monkeypatch.setattr(main.gitea_proxy, "current_user", user)
-    monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
-    monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
-    transport = httpx.ASGITransport(app=main.app)
-    headers = {"Idempotency-Key": "restart-create-201"}
-
-    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
-        created = await client.post(
-            "/api/v1/repos/stackchain/api/issues",
-            json={"title": "Durable capture"},
-            headers=headers,
-        )
-        main._authored_action_operations.clear()
-        monkeypatch.setattr(
-            main,
-            "_idempotency_ledger",
-            IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
-        )
-        replayed = await client.post(
-            "/api/v1/repos/stackchain/api/issues",
-            json={"title": "Durable capture"},
-            headers=headers,
-        )
-
-    assert [created.status_code, replayed.status_code] == [201, 201]
-    assert replayed.json()["number"] == 201
-    assert calls == 1
-
-
 @pytest.mark.anyio
 async def test_create_issue_rejects_changed_payload_for_an_existing_idempotency_key(monkeypatch):
     calls = []
@@ -470,9 +412,7 @@ async def test_create_issue_retry_recovers_result_after_the_first_request_times_
 
 
 @pytest.mark.anyio
-async def test_create_issue_durable_ledger_evicts_completed_entry_at_size_limit(
+async def test_create_issue_idempotency_registry_evicts_oldest_entry_at_size_limit(monkeypatch):
-    monkeypatch, tmp_path
-):
     async def user():
         return {"login": "timmy"}
 
@@ -487,10 +427,7 @@ async def test_create_issue_durable_ledger_evicts_completed_entry_at_size_limit(
             "url": "https://forge.example/stackchain/api/issues/17",
         }
 
-    ledger = IdempotencyLedger(
+    monkeypatch.setattr(main, "ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES", 2)
-        tmp_path / "bounded-issues.sqlite3", ttl_seconds=600, max_entries=2
-    )
-    monkeypatch.setattr(main, "_idempotency_ledger", ledger)
     monkeypatch.setattr(main.gitea_proxy, "current_user", user)
     monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
     monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
@@ -504,11 +441,8 @@ async def test_create_issue_durable_ledger_evicts_completed_entry_at_size_limit(
             )
             assert response.status_code == 201
 
-    replay = ledger.reserve(
+    assert len(main._issue_creation_operations) == 2
-        "capture-177-bounded-0",
+    assert "capture-177-bounded-0" not in main._issue_creation_operations
-        ("issue-create", "stackchain/api", "Capture 0", "", ()),
-    )
-    assert replay.state == "reserved"
 
 
 @pytest.mark.anyio