gabriel / muse public
audit.py python
121 lines 3.8 KB
Raw
sha256:502e812e12319e30302d07d165425ceeb21687ad30e8af3a439b1f6de36c64d6 fix: update stale tag tests to use label for free-form anno… Sonnet 4.6 19 days ago
1 """Harmony audit log — append and list the tamper-evident action log.
2
3 Single responsibility: write and read the append-only audit log under
4 .muse/harmony/audit/. Entries are never modified after writing.
5 """
6
7 from __future__ import annotations
8
9 import json
10 import logging
11 import pathlib
12
13 from muse.core.types import content_hash, load_json_file
14
15 from .fingerprint import _now_utc
16 from .paths import audit_dir
17 from .patterns import _write_atomic
18 from .types import (
19 AgentProvenance,
20 AuditEvent,
21 _AuditMetadata,
22 )
23
24 logger = logging.getLogger(__name__)
25
26 #: Maximum bytes read from a single audit log entry.
27 _MAX_AUDIT_BYTES: int = 4_096 # 4 KiB
28
29 #: Maximum audit entries returned by :func:`list_audit`.
30 _MAX_AUDIT_ENTRIES: int = 10_000
31
32 # ---------------------------------------------------------------------------
33 # Audit log
34 # ---------------------------------------------------------------------------
35
36 def append_audit(
37 root: pathlib.Path,
38 event_type: str,
39 acted_by: AgentProvenance,
40 *,
41 pattern_id: str | None = None,
42 resolution_id: str | None = None,
43 policy_id: str | None = None,
44 metadata: _AuditMetadata | None = None,
45 ) -> None:
46 """Append a single entry to the harmony audit log.
47
48 Each entry is an independent JSON file named
49 ``<YYYYMMDD>-<sha256-hex-slice>.json`` in ``.muse/harmony/audit/``. The
50 append-only design means audit entries are never modified after writing —
51 they form a tamper-evident record of all harmony engine actions.
52
53 Args:
54 root: Repository root.
55 event_type: :class:`AuditEventType` constant describing what happened.
56 acted_by: Attribution for the actor.
57 pattern_id: Involved pattern (if applicable).
58 resolution_id: Involved resolution (if applicable).
59 policy_id: Involved policy (if applicable).
60 metadata: Additional event-specific fields (arbitrary JSON-safe dict).
61 """
62 now = _now_utc()
63 payload: _AuditMetadata = {
64 "event_type": event_type,
65 "pattern_id": pattern_id,
66 "resolution_id": resolution_id,
67 "policy_id": policy_id,
68 "acted_by": acted_by.to_dict(),
69 "occurred_at": now.isoformat(),
70 "metadata": metadata or {},
71 }
72 audit_id = content_hash(payload)
73 filename = f"{now.strftime('%Y%m%d')}-{audit_id[7:19]}.json"
74
75 entry: AuditEvent = {"audit_id": audit_id, **payload} # type: ignore[misc]
76 _write_atomic(audit_dir(root) / filename, json.dumps(entry, indent=2))
77
78 def list_audit(root: pathlib.Path, limit: int = 100) -> list[AuditEvent]:
79 """Return up to *limit* recent audit log entries, newest first.
80
81 Entries are sorted by filename (``<YYYYMMDD>-<sha256-hex-slice>``), which gives
82 chronological order. Skips entries exceeding :data:`_MAX_AUDIT_BYTES`
83 or that fail JSON parsing.
84
85 Args:
86 root: Repository root.
87 limit: Maximum entries to return (default 100).
88 """
89 adir = audit_dir(root)
90 if not adir.exists():
91 return []
92
93 entries: list[tuple[str, AuditEvent]] = []
94 count = 0
95
96 for f in adir.iterdir():
97 if count >= _MAX_AUDIT_ENTRIES:
98 break
99 count += 1
100 if f.is_symlink() or not f.is_file():
101 continue
102 if not f.name.endswith(".json"):
103 continue
104 try:
105 size = f.stat().st_size
106 except OSError:
107 continue
108 if size > _MAX_AUDIT_BYTES:
109 logger.warning(
110 "⚠️ harmony: audit entry %s is too large (%d bytes); skipping",
111 f.name,
112 size,
113 )
114 continue
115 data: AuditEvent | None = load_json_file(f)
116 if data is None:
117 continue
118 entries.append((f.name, data))
119
120 entries.sort(key=lambda t: t[0], reverse=True)
121 return [e for _, e in entries[:limit]]
File History 1 commit
sha256:502e812e12319e30302d07d165425ceeb21687ad30e8af3a439b1f6de36c64d6 fix: update stale tag tests to use label for free-form anno… Sonnet 4.6 19 days ago