parallel_work_streams

Parallel Work Streams & Agent Assignment Plan

Document Purpose: Identify parallelizable work streams that can be assigned to different agents simultaneously, with clear boundaries and interfaces.

Created: 2025-11-16
Status: Active Planning
Related Documents:

• Implementation Roadmap
• Foundation & Smallest Runnable

---

Parallelization Strategy

Principles

1. Independent Work Streams: Tasks that don't share code paths or have minimal overlap
2. Clear Interfaces: Well-defined contracts between components
3. Database Schema Coordination: Shared schema changes need coordination
4. Test Isolation: Each stream has its own test suite

---

Phase 1: Core Workflow Enhancement - Parallel Streams

Stream A: Task Dependencies & Blocking

Agent Assignment: Agent A
Files to Modify:

• omoi_os/models/task.py (add dependencies JSONB field)
• omoi_os/services/task_queue.py (dependency resolution logic)
• omoi_os/api/routes/tasks.py (dependency API endpoints)
• tests/test_task_dependencies.py (new test file)

Database Changes:

• Migration: Add dependencies JSONB column to tasks table
• Index: Add GIN index on dependencies for query performance

Interface Contracts:

# Task model
dependencies: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True)
# Format: {"depends_on": ["task_id_1", "task_id_2"], "blocks": ["task_id_3"]}

# TaskQueueService methods
def get_next_task(self, phase_id: str) -> Task | None:
    # Must filter out tasks with incomplete dependencies
    
def check_dependencies_complete(self, task_id: str) -> bool:
    # New method: Check if all dependencies are completed
    
def get_blocked_tasks(self, task_id: str) -> list[Task]:
    # New method: Get tasks blocked by this task

Dependencies: None (can start immediately)

Deliverables:

• ✅ Task model with dependencies field
• ✅ Dependency resolution in get_next_task()
• ✅ Circular dependency detection
• ✅ Tests for dependency resolution