"""Human-driven, fail-closed controller for the bounded Tasks 21-25 window. Arming performs only local integrity and quiescence checks. It does not call Telegram, create a customer, activate delivery, or start a service. A human may later use ``launch --start-service`` after the arm receipt is revalidated; the controller still never sends a Telegram message or callback. """ from __future__ import annotations import argparse import hashlib import json import os import stat import subprocess import tempfile import urllib.parse from dataclasses import dataclass from datetime import datetime, timezone from enum import Enum from pathlib import Path from typing import Any, Callable, Mapping, Sequence, cast import yaml CANDIDATE_SCHEMA = "dualcoach-provider-auth-candidate-v1" ARM_RECEIPT_SCHEMA = "dualcoach-tasks21-25-controller-arm-v1" LAUNCH_AUTHORIZATION_SCHEMA = "dualcoach-delivery-launch-authorization-v1" SERVICE_NAME = "hermes-gateway-dualcoachtest.service" TASKS = [21, 22, 23, 24, 25] FIRST_OWNER_TELEGRAM_ACTION = 'Tap "Approve" on the first owner review card.' PROVIDER_AUTH_MODULE = Path(__file__).with_name("dualcoach_admin.py") class ControllerState(str, Enum): ARMED = "armed" UNARMED = "unarmed" @dataclass(frozen=True) class ControllerArmReceipt: state: ControllerState payload: dict[str, object] receipt_path: Path | None class _PrerequisiteFailure(Exception): def __init__(self, reason: str) -> None: super().__init__(reason) self.reason: str = reason def _string_object_mapping(value: object) -> dict[str, object] | None: if not isinstance(value, Mapping): return None result: dict[str, object] = {} for key, item in value.items(): if not isinstance(key, str): return None result[key] = item return result def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _sha256_file(path: Path) -> str: return _sha256_bytes(path.read_bytes()) def _private_regular_file(path: Path, *, sealed: bool = False) -> None: info = path.lstat() allowed_modes = {0o400, 0o600} if sealed else {0o600} if ( path.is_symlink() or not path.is_file() or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) not in allowed_modes ): raise _PrerequisiteFailure("unsafe_input_file") def _read_private_json(path: Path, *, sealed: bool = False) -> dict[str, object]: _private_regular_file(path, sealed=sealed) try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: raise _PrerequisiteFailure("malformed_input") from error if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): raise _PrerequisiteFailure("malformed_input") return value def _require_digest(value: object, reason: str) -> str: if not isinstance(value, str) or len(value) != 64: raise _PrerequisiteFailure(reason) try: int(value, 16) except ValueError as error: raise _PrerequisiteFailure(reason) from error return value def _repository_status_bytes(repository: Path) -> bytes: result = subprocess.run( ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], cwd=repository, check=False, capture_output=True, ) if result.returncode != 0: raise _PrerequisiteFailure("repository_status_unavailable") return result.stdout def _owner_review_recovery_ready(profile: Path) -> bool: """Prove startup has exactly one durable owner-review card to recover.""" ledger_path = ( profile / "data" / "onboarding" / "telegram-room-bootstrap-v1" / "ledger.json" ) try: ledger = _read_private_json(ledger_path) sessions = ledger.get("sessions") if not isinstance(sessions, list): return False recoverable: list[tuple[str, str]] = [] for raw_session in sessions: session = _string_object_mapping(raw_session) draft = ( _string_object_mapping(session.get("customer_draft")) if session is not None else None ) if session is None or draft is None or session.get("state") != "AWAITING_ACTIVATION": continue session_id = session.get("session_id") customer_key = draft.get("customer_key") if ( not isinstance(session_id, str) or not session_id or not isinstance(customer_key, str) or not customer_key or Path(customer_key).name != customer_key or customer_key in {".", ".."} ): return False recoverable.append((session_id, customer_key)) if len(recoverable) != 1: return False session_id, customer_key = recoverable[0] state = _read_private_json( profile / "data" / "customers" / customer_key / "nutrition-onboarding" / "session.json" ) records = _string_object_mapping(state.get("sessions")) record = ( _string_object_mapping(records.get(session_id)) if records is not None else None ) payload = ( _string_object_mapping(record.get("payload")) if record is not None else None ) return bool( record is not None and record.get("state") == "COMMITTED" and payload is not None and payload.get("state") == "owner_review" ) except (OSError, _PrerequisiteFailure): return False def _service_is_inactive(service: str) -> bool: result = subprocess.run( ["systemctl", "--user", "show", service, "--property=ActiveState,SubState", "--value"], check=False, capture_output=True, text=True, ) return result.returncode == 0 and tuple(result.stdout.splitlines()) == ("inactive", "dead") def _profile_process_absent(profile: Path) -> bool: expected = b"HERMES_HOME=" + os.fsencode(profile) for candidate in Path("/proc").iterdir(): if not candidate.name.isdecimal(): continue try: environment = (candidate / "environ").read_bytes().split(b"\0") except (FileNotFoundError, PermissionError, ProcessLookupError, OSError): continue if expected in environment: return False return True def _empty_delivery_document(path: Path) -> bool: if not path.exists(): return True _private_regular_file(path) if path.suffix == ".jsonl": return path.read_bytes().strip() == b"" try: value = json.loads(path.read_text(encoding="utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return False if value in ({}, [], None): return True if isinstance(value, Mapping): for key in ("deliveries", "records", "items"): if key in value: return value[key] == [] return False def _profile_snapshot(profile: Path) -> dict[str, object]: config_path = profile / "config.yaml" auth_path = profile / "auth.json" registry_path = profile / "customers" / "registry.json" state_path = profile / "gateway_state.json" for path in (config_path, auth_path, registry_path, state_path): _private_regular_file(path) try: config = yaml.safe_load(config_path.read_text(encoding="utf-8")) registry = json.loads(registry_path.read_text(encoding="utf-8")) state = json.loads(state_path.read_text(encoding="utf-8")) except (UnicodeDecodeError, json.JSONDecodeError, yaml.YAMLError) as error: raise _PrerequisiteFailure("profile_state_malformed") from error if not isinstance(config, Mapping) or not isinstance(registry, Mapping) or not isinstance(state, Mapping): raise _PrerequisiteFailure("profile_state_malformed") platforms = config.get("platforms") telegram = platforms.get("telegram") if isinstance(platforms, Mapping) else None extra = telegram.get("extra") if isinstance(telegram, Mapping) else None adaptive = extra.get("adaptive_nutrition") if isinstance(extra, Mapping) else None if not isinstance(adaptive, Mapping) or adaptive.get("delivery_enabled") is not False: raise _PrerequisiteFailure("delivery_not_disabled") customers = registry.get("customers") if not isinstance(customers, list) or customers: raise _PrerequisiteFailure("customer_baseline_not_empty") gateway_platforms = state.get("platforms") gateway_telegram = gateway_platforms.get("telegram") if isinstance(gateway_platforms, Mapping) else None if state.get("gateway_state") != "stopped" or not isinstance(gateway_telegram, Mapping) or gateway_telegram.get("state") != "disconnected": raise _PrerequisiteFailure("gateway_state_not_stopped") delivery_paths = ( profile / "data" / "owner-actions" / "draft-deliveries.json", profile / "data" / "scheduled-deliveries.jsonl", profile / "data" / "onboarding" / "telegram-publication-outbox-v1" / "ledger.json", ) if not all(_empty_delivery_document(path) for path in delivery_paths): raise _PrerequisiteFailure("delivery_baseline_not_empty") files = (config_path, auth_path, registry_path, state_path, *delivery_paths) return { "config_sha256": _sha256_file(config_path), "auth_sha256": _sha256_file(auth_path), "registry_sha256": _sha256_file(registry_path), "gateway_state_sha256": _sha256_file(state_path), "delivery_path_sha256": { path.relative_to(profile).as_posix(): _sha256_file(path) if path.exists() else None for path in delivery_paths }, "delivery_count": 0, "customer_count": 0, } def _profile_authorization_digest(profile: Path) -> str: payload = { name: _sha256_file(profile / name) for name in ("config.yaml", "auth.json", "customers/registry.json") } return _sha256_bytes(json.dumps( payload, sort_keys=True, separators=(",", ":") ).encode("utf-8")) def _verify_prior_identity_evidence(path: Path, expected_sha256: str) -> None: receipt = _read_private_json(path) if _sha256_file(path) != expected_sha256: raise _PrerequisiteFailure("synthetic_identity_evidence_drift") preflight = _string_object_mapping(receipt.get("preflight_customer")) if preflight is None or ( preflight.get("archive_record_historically_enabled") is not True or preflight.get("exact_archive_customer_cardinality") != 1 or preflight.get("current_registry_customer_count") != 0 or preflight.get("current_registry_enabled_customer_count") != 0 or preflight.get("private_dm_shape") is not True or preflight.get("distinct_from_owner_and_staff_authorities") is not True ): raise _PrerequisiteFailure("synthetic_identity_evidence_invalid") def _usage_evidence_is_valid(value: Mapping[str, object] | None) -> bool: if value is None: return False input_tokens = value.get("input_tokens") output_tokens = value.get("output_tokens") total_tokens = value.get("total_tokens") if ( type(input_tokens) is not int or type(output_tokens) is not int or type(total_tokens) is not int or input_tokens < 0 or output_tokens < 0 or total_tokens < input_tokens + output_tokens ): return False return True def _verify_provider_receipt(path: Path, candidate_digest: str) -> None: receipt = _read_private_json(path) contract = _string_object_mapping(receipt.get("request_contract")) effects = _string_object_mapping(receipt.get("effects")) usage = _string_object_mapping(receipt.get("usage")) if ( receipt.get("schema") != "dualcoach-provider-auth-receipt-v2" or receipt.get("candidate_digest") != candidate_digest or receipt.get("result") != "ready" or receipt.get("success") is not True or receipt.get("exit_code") != 0 or receipt.get("provider_adapter") != "agent.auxiliary_client.resolve_provider_client" or receipt.get("provider_adapter_version") != "v1" or receipt.get("probe_kind") != "codex_nonpersistent_generation" or receipt.get("active_probe_authorized") is not True or receipt.get("store") is not False or receipt.get("billable") is not True or receipt.get("response_status") != "completed" or receipt.get("sdk_max_retries") != 0 or receipt.get("request_attempts") != 1 or receipt.get("profile_snapshot_unchanged") is not True or contract != { "store": False, "stream": True, "tools": False, "metadata": False, "previous_response_id": False, "conversation_ids": False, "thread_ids": False, "sdk_max_retries": 0, "request_attempts": 1, } or effects != { "delivery_actions": 0, "registry_mutations": 0, "service_actions": 0, "telegram_actions": 0, } or not _usage_evidence_is_valid(usage) ): raise _PrerequisiteFailure("provider_auth_not_ready") def _candidate_binding( manifest_path: Path, workspace_receipt: Path, workspace_index: Path, synthetic_identity_receipt: Path, repository: Path, ) -> tuple[dict[str, object], str]: manifest = _read_private_json(manifest_path, sealed=True) if manifest.get("schema") != CANDIDATE_SCHEMA: raise _PrerequisiteFailure("candidate_manifest_invalid") candidate_digest = _require_digest(manifest.get("candidate_digest"), "candidate_manifest_invalid") if manifest.get("provider_auth_module_sha256") != _sha256_file(PROVIDER_AUTH_MODULE): raise _PrerequisiteFailure("provider_auth_module_drift") if manifest.get("controller_module_sha256") != _sha256_file(Path(__file__)): raise _PrerequisiteFailure("controller_module_drift") bindings = ( ("historical_workspace_receipt_sha256", workspace_receipt, True), ("historical_workspace_index_sha256", workspace_index, True), ("synthetic_identity_evidence_sha256", synthetic_identity_receipt, False), ) for key, path, sealed in bindings: _private_regular_file(path, sealed=sealed) if manifest.get(key) != _sha256_file(path): raise _PrerequisiteFailure("historical_evidence_drift") expected_status_digest = _require_digest(manifest.get("repository_status_sha256"), "candidate_manifest_invalid") if _sha256_bytes(_repository_status_bytes(repository)) != expected_status_digest: raise _PrerequisiteFailure("repository_status_drift") return manifest, candidate_digest def _private_json(path: Path, value: object) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) path.parent.chmod(0o700) descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temporary_path = Path(temporary_name) try: os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as stream: descriptor = -1 json.dump(value, stream, sort_keys=True, separators=(",", ":")) stream.write("\n") stream.flush() os.fsync(stream.fileno()) os.replace(temporary_path, path) path.chmod(0o600) finally: if descriptor >= 0: os.close(descriptor) if temporary_path.exists(): temporary_path.unlink() def _unarmed(reason: str) -> ControllerArmReceipt: return ControllerArmReceipt( state=ControllerState.UNARMED, payload={"schema": ARM_RECEIPT_SCHEMA, "status": "NO_GO", "reason": reason}, receipt_path=None, ) def arm_controller( *, candidate_manifest: Path, provider_receipt: Path, profile: Path, workspace_receipt: Path, workspace_index: Path, synthetic_identity_receipt: Path, repository: Path, state_directory: Path, ) -> ControllerArmReceipt: """Create one private arm receipt after all non-mutating checks pass.""" try: manifest, candidate_digest = _candidate_binding( candidate_manifest, workspace_receipt, workspace_index, synthetic_identity_receipt, repository, ) _verify_provider_receipt(provider_receipt, candidate_digest) _verify_prior_identity_evidence( synthetic_identity_receipt, str(manifest["synthetic_identity_evidence_sha256"]), ) if not _service_is_inactive(SERVICE_NAME): raise _PrerequisiteFailure("service_not_inactive") if not _profile_process_absent(profile): raise _PrerequisiteFailure("profile_process_present") profile_snapshot = _profile_snapshot(profile) if not _owner_review_recovery_ready(profile): raise _PrerequisiteFailure("owner_review_not_ready") except _PrerequisiteFailure as error: return _unarmed(error.reason) payload: dict[str, object] = { "schema": ARM_RECEIPT_SCHEMA, "status": "READY_FOR_CLEAN_WINDOW", "tasks": TASKS, "candidate_digest": candidate_digest, "candidate_manifest_sha256": _sha256_file(candidate_manifest), "provider_receipt_sha256": _sha256_file(provider_receipt), "repository_status_sha256": manifest["repository_status_sha256"], "profile_snapshot": profile_snapshot, "delivery_count": 0, "service_action": "none", "telegram_action": "none", "first_owner_telegram_action": FIRST_OWNER_TELEGRAM_ACTION, "launch_inputs": { "candidate_manifest": str(candidate_manifest), "provider_receipt": str(provider_receipt), "profile": str(profile), "workspace_receipt": str(workspace_receipt), "workspace_index": str(workspace_index), "synthetic_identity_receipt": str(synthetic_identity_receipt), "repository": str(repository), }, } receipt_path = state_directory / f"arm-{candidate_digest}.json" try: _private_json(receipt_path, payload) except OSError: return _unarmed("arm_receipt_write_failed") return ControllerArmReceipt(ControllerState.ARMED, payload, receipt_path) def _launch_inputs(receipt: Mapping[str, object]) -> dict[str, Path]: inputs = _string_object_mapping(receipt.get("launch_inputs")) if inputs is None: raise _PrerequisiteFailure("arm_receipt_invalid") result: dict[str, Path] = {} for key in ( "candidate_manifest", "provider_receipt", "profile", "workspace_receipt", "workspace_index", "synthetic_identity_receipt", "repository", ): value = inputs.get(key) if not isinstance(value, str) or not value: raise _PrerequisiteFailure("arm_receipt_invalid") result[key] = Path(value) return result def _start_service(service: str) -> None: subprocess.run(["systemctl", "--user", "start", service], check=True) def _deployed_candidate_matches( service: str, *, manifest: Mapping[str, object], profile: Path, ) -> bool: """Bind the systemd runtime package to the sealed candidate wheel.""" unit = subprocess.run( [ "systemctl", "--user", "show", service, "--property=ExecStart,Environment,WorkingDirectory", ], check=False, capture_output=True, text=True, ) if unit.returncode != 0: return False properties: dict[str, str] = {} for line in unit.stdout.splitlines(): key, separator, value = line.partition("=") if not separator or key in properties: return False properties[key] = value expected_home = f"HERMES_HOME={profile}" if ( properties.get("WorkingDirectory") != str(profile) or expected_home not in properties.get("Environment", "").split() ): return False exec_start = properties.get("ExecStart", "") marker = "path=" start = exec_start.find(marker) if start < 0: return False start += len(marker) end = exec_start.find(" ;", start) if end < 0: return False executable = Path(exec_start[start:end]) if executable.parent.name != "bin": return False site_packages = tuple( (executable.parent.parent / "lib").glob("python*/site-packages") ) if len(site_packages) != 1: return False direct_urls = tuple( site_packages[0].glob("hermes_agent-*.dist-info/direct_url.json") ) wheel = _string_object_mapping(manifest.get("wheel")) expected_sha256 = wheel.get("sha256") if wheel is not None else None if len(direct_urls) != 1 or not isinstance(expected_sha256, str): return False try: direct_url = json.loads(direct_urls[0].read_text(encoding="utf-8")) archive = _string_object_mapping(direct_url.get("archive_info")) hashes = ( _string_object_mapping(archive.get("hashes")) if archive is not None else None ) url = direct_url.get("url") if ( hashes is None or hashes.get("sha256") != expected_sha256 or not isinstance(url, str) ): return False deployed_wheel = Path(urllib.parse.unquote(urllib.parse.urlparse(url).path)) return deployed_wheel.is_file() and _sha256_file(deployed_wheel) == expected_sha256 except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError): return False def launch_controller( arm_receipt: Path | None, *, start_service: bool, task26_authority_pin: Path | None = None, task26_candidate_digest: str | None = None, ) -> ControllerArmReceipt: """Revalidate an arm receipt and optionally make the human-requested start. No Telegram API or gateway business operation is available from this controller. The only optional mutation is the explicit user launch of the already-verified dedicated systemd service. """ if arm_receipt is None: return _unarmed("arm_receipt_missing") try: receipt = _read_private_json(arm_receipt) if receipt.get("schema") != ARM_RECEIPT_SCHEMA or receipt.get("status") != "READY_FOR_CLEAN_WINDOW": raise _PrerequisiteFailure("arm_receipt_invalid") inputs = _launch_inputs(receipt) manifest, candidate_digest = _candidate_binding( inputs["candidate_manifest"], inputs["workspace_receipt"], inputs["workspace_index"], inputs["synthetic_identity_receipt"], inputs["repository"], ) if receipt.get("candidate_digest") != candidate_digest: raise _PrerequisiteFailure("candidate_binding_drift") _verify_provider_receipt(inputs["provider_receipt"], candidate_digest) _verify_prior_identity_evidence( inputs["synthetic_identity_receipt"], str(manifest["synthetic_identity_evidence_sha256"]), ) if not _service_is_inactive(SERVICE_NAME): raise _PrerequisiteFailure("service_not_inactive") if not _profile_process_absent(inputs["profile"]): raise _PrerequisiteFailure("profile_process_present") if receipt.get("profile_snapshot") != _profile_snapshot(inputs["profile"]): raise _PrerequisiteFailure("profile_baseline_drift") if not _owner_review_recovery_ready(inputs["profile"]): raise _PrerequisiteFailure("owner_review_not_ready") if start_service and ( task26_authority_pin is None or task26_candidate_digest is None ): raise _PrerequisiteFailure("task26_runtime_authority_missing") if start_service and task26_candidate_digest != candidate_digest: raise _PrerequisiteFailure("task26_runtime_candidate_mismatch") if start_service and not _deployed_candidate_matches( SERVICE_NAME, manifest=manifest, profile=inputs["profile"], ): raise _PrerequisiteFailure("deployed_candidate_drift") runtime_snapshot: dict[str, object] | None = None if start_service: from .task26_runtime_authority import FileCandidateAuthoritySource authority_source = FileCandidateAuthoritySource( cast(Path, task26_authority_pin), forbidden_roots=( inputs["profile"], inputs["repository"], inputs["candidate_manifest"], inputs["workspace_receipt"], inputs["workspace_index"], inputs["provider_receipt"], ), ) with authority_source.authorize( candidate_digest, "service_activation" ) as runtime_snapshot: _start_service(SERVICE_NAME) except _PrerequisiteFailure as error: return _unarmed(error.reason) except ValueError: return _unarmed("task26_runtime_authority_rejected") except (OSError, subprocess.CalledProcessError): return _unarmed("service_start_failed") payload: dict[str, object] = { "schema": ARM_RECEIPT_SCHEMA, "status": "AWAITING_HUMAN_TELEGRAM_ACTION", "tasks": TASKS, "candidate_digest": candidate_digest, "service_action": "started_by_human_launch" if start_service else "none", "telegram_action": "none", "first_owner_telegram_action": FIRST_OWNER_TELEGRAM_ACTION, } receipt_path = arm_receipt if start_service: product_fields = { field: _require_digest(manifest.get(field), "candidate_product_identity_invalid") for field in ( "candidate_product_binding_sha256", "hermes_wheel_sha256", "profile_wheel_sha256", ) } if ( product_fields["hermes_wheel_sha256"] == product_fields["profile_wheel_sha256"] or candidate_digest in product_fields.values() ): return _unarmed("candidate_product_identity_invalid") launch_payload: dict[str, object] = { "schema": LAUNCH_AUTHORIZATION_SCHEMA, "status": "AUTHORIZED", "candidate_digest": candidate_digest, "full_digest": str(manifest.get("full_digest") or candidate_digest), "core_digest": str(manifest.get("core_digest") or candidate_digest), "inventory_digest": str(manifest.get("inventory_digest") or candidate_digest), "manifest_digest": _sha256_file(inputs["candidate_manifest"]), "wheel_digest": product_fields["hermes_wheel_sha256"], **product_fields, "provider_config_digest": _sha256_file(inputs["provider_receipt"]), "profile_authorization_digest": _profile_authorization_digest( inputs["profile"] ), "profile": str(inputs["profile"]), "service_action": "started_by_human_launch", "telegram_action": "none", "runtime_authority_snapshot": runtime_snapshot, } receipt_path = arm_receipt.with_name(f"launch-{candidate_digest}.json") try: _private_json(receipt_path, launch_payload) except OSError: return _unarmed("launch_authorization_write_failed") payload = launch_payload return ControllerArmReceipt( state=ControllerState.ARMED, payload=payload, receipt_path=receipt_path, ) def _load_production_nutrition_coordinator( profile: Path, authority_source: object, candidate_digest: str, ) -> object: from gateway.platforms.nutrition_coaching import ( NutritionCoachingCoordinator, load_committed_customer_registry, ) registry, registry_path = load_committed_customer_registry(profile) return NutritionCoachingCoordinator( profile, registry, registry_path=registry_path, delivery_enabled=False, task26_authority_source=authority_source, task26_candidate_digest=candidate_digest, task26_runtime_required=True, ) def issue_delivery_capability( launch_authorization: Path, *, task26_authority_pin: Path, task26_candidate_digest: str, candidate_product_binding_sha256: str, hermes_wheel_sha256: str, profile_wheel_sha256: str, now_provider: Callable[[], datetime] | None = None, ) -> ControllerArmReceipt: """Issue through the production coordinator under external runtime authority.""" now = (now_provider or (lambda: datetime.now(timezone.utc)))() try: launch = _read_private_json(launch_authorization) if ( launch.get("schema") != LAUNCH_AUTHORIZATION_SCHEMA or launch.get("status") != "AUTHORIZED" ): raise _PrerequisiteFailure("launch_authorization_invalid") profile_value = launch.get("profile") if not isinstance(profile_value, str) or not profile_value: raise _PrerequisiteFailure("launch_authorization_invalid") profile = Path(profile_value) if launch.get("profile_authorization_digest") != _profile_authorization_digest(profile): raise _PrerequisiteFailure("profile_authorization_drift") candidate = _require_digest( task26_candidate_digest, "candidate_product_identity_invalid" ) product = { "candidate_product_binding_sha256": _require_digest( candidate_product_binding_sha256, "candidate_product_identity_invalid", ), "hermes_wheel_sha256": _require_digest( hermes_wheel_sha256, "candidate_product_identity_invalid" ), "profile_wheel_sha256": _require_digest( profile_wheel_sha256, "candidate_product_identity_invalid" ), } if ( launch.get("candidate_digest") != candidate or any(launch.get(field) != value for field, value in product.items()) or launch.get("wheel_digest") != product["hermes_wheel_sha256"] or product["hermes_wheel_sha256"] == product["profile_wheel_sha256"] or candidate in product.values() ): raise _PrerequisiteFailure("candidate_product_identity_invalid") from gateway.platforms.nutrition_coaching import DeliveryLaunchAuthorization from gateway.platforms.task26_runtime_authority import ( FileCandidateAuthoritySource, validate_snapshot, ) source = FileCandidateAuthoritySource( task26_authority_pin, forbidden_roots=(profile, launch_authorization.parent), ) predecessor = validate_snapshot( launch.get("runtime_authority_snapshot") ) with source.authorize(candidate, "capability_issue", predecessor): coordinator = _load_production_nutrition_coordinator( profile, source, candidate ) deliveries = cast(Any, coordinator)._read_deliveries() candidates = [ row for row in deliveries.values() if isinstance(row, dict) and row.get("status") == "awaiting_capability" and isinstance(row.get("draft_id"), str) ] if len(candidates) != 1: raise _PrerequisiteFailure("approved_delivery_card_not_unique") row = candidates[0] draft_id = cast(str, row["draft_id"]) generation = cast(Any, coordinator).draft_generation(draft_id) if generation is None: raise _PrerequisiteFailure("delivery_generation_unavailable") launch_pins = { field: _require_digest( launch.get(field), "launch_authorization_invalid" ) for field in ( "candidate_digest", "full_digest", "core_digest", "inventory_digest", "manifest_digest", "wheel_digest", "provider_config_digest", "profile_authorization_digest", ) } authorization = DeliveryLaunchAuthorization( **launch_pins, **product, runtime_authority_snapshot=predecessor, ) capability_id = _sha256_bytes( json.dumps( { "candidate_digest": candidate, **product, "draft_id": draft_id, "generation": generation.generation, "generation_record_digest": generation.record_digest, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ) capability = cast(Any, coordinator).issue_delivery_capability( draft_id, cast(Any, coordinator).owner, launch_authorization=authorization, capability_id=capability_id, issued_at=now, expected_generation=generation.generation, expected_record_digest=generation.record_digest, expected_checkin_revision=generation.checkin_revision, expected_draft_revision=generation.draft_revision, ) ledger_path = profile / "data/owner-actions/draft-deliveries.json" except Exception as error: reason = ( error.reason if isinstance(error, _PrerequisiteFailure) else "delivery_capability_issue_failed" ) return _unarmed(reason) return ControllerArmReceipt( ControllerState.ARMED, { "schema": "draft-delivery-capability-issue-v2", "status": "ISSUED", "capability_id": capability.capability_id, "expires_at": capability.expires_at.isoformat(), "candidate_digest": candidate, **product, "runtime_authority_snapshot": capability.bindings[ "runtime_authority_snapshot" ], "provider_actions": 0, "telegram_actions": 0, }, ledger_path, ) def _public_payload(payload: Mapping[str, object]) -> dict[str, object]: return {key: value for key, value in payload.items() if key != "launch_inputs"} def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="dualcoach_tasks21_25_controller") commands = parser.add_subparsers(dest="command", required=True) arm = commands.add_parser("arm") arm.add_argument("--candidate-manifest", required=True, type=Path) arm.add_argument("--provider-receipt", required=True, type=Path) arm.add_argument("--profile", required=True, type=Path) arm.add_argument("--workspace-receipt", required=True, type=Path) arm.add_argument("--workspace-index", required=True, type=Path) arm.add_argument("--synthetic-identity-receipt", required=True, type=Path) arm.add_argument("--repository", required=True, type=Path) arm.add_argument("--state-directory", required=True, type=Path) arm.add_argument("--json", action="store_true", required=True) launch = commands.add_parser("launch") launch.add_argument("--arm-receipt", required=True, type=Path) launch.add_argument("--start-service", action="store_true") launch.add_argument("--task26-authority-pin", required=True, type=Path) launch.add_argument("--task26-candidate-digest", required=True) launch.add_argument("--json", action="store_true", required=True) issue = commands.add_parser("issue-delivery-capability") issue.add_argument("--launch-authorization", required=True, type=Path) issue.add_argument("--task26-authority-pin", required=True, type=Path) issue.add_argument("--task26-candidate-digest", required=True) issue.add_argument("--candidate-product-binding-sha256", required=True) issue.add_argument("--hermes-wheel-sha256", required=True) issue.add_argument("--profile-wheel-sha256", required=True) issue.add_argument("--json", action="store_true", required=True) return parser def main(argv: Sequence[str] | None = None) -> int: args = _parser().parse_args(argv) if args.command == "arm": receipt = arm_controller( candidate_manifest=args.candidate_manifest, provider_receipt=args.provider_receipt, profile=args.profile, workspace_receipt=args.workspace_receipt, workspace_index=args.workspace_index, synthetic_identity_receipt=args.synthetic_identity_receipt, repository=args.repository, state_directory=args.state_directory, ) elif args.command == "launch": receipt = launch_controller( args.arm_receipt, start_service=args.start_service, task26_authority_pin=args.task26_authority_pin, task26_candidate_digest=args.task26_candidate_digest, ) else: receipt = issue_delivery_capability( args.launch_authorization, task26_authority_pin=args.task26_authority_pin, task26_candidate_digest=args.task26_candidate_digest, candidate_product_binding_sha256=( args.candidate_product_binding_sha256 ), hermes_wheel_sha256=args.hermes_wheel_sha256, profile_wheel_sha256=args.profile_wheel_sha256, ) print(json.dumps(_public_payload(receipt.payload), sort_keys=True, separators=(",", ":"))) return 0 if receipt.state is ControllerState.ARMED else 1 if __name__ == "__main__": raise SystemExit(main())