Merge pull request 'feat(orchestrator): OutputPoster and drain-before-drop #146' (#168) from task/146-output-poster-v2 into main

Author: AlexMikhalev

crates/terraphim_orchestrator/src/config.rs

@@ -37,6 +37,18 @@ pub struct OrchestratorConfig {
     /// Directory for flow run state persistence.
     #[serde(default)]
     pub flow_state_dir: Option<PathBuf>,
+    /// Gitea configuration for posting agent output to issues.
+    #[serde(default)]
+    pub gitea: Option<GiteaOutputConfig>,
+}
+
+/// Configuration for posting agent output to Gitea issues.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct GiteaOutputConfig {
+    pub base_url: String,
+    pub token: String,
+    pub owner: String,
+    pub repo: String,
 }
 
 /// Lightweight reference to an SFIA skill code and level.
@@ -97,6 +109,9 @@ pub struct AgentDefinition {
     /// Maximum CPU seconds allowed per agent execution.
     #[serde(default)]
     pub max_cpu_seconds: Option<u64>,
+    /// Gitea issue number to post output to (optional).
+    #[serde(default)]
+    pub gitea_issue: Option<u64>,
 }
 
 /// Agent layer in the dark factory hierarchy.

crates/terraphim_orchestrator/src/flow/executor.rs

@@ -169,6 +169,7 @@ impl FlowExecutor {
             fallback_model: None,
             grace_period_secs: None,
             max_cpu_seconds: None,
+            gitea_issue: None,
         };
 
         // Build provider for spawner

crates/terraphim_orchestrator/src/lib.rs

@@ -40,21 +40,23 @@ pub mod handoff;
 pub mod mode;
 pub mod nightwatch;
 pub mod persona;
+pub mod output_poster;
 pub mod scheduler;
 pub mod scope;
 
 pub use compound::{CompoundReviewResult, CompoundReviewWorkflow, ReviewGroupDef, SwarmConfig};
 pub use concurrency::{ConcurrencyController, FairnessPolicy, ModeQuotas};
 pub use config::{
-    AgentDefinition, AgentLayer, CompoundReviewConfig, ConcurrencyConfig, NightwatchConfig,
-    OrchestratorConfig, TrackerConfig, TrackerStates, WorkflowConfig,
+    AgentDefinition, AgentLayer, CompoundReviewConfig, ConcurrencyConfig, GiteaOutputConfig,
+    NightwatchConfig, OrchestratorConfig, TrackerConfig, TrackerStates, WorkflowConfig,
 };
 pub use cost_tracker::{BudgetVerdict, CostSnapshot, CostTracker};
 pub use dispatcher::{DispatchTask, Dispatcher, DispatcherStats};
 pub use dual_mode::DualModeOrchestrator;
 pub use error::OrchestratorError;
 pub use handoff::{HandoffBuffer, HandoffContext, HandoffLedger};
 pub use mode::{IssueMode, TimeMode};
