audit: add query filters + file locking for chain integrity\n\naudit.py:\n - recent() gains agent, tool, since, until filters. All composable.\n - _write_entry() now acquires an exclusive file lock (fcntl.LOCK_EX)\n and re-reads the last chain hash under lock before writing. Fixes\n chain breaks caused by concurrent writers (e.g. dashboard + test\n suite both appending). Graceful fallback on Windows (no fcntl).\n\ncli.py:\n - 'heddle audit show' gains --agent, --tool, --since, --until options\n in addition to existing -n/--count and --event.\n - Example: heddle audit show --agent prometheus-bridge --since 2026-04\n heddle audit show -n 10 --event trust_violation --tool dangerous\n\ntests: 137 -> 142 (+5)\n - test_audit_filter_by_agent\n - test_audit_filter_by_tool\n - test_audit_filter_by_time_range\n - test_audit_filter_combined\n - test_audit_filter_no_matches\n\nNote: the existing audit log at ~/.heddle/audit/audit.jsonl has a chain\nbreak at line 2046 from before this fix (two processes writing without\nlocking on 2026-03-29). The break is historical — new entries written\nafter this commit will maintain chain integrity even under concurrent\nwrites. heddle audit verify correctly reports the break.\n\nv0.2 Pillar 2 (audit query CLI).

Author: goweft

src/heddle/cli.py

@@ -218,11 +218,17 @@ def audit():
 @audit.command("show")
 @click.option("-n", "--count", default=20, help="Number of entries")
 @click.option("--event", default=None, help="Filter by event type")
-def audit_show(count: int, event: str | None):
+@click.option("--agent", default=None, help="Filter by agent name")
+@click.option("--tool", default=None, help="Filter by tool name")
+@click.option("--since", default=None, help="Only entries after this ISO timestamp")
+@click.option("--until", "until_", default=None, help="Only entries before this ISO timestamp")
+def audit_show(count: int, event: str | None, agent: str | None,
+               tool: str | None, since: str | None, until_: str | None):
     """Show recent audit log entries."""
     from heddle.security.audit import get_audit_logger
     logger = get_audit_logger()
-    entries = logger.recent(count, event_type=event)
+    entries = logger.recent(count, event_type=event, agent=agent,
+                            tool=tool, since=since, until=until_)
     if not entries:
         console.print("[dim]No audit entries.[/]")
         return

src/heddle/security/audit.py

@@ -18,6 +18,12 @@
 from pathlib import Path
 from typing import Any
 
+try:
+    import fcntl
+    _HAS_FCNTL = True
+except ImportError:
+    _HAS_FCNTL = False  # Windows
+
 logger = logging.getLogger(__name__)
 
 DEFAULT_LOG_DIR = Path.home() / ".heddle" / "audit"
@@ -58,15 +64,30 @@ def _compute_last_hash(self) -> str:
         return "GENESIS"
 
     def _write_entry(self, entry: dict[str, Any]) -> None:
-        """Write a single audit entry, updating the chain hash."""
-        entry["chain_hash"] = self._prev_hash
-        entry["timestamp"] = datetime.now(timezone.utc).isoformat()
-
-        line = json.dumps(entry, default=str, separators=(",", ":"))
-        self._prev_hash = hashlib.sha256(line.encode()).hexdigest()
+        """Write a single audit entry, updating the chain hash.
 
+        Uses file locking to prevent chain breaks when multiple
+        processes (e.g. dashboard + test suite) write concurrently.
+        """
         with open(self._log_file, "a") as f:
-            f.write(line + "\n")
+            if _HAS_FCNTL:
+                fcntl.flock(f, fcntl.LOCK_EX)
+            try:
+                # Re-read last hash under lock in case another process
+                # appended since we last computed it.
+                self._prev_hash = self._compute_last_hash()
+
+                entry["chain_hash"] = self._prev_hash
+                entry["timestamp"] = datetime.now(timezone.utc).isoformat()
+
+                line = json.dumps(entry, default=str, separators=(",", ":"))
+                self._prev_hash = hashlib.sha256(line.encode()).hexdigest()
+
+                f.write(line + "\n")
+                f.flush()
+            finally:
+                if _HAS_FCNTL:
+                    fcntl.flock(f, fcntl.LOCK_UN)
 
     # ── Public logging methods ───────────────────────────────────────
 
@@ -191,8 +212,25 @@ def verify_chain(self) -> tuple[bool, int, str]:
 
         return True, count, f"Chain valid: {count} entries"
 
