orchestrata_ultimate.orchestrator.intent_parser

Reference › orchestrata_ultimate › orchestrator › intent_parser

The parser for formally typed intent strings, the structured counterpart of the natural-language path in intent_classifier. It is used by intent_broker and by the governance engine, and is traced to spec 003 story US-001. Where a human request arrives as free text through Concierge it goes to the classifier; where a caller already has a string of the form action, resource type, path and constraints, this module turns it into a typed record.

IntentParser.parse matches the string against the grammar action:type/path followed by optional :key=value constraints. It validates the action against the Action enum (read, write, execute, create, delete, modify) and the type against ResourceType (file, command, git, network, system, directory), rejects an empty path, and returns the module's own Intent dataclass with constraints as a dictionary. Any violation raises IntentParseError; try_parse returns None instead. The Intent here differs from the richer record in intent_types: it offers to_string, to_prolog and hash for caching. A ParameterExtractor (see parameter_extractor) can be injected. IntentExtractor handles a different input, the tool calls made by an agent: extract maps the tool name (Read, Write, Edit, Bash, Glob, Grep, Task, WebFetch, WebSearch, TodoWrite) to an action and builds a file, command, search, task, network or system intent from its parameters, returning None for an unmapped tool.

At a glance

Class diagram

Dependency diagram

API

Classes

Action (enum)

Bases: Enum · Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-013']) · Line 23

Valid intent actions.

ResourceType (enum)

Bases: Enum · Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-004', 'REQ-003-US-001-014']) · Line 35

Valid resource types.

Intent (dataclass)

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-023']), dataclass · Line 48

Parsed intent with all components.

IntentParseError

Bases: Exception · Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-034']) · Line 83

Error parsing intent string.

IntentParser

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-011']) · Line 88

Parser for intent strings.

IntentExtractor

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-003', 'REQ-003-US-001-013']) · Line 191

Extracts intents from tool calls.

IntentValidator

Decorators: traces(spec='003', story='US-003', reqs=['REQ-003-US-003-002', 'REQ-003-US-003-003']) · Line 326

Validates intents against security policies.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/orchestrator/intent_parser.py

"""Intent parsing for Orchestrata.

Parses and validates intents according to the formal grammar:
  intent := action ":" resource_type "/" resource_path [":" constraint]*

Based on DESIGN_DECISIONS.md specifications.
"""

import hashlib
import logging
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.parameter_extractor import ParameterExtractor

