feat(orchestrator): implement Proportional fairness and wire dual-mode run loop

- Add Proportional fairness policy enforcement in ConcurrencyController::acquire()
  that caps each mode at its fair share of global capacity
- Wire task_rx into DualModeOrchestrator::run() select loop with track_spawned_task()
- Add config() getter, task_sender(), and active agent tracking
- Remove all unused imports (no variable suppression -- every field fully used)
- Add MockTracker test exercising fetch_candidate_issues trait implementation

73 tests pass, clippy clean, fmt clean.

Co-Authored-By: Terraphim AI <noreply@anthropic.com>

Author: Terraphim CI

crates/terraphim_orchestrator/src/concurrency.rs

@@ -127,6 +127,11 @@ impl ConcurrencyController {
         }
     }
 
+    /// Get the active fairness policy.
+    pub fn fairness_policy(&self) -> FairnessPolicy {
+        self.fairness
+    }
+
     /// Acquire a slot for the given mode.
     async fn acquire(&self, mode: AgentMode) -> Option<AgentPermit> {
         // Check mode quota first
@@ -135,6 +140,31 @@ impl ConcurrencyController {
             return None;
         }
 
+        // Apply fairness policy: under Proportional, check whether the mode
+        // is consuming more than its fair share of global capacity.
+        if self.fairness == FairnessPolicy::Proportional {
+            let counts = self.running.lock().await;
+            let total = counts.time_driven + counts.issue_driven;
+            let global_cap = self.global.available_permits() + total;
+            if global_cap > 0 {
+                let mode_count = match mode {
+                    AgentMode::TimeDriven => counts.time_driven,
+                    AgentMode::IssueDriven => counts.issue_driven,
+                };
+                let mode_quota = match mode {
+                    AgentMode::TimeDriven => self.quotas.time_max,
+                    AgentMode::IssueDriven => self.quotas.issue_max,
+                };
+                let total_quota = self.quotas.time_max + self.quotas.issue_max;
+                // Fair share = global_cap * (mode_quota / total_quota)
+                let fair_share = (global_cap * mode_quota) / total_quota.max(1);
+                if mode_count >= fair_share && fair_share > 0 {
+                    tracing::debug!(?mode, mode_count, fair_share, "proportional fairness limit");
+                    return None;
+                }
+            }
+        }
+
         // Try to acquire global permit
         let global_permit = match self.global.clone().try_acquire_owned() {
             Ok(p) => p,

crates/terraphim_orchestrator/src/dispatcher.rs

@@ -3,7 +3,6 @@
 //! Provides a priority queue with fairness between different dispatch sources.
 
 use std::collections::VecDeque;
-use tokio::sync::{mpsc, oneshot};
 
 /// A dispatch task from any source.
 #[derive(Debug, Clone)]

crates/terraphim_orchestrator/src/dual_mode.rs

@@ -4,16 +4,15 @@
 //! with shared concurrency control and unified status.
 
 use crate::{
-    AgentDefinition, AgentLayer, AgentOrchestrator, CompoundReviewResult, ConcurrencyController,
-    DispatchTask, DispatcherStats, FairnessPolicy, HandoffContext, IssueMode, ModeQuotas,
-    NightwatchMonitor, OrchestratorConfig, ScheduleEvent, TimeMode, TimeScheduler, WorkflowConfig,
+    AgentDefinition, AgentOrchestrator, CompoundReviewResult, ConcurrencyController,
+    DispatcherStats, FairnessPolicy, HandoffContext, ModeQuotas, OrchestratorConfig, ScheduleEvent,
+    TimeScheduler, WorkflowConfig,
 };
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 use terraphim_tracker::{GiteaTracker, IssueTracker};
 use tokio::sync::{mpsc, watch, Mutex};
-use tokio::task::JoinHandle;
 use tracing::{error, info, warn};
 
 /// Shared state between time and issue modes.
@@ -88,8 +87,10 @@ pub struct DualModeOrchestrator {
     time_mode: Option<TimeModeComponents>,
     /// Issue mode controller (if enabled).
     issue_mode: Option<IssueModeComponents>,
-    /// Task receiver.
+    /// Task receiver from both modes.
     task_rx: mpsc::Receiver<SpawnTask>,
+    /// Task sender for dispatching from modes.
+    task_tx: mpsc::Sender<SpawnTask>,
     /// Active agents.
     active_agents: Arc<Mutex<HashMap<String, AgentId>>>,
 }
@@ -141,7 +142,7 @@ impl DualModeOrchestrator {
             shutdown_tx,
         };
 
-        // Create task channel
+        // Create task channel for modes to send spawned tasks to orchestrator
         let (task_tx, task_rx) = mpsc::channel(128);
 
         // Setup time mode
@@ -186,58 +187,95 @@ impl DualModeOrchestrator {
             time_mode,
             issue_mode,
             task_rx,
+            task_tx,
             active_agents: Arc::new(Mutex::new(HashMap::new())),
         })
     }
 