-    def recent(self, n: int = 20, event_type: str | None = None) -> list[dict]:
-        """Read the most recent N entries, optionally filtered by event type."""
+    def recent(
+        self,
+        n: int = 20,
+        event_type: str | None = None,
+        agent: str | None = None,
+        tool: str | None = None,
+        since: str | None = None,
+        until: str | None = None,
+    ) -> list[dict]:
+        """Read the most recent N entries with optional filters.
+
+        Args:
+            n: Maximum entries to return.
+            event_type: Filter by event type (tool_call, http_bridge, etc.).
+            agent: Filter by agent (config) name.
+            tool: Filter by tool name.
+            since: ISO timestamp — only entries at or after this time.
+            until: ISO timestamp — only entries at or before this time.
+        """
         if not self._log_file.exists():
             return []
 
@@ -204,11 +242,23 @@ def recent(self, n: int = 20, event_type: str | None = None) -> list[dict]:
                     continue
                 try:
                     entry = json.loads(line)
-                    if event_type is None or entry.get("event") == event_type:
-                        entries.append(entry)
                 except json.JSONDecodeError:
                     continue
 
+                if event_type is not None and entry.get("event") != event_type:
+                    continue
+                if agent is not None and entry.get("agent") != agent:
+                    continue
+                if tool is not None and entry.get("tool") != tool:
+                    continue
+                ts = entry.get("timestamp", "")
+                if since is not None and ts < since:
+                    continue
+                if until is not None and ts > until:
+                    continue
+
+                entries.append(entry)
+
         return entries[-n:]
 
 

tests/test_security.py

@@ -94,6 +94,70 @@ def test_audit_filter_by_event(audit):
     assert len(http_entries) == 1
 
 
+
+
+def test_audit_filter_by_agent(audit):
+    audit.log_tool_call("agent-a", "t1", {}, "success")
+    audit.log_tool_call("agent-b", "t2", {}, "success")
+    audit.log_tool_call("agent-a", "t3", {}, "success")
+
+    a_entries = audit.recent(10, agent="agent-a")
+    assert len(a_entries) == 2
+    assert all(e["agent"] == "agent-a" for e in a_entries)
+
+    b_entries = audit.recent(10, agent="agent-b")
+    assert len(b_entries) == 1
+
+
+def test_audit_filter_by_tool(audit):
+    audit.log_tool_call("a", "query_prometheus", {}, "success")
+    audit.log_tool_call("a", "get_alerts", {}, "success")
+    audit.log_tool_call("a", "query_prometheus", {}, "success")
+
+    entries = audit.recent(10, tool="query_prometheus")
+    assert len(entries) == 2
+    assert all(e["tool"] == "query_prometheus" for e in entries)
+
+
+def test_audit_filter_by_time_range(audit):
+    audit.log_tool_call("a", "t1", {}, "success")
+    audit.log_tool_call("a", "t2", {}, "success")
+    audit.log_tool_call("a", "t3", {}, "success")
+
+    entries = audit.recent(10)
+    assert len(entries) == 3
+
+    # Use the timestamp of the second entry as a boundary
+    mid_ts = entries[1]["timestamp"]
+    since_entries = audit.recent(10, since=mid_ts)
+    assert len(since_entries) >= 2  # second and third entry
+
+    until_entries = audit.recent(10, until=mid_ts)
+    assert len(until_entries) >= 1  # first entry at minimum
+
+
+def test_audit_filter_combined(audit):
+    audit.log_tool_call("agent-a", "t1", {}, "success")
+    audit.log_http_bridge("agent-a", "t1", "GET", "http://x", status_code=200)
+    audit.log_tool_call("agent-b", "t2", {}, "success")
+    audit.log_tool_call("agent-a", "t2", {}, "error", error="timeout")
+
+    # agent + event type
+    entries = audit.recent(10, event_type="tool_call", agent="agent-a")
+    assert len(entries) == 2
+    assert all(e["agent"] == "agent-a" and e["event"] == "tool_call" for e in entries)
+
+    # agent + tool
+    entries = audit.recent(10, agent="agent-a", tool="t1")
+    assert len(entries) == 2  # tool_call + http_bridge both have tool=t1
+
+
+def test_audit_filter_no_matches(audit):
+    audit.log_tool_call("a", "t1", {}, "success")
+    entries = audit.recent(10, agent="nonexistent")
+    assert entries == []
+
+
 # ── Trust Enforcer ───────────────────────────────────────────────────
 
 @pytest.fixture(autouse=True)