+pub use output_poster::OutputPoster;
 pub use nightwatch::{
     dual_panel_evaluate, validate_certificate, Claim, CorrectionAction, CorrectionLevel,
     DriftAlert, DriftMetrics, DriftScore, DualPanelResult, NightwatchMonitor, RateLimitTracker,
@@ -129,6 +131,8 @@ pub struct AgentOrchestrator {
     persona_registry: PersonaRegistry,
     /// Renderer for persona metaprompts.
     metaprompt_renderer: MetapromptRenderer,
+    /// Output poster for posting agent output to Gitea issues.
+    output_poster: Option<OutputPoster>,
     /// Circuit breakers for each provider to prevent cascading failures.
     #[allow(dead_code)]
     circuit_breakers: Arc<Mutex<HashMap<String, CircuitBreaker>>>,
@@ -203,6 +207,9 @@ impl AgentOrchestrator {
             None => MetapromptRenderer::new().expect("default template should always compile"),
         };
 
+        // Initialize output poster if Gitea config is provided
+        let output_poster = config.gitea.as_ref().map(OutputPoster::new);
+
         Ok(Self {
             config,
             spawner,
@@ -221,6 +228,7 @@ impl AgentOrchestrator {
             cost_tracker,
             persona_registry,
             metaprompt_renderer,
+            output_poster,
             circuit_breakers: Arc::new(Mutex::new(HashMap::new())),
             active_flows: HashMap::new(),
         })
@@ -736,7 +744,35 @@ impl AgentOrchestrator {
             }
         }
 
-        // Process exits
+        // Drain output from exiting agents BEFORE removing them
+        for (name, def, status) in &exited {
+            // Drain remaining output events
+            let mut output_lines: Vec<String> = Vec::new();
+            if let Some(managed) = self.active_agents.get_mut(name) {
+                while let Ok(event) = managed.output_rx.try_recv() {
+                    self.nightwatch.observe(name, &event);
+                    match &event {
+                        crate::OutputEvent::Stdout { line, .. } => {
+                            output_lines.push(line.clone());
+                        }
+                        crate::OutputEvent::Stderr { line, .. } => {
+                            output_lines.push(format!("[stderr] {}", line));
+                        }
+                        _ => {}
+                    }
+                }
+            }
+
+            // Post output to Gitea if configured
+            if let (Some(poster), Some(issue)) = (&self.output_poster, def.gitea_issue) {
+                let exit_code = status.code();
+                if let Err(e) = poster.post_agent_output(name, issue, &output_lines, exit_code).await {
+                    warn!(agent = %name, issue = issue, error = %e, "failed to post output to Gitea");
+                }
+            }
+        }
+
+        // NOW remove from active_agents and handle exits
         for (name, def, status) in exited {
             self.active_agents.remove(&name);
             self.handle_agent_exit(&name, &def, status);
@@ -1100,6 +1136,7 @@ mod tests {
                     fallback_model: None,
                     grace_period_secs: None,
                     max_cpu_seconds: None,
+                    gitea_issue: None,
                 },
                 AgentDefinition {
                     name: "sync".to_string(),
@@ -1120,6 +1157,7 @@ mod tests {
                     fallback_model: None,
                     grace_period_secs: None,
                     max_cpu_seconds: None,
+                    gitea_issue: None,
                 },
             ],
             restart_cooldown_secs: 60,
@@ -1129,6 +1167,7 @@ mod tests {
             persona_data_dir: None,
             flows: vec![],
             flow_state_dir: None,
+            gitea: None,
         }
     }
 
@@ -1293,6 +1332,7 @@ task = "test"
                 fallback_model: None,
                 grace_period_secs: None,
                 max_cpu_seconds: None,
+                gitea_issue: None,
             }],
             restart_cooldown_secs: 0, // instant restart for testing
             max_restart_count: 3,
@@ -1301,6 +1341,7 @@ task = "test"
             persona_data_dir: None,
             flows: vec![],
             flow_state_dir: None,
+            gitea: None,
         }
     }
 
@@ -1376,6 +1417,7 @@ task = "test"
             fallback_model: None,
             grace_period_secs: None,
             max_cpu_seconds: None,
+            gitea_issue: None,
         }];
         let mut orch = AgentOrchestrator::new(config).unwrap();
 
@@ -1510,6 +1552,7 @@ task = "test"
             fallback_model: None,
             grace_period_secs: None,
             max_cpu_seconds: None,
+            gitea_issue: None,
         }];
 
         // Set up persona data dir with a test persona
@@ -1592,6 +1635,7 @@ sfia_skills = [{ code = "TEST", name = "Testing", level = 4, description = "Desi
             fallback_model: None,
             grace_period_secs: None,
             max_cpu_seconds: None,
+            gitea_issue: None,
         }];
 
         // No persona_data_dir, so registry will be empty

crates/terraphim_orchestrator/src/mode/time.rs

@@ -176,6 +176,7 @@ mod tests {
             fallback_model: None,
             grace_period_secs: None,
             max_cpu_seconds: None,
+            gitea_issue: None,
         }
     }
 

crates/terraphim_orchestrator/src/output_poster.rs

