49637ee7b0
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 21s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
91 lines
3.1 KiB
Python
91 lines
3.1 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,
|
|
}
|
|
|
|
_CONTROLLED_TEMPLATE_COPY_BOUNDARIES: dict[str, bool] = {
|
|
**_SAFE_OPERATION_BOUNDARIES,
|
|
"workflow_modification_allowed": True,
|
|
"auto_push_or_pull_request_trigger_allowed": False,
|
|
"generic_runner_label_allowed": False,
|
|
"runner_pressure_guard_required": True,
|
|
}
|
|
|
|
_BOUNDARY_PROFILES: dict[str, dict[str, bool]] = {
|
|
"safe_read_only": _SAFE_OPERATION_BOUNDARIES,
|
|
"controlled_template_copy": _CONTROLLED_TEMPLATE_COPY_BOUNDARIES,
|
|
}
|
|
|
|
|
|
def safe_operation_boundaries() -> dict[str, bool]:
|
|
"""Return a copy of the locked P0 onboarding operation boundaries."""
|
|
return dict(_SAFE_OPERATION_BOUNDARIES)
|
|
|
|
|
|
def controlled_template_copy_boundaries() -> dict[str, bool]:
|
|
"""Return controlled workflow-copy boundaries with branch triggers closed."""
|
|
return dict(_CONTROLLED_TEMPLATE_COPY_BOUNDARIES)
|
|
|
|
|
|
def source_ready_payload(
|
|
*,
|
|
schema_version: str,
|
|
source_id: str,
|
|
status: str,
|
|
readback: dict[str, Any],
|
|
next_actions: list[str],
|
|
boundary_profile: str = "safe_read_only",
|
|
) -> dict[str, Any]:
|
|
"""Build a stable source-readiness payload without opening runtime gates."""
|
|
if boundary_profile not in _BOUNDARY_PROFILES:
|
|
raise ValueError(f"{source_id}: unknown boundary profile {boundary_profile!r}")
|
|
expected_boundaries = _BOUNDARY_PROFILES[boundary_profile]
|
|
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": dict(expected_boundaries),
|
|
"next_actions": list(next_actions),
|
|
}
|
|
require_operation_boundaries(payload, source_id, expected_boundaries)
|
|
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."""
|
|
require_operation_boundaries(payload, label, _SAFE_OPERATION_BOUNDARIES)
|
|
|
|
|
|
def require_operation_boundaries(
|
|
payload: dict[str, Any],
|
|
label: str,
|
|
expected_boundaries: dict[str, bool],
|
|
) -> None:
|
|
"""Validate a payload against one explicit operation-boundary profile."""
|
|
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 expected_boundaries.items()
|
|
if boundaries.get(key) is not expected
|
|
}
|
|
if mismatches:
|
|
raise ValueError(f"{label}: unsafe operation boundaries: {mismatches}")
|