README.md

@@ -193,11 +193,14 @@ filesystem boundary independently of the service umask: the database directory i
 owner-only `0700`, database and SQLite sidecar files are owner-only `0600`, and symlinked database
 paths are rejected before access. Worker-shared live and Find Work snapshots add AES-256-GCM
 envelopes authenticated to their store identity (and live generation), so copied databases do not
-expose issue bodies, titles, notification metadata, or repository context. Existing plaintext
-snapshot rows migrate on their first read without changing freshness, revisions, ordering, or claim
-filters. Synchronized unfiled Draft collections use a separate AES-256-GCM key and authenticate the
-account and revision; existing plaintext rows likewise migrate on first read. Other private stores
-are not encrypted at the application layer. Web Push subscriptions use a third, independent
+expose issue bodies, titles, notification metadata, or repository context. The authored-action
+idempotency ledger encrypts both request fingerprints and confirmed upstream responses, authenticating
+each envelope to its operation key and field purpose so rows and fields cannot be substituted.
+Existing plaintext snapshot and ledger rows migrate atomically on their first read without changing
+freshness, revisions, ordering, replay, or conflict semantics. Synchronized unfiled Draft collections
+use a separate AES-256-GCM key and authenticate the account and revision; existing plaintext rows
+likewise migrate on first read. Other private stores are not encrypted at the application layer. Web
+Push subscriptions use a third, independent
 AES-256-GCM key authenticated to the device session, with keyed endpoint indexes preserving
 single-device enrollment without retaining capability URLs. Existing plaintext subscriptions
 migrate atomically at startup without resetting delivery checkpoints or reminder schedules. Secure

src/idempotency.py

@@ -6,6 +6,7 @@ from pathlib import Path
 from typing import Any, Callable
 
 from src.private_state import connect_private_sqlite
+from src.state_encryption import PrivateStateCipher, private_state_encryption_key
 
 
 @dataclass(frozen=True)
@@ -29,12 +30,17 @@ class IdempotencyLedger:
         max_entries: int,
         lock_timeout_seconds: float = 0.1,
         clock: Callable[[], float] = time.time,
+        encryption_key: bytes | None = None,
     ) -> None:
         self.path = Path(path)
         self.ttl_seconds = ttl_seconds
         self.max_entries = max_entries
         self.lock_timeout_seconds = lock_timeout_seconds
         self.clock = clock
+        self._cipher = PrivateStateCipher(
+            encryption_key if encryption_key is not None else private_state_encryption_key(),
+            store="idempotency-ledger",
+        )
         self._initialize()
 
     def _connect(self) -> sqlite3.Connection:
@@ -67,6 +73,17 @@ class IdempotencyLedger:
     def _fingerprint(value: tuple[Any, ...]) -> str:
         return json.dumps(value, separators=(",", ":"), sort_keys=True)
 
+    def _seal(self, value: str, *, key: str, field: str) -> str:
+        return self._cipher.seal(value, binding=f"{key}:{field}")
+
+    def _open(self, payload: str, *, key: str, field: str) -> tuple[str, bool]:
+        value, legacy = self._cipher.open(payload, binding=f"{key}:{field}")
+        if legacy:
+            return json.dumps(value, separators=(",", ":"), sort_keys=True), True
+        if not isinstance(value, str):
+            raise RuntimeError("idempotency ledger payload is invalid")
+        return value, legacy
+
     def reserve(self, key: str, fingerprint: tuple[Any, ...]) -> Reservation:
         encoded = self._fingerprint(fingerprint)
         now = self.clock()
@@ -84,10 +101,40 @@ class IdempotencyLedger:
                     (key,),
                 ).fetchone()
                 if row is not None:
-                    if row[0] != encoded:
+                    stored_fingerprint, legacy_fingerprint = self._open(
+                        row[0], key=key, field="fingerprint"
+                    )
+                    if legacy_fingerprint:
+                        connection.execute(
+                            "UPDATE idempotency_operations SET fingerprint = ? "
+                            "WHERE key = ? AND fingerprint = ?",
+                            (
+                                self._seal(
+                                    stored_fingerprint, key=key, field="fingerprint"
+                                ),
+                                key,
+                                row[0],
+                            ),
+                        )
+                    if stored_fingerprint != encoded:
                         return Reservation("conflict")
                     if row[1] == "completed":
-                        return Reservation("completed", json.loads(row[2]))
+                        stored_response, legacy_response = self._open(
+                            row[2], key=key, field="response"
+                        )
+                        if legacy_response:
+                            connection.execute(
+                                "UPDATE idempotency_operations SET response_json = ? "
+                                "WHERE key = ? AND response_json = ?",
+                                (
+                                    self._seal(
+                                        stored_response, key=key, field="response"
+                                    ),
+                                    key,
+                                    row[2],
+                                ),
+                            )
+                        return Reservation("completed", json.loads(stored_response))
                     if row[3] <= now - self.ttl_seconds:
                         return Reservation("uncertain")
                     return Reservation("pending")
@@ -110,7 +157,7 @@ class IdempotencyLedger:
                 connection.execute(
                     "INSERT INTO idempotency_operations "
                     "(key, fingerprint, status, created_at) VALUES (?, ?, 'pending', ?)",
-                    (key, encoded, now),
+                    (key, self._seal(encoded, key=key, field="fingerprint"), now),
                 )
         except sqlite3.OperationalError as exc:
             if "locked" in str(exc).lower() or "busy" in str(exc).lower():
@@ -129,7 +176,7 @@ class IdempotencyLedger:
                 connection.execute(
                     "UPDATE idempotency_operations SET status = 'completed', "
                     "response_json = ?, completed_at = ? WHERE key = ?",
-                    (encoded, self.clock(), key),
+                    (self._seal(encoded, key=key, field="response"), self.clock(), key),
                 )
         except sqlite3.OperationalError as exc:
             if "locked" in str(exc).lower() or "busy" in str(exc).lower():

tests/test_idempotency_ledger.py

@@ -2,7 +2,132 @@ import sqlite3
 from concurrent.futures import ThreadPoolExecutor
 from threading import Barrier
 
+import pytest
+
 from src.idempotency import IdempotencyLedger
+from src.state_encryption import PrivateStateEncryptionError
+
+
+PRIVATE_KEY = b"i" * 32
+
+
+def test_authored_payloads_are_encrypted_at_rest_and_replay_after_restart(tmp_path):
+    database = tmp_path / "idempotency.sqlite3"
+    fingerprint_canary = "private-issue-body-canary"
+    response_canary = "private-gitea-response-canary"
+    ledger = IdempotencyLedger(
+        database,
+        ttl_seconds=600,
+        max_entries=256,
+        encryption_key=PRIVATE_KEY,
+    )
+
+    assert ledger.reserve("create-201", ("issue-create", fingerprint_canary)).state == "reserved"
+    ledger.complete("create-201", {"title": response_canary})
+
+    persisted = database.read_bytes()
+    assert fingerprint_canary.encode() not in persisted
+    assert response_canary.encode() not in persisted
+    with sqlite3.connect(database) as connection:
+        fingerprint, response = connection.execute(
+            "SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
+            ("create-201",),
+        ).fetchone()
+    assert fingerprint.startswith("v1:")
+    assert response.startswith("v1:")
+
+    reopened = IdempotencyLedger(
+        database,
+        ttl_seconds=600,
+        max_entries=256,
+        encryption_key=PRIVATE_KEY,
+    )
+    replay = reopened.reserve("create-201", ("issue-create", fingerprint_canary))
+    assert replay.state == "completed"
+    assert replay.response == {"title": response_canary}
+
+
+def test_legacy_plaintext_completed_row_is_migrated_without_losing_replay(tmp_path):
+    database = tmp_path / "legacy.sqlite3"
+    ledger = IdempotencyLedger(
+        database,
+        ttl_seconds=600,
+        max_entries=256,
+        encryption_key=PRIVATE_KEY,
+        clock=lambda: 100.0,
+    )
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            "INSERT INTO idempotency_operations "
+            "(key, fingerprint, status, response_json, created_at, completed_at) "
+            "VALUES (?, ?, 'completed', ?, ?, ?)",
+            (
+                "legacy-comment",
+                '["issue-comment","stackchain/api",7,"legacy secret"]',
+                '{"id":42,"body":"legacy response"}',
+                90.0,
+                95.0,
+            ),
+        )
+
+    replay = ledger.reserve(
+        "legacy-comment",
+        ("issue-comment", "stackchain/api", 7, "legacy secret"),
+    )
+
+    assert replay.state == "completed"
+    assert replay.response == {"id": 42, "body": "legacy response"}
+    with sqlite3.connect(database) as connection:
+        fingerprint, response = connection.execute(
+            "SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
+            ("legacy-comment",),
+        ).fetchone()
+    assert fingerprint.startswith("v1:")
+    assert response.startswith("v1:")
+
+
+def test_wrong_key_fails_closed_before_a_completed_result_can_replay(tmp_path):
+    database = tmp_path / "wrong-key.sqlite3"
+    ledger = IdempotencyLedger(
+        database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
+    )
+    fingerprint = ("issue-comment", "stackchain/api", 7, "secret")
+    assert ledger.reserve("comment-7", fingerprint).state == "reserved"
+    ledger.complete("comment-7", {"id": 42})
+
+    wrong_key = IdempotencyLedger(
+        database, ttl_seconds=600, max_entries=256, encryption_key=b"x" * 32
+    )
+    with pytest.raises(
+        PrivateStateEncryptionError, match="private state could not be decrypted"
+    ):
+        wrong_key.reserve("comment-7", fingerprint)
+
+
+def test_completed_ciphertext_cannot_be_substituted_between_operation_keys(tmp_path):
+    database = tmp_path / "substitution.sqlite3"
+    ledger = IdempotencyLedger(
+        database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
+    )
+    fingerprint = ("issue-create", "stackchain/api", "same authored request")
+    for key, number in (("create-a", 41), ("create-b", 42)):
+        assert ledger.reserve(key, fingerprint).state == "reserved"
+        ledger.complete(key, {"number": number})
+    with sqlite3.connect(database) as connection:
+        responses = dict(
+            connection.execute(
+                "SELECT key, response_json FROM idempotency_operations"
+            ).fetchall()
+        )
+        connection.execute(
+            "UPDATE idempotency_operations SET response_json = ? WHERE key = ?",
+            (responses["create-b"], "create-a"),
+        )
+
+    with pytest.raises(
+        PrivateStateEncryptionError, match="private state could not be decrypted"
+    ):
+        ledger.reserve("create-a", fingerprint)
 
 
 def test_completed_result_replays_after_ledger_reconstruction(tmp_path):