Runtime 

Example

>>> result = await runtime.query(run_id, "progress")
>>> print(result)  # {"completed": 5, "total": 10}

get_signal_channel(workflow_id: str) → SignalChannel | None[source]

Get the SignalChannel for a running workflow.

Parameters:: workflow_id (str) – The run_id or workflow_id of the target workflow.
Returns:: The SignalChannel instance, or None if no workflow is active.
Return type:: SignalChannel | None

get_query_registry(workflow_id: str) → QueryRegistry | None[source]

Get the QueryRegistry for a running workflow.

Parameters:: workflow_id (str) – The run_id or workflow_id of the target workflow.
Returns:: The QueryRegistry instance, or None if no workflow is active.
Return type:: QueryRegistry | None

property progress_registry: ProgressRegistry

Registry for progress callbacks during workflow execution.

Register callbacks to receive ProgressUpdate events from nodes:

>>> runtime = LocalRuntime()
>>> runtime.progress_registry.register(lambda u: print(u.message))

property shutdown_coordinator: ShutdownCoordinator

Get or lazily create the ShutdownCoordinator for this runtime.

The coordinator is created on first access and the runtime’s own cleanup is automatically registered at priority 1 (drain).

Returns:: ShutdownCoordinator instance associated with this runtime.

Example

>>> runtime = LocalRuntime()
>>> runtime.shutdown_coordinator.register("db", pool.close, priority=3)
>>> await runtime.shutdown_coordinator.shutdown()

mark_externally_managed() → LocalRuntime[source]

Declare that an owning framework manages this runtime’s lifecycle.

Frameworks that hold a long-lived LocalRuntime across many execute() calls (e.g. DataFlow’s ModelRegistry, DataFlow instance, migration inspectors) should call this method immediately after construction. The runtime will then:

NOT emit the “use context manager” DeprecationWarning on execute() — that warning targets transient ad-hoc callers, not frameworks with their own shutdown protocol.
NOT register an atexit cleanup handler for the persistent event loop — the owner is responsible for calling close() at its own shutdown.

The caller MUST invoke close() (or route cleanup through ShutdownCoordinator) when the owning framework tears down.

Returns:

self to support fluent construction:

self.runtime = LocalRuntime().mark_externally_managed()

Return type:

LocalRuntime

See also

Issue #478 — the original DataFlow internal-warning leak that motivated this public opt-out.
close() — the cleanup call the owner is now responsible for.

close() → None[source]

Explicitly close the runtime and clean up resources.

This method should be called when you’re done with the runtime instance, especially in long-running applications. It closes the persistent event loop and releases all associated resources, including: - Event loop - Pending async tasks - Connection pools (indirectly, via loop closure)

Usage Patterns:

Pattern 1 - Try/Finally (Explicit Control):

>>> runtime = LocalRuntime()
>>> try:
...     results, run_id = runtime.execute(workflow)
... finally:
...     runtime.close()  # Always clean up

Pattern 2 - Long-Running Service:

>>> class MyService:
...     def __init__(self):
...         self.runtime = LocalRuntime()
...     def shutdown(self):
...         self.runtime.close()

Pattern 3 - Context Manager (Recommended):

>>> with LocalRuntime() as runtime:
...     results = runtime.execute(workflow)
# Automatic cleanup (close() called by __exit__)

Note

After calling close(), the runtime can still be used - a new event loop will be created automatically on the next execute() call. However, for best practices, create a new runtime instance instead of reusing after close().

Thread Safety:: Safe to call from any thread. Protected by internal lock.
Idempotency:: Safe to call multiple times. Subsequent calls are no-ops.

Examples

>>> runtime = LocalRuntime()
>>> runtime.execute(workflow1)
>>> runtime.close()                    # Clean up
>>> runtime.execute(workflow2)         # New loop created (OK but not recommended)
>>> runtime.close()                    # Safe to call again

See also

__enter__, __exit__: Context manager support
_cleanup_event_loop: Internal cleanup implementation

Return type:: None

acquire() → LocalRuntime[source]

Increment reference count. Call when sharing this runtime.

Returns self for fluent usage:: subsystem = Subsystem(runtime=shared_runtime.acquire())

Raises:: RuntimeError – If the runtime has already been fully closed (ref_count <= 0).
Return type:: LocalRuntime

release() → None[source]

Decrement reference count. Alias for close().

Actual cleanup happens when count reaches 0.

Return type:: None

property ref_count: int: Current reference count (for debugging/testing).

__del__(_warnings: ModuleType = <module 'warnings' from '/opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/warnings.py'>) → None[source]

Emit ResourceWarning if runtime was not properly closed.

Parameters:: _warnings (ModuleType)
Return type:: None

__enter__() → LocalRuntime[source]

Enter context manager, ensuring event loop is created.

This method is called when entering a ‘with’ statement. It: 1. Marks the runtime as context-managed 2. Eagerly creates the persistent event loop 3. Returns self for with-statement binding

Usage:

>>> with LocalRuntime() as runtime:
...     results, run_id = runtime.execute(workflow1)
...     results2, run_id2 = runtime.execute(workflow2)  # Same loop!
# Automatic cleanup on exit (__exit__ called)

Context Management Benefits:

Automatic cleanup (even on exceptions)
Clear resource lifetime
No atexit fallback needed
Pythonic and explicit

Returns:: Self for with-statement binding
Return type:: LocalRuntime

Examples

>>> with LocalRuntime(debug=True, enable_cycles=True) as runtime:
...     for workflow in workflow_list:
...         results, run_id = runtime.execute(workflow)
# All workflows share same event loop, then cleanup

Note

The event loop is created eagerly in __enter__, not lazily in execute(). This ensures consistent behavior regardless of execution paths.

See also

__exit__: Cleanup counterpart
close(): Explicit cleanup without context manager

__exit__(exc_type: type | None, exc_val: BaseException | None, exc_tb: Any | None) → None[source]

Exit context manager, cleaning up event loop.

This method is called when exiting a ‘with’ statement. It: 1. Cleans up the persistent event loop 2. Resets context-managed flag 3. Returns False to propagate exceptions

Parameters:

exc_type (type | None) – Exception type if exception occurred in with-block
exc_val (BaseException | None) – Exception value if exception occurred
exc_tb (Any | None) – Exception traceback if exception occurred

Returns:

False (do not suppress exceptions)

Return type:

None

Exception Handling:: This method does NOT suppress exceptions from the with-block. If an exception occurs during workflow execution, it will be propagated AFTER cleanup completes.

Examples

>>> with LocalRuntime() as runtime:
...     results = runtime.execute(workflow)
...     raise ValueError("Test")  # Exception raised
# __exit__ called with exc_type=ValueError
# Cleanup happens, then ValueError propagated

Note

Cleanup happens even if an exception occurred. The event loop is guaranteed to be cleaned up regardless of execution success.

See also

__enter__: Entry counterpart
_cleanup_event_loop: Actual cleanup implementation

get_validation_metrics() → Dict[str, Any][source]

Get validation performance metrics for the runtime.

Returns:: Dictionary containing performance and security metrics
Return type:: Dict[str, Any]

reset_validation_metrics() → None[source]

Reset validation metrics collector.

Return type:: None

async execute_node_with_enterprise_features(node, node_id: str, inputs: dict[str, Any], **execution_kwargs) → Any[source]

Execute a node with automatic enterprise feature integration.

This method automatically applies: - Resource limit enforcement - Retry policies with circuit breaker integration - Performance monitoring - Error handling and recovery

Parameters:

node – Node instance to execute
node_id (str) – Node identifier for tracking
inputs (dict[str, Any]) – Input parameters for node execution
**execution_kwargs – Additional execution parameters

Returns:

Node execution result

Raises:

Various enterprise exceptions based on configured policies –

Return type:

execute_node_with_enterprise_features_sync(node, node_id: str, inputs: dict[str, Any], **execution_kwargs) → Any[source]

Execute a node with automatic enterprise features (synchronous version).

This is the sync wrapper for enterprise features that can be called from the CyclicWorkflowExecutor which runs in sync context.

Parameters:

node_id (str)
inputs (dict[str, Any])

Return type:

get_resource_metrics() → dict[str, Any] | None[source]

Get current resource usage metrics from the resource enforcer.

Returns:: Dict containing resource metrics or None if no resource enforcer
Return type:: dict[str, Any] | None

get_execution_metrics(run_id: str) → dict[str, Any] | None[source]

Get execution metrics for a specific run ID.

Parameters:: run_id (str) – The run ID to get metrics for
Returns:: Dict containing execution metrics or None if not available
Return type:: dict[str, Any] | None

get_retry_policy_engine()[source]

Get the retry policy engine instance.

Returns:: RetryPolicyEngine instance or None if not initialized

get_retry_analytics()[source]

Get comprehensive retry analytics and metrics.

Returns:: Dictionary containing retry analytics or None if retry engine not enabled

get_retry_metrics_summary()[source]

Get summary of retry metrics.

Returns:: Dictionary containing retry metrics summary or None if not available

get_strategy_effectiveness()[source]

Get effectiveness statistics for all retry strategies.

Returns:: Dictionary mapping strategy names to effectiveness stats

register_retry_strategy(name: str, strategy)[source]

Register a custom retry strategy.

Parameters:

name (str) – Strategy name for identification
strategy – RetryStrategy instance

register_retry_strategy_for_exception(exception_type: type, strategy)[source]

Register strategy for specific exception type.

Parameters:

exception_type (type) – Exception type to handle
strategy – RetryStrategy to use for this exception type

add_retriable_exception(exception_type: type)[source]

Add an exception type to the retriable exceptions list.

Parameters:: exception_type (type) – Exception type to mark as retriable

add_non_retriable_exception(exception_type: type)[source]

Add an exception type to the non-retriable exceptions list.

Parameters:: exception_type (type) – Exception type to mark as non-retriable

reset_retry_metrics()[source]: Reset all retry metrics and analytics data.

get_retry_configuration()[source]

Get current retry policy configuration.

Returns:: Dictionary containing current retry configuration

get_execution_plan_cached(workflow: Workflow, switch_results: Dict[str, Dict[str, Any]]) → List[str] | tuple[str, ...][source]

Get execution plan with caching for improved performance.

Parameters:

workflow (Workflow) – Workflow to create execution plan for
switch_results (Dict[str, Dict[str, Any]]) – Results from SwitchNode execution

Returns:

Cached or newly computed execution plan

Return type:

List[str] | tuple[str, …]

get_execution_analytics() → Dict[str, Any][source]

Get comprehensive execution analytics for monitoring and optimization.

Returns:: Dictionary containing detailed analytics data
Return type:: Dict[str, Any]

record_execution_performance(workflow: Workflow, execution_time: float, nodes_executed: int, used_conditional: bool, performance_improvement: float = 0.0)[source]

Record execution performance for analytics.

Parameters:

workflow (Workflow) – Workflow that was executed
execution_time (float) – Total execution time in seconds
nodes_executed (int) – Number of nodes actually executed
used_conditional (bool) – Whether conditional execution was used
performance_improvement (float) – Performance improvement percentage (0.0-1.0)

clear_analytics_data(keep_patterns: bool = True)[source]

Clear analytics data for fresh monitoring.

Parameters:: keep_patterns (bool) – Whether to preserve execution patterns

get_health_diagnostics() → Dict[str, Any][source]

Get health diagnostics for monitoring system health.

Returns:: Dictionary containing health check results
Return type:: Dict[str, Any]

optimize_runtime_performance() → Dict[str, Any][source]

Optimize runtime performance based on analytics data.

Returns:: Dictionary describing optimizations applied
Return type:: Dict[str, Any]

get_performance_report() → Dict[str, Any][source]

Get performance monitoring report.

Returns:: Performance statistics and recommendations
Return type:: Dict[str, Any]

generate_compatibility_report(workflow: Workflow) → Dict[str, Any][source]

Generate compatibility report for a workflow.

Parameters:: workflow (Workflow) – Workflow to analyze
Returns:: Compatibility report dictionary
Return type:: Dict[str, Any]

get_compatibility_report_markdown(workflow: Workflow) → str[source]

Generate compatibility report in markdown format.

Parameters:: workflow (Workflow) – Workflow to analyze
Returns:: Markdown formatted report
Return type:: str

set_performance_monitoring(enabled: bool) → None[source]

