from __future__ import annotations import hashlib import json import os import stat from pathlib import Path import urllib.error import urllib.request from typing import Any from .errors import DispatchError from .securefs import atomic_write_bytes_shared, ensure_shared_directory, read_secure_bytes class OpenClawSynchronizer: """Synchronize task starts, ends, dispatches, and project summaries. A durable file is always emitted. An optional HTTP endpoint is best-effort; failure leaves the lifecycle record pending for retry. """ def __init__( self, queue_dir: Path, *, endpoint: str | None = None, bearer_token: str | None = None, timeout_seconds: int = 10, ): self.queue_dir = Path(queue_dir) self.endpoint = endpoint self.bearer_token = bearer_token self.timeout_seconds = timeout_seconds ensure_shared_directory(self.queue_dir) def emit(self, record: dict[str, Any]) -> Path: sync_id = record.get("sync_id") if not isinstance(sync_id, str) or not sync_id: raise DispatchError("invalid_sync_id") payload = (json.dumps(record, indent=2, sort_keys=True) + "\n").encode() path = self.queue_dir / "pending" / f"{sync_id}.json" ensure_shared_directory(path.parent) try: st = os.lstat(path) except FileNotFoundError: atomic_write_bytes_shared(path, payload, mode=0o640) else: if not stat.S_ISREG(st.st_mode): raise DispatchError("sync_target_not_regular") if read_secure_bytes(path, expected_uid=None, max_bytes=1_048_576) != payload: raise DispatchError("sync_idempotency_conflict") if self.endpoint: headers = { "Content-Type": "application/json", "Idempotency-Key": sync_id, "User-Agent": "aios-orchestrator-gateway/0.2", } if self.bearer_token: headers["Authorization"] = f"Bearer {self.bearer_token}" request = urllib.request.Request( self.endpoint, data=payload, method="POST", headers=headers, ) try: with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response: response.read(65_536) if response.status not in {200, 201, 202, 204}: raise DispatchError(f"openclaw_sync_http_{response.status}") except urllib.error.HTTPError as exc: raise DispatchError(f"openclaw_sync_http_{exc.code}") from exc except (urllib.error.URLError, TimeoutError, OSError) as exc: raise DispatchError("openclaw_sync_failed") from exc return path def lifecycle_record( *, contract_commit: str, project_key: str, kind: str, work_unit_id: str | None, payload: dict[str, Any], ) -> dict[str, Any]: canonical = json.dumps( { "contract_commit": contract_commit, "project_key": project_key, "kind": kind, "work_unit_id": work_unit_id, "payload": payload, }, separators=(",", ":"), sort_keys=True, ) sync_id = hashlib.sha256(canonical.encode()).hexdigest() return { "schema_version": 1, "sync_id": sync_id, "contract_commit": contract_commit, "project_key": project_key, "kind": kind, "work_unit_id": work_unit_id, "payload": payload, }