ADR-069: Atomic Task Management System

Status

Accepted - January 13, 2026

Context

The Problem

Large project plans in CODITECT (e.g., PILOT-PARALLEL-EXECUTION-PLAN.md at 380KB, ~95K tokens) present fundamental challenges beyond token economics (addressed in ADR-068):

1. No Automated Assignment: Manual agent-task matching for 500+ tasks
2. No Execution Tracking: Task status scattered across markdown checkboxes
3. No Dependency Enforcement: Blocking relationships not programmatically enforced
4. No Learning Loop: Agent performance on task types not tracked
5. No Parallelization Logic: Can't identify which tasks can run concurrently
6. Context Fragmentation: Related tasks may span multiple documents

Research Foundation

Based on synthesis of:

• Hierarchical Task Networks (HTN): Formal task decomposition methodology
• Google Cloud Workflows: State machine patterns for task execution
• Microsoft AutoGen: Multi-agent task coordination
• UiPath Orchestrator: Enterprise task queuing and assignment
• Academic Research: arXiv papers on agentic task planning

Key Insight

Atomic tasks are the smallest units of work that can be:

• Assigned to a single agent
• Completed in a single session
• Verified independently
• Rolled back atomically

Decision

Implement an Atomic Task Management System that decomposes project plans into database-stored tasks with automated agent assignment, dependency resolution, and execution tracking.

1. Task Hierarchy Model

Track (A-G)
└── Section (A.1, A.2, ...)
    └── Task (A.1.1, A.1.2, ...)
        └── Subtask (A.1.1.1, A.1.1.2, ...)  ← Atomic level

Atomic Task Definition:

• Time-boxed: Completable in <4 hours
• Single-agent: Assigned to one agent type
• Testable: Has defined acceptance criteria
• Isolated: Minimal external dependencies

2. Database Schema

Core Tables

-- Primary atomic task table
CREATE TABLE atomic_tasks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    task_id VARCHAR(20) UNIQUE NOT NULL,        -- e.g., "A.9.1.1.3"
    parent_task_id VARCHAR(20),                  -- e.g., "A.9.1.1"

    -- Hierarchy
    track CHAR(1) NOT NULL CHECK (track IN ('A','B','C','D','E','F','G')),
    section INTEGER NOT NULL,
    task_number INTEGER NOT NULL,
    subtask_number INTEGER,

    -- Content
    title VARCHAR(255) NOT NULL,
    description TEXT,
    acceptance_criteria TEXT[],

    -- Classification
    task_type VARCHAR(50) NOT NULL,              -- e.g., "implementation", "documentation"
    domain VARCHAR(50),                          -- e.g., "backend", "frontend", "devops"
    complexity VARCHAR(10) CHECK (complexity IN ('trivial', 'simple', 'moderate', 'complex')),
    estimated_tokens INTEGER,

    -- State
    status VARCHAR(20) DEFAULT 'pending'
        CHECK (status IN ('pending', 'ready', 'in_progress', 'blocked', 'completed', 'failed', 'skipped')),
    priority INTEGER DEFAULT 50 CHECK (priority BETWEEN 1 AND 100),

    -- Assignment
    recommended_agent VARCHAR(100),
    assigned_agent VARCHAR(100),
    assigned_at TIMESTAMP WITH TIME ZONE,

    -- Execution
    started_at TIMESTAMP WITH TIME ZONE,
    completed_at TIMESTAMP WITH TIME ZONE,
    execution_duration_ms INTEGER,
    outcome VARCHAR(20) CHECK (outcome IN ('success', 'partial', 'failure', 'timeout')),
    output_artifacts TEXT[],                      -- Files created/modified

    -- Metrics
    retry_count INTEGER DEFAULT 0,
    token_usage INTEGER,

    -- Metadata
    source_file VARCHAR(255),                     -- Original markdown file
    source_line_start INTEGER,
    source_line_end INTEGER,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),

    -- Vector embedding for semantic search
    title_embedding vector(1536)
);

-- Dependency graph
CREATE TABLE task_dependencies (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    source_task_id VARCHAR(20) NOT NULL REFERENCES atomic_tasks(task_id),
    target_task_id VARCHAR(20) NOT NULL REFERENCES atomic_tasks(task_id),
    dependency_type VARCHAR(20) NOT NULL
        CHECK (dependency_type IN ('blocks', 'informs', 'relates')),
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    UNIQUE(source_task_id, target_task_id, dependency_type)
);

