[Medium] Background Tasks Support

[marklicata]

Summary

Add support for long-running background tasks with ability to monitor, manage, and terminate them, enabling workflows like running tests, builds, or servers while continuing to interact with Amplifier.

Current State

Amplifier CLI runs bash commands synchronously:

• Commands block until completion
• Long-running commands freeze the session
• No way to run processes in background
• No way to monitor or kill running processes

Proposed Implementation

1. Background Task Model

amplifier> run the test suite in background
[Starting background task: pytest tests/ -v]
[Task ID: task_001]
[Running in background - use /tasks to monitor]

amplifier> while that runs, let's work on the API
[Continues normal conversation]

amplifier> /tasks
Background Tasks:
  task_001  pytest tests/ -v       RUNNING   2m 15s
  task_002  npm run build          COMPLETED 5m ago (exit 0)

2. Background-Aware Tools

New tools for background execution:

BashBackground - Start Background Task

{
    "tool": "BashBackground",
    "args": {
        "command": "pytest tests/ -v",
        "timeout_seconds": 600,
        "notify_on_complete": true
    }
}
# Returns: {"task_id": "task_001", "status": "started"}

TaskStatus - Check Task Status

{
    "tool": "TaskStatus",
    "args": {
        "task_id": "task_001"
    }
}
# Returns: {"status": "running", "runtime_seconds": 135, "output_lines": 42}

TaskOutput - Get Task Output

{
    "tool": "TaskOutput",
    "args": {
        "task_id": "task_001",
        "tail": 50  # Last N lines
    }
}
# Returns: {"output": "...", "truncated": true}

TaskKill - Terminate Task

{
    "tool": "TaskKill",
    "args": {
        "task_id": "task_001",
        "signal": "SIGTERM"  # or SIGKILL
    }
}
# Returns: {"killed": true}

3. Task Lifecycle

┌──────────┐     ┌──────────┐     ┌───────────┐
│  PENDING │────▶│ RUNNING  │────▶│ COMPLETED │
└──────────┘     └────┬─────┘     └───────────┘
                      │
                      │ timeout/kill
                      ▼
                ┌───────────┐
                │ CANCELLED │
                └───────────┘

4. Output Streaming

Background tasks write to a buffer that can be read incrementally:

@dataclass
class BackgroundTask:
    id: str
    command: str
    process: subprocess.Popen
    started_at: datetime
    status: TaskStatus
    output_buffer: OutputBuffer
    
class OutputBuffer:
    """Circular buffer for task output."""
    
    def __init__(self, max_lines: int = 10000):
        self._lines: deque[str] = deque(maxlen=max_lines)
        self._lock = threading.Lock()
    
    def append(self, line: str) -> None:
        with self._lock:
            self._lines.append(line)
    
    def tail(self, n: int = 50) -> list[str]:
        with self._lock:
            return list(self._lines)[-n:]
    
    def all(self) -> list[str]:
        with self._lock:
            return list(self._lines)

5. Notifications

When background tasks complete:

tasks:
  notifications:
    on_complete: true      # Notify when task finishes
    on_error: true         # Notify on non-zero exit
    sound: false           # System sound (macOS/Linux)

Notification appears in session:

amplifier> [working on something else...]

📋 Background task completed:
   task_001: pytest tests/ -v
   Status: COMPLETED (exit code 0)
   Duration: 3m 42s
   Use /tasks output task_001 to see results

amplifier> 

6. New Module Structure

src/amplifier_app_cli/tasks/
├── __init__.py
├── manager.py        # Task lifecycle management
├── runner.py         # Process spawning and monitoring
├── buffer.py         # Output buffering
├── notifications.py  # Completion notifications
└── tools.py          # Tool implementations

7. Core Interfaces

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Callable
import subprocess

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    TIMEOUT = "timeout"

@dataclass
class TaskResult:
    exit_code: int | None
    output: str
    error: str
    runtime_seconds: float

@dataclass
class BackgroundTask:
    id: str
    command: str
    status: TaskStatus
    created_at: datetime
    started_at: datetime | None = None
    completed_at: datetime | None = None
    process: subprocess.Popen | None = None
    result: TaskResult | None = None
    timeout_seconds: int | None = None
    notify_on_complete: bool = True
    
    @property
    def runtime_seconds(self) -> float:
        if not self.started_at:
            return 0
        end = self.completed_at or datetime.now()
        return (end - self.started_at).total_seconds()

