Technical Design: Orchestrator Coordination Fix

Technical Design: Orchestrator Coordination Fix

Metadata

• Feature: orchestrator-coordination-fix
• Status: APPROVED
• Created: 2026-02-04
• GitHub Issues: Related to #65, #119 (bite-sized-planning rush failures)

1. Overview

1.1 Summary

Fix 6 critical bugs in ZERG orchestrator that cause merge conflicts, level violations, and redundant gate runs. The core fix is simple: defer all merges to ship time instead of merging after each level.

1.2 Goals

• Main branch untouched during rush execution
• Strict level enforcement at task claim time
• Strict dependency enforcement at runtime
• Quality gates run exactly once (at ship)
• Live status streaming for user visibility

1.3 Non-Goals

• Changing ship command's PR creation logic
• Adding new merge strategies
• Modifying worker spawn behavior

2. Architecture

2.1 Current Flow (Broken)

Level 1 complete → merge_level() → full_merge_flow(target="main") → MAIN MODIFIED
Level 2 complete → merge_level() → full_merge_flow(target="main") → MAIN MODIFIED (again)
...

2.2 Fixed Flow

Level 1 complete → set_level_status("complete") → advance to Level 2 → NO MERGE
Level 2 complete → set_level_status("complete") → advance to Level 3 → NO MERGE
...
All levels complete → Rush DONE (main untouched)

/z:git --action ship:
  → Create staging from main
  → Merge all worker branches to staging
  → Run quality gates ONCE
  → Merge staging to main / Create PR
  → Cleanup

2.3 Component Changes

3. Detailed Design

3.1 Config Changes

# zerg/config.py

class RushConfig(BaseModel):
    # Existing fields...

    # NEW: Coordination fixes
    defer_merge_to_ship: bool = Field(
        default=True,
        description="Defer all merges to /z:git --action ship"
    )
    gates_at_ship_only: bool = Field(
        default=True,
        description="Run quality gates only at ship time, not per level"
    )

3.2 DependencyChecker Helper

# zerg/dependency_checker.py

class DependencyChecker:
    def __init__(self, state: StateManager, parser: TaskParser):
        self._state = state
        self._parser = parser

    def are_dependencies_complete(self, task_id: str) -> bool:
        """Check if all dependencies of task_id are COMPLETE."""
        task = self._parser.get_task(task_id)
        if not task:
            return False

        for dep_id in task.get("dependencies", []):
            dep_status = self._state.get_task_status(dep_id)
            if dep_status != TaskStatus.COMPLETE.value:
                return False

        return True

    def get_incomplete_dependencies(self, task_id: str) -> list[str]:
        """Return list of dependency IDs that are not COMPLETE."""
        task = self._parser.get_task(task_id)
        if not task:
            return []

        incomplete = []
        for dep_id in task.get("dependencies", []):
            dep_status = self._state.get_task_status(dep_id)
            if dep_status != TaskStatus.COMPLETE.value:
                incomplete.append(dep_id)

        return incomplete

3.3 EventEmitter for Live Streaming

# zerg/event_emitter.py

class EventEmitter:
    def __init__(self, feature: str, state_dir: Path):
        self._event_file = state_dir / f"{feature}-events.jsonl"

    def emit(self, event_type: str, data: dict | None = None) -> None:
        """Append event to JSONL file."""
        entry = {
            "ts": datetime.now().isoformat(),
            "event": event_type,
            "data": data or {}
        }
        with open(self._event_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def subscribe(self, callback: Callable[[dict], None]) -> None:
        """Tail event file and call callback for new events."""
        # Implementation: use watchdog or simple polling

3.4 Modified claim_task()

# zerg/state.py

def claim_task(
    self,
    task_id: str,
    worker_id: int,
    current_level: int | None = None,
    dependency_checker: DependencyChecker | None = None
) -> bool:
    """Attempt to claim a task for a worker."""
    with self._atomic_update():
        task_state = self._state.get("tasks", {}).get(task_id, {})

        # NEW: Level enforcement
        if current_level is not None:
            task_level = task_state.get("level", 1)
            if task_level != current_level:
                logger.warning(f"Task {task_id} (L{task_level}) blocked - current level is {current_level}")
                return False

        # NEW: Dependency enforcement
        if dependency_checker is not None:
            if not dependency_checker.are_dependencies_complete(task_id):
                incomplete = dependency_checker.get_incomplete_dependencies(task_id)
                logger.warning(f"Task {task_id} blocked by incomplete deps: {incomplete}")
                return False

        # Existing checks...
        current_status = task_state.get("status", TaskStatus.PENDING.value)
        if current_status not in (TaskStatus.TODO.value, TaskStatus.PENDING.value):
            return False

        existing_worker = task_state.get("worker_id")
        if existing_worker is not None and existing_worker != worker_id:
            return False

        # Claim it
        self.set_task_status(task_id, TaskStatus.CLAIMED, worker_id=worker_id)
        self.record_task_claimed(task_id, worker_id)
        return True

3.5 Modified handle_level_complete()

# zerg/level_coordinator.py

def handle_level_complete(self, level: int) -> bool:
    """Handle level completion."""
    # Check config
    if self.config.rush.defer_merge_to_ship:
        # NEW: No merge during rush
        logger.info(f"Level {level} complete. Merge deferred to ship.")
        self.state.set_level_status(level, "complete")
        self._emit_event("level_complete", {"level": level, "merge": "deferred"})
        return True

    # Legacy behavior: merge to main
    # ... existing merge code ...

3.6 Modified finalize()

# zerg/merge.py

def finalize(self, staging_branch: str, target_branch: str) -> str:
    """Finalize merge to target branch."""
    # NEW: Use detached HEAD to avoid worktree lock
    current = self.git.current_branch()
    if current == staging_branch or current == target_branch:
        self.git.run(["checkout", "--detach"])

    # Checkout target and merge
    self.git.checkout(target_branch)
    commit = self.git.merge(staging_branch, message=f"ZERG: Complete level merge from {staging_branch}")

    # Cleanup
    self.git.delete_branch(staging_branch, force=True)

    return commit

4. Key Decisions

4.1 Decision: Config Flags for Backward Compatibility

Context: Need to avoid breaking existing workflows.

Options:

1. Hard-code new behavior (breaking change)
2. Config flags with new behavior as default
3. Config flags with old behavior as default

Decision: Option 2 - Config flags, new behavior default.

Rationale: Fixes bugs by default, but users can opt-out if needed.

4.2 Decision: Detached HEAD for finalize()

Context: Worktree lock errors when staging branch is checked out.

Options:

1. Dedicated merge worktree
2. Detached HEAD checkout
3. Force delete with --force

Decision: Option 2 - Detached HEAD.

Rationale: Simpler than worktree management, safer than force delete.

5. Implementation Plan

5.1 Task Summary

Total: 17 tasks, ~145m sequential, ~50m with 3 workers

5.2 File Ownership

6. Risk Assessment

7. Testing Strategy

7.1 Unit Tests

• test_dependency_checker.py: All dependency scenarios
• test_event_emitter.py: emit, subscribe, cleanup
• test_claim_task_enforcement.py: Level + dependency validation

7.2 Integration Tests

• test_deferred_merge.py: Main untouched during rush
• test_claim_enforcement.py: End-to-end level/dep enforcement

8. Parallel Execution Notes

8.1 Recommended Workers

• Minimum: 2 workers
• Optimal: 3 workers
• Maximum: 4 workers

8.2 Estimated Duration

• Single worker: ~145m
• With 3 workers: ~50m
• Speedup: ~3x

9. Approval