Improve SQLite concurrency and document EMILIA collaboration

README.md

@@ -32,7 +32,7 @@ DNS-based discovery, mandatory Ed25519 signing, server-mediated delivery.
     |                    |                     |                    |
 ```
 
-[Quick Start](#quick-start) · [Python SDK](#python-sdk) · [CLI Reference](#cli-reference) · [Documentation](#documentation)
+[Quick Start](#quick-start) · [Python SDK](#python-sdk) · [CLI Reference](#cli-reference) · [Documentation](#documentation) · [Collaboration](#collaboration)
 
 </div>
 
@@ -247,13 +247,17 @@ ATP is defined as an IETF Internet-Draft (Standards Track):
 >
 > [IETF Datatracker](https://datatracker.ietf.org/doc/draft-li-atp/) · [Full Text](../Agent%20Transfer%20Protocol%20(ATP).md)
 
+## Collaboration
+
+- **[Iman Schrock (@FutureEnterprises)](https://github.com/FutureEnterprises), EMILIA Protocol** — collaborated on the [EP receipt over ATP composition demo](https://github.com/emiliaprotocol/emilia-protocol/tree/main/examples/ep-over-atp) for the IETF 126 Hackathon. The demo carries an EMILIA human-authorization receipt as an opaque ATP payload and verifies the two layers independently: ATP authenticates the sending agent and domain, while the EMILIA receipt proves authorization of the exact action.
+
 ## Contributing
 
 ```bash
 git clone https://github.com/NKU-AOSP-Lab/AgentTransferProtocol.git
 cd atp
 pip install -e ".[dev]"
