1"""Transactional, local-only orchestration over Sigil's appliance CLI."""23from __future__ import annotations45import asyncio6import re7from pathlib import Path8from typing import Any910from .errors import ControllerError11from .runner import Runner121314FINGERPRINT = re.compile(r"^[0-9a-f]{8}…[0-9a-f]{8}$")15TRANSACTION = re.compile(r"^[0-9a-f]{32}$")16READY_TIMEOUT_SECONDS = 30.017POLL_SECONDS = 0.25181920class Controller:21 def __init__(self, runner: Runner):22 self.runner = runner23 self._mutation_lock = asyncio.Lock()2425 async def _status(self) -> dict[str, Any]:26 return await self.runner.sigil_json(27 [28 "appliance",29 "status",30 "--config",31 str(self.runner.config),32 "--json",33 "--schema-version",34 "2",35 ],36 schema_version=2,37 )3839 async def _service(self) -> dict[str, str]:40 result = await self.runner.systemctl_command("show", check=False)41 values: dict[str, str] = {}42 for raw_line in result.stdout.decode("utf-8", "strict").splitlines():43 key, separator, value = raw_line.partition("=")44 if separator and key in {45 "LoadState",46 "ActiveState",47 "SubState",48 "UnitFileState",49 }:50 values[key] = value[:64]51 return {52 "installed": values.get("LoadState") == "loaded",53 "active": values.get("ActiveState") == "active",54 "active_state": values.get("ActiveState", "unknown"),55 "sub_state": values.get("SubState", "unknown"),56 "unit_file_state": values.get("UnitFileState", "unknown"),57 }5859 async def get_snapshot(self) -> dict[str, Any]:60 service = await self._service()61 if not self.runner.sigil.is_file():62 return {63 "schema_version": 1,64 "installed": False,65 "compatible": False,66 "service": service,67 "appliance": None,68 "capabilities": {69 "stream_diagnostics": False,70 "factory_reset": False,71 },72 }73 try:74 appliance = await self._status()75 compatible = True76 error = None77 except ControllerError as failure:78 appliance = None79 compatible = failure.code != "incompatible_sigil"80 error = failure.code81 response: dict[str, Any] = {82 "schema_version": 1,83 "installed": True,84 "compatible": compatible,85 "service": service,86 "appliance": appliance,87 "capabilities": {88 "stream_diagnostics": False,89 "factory_reset": False,90 },91 }92 if error is not None:93 response["error"] = error94 return response9596 async def get_config(self) -> dict[str, Any]:97 return await self.runner.sigil_json(98 [99 "appliance",100 "config",101 "show",102 "--config",103 str(self.runner.config),104 "--json",105 ]106 )107108 async def validate_config(self, request: dict[str, Any]) -> dict[str, Any]:109 self._validate_config_request(request)110 return await self.runner.sigil_json(111 [112 "appliance",113 "config",114 "validate",115 "--config",116 str(self.runner.config),117 "--json",118 ],119 stdin_value=request,120 )121122 async def restart_service(self) -> dict[str, Any]:123 async with self._mutation_lock:124 previous_instance = await self._current_instance()125 await self.runner.systemctl_command("restart")126 status = await self._wait_ready(previous_instance=previous_instance)127 return {"schema_version": 1, "operation": "restart", "status": status}128129 async def apply_config(self, request: dict[str, Any]) -> dict[str, Any]:130 self._validate_config_request(request)131 async with self._mutation_lock:132 validation = await self.validate_config(request)133 if not validation.get("changed"):134 return {135 "schema_version": 1,136 "operation": "config_apply",137 "changed": False,138 "revision": validation.get("candidate_revision"),139 }140141 was_active = (await self._service())["active"]142 baseline_instance = await self._current_instance()143 transaction: str | None = None144 candidate_instance: str | None = None145 candidate_revision: str | None = None146 commit_response_failure: BaseException | None = None147 try:148 await self._stop_and_wait()149 installed = await self.runner.sigil_json(150 [151 "appliance",152 "config",153 "set",154 "--config",155 str(self.runner.config),156 "--json",157 ],158 stdin_value=request,159 )160 transaction = installed.get("transaction")161 if not isinstance(transaction, str) or not TRANSACTION.fullmatch(transaction):162 raise ControllerError("invalid_response")163 candidate_revision = self._revision(installed.get("candidate_revision"))164 await self.runner.systemctl_command("start")165 ready = await self._wait_ready(166 expected_revision=candidate_revision,167 previous_instance=baseline_instance,168 )169 instance = ready.get("runtime", {}).get("instance_id")170 if not isinstance(instance, str) or not TRANSACTION.fullmatch(instance):171 raise ControllerError("health_not_proven")172 candidate_instance = instance173 await self._stop_and_wait()174 try:175 committed = await self.runner.sigil_json(176 [177 "appliance",178 "config",179 "commit",180 "--config",181 str(self.runner.config),182 "--transaction",183 transaction,184 "--expected-instance",185 instance,186 "--json",187 ]188 )189 committed_revision = self._revision(committed.get("revision"))190 if committed_revision != candidate_revision:191 raise ControllerError("invalid_response")192 except BaseException as failure:193 committed_without_response = await asyncio.shield(194 self._config_is_committed(candidate_revision)195 )196 if not committed_without_response:197 raise198 committed_revision = candidate_revision199 commit_response_failure = failure200 except BaseException as failure:201 rollback_ok = await asyncio.shield(202 self._recover_apply(203 transaction,204 was_active,205 baseline_instance,206 request["expected_revision"],207 )208 )209 if isinstance(failure, asyncio.CancelledError):210 raise211 if not rollback_ok:212 raise ControllerError("rollback_failed") from failure213 raise ControllerError("apply_rolled_back") from failure214215 if was_active:216 try:217 await self._start_and_wait(candidate_instance, committed_revision)218 except BaseException as failure:219 # The transaction is already committed. Never claim that it rolled back220 # or stop the proven candidate after a UI/service-health failure.221 try:222 await asyncio.shield(self.runner.systemctl_command("start"))223 except Exception:224 pass225 if isinstance(failure, asyncio.CancelledError):226 raise227 raise ControllerError("post_commit_service_unhealthy") from failure228 if isinstance(commit_response_failure, asyncio.CancelledError):229 raise commit_response_failure230 return {231 "schema_version": 1,232 "operation": "config_apply",233 "changed": True,234 "revision": committed_revision,235 "commit_response_recovered": commit_response_failure is not None,236 }237238 async def rollback_pending(self, transaction: str) -> dict[str, Any]:239 if not TRANSACTION.fullmatch(transaction):240 raise ControllerError("invalid_request")241 async with self._mutation_lock:242 was_active = (await self._service())["active"]243 previous_instance = await self._current_instance()244 before = await self.get_config()245 pending = before.get("pending_transaction")246 if not isinstance(pending, dict) or pending.get("transaction") != transaction:247 raise ControllerError("transaction_not_found")248 base_revision = self._revision(pending.get("base_revision"))249 await self._stop_and_wait()250 try:251 result = await self.runner.sigil_json(252 [253 "appliance",254 "config",255 "rollback",256 "--config",257 str(self.runner.config),258 "--transaction",259 transaction,260 "--json",261 ]262 )263 restored_revision = self._revision(result.get("restored_revision"))264 if restored_revision != base_revision:265 raise ControllerError("invalid_response")266 except BaseException as failure:267 completed = await asyncio.shield(268 self._rollback_completed(transaction, base_revision)269 )270 if completed and was_active:271 try:272 await asyncio.shield(273 self._start_and_wait(previous_instance, base_revision)274 )275 except Exception as recovery_error:276 raise ControllerError("rollback_service_unhealthy") from recovery_error277 if isinstance(failure, asyncio.CancelledError):278 raise279 if completed:280 raise ControllerError("rollback_response_lost") from failure281 # The candidate may still be live in the config file. Staying stopped is282 # the only fail-closed outcome until the user retries explicit rollback.283 raise ControllerError("rollback_failed_safe") from failure284 if was_active:285 await self._start_and_wait(previous_instance, restored_revision)286 return result287288 async def reset_enrollment(self, expected_host_fingerprint: str) -> dict[str, Any]:289 if not FINGERPRINT.fullmatch(expected_host_fingerprint):290 raise ControllerError("invalid_request")291 async with self._mutation_lock:292 was_active = (await self._service())["active"]293 previous_instance = await self._current_instance()294 failure: BaseException | None = None295 result: dict[str, Any] | None = None296 try:297 await self._stop_and_wait()298 result = await self.runner.sigil_json(299 [300 "appliance",301 "enrollment-reset",302 "--config",303 str(self.runner.config),304 "--expected-host-fingerprint",305 expected_host_fingerprint,306 "--json",307 ]308 )309 except BaseException as error:310 failure = error311 restore_ok = True312 if was_active:313 try:314 await asyncio.shield(315 self._start_and_wait(previous_instance, None)316 )317 except Exception:318 restore_ok = False319 if failure is not None:320 if isinstance(failure, asyncio.CancelledError):321 raise failure322 if not restore_ok:323 raise ControllerError("reset_restore_failed") from failure324 raise failure325 if not restore_ok:326 raise ControllerError("reset_restore_failed")327 if result is None:328 raise ControllerError("invalid_response")329 return result330331 async def _recover_apply(332 self,333 transaction: str | None,334 was_active: bool,335 previous_instance: str | None,336 base_revision: str,337 ) -> bool:338 try:339 await self._stop_and_wait()340 if transaction is None:341 shown = await self.get_config()342 pending = shown.get("pending_transaction")343 if isinstance(pending, dict):344 candidate = pending.get("transaction")345 if isinstance(candidate, str) and TRANSACTION.fullmatch(candidate):346 transaction = candidate347 if transaction is not None:348 rolled_back = await self.runner.sigil_json(349 [350 "appliance",351 "config",352 "rollback",353 "--config",354 str(self.runner.config),355 "--transaction",356 transaction,357 "--json",358 ]359 )360 base_revision = self._revision(rolled_back.get("restored_revision"))361 if was_active:362 await self._start_and_wait(previous_instance, base_revision)363 return True364 except BaseException:365 return False366367 async def _rollback_completed(self, transaction: str, base_revision: str) -> bool:368 try:369 shown = await self.get_config()370 return (371 shown.get("revision") == base_revision372 and shown.get("pending_transaction") is None373 )374 except ControllerError:375 return False376377 async def _config_is_committed(self, candidate_revision: str) -> bool:378 try:379 shown = await self.get_config()380 return (381 shown.get("revision") == candidate_revision382 and shown.get("pending_transaction") is None383 )384 except ControllerError:385 return False386387 async def _current_instance(self) -> str | None:388 try:389 instance = (await self._status()).get("runtime", {}).get("instance_id")390 if instance is None:391 # A successful status read with no instance proves there is no stale392 # daemon snapshot to confuse with the process about to start.393 return ""394 return (395 instance396 if isinstance(instance, str) and TRANSACTION.fullmatch(instance)397 else None398 )399 except ControllerError:400 return None401402 async def _start_and_wait(403 self, previous_instance: str | None, expected_revision: str | None404 ) -> dict[str, Any]:405 await self.runner.systemctl_command("start")406 return await self._wait_ready(407 expected_revision=expected_revision,408 previous_instance=previous_instance,409 )410411 async def _stop_and_wait(self) -> None:412 await self.runner.systemctl_command("stop")413 deadline = asyncio.get_running_loop().time() + READY_TIMEOUT_SECONDS414 while asyncio.get_running_loop().time() < deadline:415 if not (await self._service())["active"]:416 return417 await asyncio.sleep(POLL_SECONDS)418 raise ControllerError("service_stop_timeout")419420 async def _wait_ready(421 self,422 *,423 expected_revision: Any = None,424 previous_instance: Any = None,425 ) -> dict[str, Any]:426 deadline = asyncio.get_running_loop().time() + READY_TIMEOUT_SECONDS427 # Remember why the last poll fell short so a timeout reports the root428 # cause instead of swallowing it: a status call that itself failed429 # (last_error) or a status that came back but never proved ready430 # (last_gap).431 last_error: ControllerError | None = None432 last_gap: str | None = None433 while asyncio.get_running_loop().time() < deadline:434 try:435 status = await self._status()436 runtime = status.get("runtime", {})437 instance = runtime.get("instance_id")438 revision_ok = (439 expected_revision is None440 or runtime.get("loaded_config_revision") == expected_revision441 )442 instance_ok = (443 isinstance(previous_instance, str) and instance != previous_instance444 )445 if (446 status.get("overall") in {"ready", "active"}447 and runtime.get("state") == "fresh"448 and runtime.get("daemon") == "ready"449 and isinstance(instance, str)450 and revision_ok451 and instance_ok452 ):453 return status454 last_error = None455 last_gap = self._describe_readiness_gap(456 status, runtime, expected_revision, previous_instance457 )458 except ControllerError as error:459 last_error = error460 last_gap = None461 await asyncio.sleep(POLL_SECONDS)462 detail = last_error.code if last_error is not None else last_gap463 raise ControllerError("service_ready_timeout", detail) from last_error464465466 def _describe_readiness_gap(467 status: dict[str, Any],468 runtime: dict[str, Any],469 expected_revision: Any,470 previous_instance: Any,471 ) -> str:472 """Name the first unmet readiness condition, for timeout diagnostics."""473 overall = status.get("overall")474 if overall not in {"ready", "active"}:475 return f"overall={overall!r}"476 if runtime.get("state") != "fresh":477 return f"runtime.state={runtime.get('state')!r}"478 if runtime.get("daemon") != "ready":479 return f"runtime.daemon={runtime.get('daemon')!r}"480 instance = runtime.get("instance_id")481 if not isinstance(instance, str):482 return "runtime.instance_id missing"483 if (484 expected_revision is not None485 and runtime.get("loaded_config_revision") != expected_revision486 ):487 return (488 f"loaded_config_revision={runtime.get('loaded_config_revision')!r} "489 f"!= expected {expected_revision!r}"490 )491 if isinstance(previous_instance, str) and instance == previous_instance:492 return "runtime.instance_id unchanged from previous"493 return "readiness conditions not met"494495496 def _revision(value: Any) -> str:497 if not isinstance(value, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", value):498 raise ControllerError("invalid_response")499 return value500501502 def _validate_config_request(request: dict[str, Any]) -> None:503 if not isinstance(request, dict) or set(request) != {504 "schema_version",505 "expected_revision",506 "settings",507 }:508 raise ControllerError("invalid_request")509 if request["schema_version"] != 1:510 raise ControllerError("invalid_request")511 revision = request["expected_revision"]512 if not isinstance(revision, str) or not re.fullmatch(513 r"sha256:[0-9a-f]{64}", revision514 ):515 raise ControllerError("invalid_request")516 settings = request["settings"]517 if not isinstance(settings, dict) or set(settings) != {518 "resolution",519 "framerate",520 "rate_control",521 }:522 raise ControllerError("invalid_request")523 framerate = settings["framerate"]524 if (525 not isinstance(framerate, int)526 or isinstance(framerate, bool)527 or not 1 <= framerate <= 240528 ):529 raise ControllerError("invalid_request")530 resolution = settings["resolution"]531 if not isinstance(resolution, dict) or resolution.get("mode") not in {532 "native",533 "fixed",534 }:535 raise ControllerError("invalid_request")536 if resolution["mode"] == "native":537 if set(resolution) != {"mode"}:538 raise ControllerError("invalid_request")539 else:540 if set(resolution) != {"mode", "width", "height"}:541 raise ControllerError("invalid_request")542 for field in ("width", "height"):543 value = resolution[field]544 if (545 not isinstance(value, int)546 or isinstance(value, bool)547 or value < 64548 or value % 2549 ):550 raise ControllerError("invalid_request")551 if resolution["width"] > 7680 or resolution["height"] > 4320:552 raise ControllerError("invalid_request")553 rate_control = settings["rate_control"]554 if rate_control is None:555 return556 if not isinstance(rate_control, dict):557 raise ControllerError("invalid_request")558 mode = rate_control.get("mode")559 if mode == "cbr":560 if set(rate_control) != {"mode", "bitrate_kbps"}:561 raise ControllerError("invalid_request")562 value = rate_control["bitrate_kbps"]563 if not isinstance(value, int) or isinstance(value, bool) or not 1000 <= value <= 100000:564 raise ControllerError("invalid_request")565 elif mode == "cqp":566 if set(rate_control) != {"mode", "quantizer"}:567 raise ControllerError("invalid_request")568 value = rate_control["quantizer"]569 if not isinstance(value, int) or isinstance(value, bool) or not 1 <= value <= 51:570 raise ControllerError("invalid_request")571 else:572 raise ControllerError("invalid_request")