orchestrata_ultimate.orchestrator.intent_template

Reference › orchestrata_ultimate › orchestrator › intent_template

Reusable patterns for creating intents, traced to spec 003 story US-007. A template fixes an action and a target, declares a typed parameter schema, and supplies default parameters and constraints, so a caller supplies only the values that vary. No other reference page imports it; it is exercised by its own tests.

IntentTemplate.validate_params checks a parameter dictionary against parameter_schema: a name that is neither supplied nor defaulted is reported as missing, and a value of the wrong type is reported with the expected and actual type names. It returns a list of error strings rather than raising. instantiate merges defaults with the supplied parameters, builds IntentConstraint objects from the default constraints, takes resource_path out of the merged parameters, and returns a pending Intent with a fresh UUID. CompositeTemplate groups several templates with an execution_order (default sequential); its instantiate produces one intent per template and stamps them all with a shared correlation_id so they can be tracked as one request. TemplateRegistry stores templates and composites by id, with register, get, unregister, list_all, search by name, description or tag, search_by_action, search_by_target, compose (which returns None if any id is unknown) and get_composite.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentTemplate (dataclass)

Decorators: traces(spec='003', story='US-007', reqs=['REQ-003-US-007-001', 'REQ-003-US-007-007', 'REQ-003-US-007-011', 'REQ-003-US-…, dataclass · Line 37

Reusable intent template with parameter schema.

CompositeTemplate (dataclass)

Decorators: traces(spec='003', story='US-007', reqs=['REQ-003-US-007-008', 'REQ-003-US-007-021', 'REQ-003-US-007-033']), dataclass · Line 152

Multi-step template combining several IntentTemplates.

TemplateRegistry

Decorators: traces(spec='003', story='US-007', reqs=['REQ-003-US-007-004', 'REQ-003-US-007-005', 'REQ-003-US-007-008', 'REQ-003-US-… · Line 216

Registry for intent templates.

Source

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

"""Intent template system for reusable intent patterns.

Provides template registration, validation, instantiation,
and composition for complex multi-step operations.
"""

import logging
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.intent_types import (
    Intent,
    IntentAction,
    IntentConstraint,
    IntentPriority,
    IntentStatus,
    IntentTarget,
)

logger = logging.getLogger(__name__)


@traces(
    spec="003",
    story="US-007",
    reqs=[
        "REQ-003-US-007-001",
        "REQ-003-US-007-007",
        "REQ-003-US-007-011",
        "REQ-003-US-007-012",
    ],
)
@dataclass
class IntentTemplate:
    """Reusable intent template with parameter schema."""

    id: str
    name: str
    action: IntentAction
    target: IntentTarget
    parameter_schema: dict[str, type] = field(default_factory=dict)
    default_parameters: dict[str, Any] = field(default_factory=dict)
    default_constraints: list[dict[str, str]] = field(default_factory=list)
    description: str = ""
    version: str = "1.0.0"
    tags: list[str] = field(default_factory=list)

    @traces(
        spec="003",
        story="US-007",
        reqs=[
            "REQ-003-US-007-003",
            "REQ-003-US-007-025",
            "REQ-003-US-007-026",
            "REQ-003-US-007-028",
        ],
    )
    def validate_params(self, params: dict[str, Any]) -> list[str]:
        """Validate parameters against the schema. Returns list of errors."""
        errors = []
        for param_name, param_type in self.parameter_schema.items():
            if param_name not in params and param_name not in self.default_parameters:
                errors.append(f"Missing required parameter: {param_name}")
            elif param_name in params and not isinstance(
                params[param_name], param_type
            ):
                errors.append(
                    f"Parameter '{param_name}' expects {param_type.__name__}, "
                    f"got {type(params[param_name]).__name__}"
                )
        return errors

    @traces(
        spec="003",
        story="US-002",
        reqs=["REQ-003-US-002-022", "REQ-003-US-007-002", "REQ-003-US-007-013"],
    )
    def instantiate(
        self,
        params: dict[str, Any] | None = None,
        requester: str = "system",
        session_id: str | None = None,
        priority: IntentPriority = IntentPriority.NORMAL,
    ) -> Intent:
        """Create an Intent from this template with given parameters."""
        merged = dict(self.default_parameters)
        if params:
            merged.update(params)

        constraints = [
            IntentConstraint(
                key=c.get("key", ""),
                value=c.get("value", ""),
                operator=c.get("operator", "eq"),
            )
            for c in self.default_constraints
        ]

        resource_path = merged.pop("resource_path", "")

        now = datetime.now(UTC)
        return Intent(
            id=str(uuid.uuid4()),
            action=self.action,
            target=self.target,
            resource_path=resource_path,
            parameters=merged,
            constraints=constraints,
            requester=requester,
            status=IntentStatus.PENDING,
            priority=priority,
            created_at=now,
            updated_at=now,
            session_id=session_id,
            correlation_id=None,
            target_volk=None,
            error_message=None,
        )

    @traces(
        spec="003",
        story="US-007",
        reqs=["REQ-003-US-007-006", "REQ-003-US-007-018", "REQ-003-US-007-031"],
    )
    def to_dict(self) -> dict[str, Any]:
        """Serialize to dictionary."""
        return {
            "id": self.id,
            "name": self.name,
            "action": self.action.value,
            "target": self.target.value,
            "parameter_schema": {
                k: v.__name__ for k, v in self.parameter_schema.items()
            },
            "default_parameters": self.default_parameters,
            "default_constraints": self.default_constraints,
            "description": self.description,
            "version": self.version,
            "tags": self.tags,
        }


@traces(
    spec="003",
    story="US-007",
    reqs=["REQ-003-US-007-008", "REQ-003-US-007-021", "REQ-003-US-007-033"],
)
@dataclass
class CompositeTemplate:
    """Multi-step template combining several IntentTemplates."""

    id: str
    name: str
    templates: list[IntentTemplate] = field(default_factory=list)
    execution_order: str = "sequential"  # "sequential" or "parallel"
    description: str = ""

    @traces(
        spec="003",
        story="US-007",
        reqs=["REQ-003-US-007-008", "REQ-003-US-007-021"],
    )
    def instantiate(
        self,
        params_list: list[dict[str, Any]] | None = None,
        requester: str = "system",
        session_id: str | None = None,
    ) -> list[Intent]:
        """Instantiate all templates in this composite."""
        if params_list is None:
            params_list = [{}] * len(self.templates)

        intents = []
        correlation_id = str(uuid.uuid4())
        for i, template in enumerate(self.templates):
            params = params_list[i] if i < len(params_list) else {}
            intent = template.instantiate(
                params=params,
                requester=requester,
                session_id=session_id,
            )
            intent.correlation_id = correlation_id
            intents.append(intent)
        return intents

    @traces(
        spec="003",
        story="US-007",
        reqs=["REQ-003-US-007-006", "REQ-003-US-007-018", "REQ-003-US-007-031"],
    )
    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "name": self.name,
            "templates": [t.to_dict() for t in self.templates],
            "execution_order": self.execution_order,
            "description": self.description,
        }


@traces(
    spec="003",
    story="US-007",
    reqs=[
        "REQ-003-US-007-004",
        "REQ-003-US-007-005",
        "REQ-003-US-007-008",
        "REQ-003-US-007-014",
        "REQ-003-US-007-016",
        "REQ-003-US-007-021",
    ],
)
class TemplateRegistry:
    """Registry for intent templates."""

    def __init__(self) -> None:
        self._templates: dict[str, IntentTemplate] = {}
        self._composites: dict[str, CompositeTemplate] = {}

    @traces(
        spec="003",
        story="US-007",
        reqs=[
            "REQ-003-US-007-005",
            "REQ-003-US-007-015",
            "REQ-003-US-007-017",
            "REQ-003-US-007-019",
            "REQ-003-US-007-030",
        ],
    )
    def register(self, template: IntentTemplate) -> str:
        """Register a template and return its ID."""
        self._templates[template.id] = template
        return template.id

    @traces(spec="003", story="US-007", reqs=["REQ-003-US-007-014"])
    def get(self, template_id: str) -> IntentTemplate | None:
        """Get a template by ID."""
        return self._templates.get(template_id)

    @traces(spec="003", story="US-007", reqs=["REQ-003-US-007-014"])
    def unregister(self, template_id: str) -> bool:
        """Remove a template. Returns True if found."""
        if template_id in self._templates:
            del self._templates[template_id]
            return True
        return False

    @traces(spec="003", story="US-007", reqs=["REQ-003-US-007-014"])
    def list_all(self) -> list[IntentTemplate]:
        """List all registered templates."""
        return list(self._templates.values())

    @traces(
        spec="003",
        story="US-007",
        reqs=["REQ-003-US-007-004", "REQ-003-US-007-016", "REQ-003-US-007-029"],
    )
    def search(self, query: str) -> list[IntentTemplate]:
        """Search templates by name, description, or tags."""
        query_lower = query.lower()
        results = []
        for t in self._templates.values():
            if (
                query_lower in t.name.lower()
                or query_lower in t.description.lower()
                or any(query_lower in tag.lower() for tag in t.tags)
            ):
                results.append(t)
        return results

    @traces(spec="003", story="US-007", reqs=["REQ-003-US-007-004"])
    def search_by_action(self, action: IntentAction) -> list[IntentTemplate]:
        """Find templates matching an action."""
        return [t for t in self._templates.values() if t.action == action]

    @traces(spec="003", story="US-007", reqs=["REQ-003-US-007-004"])
    def search_by_target(self, target: IntentTarget) -> list[IntentTemplate]:
        """Find templates matching a target."""
        return [t for t in self._templates.values() if t.target == target]

    @traces(
        spec="003", story="US-007", reqs=["REQ-003-US-007-008", "REQ-003-US-007-021"]
    )
    def compose(
        self,
        template_ids: list[str],
        name: str = "composite",
        execution_order: str = "sequential",
    ) -> CompositeTemplate | None:
        """Compose multiple templates into a CompositeTemplate."""
        templates = []
        for tid in template_ids:
            t = self.get(tid)
            if t is None:
                return None
            templates.append(t)

        composite = CompositeTemplate(
            id=str(uuid.uuid4()),
            name=name,
            templates=templates,
            execution_order=execution_order,
        )
        self._composites[composite.id] = composite
        return composite

    @traces(spec="003", story="US-007", reqs=["REQ-003-US-007-008"])
    def get_composite(self, composite_id: str) -> CompositeTemplate | None:
        """Get a composite template by ID."""
        return self._composites.get(composite_id)

    @traces(spec="003", story="US-007", reqs=["REQ-003-US-007-024"])
    def count(self) -> int:
        """Total registered templates."""
        return len(self._templates)