wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

feat(api): expose github owner response intake readiness

Browse Source
This commit is contained in:
Your Name
2026-06-27 20:51:17 +08:00
parent de609a79ae
commit 80138e9854
4 changed files with 537 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

416
apps/api/src/services/github_target_private_backup_evidence_gate.py
View File

@@ -34,7 +34,9 @@ def load_latest_github_target_private_backup_evidence_gate(
approval_package = _load_snapshot(directory / _APPROVAL_PACKAGE_FILE)
probe = _load_snapshot(directory / _PROBE_FILE)
connector_readback = _load_optional_snapshot(directory / _CONNECTOR_READBACK_FILE)
missing_source_readiness = _load_optional_snapshot(directory / _MISSING_SOURCE_READINESS_FILE)
missing_source_readiness = _load_optional_snapshot(
directory / _MISSING_SOURCE_READINESS_FILE
)
_require_source_contracts(
decision=decision,
@@ -90,6 +92,12 @@ def build_github_target_private_backup_evidence_gate(
if row.get("github_repo")
}
owner_summary = _dict(owner_response.get("summary"))
owner_response_intake_readiness = _owner_response_intake_readiness(owner_response)
owner_template_by_repo = {
str(row.get("github_repo")): _dict(row)
for row in _list(owner_response.get("response_templates"))
if row.get("github_repo")
}
targets = [
_build_target(
@@ -97,7 +105,12 @@ def build_github_target_private_backup_evidence_gate(
probe=probe_by_repo.get(str(row.get("github_repo")), {}),
approval_item=package_by_repo.get(str(row.get("github_repo")), {}),
connector_readback=connector_by_repo.get(str(row.get("github_repo")), {}),
missing_source_readiness=missing_source_by_repo.get(str(row.get("github_repo")), {}),
missing_source_readiness=missing_source_by_repo.get(
str(row.get("github_repo")), {}
),
owner_response_template=owner_template_by_repo.get(
str(row.get("github_repo")), {}
),
)
for row in decisions
]
@@ -106,7 +119,8 @@ def build_github_target_private_backup_evidence_gate(
public_probe_visible_targets = [
row
for row in approval_required_targets
if row["visibility_evidence_status"] == "blocked_public_probe_visible_private_evidence_required"
if row["visibility_evidence_status"]
== "blocked_public_probe_visible_private_evidence_required"
]
not_found_or_private_targets = [
row
@@ -114,7 +128,9 @@ def build_github_target_private_backup_evidence_gate(
if row["visibility_evidence_status"] == "blocked_private_or_absent_not_verified"
]
external_scope_targets = [
row for row in targets if row["visibility_evidence_status"] == "external_scope_not_backup_target"
row
for row in targets
if row["visibility_evidence_status"] == "external_scope_not_backup_target"
]
blocked_targets = [
row
@@ -122,8 +138,12 @@ def build_github_target_private_backup_evidence_gate(
if not row["private_backup_verified"] or not row["execution_ready"]
]
private_backup_verified_count = sum(1 for row in approval_required_targets if row["private_backup_verified"])
forbidden_actions = sorted({item for row in targets for item in row["forbidden_actions"]})
private_backup_verified_count = sum(
1 for row in approval_required_targets if row["private_backup_verified"]
)
forbidden_actions = sorted(
{item for row in targets for item in row["forbidden_actions"]}
)
return {
"schema_version": _SCHEMA_VERSION,
@@ -143,14 +163,18 @@ def build_github_target_private_backup_evidence_gate(
"summary": {
"target_decision_count": len(targets),
"approval_required_target_count": len(approval_required_targets),
"approval_package_item_count": len(_list(approval_package.get("approval_items"))),
"approval_package_item_count": len(
_list(approval_package.get("approval_items"))
),
"github_connector_owner_visible_repository_count": _int(
owner_readback.get("visible_repository_count")
),
"github_connector_owner_visible_private_repository_count": _int(
owner_readback.get("visible_private_repository_count")
),
"github_connector_readback_count": len(_list(connector_payload.get("targets"))),
"github_connector_readback_count": len(
_list(connector_payload.get("targets"))
),
"github_connector_private_visibility_count": sum(
1
for row in _list(connector_payload.get("targets"))
@@ -161,7 +185,9 @@ def build_github_target_private_backup_evidence_gate(
for row in _list(connector_payload.get("targets"))
if _dict(row).get("readback_status") == "not_found_or_inaccessible"
),
"github_missing_target_resolution_count": len(_list(missing_payload.get("targets"))),
"github_missing_target_resolution_count": len(
_list(missing_payload.get("targets"))
),
"github_missing_target_gitea_source_candidate_count": _int(
missing_summary.get("gitea_source_candidate_count")
),
@@ -177,17 +203,52 @@ def build_github_target_private_backup_evidence_gate(
"github_missing_target_create_private_repo_ready_count": _int(
missing_summary.get("create_private_repo_ready_count")
),
"github_missing_target_refs_sync_ready_count": _int(missing_summary.get("refs_sync_ready_count")),
"github_missing_target_refs_sync_ready_count": _int(
missing_summary.get("refs_sync_ready_count")
),
"public_probe_visible_target_count": len(public_probe_visible_targets),
"not_found_or_private_target_count": len(not_found_or_private_targets),
"private_backup_verified_count": private_backup_verified_count,
"private_visibility_verified_count": private_backup_verified_count,
"private_visibility_evidence_missing_count": len(approval_required_targets) - private_backup_verified_count,
"private_visibility_evidence_missing_count": len(approval_required_targets)
- private_backup_verified_count,
"safe_credential_required_count": len(approval_required_targets),
"safe_credential_accepted_evidence_count": 0,
"owner_response_received_count": _int(owner_summary.get("received_response_count")),
"owner_response_accepted_count": _int(owner_summary.get("accepted_response_count")),
"execution_ready_count": sum(1 for row in approval_required_targets if row["execution_ready"]),
"owner_response_received_count": _int(
owner_summary.get("received_response_count")
),
"owner_response_accepted_count": _int(
owner_summary.get("accepted_response_count")
),
"owner_response_request_ready": owner_response_intake_readiness[
"request_ready"
],
"owner_response_required_response_item_count": owner_response_intake_readiness[
"required_response_item_count"
],
"owner_response_requested_template_count": owner_response_intake_readiness[
"requested_template_count"
],
"owner_response_template_count": owner_response_intake_readiness[
"response_template_count"
],
"owner_response_allowed_response_field_count": owner_response_intake_readiness[
"allowed_response_field_count"
],
"owner_response_forbidden_payload_count": owner_response_intake_readiness[
"forbidden_payload_count"
],
"owner_response_collection_check_count": owner_response_intake_readiness[
"collection_check_count"
],
"owner_response_intake_preflight_check_count": owner_response_intake_readiness[
"intake_preflight_check_count"
],
"owner_response_request_execution_authorized": False,
"github_target_owner_response_handoff_not_approval": True,
"execution_ready_count": sum(
1 for row in approval_required_targets if row["execution_ready"]
),
"blocked_target_count": len(blocked_targets),
"external_scope_target_count": len(external_scope_targets),
"forbidden_action_count": len(forbidden_actions),
@@ -202,6 +263,7 @@ def build_github_target_private_backup_evidence_gate(
"not_found_or_private_as_absent_allowed": False,
"public_repo_allowed": False,
},
"owner_response_intake_readiness": owner_response_intake_readiness,
"targets": targets,
"acceptance_requirements": _acceptance_requirements(owner_response),
"rejection_rules": _rejection_rules(owner_response),
@@ -238,6 +300,7 @@ def _build_target(
approval_item: dict[str, Any],
connector_readback: dict[str, Any],
missing_source_readiness: dict[str, Any],
owner_response_template: dict[str, Any],
) -> dict[str, Any]:
github_repo = str(decision.get("github_repo") or "")
probe_status = str(probe.get("status") or decision.get("probe_status") or "unknown")
@@ -255,7 +318,9 @@ def _build_target(
probe_status=probe_status,
private_visibility_verified=private_visibility_verified,
)
blockers = _target_blockers(visibility_status, approval_required, private_visibility_verified)
blockers = _target_blockers(
visibility_status, approval_required, private_visibility_verified
)
forbidden_actions = _strings(approval_item.get("still_forbidden")) or [
"create_github_repo",
"change_repo_visibility",
@@ -275,14 +340,26 @@ def _build_target(
"target_state": target_state,
"risk": str(decision.get("risk") or "UNKNOWN"),
"visibility_evidence_status": visibility_status,
"github_connector_readback_status": str(connector_readback.get("readback_status") or "not_checked"),
"github_connector_readback_status": str(
connector_readback.get("readback_status") or "not_checked"
),
"github_connector_visibility": connector_readback.get("visibility"),
"missing_target_source_resolution_status": missing_source_readiness.get("source_resolution_status"),
"missing_target_source_candidate_type": missing_source_readiness.get("source_candidate_type"),
"missing_target_source_resolution_status": missing_source_readiness.get(
"source_resolution_status"
),
"missing_target_source_candidate_type": missing_source_readiness.get(
"source_candidate_type"
),
"missing_target_gitea_repo": missing_source_readiness.get("gitea_repo"),
"missing_target_local_worktree_status": missing_source_readiness.get("local_worktree_status"),
"missing_target_create_private_repo_ready": missing_source_readiness.get("create_private_repo_ready"),
"missing_target_refs_sync_ready": missing_source_readiness.get("refs_sync_ready"),
"missing_target_local_worktree_status": missing_source_readiness.get(
"local_worktree_status"
),
"missing_target_create_private_repo_ready": missing_source_readiness.get(
"create_private_repo_ready"
),
"missing_target_refs_sync_ready": missing_source_readiness.get(
"refs_sync_ready"
),
"missing_target_required_owner_decisions": _strings(
missing_source_readiness.get("required_owner_decisions")
),
@@ -290,12 +367,42 @@ def _build_target(
"private_visibility_owner_evidence_ref": connector_readback.get("evidence_ref"),
"safe_credential_evidence_status": "not_collected",
"safe_credential_evidence_ref": None,
"owner_response_template_id": owner_response_template.get("template_id"),
"owner_response_submission_status": "waiting_owner_response"
if approval_required
else "not_required",
"owner_response_requested_decision": owner_response_template.get(
"requested_owner_decision"
),
"owner_response_required_fields": _strings(
owner_response_template.get("required_owner_fields")
),
"owner_response_acceptable_decisions": _strings(
owner_response_template.get("acceptable_decisions")
),
"owner_response_minimum_evidence_refs": _strings(
owner_response_template.get("minimum_evidence_refs")
),
"owner_response_acceptance_criteria": _strings(
owner_response_template.get("acceptance_criteria")
),
"owner_response_rejection_conditions": _strings(
owner_response_template.get("rejection_conditions")
),
"owner_response_allowed_outputs": _strings(
owner_response_template.get("allowed_outputs")
),
"owner_response_execution_authorized": False,
"owner_response_accepted": False,
"refs_sync_ready": False,
"execution_ready": False,
"blockers": blockers,
"evidence_refs": _strings(decision.get("evidence_refs")),
"next_action": str(approval_item.get("approval_action") or decision.get("recommended_action") or ""),
"next_action": str(
approval_item.get("approval_action")
or decision.get("recommended_action")
or ""
),
"forbidden_actions": forbidden_actions,
"repo_creation_authorized": False,
"visibility_change_authorized": False,
@@ -367,11 +474,21 @@ def _require_source_contracts(
missing_source_readiness: dict[str, Any],
) -> None:
_require_schema(decision, "github_target_decision_v1", _DECISION_FILE)
_require_schema(owner_response, "github_target_owner_decision_response_v1", _OWNER_RESPONSE_FILE)
_require_schema(approval_package, "github_target_repo_approval_package_v1", _APPROVAL_PACKAGE_FILE)
_require_schema(
owner_response, "github_target_owner_decision_response_v1", _OWNER_RESPONSE_FILE
)
_require_schema(
approval_package,
"github_target_repo_approval_package_v1",
_APPROVAL_PACKAGE_FILE,
)
_require_schema(probe, "github_target_probe_v1", _PROBE_FILE)
if connector_readback:
_require_schema(connector_readback, "github_target_connector_readback_v1", _CONNECTOR_READBACK_FILE)
_require_schema(
connector_readback,
"github_target_connector_readback_v1",
_CONNECTOR_READBACK_FILE,
)
if missing_source_readiness:
_require_schema(
missing_source_readiness,
@@ -382,9 +499,13 @@ def _require_source_contracts(
_require_probe_consistency(probe, _PROBE_FILE)
_require_approval_package_consistency(approval_package, _APPROVAL_PACKAGE_FILE)
if connector_readback:
_require_connector_readback_consistency(connector_readback, _CONNECTOR_READBACK_FILE)
_require_connector_readback_consistency(
connector_readback, _CONNECTOR_READBACK_FILE
)
if missing_source_readiness:
_require_missing_source_readiness_consistency(missing_source_readiness, _MISSING_SOURCE_READINESS_FILE)
_require_missing_source_readiness_consistency(
missing_source_readiness, _MISSING_SOURCE_READINESS_FILE
)
_require_owner_response_boundaries(owner_response, _OWNER_RESPONSE_FILE)
@@ -398,7 +519,9 @@ def _require_decision_consistency(payload: dict[str, Any], label: str) -> None:
decisions = _list(payload.get("decisions"))
if _int(payload.get("decision_count")) != len(decisions):
raise ValueError(f"{label}: decision_count must equal decisions length")
actual_approval_required = sum(1 for row in decisions if _dict(row).get("approval_required") is True)
actual_approval_required = sum(
1 for row in decisions if _dict(row).get("approval_required") is True
)
if _int(payload.get("approval_required_count")) != actual_approval_required:
raise ValueError(f"{label}: approval_required_count must match decisions")
@@ -408,7 +531,9 @@ def _require_probe_consistency(payload: dict[str, Any], label: str) -> None:
if _int(payload.get("candidate_count")) != len(candidates):
raise ValueError(f"{label}: candidate_count must equal candidates length")
exists_count = sum(1 for row in candidates if _dict(row).get("status") == "exists")
not_found_count = sum(1 for row in candidates if _dict(row).get("status") == "not_found_or_private")
not_found_count = sum(
1 for row in candidates if _dict(row).get("status") == "not_found_or_private"
)
if _int(payload.get("exists_count")) != exists_count:
raise ValueError(f"{label}: exists_count must match candidates")
if _int(payload.get("not_found_or_private_count")) != not_found_count:
@@ -421,7 +546,9 @@ def _require_approval_package_consistency(payload: dict[str, Any], label: str) -
raise ValueError(f"{label}: package_count must equal approval_items length")
def _require_connector_readback_consistency(payload: dict[str, Any], label: str) -> None:
def _require_connector_readback_consistency(
payload: dict[str, Any], label: str
) -> None:
summary = _dict(payload.get("summary"))
owner_readback = _dict(payload.get("owner_readback"))
targets = [_dict(row) for row in _list(payload.get("targets"))]
@@ -429,12 +556,26 @@ def _require_connector_readback_consistency(payload: dict[str, Any], label: str)
raise ValueError(f"{label}: summary.target_count must equal targets length")
if owner_readback:
visible_repositories = _list(owner_readback.get("visible_repositories"))
if _int(owner_readback.get("visible_repository_count")) != len(visible_repositories):
raise ValueError(f"{label}: owner visible repository count must match visible repositories")
if _int(owner_readback.get("visible_private_repository_count")) > len(visible_repositories):
raise ValueError(f"{label}: owner private repository count cannot exceed visible repositories")
private_count = sum(1 for row in targets if row.get("private_visibility_verified") is True)
not_found_count = sum(1 for row in targets if row.get("readback_status") == "not_found_or_inaccessible")
if _int(owner_readback.get("visible_repository_count")) != len(
visible_repositories
):
raise ValueError(
f"{label}: owner visible repository count must match visible repositories"
)
if _int(owner_readback.get("visible_private_repository_count")) > len(
visible_repositories
):
raise ValueError(
f"{label}: owner private repository count cannot exceed visible repositories"
)
private_count = sum(
1 for row in targets if row.get("private_visibility_verified") is True
)
not_found_count = sum(
1
for row in targets
if row.get("readback_status") == "not_found_or_inaccessible"
)
if _int(summary.get("private_visibility_verified_count")) != private_count:
raise ValueError(f"{label}: private visibility count must match targets")
if _int(summary.get("not_found_or_inaccessible_count")) != not_found_count:
@@ -447,19 +588,33 @@ def _require_connector_readback_consistency(payload: dict[str, Any], label: str)
"secret_values_collected",
}
enabled = sorted(flag for flag in false_flags if summary.get(flag) is not False)
enabled.extend(sorted(flag for flag in false_flags if owner_readback and owner_readback.get(flag) is not False))
enabled.extend(
sorted(
flag
for flag in false_flags
if owner_readback and owner_readback.get(flag) is not False
)
)
if enabled:
raise ValueError(f"{label}: connector readback must remain read-only: {enabled}")
raise ValueError(
f"{label}: connector readback must remain read-only: {enabled}"
)
def _require_missing_source_readiness_consistency(payload: dict[str, Any], label: str) -> None:
def _require_missing_source_readiness_consistency(
payload: dict[str, Any], label: str
) -> None:
summary = _dict(payload.get("summary"))
targets = [_dict(row) for row in _list(payload.get("targets"))]
if _int(summary.get("target_count")) != len(targets):
raise ValueError(f"{label}: summary.target_count must equal targets length")
create_ready = sum(1 for row in targets if row.get("create_private_repo_ready") is True)
create_ready = sum(
1 for row in targets if row.get("create_private_repo_ready") is True
)
refs_ready = sum(1 for row in targets if row.get("refs_sync_ready") is True)
owner_required = sum(1 for row in targets if row.get("owner_response_required") is True)
owner_required = sum(
1 for row in targets if row.get("owner_response_required") is True
)
if _int(summary.get("create_private_repo_ready_count")) != create_ready:
raise ValueError(f"{label}: create ready count must match targets")
if _int(summary.get("refs_sync_ready_count")) != refs_ready:
@@ -474,7 +629,9 @@ def _require_missing_source_readiness_consistency(payload: dict[str, Any], label
"refs_sync_performed",
"secret_values_collected",
}
enabled = sorted(flag for flag in false_summary_flags if summary.get(flag) is not False)
enabled = sorted(
flag for flag in false_summary_flags if summary.get(flag) is not False
)
boundaries = _dict(payload.get("operation_boundaries"))
false_boundary_flags = {
"github_api_write_allowed",
@@ -483,9 +640,15 @@ def _require_missing_source_readiness_consistency(payload: dict[str, Any], label
"refs_sync_allowed",
"secret_value_collection_allowed",
}
enabled.extend(sorted(flag for flag in false_boundary_flags if boundaries.get(flag) is not False))
enabled.extend(
sorted(
flag for flag in false_boundary_flags if boundaries.get(flag) is not False
)
)
if enabled:
raise ValueError(f"{label}: missing source readiness must remain read-only: {enabled}")
raise ValueError(
f"{label}: missing source readiness must remain read-only: {enabled}"
)
def _require_owner_response_boundaries(payload: dict[str, Any], label: str) -> None:
@@ -506,7 +669,170 @@ def _require_owner_response_boundaries(payload: dict[str, Any], label: str) -> N
}
enabled = sorted(flag for flag in false_flags if summary.get(flag) is not False)
if enabled:
raise ValueError(f"{label}: owner response boundary flags must remain false: {enabled}")
raise ValueError(
f"{label}: owner response boundary flags must remain false: {enabled}"
)
_require_owner_response_intake_consistency(payload, label)
def _require_owner_response_intake_consistency(
payload: dict[str, Any], label: str
) -> None:
packet = _dict(payload.get("owner_response_request_packet"))
templates = [_dict(row) for row in _list(payload.get("response_templates"))]
collection_checks = [
_dict(row) for row in _list(payload.get("owner_response_collection_checks"))
]
preflight_checks = [
_dict(row) for row in _list(payload.get("intake_preflight_checks"))
]
acceptance_checks = [_dict(row) for row in _list(payload.get("acceptance_checks"))]
summary = _dict(payload.get("summary"))
if (
packet.get("execution_authorized") is not False
or packet.get("not_approval") is not True
):
raise ValueError(
f"{label}: owner response request packet must remain non-executing"
)
requested_template_ids = _strings(packet.get("requested_template_ids"))
if _int(packet.get("required_response_item_count")) != len(requested_template_ids):
raise ValueError(
f"{label}: required response item count must match requested templates"
)
if _int(summary.get("response_template_count")) != len(templates):
raise ValueError(f"{label}: response template count must match templates")
if _int(summary.get("owner_response_collection_check_count")) != len(
collection_checks
):
raise ValueError(f"{label}: collection check count must match checks")
if _int(summary.get("intake_preflight_check_count")) != len(preflight_checks):
raise ValueError(f"{label}: intake preflight check count must match checks")
if _int(summary.get("acceptance_check_count")) != len(acceptance_checks):
raise ValueError(f"{label}: acceptance check count must match checks")
template_ids = {
str(row.get("template_id")) for row in templates if row.get("template_id")
}
missing_templates = sorted(set(requested_template_ids) - template_ids)
if missing_templates:
raise ValueError(
f"{label}: requested templates missing from response templates: {missing_templates}"
)
executable_rows = [
str(row.get("template_id") or row.get("check_id") or "unknown")
for row in [
*templates,
*collection_checks,
*preflight_checks,
*acceptance_checks,
]
if row.get("execution_authorized") is not False
]
if executable_rows:
raise ValueError(
f"{label}: owner response intake rows must remain non-executing: {executable_rows}"
)
def _owner_response_intake_readiness(owner_response: dict[str, Any]) -> dict[str, Any]:
summary = _dict(owner_response.get("summary"))
packet = _dict(owner_response.get("owner_response_request_packet"))
requested_template_ids = _strings(packet.get("requested_template_ids"))
response_templates = [
_owner_response_template_summary(row)
for row in _list(owner_response.get("response_templates"))
]
collection_checks = [
_intake_check_summary(row)
for row in _list(owner_response.get("owner_response_collection_checks"))
]
preflight_checks = [
_intake_check_summary(row)
for row in _list(owner_response.get("intake_preflight_checks"))
]
acceptance_checks = [
_intake_check_summary(row)
for row in _list(owner_response.get("acceptance_checks"))
]
request_ready = (
packet.get("execution_authorized") is False
and packet.get("not_approval") is True
and bool(requested_template_ids)
and len(requested_template_ids) == len(response_templates)
and _int(summary.get("target_owner_handoff_package_ready")) == 1
)
return {
"status": "ready_to_collect_read_only_owner_response_not_authorization"
if request_ready
else "blocked_owner_response_request_packet_incomplete",
"request_ready": request_ready,
"request_id": str(packet.get("request_id") or ""),
"display_status": str(packet.get("display_status") or ""),
"requested_packet": str(packet.get("requested_packet") or ""),
"required_response_item_count": _int(
packet.get("required_response_item_count")
),
"requested_template_count": len(requested_template_ids),
"requested_template_ids": requested_template_ids,
"response_template_count": len(response_templates),
"response_templates": response_templates,
"allowed_response_field_count": len(
_strings(packet.get("allowed_response_fields"))
),
"allowed_response_fields": _strings(packet.get("allowed_response_fields")),
"evidence_ref_rules": _strings(packet.get("evidence_ref_rules")),
"forbidden_payload_count": len(_strings(packet.get("forbidden_payloads"))),
"forbidden_payloads": _strings(packet.get("forbidden_payloads")),
"allowed_submission_modes": _strings(packet.get("allowed_submission_modes")),
"collection_check_count": len(collection_checks),
"collection_checks": collection_checks,
"intake_preflight_check_count": len(preflight_checks),
"intake_preflight_checks": preflight_checks,
"acceptance_check_count": len(acceptance_checks),
"acceptance_checks": acceptance_checks,
"still_forbidden": _strings(packet.get("still_forbidden")),
"received_response_count": _int(summary.get("received_response_count")),
"accepted_response_count": _int(summary.get("accepted_response_count")),
"rejected_response_count": _int(summary.get("rejected_response_count")),
"target_owner_request_dispatch_authorized": False,
"action_buttons_allowed": False,
"execution_authorized": False,
"not_approval": True,
}
def _owner_response_template_summary(value: Any) -> dict[str, Any]:
row = _dict(value)
return {
"template_id": str(row.get("template_id") or ""),
"github_repo": str(row.get("github_repo") or ""),
"source_key": str(row.get("source_key") or ""),
"target_state": str(row.get("target_state") or ""),
"risk": str(row.get("risk") or ""),
"requested_owner_decision": str(row.get("requested_owner_decision") or ""),
"required_owner_fields": _strings(row.get("required_owner_fields")),
"acceptable_decisions": _strings(row.get("acceptable_decisions")),
"minimum_evidence_refs": _strings(row.get("minimum_evidence_refs")),
"execution_authorized": False,
}
def _intake_check_summary(value: Any) -> dict[str, Any]:
row = _dict(value)
return {
"check_id": str(row.get("check_id") or ""),
"title": str(row.get("title") or ""),
"required": row.get("required") is True,
"pass_condition": str(row.get("pass_condition") or ""),
"failure_lane": str(row.get("failure_lane") or ""),
"awooop_display": str(row.get("awooop_display") or ""),
"execution_authorized": False,
"not_approval": row.get("not_approval") is True,
}
def _generated_at(owner_response: dict[str, Any]) -> str:

132
apps/api/tests/test_github_target_private_backup_evidence_gate.py
View File

@@ -17,23 +17,56 @@ def test_load_github_target_private_backup_evidence_gate_from_committed_snapshot
assert snapshot["schema_version"] == "github_target_private_backup_evidence_gate_v1"
assert snapshot["mode"] == "read_only_private_backup_evidence_gate"
assert snapshot["status"] == "blocked_private_visibility_and_safe_credential_evidence_required"
assert (
snapshot["status"]
== "blocked_private_visibility_and_safe_credential_evidence_required"
)
assert snapshot["summary"]["target_decision_count"] == 10
assert snapshot["summary"]["approval_required_target_count"] == 9
assert snapshot["summary"]["github_connector_owner_visible_repository_count"] == 4
assert snapshot["summary"]["github_connector_owner_visible_private_repository_count"] == 4
assert (
snapshot["summary"]["github_connector_owner_visible_private_repository_count"]
== 4
)
assert snapshot["summary"]["github_connector_readback_count"] == 9
assert snapshot["summary"]["github_connector_private_visibility_count"] == 4
assert snapshot["summary"]["github_connector_not_found_or_inaccessible_count"] == 5
assert snapshot["summary"]["github_missing_target_resolution_count"] == 5
assert snapshot["summary"]["github_missing_target_gitea_source_candidate_count"] == 3
assert snapshot["summary"]["github_missing_target_internal_remote_source_candidate_count"] == 2
assert snapshot["summary"]["github_missing_target_local_worktree_candidate_count"] == 5
assert snapshot["summary"]["github_missing_target_canonical_source_ambiguous_count"] == 5
assert snapshot["summary"]["github_missing_target_create_private_repo_ready_count"] == 0
assert (
snapshot["summary"]["github_missing_target_gitea_source_candidate_count"] == 3
)
assert (
snapshot["summary"][
"github_missing_target_internal_remote_source_candidate_count"
]
== 2
)
assert (
snapshot["summary"]["github_missing_target_local_worktree_candidate_count"] == 5
)
assert (
snapshot["summary"]["github_missing_target_canonical_source_ambiguous_count"]
== 5
)
assert (
snapshot["summary"]["github_missing_target_create_private_repo_ready_count"]
== 0
)
assert snapshot["summary"]["github_missing_target_refs_sync_ready_count"] == 0
assert snapshot["summary"]["private_backup_verified_count"] == 4
assert snapshot["summary"]["private_visibility_verified_count"] == 4
assert snapshot["summary"]["owner_response_request_ready"] is True
assert snapshot["summary"]["owner_response_required_response_item_count"] == 9
assert snapshot["summary"]["owner_response_requested_template_count"] == 9
assert snapshot["summary"]["owner_response_template_count"] == 9
assert snapshot["summary"]["owner_response_allowed_response_field_count"] == 25
assert snapshot["summary"]["owner_response_forbidden_payload_count"] == 15
assert snapshot["summary"]["owner_response_collection_check_count"] == 6
assert snapshot["summary"]["owner_response_intake_preflight_check_count"] == 6
assert snapshot["summary"]["owner_response_request_execution_authorized"] is False
assert (
snapshot["summary"]["github_target_owner_response_handoff_not_approval"] is True
)
assert snapshot["summary"]["blocked_target_count"] == 9
assert snapshot["summary"]["public_repo_allowed"] is False
assert snapshot["summary"]["repo_creation_authorized"] is False
@@ -45,6 +78,25 @@ def test_load_github_target_private_backup_evidence_gate_from_committed_snapshot
assert snapshot["operation_boundaries"]["visibility_change_allowed"] is False
assert snapshot["operation_boundaries"]["refs_sync_allowed"] is False
assert snapshot["operation_boundaries"]["secret_value_collection_allowed"] is False
intake = snapshot["owner_response_intake_readiness"]
assert (
intake["status"]
== "ready_to_collect_read_only_owner_response_not_authorization"
)
assert intake["request_ready"] is True
assert intake["required_response_item_count"] == 9
assert intake["requested_template_count"] == 9
assert intake["response_template_count"] == 9
assert intake["allowed_response_field_count"] == 25
assert intake["forbidden_payload_count"] == 15
assert intake["collection_check_count"] == 6
assert intake["intake_preflight_check_count"] == 6
assert intake["acceptance_check_count"] == 8
assert intake["execution_authorized"] is False
assert intake["not_approval"] is True
assert "private_clone_url_credential" in intake["forbidden_payloads"]
assert "read_only_markdown_response" in intake["allowed_submission_modes"]
assert "create_github_repo" in intake["still_forbidden"]
targets = {target["github_repo"]: target for target in snapshot["targets"]}
assert targets["owenhytsai/awoooi"]["visibility_evidence_status"] == (
@@ -52,6 +104,19 @@ def test_load_github_target_private_backup_evidence_gate_from_committed_snapshot
)
assert targets["owenhytsai/awoooi"]["private_backup_verified"] is True
assert targets["owenhytsai/awoooi"]["github_connector_visibility"] == "private"
assert (
targets["owenhytsai/awoooi"]["owner_response_template_id"]
== "target-awoooi-refs-blocked"
)
assert (
targets["owenhytsai/awoooi"]["owner_response_submission_status"]
== "waiting_owner_response"
)
assert targets["owenhytsai/awoooi"]["owner_response_execution_authorized"] is False
assert (
"canonical_source"
in targets["owenhytsai/awoooi"]["owner_response_required_fields"]
)
assert targets["owenhytsai/ewoooc"]["visibility_evidence_status"] == (
"blocked_private_or_absent_not_verified"
)
@@ -59,21 +124,41 @@ def test_load_github_target_private_backup_evidence_gate_from_committed_snapshot
"blocked_canonical_source_ambiguous"
)
assert targets["owenhytsai/ewoooc"]["missing_target_gitea_repo"] == "wooo/ewoooc"
assert targets["owenhytsai/ewoooc"]["missing_target_create_private_repo_ready"] is False
assert (
targets["owenhytsai/ewoooc"]["missing_target_create_private_repo_ready"]
is False
)
assert targets["owenhytsai/ewoooc"]["missing_target_refs_sync_ready"] is False
assert targets["owenhytsai/ewoooc"]["private_backup_verified"] is False
assert targets["owenhytsai/VibeWork"]["missing_target_source_resolution_status"] == (
"blocked_product_boundary_and_local_divergence"
assert (
targets["owenhytsai/ewoooc"]["owner_response_template_id"]
== "target-ewoooc-private-or-new"
)
assert (
"approve_new_target_creation_candidate"
in targets["owenhytsai/ewoooc"]["owner_response_acceptable_decisions"]
)
assert targets["owenhytsai/VibeWork"][
"missing_target_source_resolution_status"
] == ("blocked_product_boundary_and_local_divergence")
assert (
targets["owenhytsai/VibeWork"]["missing_target_gitea_repo"] == "wooo/vibework"
)
assert targets["owenhytsai/VibeWork"]["missing_target_gitea_repo"] == "wooo/vibework"
assert targets["nexu-io/open-design"]["visibility_evidence_status"] == (
"external_scope_not_backup_target"
)
assert targets["nexu-io/open-design"]["owner_response_template_id"] is None
assert (
targets["nexu-io/open-design"]["owner_response_submission_status"]
== "not_required"
)
def test_github_target_private_backup_gate_rejects_runtime_authorization(tmp_path):
_copy_security_snapshots(tmp_path)
owner_response_path = tmp_path / "github-target-owner-decision-response.snapshot.json"
owner_response_path = (
tmp_path / "github-target-owner-decision-response.snapshot.json"
)
owner_response = json.loads(owner_response_path.read_text(encoding="utf-8"))
owner_response["summary"]["repo_creation_authorized"] = True
owner_response_path.write_text(json.dumps(owner_response), encoding="utf-8")
@@ -82,7 +167,24 @@ def test_github_target_private_backup_gate_rejects_runtime_authorization(tmp_pat
load_latest_github_target_private_backup_evidence_gate(tmp_path)
def test_github_target_private_backup_gate_requires_decision_rollup_consistency(tmp_path):
def test_github_target_private_backup_gate_rejects_owner_response_execution_packet(
tmp_path,
):
_copy_security_snapshots(tmp_path)
owner_response_path = (
tmp_path / "github-target-owner-decision-response.snapshot.json"
)
owner_response = json.loads(owner_response_path.read_text(encoding="utf-8"))
owner_response["owner_response_request_packet"]["execution_authorized"] = True
owner_response_path.write_text(json.dumps(owner_response), encoding="utf-8")
with pytest.raises(ValueError, match="owner response request packet"):
load_latest_github_target_private_backup_evidence_gate(tmp_path)
def test_github_target_private_backup_gate_requires_decision_rollup_consistency(
tmp_path,
):
_copy_security_snapshots(tmp_path)
decision_path = tmp_path / "github-target-decision.snapshot.json"
decision = json.loads(decision_path.read_text(encoding="utf-8"))
@@ -100,7 +202,9 @@ def test_github_target_private_backup_gate_rejects_missing_source_write_flags(tm
readiness["summary"]["repo_creation_performed"] = True
readiness_path.write_text(json.dumps(readiness), encoding="utf-8")
with pytest.raises(ValueError, match="missing source readiness must remain read-only"):
with pytest.raises(
ValueError, match="missing source readiness must remain read-only"
):
load_latest_github_target_private_backup_evidence_gate(tmp_path)

18
apps/api/tests/test_github_target_private_backup_evidence_gate_api.py
View File

@@ -27,6 +27,13 @@ def test_github_target_private_backup_evidence_gate_endpoint_returns_read_only_g
assert data["summary"]["github_missing_target_refs_sync_ready_count"] == 0
assert data["summary"]["private_backup_verified_count"] == 4
assert data["summary"]["private_visibility_verified_count"] == 4
assert data["summary"]["owner_response_request_ready"] is True
assert data["summary"]["owner_response_required_response_item_count"] == 9
assert data["summary"]["owner_response_requested_template_count"] == 9
assert data["summary"]["owner_response_template_count"] == 9
assert data["summary"]["owner_response_collection_check_count"] == 6
assert data["summary"]["owner_response_intake_preflight_check_count"] == 6
assert data["summary"]["owner_response_request_execution_authorized"] is False
assert data["summary"]["blocked_target_count"] == 9
assert data["summary"]["public_repo_allowed"] is False
assert data["summary"]["repo_creation_authorized"] is False
@@ -40,4 +47,15 @@ def test_github_target_private_backup_evidence_gate_endpoint_returns_read_only_g
assert data["operation_boundaries"]["refs_sync_allowed"] is False
assert data["operation_boundaries"]["workflow_trigger_allowed"] is False
assert data["operation_boundaries"]["private_clone_url_collection_allowed"] is False
intake = data["owner_response_intake_readiness"]
assert (
intake["status"]
== "ready_to_collect_read_only_owner_response_not_authorization"
)
assert intake["request_ready"] is True
assert intake["requested_template_count"] == 9
assert intake["forbidden_payload_count"] == 15
assert intake["execution_authorized"] is False
assert intake["not_approval"] is True
assert data["targets"][0]["owner_response_execution_authorized"] is False
assert "192.168.0." not in response.text

30
docs/LOGBOOK.md
View File

@@ -1,3 +1,33 @@
## 2026-06-27 — 20:49 GitHub backup owner response intake readiness 本地完成
**背景**：GitHub private backup gate 已能讀回 5 個 missing target source readiness，但下一個安全步驟需要讓產品端清楚知道 owner response 可以收哪些欄位、必須拒收哪些 payload，以及每個 target 對應哪個 response template；同時不能把一般「批准 / 繼續」誤當成 repo creation、visibility change 或 refs sync 授權。
**完成內容**：
- `apps/api/src/services/github_target_private_backup_evidence_gate.py` 新增 `owner_response_intake_readiness` read-only 投影，從 committed `github-target-owner-decision-response.snapshot.json` 讀出 request packet、9 個 response templates、allowed fields、forbidden payloads、collection checks、intake preflight checks 與 acceptance checks。
- `summary` 新增 owner-response intake counters：`owner_response_request_ready=true`、required response items `9`、requested templates `9`、response templates `9`、allowed response fields `25`、forbidden payloads `15`、collection checks `6`、intake preflight checks `6`。
- 每個 target 新增 owner response template mapping：`owner_response_template_id`、`owner_response_submission_status`、required fields、acceptable decisions、minimum evidence refs、acceptance / rejection criteria 與 `owner_response_execution_authorized=false`。
- Validator 新增 owner response intake consistency guard，拒收 request packet、templates 或 checks 夾帶 `execution_authorized=true`，並檢查 template / collection / preflight / acceptance count 一致。
**本地驗證結果**：
- `python3 -m py_compile apps/api/src/services/github_target_private_backup_evidence_gate.py`：通過。
- `DATABASE_URL=sqlite:///test.db PYTHONPATH=apps/api python3.11 -m pytest apps/api/tests/test_github_target_private_backup_evidence_gate.py apps/api/tests/test_github_target_private_backup_evidence_gate_api.py apps/api/tests/test_delivery_closure_workbench_api.py -q`：`7 passed`。
- `git diff --check`：通過。
**完成度 / 邊界**：
- 本段「GitHub owner response intake readiness API 欄位」本地：`0% -> 85%`。尚待 commit、normal push、Gitea CD、production API readback。
- GitHub backup mirror governance 仍為 blocked；本段只提升 owner-response 收件與拒收規則可讀性，不代表 private backup 已全數 verified，也不代表 refs sync 或 repo creation ready。
**仍維持 0 / false**：
- `owner_response_received_count=0`、`owner_response_accepted_count=0`、`safe_credential_accepted_evidence_count=0`、`github_missing_target_create_private_repo_ready_count=0`、`github_missing_target_refs_sync_ready_count=0`、`execution_ready_count=0`。
- `github_api_write_allowed=false`、`repo_creation_allowed=false`、`visibility_change_allowed=false`、`refs_sync_allowed=false`、`workflow_trigger_allowed=false`、`private_clone_url_collection_allowed=false`、`secret_value_collection_allowed=false`。
**未做**：
- 沒有 GitHub repo creation、visibility change、refs sync、workflow trigger、private clone URL collection、secret value collection；沒有 host / Docker / systemd / Nginx / firewall / K8s / DB / Wazuh runtime 寫操作；沒有 force push。
**下一步**：
- normal push feature 與 main，等待 Gitea code-review / CD。
- 部署後讀回 `GET /api/v1/agents/github-target-private-backup-evidence-gate`，目標確認 `owner_response_request_ready=true`、requested templates `9`、forbidden payloads `15`、collection / preflight checks `6/6`，且 create/private ready 與 refs sync ready 仍維持 `0`。
## 2026-06-27｜P2-416 D1N：目前有效 AI Agent 自主化控制層與日週月報 Telegram Gateway 接線
**背景**：使用者已明確要求不再依舊 no-send / no-live / 高風險預設人工規範推進；目前有效方向是 low / medium / high 風險在 allowlist、Ansible check-mode、controlled apply、post-apply verifier、KM / PlayBook writeback 與 Telegram receipt 下由 AI Agent 受控自動處理。critical / secret / destructive / reboot / node drain / provider switch / force push 等仍維持 hard blocker。