Action Framework

The action framework provides the foundational classes for building reusable workflow components that follow GitHub Actions patterns.

causaliq-workflow uses the CausalIQActionProvider base class from causaliq-core for all action implementations. This ensures consistency across the CausalIQ ecosystem and provides standardised interfaces for action validation, execution, and result handling.

causaliq-core Integration

causaliq-workflow imports the following from causaliq-core:

• CausalIQActionProvider - Abstract base class for all workflow actions
• ActionInput - Type-safe input specification for action parameters
• ActionResult - Tuple type for action return values (status, metadata, objects)
• ActionValidationError - Exception for parameter validation failures
• ActionExecutionError - Exception for runtime execution failures

Core Classes

causaliq_core.CausalIQActionProvider

The base class for all action providers in the CausalIQ ecosystem.

CausalIQActionProvider

Base class for action providers that expose multiple workflow actions.

Action providers group related actions together. Each provider must declare which actions it supports via the supported_actions attribute. The 'action' input parameter selects which action to execute.

Providers can also declare supported_types for compress/decompress operations. Workflow uses these to build a registry of which provider handles compression for each data type.

Action patterns define how actions interact with workflow caches: - CREATE: Creates new entries (output + matrix required, input prohibited) - UPDATE: Modifies existing entries (input required, output/matrix prohibited) - AGGREGATE: Combines entries (input + output + matrix required)

Attributes:

• name (str) –

  Provider identifier for workflow 'uses' field.

• version (str) –

  Provider version string.

• description (str) –

  Human-readable description.

• author (str) –

  Provider author/maintainer.

• supported_actions (Set[str]) –

  Set of action names this provider supports.

• action_patterns (Dict[str, ActionPattern]) –

  Mapping of action names to their execution patterns.

• supported_types (Set[str]) –

  Set of data types this provider can compress.

• inputs (Dict[str, ActionInput]) –

  Input parameter specifications.

• outputs (Dict[str, str]) –

  Output name to description mapping.

Methods:

• compress –

  Compress textual content to compact binary blob for cache storage.

• decompress –

  Decompress compact binary blob back to textual content.

• run –

  Template method for action execution.

• validate_parameters –

  Validate action name and parameters.

compress

compress(data_type: str, content: str, cache: TokenCache) -> bytes

Compress textual content to compact binary blob for cache storage.

Converts interchange format strings (e.g., GraphML, JSON) to compact binary blobs suitable for SQLite storage. Subclasses should override this to support their data types. The data_type must be declared in supported_types.

Compression may use type-specific algorithms. For example, graphml compression parses the XML to extract edges as binary, and uses TokenCache to tokenise node names.

Parameters:

• data_type

  (str) –

  Type of data to compress (must be in supported_types).

• content

  (str) –

  Textual interchange format string.

• cache

  (TokenCache) –

  TokenCache instance for tokenisation.

Returns:

• bytes –

  Compact binary blob representation.

Raises:

• NotImplementedError –

  If the data type is not supported.

decompress

decompress(data_type: str, blob: bytes, cache: TokenCache) -> str

Decompress compact binary blob back to textual content.

Converts compact binary blobs from cache storage back to interchange format strings. Subclasses should override this to support their data types. The data_type must be declared in supported_types.

Parameters:

• data_type

  (str) –

  Type of data to decompress (must be in supported_types).

• blob

  (bytes) –

  Compact binary blob from cache.

• cache

  (TokenCache) –

  TokenCache instance for detokenisation.

Returns:

• str –

  Textual interchange format string.

Raises:

• NotImplementedError –

  If the data type is not supported.

run

run(
    action: str,
    parameters: Dict[str, Any],
    mode: str = "dry-run",
    context: Optional[Any] = None,
    logger: Optional[Any] = None,
) -> ActionResult

Template method for action execution.

This method implements the standard execution flow: 1. Validate parameters via validate_parameters() 2. If dry-run mode, return via _dry_run_result() 3. Otherwise, execute via _execute()

Subclasses should NOT override this method. Instead override: - validate_parameters() for parameter validation - _dry_run_result() for custom dry-run responses - _execute() for actual execution logic

Parameters:

• action

  (str) –

  Name of the action to execute (must be in supported_actions)

• parameters

  (Dict[str, Any]) –

  Dictionary of parameter values for the action

• mode

  (str, default: 'dry-run' ) –

  Execution mode ('dry-run', 'run', 'force', 'compare')

• context

  (Optional[Any], default: None ) –

  Workflow context for optimisation and intelligence

• logger

  (Optional[Any], default: None ) –

  Optional logger for task execution reporting

Returns:

• ActionResult –

  Tuple of (status, metadata, objects) where:

• ActionResult –
  • status: "success", "skipped", or "error"
• ActionResult –
  • metadata: Dictionary of execution metadata
• ActionResult –
  • objects: List of object dicts with keys:
  • type: Data type (e.g., "graphml", "json")
  • name: Object name identifier
  • content: String content

Raises:

• ActionExecutionError –

  If action execution fails

• ActionValidationError –

  If validation fails

validate_parameters

validate_parameters(action: str, parameters: Dict[str, Any]) -> None

Validate action name and parameters.