-- Agent performance tracking
CREATE TABLE agent_task_affinity (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    agent_name VARCHAR(100) NOT NULL,
    task_type VARCHAR(50) NOT NULL,
    domain VARCHAR(50),

    -- Performance metrics
    total_executions INTEGER DEFAULT 0,
    successful_executions INTEGER DEFAULT 0,
    success_rate DECIMAL(5,4) GENERATED ALWAYS AS
        (CASE WHEN total_executions > 0
              THEN successful_executions::DECIMAL / total_executions
              ELSE 0 END) STORED,
    avg_duration_ms INTEGER,
    avg_token_usage INTEGER,

    -- Confidence
    affinity_score DECIMAL(5,4) DEFAULT 0.5,
    last_updated TIMESTAMP WITH TIME ZONE DEFAULT NOW(),

    UNIQUE(agent_name, task_type, domain)
);

-- Execution log
CREATE TABLE task_executions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    task_id VARCHAR(20) NOT NULL REFERENCES atomic_tasks(task_id),
    agent_name VARCHAR(100) NOT NULL,

    -- Execution details
    started_at TIMESTAMP WITH TIME ZONE NOT NULL,
    completed_at TIMESTAMP WITH TIME ZONE,
    duration_ms INTEGER,
    outcome VARCHAR(20) CHECK (outcome IN ('success', 'partial', 'failure', 'timeout', 'aborted')),

    -- Context
    session_id VARCHAR(100),
    machine_id VARCHAR(100),
    model_used VARCHAR(50),

    -- Metrics
    tokens_input INTEGER,
    tokens_output INTEGER,
    retry_number INTEGER DEFAULT 0,

    -- Output
    artifacts_created TEXT[],
    error_message TEXT,
    execution_log JSONB,

    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

Indexes

-- Performance indexes
CREATE INDEX idx_tasks_status ON atomic_tasks(status);
CREATE INDEX idx_tasks_track ON atomic_tasks(track);
CREATE INDEX idx_tasks_priority ON atomic_tasks(priority DESC);
CREATE INDEX idx_tasks_ready ON atomic_tasks(status) WHERE status = 'ready';
CREATE INDEX idx_deps_source ON task_dependencies(source_task_id);
CREATE INDEX idx_deps_target ON task_dependencies(target_task_id);
CREATE INDEX idx_affinity_agent ON agent_task_affinity(agent_name);

-- Vector similarity search
CREATE INDEX idx_tasks_embedding ON atomic_tasks
    USING ivfflat (title_embedding vector_cosine_ops) WITH (lists = 100);

3. Task States

                    ┌──────────┐
                    │ pending  │
                    └────┬─────┘
                         │ dependencies resolved
                         ▼
                    ┌──────────┐
          ┌─────── │  ready   │ ◀──────────┐
          │        └────┬─────┘            │
          │             │ agent assigned   │ retry
          │             ▼                  │
          │        ┌──────────┐            │
          │        │in_progress├───────────┤
          │        └────┬─────┘            │
          │             │                  │
          │    ┌────────┼────────┐         │
          │    ▼        ▼        ▼         │
          │ ┌──────┐ ┌──────┐ ┌──────┐     │
          │ │success│ │partial│ │failure├──┘
          │ └───┬──┘ └───┬──┘ └───┬──┘
          │     │        │        │
          │     ▼        ▼        ▼
          │ ┌───────────────────────┐
blocked ◀─┴─│      completed        │
            └───────────────────────┘

4. Agent Matching Algorithm

def calculate_affinity_score(agent: str, task: AtomicTask) -> float:
    """
    Calculate agent-task affinity score.

    Weights:
    - Semantic similarity: 40%
    - Historical success rate: 30%
    - Domain expertise: 20%
    - Current availability: 10%
    """
    # Get agent capabilities embedding
    agent_embedding = get_agent_embedding(agent)
    task_embedding = task.title_embedding

    # Semantic similarity (cosine)
    semantic_score = cosine_similarity(agent_embedding, task_embedding)

    # Historical performance
    affinity = get_agent_affinity(agent, task.task_type, task.domain)
    historical_score = affinity.success_rate if affinity else 0.5

    # Domain expertise
    domain_score = get_domain_expertise(agent, task.domain)

    # Availability (active tasks count)
    availability = get_agent_availability(agent)

    # Weighted combination
    return (
        semantic_score * 0.4 +
        historical_score * 0.3 +
        domain_score * 0.2 +
        availability * 0.1
    )

5. Dependency Resolution

def get_ready_tasks(limit: int = 10) -> List[AtomicTask]:
    """
    Get tasks ready for execution.

    A task is ready when:
    1. Status is 'pending' or 'ready'
    2. All 'blocks' dependencies are completed
    3. Not currently assigned
    """
    return db.execute("""
        SELECT t.* FROM atomic_tasks t
        WHERE t.status IN ('pending', 'ready')
          AND t.assigned_agent IS NULL
          AND NOT EXISTS (
              SELECT 1 FROM task_dependencies d
              JOIN atomic_tasks dep ON d.source_task_id = dep.task_id
              WHERE d.target_task_id = t.task_id
                AND d.dependency_type = 'blocks'
                AND dep.status != 'completed'
          )
        ORDER BY t.priority DESC, t.created_at ASC
        LIMIT :limit
    """, limit=limit)

