script-build-and-framework-tmp/visualization/backend/app/source_lineage/resolver.py

from __future__ import annotations

import hashlib
import json
import re
from typing import Any

from .schema import EvidenceSpec, SourceBinding, SourceRecord, unresolved_binding


_INDEX = re.compile(r"\[([0-9]+)\]")


def json_pointer(path: str | None) -> str:
    """Convert the visualization's legacy dotted paths to RFC 6901 pointers."""
    value = str(path or "").strip()
    if not value:
        return ""
    if " / " in value:
        return ""
    value = _INDEX.sub(r".\1", value)
    parts = [part for part in value.split(".") if part]
    return "/" + "/".join(
        part.replace("~", "~0").replace("/", "~1") for part in parts
    )


def pointer_lookup(value: Any, pointer: str) -> tuple[bool, Any]:
    if not pointer:
        return True, value
    current = value
    for raw in pointer.lstrip("/").split("/"):
        part = raw.replace("~1", "/").replace("~0", "~")
        if isinstance(current, list):
            try:
                current = current[int(part)]
            except (ValueError, IndexError):
                return False, None
        elif isinstance(current, dict):
            if part not in current:
                return False, None
            current = current[part]
        else:
            return False, None
    return True, current


def pointer_value(value: Any, pointer: str) -> Any:
    return pointer_lookup(value, pointer)[1]


def source_digest(value: Any) -> str:
    serialized = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()


