feat: Python task execution and enable/disable controls

[mortenoh]

Summary

This PR introduces comprehensive enhancements to the task execution system, adding Python function execution capabilities, task enable/disable controls with automatic orphaned task validation, type-based dependency injection, and type system improvements.

Features

1. Python Task Execution

• TaskRegistry: Register Python functions (sync or async) for execution as tasks
• Task Types: New task_type field distinguishes "shell" vs "python" tasks
• Parameters: JSON parameters dict passed as kwargs to Python functions
• Artifact Structure: Python tasks store result/error instead of stdout/stderr
• Error Handling: Full exception capture with type, message, and traceback

Example with Parameters:

# Register function
@TaskRegistry.register("calculate_sum")
async def calculate_sum(a: int, b: int) -> dict:
    return {"result": a + b, "operation": "sum"}

# Create task with parameters
TaskIn(
    command="calculate_sum",
    task_type="python",
    parameters={"a": 10, "b": 32}  # Passed as kwargs to function
)

2. Type-Based Dependency Injection

• Automatic Injection: Framework services injected based on function parameter type hints
• Injectable Types: AsyncSession, Database, ArtifactManager, JobScheduler
• User Parameters: Primitives and generic types from task.parameters
• Optional Support: Handles Optional[AsyncSession] and other Optional types correctly
• Flexible Naming: Parameter names don't matter, only types (e.g., session, db, conn all work for AsyncSession)

Example with Injection:

# Function with injected database session
@TaskRegistry.register("query_task_count")
async def query_task_count(session: AsyncSession) -> dict:
    """Uses injected session - no user parameters needed."""
    stmt = select(func.count()).select_from(Task)
    result = await session.execute(stmt)
    return {"total_tasks": result.scalar()}

# Execute - session injected automatically
TaskIn(
    command="query_task_count",
    task_type="python",
    parameters={}  # Empty - session provided by framework
)

Mixed Parameters:

@TaskRegistry.register("process_with_db")
async def process_with_db(
    input_text: str,        # From task.parameters
    count: int,             # From task.parameters
    session: AsyncSession,  # Injected by framework
) -> dict:
    # Use both user params and injected services
    return {"processed": input_text, "count": count}

3. Task Enable/Disable Controls

• Enabled Field: Boolean enabled field (default: true) to control task execution
• Execution Prevention: Disabled tasks cannot be executed (raises ValueError)
• API Filtering: Query parameter ?enabled=true/false for listing tasks
• Repository Methods: find_by_enabled() and find_all(enabled=...)
• Soft Delete Pattern: Preserves task history while preventing execution

Example:

# Create disabled task
TaskIn(command="echo test", enabled=False)

# Filter tasks by enabled status
GET /api/v1/tasks?enabled=true   # Only enabled
GET /api/v1/tasks?enabled=false  # Only disabled

4. Orphaned Task Validation

• Startup Utility: validate_and_disable_orphaned_tasks() checks Python tasks
• Auto-Disable: Orphaned tasks (referencing unregistered functions) auto-disabled
• Structured Logging: Warnings with task IDs, function names, and counts
• Production Ready: Non-destructive, idempotent, preserves audit trail

Example:

app = (
    ServiceBuilder(info=info)
    .with_tasks()
    .on_startup(validate_and_disable_orphaned_tasks)  # Auto-disable orphaned
    .build()
)

5. Read-Only API Pattern

• Security Example: readonly_task_api.py demonstrates pre-seeded tasks
• CRUD Permissions: Support for create/read/update/delete flags
• Version Control: Tasks defined in code, deployed via startup hooks
• Command Injection Prevention: No runtime task creation in production

Example:

task_permissions = CrudPermissions(
    create=False, read=True, update=False, delete=False
)

app = ServiceBuilder(info=info).with_tasks(permissions=task_permissions).build()

6. Type System Improvements

• Renamed Type: SerializableDict → JsonSafe for clarity
• Better Name: Reflects actual behavior (accepts any type, ensures JSON-safe serialization)
• Comprehensive Docs: Multi-line documentation explaining serialization behavior
• Graceful Handling: Non-JSON-serializable values (PyTorch models, sklearn models) replaced with metadata in API responses while preserving originals in storage

Behavior:

# JSON-serializable: works as expected
{"result": 42, "status": "ok"} → {"result": 42, "status": "ok"}

# Non-serializable: replaced with metadata in API responses
{"model": <PyTorch model>} → {"model": {"_type": "Module", "_module": "torch.nn", ...}}
# Original object remains in storage via PickleType

Examples

New Examples:

