diff --git a/apps/api/src/services/independent_verifier_registry.py b/apps/api/src/services/independent_verifier_registry.py index aa3828b2b..3eb2ee15c 100644 --- a/apps/api/src/services/independent_verifier_registry.py +++ b/apps/api/src/services/independent_verifier_registry.py @@ -132,6 +132,26 @@ _EXTERNAL_READBACKS: tuple[dict[str, Any], ...] = ( "source_ref": "apps/api/src/services/backup_restore_evidence_verifier.py", "runtime_status": "receipt_pending", }, + { + "readback_id": "stockplatform_cron_exit_code_postcondition", + "verifier": "stockplatform_cron_exit_code_independent_verifier", + "required_source": "stockplatform_cron_scheduler_readback", + "source_ref": ( + "apps/api/src/services/" + "stockplatform_cron_intelligence_sync_control.py" + ), + "runtime_status": "receipt_pending", + }, + { + "readback_id": "stockplatform_cron_freshness_postcondition", + "verifier": "stockplatform_cron_freshness_independent_verifier", + "required_source": "stockplatform_public_api_freshness_readback", + "source_ref": ( + "apps/api/src/services/" + "stockplatform_cron_intelligence_sync_control.py" + ), + "runtime_status": "receipt_pending", + }, ) diff --git a/apps/api/src/services/stockplatform_cron_intelligence_sync_control.py b/apps/api/src/services/stockplatform_cron_intelligence_sync_control.py new file mode 100644 index 000000000..fe7085ed0 --- /dev/null +++ b/apps/api/src/services/stockplatform_cron_intelligence_sync_control.py @@ -0,0 +1,943 @@ +"""StockPlatform cron_intelligence_sync repair and closure contract. + +This control-plane module accepts only public-safe durable receipts. It can +record an exact, fingerprint-deduplicated source-fix candidate, but it never +calls an AI provider, deploys code, invokes Agent99, runs a host command, or +mutates StockPlatform runtime state. Runtime closure requires an exact +controlled deployment receipt plus independent exit-code and freshness +verifier receipts from the same deployment run. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Awaitable, Callable, Mapping +from datetime import datetime +from typing import Any +from uuid import UUID + +import structlog +from sqlalchemy import text + +from src.services.audit_sink import sanitize +from src.services.service_registry import get_service_registry + +logger = structlog.get_logger(__name__) + +CANDIDATE_SCHEMA_VERSION = "stockplatform_cron_repair_candidate_v1" +HANDOFF_SCHEMA_VERSION = "stockplatform_cron_repair_handoff_v1" +CLOSURE_SCHEMA_VERSION = "stockplatform_cron_repair_closure_v1" +CANONICAL_ASSET_ID = "schedule:stockplatform-v2:cron_intelligence_sync" +TARGET_RESOURCE = "cron_intelligence_sync" +PRODUCT_ID = "stockplatform-v2" +REPOSITORY = "wooo/stockplatform-v2" +TYPED_DOMAIN = "docker_container" +TYPED_EXECUTOR = "host_ansible_executor" +ROUTE_VERIFIER = "stockplatform_cron_independent_verifier" +DEPLOY_EXECUTOR = "gitea_controlled_cd_executor" +EXIT_CODE_VERIFIER = "stockplatform_cron_exit_code_independent_verifier" +FRESHNESS_VERIFIER = "stockplatform_cron_freshness_independent_verifier" +MAX_EVIDENCE_SKEW_SECONDS = 15 * 60 + +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$") +_COMMIT_SHA = re.compile(r"^[0-9a-f]{40}$") + +PersistCandidate = Callable[[dict[str, Any], str], Awaitable[dict[str, Any]]] +PersistClosure = Callable[[dict[str, Any], str], Awaitable[dict[str, Any]]] + + +def _safe_text(value: Any) -> str: + return str(value or "").strip() + + +def _number(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, int | float): + return None + return float(value) + + +def _timestamp(receipt: Mapping[str, Any], field: str) -> datetime | None: + raw = _safe_text(receipt.get(field)) + if not raw: + return None + try: + value = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError: + return None + return value if value.tzinfo is not None else None + + +def _valid_receipt_id(value: Any) -> bool: + receipt_id = _safe_text(value) + return bool( + _SAFE_ID.fullmatch(receipt_id) + and re.fullmatch(r"[0-9a-f]{64}", receipt_id) is None + ) + + +def _valid_uuid(value: Any) -> bool: + try: + UUID(_safe_text(value)) + except (TypeError, ValueError): + return False + return True + + +def _derived_receipt_id(prefix: str, *values: str) -> str: + digest = hashlib.sha256("|".join(values).encode("utf-8")).hexdigest()[:24] + return f"{prefix}:{digest}" + + +def _public_safe_record(record: Mapping[str, Any]) -> dict[str, Any]: + safe = sanitize(dict(record)) + for field in ("fingerprint", "candidate_fingerprint"): + value = _safe_text(record.get(field)) + if re.fullmatch(r"[0-9a-f]{64}", value): + safe[field] = value + return safe + + +def _common_source_correlation( + receipts: tuple[Mapping[str, Any], ...], +) -> tuple[dict[str, str] | None, list[str]]: + first = receipts[0] + correlation = { + field: _safe_text(first.get(field)) + for field in ("trace_id", "run_id", "work_item_id") + } + blockers: list[str] = [] + for field, value in correlation.items(): + if not _SAFE_ID.fullmatch(value): + blockers.append(f"{field}_missing_or_invalid") + if not _valid_uuid(correlation["run_id"]): + blockers.append("run_id_not_uuid") + + receipt_ids: list[str] = [] + observed_at: list[datetime] = [] + for receipt in receipts: + receipt_id = _safe_text(receipt.get("receipt_id")) + if not _valid_receipt_id(receipt_id): + blockers.append("evidence_receipt_id_missing_or_invalid") + else: + receipt_ids.append(receipt_id) + for field, expected in correlation.items(): + if _safe_text(receipt.get(field)) != expected: + blockers.append(f"same_run_{field}_mismatch") + if receipt.get("durable_readback_ack") is not True: + blockers.append("evidence_not_durable") + if receipt.get("freshness_verified") is not True: + blockers.append("evidence_freshness_unverified") + max_age = _number(receipt.get("max_age_seconds")) + if max_age is None or max_age <= 0 or max_age > MAX_EVIDENCE_SKEW_SECONDS: + blockers.append("evidence_max_age_invalid") + observed = _timestamp(receipt, "observed_at") + if observed is None: + blockers.append("evidence_observed_at_missing_or_invalid") + else: + observed_at.append(observed) + + if len(receipt_ids) != len(set(receipt_ids)): + blockers.append("evidence_receipt_replayed") + if observed_at: + try: + skew = (max(observed_at) - min(observed_at)).total_seconds() + except TypeError: + blockers.append("evidence_timezone_mismatch") + else: + if skew > MAX_EVIDENCE_SKEW_SECONDS: + blockers.append("evidence_window_not_correlated") + + blockers = list(dict.fromkeys(blockers)) + return (None if blockers else correlation), blockers + + +def _candidate_fingerprint( + *, + correlation: Mapping[str, str], + receipt_ids: list[str], + commit_sha: str, + canonical_asset_id: str, +) -> str: + material = { + "schema_version": CANDIDATE_SCHEMA_VERSION, + **correlation, + "canonical_asset_id": canonical_asset_id, + "typed_domain": TYPED_DOMAIN, + "repository": REPOSITORY, + "commit_sha": commit_sha, + "evidence_receipt_ids": receipt_ids, + } + canonical = json.dumps(material, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def _blocked_result( + *, + status: str, + blockers: list[str], + next_safe_action: str, + correlation: Mapping[str, str] | None = None, +) -> dict[str, Any]: + correlation = correlation or {} + return { + "schema_version": HANDOFF_SCHEMA_VERSION, + "accepted": False, + "created": False, + "deduplicated": False, + "candidate_created": False, + "controlled_release_allowed": False, + "status": status, + "trace_id": correlation.get("trace_id"), + "run_id": correlation.get("run_id"), + "work_item_id": correlation.get("work_item_id"), + "active_blockers": blockers, + "next_safe_action": next_safe_action, + "provider_call_performed": False, + "paid_provider_call_performed": False, + "executor_invoked": False, + "agent99_dispatch_performed": False, + "runtime_mutation_performed": False, + "cross_domain_fallback_allowed": False, + } + + +def _candidate_record_matches(expected: Mapping[str, Any], stored: Any) -> bool: + if not isinstance(stored, Mapping): + return False + immutable_fields = ( + "schema_version", + "kind", + "status", + "trace_id", + "run_id", + "work_item_id", + "fingerprint", + "canonical_asset_id", + "target_resource", + "typed_domain", + "executor", + "verifier", + "evidence", + "source_fix", + "controlled_release_allowed", + "next_safe_action", + "policy", + "runtime_mutation_performed", + "cross_domain_fallback_allowed", + "source_commitment", + ) + public_expected = _public_safe_record(expected) + return all( + stored.get(field) == expected.get(field) for field in immutable_fields + ) or all( + stored.get(field) == public_expected.get(field) for field in immutable_fields + ) + + +async def persist_stockplatform_cron_candidate( + record: dict[str, Any], project_id: str +) -> dict[str, Any]: + """Persist one exact source-fix candidate or return its duplicate receipt.""" + + from src.db.base import get_db_context + + fingerprint = _safe_text(record.get("fingerprint")) + provider_event_id = f"stockplatform-cron-candidate:{fingerprint}" + safe_record = _public_safe_record(record) + envelope = json.dumps(safe_record, ensure_ascii=False, default=str) + preview = f"stockplatform_cron:{record['status']}:{record['work_item_id']}"[:256] + async with get_db_context(project_id) as db: + inserted = await db.execute( + text( + """ + INSERT INTO awooop_conversation_event ( + project_id, channel_type, provider_event_id, + run_id, content_type, content_hash, content_preview, + content_redacted, redaction_version, source_envelope, + is_duplicate, received_at + ) VALUES ( + :project_id, 'internal', :provider_event_id, + :run_id, 'command', :content_hash, :preview, + :preview, 'audit_sink_v1', CAST(:source_envelope AS jsonb), + FALSE, NOW() + ) + ON CONFLICT (project_id, channel_type, provider_event_id) + DO NOTHING + RETURNING event_id + """ + ), + { + "project_id": project_id, + "provider_event_id": provider_event_id, + "run_id": UUID(_safe_text(record["run_id"])), + "content_hash": fingerprint, + "preview": preview, + "source_envelope": envelope, + }, + ) + row = inserted.fetchone() + if row is not None: + return {"event_id": str(row[0]), "created": True, "record": safe_record} + + existing = await db.execute( + text( + """ + SELECT event_id, source_envelope + FROM awooop_conversation_event + WHERE project_id = :project_id + AND channel_type = 'internal' + AND provider_event_id = :provider_event_id + LIMIT 1 + """ + ), + {"project_id": project_id, "provider_event_id": provider_event_id}, + ) + existing_row = existing.fetchone() + if existing_row is None: + raise RuntimeError("stockplatform_cron_candidate_dedupe_receipt_missing") + stored = existing_row[1] + if isinstance(stored, str): + stored = json.loads(stored) + if not _candidate_record_matches(record, stored): + raise RuntimeError("stockplatform_cron_candidate_dedupe_receipt_mismatch") + return { + "event_id": str(existing_row[0]), + "created": False, + "record": dict(stored), + } + + +async def build_stockplatform_cron_repair_candidate( + *, + target_resource: str, + failure_receipt: Mapping[str, Any], + diagnosis_receipt: Mapping[str, Any], + source_fix_receipt: Mapping[str, Any], + project_id: str = "awoooi", + registry: Any | None = None, + persist_candidate: PersistCandidate | None = None, +) -> dict[str, Any]: + """Record a source-fix candidate after exact identity and RCA verification.""" + + target_resource = _safe_text(target_resource) + receipts = (failure_receipt, diagnosis_receipt, source_fix_receipt) + correlation, blockers = _common_source_correlation(receipts) + if correlation is None: + return _blocked_result( + status="source_evidence_unverified", + blockers=blockers, + next_safe_action="recollect_same_run_failure_rca_and_source_fix_receipts", + ) + + registry = registry or get_service_registry() + identity = registry.resolve_identity(target_resource) + identity_exact = bool( + identity.get("resolution_status") == "resolved" + and identity.get("canonical_id") == CANONICAL_ASSET_ID + and identity.get("asset_domain") == TYPED_DOMAIN + and identity.get("executor") == TYPED_EXECUTOR + and identity.get("verifier") == ROUTE_VERIFIER + and _safe_text(identity.get("host")) == "192.168.0.110" + ) + if not identity_exact: + drift_work_item_id = _safe_text(identity.get("drift_work_item_id")) + if not drift_work_item_id: + digest = hashlib.sha256(target_resource.encode("utf-8")).hexdigest()[:12] + drift_work_item_id = f"AIA-ASSET-DRIFT-{digest.upper()}" + fingerprint = _candidate_fingerprint( + correlation=correlation, + receipt_ids=[_safe_text(row.get("receipt_id")) for row in receipts], + commit_sha=_safe_text(source_fix_receipt.get("commit_sha")), + canonical_asset_id="unresolved", + ) + record = { + "schema_version": CANDIDATE_SCHEMA_VERSION, + "kind": "asset_identity_drift_work_item", + "status": "asset_identity_unresolved", + **correlation, + "work_item_id": drift_work_item_id, + "fingerprint": fingerprint, + "canonical_asset_id": "", + "target_resource": "unresolved", + "typed_domain": "unknown", + "executor": None, + "verifier": None, + "evidence": { + "target_resource_digest": hashlib.sha256( + target_resource.encode("utf-8") + ).hexdigest(), + "raw_command_recorded": False, + "raw_log_recorded": False, + }, + "source_fix": {}, + "controlled_release_allowed": False, + "next_safe_action": "repair_exact_schedule_identity_before_any_release", + "policy": { + "provider_call_allowed": False, + "agent99_dispatch_allowed": False, + "direct_executor_invocation_allowed": False, + "runtime_mutation_allowed": False, + }, + "runtime_mutation_performed": False, + "cross_domain_fallback_allowed": False, + "source_commitment": "AIA-CONV-041", + } + try: + persistence = await (persist_candidate or persist_stockplatform_cron_candidate)( + record, project_id + ) + except Exception as exc: # noqa: BLE001 - no receipt means no drift item + logger.warning( + "stockplatform_cron_drift_persistence_failed", + error_type=type(exc).__name__, + trace_id=correlation["trace_id"], + ) + return _blocked_result( + status="durable_drift_receipt_failed", + blockers=["durable_drift_receipt_failed"], + next_safe_action="restore_durable_store_before_any_release", + correlation=correlation, + ) + if not _candidate_record_matches(record, persistence.get("record")): + return _blocked_result( + status="durable_drift_receipt_mismatch", + blockers=["durable_drift_receipt_mismatch"], + next_safe_action="repair_durable_store_before_any_release", + correlation=correlation, + ) + return { + "schema_version": HANDOFF_SCHEMA_VERSION, + "accepted": True, + "created": persistence.get("created") is True, + "deduplicated": persistence.get("created") is not True, + "candidate_created": False, + "controlled_release_allowed": False, + "status": "asset_identity_unresolved", + **correlation, + "work_item_id": drift_work_item_id, + "candidate_event_id": _safe_text(persistence.get("event_id")), + "canonical_asset_id": "", + "typed_domain": "unknown", + "active_blockers": ["canonical_asset_identity_unresolved"], + "next_safe_action": "repair_exact_schedule_identity_before_any_release", + "provider_call_performed": False, + "paid_provider_call_performed": False, + "executor_invoked": False, + "agent99_dispatch_performed": False, + "runtime_mutation_performed": False, + "cross_domain_fallback_allowed": False, + } + + evidence_blockers: list[str] = [] + consecutive_failures = _number(failure_receipt.get("consecutive_failures")) + last_exit_code = _number(failure_receipt.get("last_exit_code")) + if ( + failure_receipt.get("schema_version") + != "stockplatform_cron_failure_receipt_v1" + or failure_receipt.get("source") != "stockplatform_cron_runtime_observer" + or failure_receipt.get("product_id") != PRODUCT_ID + or failure_receipt.get("job_name") != TARGET_RESOURCE + or failure_receipt.get("canonical_asset_id") != CANONICAL_ASSET_ID + or consecutive_failures is None + or consecutive_failures < 2 + or last_exit_code is None + or last_exit_code == 0 + ): + evidence_blockers.append("consecutive_cron_failure_unverified") + + if ( + diagnosis_receipt.get("schema_version") + != "stockplatform_cron_wrapped_command_rca_receipt_v1" + or diagnosis_receipt.get("source") != "sanitized_stockplatform_cron_rca" + or diagnosis_receipt.get("product_id") != PRODUCT_ID + or diagnosis_receipt.get("job_name") != TARGET_RESOURCE + or diagnosis_receipt.get("canonical_asset_id") != CANONICAL_ASSET_ID + or diagnosis_receipt.get("root_cause") + != "wrapped_command_exit_code_not_propagated" + or diagnosis_receipt.get("root_cause_verified") is not True + or diagnosis_receipt.get("sanitized") is not True + or diagnosis_receipt.get("untrusted_evidence") is not True + or diagnosis_receipt.get("raw_command_recorded") is not False + or diagnosis_receipt.get("raw_log_recorded") is not False + or not diagnosis_receipt.get("evidence_refs") + ): + evidence_blockers.append("wrapped_command_root_cause_unverified") + + commit_sha = _safe_text(source_fix_receipt.get("commit_sha")) + if ( + source_fix_receipt.get("schema_version") + != "stockplatform_cron_source_fix_receipt_v1" + or source_fix_receipt.get("source") != "stockplatform_source_change_receipt" + or source_fix_receipt.get("repository") != REPOSITORY + or source_fix_receipt.get("product_id") != PRODUCT_ID + or source_fix_receipt.get("job_name") != TARGET_RESOURCE + or source_fix_receipt.get("canonical_asset_id") != CANONICAL_ASSET_ID + or _COMMIT_SHA.fullmatch(commit_sha) is None + or source_fix_receipt.get("fix_kind") + != "wrapped_command_exit_code_propagation" + or source_fix_receipt.get("wrapped_command_fix_present") is not True + or source_fix_receipt.get("focused_tests_passed") is not True + or source_fix_receipt.get("raw_patch_recorded") is not False + ): + evidence_blockers.append("source_fix_receipt_unverified") + + evidence_blockers = list(dict.fromkeys(evidence_blockers)) + if evidence_blockers: + return _blocked_result( + status="source_fix_candidate_blocked", + blockers=evidence_blockers, + next_safe_action="repair_exact_wrapped_command_source_receipt", + correlation=correlation, + ) + + pushed = source_fix_receipt.get("pushed_to_gitea") is True + reachable = source_fix_receipt.get("commit_reachable_from_gitea_main") is True + status = ( + "source_fix_candidate_ready_for_controlled_release" + if pushed and reachable + else "source_fix_candidate_recorded_release_pending" + ) + next_safe_action = ( + "controlled_release_lane_consumes_exact_commit_then_emits_deploy_receipt" + if pushed and reachable + else "publish_exact_source_commit_before_controlled_release" + ) + receipt_ids = [_safe_text(row.get("receipt_id")) for row in receipts] + fingerprint = _candidate_fingerprint( + correlation=correlation, + receipt_ids=receipt_ids, + commit_sha=commit_sha, + canonical_asset_id=CANONICAL_ASSET_ID, + ) + record = { + "schema_version": CANDIDATE_SCHEMA_VERSION, + "kind": "stockplatform_cron_source_fix_candidate", + "status": status, + **correlation, + "fingerprint": fingerprint, + "canonical_asset_id": CANONICAL_ASSET_ID, + "target_resource": TARGET_RESOURCE, + "typed_domain": TYPED_DOMAIN, + "executor": TYPED_EXECUTOR, + "verifier": ROUTE_VERIFIER, + "evidence": { + "failure_receipt_id": receipt_ids[0], + "diagnosis_receipt_id": receipt_ids[1], + "source_fix_receipt_id": receipt_ids[2], + "consecutive_failures": failure_receipt.get("consecutive_failures"), + "last_exit_code": failure_receipt.get("last_exit_code"), + "root_cause": diagnosis_receipt.get("root_cause"), + "raw_command_recorded": False, + "raw_log_recorded": False, + }, + "source_fix": { + "repository": REPOSITORY, + "commit_sha": commit_sha, + "fix_kind": source_fix_receipt.get("fix_kind"), + "focused_tests_passed": True, + "pushed_to_gitea": pushed, + "commit_reachable_from_gitea_main": reachable, + "runtime_verified": False, + }, + "controlled_release_allowed": False, + "next_safe_action": next_safe_action, + "policy": { + "provider_call_allowed": False, + "paid_provider_call_allowed": False, + "agent99_dispatch_allowed": False, + "direct_executor_invocation_allowed": False, + "runtime_mutation_allowed": False, + "controlled_release_must_use_exact_commit": True, + "runtime_closure_requires_same_run_dual_verifiers": True, + }, + "runtime_mutation_performed": False, + "cross_domain_fallback_allowed": False, + "source_commitment": "AIA-CONV-041", + } + try: + persistence = await (persist_candidate or persist_stockplatform_cron_candidate)( + record, project_id + ) + except Exception as exc: # noqa: BLE001 - candidate must be durable + logger.warning( + "stockplatform_cron_candidate_persistence_failed", + error_type=type(exc).__name__, + trace_id=correlation["trace_id"], + ) + return _blocked_result( + status="durable_candidate_receipt_failed", + blockers=["durable_candidate_receipt_failed"], + next_safe_action="restore_durable_store_before_controlled_release", + correlation=correlation, + ) + if not _candidate_record_matches(record, persistence.get("record")): + return _blocked_result( + status="durable_candidate_receipt_mismatch", + blockers=["durable_candidate_receipt_mismatch"], + next_safe_action="repair_durable_store_before_controlled_release", + correlation=correlation, + ) + + return { + "schema_version": HANDOFF_SCHEMA_VERSION, + "accepted": True, + "created": persistence.get("created") is True, + "deduplicated": persistence.get("created") is not True, + "candidate_created": True, + "controlled_release_allowed": False, + "status": status, + **correlation, + "candidate_event_id": _safe_text(persistence.get("event_id")), + "fingerprint": fingerprint, + "canonical_asset_id": CANONICAL_ASSET_ID, + "typed_domain": TYPED_DOMAIN, + "executor": TYPED_EXECUTOR, + "verifier": ROUTE_VERIFIER, + "commit_sha": commit_sha, + "active_blockers": ( + [] if pushed and reachable else ["source_fix_not_reachable_from_gitea_main"] + ), + "next_safe_action": next_safe_action, + "provider_call_performed": False, + "paid_provider_call_performed": False, + "executor_invoked": False, + "agent99_dispatch_performed": False, + "runtime_mutation_performed": False, + "cross_domain_fallback_allowed": False, + } + + +def _closure_blockers( + candidate: Mapping[str, Any], + deployment_receipt: Mapping[str, Any], + exit_code_receipt: Mapping[str, Any], + freshness_receipt: Mapping[str, Any], +) -> list[str]: + blockers: list[str] = [] + fingerprint = _safe_text(candidate.get("fingerprint")) + source_fix = candidate.get("source_fix") + source_fix = source_fix if isinstance(source_fix, Mapping) else {} + commit_sha = _safe_text(source_fix.get("commit_sha")) + if ( + candidate.get("schema_version") != CANDIDATE_SCHEMA_VERSION + or candidate.get("kind") != "stockplatform_cron_source_fix_candidate" + or re.fullmatch(r"[0-9a-f]{64}", fingerprint) is None + or candidate.get("canonical_asset_id") != CANONICAL_ASSET_ID + or candidate.get("target_resource") != TARGET_RESOURCE + or candidate.get("typed_domain") != TYPED_DOMAIN + or candidate.get("executor") != TYPED_EXECUTOR + or candidate.get("verifier") != ROUTE_VERIFIER + or _COMMIT_SHA.fullmatch(commit_sha) is None + or candidate.get("runtime_mutation_performed") is not False + or candidate.get("cross_domain_fallback_allowed") is not False + ): + blockers.append("candidate_contract_invalid") + + deployment_id = _safe_text(deployment_receipt.get("receipt_id")) + execution_run_id = _safe_text(deployment_receipt.get("execution_run_id")) + deployed_at = _timestamp(deployment_receipt, "deployed_at") + if ( + deployment_receipt.get("schema_version") + != "stockplatform_cron_deployment_receipt_v1" + or not _valid_receipt_id(deployment_id) + or not _valid_uuid(execution_run_id) + or deployment_receipt.get("source") + != "gitea_controlled_cd_release_readback" + or deployment_receipt.get("executor") != DEPLOY_EXECUTOR + or deployment_receipt.get("status") != "deployed" + or deployment_receipt.get("durable_readback_ack") is not True + or deployment_receipt.get("production_deploy") is not True + or deployment_receipt.get("repository") != REPOSITORY + or deployment_receipt.get("commit_sha") != commit_sha + or deployment_receipt.get("canonical_asset_id") != CANONICAL_ASSET_ID + or deployment_receipt.get("job_name") != TARGET_RESOURCE + or deployment_receipt.get("candidate_fingerprint") != fingerprint + or deployed_at is None + ): + blockers.append("exact_controlled_deployment_receipt_unverified") + + exit_id = _safe_text(exit_code_receipt.get("receipt_id")) + exit_observed = _timestamp(exit_code_receipt, "observed_at") + invocation_count = _number(exit_code_receipt.get("invocation_count")) + exit_max_age = _number(exit_code_receipt.get("max_age_seconds")) + if ( + exit_code_receipt.get("schema_version") + != "stockplatform_cron_exit_code_verifier_receipt_v1" + or not _valid_receipt_id(exit_id) + or exit_code_receipt.get("verifier") != EXIT_CODE_VERIFIER + or exit_code_receipt.get("source") + != "stockplatform_cron_scheduler_readback" + or exit_code_receipt.get("status") != "verified" + or exit_code_receipt.get("durable_readback_ack") is not True + or exit_code_receipt.get("freshness_verified") is not True + or exit_max_age is None + or exit_max_age <= 0 + or exit_max_age > MAX_EVIDENCE_SKEW_SECONDS + or exit_code_receipt.get("execution_run_id") != execution_run_id + or exit_code_receipt.get("candidate_fingerprint") != fingerprint + or exit_code_receipt.get("deployment_receipt_id") != deployment_id + or exit_code_receipt.get("canonical_asset_id") != CANONICAL_ASSET_ID + or exit_code_receipt.get("job_name") != TARGET_RESOURCE + or exit_code_receipt.get("process_exit_code") != 0 + or exit_code_receipt.get("wrapped_command_exit_code_propagated") is not True + or invocation_count is None + or invocation_count < 1 + or exit_code_receipt.get("active_blockers") not in ([], ()) + or exit_code_receipt.get("runtime_mutation_performed") is not False + or exit_observed is None + or deployed_at is None + or exit_observed <= deployed_at + ): + blockers.append("exit_code_independent_verifier_not_closed") + + freshness_id = _safe_text(freshness_receipt.get("receipt_id")) + freshness_observed = _timestamp(freshness_receipt, "observed_at") + latest_success = _timestamp(freshness_receipt, "latest_success_at") + age_seconds = _number(freshness_receipt.get("age_seconds")) + slo_seconds = _number(freshness_receipt.get("freshness_slo_seconds")) + freshness_max_age = _number(freshness_receipt.get("max_age_seconds")) + if ( + freshness_receipt.get("schema_version") + != "stockplatform_cron_freshness_verifier_receipt_v1" + or not _valid_receipt_id(freshness_id) + or freshness_receipt.get("verifier") != FRESHNESS_VERIFIER + or freshness_receipt.get("source") + != "stockplatform_public_api_freshness_readback" + or freshness_receipt.get("status") != "verified" + or freshness_receipt.get("durable_readback_ack") is not True + or freshness_receipt.get("freshness_verified") is not True + or freshness_max_age is None + or freshness_max_age <= 0 + or freshness_max_age > MAX_EVIDENCE_SKEW_SECONDS + or freshness_receipt.get("execution_run_id") != execution_run_id + or freshness_receipt.get("candidate_fingerprint") != fingerprint + or freshness_receipt.get("deployment_receipt_id") != deployment_id + or freshness_receipt.get("canonical_asset_id") != CANONICAL_ASSET_ID + or freshness_receipt.get("job_name") != TARGET_RESOURCE + or freshness_receipt.get("active_blockers") not in ([], ()) + or freshness_receipt.get("runtime_mutation_performed") is not False + or freshness_observed is None + or latest_success is None + or deployed_at is None + or latest_success <= deployed_at + or freshness_observed < latest_success + or age_seconds is None + or slo_seconds is None + or age_seconds < 0 + or slo_seconds <= 0 + or age_seconds > slo_seconds + ): + blockers.append("freshness_independent_verifier_not_closed") + + if len({deployment_id, exit_id, freshness_id}) != 3: + blockers.append("closure_receipt_replayed") + if exit_observed is not None and freshness_observed is not None: + try: + verifier_skew = abs( + (freshness_observed - exit_observed).total_seconds() + ) + except TypeError: + blockers.append("verifier_timezone_mismatch") + else: + if verifier_skew > MAX_EVIDENCE_SKEW_SECONDS: + blockers.append("same_run_verifier_window_not_correlated") + if DEPLOY_EXECUTOR in {EXIT_CODE_VERIFIER, FRESHNESS_VERIFIER}: + blockers.append("executor_self_verification_forbidden") + return list(dict.fromkeys(blockers)) + + +def build_stockplatform_cron_closure( + *, + candidate: Mapping[str, Any], + deployment_receipt: Mapping[str, Any], + exit_code_receipt: Mapping[str, Any], + freshness_receipt: Mapping[str, Any], +) -> dict[str, Any]: + """Build a fail-closed dual-verifier runtime closure candidate.""" + + blockers = _closure_blockers( + candidate, deployment_receipt, exit_code_receipt, freshness_receipt + ) + closed = not blockers + fingerprint = _safe_text(candidate.get("fingerprint")) + execution_run_id = _safe_text(deployment_receipt.get("execution_run_id")) + receipt_id = _derived_receipt_id( + "stockplatform-cron-closure", + fingerprint, + execution_run_id, + _safe_text(deployment_receipt.get("receipt_id")), + _safe_text(exit_code_receipt.get("receipt_id")), + _safe_text(freshness_receipt.get("receipt_id")), + ) + return { + "schema_version": CLOSURE_SCHEMA_VERSION, + "receipt_id": receipt_id, + "status": "runtime_closed" if closed else "blocked_waiting_dual_verifiers", + "runtime_closed": closed, + "durable_closure_written": False, + "candidate_fingerprint": fingerprint, + "trace_id": candidate.get("trace_id"), + "source_run_id": candidate.get("run_id"), + "execution_run_id": execution_run_id or None, + "work_item_id": candidate.get("work_item_id"), + "canonical_asset_id": CANONICAL_ASSET_ID, + "typed_domain": TYPED_DOMAIN, + "deployment_receipt_id": deployment_receipt.get("receipt_id"), + "exit_code_verifier_receipt_id": exit_code_receipt.get("receipt_id"), + "freshness_verifier_receipt_id": freshness_receipt.get("receipt_id"), + "active_blockers": blockers, + "next_safe_action": ( + "persist_durable_runtime_closure_receipt" + if closed + else "collect_exact_same_run_deploy_exit_code_and_freshness_receipts" + ), + "executor": DEPLOY_EXECUTOR, + "independent_verifiers": [EXIT_CODE_VERIFIER, FRESHNESS_VERIFIER], + "provider_call_performed": False, + "agent99_dispatch_performed": False, + "runtime_mutation_performed_by_verifier": False, + "cross_domain_fallback_allowed": False, + "source_commitment": "AIA-CONV-041", + } + + +async def persist_stockplatform_cron_closure( + record: dict[str, Any], project_id: str +) -> dict[str, Any]: + """Persist one verified closure receipt without executing runtime work.""" + + from src.db.base import get_db_context + + receipt_id = _safe_text(record.get("receipt_id")) + provider_event_id = f"stockplatform-cron-closure:{receipt_id}" + safe_record = _public_safe_record(record) + envelope = json.dumps(safe_record, ensure_ascii=False, default=str) + preview = f"stockplatform_cron:runtime_closed:{record['work_item_id']}"[:256] + async with get_db_context(project_id) as db: + inserted = await db.execute( + text( + """ + INSERT INTO awooop_conversation_event ( + project_id, channel_type, provider_event_id, + run_id, content_type, content_hash, content_preview, + content_redacted, redaction_version, source_envelope, + is_duplicate, received_at + ) VALUES ( + :project_id, 'internal', :provider_event_id, + :run_id, 'receipt', :content_hash, :preview, + :preview, 'audit_sink_v1', CAST(:source_envelope AS jsonb), + FALSE, NOW() + ) + ON CONFLICT (project_id, channel_type, provider_event_id) + DO NOTHING + RETURNING event_id + """ + ), + { + "project_id": project_id, + "provider_event_id": provider_event_id, + "run_id": UUID(_safe_text(record["execution_run_id"])), + "content_hash": hashlib.sha256(envelope.encode("utf-8")).hexdigest(), + "preview": preview, + "source_envelope": envelope, + }, + ) + row = inserted.fetchone() + if row is not None: + return {"event_id": str(row[0]), "created": True, "record": safe_record} + + existing = await db.execute( + text( + """ + SELECT event_id, source_envelope + FROM awooop_conversation_event + WHERE project_id = :project_id + AND channel_type = 'internal' + AND provider_event_id = :provider_event_id + LIMIT 1 + """ + ), + {"project_id": project_id, "provider_event_id": provider_event_id}, + ) + existing_row = existing.fetchone() + if existing_row is None: + raise RuntimeError("stockplatform_cron_closure_dedupe_receipt_missing") + stored = existing_row[1] + if isinstance(stored, str): + stored = json.loads(stored) + if ( + not isinstance(stored, Mapping) + or stored.get("schema_version") != CLOSURE_SCHEMA_VERSION + or stored.get("receipt_id") != receipt_id + or stored.get("candidate_fingerprint") + != record.get("candidate_fingerprint") + ): + raise RuntimeError("stockplatform_cron_closure_dedupe_receipt_mismatch") + return { + "event_id": str(existing_row[0]), + "created": False, + "record": dict(stored), + } + + +async def record_stockplatform_cron_closure( + *, + candidate: Mapping[str, Any], + deployment_receipt: Mapping[str, Any], + exit_code_receipt: Mapping[str, Any], + freshness_receipt: Mapping[str, Any], + project_id: str = "awoooi", + persist_closure: PersistClosure | None = None, +) -> dict[str, Any]: + """Persist closure only after both independent verifiers close.""" + + closure = build_stockplatform_cron_closure( + candidate=candidate, + deployment_receipt=deployment_receipt, + exit_code_receipt=exit_code_receipt, + freshness_receipt=freshness_receipt, + ) + if closure["runtime_closed"] is not True: + return closure + try: + persistence = await (persist_closure or persist_stockplatform_cron_closure)( + closure, project_id + ) + except Exception as exc: # noqa: BLE001 - no durable receipt means no closure + logger.warning( + "stockplatform_cron_closure_persistence_failed", + error_type=type(exc).__name__, + candidate_fingerprint=closure["candidate_fingerprint"], + ) + return { + **closure, + "status": "blocked_closure_receipt_not_durable", + "runtime_closed": False, + "active_blockers": ["durable_closure_receipt_failed"], + "next_safe_action": "restore_durable_store_then_retry_same_receipts", + } + stored = persistence.get("record") + if ( + not isinstance(stored, Mapping) + or stored.get("receipt_id") != closure["receipt_id"] + or stored.get("candidate_fingerprint") != closure["candidate_fingerprint"] + or stored.get("execution_run_id") != closure["execution_run_id"] + or stored.get("runtime_closed") is not True + ): + return { + **closure, + "status": "blocked_closure_receipt_mismatch", + "runtime_closed": False, + "active_blockers": ["durable_closure_receipt_mismatch"], + "next_safe_action": "repair_durable_store_then_retry_same_receipts", + } + return { + **closure, + "durable_closure_written": True, + "created": persistence.get("created") is True, + "deduplicated": persistence.get("created") is not True, + "closure_event_id": _safe_text(persistence.get("event_id")), + "next_safe_action": "emit_destination_bound_recovery_receipt", + } diff --git a/apps/api/tests/test_independent_verifier_registry.py b/apps/api/tests/test_independent_verifier_registry.py index 9276bd6a2..9285f45a3 100644 --- a/apps/api/tests/test_independent_verifier_registry.py +++ b/apps/api/tests/test_independent_verifier_registry.py @@ -71,6 +71,21 @@ def test_registry_covers_every_typed_domain_without_false_runtime_closure() -> N assert backup_gate["required_source"] == ( "same_run_backup_freshness_escrow_restore_metadata" ) + stock_cron = { + row["readback_id"]: row + for row in registry["external_readbacks"] + if row["readback_id"].startswith("stockplatform_cron_") + } + assert set(stock_cron) == { + "stockplatform_cron_exit_code_postcondition", + "stockplatform_cron_freshness_postcondition", + } + assert stock_cron["stockplatform_cron_exit_code_postcondition"][ + "required_source" + ] == "stockplatform_cron_scheduler_readback" + assert stock_cron["stockplatform_cron_freshness_postcondition"][ + "required_source" + ] == "stockplatform_public_api_freshness_readback" def test_backup_readback_route_matches_independent_verifier_contract() -> None: diff --git a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py index d98230310..f68330819 100644 --- a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py +++ b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py @@ -155,9 +155,9 @@ def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() - "active_or_completed_commitments": 68, "by_status": { "analysis_or_governance_complete": 6, - "source_implemented_runtime_pending": 33, + "source_implemented_runtime_pending": 34, "in_progress": 27, - "not_started_or_no_current_evidence": 2, + "not_started_or_no_current_evidence": 1, "superseded": 2, }, "product_runtime_closed_commitments": 0, @@ -200,6 +200,12 @@ def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() - assert "DR scorecard remains blocked" in " ".join( commitments["AIA-CONV-035"]["source_evidence"] ) + assert commitments["AIA-CONV-041"]["status"] == ( + "source_implemented_runtime_pending" + ) + assert "independent exit-code and freshness receipts" in " ".join( + commitments["AIA-CONV-041"]["source_evidence"] + ) assert "Host112" in commitments["AIA-CONV-049"]["title"] assert commitments["AIA-CONV-049"]["status"] == ( "source_implemented_runtime_pending" diff --git a/apps/api/tests/test_stockplatform_cron_intelligence_sync_control.py b/apps/api/tests/test_stockplatform_cron_intelligence_sync_control.py new file mode 100644 index 000000000..19ec65274 --- /dev/null +++ b/apps/api/tests/test_stockplatform_cron_intelligence_sync_control.py @@ -0,0 +1,396 @@ +from __future__ import annotations + +# ruff: noqa: E402 +import os +from pathlib import Path + +import pytest + +os.environ.setdefault( + "DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test" +) + +from src.services.service_registry import ServiceRegistryClient # noqa: E402 +from src.services.stockplatform_cron_intelligence_sync_control import ( # noqa: E402 + CANONICAL_ASSET_ID, + EXIT_CODE_VERIFIER, + FRESHNESS_VERIFIER, + build_stockplatform_cron_closure, + build_stockplatform_cron_repair_candidate, + record_stockplatform_cron_closure, +) + +ROOT = Path(__file__).resolve().parents[3] +REGISTRY_PATH = ROOT / "ops" / "config" / "service-registry.yaml" +SOURCE_RUN_ID = "6a81ca92-8cc7-4af7-b84c-d988ecb4552b" +EXECUTION_RUN_ID = "786defde-85fc-49ca-97b6-6456ca8ed69d" +TRACE_ID = "stockplatform-cron-trace-041" +WORK_ITEM_ID = "AIA-STOCK-CRON-041" +LOCAL_UNPUSHED_SHA = "af9a47379a2e26198b68fff0b8b423417b3e9a70" + + +class _CandidateStore: + def __init__(self) -> None: + self.records: dict[str, dict] = {} + + async def __call__(self, record: dict, project_id: str) -> dict: + assert project_id == "awoooi" + fingerprint = record["fingerprint"] + existing = self.records.get(fingerprint) + if existing is not None: + return { + "event_id": f"candidate:{fingerprint[:12]}", + "created": False, + "record": existing, + } + self.records[fingerprint] = record + return { + "event_id": f"candidate:{fingerprint[:12]}", + "created": True, + "record": record, + } + + +class _ClosureStore: + def __init__(self) -> None: + self.records: dict[str, dict] = {} + + async def __call__(self, record: dict, project_id: str) -> dict: + assert project_id == "awoooi" + receipt_id = record["receipt_id"] + existing = self.records.get(receipt_id) + if existing is not None: + return { + "event_id": f"closure:{receipt_id}", + "created": False, + "record": existing, + } + self.records[receipt_id] = record + return { + "event_id": f"closure:{receipt_id}", + "created": True, + "record": record, + } + + +def _base(receipt_id: str) -> dict: + return { + "receipt_id": receipt_id, + "trace_id": TRACE_ID, + "run_id": SOURCE_RUN_ID, + "work_item_id": WORK_ITEM_ID, + "durable_readback_ack": True, + "freshness_verified": True, + "max_age_seconds": 300, + "observed_at": "2026-07-22T05:00:00+00:00", + } + + +def _failure() -> dict: + return { + **_base("stock-cron-failure:041"), + "schema_version": "stockplatform_cron_failure_receipt_v1", + "source": "stockplatform_cron_runtime_observer", + "product_id": "stockplatform-v2", + "job_name": "cron_intelligence_sync", + "canonical_asset_id": CANONICAL_ASSET_ID, + "consecutive_failures": 4, + "last_exit_code": 1, + } + + +def _diagnosis() -> dict: + return { + **_base("stock-cron-rca:041"), + "schema_version": "stockplatform_cron_wrapped_command_rca_receipt_v1", + "source": "sanitized_stockplatform_cron_rca", + "product_id": "stockplatform-v2", + "job_name": "cron_intelligence_sync", + "canonical_asset_id": CANONICAL_ASSET_ID, + "root_cause": "wrapped_command_exit_code_not_propagated", + "root_cause_verified": True, + "sanitized": True, + "untrusted_evidence": True, + "raw_command_recorded": False, + "raw_log_recorded": False, + "evidence_refs": ["cron-run:041", "wrapped-command:041"], + } + + +def _source_fix(*, pushed: bool = False, reachable: bool = False) -> dict: + return { + **_base("stock-cron-source-fix:041"), + "schema_version": "stockplatform_cron_source_fix_receipt_v1", + "source": "stockplatform_source_change_receipt", + "repository": "wooo/stockplatform-v2", + "product_id": "stockplatform-v2", + "job_name": "cron_intelligence_sync", + "canonical_asset_id": CANONICAL_ASSET_ID, + "commit_sha": LOCAL_UNPUSHED_SHA, + "fix_kind": "wrapped_command_exit_code_propagation", + "wrapped_command_fix_present": True, + "focused_tests_passed": True, + "raw_patch_recorded": False, + "pushed_to_gitea": pushed, + "commit_reachable_from_gitea_main": reachable, + } + + +async def _candidate(store: _CandidateStore) -> tuple[dict, dict]: + result = await build_stockplatform_cron_repair_candidate( + target_resource="cron_intelligence_sync", + failure_receipt=_failure(), + diagnosis_receipt=_diagnosis(), + source_fix_receipt=_source_fix(), + registry=ServiceRegistryClient(REGISTRY_PATH), + persist_candidate=store, + ) + return result, store.records[result["fingerprint"]] + + +def _deployment(candidate: dict) -> dict: + return { + "schema_version": "stockplatform_cron_deployment_receipt_v1", + "receipt_id": "stock-cron-deploy:041", + "source": "gitea_controlled_cd_release_readback", + "executor": "gitea_controlled_cd_executor", + "status": "deployed", + "durable_readback_ack": True, + "production_deploy": True, + "repository": "wooo/stockplatform-v2", + "commit_sha": LOCAL_UNPUSHED_SHA, + "canonical_asset_id": CANONICAL_ASSET_ID, + "job_name": "cron_intelligence_sync", + "candidate_fingerprint": candidate["fingerprint"], + "execution_run_id": EXECUTION_RUN_ID, + "deployed_at": "2026-07-22T06:00:00+00:00", + } + + +def _exit_code(candidate: dict, deployment: dict) -> dict: + return { + "schema_version": "stockplatform_cron_exit_code_verifier_receipt_v1", + "receipt_id": "stock-cron-exit:041", + "verifier": EXIT_CODE_VERIFIER, + "source": "stockplatform_cron_scheduler_readback", + "status": "verified", + "durable_readback_ack": True, + "freshness_verified": True, + "max_age_seconds": 300, + "execution_run_id": EXECUTION_RUN_ID, + "candidate_fingerprint": candidate["fingerprint"], + "deployment_receipt_id": deployment["receipt_id"], + "canonical_asset_id": CANONICAL_ASSET_ID, + "job_name": "cron_intelligence_sync", + "process_exit_code": 0, + "wrapped_command_exit_code_propagated": True, + "invocation_count": 1, + "active_blockers": [], + "runtime_mutation_performed": False, + "observed_at": "2026-07-22T06:05:00+00:00", + } + + +def _freshness(candidate: dict, deployment: dict) -> dict: + return { + "schema_version": "stockplatform_cron_freshness_verifier_receipt_v1", + "receipt_id": "stock-cron-freshness:041", + "verifier": FRESHNESS_VERIFIER, + "source": "stockplatform_public_api_freshness_readback", + "status": "verified", + "durable_readback_ack": True, + "freshness_verified": True, + "max_age_seconds": 300, + "execution_run_id": EXECUTION_RUN_ID, + "candidate_fingerprint": candidate["fingerprint"], + "deployment_receipt_id": deployment["receipt_id"], + "canonical_asset_id": CANONICAL_ASSET_ID, + "job_name": "cron_intelligence_sync", + "latest_success_at": "2026-07-22T06:07:00+00:00", + "observed_at": "2026-07-22T06:08:00+00:00", + "age_seconds": 60, + "freshness_slo_seconds": 900, + "active_blockers": [], + "runtime_mutation_performed": False, + } + + +def test_registry_resolves_only_the_exact_stockplatform_schedule_identity() -> None: + registry = ServiceRegistryClient(REGISTRY_PATH) + exact = registry.resolve_identity("cron_intelligence_sync") + unknown = registry.resolve_identity("cron_intelligence_sync_guess") + + assert exact["canonical_id"] == CANONICAL_ASSET_ID + assert exact["asset_domain"] == "docker_container" + assert exact["host"] == "192.168.0.110" + assert exact["executor"] == "host_ansible_executor" + assert exact["verifier"] == "stockplatform_cron_independent_verifier" + assert exact["controlled_apply_allowed"] is False + assert unknown["resolution_status"] == "asset_identity_unresolved" + assert unknown["controlled_apply_allowed"] is False + + +@pytest.mark.asyncio +async def test_local_unpushed_fix_creates_one_durable_release_pending_candidate() -> None: + store = _CandidateStore() + first, record = await _candidate(store) + second, _ = await _candidate(store) + + assert first["status"] == "source_fix_candidate_recorded_release_pending" + assert first["created"] is True + assert first["candidate_created"] is True + assert first["controlled_release_allowed"] is False + assert first["active_blockers"] == [ + "source_fix_not_reachable_from_gitea_main" + ] + assert first["executor_invoked"] is False + assert first["agent99_dispatch_performed"] is False + assert first["provider_call_performed"] is False + assert first["runtime_mutation_performed"] is False + assert second["deduplicated"] is True + assert second["fingerprint"] == first["fingerprint"] + assert len(store.records) == 1 + assert record["source_fix"] == { + "repository": "wooo/stockplatform-v2", + "commit_sha": LOCAL_UNPUSHED_SHA, + "fix_kind": "wrapped_command_exit_code_propagation", + "focused_tests_passed": True, + "pushed_to_gitea": False, + "commit_reachable_from_gitea_main": False, + "runtime_verified": False, + } + assert record["policy"]["direct_executor_invocation_allowed"] is False + + +@pytest.mark.asyncio +async def test_unknown_schedule_creates_only_durable_identity_drift_item() -> None: + store = _CandidateStore() + result = await build_stockplatform_cron_repair_candidate( + target_resource="unknown-stock-cron", + failure_receipt=_failure(), + diagnosis_receipt=_diagnosis(), + source_fix_receipt=_source_fix(), + registry=ServiceRegistryClient(REGISTRY_PATH), + persist_candidate=store, + ) + + assert result["status"] == "asset_identity_unresolved" + assert result["candidate_created"] is False + assert result["controlled_release_allowed"] is False + assert result["typed_domain"] == "unknown" + assert result["work_item_id"].startswith("AIA-ASSET-DRIFT-") + assert result["active_blockers"] == ["canonical_asset_identity_unresolved"] + assert result["executor_invoked"] is False + assert result["cross_domain_fallback_allowed"] is False + assert len(store.records) == 1 + record = next(iter(store.records.values())) + assert record["kind"] == "asset_identity_drift_work_item" + assert record["source_fix"] == {} + + +@pytest.mark.asyncio +async def test_unverified_or_raw_wrapped_command_evidence_never_persists_candidate() -> None: + store = _CandidateStore() + diagnosis = _diagnosis() + diagnosis["raw_command_recorded"] = True + diagnosis["root_cause_verified"] = False + + result = await build_stockplatform_cron_repair_candidate( + target_resource="cron_intelligence_sync", + failure_receipt=_failure(), + diagnosis_receipt=diagnosis, + source_fix_receipt=_source_fix(), + registry=ServiceRegistryClient(REGISTRY_PATH), + persist_candidate=store, + ) + + assert result["status"] == "source_fix_candidate_blocked" + assert result["active_blockers"] == [ + "wrapped_command_root_cause_unverified" + ] + assert result["candidate_created"] is False + assert result["runtime_mutation_performed"] is False + assert store.records == {} + + +@pytest.mark.asyncio +async def test_closure_fails_closed_on_missing_mismatched_or_stale_verifier() -> None: + _, candidate = await _candidate(_CandidateStore()) + deployment = _deployment(candidate) + exit_code = _exit_code(candidate, deployment) + freshness = _freshness(candidate, deployment) + + mismatched_exit = {**exit_code, "execution_run_id": SOURCE_RUN_ID} + result = build_stockplatform_cron_closure( + candidate=candidate, + deployment_receipt=deployment, + exit_code_receipt=mismatched_exit, + freshness_receipt=freshness, + ) + assert result["runtime_closed"] is False + assert "exit_code_independent_verifier_not_closed" in result["active_blockers"] + + stale_freshness = { + **freshness, + "latest_success_at": "2026-07-22T05:59:00+00:00", + } + result = build_stockplatform_cron_closure( + candidate=candidate, + deployment_receipt=deployment, + exit_code_receipt=exit_code, + freshness_receipt=stale_freshness, + ) + assert result["runtime_closed"] is False + assert "freshness_independent_verifier_not_closed" in result["active_blockers"] + + wrong_commit = {**deployment, "commit_sha": "0" * 40} + result = build_stockplatform_cron_closure( + candidate=candidate, + deployment_receipt=wrong_commit, + exit_code_receipt=exit_code, + freshness_receipt=freshness, + ) + assert result["runtime_closed"] is False + assert "exact_controlled_deployment_receipt_unverified" in result[ + "active_blockers" + ] + assert result["durable_closure_written"] is False + + +@pytest.mark.asyncio +async def test_same_run_dual_verifiers_create_one_durable_closure_receipt() -> None: + _, candidate = await _candidate(_CandidateStore()) + deployment = _deployment(candidate) + exit_code = _exit_code(candidate, deployment) + freshness = _freshness(candidate, deployment) + store = _ClosureStore() + + first = await record_stockplatform_cron_closure( + candidate=candidate, + deployment_receipt=deployment, + exit_code_receipt=exit_code, + freshness_receipt=freshness, + persist_closure=store, + ) + second = await record_stockplatform_cron_closure( + candidate=candidate, + deployment_receipt=deployment, + exit_code_receipt=exit_code, + freshness_receipt=freshness, + persist_closure=store, + ) + + assert first["status"] == "runtime_closed" + assert first["runtime_closed"] is True + assert first["durable_closure_written"] is True + assert first["created"] is True + assert first["active_blockers"] == [] + assert first["independent_verifiers"] == [ + EXIT_CODE_VERIFIER, + FRESHNESS_VERIFIER, + ] + assert first["executor"] not in first["independent_verifiers"] + assert first["runtime_mutation_performed_by_verifier"] is False + assert second["deduplicated"] is True + assert second["receipt_id"] == first["receipt_id"] + assert len(store.records) == 1 diff --git a/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json b/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json index 6fc0d083f..842d77d00 100644 --- a/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json +++ b/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json @@ -58,7 +58,7 @@ {"id":"AIA-CONV-038","category":"named_incident","status":"source_implemented_runtime_pending","title":"provider_freshness_signal 改成 readback、repair 與 verify","linked_work_items":["AIA-SRE-013","AIA-SRE-017"],"terminal_condition":"provider window、last seen、direct reference 與 no-false-green verifier 決定 terminal。"}, {"id":"AIA-CONV-039","category":"named_incident","status":"source_implemented_runtime_pending","title":"Backup.awoooi-frequent、Langfuse、Signoz、AI 等 backup family 分類與閉環","linked_work_items":["AIA-SRE-011","AIA-SRE-017"],"terminal_condition":"每類 backup signal 有正確 route、read-only receipt、去重與 closure。"}, {"id":"AIA-CONV-040","category":"named_incident","status":"source_implemented_runtime_pending","title":"Host188 PostgreSQL CPU、長查詢、MOMO provider backoff 由 Agent99 處置","linked_work_items":["AIA-SRE-006","AIA-SRE-010"],"terminal_condition":"typed Host188/DB dispatch、bounded apply、獨立 verifier 與 recipient receipt 完成。"}, - {"id":"AIA-CONV-041","category":"named_incident","status":"not_started_or_no_current_evidence","title":"StockPlatform cron_intelligence_sync 連續失敗自動診斷、修正與驗證","linked_work_items":["AIA-SRE-004","AIA-SRE-006"],"terminal_condition":"wrapped-command/root cause 修正後 cron freshness 與 exit-code verifier 關閉。"}, + {"id":"AIA-CONV-041","category":"named_incident","status":"source_implemented_runtime_pending","title":"StockPlatform cron_intelligence_sync 連續失敗自動診斷、修正與驗證","linked_work_items":["AIA-SRE-004","AIA-SRE-006"],"terminal_condition":"wrapped-command/root cause 修正後 cron freshness 與 exit-code verifier 關閉。","source_evidence":["exact StockPlatform cron_intelligence_sync schedule identity fails closed without cross-domain fallback","same-run durable failure, sanitized wrapped-command RCA and exact source-fix receipts create one fingerprint-deduplicated candidate","local or unreachable source commit remains release-pending and cannot invoke a provider, Agent99, executor or runtime mutation","durable runtime closure requires exact controlled deployment plus independent exit-code and freshness receipts from one deployment run"]}, {"id":"AIA-CONV-042","category":"knowledge_closure","status":"in_progress","title":"所有告警推理前整合 MCP 與 RAG context","linked_work_items":["AIA-SRE-012","AIA-SRE-015"],"terminal_condition":"每次推理記錄 context source IDs、freshness 與 retrieval receipt。"}, {"id":"AIA-CONV-043","category":"knowledge_closure","status":"source_implemented_runtime_pending","title":"診斷、候選、執行、驗證沉澱至 KM/RAG/MCP/PlayBook","linked_work_items":["AIA-SRE-015"],"terminal_condition":"同 run 六個 durable write receipts 唯一、可讀回且與 target/source 一致。"}, @@ -98,9 +98,9 @@ "active_or_completed_commitments": 68, "by_status": { "analysis_or_governance_complete": 6, - "source_implemented_runtime_pending": 33, + "source_implemented_runtime_pending": 34, "in_progress": 27, - "not_started_or_no_current_evidence": 2, + "not_started_or_no_current_evidence": 1, "superseded": 2 }, "product_runtime_closed_commitments": 0 diff --git a/k8s/awoooi-prod/15-service-registry-configmap.yaml b/k8s/awoooi-prod/15-service-registry-configmap.yaml index 8356692d6..bf2b35a68 100644 --- a/k8s/awoooi-prod/15-service-registry-configmap.yaml +++ b/k8s/awoooi-prod/15-service-registry-configmap.yaml @@ -9,7 +9,7 @@ metadata: app: awoooi component: service-registry annotations: - awoooi.wooo.work/source-sha256: "b1ad3cdd9ad3e041afd57f867738027b7668f72df63d0f238d7fa4befc584f9f" + awoooi.wooo.work/source-sha256: "d87300a155a528d15aeceb047bda9da57b3b15a8be2c37f24f2c047d95e175ab" data: service-registry.yaml: | # ops/config/service-registry.yaml @@ -299,6 +299,17 @@ data: stateful_level: AUTO containers: ["stock-platform"] + - name: stockplatform-cron-intelligence-sync + canonical_id: "schedule:stockplatform-v2:cron_intelligence_sync" + display_name: "StockPlatform cron_intelligence_sync" + host: "192.168.0.110" + stateful_level: STANDARD_HITL + asset_domain: docker_container + executor: host_ansible_executor + verifier: stockplatform_cron_independent_verifier + aliases: ["cron_intelligence_sync"] + reason: "精確排程身分；source release 與 runtime closure 均維持 receipt gate。" + # ─── 備份策略參考 ──────────────────────────────────────────────────────── backup_policies: velero_max_age_hours: 4 # Velero 備份過期閾值（Q2 決策） diff --git a/ops/config/service-registry.yaml b/ops/config/service-registry.yaml index 1072efada..64b134531 100644 --- a/ops/config/service-registry.yaml +++ b/ops/config/service-registry.yaml @@ -285,6 +285,17 @@ services: stateful_level: AUTO containers: ["stock-platform"] + - name: stockplatform-cron-intelligence-sync + canonical_id: "schedule:stockplatform-v2:cron_intelligence_sync" + display_name: "StockPlatform cron_intelligence_sync" + host: "192.168.0.110" + stateful_level: STANDARD_HITL + asset_domain: docker_container + executor: host_ansible_executor + verifier: stockplatform_cron_independent_verifier + aliases: ["cron_intelligence_sync"] + reason: "精確排程身分；source release 與 runtime closure 均維持 receipt gate。" + # ─── 備份策略參考 ──────────────────────────────────────────────────────── backup_policies: velero_max_age_hours: 4 # Velero 備份過期閾值（Q2 決策）