logger = logging.getLogger(__name__)


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-013"])
class Action(Enum):
    """Valid intent actions."""

    READ = "read"
    WRITE = "write"
    EXECUTE = "execute"
    CREATE = "create"
    DELETE = "delete"
    MODIFY = "modify"


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-004", "REQ-003-US-001-014"])
class ResourceType(Enum):
    """Valid resource types."""

    FILE = "file"
    COMMAND = "command"
    GIT = "git"
    NETWORK = "network"
    SYSTEM = "system"
    DIRECTORY = "directory"


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-023"])
@dataclass
class Intent:
    """Parsed intent with all components."""

    action: Action
    resource_type: ResourceType
    resource_path: str
    constraints: dict[str, str] = field(default_factory=dict)

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-023"])
    def to_string(self) -> str:
        """Convert intent to canonical string format."""
        base = f"{self.action.value}:{self.resource_type.value}/{self.resource_path}"
        if self.constraints:
            constraint_str = ":".join(f"{k}={v}" for k, v in self.constraints.items())
            return f"{base}:{constraint_str}"
        return base

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-012"])
    def to_prolog(self) -> str:
        """Convert intent to Prolog term."""
        constraints_list = ",".join(
            f"''{k}'='{v}'" for k, v in self.constraints.items()
        )
        return (
            f"intent('{self.action.value}', '{self.resource_type.value}', "
            f"'{self.resource_path}', [{constraints_list}])"
        )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-023"])
    def hash(self) -> str:
        """Generate hash for caching."""
        return hashlib.sha256(self.to_string().encode()).hexdigest()[:16]


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-034"])
class IntentParseError(Exception):
    """Error parsing intent string."""


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-011"])
class IntentParser:
    """Parser for intent strings.

    Grammar:
        intent := action ":" resource_type "/" resource_path (":" constraint)*
        action := "read" | "write" | "execute" | "create" | "delete" | "modify"
        resource_type := "file" | "command" | "git" | "network" | "system" | "directory"
        constraint := key "=" value
    """

    def __init__(self, extractor: ParameterExtractor | None = None) -> None:
        self._extractor = extractor or ParameterExtractor()

    # Pattern for parsing intent strings
    INTENT_PATTERN = re.compile(
        r"^(?P<action>\w+):(?P<type>\w+)/(?P<path>[^:]+)(?::(?P<constraints>.+))?$"
    )

    # Pattern for parsing constraints
    CONSTRAINT_PATTERN = re.compile(r"(?P<key>\w+)=(?P<value>[^:]+)")

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-011"])
    def parse(self, intent_string: str) -> Intent:
        """Parse an intent string.

        Args:
            intent_string: String in format "action:type/path:constraints"

        Returns:
            Parsed Intent object.

        Raises:
            IntentParseError: If string is malformed.
        """
        match = self.INTENT_PATTERN.match(intent_string.strip())
        if not match:
            raise IntentParseError(f"Invalid intent format: {intent_string}")

        # Parse action
        action_str = match.group("action")
        try:
            action = Action(action_str)
        except ValueError:
            raise IntentParseError(f"Invalid action: {action_str}")

        # Parse resource type
        type_str = match.group("type")
        try:
            resource_type = ResourceType(type_str)
        except ValueError:
            raise IntentParseError(f"Invalid resource type: {type_str}")

        # Parse path
        resource_path = match.group("path")
        if not resource_path:
            raise IntentParseError("Resource path cannot be empty")

        # Parse constraints
        constraints = {}
        constraints_str = match.group("constraints")
        if constraints_str:
            # Split on colons to get individual constraint strings
            constraint_parts = constraints_str.split(":")
            for constraint_part in constraint_parts:
                constraint_match = self.CONSTRAINT_PATTERN.match(constraint_part)
                if constraint_match:
                    key = constraint_match.group("key")
                    value = constraint_match.group("value")
                    constraints[key] = value

        return Intent(
            action=action,
            resource_type=resource_type,
            resource_path=resource_path,
            constraints=constraints,
        )

    @traces(
        spec="003",
        story="US-001",
        reqs=[
            "REQ-003-US-001-010",
            "REQ-003-US-001-021",
            "REQ-003-US-001-034",
            "REQ-003-US-001-035",
        ],
    )
    def try_parse(self, intent_string: str) -> Intent | None:
        """Try to parse an intent string, returning None on failure.

        Args:
            intent_string: String to parse.

        Returns:
            Parsed Intent or None.
        """
        try:
            return self.parse(intent_string)
        except IntentParseError:
            return None


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-003", "REQ-003-US-001-013"])
class IntentExtractor:
    """Extracts intents from tool calls.

    Maps Claude Code tools to their corresponding intents.
    """

    # Tool to action mapping
    TOOL_ACTIONS: dict[str, Action] = {
        "Read": Action.READ,
        "Write": Action.WRITE,
        "Edit": Action.MODIFY,
        "Bash": Action.EXECUTE,
        "Glob": Action.READ,
        "Grep": Action.READ,
        "Task": Action.EXECUTE,
        "TodoWrite": Action.WRITE,
        "WebFetch": Action.READ,
        "WebSearch": Action.READ,
    }

    def extract(self, tool_name: str, parameters: dict[str, Any]) -> Intent | None:
        """Extract intent from a tool call.

        Args:
            tool_name: Name of the tool being called.
            parameters: Tool parameters.

        Returns:
            Extracted Intent or None if tool is not mapped.
        """
        action = self.TOOL_ACTIONS.get(tool_name)
        if not action:
            return None

        # Determine resource type and path based on tool
        if tool_name in ("Read", "Write", "Edit"):
            return self._extract_file_intent(action, parameters)
        elif tool_name == "Bash":
            return self._extract_command_intent(parameters)
        elif tool_name in ("Glob", "Grep"):
            return self._extract_search_intent(action, tool_name, parameters)
        elif tool_name == "Task":
            return self._extract_task_intent(parameters)
        elif tool_name in ("WebFetch", "WebSearch"):
            return self._extract_network_intent(action, parameters)
        elif tool_name == "TodoWrite":
            return self._extract_system_intent(parameters)

        return None  # pragma: no cover

    def _extract_file_intent(
        self, action: Action, parameters: dict[str, Any]
    ) -> Intent:
        """Extract intent from file operations."""
        file_path = parameters.get("file_path", "unknown")
        return Intent(
            action=action,
            resource_type=ResourceType.FILE,
            resource_path=file_path,
        )

    def _extract_command_intent(self, parameters: dict[str, Any]) -> Intent:
        """Extract intent from Bash command."""
        command = parameters.get("command", "unknown")

        # Detect git commands
        if command.strip().startswith("git"):
            parts = command.strip().split()
            subcommand = parts[1] if len(parts) > 1 else "unknown"
            return Intent(
                action=Action.EXECUTE,
                resource_type=ResourceType.GIT,
                resource_path=command,
                constraints={"subcommand": subcommand},
            )

        return Intent(
            action=Action.EXECUTE,
            resource_type=ResourceType.COMMAND,
            resource_path=command,
        )

    def _extract_search_intent(
        self, action: Action, tool_name: str, parameters: dict[str, Any]
    ) -> Intent:
        """Extract intent from search operations."""
        pattern = parameters.get("pattern", "unknown")
        path = parameters.get("path", ".")

        return Intent(
            action=action,
            resource_type=(
                ResourceType.DIRECTORY if tool_name == "Glob" else ResourceType.FILE
            ),
            resource_path=path,
            constraints={"pattern": pattern},
        )

    def _extract_task_intent(self, parameters: dict[str, Any]) -> Intent:
        """Extract intent from Task agent spawn."""
        subagent_type = parameters.get("subagent_type", "unknown")
        prompt = parameters.get("prompt", "")[:50]  # First 50 chars

        return Intent(
            action=Action.EXECUTE,
            resource_type=ResourceType.SYSTEM,
            resource_path=f"agent/{subagent_type}",
            constraints={"prompt_preview": prompt},
        )

    def _extract_network_intent(
        self, action: Action, parameters: dict[str, Any]
    ) -> Intent:
        """Extract intent from network operations."""
        url = parameters.get("url", parameters.get("query", "unknown"))

        return Intent(
            action=action,
            resource_type=ResourceType.NETWORK,
            resource_path=url,
        )

    def _extract_system_intent(self, parameters: dict[str, Any]) -> Intent:
        """Extract intent from system operations."""
        todos = parameters.get("todos", [])

        return Intent(
            action=Action.WRITE,
            resource_type=ResourceType.SYSTEM,
            resource_path="todo_list",
            constraints={"count": str(len(todos))},
        )


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-002", "REQ-003-US-003-003"])
class IntentValidator:
    """Validates intents against security policies."""

    # Protected paths that cannot be written to
    PROTECTED_PATHS = {"/etc", "/usr", "/var", "/root", "/boot", "/sys", "/proc"}

    # Destructive commands that are always blocked
    DESTRUCTIVE_COMMANDS = {"rm -rf /", "dd if=/dev/zero", "mkfs", ":(){:|:&};:"}

    # Safe commands (whitelist)
    SAFE_COMMANDS = {"git", "ls", "cat", "grep", "find", "uv", "pytest", "python"}

    @traces(
        spec="003", story="US-003", reqs=["REQ-003-US-003-002", "REQ-003-US-003-003"]
    )
    def validate(self, intent: Intent) -> tuple[bool, str]:
        """Validate an intent against security policies.

        Args:
            intent: Intent to validate.

        Returns:
            Tuple of (is_valid, reason).
        """
        # Check protected paths
        if intent.action in (Action.WRITE, Action.DELETE, Action.MODIFY):
            for protected in self.PROTECTED_PATHS:
                if intent.resource_path.startswith(protected):
                    return False, f"Path {intent.resource_path} is protected"

        # Check destructive commands
        if intent.resource_type == ResourceType.COMMAND:
            for destructive in self.DESTRUCTIVE_COMMANDS:
                if destructive in intent.resource_path:
                    return False, f"Command contains destructive pattern: {destructive}"

        return True, "Valid"

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-007"])
    def is_safe_command(self, intent: Intent) -> bool:
        """Check if a command intent is in the safe list.

        Args:
            intent: Intent to check.

        Returns:
            True if command is safe.
        """
        if intent.resource_type != ResourceType.COMMAND:
            return True

        command = intent.resource_path.strip().split()[0]
        return command in self.SAFE_COMMANDS