• examples/python_task_execution_api.py - Complete Python task service with:
  • Multiple registered functions (async and sync)
  • Tasks with parameters ({"a": 10, "b": 32})
  • Dependency injection example (database queries)
  • Error handling demonstrations
  • Disabled and orphaned task examples
• examples/readonly_task_api.py - Secure read-only task API

New Documentation:

• examples/docs/task_python_execution.md - Comprehensive cURL guide
• examples/docs/task_python_execution.postman_collection.json - Full Postman collection

Documentation

Updated Guides:

• docs/guides/task-execution.md - Comprehensive updates including:
  • Python task execution workflow and examples
  • TaskRegistry usage and patterns
  • Parameters field documentation and examples
  • Dependency injection documentation
  • Enable/disable API documentation
  • Orphaned task handling
  • Read-only API security patterns

Design Documents:

• designs/python-tasks-and-scheduling.md - Architecture and implementation details

Tests

New Test Files:

• tests/test_task_registry.py - TaskRegistry functionality (9 tests)
• tests/test_task_repository.py - Enabled filtering (6 tests)
• tests/test_task_validation.py - Orphaned task detection (7 tests)
• tests/test_task_injection.py - Type-based dependency injection (7 tests)
• tests/test_example_python_task_execution_api.py - Integration tests (9 tests)

Updated Tests:

• tests/test_manager_task.py - Added enabled/disabled execution tests
• tests/test_task_router.py - Added enabled filtering API tests

Test Results:

• 683 tests passed
• 6 tests skipped
• All linting checks passed
• No type errors

API Changes

New Fields

{
  "task_type": "python",               // "shell" or "python"
  "parameters": {"a": 10, "b": 32},   // Dict passed as kwargs (Python only)
  "enabled": true                      // Control execution
}

New Query Parameters

GET /api/v1/tasks?enabled=true   # Only enabled tasks
GET /api/v1/tasks?enabled=false  # Only disabled tasks

New Task Types

Shell Task (unchanged):

{
  "command": "echo 'Hello World'",
  "task_type": "shell"
}

Python Task (new):

{
  "command": "calculate_sum",
  "task_type": "python",
  "parameters": {"a": 10, "b": 32}
}

Python Task with Injection:

{
  "command": "query_task_count",
  "task_type": "python",
  "parameters": {}  // Session injected automatically
}

Error Responses

{
  "detail": "Cannot execute disabled task 01TASK..."
}

Migration

Database Changes:

• Added task_type column (default: "shell")
• Added parameters JSON column (nullable, stores parameter dict)
• Added enabled boolean column (default: true)
• Migration applied to initial schema (backwards compatible)

Usage

Python Task Registration

from chapkit import TaskRegistry

@TaskRegistry.register("my_function")
async def my_function(param: str) -> dict:
    return {"result": f"Processed {param}"}

Python Task with Dependency Injection

from sqlalchemy.ext.asyncio import AsyncSession

@TaskRegistry.register("db_query")
async def db_query(
    query_param: str,       # From task.parameters
    session: AsyncSession,  # Injected by framework
) -> dict:
    # Use session for database operations
    result = await session.execute(...)
    return {"results": result.all()}

Creating Python Tasks with Parameters

# Via API
POST /api/v1/tasks
{
  "command": "my_function",
  "task_type": "python",
  "parameters": {"param": "test_value"}
}

# Via code
TaskIn(
    command="my_function",
    task_type="python",
    parameters={"param": "test_value"}
)

Startup Validation

from chapkit import validate_and_disable_orphaned_tasks

async def validate_tasks_on_startup(app: FastAPI) -> None:
    await validate_and_disable_orphaned_tasks(app)

app = (
    ServiceBuilder(info=info)
    .with_tasks()
    .on_startup(validate_tasks_on_startup)
    .build()
)

Read-Only API

from chapkit.core.api.crud import CrudPermissions

task_permissions = CrudPermissions(
    create=False, read=True, update=False, delete=False
)

app = ServiceBuilder(info=info).with_tasks(permissions=task_permissions).build()

Breaking Changes

None. All changes are backwards compatible:

• Default task_type is "shell"
• Default enabled is true
• parameters is optional (null for shell tasks)
• Existing shell tasks continue to work unchanged
• SerializableDict renamed to JsonSafe (internal type, exported from core)
• Dependency injection only activates for injectable types

Checklist

• All tests pass (make test)
• All linting passes (make lint)
• Documentation updated
• Examples provided with parameters demonstrations
• Dependency injection examples and tests
• cURL guide and Postman collection created
• Migration included
• Backwards compatible
• Type system improvements with better naming