From d11c792ef229c918a121b465a1beaf93174dc558 Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 29 May 2026 19:45:20 -0700 Subject: [PATCH] feat(sync): add local sync journal --- core/sync/journal.py | 197 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 core/sync/journal.py diff --git a/core/sync/journal.py b/core/sync/journal.py new file mode 100644 index 00000000..be826022 --- /dev/null +++ b/core/sync/journal.py @@ -0,0 +1,197 @@ +"""Local sync journal for edge/cloud artifact transfer state. + +The journal is deliberately local and transport-agnostic. It records pending +uploads/downloads and their outcomes without blocking active reasoning, +refusal, safety, or action gates. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum + + +class JournalDirection(str, Enum): + """Direction of a sync journal entry.""" + + UPLOAD = "upload" + DOWNLOAD = "download" + + +class JournalStatus(str, Enum): + """Lifecycle status for a sync journal entry.""" + + PENDING = "pending" + ACKNOWLEDGED = "acknowledged" + COMPLETED = "completed" + REJECTED = "rejected" + FAILED = "failed" + + +@dataclass(frozen=True, slots=True) +class JournalEntry: + """One local sync work item or outcome.""" + + entry_id: str + direction: JournalDirection + artifact_id: str + artifact_type: str + status: JournalStatus = JournalStatus.PENDING + target_uri: str | None = None + reason: str | None = None + attempts: int = 0 + + +@dataclass(frozen=True, slots=True) +class JournalDecision: + """Result of mutating the local journal.""" + + accepted: bool + reason: str + entry: JournalEntry | None = None + + +@dataclass(slots=True) +class LocalSyncJournal: + """In-memory local journal for sync work. + + A later file-backed journal can implement the same semantics. This first + version proves the contract: sync failures are recorded locally and never + imply hot-path failure or active release mutation. + """ + + _entries: dict[str, JournalEntry] = field(default_factory=dict) + + def append_upload( + self, + *, + entry_id: str, + artifact_id: str, + artifact_type: str, + target_uri: str | None = None, + ) -> JournalDecision: + return self._append( + entry_id=entry_id, + direction=JournalDirection.UPLOAD, + artifact_id=artifact_id, + artifact_type=artifact_type, + target_uri=target_uri, + ) + + def append_download( + self, + *, + entry_id: str, + artifact_id: str, + artifact_type: str, + target_uri: str | None = None, + ) -> JournalDecision: + return self._append( + entry_id=entry_id, + direction=JournalDirection.DOWNLOAD, + artifact_id=artifact_id, + artifact_type=artifact_type, + target_uri=target_uri, + ) + + def mark_acknowledged(self, entry_id: str) -> JournalDecision: + return self._transition(entry_id, JournalStatus.ACKNOWLEDGED, "acknowledged") + + def mark_completed(self, entry_id: str) -> JournalDecision: + return self._transition(entry_id, JournalStatus.COMPLETED, "completed") + + def mark_rejected(self, entry_id: str, reason: str) -> JournalDecision: + return self._transition(entry_id, JournalStatus.REJECTED, reason) + + def mark_failed(self, entry_id: str, reason: str) -> JournalDecision: + return self._transition(entry_id, JournalStatus.FAILED, reason, increment_attempts=True) + + def retry(self, entry_id: str) -> JournalDecision: + entry = self._entries.get(entry_id) + if entry is None: + return JournalDecision(False, "unknown_entry") + if entry.status not in {JournalStatus.FAILED, JournalStatus.REJECTED}: + return JournalDecision(False, "retry_not_needed", entry) + retried = JournalEntry( + entry_id=entry.entry_id, + direction=entry.direction, + artifact_id=entry.artifact_id, + artifact_type=entry.artifact_type, + status=JournalStatus.PENDING, + target_uri=entry.target_uri, + reason=None, + attempts=entry.attempts, + ) + self._entries[entry_id] = retried + return JournalDecision(True, "pending", retried) + + def get(self, entry_id: str) -> JournalEntry | None: + return self._entries.get(entry_id) + + def pending(self) -> tuple[JournalEntry, ...]: + return tuple(entry for entry in self._entries.values() if entry.status is JournalStatus.PENDING) + + def entries(self) -> tuple[JournalEntry, ...]: + return tuple(self._entries.values()) + + def as_dicts(self) -> tuple[dict[str, object], ...]: + return tuple( + { + "entry_id": entry.entry_id, + "direction": entry.direction.value, + "artifact_id": entry.artifact_id, + "artifact_type": entry.artifact_type, + "status": entry.status.value, + "target_uri": entry.target_uri, + "reason": entry.reason, + "attempts": entry.attempts, + } + for entry in self._entries.values() + ) + + def _append( + self, + *, + entry_id: str, + direction: JournalDirection, + artifact_id: str, + artifact_type: str, + target_uri: str | None, + ) -> JournalDecision: + if not entry_id: + return JournalDecision(False, "missing_entry_id") + if entry_id in self._entries: + return JournalDecision(False, "duplicate_entry_id", self._entries[entry_id]) + entry = JournalEntry( + entry_id=entry_id, + direction=direction, + artifact_id=artifact_id, + artifact_type=artifact_type, + target_uri=target_uri, + ) + self._entries[entry_id] = entry + return JournalDecision(True, "pending", entry) + + def _transition( + self, + entry_id: str, + status: JournalStatus, + reason: str, + *, + increment_attempts: bool = False, + ) -> JournalDecision: + entry = self._entries.get(entry_id) + if entry is None: + return JournalDecision(False, "unknown_entry") + updated = JournalEntry( + entry_id=entry.entry_id, + direction=entry.direction, + artifact_id=entry.artifact_id, + artifact_type=entry.artifact_type, + status=status, + target_uri=entry.target_uri, + reason=reason, + attempts=entry.attempts + 1 if increment_attempts else entry.attempts, + ) + self._entries[entry_id] = updated + return JournalDecision(True, reason, updated)