Implement code-gated viz mapping with clarification disambiguation

Author: jmsexton03

database/configs/base_amrex_config.py

@@ -2384,6 +2384,68 @@ def get_viz_variable_catalog(cls, repo_root: Path | None = None) -> list[dict[st
         return []
 
     @classmethod
+    def get_viz_tier1_intents(cls) -> dict[str, dict[str, Any]]:
+        """
+        Return canonical semantic visualization intents (Tier 1).
+        """
+        return {
+            "temperature": {"aliases": ["temperature", "temp", "thermal"]},
+            "velocity": {"aliases": ["velocity", "speed"]},
+            "vertical_velocity": {
+                "aliases": [
+                    "vertical velocity",
+                    "vertical_velocity",
+                    "w-velocity",
+                    "w velocity",
+                    "updraft",
+                    "downdraft",
+                ]
+            },
+            "pressure": {"aliases": ["pressure", "pres"]},
+            "density": {"aliases": ["density", "rho"]},
+            "vorticity": {"aliases": ["vorticity", "vort"]},
+            "cloud_water": {
+                "aliases": [
+                    "cloud water",
+                    "cloud_water",
+                    "liquid water",
+                    "cloud liquid",
+                    "qc",
+                ]
+            },
+        }
+
+    @classmethod
+    def build_viz_tier2_candidates(cls, repo_root: Path | None = None) -> dict[str, list[dict[str, Any]]]:
+        """
+        Build solver candidates (Tier 2) from live source catalog.
+        """
+        catalog = cls.get_viz_variable_catalog(repo_root=repo_root)
+        if not catalog:
+            return {}
+
+        intents = cls.get_viz_tier1_intents()
+        candidates: dict[str, list[dict[str, Any]]] = {}
+        for token, spec in intents.items():
+            aliases = {token.lower()}
+            for alias in spec.get("aliases", []) or []:
+                aliases.add(str(alias).strip().lower())
+            token_candidates: list[dict[str, Any]] = []
+            for entry in catalog:
+                if not isinstance(entry, dict):
+                    continue
+                name = str(entry.get("name", "")).strip()
+                if not name:
+                    continue
+                entry_aliases = {name.lower()}
+                for alias in entry.get("aliases", []) or []:
+                    entry_aliases.add(str(alias).strip().lower())
+                if aliases.intersection(entry_aliases):
+                    token_candidates.append(dict(entry))
+            if token_candidates:
+                candidates[token] = token_candidates
+        return candidates
+
     def get_default_slice_axis(cls) -> str | None:
         """
         Return preferred default slice-normal axis for visualization.
@@ -2399,6 +2461,13 @@ def get_plotfile_var_param(cls) -> str:
         """
         return "amr.plot_vars"
 
+    @classmethod
+    def get_plot_var_param_candidates(cls) -> list[str]:
+        """
+        Return ordered plot-var ParmParse candidate keys (primary first).
+        """
+        return [cls.get_plotfile_var_param()]
+
     @classmethod
     def get_plotfile_period_param(cls) -> str | None:
         """

database/configs/erf_config.py

@@ -41,6 +41,10 @@ def get_plotfile_period_param(cls) -> str | None:
     def get_plotfile_step_interval_param(cls) -> str | None:
         return "erf.plot_int_1"
 
+    @classmethod
+    def get_plot_var_param_candidates(cls) -> list[str]:
+        return ["erf.plot_vars_1", "amr.plot_vars", "plot_vars"]
+
     # === Registry Metadata ===
     github_org = "erf-model"
     github_repo = "ERF"

database/configs/pelelmex_config.py

@@ -171,6 +171,10 @@ class PeleLMeXConfig(BaseAMReXConfig):
     def get_plotfile_var_param(cls) -> str:
         return "peleLM.derive_plot_vars"
 
+    @classmethod
+    def get_plot_var_param_candidates(cls) -> list[str]:
+        return ["peleLM.derive_plot_vars", "amr.plot_vars", "plot_vars"]
+
     @classmethod
     def get_plotfile_period_param(cls) -> str | None:
         return "amr.plot_per"

docs/intent_runtime_routing.md

@@ -18,6 +18,26 @@ mixing it into inputs-file modifications.
 Runtime intent is consumed by runner and visualization paths. It is not passed
 into `ConfigModelFactory.apply_modifications`.
 
+## Visualization Mapping Contract
+
+Visualization field resolution is code-gated and tiered:
+
+1. Tier 1 semantic intents: canonical tokens extracted from prompt text (for
+   example `cloud_water`, `temperature`, `velocity`).
+2. Tier 2 solver candidates: live solver-code catalog candidates resolved from
+   `build_viz_tier2_candidates(repo_root)`.
+3. Tier 3 executable selection: final `requested_fields` consumed by input
+   writing/visualization, selected only from Tier 2 candidates.
+
+Policy:
+
+- Missing solver code/catalog during visualization mapping is a hard runtime
+  failure with remediation guidance.
+- No blind fallback to unresolved generic semantic tokens in final
+  `requested_fields`.
+- Ambiguity or no-match is routed to clarification with candidate-constrained
+  question payloads.
+
 ## Precedence
 
 Effective runtime values are resolved in this order:

docs/workflow_visualization.md

@@ -12,6 +12,27 @@ The workflow generates visualizations from AMReX plotfiles after a successful
 analysis pass. Visualization uses the multi-backend service (AMReX tools first,
 then optional pyamrex, then yt) and records metadata in the graph state.
 
+### Intent Resolution and Clarification
+
+Visualization intent resolution is deterministic-first and code-derived:
+
+- Architect emits semantic visualization intent only.
+- `visualization_intent_node` performs canonical mapping through solver
+  Tier-2 candidates from live code catalogs.
+- If mapping is ambiguous/unresolved, the node records diagnostics:
+  - `visualization_mapping_candidates`
+  - `visualization_mapping_unresolved`
+  - `visualization_mapping_source`
+  - `visualization_mapping_confidence`
+- Clarification node asks constrained disambiguation questions using provided
+  candidate sets, and AI responses outside the candidate set are rejected.
+
+Hard-fail behavior:
+
+- If solver catalog/source code is unavailable for a visualization mapping
+  request, runtime fails with actionable remediation instead of silently
+  guessing fields.
+
 ### Workflow Flow
 
 ```

src/models/graph_state_canonical.py

@@ -55,6 +55,10 @@ class GraphState(TypedDict, total=False):
     requested_plot_vars: Optional[List[str]]       # Prompt-extracted plot quantities
     visualization_config: Optional[Dict[str, Any]] # Prompt-extracted visualization settings
     visualization_intent: Optional[Dict[str, Any]] # Canonical visualization intent contract
+    visualization_mapping_candidates: Optional[Dict[str, Any]]
+    visualization_mapping_unresolved: Optional[List[str]]
+    visualization_mapping_source: Optional[str]
+    visualization_mapping_confidence: Optional[float]
 
     # ========================================
     # PLANNING PHASE (Architect outputs)

src/nodes/clarification_handler_node.py

@@ -48,6 +48,16 @@ def clarification_handler_node(state: dict[str, Any]) -> dict[str, Any]:
             "clarification_turns": turns + 1,
             "skip_further_clarification": False,
         }
+    if answered_by == "ai_agent" and not _is_allowed_candidate_response(
+        pending_record.get("question", {}),
+        response,
+    ):
+        return {
+            "clarification_history": history,
+            "clarification_needed": True,
+            "clarification_turns": turns + 1,
+            "skip_further_clarification": False,
+        }
 
     updated_record = _resolve_record(pending_record, response, answered_by)
     history[pending_index] = updated_record
@@ -177,6 +187,18 @@ def _merge_plot_vars(current: list[Any], resolved_value: str) -> list[str]:
     return merged
 
 
+def _is_allowed_candidate_response(question: dict[str, Any], response: str) -> bool:
+    question_data = _question_to_dict(question)
+    context = question_data.get("context", {})
+    if not isinstance(context, dict):
+        return True
+    allowed = context.get("candidates")
+    if not isinstance(allowed, list) or not allowed:
+        return True
+    allowed_set = {str(v).strip() for v in allowed if str(v).strip()}
+    return str(response).strip() in allowed_set
+
+
 def _as_history(value: Any) -> list[dict[str, Any]]:
     if not isinstance(value, list):
         return []

src/nodes/clarification_node.py

@@ -6,11 +6,11 @@
 
 from src.models import GraphState
 from src.models.clarification_schemas import ClarificationQuestion, ClarificationRecord
+from src.services.viz_param_extractor import get_plot_var_param_candidates
 
 
 REQUIRED_FIELDS = ("n_cell", "max_level", "stop_time", "max_step")
 RESOURCE_FIELDS = ("node_count", "time_limit", "cluster")
-PLOTFILE_VAR_KEYS = ("amr.plot_vars", "peleLM.derive_plot_vars", "plot_vars")
 MAX_CLARIFICATION_TURNS = 3
 
 
@@ -261,8 +261,41 @@ def _check_level_4_visualization(
     prompt: str,
     requested_plot_vars: Any,
 ) -> ClarificationQuestion | None:
-    del state, locked_fields, prompt
-    if requested_plot_vars == [] and not _has_plotfile_var(resolved):
+    del locked_fields, prompt
+    mapping_candidates = state.get("visualization_mapping_candidates")
+    mapping_unresolved = state.get("visualization_mapping_unresolved")
+    if isinstance(mapping_candidates, dict) and mapping_candidates:
+        token = next(iter(mapping_candidates.keys()))
+        entries = mapping_candidates.get(token, [])
+        names: list[str] = []
+        if isinstance(entries, list):
+            for entry in entries:
+                if isinstance(entry, dict):
+                    name = str(entry.get("name", "")).strip()
+                    if name:
+                        names.append(name)
+        if names:
+            return ClarificationQuestion(
+                field_name="requested_plot_vars",
+                question_text=f"Which field should represent '{token}' in visualization output?",
+                decision_level=4,
+                fallback_tier="amrex_generic",
+                context={
+                    "reason": "visualization_mapping_ambiguity",
+                    "token": token,
+                    "candidates": names,
+                },
+            )
+    if isinstance(mapping_unresolved, list) and mapping_unresolved:
+        token = str(mapping_unresolved[0]).strip()
+        return ClarificationQuestion(
+            field_name="requested_plot_vars",
+            question_text=f"No solver field matched '{token}'. Which plot variable should be used?",
+            decision_level=4,
+            fallback_tier="amrex_generic",
+            context={"reason": "visualization_mapping_unresolved", "token": token, "candidates": []},
+        )
+    if requested_plot_vars == [] and not _has_plotfile_var(resolved, state):
         return ClarificationQuestion(
             field_name="plot_vars",
             question_text="Which variables should be written to plotfiles for visualization?",
@@ -386,7 +419,7 @@ def _base_context(
         "missing_fields": missing_fields,
         "ambiguous_fields": [],
         "requested_plot_vars_empty": requested_plot_vars == [],
-        "missing_plotfile_vars": requested_plot_vars == [] and not _has_plotfile_var(resolved),
+        "missing_plotfile_vars": requested_plot_vars == [] and not _has_plotfile_var(resolved, None),
     }
 
 
@@ -406,8 +439,12 @@ def _fallback_tier_for_field(field: str) -> str:
     return "free_text"
 
 
-def _has_plotfile_var(resolved: dict[str, Any]) -> bool:
-    for key in PLOTFILE_VAR_KEYS:
+def _has_plotfile_var(resolved: dict[str, Any], state: GraphState | None) -> bool:
+    solver = ""
+    if isinstance(state, dict):
+        solver = str(state.get("selected_solver") or "").strip()
+    keys = get_plot_var_param_candidates(solver) + ["plot_vars"]
+    for key in keys:
         if _has_value(resolved.get(key)):
             return True
     return False

src/nodes/visualization_intent_node.py

@@ -12,9 +12,9 @@
 from src.models import GraphState
 from src.models.visualization_intent import VisualizationIntent, VisualizationPlotSpec
 from src.services.viz_param_extractor import (
-    canonicalize_requested_plot_vars,
     convert_prompt_seconds_to_solver_time,
     extract_viz_params_from_prompt,
+    resolve_viz_field_mapping,
 )
 
 
@@ -84,6 +84,7 @@ def build_visualization_intent(
     requested_plot_vars: list[str] | None = None,
     visualization_config: dict[str, Any] | None = None,
     prior_intent: dict[str, Any] | None = None,
+    mapping_diagnostics_out: dict[str, Any] | None = None,
 ) -> VisualizationIntent:
     """
     Build a canonical visualization_intent payload.
@@ -112,12 +113,21 @@ def build_visualization_intent(
         + list(prior_fields)
         + list(prior_vars)
     )
-    if solver_name:
-        merged_requested = canonicalize_requested_plot_vars(
+    mapping_diagnostics: dict[str, Any] = {
+        "resolved_fields": list(merged_requested),
+        "candidate_fields_by_token": {},
+        "unresolved_tokens": [],
+        "ambiguous_tokens": [],
+        "mapping_source": "semantic_only" if merged_requested else "none",
+        "mapping_confidence": 1.0,
+    }
+    if solver_name and merged_requested:
+        mapping_diagnostics = resolve_viz_field_mapping(
             merged_requested,
             code_name=solver_name,
             repo_root=repo_root,
         )
+        merged_requested = list(mapping_diagnostics.get("resolved_fields", []))
 
     merged_config: dict[str, Any] = {}
     if isinstance(extracted_config, dict):
@@ -184,6 +194,10 @@ def build_visualization_intent(
     if source not in {"prompt", "clarification", "default"}:
         source = "default"
 
+    if isinstance(mapping_diagnostics_out, dict):
+        mapping_diagnostics_out.clear()
+        mapping_diagnostics_out.update(mapping_diagnostics)
+
     return VisualizationIntent(
         requested_fields=merged_requested,
         cadence_prompt_seconds=cadence_prompt_seconds,
@@ -248,18 +262,24 @@ def visualization_intent_node(state: GraphState) -> dict[str, Any]:
     prompt = str(state.get("prompt") or state.get("user_requirement") or "")
     prior_intent = state.get("visualization_intent") if isinstance(state.get("visualization_intent"), dict) else {}
 
+    mapping: dict[str, Any] = {}
     model = build_visualization_intent(
         prompt=prompt,
         solver_name=solver_name,
         repo_root=repo_root,
         requested_plot_vars=[],
         visualization_config={},
         prior_intent=prior_intent,
+        mapping_diagnostics_out=mapping,
     )
     intent = model.model_dump()
 
     return {
         "visualization_intent": intent,
         "requested_plot_vars": list(model.requested_fields),
         "visualization_config": dict(intent.get("visualization_config", {})),
+        "visualization_mapping_candidates": dict(mapping.get("candidate_fields_by_token", {})),
+        "visualization_mapping_unresolved": list(mapping.get("unresolved_tokens", [])),
+        "visualization_mapping_source": mapping.get("mapping_source"),
+        "visualization_mapping_confidence": mapping.get("mapping_confidence"),
     }