src/effective_status.py

from __future__ import annotations

from typing import Any


SUCCESS_VALIDATION_TOKENS = {
    "success",
    "succeeded",
    "passed",
    "full_inference_success",
    "validated",
    "validated_after_manual_space_test",
    "validated_after_space_test",
    "manual_validation_passed",
    "manual_validated",
    "recovered_by_manual_validation",
    "recovered_by_space_test",
}

PARTIAL_BUILD_TOKENS = {
    "partial",
    "partial_validation",
    "health_only",
    "full_inference_candidate_health_passed",
    "demo_usable_full_promise_not_verified",
    "placeholder_scaffold_deployed",
    "interactive_app_available_smoke_failed",
    "manual_test_required_smoke_failed",
    "completed_with_warnings",
    "unknown",
    "stale",
}

FAILED_BUILD_TOKENS = {"failed", "failure", "error"}
MANUAL_BUILD_TOKENS = {
    "manual_hardware_required",
    "generated_needs_manual_hardware",
    "waiting_manual_hardware",
    "manual_action_required",
    "waiting_manual_action",
}
BLOCKED_BUILD_TOKENS = {"technical_blocker", "technical_blocker_boot_only", "blocked"}


def _lower(value: Any) -> str:
    return str(value or "").strip().lower()


def _first_nonempty(*values: Any) -> str:
    for value in values:
        if value is not None and str(value).strip():
            return str(value).strip()
    return ""


def _target_space_from_bundle(bundle: dict[str, Any]) -> str:
    summary = bundle.get("summary") or bundle.get("summary_file") or {}
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    return _first_nonempty(
        summary.get("target_space"),
        summary.get("target_space_id"),
        state.get("target_space"),
        state.get("target_space_id"),
        launch.get("target_space"),
        launch.get("target_space_id"),
    )


def _contract_declares_no_full_inference(bundle: dict[str, Any]) -> bool:
    contract = bundle.get("inference_contract") or {}
    quality = bundle.get("demo_quality_contract") or {}
    gate = bundle.get("inference_gate") or {}
    blockers = bundle.get("technical_blockers") or {}
    for src in (contract, quality, gate, blockers):
        if not isinstance(src, dict):
            continue
        if src.get("full_inference_implemented") is False or src.get("real_inference_implemented") is False:
            return True
        text = " ".join(
            _lower(src.get(key))
            for key in (
                "status",
                "validation_level",
                "inference_strategy",
                "ui_status",
                "ui_badge",
                "reason",
                "message",
            )
        )
        if any(marker in text for marker in ("diagnostic_only", "diagnostic-only", "boot_only", "boot-only", "no_full_inference", "technical_blocker_boot_only")):
            return True
    return False


def _linked_rows(bundle: dict[str, Any]) -> list[dict[str, Any]]:
    linked = bundle.get("linked_validations") or {}
    rows = linked.get("validations") if isinstance(linked, dict) else linked
    if isinstance(rows, list):
        return [row for row in rows if isinstance(row, dict)]
    return []


def _manual_or_post_build_payload(bundle: dict[str, Any]) -> dict[str, Any]:
    post_build = bundle.get("post_build_validation_status") or {}
    if isinstance(post_build, dict) and post_build:
        return post_build
    manual = bundle.get("manual_validation_status") or {}
    if isinstance(manual, dict) and manual:
        return manual
    summary = bundle.get("summary") or {}
    manual = summary.get("manual_validation_status") if isinstance(summary, dict) else {}
    if isinstance(manual, dict) and manual:
        return manual
    linked = bundle.get("linked_validations") or {}
    if isinstance(linked, dict) and _lower(linked.get("effective_status")) in SUCCESS_VALIDATION_TOKENS:
        rows = _linked_rows(bundle)
        first = next((row for row in rows if _lower(row.get("status") or row.get("effective_status")) in SUCCESS_VALIDATION_TOKENS), {})
        return {
            "status": "success",
            "effective_status": "validated_after_manual_space_test",
            "validation_run_id": first.get("validation_run_id") or first.get("run_id") or "",
            "target_space": first.get("target_space") or "",
            "api_name": first.get("api_name") or "",
            "updated_at": linked.get("updated_at") or first.get("updated_at") or "",
        }
    return {}


def _successful_post_build_validation(bundle: dict[str, Any]) -> dict[str, Any]:
    payload = _manual_or_post_build_payload(bundle)
    if not isinstance(payload, dict) or not payload:
        return {}
    if _lower(payload.get("status") or payload.get("effective_status") or payload.get("effective_verdict")) not in SUCCESS_VALIDATION_TOKENS:
        return {}
    if _contract_declares_no_full_inference(bundle):
        return {}
    target = _target_space_from_bundle(bundle)
    validation_target = _first_nonempty(payload.get("target_space"), payload.get("target_space_id"))
    if target and validation_target and target != validation_target:
        return {}
    return payload


