ea50d453a5
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""Shared read-only contracts for AWOOOI onboarding source readiness."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
_SAFE_OPERATION_BOUNDARIES: dict[str, bool] = {
|
|
"read_only_api_allowed": True,
|
|
"workflow_modification_allowed": False,
|
|
"workflow_trigger_allowed": False,
|
|
"repo_creation_allowed": False,
|
|
"refs_sync_allowed": False,
|
|
"github_api_allowed": False,
|
|
"host_or_k8s_write_allowed": False,
|
|
"secret_read_allowed": False,
|
|
"raw_session_or_sqlite_read_allowed": False,
|
|
}
|
|
|
|
|
|
def safe_operation_boundaries() -> dict[str, bool]:
|
|
"""Return a copy of the locked P0 onboarding operation boundaries."""
|
|
return dict(_SAFE_OPERATION_BOUNDARIES)
|
|
|
|
|
|
def source_ready_payload(
|
|
*,
|
|
schema_version: str,
|
|
source_id: str,
|
|
status: str,
|
|
readback: dict[str, Any],
|
|
next_actions: list[str],
|
|
) -> dict[str, Any]:
|
|
"""Build a stable source-readiness payload without opening runtime gates."""
|
|
payload: dict[str, Any] = {
|
|
"schema_version": schema_version,
|
|
"generated_at": "2026-06-29T12:45:00+08:00",
|
|
"source_id": source_id,
|
|
"status": status,
|
|
"readback": deepcopy(readback),
|
|
"operation_boundaries": safe_operation_boundaries(),
|
|
"next_actions": list(next_actions),
|
|
}
|
|
require_safe_operation_boundaries(payload, source_id)
|
|
return payload
|
|
|
|
|
|
def require_safe_operation_boundaries(payload: dict[str, Any], label: str) -> None:
|
|
"""Keep recreated onboarding services read-only until the apply gate opens."""
|
|
boundaries = payload.get("operation_boundaries")
|
|
if not isinstance(boundaries, dict):
|
|
raise ValueError(f"{label}: operation_boundaries must be an object")
|
|
mismatches = {
|
|
key: boundaries.get(key)
|
|
for key, expected in _SAFE_OPERATION_BOUNDARIES.items()
|
|
if boundaries.get(key) is not expected
|
|
}
|
|
if mismatches:
|
|
raise ValueError(f"{label}: unsafe operation boundaries: {mismatches}")
|