"""Finite sanitized execution-refusal copy; no graph/input/error interpolation."""

from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import replace
from typing import TYPE_CHECKING, Any

from runtime_contracts.builder.copy_ids import BuilderCopyId
from runtime_contracts.builder.execution import ExecutionEvidenceSummary
from runtime_contracts.builder.presentation import BuilderText, copy_text, text_parts
from services.dify_builder.native_launch_observation import FailureResource

if TYPE_CHECKING:
    from core.dify_builder.models import Run

FALLBACK = copy_text(BuilderCopyId.REPLIES_EXECUTION_SAFETY_REQUIRES_REVIEW_OR_REVISED)
_MESSAGES = {
    "execution_permission_denied": copy_text(BuilderCopyId.TEST_PERMISSION_DENIED),
    "execution_owner_mismatch": copy_text(BuilderCopyId.TEST_PERMISSION_DENIED),
    "agent_access_denied": copy_text(BuilderCopyId.TEST_PERMISSION_DENIED),
    "stale_execution_revision": copy_text(BuilderCopyId.TEST_CONTEXT_CHANGED),
    "test_input_binding_mismatch": copy_text(BuilderCopyId.TEST_CONTEXT_CHANGED),
    "effective_input_binding_mismatch": copy_text(BuilderCopyId.TEST_CONTEXT_CHANGED),
    "context_digest_mismatch": copy_text(BuilderCopyId.TEST_CONTEXT_CHANGED),
    "agent_capability_changed": copy_text(BuilderCopyId.TEST_CONTEXT_CHANGED),
    "code_risk_assessment_stale": copy_text(BuilderCopyId.TEST_POLICY_REVIEW),
    "installed_tool_unavailable": copy_text(BuilderCopyId.RESOURCES_CATALOGUE_UNAVAILABLE),
    "agent_tool_metadata_unavailable": copy_text(BuilderCopyId.RESOURCES_CATALOGUE_UNAVAILABLE),
    "agent_not_available": copy_text(BuilderCopyId.RESOURCES_CATALOGUE_UNAVAILABLE),
    "unsupported_node_implementation": copy_text(BuilderCopyId.REPLIES_BUILDER_DOES_NOT_YET_HAVE_SAFE),
    "unsupported_json_tool_configuration": copy_text(BuilderCopyId.REPLIES_BUILDER_CANNOT_SAFELY_TEST_THIS_TOOL),
    "unsupported_json_credentials": copy_text(BuilderCopyId.REPLIES_BUILDER_CANNOT_SAFELY_TEST_THIS_TOOL2),
    "unsupported_json_content_constant": copy_text(BuilderCopyId.REPLIES_PUT_THIS_J_S_O_N_CONTENT_IN_A),
    "unsupported_json_filter": copy_text(BuilderCopyId.REPLIES_USE_A_SUPPORTED_J_S_O_N_PATH_SELECTION),
    "unsupported_json_selector": copy_text(BuilderCopyId.REPLIES_BIND_J_S_O_N_PARSE_TEXT_FROM_A),
    "invalid_json_parameters": copy_text(BuilderCopyId.REPLIES_SET_J_S_O_N_PARSE_CONTENT_AND_FILTER),
    "json_execution_limit": copy_text(BuilderCopyId.REPLIES_REDUCE_J_S_O_N_PARSE_INPUT_OR_OUTPUT),
    "installed_json_tool_unavailable": copy_text(BuilderCopyId.REPLIES_INSTALL_THE_QUALIFIED_J_S_O_N_PROCESS_VERSION),
    "unsupported_json_tool_installation": copy_text(
        BuilderCopyId.REPLIES_INSTALL_THE_QUALIFIED_J_S_O_N_PROCESS_VERSION
    ),
    "json_tool_installation_changed": copy_text(BuilderCopyId.REPLIES_THE_INSTALLED_J_S_O_N_PROCESS_VERSION_CHANGED),
    "unsupported_template_syntax": copy_text(BuilderCopyId.REPLIES_TEMPLATE_SUPPORTS_SCALAR_TEXT_IF_ELIF),
    "unsupported_template_configuration": copy_text(BuilderCopyId.REPLIES_EDIT_TEMPLATE_VARIABLES_TO_USE_UNIQUE),
    "unsupported_template_selector": copy_text(BuilderCopyId.REPLIES_USE_TEMPLATE_INPUTS_FROM_START_SCALARS),
    "template_execution_limit": copy_text(BuilderCopyId.REPLIES_REDUCE_TEMPLATE_SIZE_OR_SELECTED_TEXT),
}


def execution_refusal_message(reason_code: str) -> BuilderText:
    return _MESSAGES.get(reason_code, FALLBACK)


