fix: prompt module hardening — null guard, token preflight, dead code removal

agent_actions/prompt/context/scope_application.py

@@ -6,7 +6,7 @@
 from copy import deepcopy
 from typing import Any
 
-from agent_actions.errors import RecordContextError
+from agent_actions.errors import ConfigurationError, RecordContextError
 from agent_actions.logging.core.manager import fire_event
 from agent_actions.logging.events.io_events import (
     ContextFieldSkippedEvent,
@@ -144,12 +144,15 @@ def apply_context_scope(
         logger.debug("[STATIC_DATA] Merging %s static data fields into context", len(static_data))
         logger.debug("[STATIC_DATA] Fields: %s", list(static_data.keys()))
 
-        # Add under 'seed' namespace in prompt_context (for field reference replacement)
-        # This allows references like {{seed.exam_syllabus}} in prompts
+        # 'seed' is reserved for static data injection (SPECIAL_NAMESPACES blocks it upstream too).
         if "seed" in prompt_context:
-            logger.warning(
-                "Seed data namespace 'seed' conflicts with existing action. "
-                "Seed data will overwrite it."
+            raise ConfigurationError(
+                "Namespace collision: action named 'seed' conflicts with the seed data namespace. "
+                "Rename the action to avoid overwriting its output with static seed data.",
+                context={
+                    "action_name": action_name,
+                    "conflicting_namespace": "seed",
+                },
             )
         prompt_context["seed"] = static_data
         logger.debug("[SEED_DATA] Added to prompt_context under 'seed' namespace")

agent_actions/prompt/context/scope_namespace.py

@@ -53,28 +53,6 @@ def _extract_content_data(source_content: Any) -> dict:
     return {k: v for k, v in source_content.items() if k not in _RECORD_METADATA_KEYS}
 
 
-def _enrich_source_namespace(
-    base_namespace: dict[str, Any], current_item: dict[str, Any] | None
-) -> dict[str, Any]:
-    """
-    Merge fallback fields into the source namespace from the current item.
-
-    This helps downstream actions get at least one source-like namespace even if the
-    stored source file was sparse (e.g., only identifiers).
-    """
-    merged = dict(base_namespace or {})
-
-    if not current_item or not isinstance(current_item, dict):
-        return merged
-
-    fallback = _extract_content_data(current_item)
-    for key, value in fallback.items():
-        if key not in merged:
-            merged[key] = value
-
-    return merged
-
-
 def _filter_and_store_fields(
     field_context: dict,
     name: str,

agent_actions/prompt/formatter.py

@@ -21,18 +21,21 @@ def get_raw_prompt(agent_config):
             Raw prompt string
 
         Raises:
-            ConfigValidationError: If prompt is an empty/whitespace string for non-tool/hitl/seed actions
+            ConfigValidationError: If prompt is empty/whitespace or null for non-tool/hitl/seed/source actions
             PromptValidationError: If prompt retrieval or loading fails
         """
         raw_prompt = agent_config.get(PROMPT_KEY)
-        if (
-            agent_config.get("kind") not in ("tool", "hitl", "seed", "source")
-            and isinstance(raw_prompt, str)
-            and not raw_prompt.strip()
-        ):
-            raise ConfigValidationError(
-                f"prompt cannot be an empty string for action '{agent_config.get('agent_type', 'unknown')}'"
-            )
+        if agent_config.get("kind") not in ("tool", "hitl", "seed", "source"):
+            if isinstance(raw_prompt, str) and not raw_prompt.strip():
+                raise ConfigValidationError(
+                    f"prompt cannot be an empty string for action "
+                    f"'{agent_config.get('agent_type', 'unknown')}'"
+                )
+            if PROMPT_KEY in agent_config and raw_prompt is None:
+                raise ConfigValidationError(
+                    f"prompt is null for action '{agent_config.get('agent_type', 'unknown')}'. "
+                    f"Remove the prompt key or provide a valid prompt string."
+                )
         try:
             if PROMPT_KEY not in agent_config:
                 return "Process the following content: {content}"

agent_actions/prompt/message_builder.py

@@ -334,21 +334,25 @@ def build(
                 for m in messages
             ]
 
-        # Token overflow pre-flight guard
-        if model_name is not None:
-            estimated_tokens = sum(len(m.content) for m in messages) // 4
-            model_limit = _MODEL_CONTEXT_LIMITS.get(model_name, _DEFAULT_CONTEXT_LIMIT)
-            if estimated_tokens > model_limit:
-                raise PromptTooLargeError(
-                    f"Estimated prompt size ({estimated_tokens} tokens) exceeds "
-                    f"model context window ({model_limit} tokens)",
-                    context={
-                        "estimated_tokens": estimated_tokens,
-                        "model_limit": model_limit,
-                        "provider": provider,
-                        "model_name": model_name,
-                    },
-                )
+        # Token overflow pre-flight guard — always runs.
+        # When model_name is unknown, falls back to _DEFAULT_CONTEXT_LIMIT.
+        estimated_tokens = sum(len(m.content) for m in messages) // 4
+        model_limit = (
+            _MODEL_CONTEXT_LIMITS.get(model_name, _DEFAULT_CONTEXT_LIMIT)
+            if model_name is not None
+            else _DEFAULT_CONTEXT_LIMIT
+        )
+        if estimated_tokens > model_limit:
+            raise PromptTooLargeError(
+                f"Estimated prompt size ({estimated_tokens} tokens) exceeds "
+                f"model context window ({model_limit} tokens)",
+                context={
+                    "estimated_tokens": estimated_tokens,
+                    "model_limit": model_limit,
+                    "provider": provider,
+                    "model_name": model_name,
+                },
+            )
 
         return LLMMessageEnvelope(
             messages=messages,

agent_actions/prompt/prompt_utils.py

@@ -1,6 +1,5 @@
 """Module for String Processing Functions"""
 
-import json
 import re
 from typing import Any
 
@@ -135,100 +134,3 @@ def inject_function_outputs_into_prompt(
         else:
             processed_prompt = prompt_config
         return (processed_prompt, captured_results)
-
-    @staticmethod
-    def parse_field_references(prompt: str) -> list:
-        """
-        Parse {reference.field} patterns from prompt.
-
-        Pattern matches:
-        - {source.field}
-        - {agent.field}
-        - {agent.nested.field}
-        - {agent.items.0} (array index)
-
-        Args:
-            prompt: Prompt string with field references
-
-        Returns:
-            List of dicts with 'reference', 'field_path', and 'full_match'
-        """
-        pattern = r"\{([a-zA-Z_][a-zA-Z0-9_]*(?:\.[a-zA-Z0-9_]+)+)\}"
-        references = []
-        for match in re.finditer(pattern, prompt):
-            full_ref = match.group(1)
-            parts = full_ref.split(".")
-            references.append(
-                {
-                    "reference": parts[0],
-                    "field_path": parts[1:],
-                    "full_match": match.group(0),
-                    "start": match.start(),
-                    "end": match.end(),
-                }
-            )
-        return references
-
-    @staticmethod
-    def resolve_field_reference(reference: str, field_path: list, context: dict):
-        """
-        Resolve a field reference to its value in the context.
-
-        Args:
-            reference: Reference name (e.g., 'source', 'extractor')
-            field_path: List of field names (e.g., ['metrics', 'count'])
-            context: Dict with available references
-
-        Returns:
-            Resolved value
-
-        Raises:
-            ValueError: If reference or field not found
-        """
-        if reference not in context:
-            available = ", ".join(context.keys())
-            raise ValueError(f"Reference '{reference}' not found. Available: [{available}]")
-        data = context[reference]
-        for field in field_path:
-            if isinstance(data, dict) and field in data:
-                data = data[field]
-            elif isinstance(data, list) and field.isdigit():
-                idx = int(field)
-                if 0 <= idx < len(data):
-                    data = data[idx]
-                else:
-                    raise ValueError(f"Index {idx} out of range for array in '{reference}'")
-            else:
-                field_str = ".".join(field_path)
-                raise ValueError(f"Field '{field_str}' not found in '{reference}'")
-        return data
-
-    @staticmethod
-    def replace_field_references(prompt: str, context: dict) -> str:
-        """
-        Replace all {reference.field} patterns with their values.
-
-        Args:
-            prompt: Prompt string with field references
-            context: Dict with available references
-
-        Returns:
-            Prompt with all references replaced
-
-        Raises:
-            ValueError: If reference or field not found
-        """
-        references = PromptUtils.parse_field_references(prompt)
-        for ref in reversed(references):
-            try:
-                value = PromptUtils.resolve_field_reference(
-                    ref["reference"], ref["field_path"], context
-                )
-                if isinstance(value, dict | list):
-                    value_str = json.dumps(value, indent=2)
-                else:
-                    value_str = str(value)
-                prompt = prompt[: ref["start"]] + value_str + prompt[ref["end"] :]
-            except ValueError as e:
-                raise ValueError(f"Error resolving {ref['full_match']}: {str(e)}") from e
-        return prompt

agent_actions/prompt/service.py

@@ -364,7 +364,6 @@ def _render_prompt_template(
                                 ue,
                             )
                             prompt_context[action] = _PermissiveNamespace()
-                            # Re-render with all known skipped deps injected
                             for other in skipped_actions:
                                 if other not in prompt_context:
                                     prompt_context[other] = _PermissiveNamespace()