Stage 4: ARCHITECTURE — Fleet Evolution Engine v2

DISCOVER (KARL-ranked cohort selection)
  ↓
DISPATCH (Python orchestrator → AuraGateway skill command)
  ↓
PARSE (structured output → Supabase tasks)
  ↓
REVIEW (Codex adversarial → issue blocks → Supabase tasks)
  ↓
EVOLVE (Hydra cycles → Swift bridge polls quality gate)
  ↓
SHIP (quality gate passes → auto TestFlight trigger)
  ↓
LEARN (run as KARL trajectory → improve next cohort routing)

┌─────────────────────────────────────────────────────────┐
│                    SUPABASE (source of truth)            │
│  app_evolution_states  app_evolution_tasks  mesh_health  │
└──────────────┬────────────────────┬────────────────────-┘
               │ reads/writes        │ reads
               ▼                    ▼
┌──────────────────────┐   ┌───────────────────────────┐
│  Python Orchestrator  │   │   MeshControl Swift App   │
│  fleet_evolution_     │   │   FleetEvolutionClient    │
│  orchestrator.py      │   │   (read-only display)     │
│  (cloud-vm Prefect)   │   │   + enqueue UI            │
│                       │   │   + manual override       │
│  Writes: stage        │   │   + Hydra bridge (polls)  │
│  Reads: mesh_health   │   │   Writes: task completion │
└──────────────────────┘   └───────────────────────────┘
        │                              │
        ▼                              ▼
┌───────────────────┐        ┌──────────────────────┐
│  AuraGateway      │        │  HydraClient HTTP     │
│  Mac1 (:8001)     │        │  Mac1 (:8095)         │
│  Skill dispatch   │        │  Phase + quality data │
└───────────────────┘        └──────────────────────┘
        │
        ├──→ Mac1: claude skill runs (creative:forge, divergent-rail)
        ├──→ Mac4: /meta:omega + Codex adversarial review
        └──→ Mac3: creative content generation

class AppEvolution:
    def run(self, app_id: str, pipeline: str):
        # Stage 1: Skill
        task_id = aura.spawn(build_skill_cmd(app_id, pipeline))
        output = aura.poll_and_retrieve(task_id, timeout=3600)
        tasks = parse_skill_output(output, app_id)
        supabase.upsert_tasks(tasks)
        supabase.set_stage(app_id, 'skillComplete')

        # Stage 2: Adversarial
        review_task_id = aura.dispatch(build_review_cmd(app_id), target='mac4', model='codex')
        review = aura.poll_and_retrieve(review_task_id, timeout=1800)
        issues = parse_review_output(review, app_id)
        supabase.upsert_tasks(issues, source='adversarial')
        supabase.set_stage(app_id, 'adversarialComplete')

        # Stage 3: Hydra
        aura.spawn(f"/meta:hydra {app_id}")
        supabase.set_stage(app_id, 'hydraBuild')
        # Hydra bridge in Swift polls from here; orchestrator monitors
        # Timeout: 4 hours
        deadline = time.time() + 14400
        while time.time() < deadline:
            quality = hydra_client.get_quality(app_id)
            if quality['isPassing']:
                supabase.set_stage(app_id, 'complete')
                trigger_testflight(app_id)  # Auto-upload if isPassing
                record_trajectory(app_id, outcome='success')
                break
            time.sleep(300)  # Poll every 5min from Python
        else:
            supabase.set_stage(app_id, 'failed', error='Hydra timeout 4h')

Extract all actionable tasks from this output. For each task:
- priority: 1-5 (1=critical blocker, 5=nice-to-have)
- description: single sentence
- source: "skill"
Return JSON array of tasks.

Review app '{app_id}'. Prior skill tasks: {tasks_json}.
For each issue found, output an ISSUE block:
ISSUE FILE: <path> SEVERITY: critical|high|medium|low FIX: <description>
Output JSON at end: {"issues": [{"file": "...", "severity": "...", "description": "..."}]}

func startHydraMonitor(appId: String) {
    Task {
        while let state = states[appId], state.stage.phase == .hydra {
            try? await Task.sleep(nanoseconds: 10_000_000_000) // 10s
            let status = await HydraClient.shared.fetchStatus(for: appId)
            let mapped = EvolutionStage.fromHydraPhase(status.phase, cycle: status.cycle)
            if mapped != states[appId]?.stage {
                await updateStageLocally(appId: appId, stage: mapped)
                // Python orchestrator also updates Supabase; last-write-wins is fine
                // (both compute same mapped value from same Hydra status)
            }
            if status.quality.isPassing {
                // Let Python handle the official markComplete + TestFlight
                // But update UI immediately
                await MainActor.run { states[appId]?.stage = .complete }
                break
            }
        }
    }
}

def rank_apps_by_trajectory(fleet: list[str]) -> list[str]:
    # Query KARL trajectory scores from Supabase
    trajectories = supabase.select('trajectories', {'app_id': {'in': fleet}})
    scores = {t['app_id']: t['reward_mean'] for t in trajectories}
    # Apps with no trajectory: medium priority
    return sorted(fleet, key=lambda x: scores.get(x, 0.5), reverse=True)