class TaskManager:
    def __init__(
        self,
        max_concurrent: int = 5,
        on_complete: Callable[[BackgroundTask], None] | None = None
    ): ...
    
    def start(
        self,
        command: str,
        timeout_seconds: int | None = None,
        notify: bool = True,
        env: dict[str, str] | None = None,
        cwd: str | None = None
    ) -> BackgroundTask:
        """Start a new background task."""
        ...
    
    def get(self, task_id: str) -> BackgroundTask | None:
        """Get task by ID."""
        ...
    
    def list(
        self,
        status: TaskStatus | None = None,
        limit: int = 20
    ) -> list[BackgroundTask]:
        """List tasks, optionally filtered by status."""
        ...
    
    def output(
        self,
        task_id: str,
        tail: int | None = None
    ) -> str:
        """Get task output."""
        ...
    
    def kill(
        self,
        task_id: str,
        signal: str = "SIGTERM"
    ) -> bool:
        """Kill a running task."""
        ...
    
    def cleanup(self, max_age_hours: int = 24) -> int:
        """Remove old completed tasks."""
        ...

8. Slash Commands

/tasks                        # List all active tasks
/tasks list                   # Same as above
/tasks list --all             # Include completed tasks
/tasks status <id>            # Show task details
/tasks output <id>            # Show task output
/tasks output <id> --tail 100 # Last 100 lines
/tasks output <id> --follow   # Stream output (like tail -f)
/tasks kill <id>              # Terminate task
/tasks kill <id> --force      # Force kill (SIGKILL)
/tasks clean                  # Remove old completed tasks

9. Interactive Output Streaming

For /tasks output <id> --follow:

amplifier> /tasks output task_001 --follow

─── Task task_001: pytest tests/ -v ───
[Following output - Ctrl+C to stop]

tests/test_api.py::test_health_check PASSED
tests/test_api.py::test_create_user PASSED
tests/test_api.py::test_get_user PASSED
tests/test_auth.py::test_login PASSED
tests/test_auth.py::test_logout PASSED
█ [streaming...]

[Ctrl+C pressed]
amplifier>

10. AI Awareness

The AI should be able to:

1. Proactively suggest background execution for long tasks
2. Check on task status when relevant
3. Act on task results when completed

System prompt addition:

You have access to background task management:
- Use BashBackground for commands that may take >30 seconds
- Use TaskStatus to check on running tasks
- Use TaskOutput to see results when tasks complete
- Use TaskKill to stop runaway processes

Current background tasks:
{{#each active_tasks}}
- {{id}}: {{command}} ({{status}}, {{runtime}})
{{/each}}

11. Configuration

tasks:
  enabled: true
  max_concurrent: 5           # Max parallel tasks
  default_timeout: 3600       # 1 hour default
  output_buffer_lines: 10000  # Lines to keep per task
  auto_cleanup_hours: 24      # Remove old tasks after
  
  notifications:
    on_complete: true
    on_error: true
    sound: false
    
  # Commands to always run in background
  auto_background:
    - pattern: "pytest *"
      timeout: 600
    - pattern: "npm run build"
      timeout: 300
    - pattern: "cargo build *"
      timeout: 600

12. CLI Commands

# List tasks from command line
amplifier tasks list
amplifier tasks list --all

# Show task details
amplifier tasks show task_001

# Get output
amplifier tasks output task_001
amplifier tasks output task_001 --tail 100

# Kill task
amplifier tasks kill task_001

# Clean old tasks
amplifier tasks clean --older-than 24h

Acceptance Criteria

• BashBackground tool starts tasks in background
• TaskStatus returns current task state
• TaskOutput returns buffered output
• TaskKill terminates running tasks
• Task output buffered (circular buffer)
• Notifications when tasks complete
• /tasks slash commands work
• --follow streams output in real-time
• Timeout handling for long tasks
• Max concurrent task limit enforced
• Auto-cleanup of old completed tasks
• Tasks survive session reconnect
• Unit tests for task manager
• Integration tests for background execution

Related

• Depends on: None
• Enhances: Long-running workflows (tests, builds, servers)

Estimated Effort

Medium - 1.5-2 weeks

Files to Create/Modify

• src/amplifier_app_cli/tasks/ (new module)
• src/amplifier_app_cli/commands/tasks.py (new CLI commands)
• src/amplifier_app_cli/commands/slash.py (task slash commands)
• src/amplifier_app_cli/lib/tools.py (register new tools)