-python -m pytest tests/ -v    # 227 tests
+python -m pytest tests/ -v    # run the full test suite
 ```
 
 See [Architecture](docs/architecture.md) for module design and development guide.

src/atp/security/replay.py

@@ -42,7 +42,9 @@ def __init__(
 
     def _init_db(self) -> None:
         """Create the nonces table if using persistent storage."""
-        self._conn = sqlite3.connect(str(self._db_path))
+        self._conn = sqlite3.connect(str(self._db_path), timeout=30)
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA busy_timeout=30000")
         self._conn.execute(
             """CREATE TABLE IF NOT EXISTS nonces (
                 nonce TEXT PRIMARY KEY,
@@ -112,6 +114,7 @@ def check(self, nonce: str, timestamp: int, sender: str = "") -> bool:
                     self._insert_count = 0
                 self._conn.commit()
             except sqlite3.Error:
+                self._conn.rollback()  # Release any held write lock
                 pass  # Cache is authoritative; DB failure is non-fatal
 
         # 5. Fresh message
@@ -124,7 +127,9 @@ def _prune_db(self) -> None:
         cutoff = int(time.time()) - self._max_age
         try:
             self._conn.execute("DELETE FROM nonces WHERE timestamp < ?", (cutoff,))
+            self._conn.commit()
         except sqlite3.Error:
+            self._conn.rollback()  # Release any held write lock
             pass
 
     def clear(self) -> None:

src/atp/storage/agents.py

@@ -25,7 +25,9 @@ def __init__(self, db_path: Path):
 
     def _get_conn(self) -> sqlite3.Connection:
         if self._conn is None:
-            self._conn = sqlite3.connect(str(self._db_path))
+            self._conn = sqlite3.connect(str(self._db_path), timeout=30)
+            self._conn.execute("PRAGMA journal_mode=WAL")
+            self._conn.execute("PRAGMA busy_timeout=30000")
         return self._conn
 
     def init_db(self) -> None:
@@ -60,6 +62,7 @@ def register(self, agent_id: str, password: str) -> AgentRecord:
             )
             conn.commit()
         except sqlite3.IntegrityError:
+            conn.rollback()
             raise StorageError(
                 code=ATPErrorCode.SERVER_ERROR,
                 message=f"Agent {agent_id} already registered"

src/atp/storage/messages.py

@@ -37,8 +37,10 @@ class StoredMessage:
 class MessageStore:
     def __init__(self, db_path: Path):
         self._db_path = db_path
-        self._conn = sqlite3.connect(str(db_path))
+        self._conn = sqlite3.connect(str(db_path), timeout=30)
         self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA busy_timeout=30000")
         self.init_db()
 
     def init_db(self) -> None:
@@ -99,6 +101,7 @@ def enqueue(
             self._conn.commit()
             return cursor.lastrowid  # type: ignore[return-value]
         except sqlite3.IntegrityError as exc:
+            self._conn.rollback()
             raise StorageError(
                 ATPErrorCode.SERVER_ERROR,
                 f"Duplicate nonce: {message.nonce}",

tests/test_agents.py

@@ -47,6 +47,17 @@ def test_duplicate_register_raises_storage_error(self, store):
         with pytest.raises(StorageError):
             store.register("alice@example.com", "different")
 
+        # A rejected registration must not leave a write lock behind.
+        second_store = AgentStore(store._db_path)
+        second_store.init_db()
+        record = second_store.register("bob@example.com", "secret")
+        assert record.agent_id == "bob@example.com"
+
+    def test_sqlite_connection_uses_wal_and_busy_timeout(self, store):
+        conn = store._get_conn()
+        assert conn.execute("PRAGMA journal_mode").fetchone()[0] == "wal"
+        assert conn.execute("PRAGMA busy_timeout").fetchone()[0] == 30_000
+
     def test_change_password(self, store):
         store.register("alice@example.com", "oldpass")
         assert store.verify("alice@example.com", "oldpass") is True

tests/test_messages.py

@@ -51,7 +51,8 @@ def test_enqueue_and_get_by_nonce(self, tmp_path: Path) -> None:
 
     def test_duplicate_nonce_raises(self, tmp_path: Path) -> None:
         """enqueue() should raise StorageError on duplicate nonce."""
-        store = MessageStore(db_path=tmp_path / "test.db")
+        db_path = tmp_path / "test.db"
+        store = MessageStore(db_path=db_path)
         msg1 = _make_message(nonce="dup-nonce")
         msg2 = _make_message(nonce="dup-nonce")
 
@@ -60,6 +61,16 @@ def test_duplicate_nonce_raises(self, tmp_path: Path) -> None:
         with pytest.raises(StorageError):
             store.enqueue(msg2)
 
+        # A rejected enqueue must not leave a write lock behind.
+        second_store = MessageStore(db_path=db_path)
+        row_id = second_store.enqueue(_make_message(nonce="fresh-nonce"))
+        assert row_id > 0
+
+    def test_sqlite_connection_uses_wal_and_busy_timeout(self, tmp_path: Path) -> None:
+        store = MessageStore(db_path=tmp_path / "test.db")
+        assert store._conn.execute("PRAGMA journal_mode").fetchone()[0] == "wal"
+        assert store._conn.execute("PRAGMA busy_timeout").fetchone()[0] == 30_000
+
     def test_update_status(self, tmp_path: Path) -> None:
         """update_status() should change the message status."""
         store = MessageStore(db_path=tmp_path / "test.db")

tests/test_replay.py

@@ -1,5 +1,6 @@
-import time
+import sqlite3
 import threading
+import time
 import uuid
 
 import pytest
@@ -73,6 +74,49 @@ def test_clear_empties_cache(self):
         # After clearing, same nonce should be accepted again
         assert guard.check(nonce, timestamp) is True
 
+    def test_sqlite_connection_uses_wal_and_busy_timeout(self, tmp_path):
+        guard = ReplayGuard(db_path=tmp_path / "nonces.db")
+        assert guard._conn is not None
+        assert guard._conn.execute("PRAGMA journal_mode").fetchone()[0] == "wal"
+        assert guard._conn.execute("PRAGMA busy_timeout").fetchone()[0] == 30_000
+
+    def test_persistence_error_releases_write_lock(self, tmp_path):
+        db_path = tmp_path / "nonces.db"
+        guard = ReplayGuard(db_path=db_path)
+        assert guard._conn is not None
+        guard._conn.execute(
+            """CREATE TRIGGER reject_nonce BEFORE INSERT ON nonces
+            BEGIN SELECT RAISE(ABORT, 'injected failure'); END"""
+        )
+        guard._conn.commit()
+
+        assert guard.check("rejected", int(time.time())) is True
+
+        observer = sqlite3.connect(db_path, timeout=0.1)
+        observer.execute("DROP TRIGGER reject_nonce")
+        observer.commit()
+        observer.execute("INSERT INTO nonces VALUES (?, ?)", ("observer", int(time.time())))
+        observer.commit()
+        observer.close()
+
+    def test_prune_db_commits_deletions(self, tmp_path):
+        db_path = tmp_path / "nonces.db"
+        guard = ReplayGuard(max_age_seconds=10, db_path=db_path)
+        assert guard._conn is not None
+        guard._conn.execute(
+            "INSERT INTO nonces VALUES (?, ?)", ("expired", int(time.time()) - 100)
+        )
+        guard._conn.commit()
+
+        guard._prune_db()
+
+        observer = sqlite3.connect(db_path)
+        count = observer.execute(
+            "SELECT COUNT(*) FROM nonces WHERE nonce = 'expired'"
+        ).fetchone()[0]
+        observer.close()
+        assert count == 0
+
     def test_thread_safety(self):
         guard = ReplayGuard()
         now = int(time.time())