# One bounded semantic preparation fallback. Existing fixed-copy IDs are translated by
# Web; this backend-only disclosure cannot add a frontend catalog dependency.
_CODE_RISK_UNAVAILABLE = {
    "en": "The step risk explanation is unavailable. This test will simulate this step "
    "and cannot verify its real behavior. "
    "Prepare a new test to try the risk review again.",
    "zh": "暂时无法说明此步骤的风险。本次测试将模拟该节点，无法验证其业务行为。请重新准备测试以重试风险评估。",
}


def code_risk_unavailable_notice(reply_language: str) -> str:
    return _CODE_RISK_UNAVAILABLE["zh" if reply_language.lower().startswith("zh") else "en"]


# Only bounded backend-owned display labels are translated here. Technical
# codes remain in diagnostics/failure_code; resource names and user text are
# never interpreted as localization keys. Other locales use the English label.
_REASON_LABELS = {
    "connection_refused": ("connection refused", "连接被拒绝"),
    "timeout": ("request timed out", "请求超时"),
    "authentication": ("authentication failed", "身份认证失败"),
    "permission_denied": ("provider permission denied", "提供商权限被拒绝"),
    "credential_unavailable": ("credential missing or inaccessible", "凭据缺失或无权访问"),
    "authorization_lookup_failed": ("credential access could not be verified", "无法验证凭据访问权限"),
    "rate_limit": ("rate limit exceeded", "请求超过频率限制"),
    "service_unavailable": ("service unavailable", "服务不可用"),
    "provider_invocation_failed": ("model invocation failed", "模型调用失败"),
    "model_resolution_failed": ("model configuration unavailable", "模型配置不可用"),
    "empty_response": ("empty model response", "模型响应为空"),
    "json_decode_failed": ("invalid JSON response", "模型响应不是有效的 JSON"),
    "response_schema_invalid": ("response does not match the required schema", "模型响应不符合要求的结构"),
    "unknown": ("cause unavailable", "暂未获取到具体原因"),
}
_SCOPE_LABELS = {"structure": ("structure", "结构"), "configuration": ("configuration", "配置")}


def _display_label(labels: Mapping[str, tuple[str, str]], value: str, reply_language: str) -> str:
    pair = labels.get(value)
    return pair[1 if reply_language.lower().startswith("zh") else 0] if pair else value


def change_scope_label(scope: str, reply_language: str) -> str:
    """Translate the display scope without mutating the ChangeSet's machine scope."""
    return _display_label(_SCOPE_LABELS, scope, reply_language)


def observed_failure_message(
    summary: ExecutionEvidenceSummary | None, *, reply_language: str = "en-US"
) -> BuilderText | None:
    """Observed preparation failure and unknown effects need different recovery."""
    if summary is None:
        return None
    if (
        summary.failure_resource
        and summary.failure_reason
        in {"authentication", "permission_denied", "credential_unavailable", "authorization_lookup_failed"}
        and summary.sealed
        and (
            summary.safety_outcome == "preparation_failed"
            or (summary.safety_outcome == "native_failed" and summary.failure_phase == "execution")
        )
    ):
        return _resource_setup_guidance(
            reason=summary.failure_reason,
            resource=summary.failure_resource,
            node_ids=summary.failure_node_ids,
            reply_language=reply_language,
        )
    if summary.safety_outcome == "preparation_failed" or (
        summary.safety_outcome == "native_failed" and summary.failure_phase == "execution"
    ):
        message = (
            copy_text(BuilderCopyId.TEST_RUNTIME_SERVICE_FAILED)
            if summary.failure_phase == "execution"
            else (
                "测试准备失败（{reason}），工作流尚未开始执行。请检查模型/工具服务和配置，"
                "或选择其他工作区资源，再准备并确认一次新测试。"
                if reply_language.lower().startswith("zh")
                else "Test preparation failed ({reason}) before workflow execution started. "
                "Check model/tool services and configuration, or choose another workspace resource; "
                "then prepare and confirm a fresh test."
            ).format(reason=_display_label(_REASON_LABELS, summary.failure_reason or "unknown", reply_language))
        )
        if summary.failure_resource:
            _, provider, name = summary.failure_resource
            return text_parts(
                message, "\n", "/".join(filter(None, (provider, name))), "\n", ", ".join(summary.failure_node_ids)
            )
        return message
    if summary.safety_outcome == "execution_evidence_unknown":
        return copy_text(BuilderCopyId.REPLIES_EXECUTION_OUTCOME_IS_UNKNOWN_RUN_AGAIN_BEFORE_PUBLISHING)
    return None


