From 7bf6e201aa745431ed454c27a34543d057ea7e86 Mon Sep 17 00:00:00 2001 From: neason Date: Sat, 1 Aug 2026 23:27:49 +0800 Subject: [PATCH] perf(audit): use WAL journal to avoid fsync per commit on event loop AuditStore.append runs synchronously from TurnEngine._audit at every tool stage (17 call sites inside the async engine loop). With SQLite's default DELETE journal + synchronous=FULL, each append's commit() fsyncs, blocking the event loop for ~1-10ms per tool call. Set journal_mode=WAL and synchronous=NORMAL at connect time so a commit returns without an fsync while preserving transaction atomicity; WAL checkpoints flush to disk periodically. Add tests/test_audit.py covering the pragma, round-trip/filters, secret/body redaction, browser_type input redaction, resource extraction, close idempotency, and truncation. --- coworker/audit.py | 7 +++ tests/test_audit.py | 134 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 tests/test_audit.py diff --git a/coworker/audit.py b/coworker/audit.py index 349d20b10..8e7cb25b2 100644 --- a/coworker/audit.py +++ b/coworker/audit.py @@ -29,6 +29,13 @@ def __init__(self, db_path: str | Path) -> None: self._lock = threading.RLock() self._conn = sqlite3.connect(self.db_path, check_same_thread=False) self._conn.row_factory = sqlite3.Row + # WAL + synchronous=NORMAL: a commit no longer fsyncs on every write. The + # audit sink is called synchronously from the engine's async loop (one row + # per tool stage), so the default DELETE journal + FULL synchronous was + # blocking the event loop by ~1-10ms per tool call. WAL keeps transaction + # durability (rollback/atomicity) and only checkpoints to disk periodically. + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=NORMAL") self._conn.execute(""" CREATE TABLE IF NOT EXISTS audit_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 000000000..0b0bce885 --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,134 @@ +"""Durable audit log: storage round-trip, redaction, resource extraction, and +WAL journal mode so per-commit fsync doesn't stall the event loop.""" + +from coworker.audit import AuditStore, _resource, _sanitize_args + + +def _pragma(conn, name): + row = conn.execute(f"PRAGMA {name}").fetchone() + return row[0] if row else None + + +def test_audit_store_uses_wal_journal(tmp_path): + """WAL + synchronous=NORMAL: commit returns without an fsync per tool audit, + which otherwise blocks the asyncio event loop at every _audit() call site.""" + store = AuditStore(tmp_path / "audit.db") + try: + assert _pragma(store._conn, "journal_mode").lower() == "wal" + assert _pragma(store._conn, "synchronous") == 1 # NORMAL + finally: + store.close() + + +def test_append_and_list_roundtrip(tmp_path): + store = AuditStore(tmp_path / "audit.db") + try: + store.append( + { + "session_id": "s1", + "agent": "code", + "workspace": "/tmp/ws", + "tool": "shell", + "stage": "finished", + "status": "ok", + "arguments": {"command": "ls"}, + "result_preview": "file.txt", + } + ) + rows = store.list() + assert len(rows) == 1 + row = rows[0] + assert row["session_id"] == "s1" + assert row["tool"] == "shell" + assert row["stage"] == "finished" + assert row["status"] == "ok" + assert row["args"] == {"command": "ls"} + finally: + store.close() + + +def test_list_filters(tmp_path): + store = AuditStore(tmp_path / "audit.db") + try: + store.append({"session_id": "s1", "tool": "shell", "arguments": {}}) + store.append({"session_id": "s2", "tool": "read_file", "arguments": {}}) + assert len(store.list(session_id="s1")) == 1 + assert len(store.list(tool="read_file")) == 1 + assert len(store.list(session_id="s1", tool="read_file")) == 0 + finally: + store.close() + + +def test_sanitize_args_redacts_secrets(): + args = { + "token": "t", + "api_key": "k", + "password": "p", + "access_token": "a", + "bot_token": "b", + "app_token": "ap", + "secret": "s", + "raw": "r", + "body": "hi", + "content": "x", + "html": "

", + "safe": "visible", + "nested_token": "also-redacted", + } + out = _sanitize_args("http_get", args) + for key in ( + "token", + "api_key", + "password", + "access_token", + "bot_token", + "app_token", + "secret", + "raw", + "body", + "content", + "html", + "nested_token", + ): + assert out[key] == "[redacted]" or out[key] == "[redacted body]", key + assert out["safe"] == "visible" + + +def test_sanitize_args_redacts_browser_type_text(): + out = _sanitize_args("browser_type", {"text": "my password is hunter2"}) + assert out["text"] == "[redacted input]" + + +def test_resource_extracts_well_known_keys(): + assert ( + _resource("http_get", {"url": "https://example.com"}, {}) + == "https://example.com" + ) + assert _resource("github_issue", {"owner": "o", "repo": "r"}, {}) == "o" + assert ( + _resource("zendesk_search", {"subdomain": "acme", "query": "foo"}, {}) + == "acme.zendesk.com" + ) + assert _resource("read_file", {"path": "/etc/hosts"}, {}) == "" + assert ( + _resource("http_get", {}, {"url": "https://from-result"}) + == "https://from-result" + ) + + +def test_close_is_idempotent(tmp_path): + store = AuditStore(tmp_path / "audit.db") + store.close() + store.close() # must not raise + + +def test_summarize_truncates_long_strings(tmp_path): + store = AuditStore(tmp_path / "audit.db") + try: + long = "x" * 5000 + store.append({"tool": "echo", "arguments": {"message": long}}) + row = store.list(tool="echo")[0] + assert len(row["args"]["message"]) <= 500 + assert row["args"]["message"].endswith("...") + finally: + store.close()