Override this method to add provider-specific parameter validation. Always call super().validate_parameters() first to check the action name is supported.

This method is called by run() before any execution, ensuring all validation errors are caught early (during workflow parsing).

Parameters:

• action

  (str) –

  Name of the action to validate.

• parameters

  (Dict[str, Any]) –

  Dictionary of parameter values to validate.

Raises:

• ActionValidationError –

  If action is not supported or parameters are invalid.

causaliq_core.ActionInput

ActionInput dataclass

ActionInput(
    name: str,
    description: str,
    required: bool = False,
    default: Any = None,
    type_hint: str = "Any",
)

Define action input specification.

Attributes:

• name (str) –

  Parameter name.

• description (str) –

  Human-readable description.

• required (bool) –

  Whether the parameter is required.

• default (Any) –

  Default value if not provided.

• type_hint (str) –

  Type hint string for documentation.

Exception Handling

causaliq_core.ActionExecutionError

ActionExecutionError

Raised when action execution fails.

causaliq_core.ActionValidationError

ActionValidationError

Raised when action input validation fails.

---

Quick Example

from typing import Any, Dict, Optional, Set, TYPE_CHECKING

from causaliq_core import (
    ActionInput,
    ActionResult,
    ActionValidationError,
    CausalIQActionProvider,
)

if TYPE_CHECKING:
    from causaliq_workflow.registry import WorkflowContext
    from causaliq_workflow.logger import WorkflowLogger


class MyStructureLearnerAction(CausalIQActionProvider):
    """Custom structure learning action."""

    name = "my-structure-learner"
    version = "1.0.0"
    description = "Custom causal structure learning implementation"
    author = "CausalIQ"

    supported_actions: Set[str] = {"learn_structure", "evaluate_graph"}
    supported_types: Set[str] = set()

    inputs = {
        "action": ActionInput(
            name="action",
            description="Action to perform",
            required=True,
            type_hint="str",
        ),
        "data_path": ActionInput(
            name="data_path",
            description="Path to input data file",
            required=True,
            type_hint="str",
        ),
        "alpha": ActionInput(
            name="alpha",
            description="Significance level",
            required=False,
            default=0.05,
            type_hint="float",
        ),
    }

    outputs = {
        "graph_path": "Path to output GraphML file",
        "node_count": "Number of nodes in learned graph",
        "edge_count": "Number of edges in learned graph",
    }

    def run(
        self,
        action: str,
        parameters: Dict[str, Any],
        mode: str = "dry-run",
        context: Optional[Any] = None,
        logger: Optional[Any] = None,
    ) -> ActionResult:
        """Execute the structure learning algorithm.

        Args:
            action: Name of action to execute (e.g., "learn_structure")
            parameters: Action parameters (data_path, output_dir, etc.)
            mode: Execution mode ("dry-run", "run", "compare")
            context: Workflow context with matrix values and cache
            logger: Optional logger for progress reporting

        Returns:
            ActionResult tuple (status, metadata, objects)
        """
        if action not in self.supported_actions:
            raise ActionValidationError(f"Unknown action: {action}")

        # Dry-run mode: return metadata only
        if mode == "dry-run":
            return ("skipped", {"dry_run": True}, [])

        # Your implementation here
        metadata = {
            "graph_path": "/path/to/output.graphml",
            "node_count": 5,
            "edge_count": 8,
        }

        # Objects list contains graph data
        objects = [
            {
                "type": "graphml",
                "name": "graph",
                "content": "<graphml>...</graphml>",
            }
        ]

        return ("success", metadata, objects)

Design Patterns

Action Implementation Guidelines

1. Inherit from CausalIQActionProvider - The base class from causaliq-core provides the standardised interface
2. Define comprehensive inputs - Use ActionInput for type safety and documentation
3. Document outputs clearly - Help users understand action results
4. Handle errors gracefully - Use ActionValidationError for parameter issues and ActionExecutionError for runtime failures
5. Follow semantic versioning - Enable workflow compatibility tracking
6. Return ActionResult tuples - Status string, metadata dict, and objects list
7. Support dry-run mode - Return early with skipped status for validation

Testing Your Actions

import pytest

from causaliq_core import ActionValidationError


# Test successful action execution.
def test_my_action_success() -> None:
    action = MyStructureLearnerAction()
    parameters = {
        "data_path": "/path/to/test_data.csv",
        "alpha": 0.05,
    }

    status, metadata, objects = action.run(
        "learn_structure", parameters, mode="run"
    )

    assert status == "success"
    assert "node_count" in metadata
    assert "edge_count" in metadata


# Test action validation with invalid action name.
def test_my_action_invalid_action() -> None:
    action = MyStructureLearnerAction()
    parameters = {"data_path": "/path/to/data.csv"}

    with pytest.raises(ActionValidationError):
        action.run("invalid_action", parameters, mode="run")


# Test dry-run mode returns skipped status.
def test_my_action_dry_run() -> None:
    action = MyStructureLearnerAction()
    parameters = {"data_path": "/path/to/data.csv"}

    status, metadata, objects = action.run(
        "learn_structure", parameters, mode="dry-run"
    )

    assert status == "skipped"
    assert metadata.get("dry_run") is True
    assert objects == []

---

← Back to API Overview | Next: Action Registry →