aaronrene / gabriel-muse public
op_log.py python
413 lines 14.5 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
1 """Append-only operation log for Muse live collaboration.
2
3 The op log is the bridge between real-time collaborative editing and the
4 immutable commit DAG. During a live session, operations are appended to
5 the log as they occur. At commit time the log is collapsed into a
6 :class:`~muse.domain.StructuredDelta` and stored with the commit record.
7
8 Design principles
9 -----------------
10 - **Append-only** — entries are never modified or deleted; the file grows
11 monotonically. Compaction happens through checkpoints (see below).
12 - **Lamport-clocked** — every entry carries a logical Lamport timestamp
13 that imposes a total order across concurrent actors without wall-clock
14 coordination.
15 - **Causally linked** — ``parent_op_ids`` lets any entry declare the ops it
16 depends on, enabling causal replay and CRDT join operations downstream.
17 - **Domain-neutral** — the log stores :class:`~muse.domain.DomainOp` values
18 unchanged; the core engine has no opinion about what those ops mean.
19 - **Checkpoint / compaction** — when a live session crystallises into a Muse
20 commit, a checkpoint record is written that marks the current snapshot.
21 Subsequent reads return only ops that arrived after the checkpoint.
22
23 Layout::
24
25 .muse/op_log/<session_id>/
26 ops.jsonl — one JSON line per OpEntry (append-only)
27 checkpoint.json — most recent checkpoint (snapshot_id + lamport_ts)
28
29 Relationship to the commit DAG
30 -------------------------------
31 The op log does **not** replace the commit DAG. It is a staging area:
32
33 live edits → OpLog.append() → ops.jsonl
34 session end → OpLog.checkpoint(snapshot_id) → commit record
35 commit record → normal Muse commit DAG
36
37 Replaying the log from a checkpoint reproduces the snapshot deterministically,
38 giving the same guarantee as re-running ``git apply`` from a patch file.
39
40 Usage::
41
42 from muse.core.op_log import OpLog, make_op_entry
43
44 log = OpLog(repo_root, session_id="session-abc")
45 entry = make_op_entry(
46 actor_id="counterpoint-bot",
47 domain="midi",
48 domain_op=my_insert_op,
49 lamport_ts=log.next_lamport_ts(),
50 )
51 log.append(entry)
52
53 delta = log.to_structured_delta("midi") # collapse for commit
54 ckpt = log.checkpoint(snapshot_id) # crystallise
55 """
56
57 from __future__ import annotations
58
59 import datetime
60 import json
61 import logging
62 import pathlib
63 import uuid as _uuid_mod
64 from typing import TypedDict
65
66 from muse.core.store import write_text_atomic
67 from muse.domain import DomainOp, StructuredDelta
68
69
70 type _IntMap = dict[str, int]
71 logger = logging.getLogger(__name__)
72
73 _OP_LOG_DIR = ".muse/op_log"
74
75
76 # ---------------------------------------------------------------------------
77 # Wire-format TypedDicts
78 # ---------------------------------------------------------------------------
79
80
81 class OpEntry(TypedDict):
82 """A single operation in the append-only op log.
83
84 ``op_id``
85 Stable UUID4 for this entry — used by consumers to deduplicate
86 on replay and by CRDT join to establish causal identity.
87 ``actor_id``
88 The agent or human identity that produced this op.
89 ``lamport_ts``
90 Logical Lamport timestamp. Monotonically increasing within a
91 session; used to establish total ordering when wall-clock times
92 are unavailable or unreliable.
93 ``parent_op_ids``
94 Causal parents — op IDs that this entry depends on. Empty list
95 means this entry has no explicit causal dependency (root entry).
96 Used by CRDT merge and causal replay.
97 ``domain``
98 Domain tag matching the :class:`~muse.domain.MuseDomainPlugin`
99 that produced this op (e.g. ``"midi"``, ``"code"``).
100 ``domain_op``
101 The actual typed domain operation. Stored verbatim.
102 ``created_at``
103 ISO 8601 UTC wall-clock timestamp when the entry was appended.
104 Informational only — use ``lamport_ts`` for ordering.
105 ``intent_id``
106 Links this op to a coordination intent (from
107 :mod:`muse.core.coordination`). Empty string if not applicable.
108 ``reservation_id``
109 Links this op to a coordination reservation. Empty string if not
110 applicable.
111 """
112
113 op_id: str
114 actor_id: str
115 lamport_ts: int
116 parent_op_ids: list[str]
117 domain: str
118 domain_op: DomainOp
119 created_at: str
120 intent_id: str
121 reservation_id: str
122
123
124 class OpLogCheckpoint(TypedDict):
125 """A snapshot of the op log state at commit time.
126
127 Written by :meth:`OpLog.checkpoint` when a live session crystallises
128 into a Muse commit. Subsequent :meth:`OpLog.replay_since_checkpoint`
129 calls return only ops that arrived after this checkpoint.
130
131 ``session_id``
132 The session this checkpoint belongs to.
133 ``snapshot_id``
134 The commit snapshot ID that this checkpoint materialises. All ops
135 up to and including ``lamport_ts`` are captured by this snapshot.
136 ``lamport_ts``
137 The Lamport timestamp of the last op included in this checkpoint.
138 ``op_count``
139 Number of op entries in the log at checkpoint time.
140 ``created_at``
141 ISO 8601 UTC timestamp.
142 """
143
144 session_id: str
145 snapshot_id: str
146 lamport_ts: int
147 op_count: int
148 created_at: str
149
150
151 # ---------------------------------------------------------------------------
152 # Factory
153 # ---------------------------------------------------------------------------
154
155
156 def make_op_entry(
157 actor_id: str,
158 domain: str,
159 domain_op: DomainOp,
160 lamport_ts: int,
161 *,
162 parent_op_ids: list[str] | None = None,
163 intent_id: str = "",
164 reservation_id: str = "",
165 ) -> OpEntry:
166 """Create a new :class:`OpEntry` with a fresh UUID op_id.
167
168 Args:
169 actor_id: Agent or human identity string.
170 domain: Domain tag (e.g. ``"midi"``).
171 domain_op: The typed domain operation to log.
172 lamport_ts: Logical Lamport timestamp for this entry.
173 parent_op_ids: Causal dependencies. Defaults to empty list.
174 intent_id: Optional coordination intent linkage.
175 reservation_id: Optional coordination reservation linkage.
176
177 Returns:
178 A fully populated :class:`OpEntry`.
179 """
180 return OpEntry(
181 op_id=str(_uuid_mod.uuid4()),
182 actor_id=actor_id,
183 lamport_ts=lamport_ts,
184 parent_op_ids=list(parent_op_ids or []),
185 domain=domain,
186 domain_op=domain_op,
187 created_at=datetime.datetime.now(datetime.timezone.utc).isoformat(),
188 intent_id=intent_id,
189 reservation_id=reservation_id,
190 )
191
192
193 # ---------------------------------------------------------------------------
194 # OpLog
195 # ---------------------------------------------------------------------------
196
197
198 class OpLog:
199 """Append-only operation log for a single live collaboration session.
200
201 Each session gets its own directory under ``.muse/op_log/<session_id>/``.
202 The log file is JSON-lines: one :class:`OpEntry` per line. The checkpoint
203 file is a single JSON object written atomically when a session is committed.
204
205 Args:
206 repo_root: Repository root (the directory containing ``.muse/``).
207 session_id: Stable identifier for this collaboration session. Use a
208 UUID, a branch name, or any stable string. The session
209 directory is created on first :meth:`append`.
210 """
211
212 def __init__(self, repo_root: pathlib.Path, session_id: str) -> None:
213 self._repo_root = repo_root
214 self._session_id = session_id
215 self._session_dir = repo_root / _OP_LOG_DIR / session_id
216 self._ops_path = self._session_dir / "ops.jsonl"
217 self._checkpoint_path = self._session_dir / "checkpoint.json"
218 self._lamport: int = 0
219
220 # ------------------------------------------------------------------
221 # Internal helpers
222 # ------------------------------------------------------------------
223
224 def _ensure_dir(self) -> None:
225 self._session_dir.mkdir(parents=True, exist_ok=True)
226
227 def _load_lamport(self) -> int:
228 """Return the highest lamport_ts seen in the log so far."""
229 if not self._ops_path.exists():
230 return 0
231 highest = 0
232 with self._ops_path.open() as fh:
233 for line in fh:
234 line = line.strip()
235 if not line:
236 continue
237 try:
238 entry: OpEntry = json.loads(line)
239 highest = max(highest, entry.get("lamport_ts", 0))
240 except json.JSONDecodeError:
241 continue
242 return highest
243
244 # ------------------------------------------------------------------
245 # Public API
246 # ------------------------------------------------------------------
247
248 def next_lamport_ts(self) -> int:
249 """Return the next Lamport timestamp to use, advancing the counter.
250
251 The counter is initialised lazily from the highest value found in the
252 log on first call (so that a reopened session continues from where it
253 left off).
254
255 Returns:
256 Monotonically increasing integer.
257 """
258 if self._lamport == 0:
259 self._lamport = self._load_lamport()
260 self._lamport += 1
261 return self._lamport
262
263 def append(self, entry: OpEntry) -> None:
264 """Append *entry* to the op log.
265
266 The entry is serialised as a single JSON line and flushed to disk.
267 This is the only write operation on the log file; entries are never
268 modified or deleted.
269
270 Args:
271 entry: A fully populated :class:`OpEntry`.
272 """
273 self._ensure_dir()
274 line = json.dumps(entry, separators=(",", ":")) + "\n"
275 with self._ops_path.open("a") as fh:
276 fh.write(line)
277 logger.debug(
278 "✅ OpLog append: actor=%r domain=%r ts=%d",
279 entry["actor_id"],
280 entry["domain"],
281 entry["lamport_ts"],
282 )
283
284 def read_all(self) -> list[OpEntry]:
285 """Return all entries in the log, in append order.
286
287 Returns:
288 List of :class:`OpEntry` dicts, oldest first.
289 """
290 if not self._ops_path.exists():
291 return []
292 entries: list[OpEntry] = []
293 with self._ops_path.open() as fh:
294 for line in fh:
295 line = line.strip()
296 if not line:
297 continue
298 try:
299 entries.append(json.loads(line))
300 except json.JSONDecodeError as exc:
301 logger.warning("⚠️ Corrupt op log line in %s: %s", self._ops_path, exc)
302 return entries
303
304 def replay_since_checkpoint(self) -> list[OpEntry]:
305 """Return entries that arrived after the last checkpoint.
306
307 If no checkpoint exists, returns all entries (equivalent to
308 :meth:`read_all`).
309
310 Returns:
311 List of :class:`OpEntry` dicts since last checkpoint, oldest first.
312 """
313 checkpoint = self.read_checkpoint()
314 all_entries = self.read_all()
315 if checkpoint is None:
316 return all_entries
317 cutoff = checkpoint["lamport_ts"]
318 return [e for e in all_entries if e["lamport_ts"] > cutoff]
319
320 def to_structured_delta(self, domain: str) -> StructuredDelta:
321 """Collapse all entries since the last checkpoint into a StructuredDelta.
322
323 Ops are ordered by Lamport timestamp. Ops from domains other than
324 *domain* are filtered out (a session may carry cross-domain ops from
325 coordinated agents; each domain collapses its own slice).
326
327 Args:
328 domain: Domain tag to filter by (e.g. ``"midi"``).
329
330 Returns:
331 A :class:`~muse.domain.StructuredDelta` with the ordered op list
332 and a simple count summary.
333 """
334 entries = self.replay_since_checkpoint()
335 entries.sort(key=lambda e: e["lamport_ts"])
336 ops = [e["domain_op"] for e in entries if e["domain"] == domain]
337
338 counts: _IntMap = {}
339 for op in ops:
340 kind = op.get("op", "unknown")
341 counts[kind] = counts.get(kind, 0) + 1
342 parts = [f"{v} {k}" for k, v in sorted(counts.items())]
343 summary = ", ".join(parts) if parts else "no ops"
344
345 return StructuredDelta(domain=domain, ops=ops, summary=summary)
346
347 def checkpoint(self, snapshot_id: str) -> OpLogCheckpoint:
348 """Write a checkpoint recording that all current ops are in *snapshot_id*.
349
350 After a checkpoint, :meth:`replay_since_checkpoint` will only return
351 ops that arrive after this call. The op log file itself is never
352 truncated — the checkpoint is a logical marker.
353
354 Args:
355 snapshot_id: The Muse snapshot ID that captured all ops to date.
356
357 Returns:
358 The written :class:`OpLogCheckpoint`.
359 """
360 all_entries = self.read_all()
361 highest_ts = max((e["lamport_ts"] for e in all_entries), default=0)
362 ckpt = OpLogCheckpoint(
363 session_id=self._session_id,
364 snapshot_id=snapshot_id,
365 lamport_ts=highest_ts,
366 op_count=len(all_entries),
367 created_at=datetime.datetime.now(datetime.timezone.utc).isoformat(),
368 )
369 self._ensure_dir()
370 write_text_atomic(self._checkpoint_path, json.dumps(ckpt, indent=2) + "\n")
371 logger.info(
372 "✅ OpLog checkpoint: session=%r snapshot=%s ts=%d ops=%d",
373 self._session_id,
374 snapshot_id[:8],
375 highest_ts,
376 len(all_entries),
377 )
378 return ckpt
379
380 def read_checkpoint(self) -> OpLogCheckpoint | None:
381 """Load the most recent checkpoint, or ``None`` if none exists."""
382 if not self._checkpoint_path.exists():
383 return None
384 try:
385 raw: OpLogCheckpoint = json.loads(self._checkpoint_path.read_text())
386 return raw
387 except (json.JSONDecodeError, KeyError) as exc:
388 logger.warning("⚠️ Corrupt checkpoint file %s: %s", self._checkpoint_path, exc)
389 return None
390
391 def session_id(self) -> str:
392 """Return the session ID for this log."""
393 return self._session_id
394
395
396 # ---------------------------------------------------------------------------
397 # Session listing
398 # ---------------------------------------------------------------------------
399
400
401 def list_sessions(repo_root: pathlib.Path) -> list[str]:
402 """Return all session IDs that have op log directories under *repo_root*.
403
404 Args:
405 repo_root: Repository root.
406
407 Returns:
408 Sorted list of session ID strings.
409 """
410 log_dir = repo_root / _OP_LOG_DIR
411 if not log_dir.exists():
412 return []
413 return sorted(p.name for p in log_dir.iterdir() if p.is_dir())
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 32 days ago