Enable or disable performance monitoring.

Parameters:: enabled (bool) – Whether to enable performance monitoring
Return type:: None

set_automatic_mode_switching(enabled: bool) → None[source]

Enable or disable automatic mode switching based on performance.

Parameters:: enabled (bool) – Whether to enable automatic switching
Return type:: None

set_compatibility_reporting(enabled: bool) → None[source]

Enable or disable compatibility reporting.

Parameters:: enabled (bool) – Whether to enable compatibility reporting
Return type:: None

get_execution_path_debug_info() → Dict[str, Any][source]

Get detailed debug information about execution paths.

Returns:: Debug information including execution decisions and paths
Return type:: Dict[str, Any]

async start_persistent_mode() → None[source]

Start runtime in persistent mode for long-running applications.

This enables connection pool sharing, resource coordination, and enterprise monitoring features. Only available when persistent_mode=True.

Raises:: RuntimeError – If persistent mode is not enabled or startup fails.
Return type:: None

async shutdown_gracefully(timeout: int = 30) → None[source]

Gracefully shutdown runtime with connection drain and cleanup.

Parameters:: timeout (int) – Maximum time to wait for shutdown completion (seconds).
Return type:: None

async get_shared_connection_pool(pool_name: str, pool_config: Dict[str, Any]) → Any[source]

Get shared connection pool for database operations.

Parameters:

pool_name (str) – Name for the connection pool
pool_config (Dict[str, Any]) – Pool configuration parameters

Returns:

Connection pool instance

Raises:

RuntimeError – If persistent mode is not started
ValueError – If pool configuration is invalid

Return type:

can_execute_workflow() → bool[source]

Check if runtime can execute another workflow based on limits.

Returns:: True if workflow can be executed, False otherwise.
Return type:: bool

get_runtime_metrics() → Dict[str, Any][source]

Get comprehensive runtime health and performance metrics.

Returns:: Dictionary containing runtime metrics across all categories.
Return type:: Dict[str, Any]

get_health_status() → Dict[str, Any][source]

Get current health status of the runtime.

Returns:: Health status information including overall status and details.
Return type:: Dict[str, Any]

property connection_pool_manager: Access the connection pool manager.

async cleanup()[source]: Clean up runtime resources.

__repr__() → str

Get string representation of runtime instance.

Returns:: String representation with key configuration
Return type:: str

Example

>>> runtime = LocalRuntime(debug=True, enable_cycles=True)
>>> print(repr(runtime))
<LocalRuntime(id=runtime_..., debug=True, cycles=True, async=False)>

validate_workflow(workflow: Workflow) → list[str]

Validate a workflow before execution.

Performs comprehensive validation including: - Basic workflow structure validation - Disconnected node detection - Required parameter checking - Connection validation - Performance warnings for large workflows

EXTRACTED FROM: LocalRuntime.validate_workflow() (lines 2054-2119) SHARED LOGIC: 100% - Pure validation with no I/O

Parameters:: workflow (Workflow) – Workflow to validate
Returns:: List of validation warnings (empty if valid)
Raises:: WorkflowValidationError – If workflow is invalid
Return type:: list[str]

Implementation Notes:

Calls workflow.validate() for basic structure
Checks for disconnected nodes in multi-node workflows
Validates required parameters are provided or connected
Warns about performance implications for large workflows
All logic is synchronous and shared between runtimes

Examples

# Basic validation
runtime = LocalRuntime()
warnings = runtime.validate_workflow(workflow)
if warnings:
    for warning in warnings:
        print(f"Warning: {warning}")

# Validation with error handling
try:
    warnings = runtime.validate_workflow(workflow)
    if not warnings:
        print("Workflow is valid")
except WorkflowValidationError as e:
    print(f"Invalid workflow: {e}")

logger

Validation capabilities for workflow runtimes.

This mixin provides comprehensive validation logic for workflows, nodes, parameters, and connections. All methods are 100% shared between sync and async runtimes as they perform pure validation with no I/O operations.

Shared Logic (100%):

All validation methods are pure logic with no sync/async variants needed. They validate data structures and configurations without performing any I/O or execution.

Dependencies:

Requires workflow.graph attribute (from Workflow)
Requires self.logger attribute (from BaseRuntime)
Requires self.debug attribute (from BaseRuntime)
No dependencies on other mixins

Usage:

class LocalRuntime(BaseRuntime, ValidationMixin):: # Inherits all 5 validation methods pass
class AsyncLocalRuntime(BaseRuntime, ValidationMixin):: # Inherits same 5 validation methods pass

Examples

# Validation is called automatically during execution runtime = LocalRuntime() runtime.execute(workflow) # Calls validation methods

# Can also validate manually warnings = runtime.validate_workflow(workflow) runtime._validate_connection_contracts(workflow, node_id, inputs, outputs)

See also

BaseRuntime: Base runtime class
ADR-XXX: Runtime Refactoring for Feature Parity

Version:: Added in: v0.10.0 Part of: Runtime parity remediation

debug

enable_cycles

enable_monitoring

conditional_execution

Characteristics:

Executes nodes sequentially
Good for debugging and development
Lower overhead for simple workflows
Direct access to local filesystem

Example Usage:

from kailash.runtime import LocalRuntime
from kailash import Workflow

workflow = Workflow("local_example")

# Add nodes...

# Execute with local runtime
with LocalRuntime() as runtime:
    results = runtime.execute(workflow)

Connection Parameter Validation (v0.8.4+)

Configure connection parameter validation modes for enhanced security:

from kailash.runtime import LocalRuntime

# Development mode - strict validation
runtime = LocalRuntime(
    connection_validation="strict"    # Block invalid connection parameters
)

# Production mode - warnings only (default)
runtime = LocalRuntime(
    connection_validation="warn"      # Log warnings but continue
)

# Performance mode - no validation
runtime = LocalRuntime(
    connection_validation="off"       # No validation overhead
)

Connection Validation Modes:

off: No connection parameter validation
warn: Log warnings for parameter issues but continue execution (default)
strict: Block execution for invalid connection parameters

Performance Characteristics:

Validation Overhead: <1ms per workflow execution
Connection Security: Prevents parameter injection through workflow connections
Enterprise Ready: Comprehensive metrics collection and alerting

Validation Features:

from kailash.runtime import LocalRuntime

# Create runtime with validation
with LocalRuntime(connection_validation="strict") as runtime:
    # Execute workflow - validation runs automatically
    results, run_id = runtime.execute(workflow, parameters=params)

    # Access validation metrics
    metrics = runtime.get_validation_metrics()
    print(f"Performance: {metrics['performance_summary']}")
    print(f"Security: {metrics['security_report']}")

    # Reset metrics for next run
    runtime.reset_validation_metrics()

AsyncLocalRuntime 

Asynchronous runtime for I/O-bound operations.

class kailash.runtime.async_local.AsyncLocalRuntime(resource_registry: ResourceRegistry | None = None, max_concurrent_nodes: int = 10, enable_analysis: bool = True, enable_profiling: bool = True, thread_pool_size: int = 4, execution_timeout: int | None = None, **kwargs)[source]

Bases: LocalRuntime

Async-optimized runtime for Kailash workflows.

Extends LocalRuntime with advanced async execution capabilities while inheriting all enterprise features through shared mixin architecture.

Inherits from:

LocalRuntime: Provides 100% feature parity with sync runtime: ├─ BaseRuntime: Core runtime foundation and configuration ├─ CycleExecutionMixin: Cyclic workflow execution delegation ├─ ValidationMixin: Workflow validation and contract checking └─ ConditionalExecutionMixin: Conditional execution and branching logic

Async-Specific Extensions:

WorkflowAnalyzer: Analyzes workflows for optimization opportunities
ExecutionContext: Async context with integrated resource access
Level-based parallel execution: Executes independent nodes concurrently
Semaphore-based concurrency control: Limits concurrent node execution
Thread pool for sync nodes: Executes sync nodes without blocking async loop
Advanced performance tracking: Detailed metrics collection

Execution Strategies:

The runtime automatically selects the optimal execution strategy: - Pure async: All nodes are async (fastest, full concurrency) - Mixed: Combination of sync and async nodes (balanced) - Sync in thread pool: All sync nodes (compatibility mode)

Example

from kailash.resources import ResourceRegistry, DatabasePoolFactory
from kailash.runtime.async_local import AsyncLocalRuntime

# Setup resources
registry = ResourceRegistry()
registry.register_factory("db", DatabasePoolFactory(...))

# Create async runtime
runtime = AsyncLocalRuntime(
    resource_registry=registry,
    max_concurrent_nodes=10,
    enable_analysis=True
)

# Execute workflow
result = await runtime.execute_workflow_async(workflow, inputs)

Parameters:

resource_registry (ResourceRegistry | None)
max_concurrent_nodes (int)
enable_analysis (bool)
enable_profiling (bool)
thread_pool_size (int)
execution_timeout (int | None)

__init__(resource_registry: ResourceRegistry | None = None, max_concurrent_nodes: int = 10, enable_analysis: bool = True, enable_profiling: bool = True, thread_pool_size: int = 4, execution_timeout: int | None = None, **kwargs)[source]

Initialize AsyncLocalRuntime.

Parameters:

resource_registry (ResourceRegistry | None) – Optional ResourceRegistry for resource management
max_concurrent_nodes (int) – Maximum number of nodes to execute concurrently
enable_analysis (bool) – Whether to analyze workflows for optimization
enable_profiling (bool) – Whether to collect detailed performance metrics
thread_pool_size (int) – Size of thread pool for sync node execution
execution_timeout (int | None) – Workflow execution timeout in seconds (default: 300 or DATAFLOW_EXECUTION_TIMEOUT env var)
**kwargs – Additional arguments passed to LocalRuntime

property execution_semaphore: Semaphore

Lazily create execution semaphore when accessed.

P0-7 FIX: Semaphore must be created in async context (with running event loop). Creating in __init__ causes race conditions in FastAPI/Docker deployments.

execute(workflow, task_manager: TaskManager | None = None, parameters: Dict[str, Any] | None = None, cancellation_token: Any = None, search_attributes: Dict[str, Any] | None = None, *, soft_time_limit: float | None = None, time_limit: float | None = None, **kwargs: Any) → Tuple[Dict[str, Any], str | None][source]

Execute workflow without creating threads (Docker-safe).

This override prevents the parent’s threading-based execution that causes Docker file descriptor issues. Uses pure async execution via asyncio.run or returns the async task if already in an event loop.

Parameters:

workflow – Workflow to execute
task_manager (TaskManager | None) – Optional task manager for tracking
parameters (Dict[str, Any] | None) – Input parameters for the workflow
soft_time_limit (float | None) – Optional advisory deadline in seconds (#912). Raises SoftTimeLimitExceeded when reached; user code MAY catch and exit cleanly.
time_limit (float | None) – Optional unconditional kill deadline in seconds (#912). Raises HardTimeLimitExceeded after time_limit + grace regardless of acknowledgement.
cancellation_token (Any)
search_attributes (Dict[str, Any] | None)
kwargs (Any)

Returns:

Tuple of (results dict, run_id)

Raises:

RuntimeError – If called from async context (use execute_workflow_async instead)
SoftTimeLimitExceeded – If soft_time_limit elapses.
HardTimeLimitExceeded – If time_limit + grace elapses.

Return type:

Tuple[Dict[str, Any], str | None]

Time-Limit Example:

from kailash.runtime.async_local import AsyncLocalRuntime
from kailash.sdk_exceptions import SoftTimeLimitExceeded

runtime = AsyncLocalRuntime()
try:
    results, run_id = runtime.execute(
        workflow.build(),
        soft_time_limit=2.0,
        time_limit=5.0,
    )
except SoftTimeLimitExceeded:
    ...  # save partial work, exit cleanly

async execute_async(workflow, task_manager: TaskManager | None = None, parameters: Dict[str, Any] | None = None, cancellation_token: Any = None, execution_tracker: Any = None, search_attributes: Dict[str, Any] | None = None, **kwargs: Any) → Tuple[Dict[str, Any], str | None][source]

Execute workflow asynchronously (for LocalRuntime compatibility).

This method provides compatibility with LocalRuntime’s execute_async() interface while using AsyncLocalRuntime’s execution engine.

Parameters:

workflow – Workflow to execute
task_manager (TaskManager | None) – Optional task manager for tracking
parameters (Dict[str, Any] | None) – Input parameters for the workflow
cancellation_token (Any)
execution_tracker (Any)
search_attributes (Dict[str, Any] | None)
kwargs (Any)

Returns:

Tuple of (results dict, run_id)

Return type:

Tuple[Dict[str, Any], str | None]

async execute_workflow_async(workflow, inputs: Dict[str, Any], context: ExecutionContext | None = None, *, idempotency_key: str | None = None, force_resume_with_drift: bool = False, soft_time_limit: float | None = None, time_limit: float | None = None) → Tuple[Dict[str, Any], str][source]

Execute workflow with native async support and production safeguards.

P0 Component 1 Features: - Timeout protection (configurable via execution_timeout) - Connection lifecycle management - Task cancellation on timeout - Cleanup guarantees

This method provides first-class async execution with: - Concurrent node execution where dependencies allow - Integrated resource management - Performance optimization based on workflow analysis - Advanced error handling and recovery

Parameters:

workflow – Workflow to execute
inputs (Dict[str, Any]) – Input data for the workflow
context (ExecutionContext | None) – Optional execution context
soft_time_limit (float | None) – Optional advisory deadline in seconds (#912). Raises SoftTimeLimitExceeded when reached; user code MAY catch and exit cleanly.
time_limit (float | None) – Optional unconditional kill deadline in seconds (#912). Raises HardTimeLimitExceeded after time_limit + grace.
idempotency_key (str | None)
force_resume_with_drift (bool)

Returns:

Tuple of (results dict, run_id) - For compatibility with tests - results: Dictionary mapping node_id -> node output - run_id: Unique execution identifier

Return type:

Tuple[Dict[str, Any], str]

Note

Returns tuple for compatibility with LocalRuntime.execute() pattern. Existing tests may expect dict - use results, run_id = await execute_workflow_async()

Raises:

asyncio.TimeoutError – If execution exceeds configured timeout
WorkflowExecutionError – If execution fails
SoftTimeLimitExceeded – If soft_time_limit elapses.
HardTimeLimitExceeded – If time_limit + grace elapses.

Parameters:

inputs (Dict[str, Any])
context (ExecutionContext | None)
idempotency_key (str | None)
force_resume_with_drift (bool)
soft_time_limit (float | None)
time_limit (float | None)

Return type:

Tuple[Dict[str, Any], str]

Time-Limit Example (async):

from kailash.runtime.async_local import AsyncLocalRuntime
from kailash.sdk_exceptions import SoftTimeLimitExceeded

runtime = AsyncLocalRuntime()
try:
    results, run_id = await runtime.execute_workflow_async(
        workflow.build(),
        inputs={},
        soft_time_limit=2.0,
        time_limit=5.0,
    )
except SoftTimeLimitExceeded:
    ...  # save partial work, exit cleanly

async cleanup() → None[source]

Clean up runtime resources (idempotent).

P0-8 FIX: Enhanced cleanup with proper resource management. Safe to call multiple times - tracks cleanup state.

Recommended usage with a web framework lifespan:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    runtime = AsyncLocalRuntime()
    yield {"runtime": runtime}
    # Shutdown
    await runtime.cleanup()

app = FastAPI(lifespan=lifespan)

Return type:: None

close() → None[source]

Synchronous close that properly cleans up ALL async resources.

Overrides LocalRuntime.close() to also handle thread pool, resource registry, semaphore, and SQL connection pools.

Reference-count aware: decrements _ref_count. Actual cleanup only happens when _ref_count reaches 0.

Return type:: None

__del__(_warnings: ModuleType = <module 'warnings' from '/opt/hostedtoolcache/Python/3.11.15/x64/lib/python3.11/warnings.py'>) → None[source]

Emit ResourceWarning if runtime was not properly closed.

Parameters:: _warnings (ModuleType)
Return type:: None

async __aenter__() → AsyncLocalRuntime[source]

Async context manager entry.

Usage:

async with AsyncLocalRuntime() as runtime:: results = await runtime.execute_workflow_async(workflow, inputs)

Return type:: AsyncLocalRuntime

__enter__() → LocalRuntime

Enter context manager, ensuring event loop is created.

This method is called when entering a ‘with’ statement. It: 1. Marks the runtime as context-managed 2. Eagerly creates the persistent event loop 3. Returns self for with-statement binding

Usage:

>>> with LocalRuntime() as runtime:
...     results, run_id = runtime.execute(workflow1)
...     results2, run_id2 = runtime.execute(workflow2)  # Same loop!
# Automatic cleanup on exit (__exit__ called)

Context Management Benefits:

Automatic cleanup (even on exceptions)
Clear resource lifetime
No atexit fallback needed
Pythonic and explicit

Returns:: Self for with-statement binding
Return type:: LocalRuntime

Examples

>>> with LocalRuntime(debug=True, enable_cycles=True) as runtime:
...     for workflow in workflow_list:
...         results, run_id = runtime.execute(workflow)
# All workflows share same event loop, then cleanup

Note

The event loop is created eagerly in __enter__, not lazily in execute(). This ensures consistent behavior regardless of execution paths.

See also

__exit__: Cleanup counterpart
close(): Explicit cleanup without context manager

__exit__(exc_type: type | None, exc_val: BaseException | None, exc_tb: Any | None) → None

Exit context manager, cleaning up event loop.

This method is called when exiting a ‘with’ statement. It: 1. Cleans up the persistent event loop 2. Resets context-managed flag 3. Returns False to propagate exceptions

Parameters:

exc_type (type | None) – Exception type if exception occurred in with-block
exc_val (BaseException | None) – Exception value if exception occurred
exc_tb (Any | None) – Exception traceback if exception occurred

Returns:

False (do not suppress exceptions)

Return type:

None

Exception Handling:: This method does NOT suppress exceptions from the with-block. If an exception occurs during workflow execution, it will be propagated AFTER cleanup completes.

Examples

>>> with LocalRuntime() as runtime:
...     results = runtime.execute(workflow)
...     raise ValueError("Test")  # Exception raised
# __exit__ called with exc_type=ValueError
# Cleanup happens, then ValueError propagated

Note

Cleanup happens even if an exception occurred. The event loop is guaranteed to be cleaned up regardless of execution success.

See also

__enter__: Entry counterpart
_cleanup_event_loop: Actual cleanup implementation

__repr__() → str

Get string representation of runtime instance.

Returns:: String representation with key configuration
Return type:: str

Example

>>> runtime = LocalRuntime(debug=True, enable_cycles=True)
>>> print(repr(runtime))
<LocalRuntime(id=runtime_..., debug=True, cycles=True, async=False)>

acquire() → LocalRuntime

Increment reference count. Call when sharing this runtime.

Returns self for fluent usage:: subsystem = Subsystem(runtime=shared_runtime.acquire())

Raises:: RuntimeError – If the runtime has already been fully closed (ref_count <= 0).
Return type:: LocalRuntime

add_non_retriable_exception(exception_type: type)

Add an exception type to the non-retriable exceptions list.

Parameters:: exception_type (type) – Exception type to mark as non-retriable

add_retriable_exception(exception_type: type)

Add an exception type to the retriable exceptions list.

Parameters:: exception_type (type) – Exception type to mark as retriable

can_execute_workflow() → bool

Check if runtime can execute another workflow based on limits.

Returns:: True if workflow can be executed, False otherwise.
Return type:: bool

clear_analytics_data(keep_patterns: bool = True)

Clear analytics data for fresh monitoring.

Parameters:: keep_patterns (bool) – Whether to preserve execution patterns

property connection_pool_manager: Access the connection pool manager.

async execute_node_with_enterprise_features(node, node_id: str, inputs: dict[str, Any], **execution_kwargs) → Any

Execute a node with automatic enterprise feature integration.

This method automatically applies: - Resource limit enforcement - Retry policies with circuit breaker integration - Performance monitoring - Error handling and recovery

Parameters:

node – Node instance to execute
node_id (str) – Node identifier for tracking
inputs (dict[str, Any]) – Input parameters for node execution
**execution_kwargs – Additional execution parameters

Returns:

Node execution result

Raises:

Various enterprise exceptions based on configured policies –

Return type:

execute_node_with_enterprise_features_sync(node, node_id: str, inputs: dict[str, Any], **execution_kwargs) → Any

Execute a node with automatic enterprise features (synchronous version).

This is the sync wrapper for enterprise features that can be called from the CyclicWorkflowExecutor which runs in sync context.

Parameters:

node_id (str)
inputs (dict[str, Any])

Return type:

generate_compatibility_report(workflow: Workflow) → Dict[str, Any]

Generate compatibility report for a workflow.

Parameters:: workflow (Workflow) – Workflow to analyze
Returns:: Compatibility report dictionary
Return type:: Dict[str, Any]

get_compatibility_report_markdown(workflow: Workflow) → str

Generate compatibility report in markdown format.

Parameters:: workflow (Workflow) – Workflow to analyze
Returns:: Markdown formatted report
Return type:: str

get_execution_analytics() → Dict[str, Any]

Get comprehensive execution analytics for monitoring and optimization.

Returns:: Dictionary containing detailed analytics data
Return type:: Dict[str, Any]

get_execution_metrics(run_id: str) → dict[str, Any] | None

Get execution metrics for a specific run ID.

Parameters:: run_id (str) – The run ID to get metrics for
Returns:: Dict containing execution metrics or None if not available
Return type:: dict[str, Any] | None

get_execution_path_debug_info() → Dict[str, Any]

Get detailed debug information about execution paths.

Returns:: Debug information including execution decisions and paths
Return type:: Dict[str, Any]

get_execution_plan_cached(workflow: Workflow, switch_results: Dict[str, Dict[str, Any]]) → List[str] | tuple[str, ...]

Get execution plan with caching for improved performance.

Parameters:

workflow (Workflow) – Workflow to create execution plan for
switch_results (Dict[str, Dict[str, Any]]) – Results from SwitchNode execution

Returns:

Cached or newly computed execution plan

Return type:

List[str] | tuple[str, …]

get_health_diagnostics() → Dict[str, Any]

Get health diagnostics for monitoring system health.

Returns:: Dictionary containing health check results
Return type:: Dict[str, Any]

get_health_status() → Dict[str, Any]

Get current health status of the runtime.

Returns:: Health status information including overall status and details.
Return type:: Dict[str, Any]

get_performance_report() → Dict[str, Any]

Get performance monitoring report.

Returns:: Performance statistics and recommendations
Return type:: Dict[str, Any]

get_query_registry(workflow_id: str) → QueryRegistry | None

Get the QueryRegistry for a running workflow.

Parameters:: workflow_id (str) – The run_id or workflow_id of the target workflow.
Returns:: The QueryRegistry instance, or None if no workflow is active.
Return type:: QueryRegistry | None

get_resource_metrics() → dict[str, Any] | None

Get current resource usage metrics from the resource enforcer.

Returns:: Dict containing resource metrics or None if no resource enforcer
Return type:: dict[str, Any] | None

get_retry_analytics()

Get comprehensive retry analytics and metrics.

Returns:: Dictionary containing retry analytics or None if retry engine not enabled

get_retry_configuration()

Get current retry policy configuration.

Returns:: Dictionary containing current retry configuration

get_retry_metrics_summary()

Get summary of retry metrics.

Returns:: Dictionary containing retry metrics summary or None if not available

get_retry_policy_engine()

Get the retry policy engine instance.

Returns:: RetryPolicyEngine instance or None if not initialized

get_runtime_metrics() → Dict[str, Any]

Get comprehensive runtime health and performance metrics.

Returns:: Dictionary containing runtime metrics across all categories.
Return type:: Dict[str, Any]

async get_shared_connection_pool(pool_name: str, pool_config: Dict[str, Any]) → Any

Get shared connection pool for database operations.

Parameters:

pool_name (str) – Name for the connection pool
pool_config (Dict[str, Any]) – Pool configuration parameters

Returns:

Connection pool instance

Raises:

RuntimeError – If persistent mode is not started
ValueError – If pool configuration is invalid

Return type:

get_signal_channel(workflow_id: str) → SignalChannel | None

Get the SignalChannel for a running workflow.

Parameters:: workflow_id (str) – The run_id or workflow_id of the target workflow.
Returns:: The SignalChannel instance, or None if no workflow is active.
Return type:: SignalChannel | None

get_strategy_effectiveness()

Get effectiveness statistics for all retry strategies.

Returns:: Dictionary mapping strategy names to effectiveness stats

get_validation_metrics() → Dict[str, Any]

Get validation performance metrics for the runtime.

Returns:: Dictionary containing performance and security metrics
Return type:: Dict[str, Any]

mark_externally_managed() → LocalRuntime

Declare that an owning framework manages this runtime’s lifecycle.

Frameworks that hold a long-lived LocalRuntime across many execute() calls (e.g. DataFlow’s ModelRegistry, DataFlow instance, migration inspectors) should call this method immediately after construction. The runtime will then:

NOT emit the “use context manager” DeprecationWarning on execute() — that warning targets transient ad-hoc callers, not frameworks with their own shutdown protocol.
NOT register an atexit cleanup handler for the persistent event loop — the owner is responsible for calling close() at its own shutdown.

The caller MUST invoke close() (or route cleanup through ShutdownCoordinator) when the owning framework tears down.

Returns:

self to support fluent construction:

self.runtime = LocalRuntime().mark_externally_managed()

Return type:

LocalRuntime

See also

Issue #478 — the original DataFlow internal-warning leak that motivated this public opt-out.
close() — the cleanup call the owner is now responsible for.

on_node_complete(callback: Callable[[NodeCompletionEvent], None | Awaitable[None]]) → Any

Register callback for every node completion this runtime emits.

Returns an unregister function — call it to remove the callback.

Subscribers see one NodeCompletionEvent per node, regardless of whether the node succeeded or failed; the event’s error field carries the failure repr when set. The runtime applies redact_event_for_persistence() to the event BEFORE dispatch so no subscriber ever observes a classified PK or a redacted field’s raw value.

Subscriber exceptions are caught and logged at WARN level — a misbehaving subscriber MUST NOT take down workflow execution.

Parameters:: callback (Callable[[NodeCompletionEvent], None | Awaitable[None]])
Return type:: Any

optimize_runtime_performance() → Dict[str, Any]

Optimize runtime performance based on analytics data.

Returns:: Dictionary describing optimizations applied
Return type:: Dict[str, Any]

property progress_registry: ProgressRegistry

Registry for progress callbacks during workflow execution.

Register callbacks to receive ProgressUpdate events from nodes:

>>> runtime = LocalRuntime()
>>> runtime.progress_registry.register(lambda u: print(u.message))

async query(workflow_id: str, query_name: str, **kwargs: Any) → Any

Query the state of a running workflow.

Executes a registered query handler on the target workflow. Query handlers are registered by nodes or the runtime via the QueryRegistry.

Parameters:

workflow_id (str) – The run_id or workflow_id of the target workflow.
query_name (str) – Name of the query to execute.
**kwargs (Any) – Keyword arguments passed to the query handler.

Returns:

The return value of the query handler.

Raises:

KeyError – If no workflow with the given ID is active, or if no handler is registered for the given query name.

Return type:

Example

>>> result = await runtime.query(run_id, "progress")
>>> print(result)  # {"completed": 5, "total": 10}

record_execution_performance(workflow: Workflow, execution_time: float, nodes_executed: int, used_conditional: bool, performance_improvement: float = 0.0)

Record execution performance for analytics.

Parameters:

workflow (Workflow) – Workflow that was executed
execution_time (float) – Total execution time in seconds
nodes_executed (int) – Number of nodes actually executed
used_conditional (bool) – Whether conditional execution was used
performance_improvement (float) – Performance improvement percentage (0.0-1.0)

property ref_count: int: Current reference count (for debugging/testing).

register_retry_strategy(name: str, strategy)

Register a custom retry strategy.

Parameters:

name (str) – Strategy name for identification
strategy – RetryStrategy instance

register_retry_strategy_for_exception(exception_type: type, strategy)

Register strategy for specific exception type.

Parameters:

exception_type (type) – Exception type to handle
strategy – RetryStrategy to use for this exception type

release() → None

Decrement reference count. Alias for close().

Actual cleanup happens when count reaches 0.

Return type:: None

reset_retry_metrics(): Reset all retry metrics and analytics data.

reset_validation_metrics() → None

Reset validation metrics collector.

Return type:: None

set_automatic_mode_switching(enabled: bool) → None

Enable or disable automatic mode switching based on performance.

Parameters:: enabled (bool) – Whether to enable automatic switching
Return type:: None

set_compatibility_reporting(enabled: bool) → None

Enable or disable compatibility reporting.

Parameters:: enabled (bool) – Whether to enable compatibility reporting
Return type:: None

set_performance_monitoring(enabled: bool) → None

Enable or disable performance monitoring.

Parameters:: enabled (bool) – Whether to enable performance monitoring
Return type:: None

property shutdown_coordinator: ShutdownCoordinator

Get or lazily create the ShutdownCoordinator for this runtime.

The coordinator is created on first access and the runtime’s own cleanup is automatically registered at priority 1 (drain).

Returns:: ShutdownCoordinator instance associated with this runtime.

Example

>>> runtime = LocalRuntime()
>>> runtime.shutdown_coordinator.register("db", pool.close, priority=3)
>>> await runtime.shutdown_coordinator.shutdown()

async shutdown_gracefully(timeout: int = 30) → None

Gracefully shutdown runtime with connection drain and cleanup.

Parameters:: timeout (int) – Maximum time to wait for shutdown completion (seconds).
Return type:: None

signal(workflow_id: str, signal_name: str, data: Any = None) → None

Send a signal to a running workflow.

Delivers a named signal with optional data to a workflow identified by its run_id (or workflow_id). If the workflow has a SignalWaitNode waiting for this signal, it will receive the data and resume.

Parameters:

workflow_id (str) – The run_id or workflow_id of the target workflow.
signal_name (str) – Name of the signal to send.
data (Any) – Arbitrary data payload to deliver with the signal.

Raises:

KeyError – If no workflow with the given ID is currently active.

Return type:

None

Example

>>> runtime = LocalRuntime()
>>> # After starting a workflow with a SignalWaitNode:
>>> runtime.signal(run_id, "approval", {"approved": True})

async start_persistent_mode() → None

Start runtime in persistent mode for long-running applications.

This enables connection pool sharing, resource coordination, and enterprise monitoring features. Only available when persistent_mode=True.

Raises:: RuntimeError – If persistent mode is not enabled or startup fails.
Return type:: None

validate_workflow(workflow: Workflow) → list[str]

Validate a workflow before execution.

Performs comprehensive validation including: - Basic workflow structure validation - Disconnected node detection - Required parameter checking - Connection validation - Performance warnings for large workflows

EXTRACTED FROM: LocalRuntime.validate_workflow() (lines 2054-2119) SHARED LOGIC: 100% - Pure validation with no I/O

Parameters:: workflow (Workflow) – Workflow to validate
Returns:: List of validation warnings (empty if valid)
Raises:: WorkflowValidationError – If workflow is invalid
Return type:: list[str]

Implementation Notes:

Calls workflow.validate() for basic structure
Checks for disconnected nodes in multi-node workflows
Validates required parameters are provided or connected
Warns about performance implications for large workflows
All logic is synchronous and shared between runtimes

Examples

# Basic validation
runtime = LocalRuntime()
warnings = runtime.validate_workflow(workflow)
if warnings:
    for warning in warnings:
        print(f"Warning: {warning}")

# Validation with error handling
try:
    warnings = runtime.validate_workflow(workflow)
    if not warnings:
        print("Workflow is valid")
except WorkflowValidationError as e:
    print(f"Invalid workflow: {e}")

logger

Validation capabilities for workflow runtimes.

This mixin provides comprehensive validation logic for workflows, nodes, parameters, and connections. All methods are 100% shared between sync and async runtimes as they perform pure validation with no I/O operations.

Shared Logic (100%):

All validation methods are pure logic with no sync/async variants needed. They validate data structures and configurations without performing any I/O or execution.

Dependencies:

Requires workflow.graph attribute (from Workflow)
Requires self.logger attribute (from BaseRuntime)
Requires self.debug attribute (from BaseRuntime)
No dependencies on other mixins

Usage:

class LocalRuntime(BaseRuntime, ValidationMixin):: # Inherits all 5 validation methods pass
class AsyncLocalRuntime(BaseRuntime, ValidationMixin):: # Inherits same 5 validation methods pass

Examples

# Validation is called automatically during execution runtime = LocalRuntime() runtime.execute(workflow) # Calls validation methods

# Can also validate manually warnings = runtime.validate_workflow(workflow) runtime._validate_connection_contracts(workflow, node_id, inputs, outputs)

See also

BaseRuntime: Base runtime class
ADR-XXX: Runtime Refactoring for Feature Parity

Version:: Added in: v0.10.0 Part of: Runtime parity remediation

debug

enable_cycles

cyclic_executor

enable_monitoring

conditional_execution

async __aexit__(exc_type, exc_val, exc_tb) → None[source]

Async context manager exit — calls close() which respects ref counting.

Uses close() instead of directly calling cleanup() to ensure the ref counting contract is honored. If this runtime is shared via acquire(), close() will only decrement — not destroy resources.

Return type:: None

Characteristics:

Concurrent execution of async nodes
Efficient for API calls and file I/O
Better resource utilization
Requires async-compatible nodes

Example Usage:

from kailash.runtime import AsyncLocalRuntime
import asyncio

workflow = Workflow("async_example")
runtime = AsyncLocalRuntime()

# Add async nodes
workflow.add_node("AsyncHTTPClient", "fetch1", config={"url": "..."})
workflow.add_node("AsyncHTTPClient", "fetch2", config={"url": "..."})

# Execute asynchronously
async def run():
    try:
        results = await runtime.execute_async(workflow)
        return results
    finally:
        runtime.close()

results = asyncio.run(run())

BaseRuntime 

The abstract base class all runtimes implement.

class kailash.runtime.base.BaseRuntime(debug: bool = False, enable_cycles: bool = True, enable_async: bool = True, max_concurrency: int = 10, user_context: Any | None = None, enable_monitoring: bool = True, enable_resource_limits: bool = False, enable_security: bool = False, enable_audit: bool = False, resource_limits: Dict[str, Any] | None = None, secret_provider: Any | None = None, connection_validation: str = 'warn', conditional_execution: str = 'route_data', content_aware_success_detection: bool = True, persistent_mode: bool = False, enable_connection_sharing: bool = True, max_concurrent_workflows: int = 10, connection_pool_size: int = 20, enable_enterprise_monitoring: bool = False, enable_health_monitoring: bool = False, enable_resource_coordination: bool = True, circuit_breaker_config: Dict | None = None, retry_policy_config: Dict | None = None, connection_pool_config: Dict | None = None, trust_context: 'RuntimeTrustContext' | None = None, trust_verifier: Any | None = None, trust_verification_mode: str = 'disabled', audit_generator: Any | None = None, audit_log_to_stdout: bool = False, **kwargs)[source]

Bases: ABC

Base class for all workflow runtimes.

This class provides shared logic that is common to both LocalRuntime (synchronous) and AsyncLocalRuntime (asynchronous), eliminating code duplication and ensuring consistent behavior.

Architecture:

BaseRuntime provides foundational capabilities: - Configuration validation and initialization - Workflow metadata management - Result tracking and run ID generation - Execution metadata management - Workflow caching - Enterprise feature initialization helpers

Subclasses (LocalRuntime, AsyncLocalRuntime) inherit this base and add runtime-specific execution logic through mixins and concrete implementations.

Design Pattern:

Follows the SecureGovernedNode mixin pattern established in the SDK: - Base class provides shared initialization via super().__init__() - Subclasses call super().__init__(**kwargs) to initialize base - Mixins can be added to subclasses for additional capabilities - Abstract methods define the runtime-specific contract

Extracted Logic:

This class extracts ~500 lines of shared logic from LocalRuntime: - Lines 190-350: Configuration initialization and validation - Lines 2100-2200: Enterprise feature helpers (placeholders) - Utility methods for run ID generation, metadata tracking - Workflow caching and state management

Usage:

This class is not meant to be instantiated directly. Use LocalRuntime or AsyncLocalRuntime instead.

>>> # DON'T: Direct instantiation (will fail - abstract class)
>>> runtime = BaseRuntime()  # Raises TypeError
>>>
>>> # DO: Use concrete implementations
>>> from kailash.runtime.local import LocalRuntime
>>> runtime = LocalRuntime(debug=True, enable_cycles=True)
>>>
>>> from kailash.runtime.async_local import AsyncLocalRuntime
>>> async_runtime = AsyncLocalRuntime(debug=True, enable_async=True)

See also

LocalRuntime: Synchronous workflow execution
AsyncLocalRuntime: Asynchronous workflow execution
ADR-048: Unified Runtime Architecture
ADR-XXX: Runtime Refactoring for Feature Parity

Version:: Added in: v0.10.0 Part of: Runtime parity remediation (2025-10-25)

Parameters:

debug (bool)
enable_cycles (bool)
enable_async (bool)
max_concurrency (int)
user_context (Optional[Any])
enable_monitoring (bool)
enable_resource_limits (bool)
enable_security (bool)
enable_audit (bool)
resource_limits (Optional[Dict[str, Any]])
secret_provider (Optional[Any])
connection_validation (str)
conditional_execution (str)
content_aware_success_detection (bool)
persistent_mode (bool)
enable_connection_sharing (bool)
max_concurrent_workflows (int)
connection_pool_size (int)
enable_enterprise_monitoring (bool)
enable_health_monitoring (bool)
enable_resource_coordination (bool)
circuit_breaker_config (Optional[Dict])
retry_policy_config (Optional[Dict])
connection_pool_config (Optional[Dict])
trust_context (Optional['RuntimeTrustContext'])
trust_verifier (Optional[Any])
trust_verification_mode (str)
audit_generator (Optional[Any])
audit_log_to_stdout (bool)

__init__(debug: bool = False, enable_cycles: bool = True, enable_async: bool = True, max_concurrency: int = 10, user_context: Any | None = None, enable_monitoring: bool = True, enable_resource_limits: bool = False, enable_security: bool = False, enable_audit: bool = False, resource_limits: Dict[str, Any] | None = None, secret_provider: Any | None = None, connection_validation: str = 'warn', conditional_execution: str = 'route_data', content_aware_success_detection: bool = True, persistent_mode: bool = False, enable_connection_sharing: bool = True, max_concurrent_workflows: int = 10, connection_pool_size: int = 20, enable_enterprise_monitoring: bool = False, enable_health_monitoring: bool = False, enable_resource_coordination: bool = True, circuit_breaker_config: Dict | None = None, retry_policy_config: Dict | None = None, connection_pool_config: Dict | None = None, trust_context: 'RuntimeTrustContext' | None = None, trust_verifier: Any | None = None, trust_verification_mode: str = 'disabled', audit_generator: Any | None = None, audit_log_to_stdout: bool = False, **kwargs)[source]

Initialize base runtime.

This method extracts and consolidates common initialization logic from LocalRuntime (lines 190-350), providing a unified foundation for both sync and async runtimes.

Parameters:

debug (bool) – Whether to enable debug logging.
enable_cycles (bool) – Whether to enable cyclic workflow support.
enable_async (bool) – Whether to enable async execution for async nodes.
max_concurrency (int) – Maximum concurrent async operations.
user_context (Optional[Any]) – User context for access control (optional).
enable_monitoring (bool) – Whether to enable performance monitoring.
enable_security (bool) – Whether to enable security features.
enable_audit (bool) – Whether to enable audit logging.
resource_limits (Optional[Dict[str, Any]]) – Resource limits (memory_mb, cpu_cores, etc.).
secret_provider (Optional[Any]) – Optional secret provider for runtime secret injection.
connection_validation (str) – Connection parameter validation mode: - “off”: No validation (backward compatibility) - “warn”: Log warnings on validation errors (default) - “strict”: Raise errors on validation failures
conditional_execution (str) – Execution strategy for conditional routing: - “route_data”: Current behavior - all nodes execute, data routing only (default) - “skip_branches”: New behavior - skip unreachable branches entirely
content_aware_success_detection (bool) – Whether to enable content-aware success detection: - True: Check return value content for success/failure patterns (default) - False: Only use exception-based failure detection (legacy mode)
persistent_mode (bool) – Whether to enable persistent runtime mode for long-running applications.
enable_connection_sharing (bool) – Whether to enable connection pool sharing across runtime instances.
max_concurrent_workflows (int) – Maximum number of concurrent workflows in persistent mode.
connection_pool_size (int) – Default size for connection pools.
enable_enterprise_monitoring (bool) – Enable enterprise monitoring features.
enable_health_monitoring (bool) – Enable health monitoring.
enable_resource_coordination (bool) – Enable resource coordination.
circuit_breaker_config (Optional[Dict]) – Circuit breaker configuration.
retry_policy_config (Optional[Dict]) – Retry policy configuration.
connection_pool_config (Optional[Dict]) – Connection pool configuration.
trust_context (Optional['RuntimeTrustContext']) – Optional RuntimeTrustContext for trust propagation (CARE-015).
trust_verifier (Optional[Any]) – Optional TrustVerifier for trust verification (CARE-016).
trust_verification_mode (str) – Trust verification mode: - “disabled”: No trust verification (default for backward compatibility) - “permissive”: Log trust violations but allow execution - “enforcing”: Block execution on trust violations
audit_generator (Optional[Any]) – Optional RuntimeAuditGenerator for EATP-compliant audit trails (CARE-018).
audit_log_to_stdout (bool) – Whether to log audit events to stdout (default False).
**kwargs – Additional configuration (passed to mixins via super())
enable_resource_limits (bool)

Raises:

ValueError – If configuration parameters are invalid

Extracted From:: LocalRuntime.__init__ (lines 190-350 in local.py) - Configuration validation logic (lines 246-275) - Parameter initialization (lines 276-304) - Enterprise feature setup (lines 298-350)

abstractmethod close() → None[source]

Release runtime resources.

All runtime subclasses MUST implement proper cleanup. This ensures event loops, thread pools, connection pools, and other resources are released when the runtime is no longer needed.

Usage:

>>> runtime = LocalRuntime()
>>> try:
...     results = runtime.execute(workflow)
... finally:
...     runtime.close()

Or use context manager (preferred):

>>> with LocalRuntime() as runtime:
...     results = runtime.execute(workflow)

See also

acquire(): Increment reference count for shared runtimes
release(): Alias for close()
__enter__/__exit__: Context manager protocol

Added in: v0.12.0 (issue #71 — runtime lifecycle enforcement)

Return type:: None

__enter__() → BaseRuntime[source]

Enter context manager.

Return type:: BaseRuntime

__exit__(exc_type, exc_val, exc_tb) → None[source]

Exit context manager — calls close().

Return type:: None

abstractmethod execute(workflow: Workflow, *, soft_time_limit: float | None = None, time_limit: float | None = None, **kwargs)[source]

Execute workflow (runtime-specific implementation).

This method MUST be implemented by subclasses with appropriate signatures for sync/async execution:

Sync Implementation (LocalRuntime):

def execute(
    self,
    workflow: Workflow,
    parameters: Optional[Dict] = None,
    *,
    soft_time_limit: float | None = None,
    time_limit: float | None = None,
    **kwargs,
) -> Tuple[Dict[str, Any], str]:
    '''Execute workflow synchronously.'''
    # Implementation
    pass

Async Implementation (AsyncLocalRuntime):

async def execute(
    self,
    workflow: Workflow,
    parameters: Optional[Dict] = None,
    *,
    soft_time_limit: float | None = None,
    time_limit: float | None = None,
    **kwargs,
) -> Tuple[Dict[str, Any], str]:
    '''Execute workflow asynchronously.'''
    # Implementation
    pass

Parameters:

workflow (Workflow) – The workflow to execute
soft_time_limit (float | None) – Optional advisory deadline in seconds. When reached, the running workflow is signalled via the cancellation token; user code MAY catch SoftTimeLimitExceeded, finish in-flight work, and exit cleanly before the hard limit fires. Enforcement lands in #912 Shard 2.
time_limit (float | None) – Optional unconditional kill deadline in seconds. When time_limit + grace elapses, the wrapper raises HardTimeLimitExceeded regardless of acknowledgement. Enforcement lands in #912 Shard 2.
**kwargs – Additional execution parameters (runtime-specific). Retained per the additive #912 Shard 1 contract; a future shard MAY tighten via a Rule 6a deprecation cycle.

Returns:

Tuple of (results_dict, run_id)

Raises:

RuntimeExecutionError – If execution fails
NotImplementedError – If called directly on BaseRuntime

Implementation Requirements:

Generate run_id using self._generate_run_id()
Initialize metadata using self._initialize_execution_metadata()
Validate workflow before execution
Execute nodes in proper order
Collect and return results
Update metadata on completion/failure

Example Implementations:

Sync (LocalRuntime):

>>> def execute(self, workflow, parameters=None, **kwargs):
...     run_id = self._generate_run_id()
...     metadata = self._initialize_execution_metadata(workflow, run_id)
...     # Execute workflow synchronously
...     results = self._execute_sync(workflow, parameters)
...     return results, run_id

Async (AsyncLocalRuntime):

>>> async def execute(self, workflow, parameters=None, **kwargs):
...     run_id = self._generate_run_id()
...     metadata = self._initialize_execution_metadata(workflow, run_id)
...     # Execute workflow asynchronously
...     results = await self._execute_async(workflow, parameters)
...     return results, run_id

__repr__() → str[source]

Get string representation of runtime instance.

Returns:: String representation with key configuration
Return type:: str

Example

>>> runtime = LocalRuntime(debug=True, enable_cycles=True)
>>> print(repr(runtime))
<LocalRuntime(id=runtime_..., debug=True, cycles=True, async=False)>

Trust Runtime 

Trust-plane integration for runtime execution (EATP).

Trust context module for Kailash runtime (CARE-015, CARE-016, CARE-018).

This module provides RuntimeTrustContext for propagating trust information through workflow execution, TrustVerifier for bridging to Kaizen TrustOperations, and RuntimeAuditGenerator for EATP-compliant audit trails: - Human origin tracking across agent delegation - Constraint propagation and tightening - Audit trail for compliance - Bridge to Kaizen execution context - Trust verification for workflows, nodes, and resources - EATP-compliant audit generation

Usage:

from kailash.runtime.trust import (
    RuntimeTrustContext,
    TrustVerificationMode,
    get_runtime_trust_context,
    set_runtime_trust_context,
    runtime_trust_context,
    TrustVerifier,
    TrustVerifierConfig,
    VerificationResult,
    MockTrustVerifier,
    AuditEventType,
    AuditEvent,
    RuntimeAuditGenerator,
)

# Create and use trust context
ctx = RuntimeTrustContext(
    trace_id="trace-123",
    verification_mode=TrustVerificationMode.ENFORCING,
)

with runtime_trust_context(ctx):
    runtime = LocalRuntime()
    results, run_id = runtime.execute(workflow)

# Use trust verifier
verifier = TrustVerifier(
    config=TrustVerifierConfig(mode="enforcing"),
)
result = await verifier.verify_workflow_access(
    workflow_id="my-workflow",
    agent_id="agent-123",
)

# Use audit generator
generator = RuntimeAuditGenerator(enabled=True)
await generator.workflow_started("run-1", "my-workflow", ctx)

Version:: Added in: v0.11.0 Part of: CARE trust implementation (Phase 2)

class kailash.runtime.trust.RuntimeTrustContext(trace_id: str = <factory>, human_origin: Any | None = None, delegation_chain: List[str] = <factory>, delegation_depth: int = 0, constraints: Dict[str, ~typing.Any]=<factory>, verification_mode: TrustVerificationMode = TrustVerificationMode.DISABLED, workflow_id: str | None = None, node_path: List[str] = <factory>, metadata: Dict[str, ~typing.Any]=<factory>, created_at: datetime = <factory>)[source]

Bases: object

Trust context for workflow execution.

This dataclass captures trust information that propagates through workflow execution, enabling: - Human origin tracking across agent delegation chains - Constraint propagation with tightening semantics - Audit trail for compliance - Verification mode control

Immutability:: with_node() and with_constraints() create new instances, preserving the original context unchanged.

Variables:

trace_id (str) – Unique identifier for tracing execution (default: UUID)
human_origin (Any | None) – Origin information (compatible with Kaizen HumanOrigin)
delegation_chain (List[str]) – List of agent IDs in delegation chain
delegation_depth (int) – Current depth in delegation chain
constraints (Dict[str, Any]) – Dictionary of constraints (can only be tightened)
verification_mode (kailash.runtime.trust.context.TrustVerificationMode) – How to handle trust verification
workflow_id (str | None) – ID of the workflow being executed
node_path (List[str]) – Execution path through nodes (for audit)
metadata (Dict[str, Any]) – Additional metadata for extensibility
created_at (datetime.datetime) – When this context was created

Parameters:

trace_id (str)
human_origin (Any | None)
delegation_chain (List[str])
delegation_depth (int)
constraints (Dict[str, Any])
verification_mode (TrustVerificationMode)
workflow_id (str | None)
node_path (List[str])
metadata (Dict[str, Any])
created_at (datetime)

Example

>>> ctx = RuntimeTrustContext(
...     trace_id="trace-123",
...     verification_mode=TrustVerificationMode.ENFORCING,
...     constraints={"max_tokens": 1000},
... )
>>> node_ctx = ctx.with_node("process_data")
>>> node_ctx.node_path
['process_data']

trace_id: str

human_origin: Any | None = None

delegation_chain: List[str]

delegation_depth: int = 0

constraints: Dict[str, Any]

verification_mode: TrustVerificationMode = 'disabled'

workflow_id: str | None = None

node_path: List[str]

metadata: Dict[str, Any]

created_at: datetime

with_node(node_id: str) → RuntimeTrustContext[source]

Create new context with node_id appended to node_path.

This method creates a new RuntimeTrustContext instance with the given node_id added to the execution path. The original context remains unchanged (immutable pattern).

Parameters:: node_id (str) – The node ID to append to the path
Returns:: New RuntimeTrustContext with extended node_path
Return type:: RuntimeTrustContext

Example

>>> ctx = RuntimeTrustContext(node_path=["node1"])
>>> new_ctx = ctx.with_node("node2")
>>> ctx.node_path
['node1']
>>> new_ctx.node_path
['node1', 'node2']

with_constraints(additional_constraints: Dict[str, Any]) → RuntimeTrustContext[source]

Create new context with tightened constraints.

This method creates a new RuntimeTrustContext instance with the additional constraints merged into the existing ones using tightening semantics. The original context remains unchanged (immutable pattern).

Tightening rules:

Numeric values: takes the minimum (tighter limit)
Sets/lists: takes the intersection (fewer allowed items)
Booleans: once False (restricted), stays False
New keys: added freely (adding a constraint is tightening)
Strings: new value replaces old (caller’s responsibility)

Parameters:: additional_constraints (Dict[str, Any]) – Constraints to merge/tighten
Returns:: New RuntimeTrustContext with tightened constraints
Return type:: RuntimeTrustContext

Example

>>> ctx = RuntimeTrustContext(constraints={"max_tokens": 1000})
>>> new_ctx = ctx.with_constraints({"max_tokens": 500, "allowed_tools": ["read"]})
>>> new_ctx.constraints
{'max_tokens': 500, 'allowed_tools': ['read']}
>>> loosened = ctx.with_constraints({"max_tokens": 9999})
>>> loosened.constraints["max_tokens"]  # Stays at 1000 (tighter)
1000

to_dict() → Dict[str, Any][source]

Serialize context to dictionary.

Handles: - datetime fields as ISO format strings - human_origin via its to_dict() method if available - TrustVerificationMode as string value

Returns:: Dictionary representation of the context
Return type:: Dict[str, Any]

Example

>>> ctx = RuntimeTrustContext(trace_id="trace-123")
>>> data = ctx.to_dict()
>>> data["trace_id"]
'trace-123'

classmethod from_dict(data: Dict[str, Any]) → RuntimeTrustContext[source]

Deserialize context from dictionary.

Handles: - ISO format strings as datetime - String verification mode values

Parameters:: data (Dict[str, Any]) – Dictionary representation of the context
Returns:: RuntimeTrustContext instance
Return type:: RuntimeTrustContext

Example

>>> data = {"trace_id": "trace-123", ...}
>>> ctx = RuntimeTrustContext.from_dict(data)
>>> ctx.trace_id
'trace-123'

classmethod from_kaizen_context(ctx: Any) → RuntimeTrustContext[source]

Bridge from Kaizen ExecutionContext.

Creates a RuntimeTrustContext from a Kaizen ExecutionContext, mapping relevant fields and setting verification_mode to ENFORCING since Kaizen contexts imply trust enforcement.

This method handles the case where Kaizen is not installed gracefully, using duck typing to access expected attributes.

Parameters:: ctx (Any) – Kaizen ExecutionContext (or any object with expected attributes)
Returns:: RuntimeTrustContext bridged from Kaizen context
Return type:: RuntimeTrustContext

Example

>>> from kaizen.core.context import ExecutionContext
>>> kaizen_ctx = ExecutionContext(...)
>>> runtime_ctx = RuntimeTrustContext.from_kaizen_context(kaizen_ctx)
>>> runtime_ctx.verification_mode
TrustVerificationMode.ENFORCING

__init__(trace_id: str = <factory>, human_origin: Any | None = None, delegation_chain: List[str] = <factory>, delegation_depth: int = 0, constraints: Dict[str, ~typing.Any]=<factory>, verification_mode: TrustVerificationMode = TrustVerificationMode.DISABLED, workflow_id: str | None = None, node_path: List[str] = <factory>, metadata: Dict[str, ~typing.Any]=<factory>, created_at: datetime = <factory>) → None

Parameters:

trace_id (str)
human_origin (Any | None)
delegation_chain (List[str])
delegation_depth (int)
constraints (Dict[str, Any])
verification_mode (TrustVerificationMode)
workflow_id (str | None)
node_path (List[str])
metadata (Dict[str, Any])
created_at (datetime)

Return type:

None

class kailash.runtime.trust.TrustVerificationMode(value)[source]

Bases: Enum

Trust verification modes for runtime execution.

Modes:: DISABLED: No trust verification (default for backward compatibility) PERMISSIVE: Log trust violations but allow execution ENFORCING: Block execution on trust violations

DISABLED = 'disabled'

PERMISSIVE = 'permissive'

ENFORCING = 'enforcing'

classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name): Return the member matching name.

classmethod __iter__(): Return members in definition order.

classmethod __len__(): Return the number of members (no aliases)

kailash.runtime.trust.get_runtime_trust_context() → RuntimeTrustContext | None[source]

Get the current runtime trust context.

Returns the RuntimeTrustContext from the current context variable, or None if no context is set.

Returns:: Current RuntimeTrustContext or None
Return type:: RuntimeTrustContext | None

Example

>>> ctx = get_runtime_trust_context()
>>> if ctx:
...     print(f"Trace: {ctx.trace_id}")

kailash.runtime.trust.set_runtime_trust_context(ctx: RuntimeTrustContext | None) → None[source]

Set the runtime trust context.

Sets the RuntimeTrustContext in the current context variable. Pass None to clear the context.

Parameters:: ctx (RuntimeTrustContext | None) – RuntimeTrustContext to set, or None to clear
Return type:: None

Example

>>> ctx = RuntimeTrustContext(trace_id="trace-123")
>>> set_runtime_trust_context(ctx)
>>> get_runtime_trust_context().trace_id
'trace-123'

kailash.runtime.trust.runtime_trust_context(ctx: RuntimeTrustContext) → Iterator[RuntimeTrustContext][source]

Context manager for scoped trust context propagation.

Sets the trust context for the duration of the block, then resets to the previous value on exit (even on exception).

Parameters:: ctx (RuntimeTrustContext) – RuntimeTrustContext to use within the block
Yields:: The same RuntimeTrustContext that was passed in
Return type:: Iterator[RuntimeTrustContext]

Example

>>> ctx = RuntimeTrustContext(trace_id="trace-123")
>>> with runtime_trust_context(ctx) as active_ctx:
...     assert get_runtime_trust_context() is ctx
...     print(active_ctx.trace_id)
trace-123

class kailash.runtime.trust.VerificationResult(allowed: bool, reason: str | None = None, constraints: Dict[str, ~typing.Any]=<factory>, capability_used: str | None = None, trace_id: str | None = None)[source]

Bases: object

Result of a trust verification check.

Represents the outcome of verifying whether an agent has permission to execute a workflow, node, or access a resource.

Variables:

allowed (bool) – Whether the operation is allowed
reason (str | None) – Human-readable reason for the decision
constraints (Dict[str, Any]) – Any constraints that apply to the allowed operation
capability_used (str | None) – The capability that granted access (if any)
trace_id (str | None) – Trace ID for audit trail

Parameters:

allowed (bool)
reason (str | None)
constraints (Dict[str, Any])
capability_used (str | None)
trace_id (str | None)

Example

>>> result = VerificationResult(allowed=True, reason="Access granted")
>>> if result:  # Uses __bool__
...     print("Allowed!")
Allowed!

allowed: bool

reason: str | None = None

constraints: Dict[str, Any]

capability_used: str | None = None

trace_id: str | None = None

__bool__() → bool[source]

Allow using result directly in boolean context.

Returns:: True if allowed, False otherwise
Return type:: bool

to_dict() → Dict[str, Any][source]

Serialize result to dictionary.

Returns:: Dictionary representation of the result
Return type:: Dict[str, Any]

__init__(allowed: bool, reason: str | None = None, constraints: Dict[str, ~typing.Any]=<factory>, capability_used: str | None = None, trace_id: str | None = None) → None

Parameters:

allowed (bool)
reason (str | None)
constraints (Dict[str, Any])
capability_used (str | None)
trace_id (str | None)

Return type:

None

class kailash.runtime.trust.TrustVerifierConfig(mode: str = 'disabled', cache_enabled: bool = True, cache_ttl_seconds: int = 60, fallback_allow: bool | None = None, audit_denials: bool = True, high_risk_nodes: List[str] = <factory>)[source]

Bases: object

Configuration for TrustVerifier.

Variables:

mode (str) – Verification mode - “disabled”, “permissive”, or “enforcing”
cache_enabled (bool) – Whether to cache verification results
cache_ttl_seconds (int) – Time-to-live for cached results in seconds
fallback_allow (bool | None) – Whether to allow operations when verifier unavailable
audit_denials (bool) – Whether to log denied operations
high_risk_nodes (List[str]) – List of node types that require elevated verification

Parameters:

mode (str)
cache_enabled (bool)
cache_ttl_seconds (int)
fallback_allow (bool | None)
audit_denials (bool)
high_risk_nodes (List[str])

Example

>>> config = TrustVerifierConfig(
...     mode="enforcing",
...     cache_ttl_seconds=120,
...     high_risk_nodes=["BashCommand", "FileWrite"],
... )

mode: str = 'disabled'

cache_enabled: bool = True

cache_ttl_seconds: int = 60

fallback_allow: bool | None = None

audit_denials: bool = True

high_risk_nodes: List[str]

__init__(mode: str = 'disabled', cache_enabled: bool = True, cache_ttl_seconds: int = 60, fallback_allow: bool | None = None, audit_denials: bool = True, high_risk_nodes: List[str] = <factory>) → None

Parameters:

mode (str)
cache_enabled (bool)
cache_ttl_seconds (int)
fallback_allow (bool | None)
audit_denials (bool)
high_risk_nodes (List[str])

Return type:

None

class kailash.runtime.trust.TrustVerifier(kaizen_backend: Any | None = None, config: TrustVerifierConfig | None = None)[source]

Bases: object

Bridge between Core SDK runtime and Kaizen TrustOperations for verification.

TrustVerifier provides a verification layer that can optionally integrate with Kaizen’s TrustOperations to enforce trust policies on workflow, node, and resource access.

Modes:

DISABLED: No verification, all operations allowed (default)
PERMISSIVE: Verify and log, but allow even if denied
ENFORCING: Block operations that fail verification

Caching:

Results are cached to avoid repeated backend calls. Cache entries expire after the configured TTL.

Example

>>> verifier = TrustVerifier(
...     config=TrustVerifierConfig(mode="enforcing"),
... )
>>> result = await verifier.verify_workflow_access(
...     workflow_id="my-workflow",
...     agent_id="agent-123",
... )
>>> if result.allowed:
...     print("Access granted")

Parameters:

kaizen_backend (Optional[Any])
config (Optional[TrustVerifierConfig])

__init__(kaizen_backend: Any | None = None, config: TrustVerifierConfig | None = None) → None[source]

Initialize the TrustVerifier.

Parameters:

kaizen_backend (Any | None) – Optional Kaizen TrustOperations instance for verification
config (TrustVerifierConfig | None) – Optional configuration (defaults to TrustVerifierConfig())

Return type:

None

property is_enabled: bool

Check if verification is enabled.

Returns:: True if mode is not DISABLED

property is_enforcing: bool

Check if verification is in enforcing mode.

Returns:: True if mode is ENFORCING

clear_cache() → None[source]

Clear all cached verification results.

Return type:: None

invalidate_agent(agent_id: str) → int[source]

Invalidate all cached results for a specific agent (CARE-043).

Call this when an agent is revoked to ensure revocation takes immediate effect without waiting for cache TTL expiry.

Parameters:: agent_id (str) – The agent whose cache entries should be invalidated
Returns:: Number of cache entries removed
Return type:: int

invalidate_node(node_type: str) → int[source]

Invalidate all cached results for a specific node type (CARE-043).

Parameters:: node_type (str) – The node type whose cache entries should be invalidated
Returns:: Number of cache entries removed
Return type:: int

async verify_workflow_access(workflow_id: str, agent_id: str, trust_context: RuntimeTrustContext | None = None) → VerificationResult[source]

Verify if an agent can execute a workflow.

Parameters:

workflow_id (str) – The workflow to verify access to
agent_id (str) – The agent requesting access
trust_context (RuntimeTrustContext | None) – Optional RuntimeTrustContext for additional context

Returns:

VerificationResult indicating whether access is allowed

Return type:

async verify_node_access(node_id: str, node_type: str, agent_id: str, trust_context: RuntimeTrustContext | None = None) → VerificationResult[source]

Verify if an agent can execute a specific node.

High-risk nodes (configured in TrustVerifierConfig) receive elevated verification when a backend is available.

Parameters:

node_id (str) – The node instance ID
node_type (str) – The node type (e.g., “BashCommand”, “HttpRequest”)
agent_id (str) – The agent requesting access
trust_context (RuntimeTrustContext | None) – Optional RuntimeTrustContext for additional context

Returns:

VerificationResult indicating whether access is allowed

Return type:

async verify_resource_access(resource: str, action: str, agent_id: str, trust_context: RuntimeTrustContext | None = None) → VerificationResult[source]

Verify if an agent can access a resource.

Parameters:

resource (str) – The resource path or identifier
action (str) – The action to perform (e.g., “read”, “write”)
agent_id (str) – The agent requesting access
trust_context (RuntimeTrustContext | None) – Optional RuntimeTrustContext for additional context

Returns:

VerificationResult indicating whether access is allowed

Return type:

class kailash.runtime.trust.MockTrustVerifier(default_allow: bool = True, denied_agents: List[str] | None = None, denied_nodes: List[str] | None = None, config: TrustVerifierConfig | None = None)[source]

Bases: TrustVerifier

Mock verifier for testing without Kaizen backend.

Provides configurable allow/deny behavior for testing trust verification without requiring an actual Kaizen TrustOperations backend.

Example

>>> verifier = MockTrustVerifier(
...     default_allow=True,
...     denied_agents=["blocked-agent"],
...     denied_nodes=["BashCommand"],
... )
>>> result = await verifier.verify_workflow_access(
...     workflow_id="test-wf",
...     agent_id="good-agent",
... )
>>> result.allowed
True

Parameters:

default_allow (bool)
denied_agents (Optional[List[str]])
denied_nodes (Optional[List[str]])
config (Optional[TrustVerifierConfig])

__init__(default_allow: bool = True, denied_agents: List[str] | None = None, denied_nodes: List[str] | None = None, config: TrustVerifierConfig | None = None) → None[source]

Initialize the MockTrustVerifier.

Parameters:

default_allow (bool) – Default behavior when not explicitly denied
denied_agents (List[str] | None) – List of agent IDs that should be denied
denied_nodes (List[str] | None) – List of node types that should be denied
config (TrustVerifierConfig | None) – Optional configuration (defaults to enforcing mode)

Return type:

None

async verify_workflow_access(workflow_id: str, agent_id: str, trust_context: RuntimeTrustContext | None = None) → VerificationResult[source]

Verify workflow access using mock rules.

Parameters:

workflow_id (str) – The workflow to verify access to
agent_id (str) – The agent requesting access
trust_context (RuntimeTrustContext | None) – Optional RuntimeTrustContext for additional context

Returns:

VerificationResult based on mock rules

Return type:

async verify_node_access(node_id: str, node_type: str, agent_id: str, trust_context: RuntimeTrustContext | None = None) → VerificationResult[source]

Verify node access using mock rules.

Parameters:

node_id (str) – The node instance ID
node_type (str) – The node type
agent_id (str) – The agent requesting access
trust_context (RuntimeTrustContext | None) – Optional RuntimeTrustContext for additional context

Returns:

VerificationResult based on mock rules

Return type:

clear_cache() → None

Clear all cached verification results.

Return type:: None

invalidate_agent(agent_id: str) → int

Invalidate all cached results for a specific agent (CARE-043).

Call this when an agent is revoked to ensure revocation takes immediate effect without waiting for cache TTL expiry.

Parameters:: agent_id (str) – The agent whose cache entries should be invalidated
Returns:: Number of cache entries removed
Return type:: int

invalidate_node(node_type: str) → int

Invalidate all cached results for a specific node type (CARE-043).

Parameters:: node_type (str) – The node type whose cache entries should be invalidated
Returns:: Number of cache entries removed
Return type:: int

property is_enabled: bool

Check if verification is enabled.

Returns:: True if mode is not DISABLED

property is_enforcing: bool

Check if verification is in enforcing mode.

Returns:: True if mode is ENFORCING

async verify_resource_access(resource: str, action: str, agent_id: str, trust_context: RuntimeTrustContext | None = None) → VerificationResult[source]

Verify resource access using mock rules.

Parameters:

resource (str) – The resource path or identifier
action (str) – The action to perform
agent_id (str) – The agent requesting access
trust_context (RuntimeTrustContext | None) – Optional RuntimeTrustContext for additional context

Returns:

VerificationResult based on mock rules

Return type:

class kailash.runtime.trust.AuditEventType(value)[source]

Bases: str, Enum

Well-known audit event types (cross-domain union).

String-backed enum whose .value is used to populate the canonical AuditEvent.event_type field. This enum provides a shared vocabulary for high-level trust-plane events; domain-specific modules MAY define their own enums as long as the string values are preserved in AuditEvent.event_type.

ACTION_EXECUTED = 'action_executed'

ACTION_DENIED = 'action_denied'

DELEGATION_CREATED = 'delegation_created'

DELEGATION_REVOKED = 'delegation_revoked'

TRUST_ESTABLISHED = 'trust_established'

TRUST_REVOKED = 'trust_revoked'

POLICY_CHANGED = 'policy_changed'

ACCESS_GRANTED = 'access_granted'

ACCESS_DENIED = 'access_denied'

CONSTRAINT_VIOLATED = 'constraint_violated'

SYSTEM_EVENT = 'system_event'

CUSTOM = 'custom'

WORKFLOW_START = 'workflow_start'

WORKFLOW_END = 'workflow_end'

WORKFLOW_ERROR = 'workflow_error'

NODE_START = 'node_start'

NODE_END = 'node_end'

NODE_ERROR = 'node_error'

TRUST_VERIFICATION = 'trust_verification'

TRUST_DENIED = 'trust_denied'

RESOURCE_ACCESS = 'resource_access'

DELEGATION_USED = 'delegation_used'

DISCLOSURE = 'disclosure'

encode(encoding='utf-8', errors='strict')

Encode the string using the codec registered for encoding.

encoding: The encoding in which to encode the string.
errors: The error handling scheme to use for encoding errors. The default is ‘strict’ meaning that encoding errors raise a UnicodeEncodeError. Other possible values are ‘ignore’, ‘replace’ and ‘xmlcharrefreplace’ as well as any other name registered with codecs.register_error that can handle UnicodeEncodeErrors.

replace(old, new, count=-1, /)

Return a copy with all occurrences of substring old replaced by new.

count
Maximum number of occurrences to replace. -1 (the default value) means replace all occurrences.

If the optional argument count is given, only the first count occurrences are replaced.

split(sep=None, maxsplit=-1)

Return a list of the substrings in the string, using sep as the separator string.

sep
The separator used to split the string.

When set to None (the default value), will split on any whitespace character (including n r t f and spaces) and will discard empty strings from the result.

maxsplit
Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the front of the string and works to the end.

Note, str.split() is mainly useful for data that has been intentionally delimited. With natural text that includes punctuation, consider using the regular expression module.

rsplit(sep=None, maxsplit=-1)

Return a list of the substrings in the string, using sep as the separator string.

sep
The separator used to split the string.

When set to None (the default value), will split on any whitespace character (including n r t f and spaces) and will discard empty strings from the result.

maxsplit
Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the end of the string and works to the front.

join(iterable, /)

Concatenate any number of strings.

The string whose method is called is inserted in between each given string. The result is returned as a new string.

Example: ‘.’.join([‘ab’, ‘pq’, ‘rs’]) -> ‘ab.pq.rs’

capitalize()

Return a capitalized version of the string.

More specifically, make the first character have upper case and the rest lower case.

casefold(): Return a version of the string suitable for caseless comparisons.

title()

Return a version of the string where each word is titlecased.

More specifically, words start with uppercased characters and all remaining cased characters have lower case.

center(width, fillchar=' ', /)

Return a centered string of length width.

Padding is done using the specified fill character (default is a space).

count(sub[, start[, end]]) → int: Return the number of non-overlapping occurrences of substring sub in string S[start:end]. Optional arguments start and end are interpreted as in slice notation.

expandtabs(tabsize=8)

Return a copy where all tab characters are expanded using spaces.

If tabsize is not given, a tab size of 8 characters is assumed.

find(sub[, start[, end]]) → int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

partition(sep, /)

Partition the string into three parts using the given separator.

This will search for the separator in the string. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing the original string and two empty strings.

index(sub[, start[, end]]) → int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

ljust(width, fillchar=' ', /)

Return a left-justified string of length width.

Padding is done using the specified fill character (default is a space).

lower(): Return a copy of the string converted to lowercase.

lstrip(chars=None, /)

Return a copy of the string with leading whitespace removed.

If chars is given and not None, remove characters in chars instead.

rfind(sub[, start[, end]]) → int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

rindex(sub[, start[, end]]) → int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

rjust(width, fillchar=' ', /)

Return a right-justified string of length width.

Padding is done using the specified fill character (default is a space).

rstrip(chars=None, /)

Return a copy of the string with trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

rpartition(sep, /)

Partition the string into three parts using the given separator.

This will search for the separator in the string, starting at the end. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing two empty strings and the original string.

splitlines(keepends=False)

Return a list of the lines in the string, breaking at line boundaries.

Line breaks are not included in the resulting list unless keepends is given and true.

strip(chars=None, /)

Return a copy of the string with leading and trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

swapcase(): Convert uppercase characters to lowercase and lowercase characters to uppercase.

translate(table, /)

Replace each character in the string using the given translation table.

table
Translation table, which must be a mapping of Unicode ordinals to Unicode ordinals, strings, or None.

The table must implement lookup/indexing via __getitem__, for instance a dictionary or list. If this operation raises LookupError, the character is left untouched. Characters mapped to None are deleted.

upper(): Return a copy of the string converted to uppercase.

startswith(prefix[, start[, end]]) → bool: Return True if S starts with the specified prefix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. prefix can also be a tuple of strings to try.

endswith(suffix[, start[, end]]) → bool: Return True if S ends with the specified suffix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. suffix can also be a tuple of strings to try.

removeprefix(prefix, /)

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

removesuffix(suffix, /)

Return a str with the given suffix string removed if present.

If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.

isascii()

Return True if all characters in the string are ASCII, False otherwise.

ASCII characters have code points in the range U+0000-U+007F. Empty string is ASCII too.

islower()

Return True if the string is a lowercase string, False otherwise.

A string is lowercase if all cased characters in the string are lowercase and there is at least one cased character in the string.

isupper()

Return True if the string is an uppercase string, False otherwise.

A string is uppercase if all cased characters in the string are uppercase and there is at least one cased character in the string.

istitle()

Return True if the string is a title-cased string, False otherwise.

In a title-cased string, upper- and title-case characters may only follow uncased characters and lowercase characters only cased ones.

isspace()

Return True if the string is a whitespace string, False otherwise.

A string is whitespace if all characters in the string are whitespace and there is at least one character in the string.

isdecimal()

Return True if the string is a decimal string, False otherwise.

A string is a decimal string if all characters in the string are decimal and there is at least one character in the string.

isdigit()

Return True if the string is a digit string, False otherwise.

A string is a digit string if all characters in the string are digits and there is at least one character in the string.

isnumeric()

Return True if the string is a numeric string, False otherwise.

A string is numeric if all characters in the string are numeric and there is at least one character in the string.

isalpha()

Return True if the string is an alphabetic string, False otherwise.

A string is alphabetic if all characters in the string are alphabetic and there is at least one character in the string.

isalnum()

Return True if the string is an alpha-numeric string, False otherwise.

A string is alpha-numeric if all characters in the string are alpha-numeric and there is at least one character in the string.

isidentifier()

Return True if the string is a valid Python identifier, False otherwise.

Call keyword.iskeyword(s) to test whether string s is a reserved identifier, such as “def” or “class”.

isprintable()

Return True if the string is printable, False otherwise.

A string is printable if all of its characters are considered printable in repr() or if it is empty.

zfill(width, /)

Pad a numeric string with zeros on the left, to fill a field of the given width.

The string is never truncated.

format(*args, **kwargs) → str: Return a formatted version of S, using substitutions from args and kwargs. The substitutions are identified by braces (‘{’ and ‘}’).

format_map(mapping) → str: Return a formatted version of S, using substitutions from mapping. The substitutions are identified by braces (‘{’ and ‘}’).

static maketrans()

Return a translation table usable for str.translate().

If there is only one argument, it must be a dictionary mapping Unicode ordinals (integers) or characters to Unicode ordinals, strings or None. Character keys will be then converted to ordinals. If there are two arguments, they must be strings of equal length, and in the resulting dictionary, each character in x will be mapped to the character at the same position in y. If there is a third argument, it must be a string, whose characters will be mapped to None in the result.

__dir__(): Returns public methods and other interesting attributes.

__init__(*args, **kwds)

class kailash.runtime.trust.AuditEvent(event_id: str, timestamp: str, actor: str, action: str, resource: str, outcome: str, prev_hash: str, hash: str, parent_anchor_id: str | None = None, duration_ms: float | None = None, metadata: Dict[str, ~typing.Any]=<factory>, event_type: str | None = None, severity: str | None = None, description: str | None = None, user_id: str | None = None, tenant_id: str | None = None, resource_id: str | None = None, ip_address: str | None = None, user_agent: str | None = None, session_id: str | None = None, correlation_id: str | None = None, trace_id: str | None = None, workflow_id: str | None = None, node_id: str | None = None, agent_id: str | None = None, human_origin_id: str | None = None)[source]

Bases: object

Canonical audit event – SPEC-08 single source of truth.

Consolidates every audit record type that previously existed in scattered modules (nodes/admin/audit_log.AuditEvent, runtime/trust/audit.AuditEvent, trust/immutable_audit_log.AuditEntry, pact.audit.AuditAnchor). All of those are now either migrated to import from here or are deprecated-and-scheduled-for-deletion.

The core fields (event_id, timestamp, actor, action, resource, outcome, prev_hash, hash, parent_anchor_id, duration_ms, metadata) form the Merkle-chained audit trail. The extended fields are optional attributes for domain-specific consumers (workflow runtime, enterprise admin logging) and do not participate in the hash chain.

Variables:

event_id (str) – Unique identifier for this event.
timestamp (str) – UTC timestamp as ISO-8601 string for deterministic hashing.
actor (str) – Identifier of the entity that performed the action.
action (str) – Short description of what was done.
resource (str) – Identifier of the resource acted upon.
outcome (str) – Result of the action (success, failure, denied, error).
prev_hash (str) – SHA-256 hex digest of the previous event (or genesis sentinel).
hash (str) – SHA-256 hex digest covering all core fields above.
parent_anchor_id (str | None) – Link to triggering action for causal chains.
duration_ms (float | None) – Execution duration in milliseconds.
metadata (Dict[str, Any]) – Arbitrary additional context.
event_type (str | None) – High-level event category (see AuditEventType).
severity (str | None) – Severity level string (e.g. "low", "high", "critical").
description (str | None) – Human-readable event description.
user_id (str | None) – Authenticated user identifier (enterprise audit).
tenant_id (str | None) – Multi-tenant isolation key.
resource_id (str | None) – Structured resource identifier (distinct from resource).
ip_address (str | None) – Source IP address of the actor.
user_agent (str | None) – User-Agent string of the actor.
session_id (str | None) – Session correlation ID.
correlation_id (str | None) – Request/trace correlation ID.
trace_id (str | None) – Distributed trace ID for correlation across services.
workflow_id (str | None) – Workflow run identifier (runtime audit).
node_id (str | None) – Node instance identifier (runtime audit).
agent_id (str | None) – Agent identifier from delegation chain.
human_origin_id (str | None) – Identifier of the human at the root of the delegation chain.

Parameters:

event_id (str)
timestamp (str)
actor (str)
action (str)
resource (str)
outcome (str)
prev_hash (str)
hash (str)
parent_anchor_id (str | None)
duration_ms (float | None)
metadata (Dict[str, Any])
event_type (str | None)
severity (str | None)
description (str | None)
user_id (str | None)
tenant_id (str | None)
resource_id (str | None)
ip_address (str | None)
user_agent (str | None)
session_id (str | None)
correlation_id (str | None)
trace_id (str | None)
workflow_id (str | None)
node_id (str | None)
agent_id (str | None)
human_origin_id (str | None)

event_id: str

timestamp: str

actor: str

action: str

resource: str

outcome: str

prev_hash: str

hash: str

parent_anchor_id: str | None = None

duration_ms: float | None = None

metadata: Dict[str, Any]

event_type: str | None = None

severity: str | None = None

description: str | None = None

user_id: str | None = None

tenant_id: str | None = None

resource_id: str | None = None

ip_address: str | None = None

user_agent: str | None = None

session_id: str | None = None

correlation_id: str | None = None

trace_id: str | None = None

workflow_id: str | None = None

node_id: str | None = None

agent_id: str | None = None

human_origin_id: str | None = None

to_dict() → Dict[str, Any][source]

Serialize to a plain dictionary.

All extended fields are included; None values are preserved so round-trips are lossless.

Return type:: Dict[str, Any]

classmethod from_dict(data: Dict[str, Any]) → AuditEvent[source]

Deserialize from a plain dictionary.

Unknown extra keys are ignored. Missing optional keys default to None or the field default.

Parameters:: data (Dict[str, Any])
Return type:: AuditEvent

verify_integrity() → bool[source]

Verify this event’s hash matches its content.

Uses hmac.compare_digest to prevent timing side-channels. Only the core Merkle-chained fields participate in the hash; extended domain fields are excluded so pre-SPEC-08 chains remain verifiable after migration.

Returns:: True if the hash is valid.
Return type:: bool

Parameters:

event_id (str)
timestamp (str)
actor (str)
action (str)
resource (str)
outcome (str)
prev_hash (str)
hash (str)
parent_anchor_id (str | None)
duration_ms (float | None)
metadata (Dict[str, Any])
event_type (str | None)
severity (str | None)
description (str | None)
user_id (str | None)
tenant_id (str | None)
resource_id (str | None)
ip_address (str | None)
user_agent (str | None)
session_id (str | None)
correlation_id (str | None)
trace_id (str | None)
workflow_id (str | None)
node_id (str | None)
agent_id (str | None)
human_origin_id (str | None)

Return type:

None

class kailash.runtime.trust.RuntimeAuditGenerator(audit_store: Any | None = None, enabled: bool = True, log_to_stdout: bool = False)[source]

Bases: object

Audit trail generator for runtime workflow execution.

Generates EATP-compliant audit events for workflow and node execution, with optional persistence to Kaizen AuditStore.

Features:

In-memory event storage for queries
Optional Kaizen AuditStore integration
Optional stdout logging
Extraction of trust context fields

Example

>>> generator = RuntimeAuditGenerator(enabled=True)
>>> trust_ctx = RuntimeTrustContext(trace_id="trace-123")
>>> await generator.workflow_started("run-1", "wf-1", trust_ctx)
>>> events = generator.get_events()

Parameters:

audit_store (Optional[Any])
enabled (bool)
log_to_stdout (bool)

__init__(audit_store: Any | None = None, enabled: bool = True, log_to_stdout: bool = False) → None[source]

Initialize the RuntimeAuditGenerator.

Parameters:

audit_store (Any | None) – Optional Kaizen AuditStore for persistence
enabled (bool) – Whether audit generation is enabled (default True)
log_to_stdout (bool) – Whether to log events to stdout (default False)

Return type:

None

async workflow_started(run_id: str, workflow_name: str, trust_context: 'RuntimeTrustContext' | None) → AuditEvent[source]

Record workflow start event.

Parameters:

run_id (str) – Unique run identifier
workflow_name (str) – Name of the workflow
trust_context (Optional['RuntimeTrustContext']) – RuntimeTrustContext for trust information

Returns:

The created AuditEvent

Return type:

async workflow_completed(run_id: str, duration_ms: int, trust_context: 'RuntimeTrustContext' | None) → AuditEvent[source]

Record workflow completion event.

Parameters:

run_id (str) – Unique run identifier
duration_ms (int) – Execution duration in milliseconds
trust_context (Optional['RuntimeTrustContext']) – RuntimeTrustContext for trust information

Returns:

The created AuditEvent

Return type:

async workflow_failed(run_id: str, error: str, duration_ms: int, trust_context: 'RuntimeTrustContext' | None) → AuditEvent[source]

Record workflow failure event.

Parameters:

run_id (str) – Unique run identifier
error (str) – Error message or description
duration_ms (int) – Execution duration in milliseconds
trust_context (Optional['RuntimeTrustContext']) – RuntimeTrustContext for trust information

Returns:

The created AuditEvent

Return type:

async node_executed(run_id: str, node_id: str, node_type: str, duration_ms: int, trust_context: 'RuntimeTrustContext' | None) → AuditEvent[source]

Record successful node execution event.

Parameters:

run_id (str) – Unique run identifier
node_id (str) – Node instance ID
node_type (str) – Type of node executed
duration_ms (int) – Execution duration in milliseconds
trust_context (Optional['RuntimeTrustContext']) – RuntimeTrustContext for trust information

Returns:

The created AuditEvent

Return type:

async node_failed(run_id: str, node_id: str, node_type: str, error: str, duration_ms: int, trust_context: 'RuntimeTrustContext' | None) → AuditEvent[source]

Record node failure event.

Parameters:

run_id (str) – Unique run identifier
node_id (str) – Node instance ID
node_type (str) – Type of node that failed
error (str) – Error message or description
duration_ms (int) – Execution duration in milliseconds
trust_context (Optional['RuntimeTrustContext']) – RuntimeTrustContext for trust information

Returns:

The created AuditEvent

Return type:

async trust_verification_performed(run_id: str, target: str, allowed: bool, reason: str, trust_context: 'RuntimeTrustContext' | None) → AuditEvent[source]

Record trust verification event.

Parameters:

run_id (str) – Unique run identifier
target (str) – What was verified (e.g., “workflow:name” or “node:type:id”)
allowed (bool) – Whether the verification allowed the operation
reason (str) – Reason for the verification result
trust_context (Optional['RuntimeTrustContext']) – RuntimeTrustContext for trust information

Returns:

The created AuditEvent

Return type:

async resource_accessed(run_id: str, resource: str, action: str, result: str, trust_context: 'RuntimeTrustContext' | None) → AuditEvent[source]

Record resource access event.

Parameters:

run_id (str) – Unique run identifier
resource (str) – Resource path or identifier
action (str) – Action performed (e.g., “read”, “write”)
result (str) – Result of access (“success”, “failure”, “denied”)
trust_context (Optional['RuntimeTrustContext']) – RuntimeTrustContext for trust information

Returns:

The created AuditEvent

Return type: