f07f04404d
- Add Tag DTOs (CreateTagDto, UpdateTagDto) with validation - Implement TagsService with CRUD operations - Add TagsController with authenticated endpoints - Support automatic slug generation from tag names - Add workspace isolation for tags - Include entry count in tag responses - Add findOrCreateTags method for entry creation/update - Implement comprehensive test coverage (29 tests passing) Endpoints: - GET /api/knowledge/tags - List workspace tags - POST /api/knowledge/tags - Create tag - GET /api/knowledge/tags/:slug - Get tag by slug - PUT /api/knowledge/tags/:slug - Update tag - DELETE /api/knowledge/tags/:slug - Delete tag - GET /api/knowledge/tags/:slug/entries - List entries with tag Related: KNOW-003
1183 lines
36 KiB
Markdown
1183 lines
36 KiB
Markdown
# Agent Orchestration Layer
|
|
|
|
**Version:** 1.0
|
|
**Status:** Design Phase
|
|
**Author:** Mosaic Stack Team
|
|
**Date:** 2025-01-29
|
|
|
|
## Table of Contents
|
|
|
|
1. [Problem Statement](#problem-statement)
|
|
2. [Architecture Overview](#architecture-overview)
|
|
3. [Database Schema](#database-schema)
|
|
4. [Valkey/Redis Key Patterns](#valkeyredis-key-patterns)
|
|
5. [API Endpoints](#api-endpoints)
|
|
6. [Coordinator Service Design](#coordinator-service-design)
|
|
7. [Recovery & Resilience Patterns](#recovery--resilience-patterns)
|
|
8. [Implementation Phases](#implementation-phases)
|
|
|
|
---
|
|
|
|
## Problem Statement
|
|
|
|
### Current Limitations
|
|
|
|
1. **No Persistent Orchestration**
|
|
- Agents spawned in sessions have no lifecycle management beyond the session
|
|
- Work continuity is lost when Gateway restarts or sessions end
|
|
- No coordination between agents working on the same project
|
|
|
|
2. **No Health Monitoring**
|
|
- No visibility into agent status after spawn
|
|
- Failed agents leave tasks in limbo
|
|
- No automated recovery from agent failures
|
|
|
|
3. **No Task Persistence**
|
|
- Task state only exists in session memory
|
|
- Users can't track long-running work
|
|
- No audit trail of agent activity
|
|
|
|
4. **Conversation Interference**
|
|
- Main session conversations derail active development work
|
|
- Context switching disrupts agent focus
|
|
- No separation between user chat and background work
|
|
|
|
### Requirements
|
|
|
|
The Agent Orchestration Layer must provide:
|
|
|
|
- **Persistent Task Management** — Task definitions, state, and history survive restarts
|
|
- **Autonomous Coordination** — Independent of main session; agents work continuously
|
|
- **Health Monitoring** — Real-time tracking of agent status and progress
|
|
- **Automatic Recovery** — Detect failures and resume work intelligently
|
|
- **Complete Audit Trail** — Every state transition logged for debugging and accountability
|
|
- **Multi-Workspace Support** — Row-level security and tenant isolation
|
|
|
|
---
|
|
|
|
## Architecture Overview
|
|
|
|
### High-Level Design
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────────────┐
|
|
│ API Layer │
|
|
│ (NestJS Controllers + Guards + Interceptors) │
|
|
└────────────┬─────────────────────────────────────┬──────────────┘
|
|
│ │
|
|
│ │
|
|
┌────────▼────────┐ ┌────────▼─────────┐
|
|
│ Task Manager │ │ Agent Manager │
|
|
│ Service │ │ Service │
|
|
└────────┬────────┘ └────────┬─────────┘
|
|
│ │
|
|
│ ┌───────────────────┐ │
|
|
└─────────► Coordinator ◄──────┘
|
|
│ Service │
|
|
└─────────┬─────────┘
|
|
│
|
|
┌────────────────────┼────────────────────┐
|
|
│ │ │
|
|
┌───────▼────────┐ ┌────────▼────────┐ ┌──────▼──────┐
|
|
│ PostgreSQL │ │ Valkey/Redis │ │ Gateway │
|
|
│ (Persistent) │ │ (Runtime) │ │ (Agent │
|
|
│ │ │ │ │ Spawner) │
|
|
└────────────────┘ └─────────────────┘ └─────────────┘
|
|
```
|
|
|
|
### Component Responsibilities
|
|
|
|
| Component | Responsibility |
|
|
|-----------|----------------|
|
|
| **Task Manager** | CRUD operations on tasks, state transitions, assignment logic |
|
|
| **Agent Manager** | Agent lifecycle, health tracking, session management |
|
|
| **Coordinator** | Heartbeat processing, failure detection, recovery orchestration |
|
|
| **PostgreSQL** | Persistent storage of tasks, agents, sessions, logs |
|
|
| **Valkey/Redis** | Runtime state, heartbeats, quick lookups, pub/sub |
|
|
| **Gateway** | Agent spawning, session management, message routing |
|
|
|
|
---
|
|
|
|
## Database Schema
|
|
|
|
### Task State Machine
|
|
|
|
```
|
|
┌──────────────┐
|
|
│ PENDING │
|
|
└──────┬───────┘
|
|
│
|
|
┌──────▼───────┐
|
|
┌────┤ ASSIGNED ├────┐
|
|
│ └──────┬───────┘ │
|
|
│ │ │
|
|
┌──────▼───────┐ │ ┌────────▼──────┐
|
|
│ RUNNING │ │ │ PAUSED │
|
|
└──────┬───────┘ │ └────────┬──────┘
|
|
│ │ │
|
|
│ ┌──────▼───────┐ │
|
|
└────► COMPLETED ◄────┘
|
|
└──────┬───────┘
|
|
│
|
|
┌──────▼───────┐
|
|
│ ARCHIVED │
|
|
└──────────────┘
|
|
|
|
┌──────────────┐
|
|
│ FAILED │
|
|
└──────┬───────┘
|
|
│
|
|
┌──────▼───────┐
|
|
│ ABORTED │────┐
|
|
└──────────────┘ │
|
|
│ │
|
|
└────────────┘
|
|
(retry → PENDING)
|
|
```
|
|
|
|
### New Tables
|
|
|
|
#### `agent_tasks`
|
|
|
|
Extends existing task management with orchestration-specific fields.
|
|
|
|
```sql
|
|
CREATE TYPE task_orchestration_status AS ENUM (
|
|
'PENDING', -- Awaiting assignment
|
|
'ASSIGNED', -- Assigned to an agent
|
|
'RUNNING', -- Agent actively working
|
|
'PAUSED', -- User paused
|
|
'COMPLETED', -- Successfully finished
|
|
'FAILED', -- Failed with error
|
|
'ABORTED', -- Terminated (stale agent, manual cancel)
|
|
'ARCHIVED' -- Historical record
|
|
);
|
|
|
|
CREATE TABLE agent_tasks (
|
|
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
|
|
workspace_id UUID NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
|
|
|
|
-- Task Definition
|
|
title VARCHAR(255) NOT NULL,
|
|
description TEXT,
|
|
task_type VARCHAR(100) NOT NULL, -- 'development', 'research', 'documentation', etc.
|
|
|
|
-- Status & Priority
|
|
status task_orchestration_status DEFAULT 'PENDING',
|
|
priority INT DEFAULT 5, -- 1 (low) to 10 (high)
|
|
|
|
-- Assignment
|
|
agent_id UUID REFERENCES agents(id) ON DELETE SET NULL,
|
|
session_key VARCHAR(255), -- Current active session
|
|
|
|
-- Progress Tracking
|
|
progress_percent INT DEFAULT 0 CHECK (progress_percent BETWEEN 0 AND 100),
|
|
current_step TEXT,
|
|
estimated_completion_at TIMESTAMPTZ,
|
|
|
|
-- Retry Logic
|
|
retry_count INT DEFAULT 0,
|
|
max_retries INT DEFAULT 3,
|
|
retry_backoff_seconds INT DEFAULT 300, -- 5 minutes
|
|
last_error TEXT,
|
|
|
|
-- Dependencies
|
|
depends_on UUID[] DEFAULT ARRAY[]::UUID[], -- Array of task IDs
|
|
blocks UUID[] DEFAULT ARRAY[]::UUID[], -- Tasks blocked by this one
|
|
|
|
-- Context
|
|
input_context JSONB DEFAULT '{}', -- Input data/params for agent
|
|
output_result JSONB, -- Final result from agent
|
|
checkpoint_data JSONB DEFAULT '{}', -- Resumable state
|
|
|
|
-- Metadata
|
|
metadata JSONB DEFAULT '{}',
|
|
tags VARCHAR(100)[] DEFAULT ARRAY[]::VARCHAR[],
|
|
|
|
-- Audit
|
|
created_by UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
created_at TIMESTAMPTZ DEFAULT NOW(),
|
|
updated_at TIMESTAMPTZ DEFAULT NOW(),
|
|
assigned_at TIMESTAMPTZ,
|
|
started_at TIMESTAMPTZ,
|
|
completed_at TIMESTAMPTZ,
|
|
failed_at TIMESTAMPTZ,
|
|
|
|
CONSTRAINT fk_workspace FOREIGN KEY (workspace_id) REFERENCES workspaces(id),
|
|
CONSTRAINT fk_agent FOREIGN KEY (agent_id) REFERENCES agents(id)
|
|
);
|
|
|
|
CREATE INDEX idx_agent_tasks_workspace ON agent_tasks(workspace_id);
|
|
CREATE INDEX idx_agent_tasks_status ON agent_tasks(status) WHERE status IN ('PENDING', 'ASSIGNED', 'RUNNING');
|
|
CREATE INDEX idx_agent_tasks_agent ON agent_tasks(agent_id) WHERE agent_id IS NOT NULL;
|
|
CREATE INDEX idx_agent_tasks_priority ON agent_tasks(priority DESC, created_at ASC) WHERE status = 'PENDING';
|
|
CREATE INDEX idx_agent_tasks_depends_on ON agent_tasks USING gin(depends_on);
|
|
CREATE INDEX idx_agent_tasks_tags ON agent_tasks USING gin(tags);
|
|
```
|
|
|
|
#### `agent_task_logs`
|
|
|
|
Immutable audit trail of all task state transitions and agent actions.
|
|
|
|
```sql
|
|
CREATE TYPE task_log_level AS ENUM ('DEBUG', 'INFO', 'WARN', 'ERROR', 'CRITICAL');
|
|
|
|
CREATE TABLE agent_task_logs (
|
|
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
|
|
workspace_id UUID NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
|
|
task_id UUID NOT NULL REFERENCES agent_tasks(id) ON DELETE CASCADE,
|
|
agent_id UUID REFERENCES agents(id) ON DELETE SET NULL,
|
|
|
|
-- Log Entry
|
|
level task_log_level DEFAULT 'INFO',
|
|
event VARCHAR(100) NOT NULL, -- 'state_transition', 'progress_update', 'error', etc.
|
|
message TEXT,
|
|
details JSONB DEFAULT '{}',
|
|
|
|
-- State Snapshot
|
|
previous_status task_orchestration_status,
|
|
new_status task_orchestration_status,
|
|
|
|
-- Context
|
|
session_key VARCHAR(255),
|
|
stack_trace TEXT,
|
|
|
|
created_at TIMESTAMPTZ DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX idx_task_logs_task ON agent_task_logs(task_id, created_at DESC);
|
|
CREATE INDEX idx_task_logs_workspace ON agent_task_logs(workspace_id, created_at DESC);
|
|
CREATE INDEX idx_task_logs_level ON agent_task_logs(level) WHERE level IN ('ERROR', 'CRITICAL');
|
|
CREATE INDEX idx_task_logs_event ON agent_task_logs(event);
|
|
```
|
|
|
|
#### `agent_heartbeats`
|
|
|
|
Short-lived records for health monitoring (TTL enforced in Valkey).
|
|
|
|
```sql
|
|
CREATE TABLE agent_heartbeats (
|
|
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
|
|
workspace_id UUID NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
|
|
agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
|
|
|
|
-- Health Data
|
|
status VARCHAR(50) NOT NULL, -- 'healthy', 'degraded', 'stale'
|
|
current_task_id UUID REFERENCES agent_tasks(id) ON DELETE SET NULL,
|
|
progress_percent INT,
|
|
|
|
-- Resource Usage
|
|
memory_mb INT,
|
|
cpu_percent INT,
|
|
|
|
-- Timing
|
|
last_seen_at TIMESTAMPTZ DEFAULT NOW(),
|
|
next_expected_at TIMESTAMPTZ,
|
|
|
|
metadata JSONB DEFAULT '{}'
|
|
);
|
|
|
|
CREATE INDEX idx_heartbeats_agent ON agent_heartbeats(agent_id, last_seen_at DESC);
|
|
CREATE INDEX idx_heartbeats_stale ON agent_heartbeats(last_seen_at) WHERE status != 'healthy';
|
|
|
|
-- Cleanup function (run periodically)
|
|
CREATE OR REPLACE FUNCTION cleanup_old_heartbeats()
|
|
RETURNS void AS $$
|
|
BEGIN
|
|
DELETE FROM agent_heartbeats
|
|
WHERE last_seen_at < NOW() - INTERVAL '1 hour';
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
```
|
|
|
|
### Updates to Existing Tables
|
|
|
|
#### Extend `agents` table
|
|
|
|
```sql
|
|
-- Add orchestration fields to existing agents table
|
|
ALTER TABLE agents ADD COLUMN IF NOT EXISTS coordinator_enabled BOOLEAN DEFAULT true;
|
|
ALTER TABLE agents ADD COLUMN IF NOT EXISTS max_concurrent_tasks INT DEFAULT 1;
|
|
ALTER TABLE agents ADD COLUMN IF NOT EXISTS heartbeat_interval_seconds INT DEFAULT 30;
|
|
ALTER TABLE agents ADD COLUMN IF NOT EXISTS stale_threshold_seconds INT DEFAULT 180; -- 3 minutes
|
|
ALTER TABLE agents ADD COLUMN IF NOT EXISTS capabilities VARCHAR(100)[] DEFAULT ARRAY[]::VARCHAR[];
|
|
|
|
CREATE INDEX idx_agents_coordinator ON agents(coordinator_enabled) WHERE coordinator_enabled = true;
|
|
```
|
|
|
|
---
|
|
|
|
## Valkey/Redis Key Patterns
|
|
|
|
Valkey is used for:
|
|
- **Real-time state** (fast reads/writes)
|
|
- **Pub/Sub messaging** (coordination events)
|
|
- **Distributed locks** (prevent race conditions)
|
|
- **Session data** (ephemeral context)
|
|
|
|
### Key Naming Convention
|
|
|
|
```
|
|
{namespace}:{workspace_id}:{entity_type}:{entity_id}:{property}
|
|
```
|
|
|
|
### Key Patterns
|
|
|
|
#### Task Queue
|
|
|
|
```redis
|
|
# Pending tasks (sorted set by priority + timestamp)
|
|
ZADD tasks:pending:{workspace_id} {priority}:{timestamp} {task_id}
|
|
|
|
# Assigned tasks (hash per agent)
|
|
HSET agent:tasks:{agent_id} {task_id} {assigned_at}
|
|
|
|
# Active sessions (set)
|
|
SADD tasks:active:{workspace_id} {task_id}:{session_key}
|
|
```
|
|
|
|
#### Agent Status
|
|
|
|
```redis
|
|
# Agent heartbeat (string with TTL)
|
|
SETEX agent:heartbeat:{agent_id} 60 "{\"status\":\"running\",\"task_id\":\"{task_id}\",\"timestamp\":{ts}}"
|
|
|
|
# Agent status (hash)
|
|
HSET agent:status:{agent_id}
|
|
status "running"
|
|
current_task "{task_id}"
|
|
last_heartbeat {timestamp}
|
|
|
|
# Stale agents (sorted set by last heartbeat)
|
|
ZADD coordinator:stale_agents {timestamp} {agent_id}
|
|
```
|
|
|
|
#### Coordination
|
|
|
|
```redis
|
|
# Coordinator lock (to prevent multiple coordinators)
|
|
SET coordinator:lock:{workspace_id} {coordinator_instance_id} NX EX 30
|
|
|
|
# Task assignment lock
|
|
SET task:assign_lock:{task_id} {agent_id} NX EX 5
|
|
|
|
# Recovery queue (list)
|
|
LPUSH coordinator:recovery_queue:{workspace_id} {task_id}
|
|
```
|
|
|
|
#### Pub/Sub Channels
|
|
|
|
```redis
|
|
# Task events
|
|
PUBLISH tasks:events:{workspace_id} "{\"event\":\"task_completed\",\"task_id\":\"{task_id}\"}"
|
|
|
|
# Agent events
|
|
PUBLISH agents:events:{workspace_id} "{\"event\":\"agent_stale\",\"agent_id\":\"{agent_id}\"}"
|
|
|
|
# Coordinator commands
|
|
PUBLISH coordinator:commands:{workspace_id} "{\"command\":\"reassign_task\",\"task_id\":\"{task_id}\"}"
|
|
```
|
|
|
|
#### Session Data (TTL: 1 hour)
|
|
|
|
```redis
|
|
# Session context (hash with TTL)
|
|
HSET session:context:{session_key}
|
|
workspace_id "{workspace_id}"
|
|
agent_id "{agent_id}"
|
|
task_id "{task_id}"
|
|
started_at {timestamp}
|
|
EXPIRE session:context:{session_key} 3600
|
|
```
|
|
|
|
### Data Lifecycle
|
|
|
|
| Key Type | TTL | Cleanup Strategy |
|
|
|----------|-----|------------------|
|
|
| `agent:heartbeat:*` | 60s | Auto-expire |
|
|
| `agent:status:*` | None | Delete on agent termination |
|
|
| `session:context:*` | 1h | Auto-expire |
|
|
| `tasks:pending:*` | None | Remove on assignment |
|
|
| `coordinator:lock:*` | 30s | Auto-expire (renewed by active coordinator) |
|
|
| `task:assign_lock:*` | 5s | Auto-expire after assignment |
|
|
|
|
---
|
|
|
|
## API Endpoints
|
|
|
|
### Task Management
|
|
|
|
```typescript
|
|
// Create a new orchestration task
|
|
POST /api/v1/agent-tasks
|
|
{
|
|
"title": "Fix TypeScript strict errors in U-Connect",
|
|
"description": "...",
|
|
"taskType": "development",
|
|
"priority": 8,
|
|
"inputContext": {
|
|
"repository": "u-connect",
|
|
"branch": "main",
|
|
"scope": ["packages/shared"]
|
|
},
|
|
"dependsOn": [],
|
|
"maxRetries": 3
|
|
}
|
|
|
|
// List tasks (with filtering)
|
|
GET /api/v1/agent-tasks?status=RUNNING&priority>=8&tags=typescript
|
|
|
|
// Get task details
|
|
GET /api/v1/agent-tasks/:id
|
|
|
|
// Update task (limited fields)
|
|
PATCH /api/v1/agent-tasks/:id
|
|
{
|
|
"priority": 10,
|
|
"status": "PAUSED"
|
|
}
|
|
|
|
// Cancel task
|
|
POST /api/v1/agent-tasks/:id/cancel
|
|
{
|
|
"reason": "Requirements changed"
|
|
}
|
|
|
|
// Retry failed task
|
|
POST /api/v1/agent-tasks/:id/retry
|
|
|
|
// Get task logs
|
|
GET /api/v1/agent-tasks/:id/logs?level=ERROR&limit=100
|
|
```
|
|
|
|
### Agent Management
|
|
|
|
```typescript
|
|
// List agents
|
|
GET /api/v1/agents?status=WORKING&coordinatorEnabled=true
|
|
|
|
// Get agent details
|
|
GET /api/v1/agents/:id
|
|
|
|
// Register/update agent
|
|
PUT /api/v1/agents/:agentId
|
|
{
|
|
"name": "Dev Agent #1",
|
|
"model": "claude-sonnet-4",
|
|
"capabilities": ["typescript", "nestjs", "prisma"],
|
|
"maxConcurrentTasks": 2,
|
|
"heartbeatIntervalSeconds": 30
|
|
}
|
|
|
|
// Get agent health
|
|
GET /api/v1/agents/:id/health
|
|
|
|
// Get agent tasks
|
|
GET /api/v1/agents/:id/tasks?status=RUNNING
|
|
|
|
// Terminate agent
|
|
POST /api/v1/agents/:id/terminate
|
|
{
|
|
"reason": "Manual shutdown",
|
|
"graceful": true // Allow task completion
|
|
}
|
|
```
|
|
|
|
### Coordination
|
|
|
|
```typescript
|
|
// Coordinator status (internal/admin only)
|
|
GET /api/v1/coordinator/status
|
|
{
|
|
"active": true,
|
|
"instanceId": "coord-abc123",
|
|
"workspaces": ["ws-1", "ws-2"],
|
|
"lastHeartbeat": "2025-01-29T14:30:00Z",
|
|
"stats": {
|
|
"pendingTasks": 3,
|
|
"runningTasks": 5,
|
|
"healthyAgents": 8,
|
|
"staleAgents": 1
|
|
}
|
|
}
|
|
|
|
// Force recovery check (admin only)
|
|
POST /api/v1/coordinator/recover
|
|
{
|
|
"workspaceId": "optional - all if omitted"
|
|
}
|
|
|
|
// Get coordination logs (admin only)
|
|
GET /api/v1/coordinator/logs?workspace={id}&level=ERROR
|
|
```
|
|
|
|
### Webhooks (for external integration)
|
|
|
|
```typescript
|
|
// Task state change webhook
|
|
POST {configured_webhook_url}
|
|
{
|
|
"event": "task.status_changed",
|
|
"taskId": "task-123",
|
|
"previousStatus": "RUNNING",
|
|
"newStatus": "COMPLETED",
|
|
"workspaceId": "ws-1",
|
|
"timestamp": "2025-01-29T14:30:00Z"
|
|
}
|
|
```
|
|
|
|
---
|
|
|
|
## Coordinator Service Design
|
|
|
|
### Core Responsibilities
|
|
|
|
1. **Health Monitoring** — Check agent heartbeats, mark stale agents
|
|
2. **Task Assignment** — Match pending tasks to available agents
|
|
3. **Recovery Orchestration** — Reassign tasks from failed/stale agents
|
|
4. **Dependency Resolution** — Ensure tasks wait for dependencies
|
|
5. **Resource Management** — Enforce agent concurrency limits
|
|
|
|
### Architecture
|
|
|
|
```typescript
|
|
// Coordinator Service (NestJS)
|
|
@Injectable()
|
|
export class CoordinatorService {
|
|
private coordinatorLock: string;
|
|
private isRunning: boolean = false;
|
|
|
|
constructor(
|
|
private readonly taskManager: TaskManagerService,
|
|
private readonly agentManager: AgentManagerService,
|
|
private readonly valkey: Valkey,
|
|
private readonly prisma: PrismaService,
|
|
private readonly logger: Logger
|
|
) {}
|
|
|
|
// Main coordination loop
|
|
@Cron('*/30 * * * * *') // Every 30 seconds
|
|
async coordinate() {
|
|
if (!await this.acquireLock()) {
|
|
return; // Another coordinator is active
|
|
}
|
|
|
|
try {
|
|
await this.checkAgentHealth();
|
|
await this.assignPendingTasks();
|
|
await this.resolveDependencies();
|
|
await this.recoverFailedTasks();
|
|
} catch (error) {
|
|
this.logger.error('Coordination cycle failed', error);
|
|
} finally {
|
|
await this.releaseLock();
|
|
}
|
|
}
|
|
|
|
// Distributed lock to prevent multiple coordinators
|
|
private async acquireLock(): Promise<boolean> {
|
|
const lockKey = `coordinator:lock:global`;
|
|
const result = await this.valkey.set(
|
|
lockKey,
|
|
process.env.HOSTNAME || 'coordinator',
|
|
'NX',
|
|
'EX',
|
|
30
|
|
);
|
|
return result === 'OK';
|
|
}
|
|
|
|
// Check agent heartbeats and mark stale
|
|
private async checkAgentHealth() {
|
|
const agents = await this.agentManager.getCoordinatorManagedAgents();
|
|
const now = Date.now();
|
|
|
|
for (const agent of agents) {
|
|
const heartbeatKey = `agent:heartbeat:${agent.id}`;
|
|
const lastHeartbeat = await this.valkey.get(heartbeatKey);
|
|
|
|
if (!lastHeartbeat) {
|
|
// No heartbeat - agent is stale
|
|
await this.handleStaleAgent(agent);
|
|
} else {
|
|
const heartbeatData = JSON.parse(lastHeartbeat);
|
|
const age = now - heartbeatData.timestamp;
|
|
|
|
if (age > agent.staleThresholdSeconds * 1000) {
|
|
await this.handleStaleAgent(agent);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Assign pending tasks to available agents
|
|
private async assignPendingTasks() {
|
|
const workspaces = await this.getActiveWorkspaces();
|
|
|
|
for (const workspace of workspaces) {
|
|
const pendingTasks = await this.taskManager.getPendingTasks(
|
|
workspace.id,
|
|
{ orderBy: { priority: 'desc', createdAt: 'asc' } }
|
|
);
|
|
|
|
for (const task of pendingTasks) {
|
|
// Check dependencies
|
|
if (!await this.areDependenciesMet(task)) {
|
|
continue;
|
|
}
|
|
|
|
// Find available agent
|
|
const agent = await this.agentManager.findAvailableAgent(
|
|
workspace.id,
|
|
task.taskType,
|
|
task.metadata.requiredCapabilities
|
|
);
|
|
|
|
if (agent) {
|
|
await this.assignTask(task, agent);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Handle stale agents
|
|
private async handleStaleAgent(agent: Agent) {
|
|
this.logger.warn(`Agent ${agent.id} is stale - recovering tasks`);
|
|
|
|
// Mark agent as ERROR
|
|
await this.agentManager.updateAgentStatus(agent.id, AgentStatus.ERROR);
|
|
|
|
// Get assigned tasks
|
|
const tasks = await this.taskManager.getTasksForAgent(agent.id);
|
|
|
|
for (const task of tasks) {
|
|
await this.recoverTask(task, 'agent_stale');
|
|
}
|
|
}
|
|
|
|
// Recover a task from failure
|
|
private async recoverTask(task: AgentTask, reason: string) {
|
|
// Log the failure
|
|
await this.taskManager.logTaskEvent(task.id, {
|
|
level: 'ERROR',
|
|
event: 'task_recovery',
|
|
message: `Task recovery initiated: ${reason}`,
|
|
previousStatus: task.status,
|
|
newStatus: 'ABORTED'
|
|
});
|
|
|
|
// Check retry limit
|
|
if (task.retryCount >= task.maxRetries) {
|
|
await this.taskManager.updateTask(task.id, {
|
|
status: 'FAILED',
|
|
lastError: `Max retries exceeded (${task.retryCount}/${task.maxRetries})`,
|
|
failedAt: new Date()
|
|
});
|
|
return;
|
|
}
|
|
|
|
// Abort current assignment
|
|
await this.taskManager.updateTask(task.id, {
|
|
status: 'ABORTED',
|
|
agentId: null,
|
|
sessionKey: null,
|
|
retryCount: task.retryCount + 1
|
|
});
|
|
|
|
// Wait for backoff period before requeuing
|
|
const backoffMs = task.retryBackoffSeconds * 1000 * Math.pow(2, task.retryCount);
|
|
setTimeout(async () => {
|
|
await this.taskManager.updateTask(task.id, {
|
|
status: 'PENDING'
|
|
});
|
|
}, backoffMs);
|
|
}
|
|
|
|
// Assign task to agent
|
|
private async assignTask(task: AgentTask, agent: Agent) {
|
|
// Acquire assignment lock
|
|
const lockKey = `task:assign_lock:${task.id}`;
|
|
const locked = await this.valkey.set(lockKey, agent.id, 'NX', 'EX', 5);
|
|
|
|
if (!locked) {
|
|
return; // Another coordinator already assigned this task
|
|
}
|
|
|
|
try {
|
|
// Update task
|
|
await this.taskManager.updateTask(task.id, {
|
|
status: 'ASSIGNED',
|
|
agentId: agent.id,
|
|
assignedAt: new Date()
|
|
});
|
|
|
|
// Spawn agent session via Gateway
|
|
const session = await this.spawnAgentSession(agent, task);
|
|
|
|
// Update task with session
|
|
await this.taskManager.updateTask(task.id, {
|
|
sessionKey: session.sessionKey,
|
|
status: 'RUNNING',
|
|
startedAt: new Date()
|
|
});
|
|
|
|
// Log assignment
|
|
await this.taskManager.logTaskEvent(task.id, {
|
|
level: 'INFO',
|
|
event: 'task_assigned',
|
|
message: `Task assigned to agent ${agent.id}`,
|
|
details: { agentId: agent.id, sessionKey: session.sessionKey }
|
|
});
|
|
} finally {
|
|
await this.valkey.del(lockKey);
|
|
}
|
|
}
|
|
|
|
// Spawn agent session via Gateway
|
|
private async spawnAgentSession(agent: Agent, task: AgentTask): Promise<AgentSession> {
|
|
// Call Gateway API to spawn subagent with task context
|
|
const response = await fetch(`${process.env.GATEWAY_URL}/api/agents/spawn`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({
|
|
workspaceId: task.workspaceId,
|
|
agentId: agent.id,
|
|
taskId: task.id,
|
|
label: `task:${task.id}:${task.title.slice(0, 30)}`,
|
|
context: {
|
|
taskTitle: task.title,
|
|
taskDescription: task.description,
|
|
inputContext: task.inputContext,
|
|
checkpointData: task.checkpointData
|
|
}
|
|
})
|
|
});
|
|
|
|
const data = await response.json();
|
|
return data.session;
|
|
}
|
|
}
|
|
```
|
|
|
|
### Coordination State Machine
|
|
|
|
```
|
|
┌─────────────────────────────────────────────────────────┐
|
|
│ Coordinator Cycle │
|
|
└─────────────────────────────────────────────────────────┘
|
|
│
|
|
▼
|
|
┌───────────────────────┐
|
|
│ Acquire Lock (30s) │
|
|
└───────────┬───────────┘
|
|
│
|
|
┌────────▼────────┐
|
|
│ Check Health │
|
|
│ - Read agents │
|
|
│ - Check ttls │
|
|
│ - Mark stale │
|
|
└────────┬────────┘
|
|
│
|
|
┌────────▼────────┐
|
|
│ Assign Tasks │
|
|
│ - Read pending │
|
|
│ - Match agents │
|
|
│ - Spawn sessions│
|
|
└────────┬────────┘
|
|
│
|
|
┌────────▼────────┐
|
|
│ Resolve Deps │
|
|
│ - Check waiting │
|
|
│ - Unblock ready │
|
|
└────────┬────────┘
|
|
│
|
|
┌────────▼────────┐
|
|
│ Recover Failed │
|
|
│ - Requeue retry │
|
|
│ - Log failures │
|
|
└────────┬────────┘
|
|
│
|
|
┌───────────▼───────────┐
|
|
│ Release Lock │
|
|
└───────────────────────┘
|
|
```
|
|
|
|
---
|
|
|
|
## Recovery & Resilience Patterns
|
|
|
|
### 1. Agent Failure Recovery
|
|
|
|
**Scenario:** Agent crashes mid-task.
|
|
|
|
**Detection:**
|
|
- Heartbeat TTL expires in Valkey
|
|
- Coordinator detects missing heartbeat
|
|
|
|
**Recovery:**
|
|
1. Mark agent as `ERROR` in database
|
|
2. Abort assigned tasks with `status = ABORTED`
|
|
3. Log failure with stack trace (if available)
|
|
4. Apply exponential backoff: `backoff = retryBackoffSeconds * 2^retryCount`
|
|
5. Requeue task to `PENDING` after backoff
|
|
6. Coordinator reassigns on next cycle
|
|
|
|
### 2. Gateway Restart Recovery
|
|
|
|
**Scenario:** Gateway restarts, killing all agent sessions.
|
|
|
|
**Detection:**
|
|
- All agent heartbeats stop simultaneously
|
|
- Coordinator detects mass stale agents
|
|
|
|
**Recovery:**
|
|
1. Coordinator marks all `RUNNING` tasks as `ABORTED`
|
|
2. Tasks with `checkpointData` can resume from last checkpoint
|
|
3. Tasks without checkpoints restart from scratch
|
|
4. Exponential backoff prevents thundering herd
|
|
|
|
### 3. Task Dependency Deadlock
|
|
|
|
**Scenario:** Task A depends on Task B, which depends on Task A (circular dependency).
|
|
|
|
**Detection:**
|
|
- Coordinator builds dependency graph
|
|
- Detects cycles during `resolveDependencies()`
|
|
|
|
**Recovery:**
|
|
1. Log `ERROR` with cycle details
|
|
2. Mark all tasks in cycle as `FAILED` with reason `dependency_cycle`
|
|
3. Notify workspace owner via webhook
|
|
|
|
### 4. Database Failure
|
|
|
|
**Scenario:** PostgreSQL becomes unavailable.
|
|
|
|
**Detection:**
|
|
- Prisma query fails with connection error
|
|
|
|
**Recovery:**
|
|
1. Coordinator catches error, logs to stderr
|
|
2. Releases lock (allowing failover to another instance)
|
|
3. Retries with exponential backoff: 5s, 10s, 20s, 40s
|
|
4. If database remains down >5 minutes, alert admin
|
|
|
|
### 5. Split-Brain Prevention
|
|
|
|
**Scenario:** Network partition causes two coordinators to run simultaneously.
|
|
|
|
**Prevention:**
|
|
- Distributed lock in Valkey with 30s TTL
|
|
- Coordinators must renew lock every cycle
|
|
- Only one coordinator can hold lock at a time
|
|
|
|
**Detection:**
|
|
- Task assigned to multiple agents (conflict detection)
|
|
|
|
**Recovery:**
|
|
1. Newer assignment wins (based on `assignedAt` timestamp)
|
|
2. Cancel older session
|
|
3. Log conflict for investigation
|
|
|
|
### 6. Stale Task Timeout
|
|
|
|
**Scenario:** Task runs longer than `estimatedCompletionAt + grace period`.
|
|
|
|
**Detection:**
|
|
- Coordinator checks `estimatedCompletionAt` field
|
|
- If exceeded by >30 minutes, mark as potentially hung
|
|
|
|
**Recovery:**
|
|
1. Send warning to agent session (via pub/sub)
|
|
2. If no progress update in 10 minutes, abort task
|
|
3. Log timeout error
|
|
4. Requeue with higher priority
|
|
|
|
---
|
|
|
|
## Implementation Phases
|
|
|
|
### Phase 1: Foundation (Week 1-2)
|
|
|
|
**Goal:** Basic task and agent models, no coordination yet.
|
|
|
|
**Deliverables:**
|
|
- [ ] Database schema migration (tables, indexes)
|
|
- [ ] Prisma models for `AgentTask`, `AgentTaskLog`, `AgentHeartbeat`
|
|
- [ ] Basic CRUD API endpoints for tasks
|
|
- [ ] Agent registration API
|
|
- [ ] Manual task assignment (no automation)
|
|
|
|
**Testing:**
|
|
- Unit tests for task state machine
|
|
- Integration tests for task CRUD
|
|
- Manual testing: create task, assign to agent, complete
|
|
|
|
### Phase 2: Coordination Core (Week 3-4)
|
|
|
|
**Goal:** Autonomous coordinator with health monitoring.
|
|
|
|
**Deliverables:**
|
|
- [ ] `CoordinatorService` with distributed locking
|
|
- [ ] Health monitoring (heartbeat TTL checks)
|
|
- [ ] Automatic task assignment to available agents
|
|
- [ ] Valkey integration for runtime state
|
|
- [ ] Pub/Sub for coordination events
|
|
|
|
**Testing:**
|
|
- Unit tests for coordinator logic
|
|
- Integration tests with Valkey
|
|
- Chaos testing: kill agents, verify recovery
|
|
- Load testing: 10+ agents, 50+ tasks
|
|
|
|
### Phase 3: Recovery & Resilience (Week 5)
|
|
|
|
**Goal:** Fault-tolerant operation with automatic recovery.
|
|
|
|
**Deliverables:**
|
|
- [ ] Agent failure detection and task recovery
|
|
- [ ] Exponential backoff for retries
|
|
- [ ] Checkpoint/resume support for long-running tasks
|
|
- [ ] Dependency resolution
|
|
- [ ] Deadlock detection
|
|
|
|
**Testing:**
|
|
- Fault injection: kill agents, restart Gateway
|
|
- Dependency cycle testing
|
|
- Retry exhaustion testing
|
|
- Split-brain simulation
|
|
|
|
### Phase 4: Observability (Week 6)
|
|
|
|
**Goal:** Full visibility into orchestration state.
|
|
|
|
**Deliverables:**
|
|
- [ ] Coordinator status dashboard
|
|
- [ ] Task progress tracking UI
|
|
- [ ] Real-time logs API
|
|
- [ ] Metrics export (Prometheus format)
|
|
- [ ] Webhook integration for external monitoring
|
|
|
|
**Testing:**
|
|
- Load testing with metrics collection
|
|
- Dashboard usability testing
|
|
- Webhook reliability testing
|
|
|
|
### Phase 5: Advanced Features (Future)
|
|
|
|
**Goal:** Production-grade features.
|
|
|
|
**Deliverables:**
|
|
- [ ] Task prioritization algorithms (SJF, priority queues)
|
|
- [ ] Agent capability matching (skills-based routing)
|
|
- [ ] Task batching (group similar tasks)
|
|
- [ ] Cost tracking (token usage, compute time)
|
|
- [ ] Multi-region coordination (geo-distributed agents)
|
|
|
|
---
|
|
|
|
## Security Considerations
|
|
|
|
### Row-Level Security (RLS)
|
|
|
|
All queries must enforce workspace isolation:
|
|
|
|
```typescript
|
|
// Example: Only return tasks for user's workspace
|
|
async getTasks(userId: string, workspaceId: string) {
|
|
const membership = await this.prisma.workspaceMember.findUnique({
|
|
where: { workspaceId_userId: { workspaceId, userId } }
|
|
});
|
|
|
|
if (!membership) {
|
|
throw new ForbiddenException('Not a member of this workspace');
|
|
}
|
|
|
|
return this.prisma.agentTask.findMany({
|
|
where: { workspaceId }
|
|
});
|
|
}
|
|
```
|
|
|
|
### API Authentication
|
|
|
|
- All endpoints require valid session token
|
|
- Coordinator endpoints restricted to `ADMIN` or `OWNER` roles
|
|
- Agent heartbeat endpoints use JWT with agent-specific claims
|
|
|
|
### Secrets Management
|
|
|
|
- Task `inputContext` may contain secrets (API keys, passwords)
|
|
- Encrypt sensitive fields at rest using Prisma middleware
|
|
- Never log secrets in task logs
|
|
- Redact secrets in API responses
|
|
|
|
---
|
|
|
|
## Monitoring & Alerting
|
|
|
|
### Key Metrics
|
|
|
|
| Metric | Description | Alert Threshold |
|
|
|--------|-------------|-----------------|
|
|
| `coordinator.cycle.duration_ms` | Coordination cycle execution time | >5000ms |
|
|
| `coordinator.stale_agents.count` | Number of stale agents detected | >5 |
|
|
| `tasks.pending.count` | Tasks waiting for assignment | >50 |
|
|
| `tasks.failed.count` | Total failed tasks (last 1h) | >10 |
|
|
| `tasks.retry.exhausted.count` | Tasks exceeding max retries | >0 |
|
|
| `agents.spawned.count` | Agent spawn rate | >100/min |
|
|
| `valkey.connection.errors` | Valkey connection failures | >0 |
|
|
|
|
### Health Checks
|
|
|
|
```http
|
|
GET /health/coordinator
|
|
{
|
|
"status": "healthy",
|
|
"coordinator": {
|
|
"active": true,
|
|
"lastCycle": "2025-01-29T14:30:00Z",
|
|
"cycleCount": 1234
|
|
},
|
|
"database": "connected",
|
|
"valkey": "connected",
|
|
"gateway": "reachable"
|
|
}
|
|
```
|
|
|
|
---
|
|
|
|
## API Integration Examples
|
|
|
|
### Creating a Task from Main Agent
|
|
|
|
```typescript
|
|
// Main agent creates a development task
|
|
const task = await fetch('/api/v1/agent-tasks', {
|
|
method: 'POST',
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': `Bearer ${sessionToken}`
|
|
},
|
|
body: JSON.stringify({
|
|
title: 'Fix TypeScript strict errors in U-Connect',
|
|
description: 'Run tsc --noEmit, fix all errors, commit changes',
|
|
taskType: 'development',
|
|
priority: 8,
|
|
inputContext: {
|
|
repository: 'u-connect',
|
|
branch: 'main',
|
|
commands: ['pnpm install', 'pnpm tsc:check']
|
|
},
|
|
maxRetries: 2,
|
|
estimatedDurationMinutes: 30
|
|
})
|
|
});
|
|
|
|
const { id } = await task.json();
|
|
console.log(`Task created: ${id}`);
|
|
```
|
|
|
|
### Agent Heartbeat (from spawned subagent)
|
|
|
|
```typescript
|
|
// Subagent sends heartbeat every 30s
|
|
setInterval(async () => {
|
|
await fetch(`/api/v1/agents/${agentId}/heartbeat`, {
|
|
method: 'POST',
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': `Bearer ${agentToken}`
|
|
},
|
|
body: JSON.stringify({
|
|
status: 'healthy',
|
|
currentTaskId: taskId,
|
|
progressPercent: 45,
|
|
currentStep: 'Running tsc --noEmit',
|
|
memoryMb: 512,
|
|
cpuPercent: 35
|
|
})
|
|
});
|
|
}, 30000);
|
|
```
|
|
|
|
### Task Progress Update
|
|
|
|
```typescript
|
|
// Agent updates task progress
|
|
await fetch(`/api/v1/agent-tasks/${taskId}/progress`, {
|
|
method: 'PATCH',
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': `Bearer ${agentToken}`
|
|
},
|
|
body: JSON.stringify({
|
|
progressPercent: 70,
|
|
currentStep: 'Fixing type errors in packages/shared',
|
|
checkpointData: {
|
|
filesProcessed: 15,
|
|
errorsFixed: 8,
|
|
remainingFiles: 5
|
|
}
|
|
})
|
|
});
|
|
```
|
|
|
|
### Task Completion
|
|
|
|
```typescript
|
|
// Agent marks task complete
|
|
await fetch(`/api/v1/agent-tasks/${taskId}/complete`, {
|
|
method: 'POST',
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': `Bearer ${agentToken}`
|
|
},
|
|
body: JSON.stringify({
|
|
outputResult: {
|
|
filesModified: 20,
|
|
errorsFixed: 23,
|
|
commitHash: 'abc123',
|
|
buildStatus: 'passing'
|
|
},
|
|
summary: 'All TypeScript strict errors resolved. Build passing.'
|
|
})
|
|
});
|
|
```
|
|
|
|
---
|
|
|
|
## Glossary
|
|
|
|
| Term | Definition |
|
|
|------|------------|
|
|
| **Agent** | Autonomous AI instance (e.g., Claude subagent) that executes tasks |
|
|
| **Task** | Unit of work to be executed by an agent |
|
|
| **Coordinator** | Background service that assigns tasks and monitors agent health |
|
|
| **Heartbeat** | Periodic signal from agent indicating it's alive and working |
|
|
| **Stale Agent** | Agent that has stopped sending heartbeats (assumed dead) |
|
|
| **Checkpoint** | Snapshot of task state allowing resumption after failure |
|
|
| **Workspace** | Tenant isolation boundary (all tasks belong to a workspace) |
|
|
| **Session** | Gateway-managed connection between user and agent |
|
|
| **Orchestration** | Automated coordination of multiple agents working on tasks |
|
|
|
|
---
|
|
|
|
## References
|
|
|
|
- [Mosaic Stack Architecture](../3-architecture/README.md)
|
|
- [Existing Agent/AgentSession Schema](../../apps/api/prisma/schema.prisma)
|
|
- [NestJS Task Scheduling](https://docs.nestjs.com/techniques/task-scheduling)
|
|
- [Valkey Documentation](https://valkey.io/docs/)
|
|
- [PostgreSQL Advisory Locks](https://www.postgresql.org/docs/current/explicit-locking.html#ADVISORY-LOCKS)
|
|
|
|
---
|
|
|
|
**Next Steps:**
|
|
1. Review and approve this design document
|
|
2. Create GitHub issues for Phase 1 tasks
|
|
3. Set up development branch: `feature/agent-orchestration`
|
|
4. Begin database schema migration
|
|
|
|
**Questions/Feedback:** Open an issue in `mosaic-stack` repo with label `orchestration`.
|