def _normalize_build_status(value: Any) -> str:
    status = _lower(value)
    if status in {"succeeded", "success", "passed", "done", "completed"}:
        return "full_inference_success"
    if status in {"cancelled", "canceled"}:
        return "stopped"
    if status in {"full_inference_candidate_health_passed", "health_only", "partial", "completed_with_warnings"}:
        return "partial_validation"
    if status in {"demo_usable_full_promise_not_verified", "placeholder_scaffold_deployed", "interactive_app_available_smoke_failed", "manual_test_required_smoke_failed"}:
        return status
    return status or "unknown"


def compute_effective_run_status(
    bundle: dict[str, Any],
    *,
    build_status: str | None = None,
    build_verdict: str | None = None,
) -> dict[str, Any]:
    """Return the immutable build verdict plus any post-build display override.

    v198.26.1 deliberately separates the original build result from linked Space
    Test results.  Linked validation may change only the effective/display
    status shown by the UI; it must not rewrite the persisted build verdict.
    """
    summary = bundle.get("summary") or bundle.get("summary_file") or {}
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    raw_build_status = _first_nonempty(
        build_status,
        summary.get("status"),
        state.get("status"),
        launch.get("status"),
        "unknown",
    )
    normalized_build = _normalize_build_status(raw_build_status)
    normalized_verdict = _normalize_build_status(build_verdict or normalized_build)
    post = _successful_post_build_validation(bundle)

    display_status = normalized_build
    display_label = ""
    post_build_status = "none"
    legacy_effective_status = ""
    can_promote = bool(post)
    promotion_reason = ""

    if can_promote:
        if normalized_build in FAILED_BUILD_TOKENS or "failed" in normalized_build:
            display_status = "recovered_by_space_test"
            legacy_effective_status = "recovered_by_manual_validation"
        elif normalized_build in BLOCKED_BUILD_TOKENS or "technical_blocker" in normalized_build:
            display_status = normalized_build
            can_promote = False
            promotion_reason = "blocked_build_not_promotable"
        else:
            display_status = "validated_after_space_test"
            legacy_effective_status = "validated_after_manual_space_test"
        if can_promote:
            post_build_status = display_status
            promotion_reason = "linked_space_test_success"

    if display_status == "validated_after_space_test":
        display_label = "Validated after Space Test"
    elif display_status == "recovered_by_space_test":
        display_label = "Recovered by Space Test"
    elif display_status == "full_inference_success":
        display_label = "Full inference success"
    elif display_status == "demo_usable_full_promise_not_verified":
        display_label = "Demo usable — full promise not verified"
    elif display_status == "placeholder_scaffold_deployed":
        display_label = "Placeholder demo — runtime not implemented"
    elif display_status == "interactive_app_available_smoke_failed":
        display_label = "Smoke input issue — Space reachable"
    elif display_status == "manual_test_required_smoke_failed":
        display_label = "Automatic smoke failed — Space Test recommended"
    elif display_status == "partial_validation":
        display_label = "Partial validation"
    elif display_status == "manual_hardware_required":
        display_label = "Manual hardware required"
    elif display_status in BLOCKED_BUILD_TOKENS:
        display_label = "Diagnostic Space"
    elif display_status in FAILED_BUILD_TOKENS:
        display_label = "Failed"
    elif display_status in {"running", "queued", "pending", "started", "scheduled"}:
        display_label = "Running"
    else:
        display_label = display_status.replace("_", " ").title() if display_status else "Unknown"

    return {
        "schema_version": "effective_run_status.v198_26_6",
        "build_status": normalized_build,
        "build_verdict": normalized_verdict,
        "build_verdict_preserved": True,
        "post_build_status": post_build_status,
        "post_build_validation": {
            "status": "success" if can_promote else ("blocked" if post and not can_promote else "none"),
            "source": "linked_space_test" if post else "none",
            "validation_run_id": post.get("validation_run_id") or post.get("run_id") or "",
            "target_space": post.get("target_space") or post.get("target_space_id") or "",
            "api_name": post.get("api_name") or "",
            "updated_at": post.get("updated_at") or post.get("validated_at") or "",
        },
        "display_status": display_status,
        "display_label": display_label,
        "effective_status": display_status,
        "effective_verdict": display_status,
        "legacy_effective_status": legacy_effective_status,
        "legacy_effective_verdict": legacy_effective_status,
        "promotion_reason": promotion_reason,
        "can_promote_from_space_test": can_promote,
    }