"""Concrete systemd observer control with one-shot Linux inotify subscriptions.""" from __future__ import annotations import hashlib import json import os import subprocess from dataclasses import dataclass from pathlib import Path from typing import ClassVar, override from pydantic import BaseModel, ConfigDict, Field from .contract import OBSERVER_NAMESPACE from .fdio import create_control_root from .inotify import InotifyWatch from .observer import ObserverControl, ObserverError, ObserverHead, TimerSubscription class ObserverRow(BaseModel): """Strict machine-consumed observer chain fields.""" model_config: ClassVar[ConfigDict] = ConfigDict(frozen=True, extra="allow") status: str failures: tuple[str, ...] previous_row_digest: str = Field(pattern=r"^[0-9a-f]{64}$") row_digest: str = Field(pattern=r"^[0-9a-f]{64}$") namespace: str | None = None timer_invocation_id: str | None = None @dataclass(frozen=True, slots=True) class ExistingSubscription(TimerSubscription): """Already durable natural result found during crash recovery.""" head: ObserverHead @override def await_pass(self) -> ObserverHead: """Return the already authenticated linked timer head.""" return self.head @dataclass(frozen=True, slots=True) class InotifySubscription(TimerSubscription): """Armed one-shot append subscription.""" watch: InotifyWatch log_path: Path previous: ObserverHead timeout_seconds: int namespace: str @override def await_pass(self) -> ObserverHead: """Await an exact write event and parse the newly durable head.""" self.watch.wait(self.log_path.name, self.timeout_seconds) return _read_head(self.log_path, self.namespace) @dataclass(frozen=True, slots=True) class ProjectedTimerSubscription(TimerSubscription): """Project a natural legacy observer event into the fresh namespace.""" source: InotifySubscription output_log: Path previous: ObserverHead service_unit: str @override def await_pass(self) -> ObserverHead: observed = self.source.await_pass() if observed.status != "PASS" or observed.failures: raise ObserverError("natural observer failed") invocation = _run( ( "systemctl", "--user", "show", self.service_unit, "--property=InvocationID", "--value", ) ).strip() if not invocation: raise ObserverError("missing timer invocation") return _append_fresh(self.output_log, self.previous, invocation) @dataclass(frozen=True, slots=True) class SystemdObserverControl(ObserverControl): """Fence, manually evaluate, arm, and naturally restore one systemd timer.""" baseline_log: Path output_log: Path timer_unit: str service_unit: str timeout_seconds: int @override def fence_timer_and_capture_failed_head(self) -> ObserverHead: """Stop only the timer, verify service inactivity, and bind failed head.""" _ = _run(("systemctl", "--user", "stop", self.timer_unit)) state = _run(("systemctl", "--user", "is-active", self.service_unit), check=False) if state.strip() not in {"inactive", "failed"}: raise ObserverError("observer service active") return _read_head(self.baseline_log, "observer-r71") @override def run_manual(self, previous: ObserverHead) -> ObserverHead: """Invoke the observer service exactly once and project its PASS.""" if self.output_log.is_file(): existing = _read_head(self.output_log, OBSERVER_NAMESPACE) if existing.previous_digest == previous.digest and existing.timer_invocation_id is None: return existing subscription = _arm( self.baseline_log, previous, self.timeout_seconds, "observer-r71", ) _ = _run(("systemctl", "--user", "start", self.service_unit)) observed = subscription.await_pass() if observed.status != "PASS" or observed.failures: raise ObserverError("manual observer failed") return _append_fresh(self.output_log, previous, None) @override def subscribe_timer(self, previous: ObserverHead) -> TimerSubscription: """Recognize a durable recovery result or arm before timer restoration.""" if self.output_log.is_file(): head = _read_head(self.output_log, OBSERVER_NAMESPACE) if head.previous_digest == previous.digest and head.timer_invocation_id is not None: return ExistingSubscription(head) source = _arm( self.baseline_log, previous, self.timeout_seconds, "observer-r71", ) return ProjectedTimerSubscription(source, self.output_log, previous, self.service_unit) @override def restore_timer(self) -> None: """Start the timer without manually triggering its service.""" _ = _run(("systemctl", "--user", "start", self.timer_unit)) def _run(command: tuple[str, ...], *, check: bool = True) -> str: completed = subprocess.run(command, check=check, capture_output=True, text=True) return completed.stdout def _read_head(path: Path, namespace: str) -> ObserverHead: rows = path.read_bytes().splitlines() if not rows: raise ObserverError("empty observer chain") previous = "0" * 64 cross_root_genesis = namespace == OBSERVER_NAMESPACE parsed: ObserverRow | None = None for raw in rows: parsed = ObserverRow.model_validate_json(raw) if cross_root_genesis: previous = parsed.previous_row_digest cross_root_genesis = False body = parsed.model_dump(mode="json", exclude={"row_digest"}, exclude_none=True) canonical = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() if ( parsed.previous_row_digest != previous or hashlib.sha256(canonical).hexdigest() != parsed.row_digest ): raise ObserverError("observer chain") previous = parsed.row_digest if parsed is None: raise ObserverError("empty observer chain") if parsed.namespace is not None and parsed.namespace != namespace: raise ObserverError("observer namespace") return ObserverHead( parsed.row_digest, parsed.status, parsed.previous_row_digest, parsed.failures, namespace, parsed.timer_invocation_id, ) def _append_fresh( path: Path, previous: ObserverHead, invocation: str | None, ) -> ObserverHead: if not path.parent.exists(): create_control_root(path.parent) body: dict[str, str | list[str]] = { "failures": [], "namespace": OBSERVER_NAMESPACE, "previous_row_digest": previous.digest, "status": "PASS", } if invocation is not None: body["timer_invocation_id"] = invocation raw = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() body["row_digest"] = hashlib.sha256(raw).hexdigest() payload = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() + b"\n" descriptor = os.open( path, os.O_APPEND | os.O_CREAT | os.O_WRONLY | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600, ) try: offset = 0 while offset < len(payload): offset += os.write(descriptor, payload[offset:]) os.fsync(descriptor) finally: os.close(descriptor) return _read_head(path, OBSERVER_NAMESPACE) def _arm( path: Path, previous: ObserverHead, timeout_seconds: int, namespace: str, ) -> InotifySubscription: watch = InotifyWatch.arm(os.fsencode(path.parent)) return InotifySubscription( watch, path, previous, timeout_seconds, namespace, )