espennilsen/extensions/pi-a2a

pi-a2a Long-Running Tasks Skill

Overview

The pi-a2a extension supports long-running tasks that can execute for hours or days without timeouts. This is essential for:

• Data processing pipelines
• Batch operations
• Research and aggregation tasks
• External API jobs with unpredictable duration
• Any A2A task that exceeds the standard timeout

When to Use

Use long-running tasks when:

• Task execution time is unpredictable or known to exceed 10 minutes
• The remote agent is processing large datasets
• Task involves multiple external API calls
• You want the task to survive Pi restarts

Don't use for:

• Quick queries (< 5 minutes)
• Interactive conversations
• Tasks requiring immediate feedback

Configuration

Enable in settings.json:

{
  "pi-a2a": {
    "longRunningTasks": {
      "enabled": true,
      "maxTaskAgeHours": 168,
      "resumeRetryAttempts": 3,
      "resumeRetryDelayMs": 5000,
      "pollingIntervalMs": 300000
    }
  }
}
```text

### Configuration Options

| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `enabled` | boolean | `false` | Enable long-running task support |
| `maxTaskAgeHours` | number | `168` (7 days) | Maximum task retention period |
| `resumeRetryAttempts` | number | `3` | Retry attempts for resume failures |
| `resumeRetryDelayMs` | number | `5000` | Delay between retries |
| `pollingIntervalMs` | number | `300000` (5 min) | Hub polling interval |

## How It Works

### Architecture

┌─────────────────────────────────────────────────────────────┐ │ 1. Task Initiated │ │ - Task state saved to SQLite │ │ - Session ID assigned │ │ - Agent continues other work │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ 2. Background Polling (every 5 min) │ │ - Checks hub for task completion │ │ - Detects state changes │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ 3. Task Completes │ │ - State updated in SQLite │ │ - Resume request queued │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ 4. Smart Resume Queue │ │ - Waits for agent to be idle │ │ - Processes one request at a time │ │ - Validates session ID (prevents stale callbacks) │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ 5. Response Delivered │ │ - Completion message injected into chat │ │ - Task result available │ └─────────────────────────────────────────────────────────────┘

### Session Persistence

Tasks are stored in `db/a2a-long-running.db` with:
- `taskId` - A2A task identifier
- `contextId` - Conversation context
- `sessionId` - Pi session that owns the task
- `state` - Current task state
- `createdAt` / `lastUpdatedAt` - Timestamps
- `response` / `error` - Task result

### Resume Queue

The resume queue ensures responses are delivered at the right time:

1. **Agent Busy**: Queue the resume request
2. **Agent Idle**: Process immediately
3. **Session Mismatch**: Skip (task belongs to old session)
4. **Failure**: Retry with backoff (up to `resumeRetryAttempts`)

## Usage Patterns

### Pattern 1: Long-Running Data Processing

```typescript
// Send task to data processing agent
const result = await a2a_send({
  agent: "data-processor",
  message: "Process this 10GB dataset...",
});

// Agent continues other work immediately
// Task result delivered when processing completes (hours later)

Pattern 2: Multi-Agent Pipeline

// Start long-running pipeline
const pipelineTask = await a2a_send({
  agent: "pipeline-orchestrator",
  message: "Run full ETL pipeline",
});

// Check status periodically or wait for completion notification
// Result delivered automatically when pipeline finishes

Pattern 3: Research Aggregation

// Start research task
const researchTask = await a2a_send({
  agent: "research-agent",
  message: "Aggregate all news about AI safety from past month",
});

// Agent can handle other requests while research runs
// Results delivered when aggregation completes

Best Practices

✅ Do

1. Enable for appropriate tasks: Use when tasks exceed 10 minutes
2. Monitor task age: Set maxTaskAgeHours appropriately for your use case
3. Handle failures: Check for error messages in completion notifications
4. Test session recovery: Verify tasks survive Pi restarts
5. Set reasonable polling intervals: Balance responsiveness with resource usage

❌ Don't

1. Don't enable for all tasks: Overhead not needed for quick operations
2. Don't set polling too low: < 60 seconds creates unnecessary load
3. Don't rely on immediate responses: Long-running tasks are asynchronous
4. Don't ignore session boundaries: Tasks belong to specific sessions

Monitoring

Check Task Status

# View pending long-running tasks
sqlite3 db/a2a-long-running.db "SELECT task_id, state, created_at FROM long_running_tasks WHERE state NOT IN ('completed', 'failed');"

Check Resume Queue

# View queued resume requests
sqlite3 db/a2a-long-running.db "SELECT task_id, priority, retry_count FROM resume_queue ORDER BY enqueued_at;"

Logs to Monitor

• long_running_task_saved - Task state persisted
• long_running_task_completed - Task finished
• resume_request_enqueued - Resume queued
• resume_queue_processed - Response delivered
• resume_queue_retry_scheduled - Retry scheduled
• long_running_task_poll_error - Polling error

Troubleshooting

Task Not Completing

Symptoms: Task stuck in "working" state for extended period

Solutions:

1. Check remote agent health
2. Verify hub connectivity
3. Increase pollingIntervalMs if hub is rate-limited
4. Check logs for long_running_task_poll_error

Resume Not Processing

Symptoms: Task completed but response not delivered

Solutions:

1. Check if agent is busy (resume waits for idle)
2. Verify session ID matches current session
3. Check retry count (may have exhausted retries)
4. Review resume_queue_* logs

Session Mismatch

Symptoms: resume_queue_stale_session in logs

Cause: Task belongs to previous Pi session

Solutions:

1. This is expected behavior - prevents cross-session contamination
2. Task result available in previous session's data
3. Consider shorter maxTaskAgeHours if this occurs frequently

Limitations

1. Hub Dependency: Requires A2A hub for polling (can't poll direct agents)
2. Session Boundaries: Tasks don't cross session boundaries
3. Polling Latency: Completion detected within polling interval (default: 5 min)
4. Storage: Tasks consume SQLite storage (pruned after maxTaskAgeHours)

Example Configuration

Development

{
  "pi-a2a": {
    "longRunningTasks": {
      "enabled": true,
      "maxTaskAgeHours": 24,
      "resumeRetryAttempts": 5,
      "pollingIntervalMs": 60000
    }
  }
}

Production

{
  "pi-a2a": {
    "longRunningTasks": {
      "enabled": true,
      "maxTaskAgeHours": 168,
      "resumeRetryAttempts": 3,
      "pollingIntervalMs": 300000
    }
  }
}

Minimal Overhead

{
  "pi-a2a": {
    "longRunningTasks": {
      "enabled": false
    }
  }
}

Related Features

• A2A Hub: Task status polling requires hub registration
• Session Persistence: Pi stores sessions on disk for recovery
• Smart Resume Queue: Respects agent workload before delivering responses
• Task Timeout: Standard tasks still have taskTimeoutMs (default: 10 min)

Version History

• v0.1.0: Initial implementation
  • LongRunningTaskStore for SQLite persistence
  • Session ID tracking
  • Smart resume queue
  • Background polling