def native_service_failure(run: Run) -> bool:
    """Observed resource failure: preserve the Run, avoid speculative graph repair."""
    summary = run.verification.execution_evidence if run.verification else None
    if (
        run.status == "failed"
        and not run.dify_run_id
        and summary is not None
        and summary.sealed
        and summary.safety_outcome == "preparation_failed"
        and summary.failure_phase == "preparation"
        and summary.failure_resource is not None
        and bool(summary.failure_node_ids)
        and summary.failure_reason in {"credential_unavailable", "authorization_lookup_failed"}
    ):
        return True
    return bool(
        run.status == "failed"
        and run.dify_run_id
        and summary is not None
        and summary.sealed
        and summary.native_run_id == run.dify_run_id
        and summary.failure_resource is not None
        and bool(summary.failure_node_ids)
        and summary.safety_outcome == "native_failed"
        and summary.failure_phase == "execution"
        and summary.failure_reason in {"service_unavailable", "authentication", "permission_denied"}
    )


def tool_credential_failure_message(
    run: Run, *, graph: Mapping[str, Any], graph_revision: str, reply_language: str = "en-US"
) -> str | None:
    """Observe an exact native Tool setup envelope, without altering Run evidence.

    Prefer sealed attribution elsewhere. This fallback requires the executed
    graph revision and one failed Tool node; titles and workflow-level text
    cannot establish the resource. Unknown, foreign and stale evidence stays
    unknown. This display observation grants no execution or retry authority.
    """
    from services.dify_builder.native_launch_observation import _public_id

    evidence = run.verification
    summary = evidence.execution_evidence if evidence else None
    if (
        run.status != "failed"
        or not run.dify_run_id
        or evidence is None
        or not graph_revision
        or evidence.executed_graph_revision != graph_revision
        or summary is None
        or not summary.sealed
        or summary.safety_outcome != "native_failed"
        or summary.native_run_id != run.dify_run_id
        or summary.failure_resource is not None
        or summary.failure_phase is not None
        or summary.failure_reason is not None
        or run.execution_refusal is not None
    ):
        return None
    failed = [node for node in run.per_node if node.status in {"failed", "exception"}]
    if len(failed) != 1 or not _public_id(failed[0].node_id):
        return None
    # These are backend-owned native runtime messages, not keyword matching.
    reasons = {
        "Failed to get tool runtime: Credentials not available: Please use ENTERPRISE global credentials": (
            "enterprise_global_credential_required"
        ),
        "Failed to get tool runtime: Credentials not available": "credential_unavailable",
    }
    reason = reasons.get(failed[0].error)
    nodes = graph.get("nodes")
    if reason is None or not isinstance(nodes, list):
        return None
    matches = [node for node in nodes if isinstance(node, Mapping) and node.get("id") == failed[0].node_id]
    if len(matches) != 1 or not isinstance(data := matches[0].get("data"), Mapping):
        return None
    if data.get("type") != "tool" or data.get("provider_type") not in {"builtin", "plugin"}:
        return None
    provider, name = data.get("provider_id"), data.get("tool_name")
    if not isinstance(provider, str) or not isinstance(name, str) or not _public_id(provider) or not _public_id(name):
        return None
    return _resource_setup_guidance(
        reason=reason,
        resource=("tool", provider, name),
        node_ids=(failed[0].node_id,),
        reply_language=reply_language,
    )


def _resource_setup_guidance(
    *, reason: str, resource: FailureResource, node_ids: tuple[str, ...], reply_language: str
) -> str:
    label = _display_label(_REASON_LABELS, reason, reply_language)
    resource_label = "/".join(filter(None, resource[1:]))
    nodes = ", ".join(node_ids)
    global_credential = reason == "enterprise_global_credential_required"
    if reply_language.lower().startswith("zh"):
        kind = "工具" if resource[0] == "tool" else "模型"
        if global_credential:
            return (
                f"{kind}设置失败：需要企业全局凭据。请联系工作区管理员配置或选择获准的企业全局凭据，"
                f"然后重新准备并确认测试。\n{resource_label}\n{nodes}"
            )
        recovery = (
            "请检查工作区凭据访问权限后重新准备测试。"
            if reason == "authorization_lookup_failed"
            else "请配置或选择有权访问此资源的凭据，并检查提供商权限后重新准备测试。"
        )
        return f"{kind}资源设置失败：{label}。{recovery}\n{resource_label}\n{nodes}"
    kind = "Tool" if resource[0] == "tool" else "Model"
    if global_credential:
        return (
            f"{kind} setup failed: enterprise global credential required. Ask a workspace administrator "
            "to configure or select an authorized enterprise global credential; "
            f"then prepare and confirm a fresh test.\n{resource_label}\n{nodes}"
        )
    recovery = (
        "Check workspace credential access before preparing a fresh test."
        if reason == "authorization_lookup_failed"
        else "Configure or select an authorized credential and check provider permissions "
        "before preparing a fresh test."
    )
    return f"{kind} resource setup failed: {label}. {recovery}\n{resource_label}\n{nodes}"


