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

feat(aiops): gate StockPlatform cron repair closure

Browse Source
This commit is contained in:
Your Name
2026-07-22 18:18:30 +08:00
parent 979d52c81f
commit d4fd87db31
8 changed files with 1408 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
apps/api/src/services/independent_verifier_registry.py
View File

@@ -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",
},
)

943
apps/api/src/services/stockplatform_cron_intelligence_sync_control.py Normal file
View File

@@ -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",
}