class SourceResolver:
    """Maps semantic CardData fields to deduplicated physical source records.

    It deliberately does not infer adoption from matching text.  The only
    evidence inputs are the field's declared relation and physical origin.
    """

    def __init__(
        self,
        *,
        bundle: dict[str, Any],
        event_details: dict[int, dict[str, Any]],
    ) -> None:
        self.bundle = bundle
        self.event_details = event_details
        self.sources: dict[str, SourceRecord] = {}
        self._span_offsets: dict[tuple[int, str, str], int] = {}

    def binding_for_field(
        self,
        field: dict[str, Any],
        *,
        binding_id: str,
        role: str | None = None,
    ) -> SourceBinding:
        source = field.get("source") if isinstance(field.get("source"), dict) else {}
        source_kind = str(source.get("kind") or "")
        source_ref = str(source.get("ref") or "")
        raw, physical = self._raw_source(source_kind, source_ref, source)
        if source_kind == "calculation":
            raw = {
                "value": field.get("value"),
                "calculation": source.get("label"),
            }
        source_id = self._source_id(source_kind, source_ref, source, physical)
        field_path = str(source.get("fieldPath") or "")
        pointer = json_pointer(field_path)
        resolution = "resolved"

        if raw is None:
            missing = self._ensure_missing_source(
                source_id,
                source,
                reason=f"未找到 {source.get('label') or source_ref or '原始记录'}",
            )
            return unresolved_binding(
                binding_id,
                reason=missing["locator"]["reason"],
                role=role or self._role(field),
                source_id=source_id,
                code="source-record-not-saved",
            )

        selector: dict[str, Any]
        found, selected = pointer_lookup(raw, pointer) if pointer else (True, raw)
        if source_kind == "calculation":
            selector = {"kind": "json-pointer", "path": "/value"}
            selected = field.get("value")
        elif source_kind == "artifact" and field_path in {"snapshotRef", "recordCounts", "historicalVersion"}:
            selector = {"kind": "whole-record"}
            selected = field.get("value")
        elif pointer and found:
            selector = {"kind": "json-pointer", "path": pointer}
        elif pointer:
            # The legacy CardData path may refer to a derived projection rather
            # than a persisted record.  Keep the source visible, but do not
            # pretend that the pointer resolves against it.
            selector = {
                "kind": "unresolved",
                "reason": f"原始记录中无法安全定位字段 {field_path}",
                "code": "field-path-mismatch",
            }
            selected = field.get("value")
            resolution = "partial"
        else:
            selector = {"kind": "whole-record"}
            selected = field.get("value")

        record = self._ensure_source(source_id, source_kind, source, raw, physical)
        record.setdefault("selectedValues", []).append(
            {
                "bindingId": binding_id,
                "fieldId": field.get("id"),
                "label": field.get("label"),
                "path": pointer or field_path or None,
                "value": selected,
            }
        )
        transform = self._transform(source_kind, source)
        if (
            transform.get("kind") == "direct"
            and found
            and not _equivalent_value(field.get("value"), selected)
        ):
            transform = {
                "kind": "parsed",
                "operation": "business-projection-v1",
                "outputKey": str(field.get("id") or binding_id),
            }
        binding: SourceBinding = {
            "id": binding_id,
            "sourceId": source_id,
            "role": role or self._role(field),  # type: ignore[typeddict-item]
            "selector": selector,
            "transform": transform,
            "evidence": self._evidence(field, source_kind),
            "resolution": resolution,  # type: ignore[typeddict-item]
        }
        if transform.get("kind") == "parsed":
            binding["evidence"]["confidence"] = "deterministic"
        if binding["resolution"] != "resolved" and transform.get("kind") == "direct":
            binding["transform"] = {
                "kind": "parsed",
                "operation": "unresolved-source-projection-v1",
                "outputKey": str(field.get("id") or binding_id),
            }
            binding["evidence"]["confidence"] = "unconfirmed"
        if "未安全关联" in str(source.get("label") or ""):
            binding["resolution"] = "unsafe"
            binding["evidence"]["confidence"] = "unconfirmed"
        return binding

    def binding_for_event(
        self,
        event_id: int,
        *,
        binding_id: str,
        path: str = "",
        role: str = "process",
        availability: str = "produced-by-run",
    ) -> SourceBinding:
        event = self.event_details.get(int(event_id))
        source_id = f"event:{event_id}"
        if event is None:
            self._ensure_missing_source(
                source_id,
                {"label": f"Event {event_id}", "ref": source_id},
                reason=f"Event {event_id} 或 Event Body 未保存",
            )
            return unresolved_binding(
                binding_id,
                reason=f"Event {event_id} 或 Event Body 未保存",
                role=role,
                source_id=source_id,
                code="event-body-not-saved",
            )
        source = {
            "kind": "runtime-event",
            "label": str(event.get("event_name") or f"Event {event_id}"),
            "ref": source_id,
            "fieldPath": path,
        }
        pointer = json_pointer(path)
        found, selected = pointer_lookup(event, pointer) if pointer else (True, event)
        resolution = "resolved" if found else "partial"
        if not pointer:
            selector = {"kind": "whole-record"}
        elif found:
            selector = {"kind": "json-pointer", "path": pointer}
        else:
            selector = {
                "kind": "unresolved",
                "reason": f"Event {event_id} 中不存在字段 {path}",
                "code": "field-path-mismatch",
            }
        record = self._ensure_source(source_id, "runtime-event", source, event, event)
        record.setdefault("selectedValues", []).append(
            {"bindingId": binding_id, "fieldId": binding_id, "label": source["label"], "path": pointer or None, "value": selected}
        )
        return {
            "id": binding_id,
            "sourceId": source_id,
            "role": role,  # type: ignore[typeddict-item]
            "selector": selector,
            "transform": {"kind": "direct"},
            "evidence": {
                "availability": availability,
                "adoption": "not-applicable" if availability == "produced-by-run" else "not-recorded",
                "confidence": "exact",
            },  # type: ignore[typeddict-item]
            "resolution": resolution,  # type: ignore[typeddict-item]
        }

    def text_span_binding(
        self,
        event_id: int,
        *,
        binding_id: str,
        path: str,
        exact_text: Any,
        role: str,
    ) -> SourceBinding:
        event = self.event_details.get(int(event_id))
        source_id = f"event:{event_id}"
        if event is None:
            self._ensure_missing_source(
                source_id,
                {"label": f"Event {event_id}", "ref": source_id},
                reason=f"Event {event_id} 或 Event Body 未保存",
            )
            return unresolved_binding(
                binding_id,
                reason=f"Event {event_id} 或 Event Body 未保存",
                role=role,
                source_id=source_id,
                code="event-body-not-saved",
            )
        pointer = json_pointer(path)
        raw_text = pointer_value(event, pointer)
        needle = str(exact_text or "")
        if not isinstance(raw_text, str) or not needle:
            binding = self.binding_for_event(
                event_id,
                binding_id=binding_id,
                path=path,
                role=role,
                availability="returned-to-agent" if role == "output" else "direct-read",
            )
            binding["transform"] = {
                "kind": "parsed",
                "operation": "structured-text-section-v1",
                "outputKey": binding_id,
            }
            binding["evidence"]["confidence"] = "deterministic"
            return binding
        span_key = (int(event_id), pointer, needle)
        search_from = self._span_offsets.get(span_key, 0)
        start = raw_text.find(needle, search_from)
        if start < 0 and search_from:
            start = raw_text.find(needle)
        if start < 0:
            binding = self.binding_for_event(
                event_id,
                binding_id=binding_id,
                path=path,
                role=role,
                availability="returned-to-agent" if role == "output" else "direct-read",
            )
            binding["transform"] = {
                "kind": "parsed",
                "operation": "normalized-text-section-v1",
                "outputKey": binding_id,
            }
            binding["evidence"]["confidence"] = "deterministic"
            return binding
        record = self._ensure_source(
            source_id,
            "runtime-event",
            {"label": str(event.get("event_name") or f"Event {event_id}"), "ref": source_id},
            event,
            event,
        )
        record.setdefault("selectedValues", []).append(
            {"bindingId": binding_id, "fieldId": binding_id, "label": binding_id, "path": pointer, "value": needle}
        )
        self._span_offsets[span_key] = start + len(needle)
        return {
            "id": binding_id,
            "sourceId": source_id,
            "role": role,  # type: ignore[typeddict-item]
            "selector": {
                "kind": "text-span",
                "path": pointer,
                "startCodePoint": start,
                "endCodePoint": start + len(needle),
                "exactText": needle,
                "sourceDigest": source_digest(raw_text),
            },
            "transform": {"kind": "direct"},
            "evidence": {
                "availability": "returned-to-agent" if role == "output" else "direct-read",
                "adoption": "not-recorded",
                "confidence": "exact",
            },
            "resolution": "resolved",
        }

    def add_unresolved_source(self, binding: SourceBinding) -> None:
        source_id = binding["sourceId"]
        reason = str(binding.get("selector", {}).get("reason") or "原始记录无法定位")
        self._ensure_missing_source(
            source_id,
            {"label": "记录缺失"},
            reason=reason,
        )

    def add_log_anchor(
        self, event_id: int, module: dict[str, Any], *, event_msg_id: str
    ) -> str:
        source_id = f"log:event:{event_id}"
        self.sources[source_id] = {
            "id": source_id,
            "kind": "log-anchor",
            "label": f"锚定日志模块 · Event {event_id}",
            "locator": {
                "table": "script_build_log",
                "eventId": event_id,
                "msgId": event_msg_id,
                "logAnchor": module.get("anchor"),
            },
            "completeness": "complete",
            "resultState": "present",
            "truncated": False,
            "selectedValues": [],
            "rawRecord": module.get("content"),
        }
        return source_id

    def empty_database_result_binding(
        self,
        *,
        binding_id: str,
        table: str,
        filters: dict[str, Any],
        label: str,
        role: str = "output",
    ) -> SourceBinding:
        """Represent a successful zero-row lookup without calling it missing."""
        filter_key = ",".join(f"{key}={filters[key]}" for key in sorted(filters))
        source_id = f"db-result:{table}:{filter_key or 'all'}"
        self.sources[source_id] = {
            "id": source_id,
            "kind": "database",
            "label": label,
            "locator": {
                "table": table,
                "filters": filters,
                "resultCount": 0,
            },
            "completeness": "complete",
            "resultState": "empty",
            "truncated": False,
            "selectedValues": [{
                "bindingId": binding_id,
                "fieldId": binding_id,
                "label": "查询结果",
                "path": "",
                "value": [],
            }],
            "rawRecord": [],
        }
        return {
            "id": binding_id,
            "sourceId": source_id,
            "role": role,  # type: ignore[typeddict-item]
            "selector": {"kind": "whole-record"},
            "transform": {
                "kind": "calculated",
                "operation": "数据库查询结果计数为 0",
            },
            "evidence": {
                "availability": "visualization-derived",
                "adoption": "not-applicable",
                "confidence": "deterministic",
            },
            "resolution": "resolved",
        }

    def members_binding(
        self,
        field: dict[str, Any],
        *,
        binding_id: str,
        members: list[dict[str, Any]],
        reducer: str,
        role: str = "output",
    ) -> SourceBinding:
        """Create a deterministic aggregate while retaining every member ref."""
        base_field = {
            **field,
            "source": {
                "kind": "calculation",
                "label": reducer,
                "ref": f"calculation:{binding_id}",
                "fieldPath": "value",
            },
            "relation": "persisted-output",
        }
        base = self.binding_for_field(base_field, binding_id=binding_id, role=role)
        member_selectors: list[dict[str, Any]] = []
        member_resolutions: list[str] = []
        for index, member in enumerate(members, 1):
            member_binding = self.binding_for_field(
                member,
                binding_id=f"{binding_id}:member:{index}",
                role=role,
            )
            member_selectors.append(
                {
                    "sourceId": member_binding["sourceId"],
                    "selector": member_binding["selector"],
                }
            )
            member_resolutions.append(str(member_binding.get("resolution") or "missing"))
            member_source = self.sources.get(member_binding["sourceId"])
            if member_source and member_source.get("selectedValues"):
                member_source["selectedValues"][-1]["aggregateBindingId"] = binding_id
        if not member_selectors:
            base["selector"] = {
                "kind": "unresolved",
                "reason": "确定性聚合没有找到可定位的参与记录",
                "code": "binding-map-missing",
            }
            base["resolution"] = "missing"
            base["evidence"]["confidence"] = "unconfirmed"
            return base
        base["selector"] = {
            "kind": "members",
            "members": member_selectors,
            "reducer": reducer,
        }
        base["transform"] = {"kind": "calculated", "operation": reducer}
        base["evidence"] = {
            "availability": "visualization-derived",
            "adoption": "not-applicable",
            "confidence": "deterministic",
        }
        if any(value in {"missing", "unsafe"} for value in member_resolutions):
            base["resolution"] = "partial"
            base["evidence"]["confidence"] = "unconfirmed"
        elif any(value == "partial" for value in member_resolutions):
            base["resolution"] = "partial"
        return base

    def _raw_source(
        self, source_kind: str, source_ref: str, source: dict[str, Any]
    ) -> tuple[Any, dict[str, Any]]:
        if source_kind == "runtime-event":
            event_id = _suffix_int(source_ref)
            if event_id is not None:
                event = self.event_details.get(event_id)
                return event, event or {}
            # A runtime source is only trustworthy when it names one concrete
            # Event.  Treating an empty ref as "all related Events" previously
            # let a multipath evaluator impersonate a missing script_evaluator
            # in round summaries.
            return None, {"detailRef": source_ref, "reason": "未保存可定位的 Event ref"}
        if source_kind == "database":
            return self._database_record(source_ref, str(source.get("label") or ""))
        if source_kind == "artifact":
            if "branch:" in source_ref:
                branch_id = _suffix_int(source_ref)
                row = _by_int(self.bundle.get("branches"), "branch_id", branch_id)
                # CardData paths are rooted at the branch record
                # (`candidate_snapshot.snapshot`), so retain that root rather
                # than pre-extracting candidate_snapshot and breaking the
                # declared JSON Pointer.
                return row, {"table": "script_build_branch", "recordId": (row or {}).get("id")}
            artifact = self.bundle.get("currentArtifact")
            return artifact, {"artifactRef": source_ref or "artifact:base:current"}
        if source_kind == "calculation":
            return {"value": None, "calculation": source.get("label")}, {"calculation": source.get("label")}
        return None, {"ref": source_ref}

    def _database_record(self, source_ref: str, label: str) -> tuple[Any, dict[str, Any]]:
        normalized = label.lower()
        if source_ref.startswith("missing-convergence:"):
            parts = source_ref.split(":", 4)
            record_type = parts[1] if len(parts) > 1 else "unknown"
            round_index = _integer(parts[3]) if len(parts) > 3 else None
            branch_ids = [
                _integer(value)
                for value in (parts[4].split(",") if len(parts) > 4 else [])
                if _integer(value) is not None
            ]
            return [], {
                "tables": (
                    ["script_build_event", "script_build_event_body"]
                    if record_type == "review"
                    else ["script_build_multipath_decision"]
                ),
                "filters": {
                    "round_index": round_index,
                    "branch_ids": branch_ids,
                },
                "resultCount": 0,
            }
        if "script_build_record" in normalized or source_ref == "run:objective" or source_ref == "run:final-result":
            row = self.bundle.get("record")
            return row, {"table": "script_build_record", "recordId": (row or {}).get("id")}
        if "data_decision" in normalized or source_ref.startswith("data-decision:"):
            row = _by_int(self.bundle.get("dataDecisions"), "id", _suffix_int(source_ref))
            return row, {"table": "script_build_data_decision", "recordId": (row or {}).get("id")}
        if "multipath" in normalized or source_ref.startswith("multipath-decision:"):
            row = _by_int(self.bundle.get("multipathDecisions"), "id", _suffix_int(source_ref))
            return row, {"table": "script_build_multipath_decision", "recordId": (row or {}).get("id")}
        if "domain_info" in normalized or source_ref.startswith("domain-info:"):
            row = _by_int(self.bundle.get("domainInfo"), "id", _suffix_int(source_ref))
            return row, {"table": "script_build_domain_info", "recordId": (row or {}).get("id")}
        if "script_build_round" in normalized or source_ref.startswith("round:") and ":branch:" not in source_ref:
            round_index = _round_index(source_ref)
            row = _by_int(self.bundle.get("rounds"), "round_index", round_index)
            return row, {"table": "script_build_round", "recordId": (row or {}).get("id")}
        if "script_build_branch" in normalized or ":branch:" in source_ref:
            branch_id = _branch_id(source_ref)
            row = _by_int(self.bundle.get("branches"), "branch_id", branch_id)
            return row, {"table": "script_build_branch", "recordId": (row or {}).get("id")}
        return None, {"table": label or "unknown", "ref": source_ref}

    def _ensure_source(
        self,
        source_id: str,
        source_kind: str,
        source: dict[str, Any],
        raw: Any,
        physical: dict[str, Any],
    ) -> SourceRecord:
        existing = self.sources.get(source_id)
        if existing is not None:
            return existing
        kind = {
            "database": "database",
            "runtime-event": "runtime-event",
            "artifact": "artifact",
            "calculation": "calculation",
            "log-anchor": "log-anchor",
        }.get(source_kind, "missing")
        if "历史回退" in str(source.get("label") or ""):
            kind = "historical-fallback"
        completeness = "complete"
        truncated = False
        omitted = 0
        if kind == "runtime-event" and isinstance(raw, dict):
            sides = [raw.get("input"), raw.get("output")]
            if raw.get("input") is None and raw.get("output") is None:
                completeness = "partial"
            for side in sides:
                if isinstance(side, dict) and side.get("truncated"):
                    truncated = True
                    omitted += int(side.get("omittedCharacters") or 0)
        record: SourceRecord = {
            "id": source_id,
            "kind": kind,  # type: ignore[typeddict-item]
            "label": str(source.get("label") or source.get("ref") or source_id),
            "locator": physical,
            "completeness": "partial" if truncated else completeness,  # type: ignore[typeddict-item]
            "resultState": (
                "empty"
                if raw == []
                or (
                    kind == "calculation"
                    and isinstance(raw, dict)
                    and raw.get("value") == []
                )
                else "present"
            ),
            "truncated": truncated,
            "selectedValues": [],
            "rawRecord": raw,
        }
        if omitted:
            record["omittedCharacters"] = omitted
        if isinstance(raw, dict) and kind == "runtime-event":
            record["status"] = str(raw.get("status") or "unknown")
            record["durationMs"] = _integer(raw.get("duration_ms"))
            record["locator"] = {
                "eventId": raw.get("id"),
                "eventName": raw.get("event_name"),
                "eventType": raw.get("event_type"),
                "table": "script_build_event + script_build_event_body",
            }
        self.sources[source_id] = record
        return record

    def _ensure_missing_source(
        self, source_id: str, source: dict[str, Any], *, reason: str
    ) -> SourceRecord:
        record = self.sources.get(source_id)
        if record is None:
            record = {
                "id": source_id,
                "kind": "missing",
                "label": str(source.get("label") or source_id),
                "locator": {"ref": source.get("ref"), "reason": reason},
                "completeness": "missing",
                "resultState": "missing",
                "truncated": False,
                "selectedValues": [],
                "rawRecord": None,
            }
            self.sources[source_id] = record
        return record

    @staticmethod
    def _source_id(
        source_kind: str,
        source_ref: str,
        source: dict[str, Any],
        physical: dict[str, Any],
    ) -> str:
        if source_kind == "runtime-event" and _suffix_int(source_ref) is not None:
            return f"event:{_suffix_int(source_ref)}"
        if source_kind == "database" and physical.get("table"):
            return f"db:{physical['table']}:{physical.get('recordId') or source_ref or 'unknown'}"
        if source_kind == "artifact":
            return f"artifact:{source_ref or physical.get('artifactRef') or 'unknown'}"
        key = f"{source_kind}:{source_ref}:{source.get('label')}"
        return f"source:{hashlib.sha1(key.encode('utf-8')).hexdigest()[:12]}"

    @staticmethod
    def _role(field: dict[str, Any]) -> str:
        relation = str(field.get("relation") or "")
        return {
            "direct-input": "input",
            "previous-result": "basis",
            "standing-constraint": "constraint",
            "explicit-basis": "basis",
            "available-upstream": "basis",
            "persisted-output": "output",
            "run-output": "output",
        }.get(relation, "basis")

    @staticmethod
    def _transform(source_kind: str, source: dict[str, Any]) -> dict[str, Any]:
        if source_kind == "calculation":
            return {"kind": "calculated", "operation": str(source.get("label") or "确定性计算")}
        if "历史回退" in str(source.get("label") or ""):
            return {
                "kind": "fallback",
                "selectedSourceId": str(source.get("ref") or ""),
                "candidateSourceIds": [str(source.get("ref") or "")],
            }
        return {"kind": "direct"}

    @staticmethod
    def _evidence(field: dict[str, Any], source_kind: str) -> EvidenceSpec:
        relation = str(field.get("relation") or "")
        if source_kind == "calculation":
            return {
                "availability": "visualization-derived",
                "adoption": "not-applicable",
                "confidence": "deterministic",
            }
        if relation == "explicit-basis":
            return {
                "availability": "direct-read",
                "adoption": "explicitly-adopted",
                "confidence": "exact",
            }
        if relation in {"direct-input", "standing-constraint"}:
            return {
                "availability": "direct-read",
                "adoption": "not-recorded",
                "confidence": "exact",
            }
        if relation == "previous-result":
            return {
                "availability": "returned-to-agent" if source_kind == "runtime-event" else "available-upstream",
                "adoption": "not-recorded",
                "confidence": "exact" if source_kind == "runtime-event" else "safe-association",
            }
        if relation == "available-upstream":
            return {
                "availability": "available-upstream",
                "adoption": "not-recorded",
                "confidence": "safe-association",
            }
        if relation == "persisted-output":
            availability = "returned-to-agent" if source_kind == "runtime-event" else "produced-by-run"
            return {
                "availability": availability,
                "adoption": "not-recorded" if availability == "returned-to-agent" else "not-applicable",
                "confidence": "exact",
            }
        if relation == "run-output":
            return {
                "availability": "produced-by-run",
                "adoption": "not-applicable",
                "confidence": "exact",
            }
        return {
            "availability": "available-upstream",
            "adoption": "not-recorded",
            "confidence": "unconfirmed",
        }


def _integer(value: Any) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _suffix_int(value: str) -> int | None:
    return _integer(str(value or "").rsplit(":", 1)[-1])


def _round_index(ref: str) -> int | None:
    match = re.search(r"(?:^|:)round:(\d+)", f":{ref}")
    return _integer(match.group(1)) if match else None


def _branch_id(ref: str) -> int | None:
    match = re.search(r":branch:(\d+)", ref)
    return _integer(match.group(1)) if match else None


def _by_int(items: Any, key: str, wanted: int | None) -> dict[str, Any] | None:
    if wanted is None:
        return None
    return next(
        (
            item
            for item in items or []
            if isinstance(item, dict) and _integer(item.get(key)) == wanted
        ),
        None,
    )


def _equivalent_value(left: Any, right: Any) -> bool:
    try:
        return json.dumps(left, ensure_ascii=False, sort_keys=True, default=str) == json.dumps(
            right, ensure_ascii=False, sort_keys=True, default=str
        )
    except (TypeError, ValueError):
        return str(left) == str(right)