Add scientific result contracts

pytest.ini

@@ -1,2 +1,3 @@
 [pytest]
 asyncio_mode = auto
+testpaths = tests

src/uxarray_mcp/domain/vector_calc.py

@@ -155,6 +155,39 @@ def _looks_velocity(unit: str) -> bool:
     return warnings, warning_codes
 
 
+def _vector_component_evidence(u: Any, v: Any) -> dict[str, Any]:
+    """Return explicit metadata evidence for physical vector interpretation."""
+
+    def evidence(var: Any) -> dict[str, str | bool | None]:
+        attrs = getattr(var, "attrs", {}) or {}
+        units = str(attrs.get("units", "")).strip()
+        standard_name = str(attrs.get("standard_name", "")).strip()
+        long_name = str(attrs.get("long_name", "")).strip()
+        direction = f"{standard_name} {long_name}".lower()
+        return {
+            "units": units or None,
+            "standard_name": standard_name or None,
+            "long_name": long_name or None,
+            "velocity_units": any(
+                hint in units.lower() for hint in _VELOCITY_LIKE_UNIT_HINTS
+            ),
+            "eastward": any(term in direction for term in ("eastward", "zonal")),
+            "northward": any(term in direction for term in ("northward", "meridional")),
+        }
+
+    u_evidence, v_evidence = evidence(u), evidence(v)
+    return {
+        "u": u_evidence,
+        "v": v_evidence,
+        "units_supported": bool(
+            u_evidence["velocity_units"] and v_evidence["velocity_units"]
+        ),
+        "component_identity_supported": bool(
+            u_evidence["eastward"] and v_evidence["northward"]
+        ),
+    }
+
+
 def compute_gradient(
     uxds: Any,
     variable_name: str,
@@ -311,6 +344,7 @@ def compute_curl(
     component_warnings, warning_codes = _vector_component_warnings(
         u_variable, v_variable, u, v, "curl"
     )
+    component_evidence = _vector_component_evidence(u, v)
 
     result, uxarray_warnings = _call_capturing_warnings(
         lambda: u.curl(v, scale_by_radius=scale_by_radius)
@@ -350,6 +384,7 @@ def compute_curl(
         "scale_by_radius": bool(scale_by_radius),
         "stats": stats,
         "component_warnings": component_warnings,
+        "component_evidence": component_evidence,
     }
     from uxarray_mcp.provenance import attach_scientific_status
 
@@ -423,6 +458,7 @@ def compute_divergence(
     component_warnings, warning_codes = _vector_component_warnings(
         u_variable, v_variable, u, v, "divergence"
     )
+    component_evidence = _vector_component_evidence(u, v)
 
     result, uxarray_warnings = _call_capturing_warnings(lambda: u.divergence(v))
     for warning in uxarray_warnings:
@@ -451,6 +487,7 @@ def compute_divergence(
         "n_face": int(uxds.uxgrid.n_face),
         "stats": stats,
         "component_warnings": component_warnings,
+        "component_evidence": component_evidence,
     }
     from uxarray_mcp.provenance import attach_scientific_status
 

src/uxarray_mcp/tools/frontdoor.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+from functools import wraps
 from typing import Any
 
 
@@ -34,6 +35,78 @@ def _reject_unsupported_remote(use_remote: bool, operation: str) -> None:
         )
 
 
+def _finalize_analysis_result(operation: str, result: dict[str, Any]) -> dict[str, Any]:
+    """Attach front-door semantics without changing low-level operation results."""
+    status = "complete"
+    physically_interpretable: bool | None = None
+    warning_codes: list[str] = []
+
+    if operation == "validate_dataset":
+        passed = result.get("passed", result.get("is_valid"))
+        physically_interpretable = None
+        if passed is False:
+            status = "invalid"
+            warning_codes.append("DATASET_VALIDATION_FAILED")
+    elif operation in {"curl", "divergence"}:
+        evidence = result.get("component_evidence", {})
+        metadata_supported = bool(
+            evidence.get("units_supported")
+            and evidence.get("component_identity_supported")
+        )
+        scaling_supported = operation != "curl" or bool(result.get("scale_by_radius"))
+        physically_interpretable = bool(
+            metadata_supported
+            and scaling_supported
+            and not result.get("component_warnings")
+        )
+        if not physically_interpretable:
+            status = "warning"
+            if not metadata_supported:
+                warning_codes.append("VECTOR_COMPONENTS_UNVERIFIED")
+            if metadata_supported and not scaling_supported:
+                warning_codes.append("PHYSICAL_SCALING_UNVERIFIED")
+
+    result["scientific_status"] = {
+        "status": status,
+        "physically_interpretable": physically_interpretable,
+        "warning_codes": warning_codes,
+    }
+    result["postconditions"] = {
+        "status": "not_evaluated",
+        "checks": [],
+        "independent_verification": False,
+    }
+    return result
+
+
+def _with_analysis_contract(func: Any) -> Any:
+    """Preserve the MCP schema signature while finalizing successful results."""
+
+    @wraps(func)
+    def wrapped(operation: str, *args: Any, **kwargs: Any) -> dict[str, Any]:
+        result = func(operation, *args, **kwargs)
+        normalized = operation.strip().lower().replace("-", "_")
+        return _finalize_analysis_result(normalized, result)
+
+    return wrapped
+
+
+def _resolve_optional_session(
+    session_id: str | None, dataset_handle: str | None
+) -> str | None:
+    """Ignore a nonexistent optional session when explicit paths are sufficient."""
+    if session_id is None or dataset_handle is not None:
+        return session_id
+    from uxarray_mcp.state import get_session
+
+    try:
+        get_session(session_id)
+    except FileNotFoundError:
+        return None
+    return session_id
+
+
+@_with_analysis_contract
 def run_analysis(
     operation: str,
     grid_path: str | None = None,
@@ -129,6 +202,7 @@ def run_analysis(
     )
 
     op = operation.strip().lower().replace("-", "_")
+    session_id = _resolve_optional_session(session_id, dataset_handle)
 
     if op == "inspect_mesh":
         return inspect_mesh(

tests/test_frontdoor_semantics.py

@@ -0,0 +1,103 @@
+"""Structured scientific contract fields on the MCP analysis front door."""
+
+from __future__ import annotations
+
+from uxarray_mcp.tools.frontdoor import (
+    _finalize_analysis_result,
+    _resolve_optional_session,
+    run_analysis,
+)
+
+
+def test_run_analysis_signature_is_preserved():
+    assert "operation" in run_analysis.__annotations__
+    assert "grid_path" in run_analysis.__annotations__
+
+
+def test_complete_operation_has_explicit_unverified_postconditions():
+    result = _finalize_analysis_result("remap_to_rectilinear", {"stats": {"mean": 1.0}})
+
+    assert result["scientific_status"] == {
+        "status": "complete",
+        "physically_interpretable": None,
+        "warning_codes": [],
+    }
+    assert result["postconditions"] == {
+        "status": "not_evaluated",
+        "checks": [],
+        "independent_verification": False,
+    }
+
+
+def test_failed_validation_is_invalid():
+    result = _finalize_analysis_result("validate_dataset", {"passed": False})
+
+    assert result["scientific_status"] == {
+        "status": "invalid",
+        "physically_interpretable": None,
+        "warning_codes": ["DATASET_VALIDATION_FAILED"],
+    }
+
+
+def test_vector_warning_marks_result_not_physically_interpretable():
+    result = _finalize_analysis_result(
+        "curl", {"component_warnings": ["components lack velocity units"]}
+    )
+
+    assert result["scientific_status"] == {
+        "status": "warning",
+        "physically_interpretable": False,
+        "warning_codes": ["VECTOR_COMPONENTS_UNVERIFIED"],
+    }
+
+
+def test_supported_vector_is_physically_interpretable():
+    result = _finalize_analysis_result(
+        "divergence",
+        {
+            "component_warnings": [],
+            "component_evidence": {
+                "units_supported": True,
+                "component_identity_supported": True,
+            },
+        },
+    )
+
+    assert result["scientific_status"] == {
+        "status": "complete",
+        "physically_interpretable": True,
+        "warning_codes": [],
+    }
+
+
+def test_unscaled_curl_is_not_physically_interpretable():
+    result = _finalize_analysis_result(
+        "curl",
+        {
+            "component_warnings": [],
+            "component_evidence": {
+                "units_supported": True,
+                "component_identity_supported": True,
+            },
+            "scale_by_radius": False,
+        },
+    )
+
+    assert result["scientific_status"] == {
+        "status": "warning",
+        "physically_interpretable": False,
+        "warning_codes": ["PHYSICAL_SCALING_UNVERIFIED"],
+    }
+
+
+def test_nonexistent_optional_session_is_ignored(tmp_path, monkeypatch):
+    monkeypatch.setenv("UXARRAY_MCP_STATE_DIR", str(tmp_path))
+
+    assert _resolve_optional_session("correlation-label", None) is None
+
+
+def test_dataset_handle_keeps_strict_session_resolution():
+    assert (
+        _resolve_optional_session("required-session", "dataset_123")
+        == "required-session"
+    )