def sanitize_observed_failure(run: Run) -> Run:
    """Bound Builder errors only; native authorized Run storage keeps original evidence."""
    summary = run.verification.execution_evidence if run.verification else None
    if (
        not native_service_failure(run)
        or summary is None
        or summary.failure_resource is None
        or summary.failure_reason is None
        or summary.failure_reason == "service_unavailable"
    ):
        return run
    guidance = _resource_setup_guidance(
        reason=summary.failure_reason,
        resource=summary.failure_resource,
        node_ids=summary.failure_node_ids,
        reply_language="en",
    )
    return replace(
        run,
        error=guidance,
        per_node=[
            replace(node, error=guidance) if node.node_id in summary.failure_node_ids and node.error else node
            for node in run.per_node
        ],
    )


def sanitize_resource_failure_event(
    payload: Mapping[str, Any], *, graph: Mapping[str, Any], recognized_errors: dict[str, str]
) -> Mapping[str, Any]:
    """Suppress recognized provider bodies in Builder streams, never seal evidence.

    The admitted graph and native stream ID scope presentation only. Terminal
    classification still requires the persisted executed graph and native rows.
    Code/unattributed errors retain their existing diagnostics.
    """
    from core.dify_builder.models import NodeOutput, Run
    from services.dify_builder.native_runtime_failure import native_resource_failure
    from services.dify_builder.run_mapping import run_id_from_stream_chunk

    data = payload.get("data")
    native_id = run_id_from_stream_chunk(payload)
    if not native_id or not isinstance(data, Mapping) or data.get("status") not in {"failed", "exception"}:
        return payload
    error = data.get("error")
    if not isinstance(error, str) or not error:
        return payload
    key = f"{native_id}:{error}"
    guidance = None
    if payload.get("event") == "node_finished" and isinstance(node_id := data.get("node_id"), str):
        failure = native_resource_failure(
            Run(
                status="failed",
                dify_run_id=native_id,
                per_node=[NodeOutput(node_id=node_id, status="failed", error=error)],
            ),
            graph,
        )
        if failure is not None and failure[2] in {"authentication", "permission_denied"}:
            guidance = _resource_setup_guidance(
                reason=failure[2], resource=failure[0], node_ids=failure[1], reply_language="en"
            )
            recognized_errors[key] = guidance
    elif payload.get("event") == "workflow_finished":
        guidance = recognized_errors.get(key)
    return {**payload, "data": {**data, "error": guidance}} if guidance else payload


def cognition_failure_message(
    records: Sequence[Mapping[str, Any]], fallback: BuilderText, *, reply_language: str = "en-US"
) -> BuilderText:
    """Use bounded observations from this operation; never provider error text."""
    for record in reversed(records):
        code = record.get("code")
        if isinstance(code, str) and code in {
            "provider_invocation_failed",
            "model_resolution_failed",
            "empty_response",
            "json_decode_failed",
            "response_schema_invalid",
        }:
            reason = record.get("reason")
            if not isinstance(reason, str) or reason not in {
                "connection_refused",
                "timeout",
                "authentication",
                "rate_limit",
                "service_unavailable",
            }:
                reason = code
            return copy_text(
                BuilderCopyId.OPERATION_MODEL_FAILED, reason=_display_label(_REASON_LABELS, reason, reply_language)
            )
    return fallback


def capability_assessment_unavailable_message(
    records: Sequence[Mapping[str, Any]], *, reply_language: str
) -> BuilderText:
    """Disclose unknown feasibility using only this assessment's observations.

    The caller supplies records captured during the assessment invocation,
    excluding prior discovery failures. Return fixed guidance and optional
    sanitized model-failure copy; never claim an unsupported capability or
    alter approval, bindings, requirements or retry policy. No I/O is performed.
    """
    guidance = (
        "计划可行性检查暂不可用。请在批准前检查计划和资源，或调整计划后重试。"
        if reply_language.lower().startswith("zh")
        else "The plan feasibility check is unavailable. Review the plan and resources before approving, "
        "or adjust the plan to retry."
    )
    detail = cognition_failure_message(records, "", reply_language=reply_language)
    return text_parts(guidance, "\n", detail) if detail else guidance