6. Task Decomposition Protocol

Converting markdown tasks to atomic tasks:

def decompose_pilot_plan(plan_path: str) -> List[AtomicTask]:
    """
    Parse PILOT plan and extract atomic tasks.
    """
    tasks = []

    with open(plan_path) as f:
        content = f.read()

    # Pattern: - [x] A.9.1.1: Task description
    pattern = r'- \[([ x])\] ([A-G])\.(\d+)\.(\d+)(?:\.(\d+))?: (.+)'

    for match in re.finditer(pattern, content):
        completed = match.group(1) == 'x'
        track = match.group(2)
        section = int(match.group(3))
        task_num = int(match.group(4))
        subtask_num = int(match.group(5)) if match.group(5) else None
        title = match.group(6)

        task_id = f"{track}.{section}.{task_num}"
        if subtask_num:
            task_id += f".{subtask_num}"

        task = AtomicTask(
            task_id=task_id,
            track=track,
            section=section,
            task_number=task_num,
            subtask_number=subtask_num,
            title=title,
            status='completed' if completed else 'pending',
            source_file=plan_path,
            source_line_start=match.start(),
        )

        # Classify task type and domain
        task.task_type = classify_task_type(title)
        task.domain = infer_domain(track, title)
        task.recommended_agent = get_default_agent(track)

        tasks.append(task)

    return tasks

7. Integration Points

System	Integration Method
CODITECT Core /orient	Load ready tasks via API
Cloud Backend	Sync via task-plan-sync.py
Session Logs	Link executions to sessions
MoE System	Route agent dispatch
Context Watcher	Track token usage per task

8. API Endpoints

# Task Orchestrator API
/api/v1/tasks:
  GET:
    - /ready              # Get tasks ready for execution
    - /by-track/{track}   # Filter by track
    - /search?q=          # Semantic search
    - /{task_id}          # Get single task

  POST:
    - /                   # Create new task
    - /import             # Bulk import from markdown
    - /{task_id}/start    # Mark task started
    - /{task_id}/complete # Mark task completed
    - /{task_id}/fail     # Mark task failed

  PATCH:
    - /{task_id}          # Update task

/api/v1/agents:
  GET:
    - /                   # List agents
    - /{agent}/affinity   # Get agent affinities
    - /recommend/{task_id}# Get recommended agent

/api/v1/dependencies:
  GET:
    - /graph/{task_id}    # Get dependency graph
    - /blocking/{task_id} # Get blocking tasks

  POST:
    - /                   # Create dependency

Consequences

Positive

1. Automated Assignment: Agent-task matching based on performance data
2. Execution Tracking: Complete audit trail of all task executions
3. Dependency Enforcement: Programmatic blocking prevents invalid execution
4. Learning Loop: Agent performance improves task routing over time
5. Parallelization: Identify independent tasks for concurrent execution
6. Searchability: Semantic search across all tasks via embeddings
7. Token Efficiency: Query only relevant tasks (~15.5K tokens vs 95K)

Negative

1. Infrastructure Overhead: Requires PostgreSQL + pgvector
2. Migration Effort: Existing 500+ tasks need initial import
3. Sync Complexity: Bidirectional sync with markdown sources
4. Learning Curve: New mental model for task management

Mitigations

Risk	Mitigation
Infrastructure overhead	Use existing Cloud SQL instance
Migration effort	Automated decomposition script
Sync complexity	Event-driven sync via hooks
Learning curve	/orient command abstracts complexity

Implementation

Phase 1: Schema & Import (4 hours)

• Create database tables
• Build markdown parser
• Import PILOT plan tasks

Phase 2: Agent Matching (4 hours)

• Generate task embeddings
• Implement affinity scoring
• Build recommendation API

Phase 3: Integration (4 hours)

• Update /orient command
• Add execution tracking hooks
• Build sync to cloud backend

Phase 4: Learning Loop (4 hours)

• Implement execution logging
• Build performance aggregation
• Create agent performance dashboard

References

Local ADRs (this repository)

• ADR-070: Task Dependency Graph
• ADR-071: Agent-Task Matching
• ADR-072: Execution Tracking
• ADR-073: Learning Feedback Loop

Related ADRs (coditect-core repository)

These ADRs are located in coditect-core/internal/architecture/adrs/

• ADR-068: Large Project Plan Token Economics (motivation for this system)
• ADR-052: Intent-Aware Context Management
• ADR-053: Cloud Context Sync Architecture
• ADR-054: Track Nomenclature Extensibility

External References

• HTN Planning Research
• Microsoft AutoGen

---

Decision Date: January 13, 2026 Decision Makers: Hal Casteel, Claude Opus 4.5 Implementation Priority: P1 (High) Estimated Effort: 16 hours