diff --git a/pytest.ini b/pytest.ini index 2f4c80e..78c5011 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,2 +1,3 @@ [pytest] asyncio_mode = auto +testpaths = tests diff --git a/src/uxarray_mcp/domain/vector_calc.py b/src/uxarray_mcp/domain/vector_calc.py index 2c0e1a4..35bb2b0 100644 --- a/src/uxarray_mcp/domain/vector_calc.py +++ b/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 diff --git a/src/uxarray_mcp/tools/frontdoor.py b/src/uxarray_mcp/tools/frontdoor.py index f6690b3..fba63aa 100644 --- a/src/uxarray_mcp/tools/frontdoor.py +++ b/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( diff --git a/tests/test_frontdoor_semantics.py b/tests/test_frontdoor_semantics.py new file mode 100644 index 0000000..cd60f29 --- /dev/null +++ b/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" + )