owner
6ffd7fe19c
Apply the approved Implementation Baseline 1.1 architecture revisions onto the current mainline, verified faithful to the delivered change bundle: - Add aios_baseline: signed owner-delegation-grant.v1 / operational-epoch.v1 / task-principal.v1 auth chain, runtime contract + deployment-truth evaluator, declarative operation registry (effect-class + fail-closed dispatch), lineage ledger, release/recipe/saga ledger, bounded optimization controller, authenticated HTTP transport, resident worker loop and systemd host probe. - Extend existing Job authority (aios_gitops_job_controller/executor) with durable intent/receipt/reconcile and fail-closed admission; harden task_plane/store heartbeat CAS and terminal-state fencing; make reconciliation_controller fail-closed on incomplete/drift evidence. - Add isolated-validation profile templates (deploy/baseline-v1.1, systemd .example), source change/rollback tooling (tools/baseline_v1_1), the AT-01..AT-23 acceptance matrix, M0-M5 runbook, installation/rollback docs and the baseline_v1_1 test suite. - Keep legacy write authorities untouched; isolated-validation profile only. Staged blobs verified byte-identical to the delivered source-changes bundle. Cross-platform baseline tests pass; systemd/fcntl-dependent paths require the declared Linux x86_64 + CPython 3.12 + systemd runtime (site-bound).
122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
"""Thin resident worker loop over the authenticated HTTP transport.
|
|
|
|
Worker policy, operation handlers, effects, receipts, and completion remain in
|
|
the trusted server. This process only maintains its registered transport
|
|
identity and asks the service to run an offered claim.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import threading
|
|
import time
|
|
from typing import Any
|
|
|
|
from .transport import TransportError
|
|
|
|
|
|
class WorkerError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class OperationWorker:
|
|
HEARTBEAT_INTERVAL_SECONDS = 15.0
|
|
|
|
def __init__(self, transport: Any, poll_seconds: float = 1.0, stop_event=None):
|
|
required = (
|
|
"register_worker", "heartbeat", "claim_next", "renew", "execute_claim"
|
|
)
|
|
if transport is None or any(not callable(getattr(transport, name, None)) for name in required):
|
|
raise WorkerError("WORKER_TRANSPORT_INVALID")
|
|
if (
|
|
isinstance(poll_seconds, bool)
|
|
or not isinstance(poll_seconds, (int, float))
|
|
or not math.isfinite(float(poll_seconds))
|
|
or not 0.001 <= poll_seconds <= 60
|
|
):
|
|
raise WorkerError("WORKER_POLL_INTERVAL_INVALID")
|
|
if stop_event is not None and (
|
|
not callable(getattr(stop_event, "is_set", None))
|
|
or not callable(getattr(stop_event, "wait", None))
|
|
):
|
|
raise WorkerError("WORKER_STOP_EVENT_INVALID")
|
|
self.transport = transport
|
|
self.poll_seconds = float(poll_seconds)
|
|
self.stop_event = stop_event if stop_event is not None else threading.Event()
|
|
self.registered = False
|
|
self.worker_id: str | None = None
|
|
self._last_heartbeat = 0.0
|
|
self.last_error: str | None = None
|
|
|
|
def _register(self) -> None:
|
|
result = self.transport.register_worker()
|
|
if (
|
|
not isinstance(result, dict) or result.get("registered") is not True
|
|
or not isinstance(result.get("worker_id"), str) or not result["worker_id"]
|
|
):
|
|
raise WorkerError("WORKER_REGISTRATION_RESULT_INVALID")
|
|
self.worker_id = result["worker_id"]
|
|
self.registered = True
|
|
|
|
def _heartbeat(self, *, force: bool = False) -> None:
|
|
now = time.monotonic()
|
|
if not force and now < self._last_heartbeat + self.HEARTBEAT_INTERVAL_SECONDS:
|
|
return
|
|
result = self.transport.heartbeat()
|
|
if (
|
|
not isinstance(result, dict) or result.get("alive") is not True
|
|
or result.get("worker_id") != self.worker_id
|
|
):
|
|
raise WorkerError("WORKER_HEARTBEAT_RESULT_INVALID")
|
|
self._last_heartbeat = time.monotonic()
|
|
|
|
def run_once(self):
|
|
"""Perform one register/heartbeat/claim/renew/execute iteration."""
|
|
if self.stop_event.is_set():
|
|
return None
|
|
just_registered = False
|
|
if not self.registered:
|
|
self._register()
|
|
just_registered = True
|
|
self._heartbeat(force=just_registered)
|
|
claim = self.transport.claim_next()
|
|
if claim is None:
|
|
self.last_error = None
|
|
return None
|
|
required = {"task_id", "claim_id", "generation", "lease_until", "operation"}
|
|
if not isinstance(claim, dict) or set(claim) != required:
|
|
raise WorkerError("WORKER_CLAIM_RESULT_INVALID")
|
|
task_id = claim["task_id"]
|
|
claim_id = claim["claim_id"]
|
|
if not isinstance(task_id, str) or not task_id or not isinstance(claim_id, str) or not claim_id:
|
|
raise WorkerError("WORKER_CLAIM_RESULT_INVALID")
|
|
renewed = self.transport.renew(task_id, claim_id)
|
|
if (
|
|
not isinstance(renewed, dict) or renewed.get("renewed") is not True
|
|
or renewed.get("task_id") != task_id or renewed.get("claim_id") != claim_id
|
|
):
|
|
raise WorkerError("WORKER_RENEW_RESULT_INVALID")
|
|
result = self.transport.execute_claim(task_id, claim_id)
|
|
if not isinstance(result, dict) or result.get("task_id") != task_id:
|
|
raise WorkerError("WORKER_EXECUTION_RESULT_INVALID")
|
|
self.last_error = None
|
|
return result
|
|
|
|
def run_forever(self) -> None:
|
|
"""Run until the supplied stop event is set, using interruptible waits."""
|
|
while not self.stop_event.is_set():
|
|
try:
|
|
self.run_once()
|
|
except TransportError as exc:
|
|
# Transport failures are observable and retried on the next
|
|
# bounded poll. Re-register after a server restart; programming
|
|
# and schema errors still terminate.
|
|
self.last_error = exc.code
|
|
self.registered = False
|
|
self.stop_event.wait(self.poll_seconds)
|
|
|
|
|
|
Worker = OperationWorker
|
|
|
|
__all__ = ["OperationWorker", "Worker", "WorkerError"]
|