@@ -0,0 +1,87 @@
+//! Posts agent output to Gitea issues after agent exit.
+
+use terraphim_tracker::GiteaTracker;
+use terraphim_tracker::gitea::GiteaConfig;
+
+use crate::config::GiteaOutputConfig;
+
+/// Posts collected agent output to a Gitea issue comment.
+pub struct OutputPoster {
+    tracker: GiteaTracker,
+}
+
+impl OutputPoster {
+    /// Create a new OutputPoster from Gitea output configuration.
+    pub fn new(config: &GiteaOutputConfig) -> Self {
+        let gitea_config = GiteaConfig {
+            base_url: config.base_url.clone(),
+            token: config.token.clone(),
+            owner: config.owner.clone(),
+            repo: config.repo.clone(),
+            active_states: vec!["open".to_string()],
+            terminal_states: vec!["closed".to_string()],
+            use_robot_api: false,
+        };
+        Self {
+            tracker: GiteaTracker::new(gitea_config).expect("Failed to create GiteaTracker"),
+        }
+    }
+
+    /// Post agent output as a comment on the given Gitea issue.
+    ///
+    /// Truncates output to 60000 bytes to stay within Gitea's comment size limit.
+    pub async fn post_agent_output(
+        &self,
+        agent_name: &str,
+        issue_number: u64,
+        output_lines: &[String],
+        exit_code: Option<i32>,
+    ) -> Result<(), String> {
+        if output_lines.is_empty() {
+            tracing::debug!(agent = %agent_name, issue = issue_number, "no output to post");
+            return Ok(());
+        }
+
+        let exit_str = match exit_code {
+            Some(code) => format!("exit code {}", code),
+            None => "unknown exit".to_string(),
+        };
+
+        let mut body = format!(
+            "**Agent `{}`** completed ({}).\n\n<details>\n<summary>Output ({} lines)</summary>\n\n```\n",
+            agent_name,
+            exit_str,
+            output_lines.len()
+        );
+
+        let joined = output_lines.join("\n");
+        // Truncate to stay within Gitea limits (~65535 bytes)
+        let max_output = 60000;
+        if joined.len() > max_output {
+            body.push_str(&joined[..max_output]);
+            body.push_str("\n... (truncated)\n");
+        } else {
+            body.push_str(&joined);
+        }
+        body.push_str("\n```\n\n</details>");
+
+        match self.tracker.post_comment(issue_number, &body).await {
+            Ok(comment) => {
+                tracing::info!(
+                    agent = %agent_name,
+                    issue = issue_number,
+                    comment_id = comment.id,
+                    "posted agent output to Gitea"
+                );
+                Ok(())
+            }
+            Err(e) => {
+                let msg = format!("failed to post output for {}: {}", agent_name, e);
+                tracing::error!("{}", msg);
+                Err(msg)
+            }
+        }
+    }
+}
+
+

crates/terraphim_orchestrator/src/scheduler.rs

@@ -163,6 +163,7 @@ mod tests {
             fallback_model: None,
             grace_period_secs: None,
             max_cpu_seconds: None,
+            gitea_issue: None,
         }
     }
 

crates/terraphim_orchestrator/tests/orchestrator_tests.rs

@@ -44,6 +44,7 @@ fn test_config() -> OrchestratorConfig {
                 fallback_model: None,
                 grace_period_secs: None,
                 max_cpu_seconds: None,
+                gitea_issue: None,
             },
             AgentDefinition {
                 name: "sync".to_string(),
@@ -64,6 +65,7 @@ fn test_config() -> OrchestratorConfig {
                 fallback_model: None,
                 grace_period_secs: None,
                 max_cpu_seconds: None,
+                gitea_issue: None,
             },
             AgentDefinition {
                 name: "reviewer".to_string(),
@@ -84,6 +86,7 @@ fn test_config() -> OrchestratorConfig {
                 fallback_model: None,
                 grace_period_secs: None,
                 max_cpu_seconds: None,
+                gitea_issue: None,
             },
         ],
         restart_cooldown_secs: 60,
@@ -93,6 +96,7 @@ fn test_config() -> OrchestratorConfig {
         persona_data_dir: None,
         flows: vec![],
         flow_state_dir: None,
+        gitea: None,
     }
 }
 

crates/terraphim_orchestrator/tests/scheduler_tests.rs

@@ -20,6 +20,7 @@ fn make_agent(name: &str, layer: AgentLayer, schedule: Option<&str>) -> AgentDef
         fallback_model: None,
         grace_period_secs: None,
         max_cpu_seconds: None,
+        gitea_issue: None,
     }
 }