+    /// Access the orchestrator configuration.
+    pub fn config(&self) -> &OrchestratorConfig {
+        &self.config
+    }
+
     /// Run the dual-mode orchestrator.
     pub async fn run(&mut self) -> Result<(), crate::OrchestratorError> {
-        info!("starting dual-mode orchestrator");
+        info!(
+            agents = self.config.agents.len(),
+            workflow_enabled = self.config.workflow.as_ref().is_some_and(|w| w.enabled),
+            "starting dual-mode orchestrator"
+        );
 
         // Start time mode task
-        let time_handle = if let Some(time_components) = self.time_mode.take() {
+        let mut time_handle = if let Some(time_components) = self.time_mode.take() {
             let state = self.state.clone();
             Some(tokio::spawn(run_time_mode(time_components, state)))
         } else {
             None
         };
 
         // Start issue mode task
-        let issue_handle = if let Some(issue_components) = self.issue_mode.take() {
+        let mut issue_handle = if let Some(issue_components) = self.issue_mode.take() {
             let state = self.state.clone();
             Some(tokio::spawn(run_issue_mode(issue_components, state)))
         } else {
             None
         };
 
-        // Wait for shutdown signal or task completion
+        // Wait for shutdown signal, task completion, or spawned tasks
         let ctrl_c = tokio::signal::ctrl_c();
         tokio::pin!(ctrl_c);
 
-        tokio::select! {
-            _ = async {
-                if let Some(h) = time_handle {
-                    let _ = h.await;
+        // Pin the optional handles for select
+        let mut time_done = false;
+        let mut issue_done = false;
+
+        loop {
+            tokio::select! {
+                // Receive spawned tasks from modes and track them
+                Some(task) = self.task_rx.recv() => {
+                    self.track_spawned_task(task).await;
                 }
-            } => {
-                info!("time mode completed");
-            }
-            _ = async {
-                if let Some(h) = issue_handle {
-                    let _ = h.await;
+                // Wait for time mode completion
+                result = async {
+                    match &mut time_handle {
+                        Some(h) => h.await,
+                        None => std::future::pending().await,
+                    }
+                }, if !time_done => {
+                    time_done = true;
+                    match result {
+                        Ok(()) => info!("time mode completed"),
+                        Err(e) => error!("time mode panicked: {}", e),
+                    }
+                    if issue_done { break; }
                 }
-            } => {
-                info!("issue mode completed");
-            }
-            result = self.base.run() => {
-                match result {
-                    Ok(()) => info!("base orchestrator completed"),
-                    Err(e) => error!("base orchestrator error: {}", e),
+                // Wait for issue mode completion
+                result = async {
+                    match &mut issue_handle {
+                        Some(h) => h.await,
+                        None => std::future::pending().await,
+                    }
+                }, if !issue_done => {
+                    issue_done = true;
+                    match result {
+                        Ok(()) => info!("issue mode completed"),
+                        Err(e) => error!("issue mode panicked: {}", e),
+                    }
+                    if time_done { break; }
+                }
+                // Base orchestrator reconciliation loop
+                result = self.base.run() => {
+                    match result {
+                        Ok(()) => info!("base orchestrator completed"),
+                        Err(e) => error!("base orchestrator error: {}", e),
+                    }
+                    break;
+                }
+                _ = &mut ctrl_c => {
+                    info!("shutdown signal received");
+                    let _ = self.state.shutdown_tx.send(true);
+                    break;
                 }
-            }
-            _ = &mut ctrl_c => {
-                info!("shutdown signal received");
-                let _ = self.state.shutdown_tx.send(true);
             }
         }
 
@@ -248,6 +286,43 @@ impl DualModeOrchestrator {
         Ok(())
     }
 
+    /// Track a spawned task from either mode.
+    async fn track_spawned_task(&self, task: SpawnTask) {
+        let mut stats = self.state.stats.lock().await;
+        stats.total_agents_spawned += 1;
+        match &task {
+            SpawnTask::TimeTask { agent } => {
+                info!(agent_name = %agent.name, "received time-driven spawn task");
+                let mut agents = self.active_agents.lock().await;
+                agents.insert(
+                    agent.name.clone(),
+                    AgentId {
+                        name: agent.name.clone(),
+                        mode: ExecutionMode::TimeDriven,
+                    },
+                );
+                *stats.active_by_mode.entry("time".into()).or_insert(0) += 1;
+            }
+            SpawnTask::IssueTask { issue_id, title } => {
+                info!(issue_id = %issue_id, title = %title, "received issue-driven spawn task");
+                let mut agents = self.active_agents.lock().await;
+                agents.insert(
+                    issue_id.clone(),
+                    AgentId {
+                        name: issue_id.clone(),
+                        mode: ExecutionMode::IssueDriven,
+                    },
+                );
+                *stats.active_by_mode.entry("issue".into()).or_insert(0) += 1;
+            }
+        }
+    }
+
+    /// Get a clone of the task sender for external dispatch.
+    pub fn task_sender(&self) -> mpsc::Sender<SpawnTask> {
+        self.task_tx.clone()
+    }
+
     /// Request shutdown.
     pub fn request_shutdown(&self) {
         let _ = self.state.shutdown_tx.send(true);

crates/terraphim_orchestrator/src/mode/issue.rs

@@ -3,7 +3,6 @@
 //! Polls Gitea/Linear for issues and dispatches agents to work on them.
 
 use crate::{ConcurrencyController, DispatchTask, Dispatcher, WorkflowConfig};
-use std::collections::HashMap;
 use std::time::Duration;
 use terraphim_tracker::{Issue, IssueTracker, PagerankClient};
 use tracing::{error, info, warn};
@@ -230,6 +229,31 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_mock_tracker_fetch_candidate_issues() {
+        let tracker = MockTracker {
+            issues: vec![Issue {
+                id: "1".into(),
+                identifier: "TEST-1".into(),
+                title: "Test Issue".into(),
+                description: None,
+                priority: Some(1),
+                state: "open".into(),
+                branch_name: None,
+                url: None,
+                labels: vec![],
+                blocked_by: vec![],
+                pagerank_score: None,
+                created_at: None,
+                updated_at: None,
+            }],
+        };
+
+        let issues = tracker.fetch_candidate_issues().await.unwrap();
+        assert_eq!(issues.len(), 1);
+        assert_eq!(issues[0].identifier, "TEST-1");
+    }
+
     #[test]
     fn test_compute_sort_score() {
         let issue1 = Issue {

crates/terraphim_orchestrator/src/mode/time.rs

@@ -3,8 +3,7 @@
 //! Manages cron-scheduled agents using the unified dispatcher.
 
 use crate::{
-    AgentDefinition, AgentLayer, ConcurrencyController, DispatchTask, Dispatcher, ScheduleEvent,
-    TimeScheduler,
+    AgentDefinition, ConcurrencyController, DispatchTask, Dispatcher, ScheduleEvent, TimeScheduler,
 };
 use tracing::{error, info, warn};
 
@@ -152,7 +151,7 @@ impl TimeMode {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::ModeQuotas;
+    use crate::{AgentLayer, ModeQuotas};
 
     fn test_agent(name: &str, layer: AgentLayer) -> AgentDefinition {
         AgentDefinition {