From 013a5da60b96e8ee4cdc88ddfc15c0b4d2995a26 Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Tue, 21 Jul 2026 15:15:41 +0000 Subject: [PATCH] huskies: merge 1243 refactor Merge jobs stop bloating replicated CRDT state --- .../src/agents/pool/pipeline/merge/runner.rs | 16 +- server/src/service/merge/mod.rs | 5 +- server/src/service/merge/summary.rs | 186 ++++++++++++++++++ 3 files changed, 203 insertions(+), 4 deletions(-) diff --git a/server/src/agents/pool/pipeline/merge/runner.rs b/server/src/agents/pool/pipeline/merge/runner.rs index 84035142..8ebb5c50 100644 --- a/server/src/agents/pool/pipeline/merge/runner.rs +++ b/server/src/agents/pool/pipeline/merge/runner.rs @@ -299,10 +299,17 @@ impl AgentPool { crate::db::write_content(crate::db::ContentKey::MergeSuccess(&sid), "1"); } - // Update CRDT with terminal status. + // Update CRDT with terminal status. The full untruncated output is + // already on disk (write_merge_report, called from + // run_merge_pipeline for the Ok(r) case above, or below for the + // Err(e) case) — only a bounded summary plus that pointer goes + // into the replicated `merge_jobs.error` field, so a large gate + // failure doesn't bloat every node's CRDT state. match &report { Ok(r) => { - let report_json = serde_json::to_string(r).unwrap_or_else(|_| String::new()); + let bounded = crate::service::merge::bound_report_for_storage(r); + let report_json = + serde_json::to_string(&bounded).unwrap_or_else(|_| String::new()); crate::crdt_state::write_merge_job( &sid, "completed", @@ -312,12 +319,15 @@ impl AgentPool { ); } Err(e) => { + let report_path = crate::service::merge::io::write_merge_report(&root, &sid, e); + let bounded = + crate::service::merge::bound_plain_error(e, report_path.as_deref()); crate::crdt_state::write_merge_job( &sid, "failed", started_at, Some(finished_at), - Some(e), + Some(&bounded), ); } } diff --git a/server/src/service/merge/mod.rs b/server/src/service/merge/mod.rs index abe4333b..d9a4c216 100644 --- a/server/src/service/merge/mod.rs +++ b/server/src/service/merge/mod.rs @@ -16,7 +16,10 @@ pub mod summary; #[allow(unused_imports)] pub use status::format_merge_status_message; #[allow(unused_imports)] -pub use summary::{summarize_merge_failure_kind, summarize_merge_result}; +pub use summary::{ + bound_plain_error, bound_report_for_storage, summarize_merge_failure_kind, + summarize_merge_result, +}; // ── Error type ──────────────────────────────────────────────────────────────── diff --git a/server/src/service/merge/summary.rs b/server/src/service/merge/summary.rs index ce8c94e6..5d7c7d97 100644 --- a/server/src/service/merge/summary.rs +++ b/server/src/service/merge/summary.rs @@ -113,6 +113,78 @@ pub fn summarize_merge_failure_kind(kind: &MergeFailureKind) -> String { summarize(failing_gate_label_for_kind(kind), &kind.to_gate_output()) } +/// Return a copy of `result` with its embedded output string (`gate_output` +/// on [`MergeResult::Success`], `output` on every other variant) bounded to +/// [`SUMMARY_TAIL_LINES`]. +/// +/// Used before storing a [`MergeResult`] in the replicated CRDT `merge_jobs` +/// collection: the full untruncated text already lives on disk (see +/// `service::merge::io::write_merge_report` and `MergeReport::report_path`), +/// so only a bounded tail needs to travel through CRDT sync. +pub fn bound_result_for_storage(result: &MergeResult) -> MergeResult { + match result { + MergeResult::Success { + conflicts_resolved, + conflict_details, + gate_output, + } => MergeResult::Success { + conflicts_resolved: *conflicts_resolved, + conflict_details: conflict_details.clone(), + gate_output: truncate_gate_output(gate_output, SUMMARY_TAIL_LINES), + }, + MergeResult::Conflict { details, output } => MergeResult::Conflict { + details: details.clone(), + output: truncate_gate_output(output, SUMMARY_TAIL_LINES), + }, + MergeResult::GateFailure { + output, + failure_kind, + } => MergeResult::GateFailure { + output: truncate_gate_output(output, SUMMARY_TAIL_LINES), + failure_kind: failure_kind.clone(), + }, + MergeResult::NoCommits { output } => MergeResult::NoCommits { + output: truncate_gate_output(output, SUMMARY_TAIL_LINES), + }, + MergeResult::Other { + output, + conflict_details, + } => MergeResult::Other { + output: truncate_gate_output(output, SUMMARY_TAIL_LINES), + conflict_details: conflict_details.clone(), + }, + } +} + +/// Return a copy of `report` whose embedded [`MergeResult`] output is bounded +/// via [`bound_result_for_storage`]; `report_path` (the pointer to the full +/// untruncated text on disk) and every other field are carried over as-is. +pub fn bound_report_for_storage( + report: &crate::agents::merge::MergeReport, +) -> crate::agents::merge::MergeReport { + crate::agents::merge::MergeReport { + story_id: report.story_id.clone(), + result: bound_result_for_storage(&report.result), + worktree_cleaned_up: report.worktree_cleaned_up, + story_archived: report.story_archived, + report_path: report.report_path.clone(), + } +} + +/// Bound a plain (non-[`MergeResult`]) error string for CRDT storage, +/// appending a pointer to the full text on disk when `report_path` is given. +/// +/// Used for the hard-error path (git/process failures that short-circuit +/// before a [`MergeResult`] is ever constructed), so the CRDT `merge_jobs` +/// error field stays bounded the same way completed-job entries do. +pub fn bound_plain_error(error: &str, report_path: Option<&str>) -> String { + let truncated = truncate_gate_output(error, SUMMARY_TAIL_LINES); + match report_path { + Some(path) => format!("{truncated}\n\nFull output: {path}"), + None => truncated, + } +} + #[cfg(test)] mod tests { use super::*; @@ -236,4 +308,118 @@ mod tests { let summary = summarize_merge_failure_kind(&kind); assert!(!summary.contains("Quality gates failed:")); } + + // ── bound_result_for_storage ──────────────────────────────────────────────── + + fn long_output(n: usize) -> String { + (1..=n) + .map(|i| format!("line{i}")) + .collect::>() + .join("\n") + } + + #[test] + fn bound_result_for_storage_truncates_gate_failure_output() { + let result = MergeResult::GateFailure { + output: long_output(100), + failure_kind: Some(GateFailureKind::Test), + }; + let bounded = bound_result_for_storage(&result); + assert!(bounded.output().len() < result.output().len()); + assert!(bounded.output().contains("line100")); + assert!(matches!( + bounded, + MergeResult::GateFailure { + failure_kind: Some(GateFailureKind::Test), + .. + } + )); + } + + #[test] + fn bound_result_for_storage_truncates_success_gate_output() { + let result = MergeResult::Success { + conflicts_resolved: true, + conflict_details: Some("resolved automatically".to_string()), + gate_output: long_output(100), + }; + let bounded = bound_result_for_storage(&result); + assert!(bounded.output().len() < result.output().len()); + assert!(bounded.output().contains("line100")); + match bounded { + MergeResult::Success { + conflicts_resolved, + conflict_details, + .. + } => { + assert!(conflicts_resolved); + assert_eq!(conflict_details.as_deref(), Some("resolved automatically")); + } + other => panic!("expected Success, got {other:?}"), + } + } + + #[test] + fn bound_result_for_storage_leaves_short_output_unchanged() { + let result = MergeResult::NoCommits { + output: "no commits to merge".to_string(), + }; + let bounded = bound_result_for_storage(&result); + assert_eq!(bounded.output(), "no commits to merge"); + } + + // ── bound_report_for_storage ──────────────────────────────────────────────── + + #[test] + fn bound_report_for_storage_truncates_output_and_keeps_report_path() { + let report = crate::agents::merge::MergeReport { + story_id: "42_story".to_string(), + result: MergeResult::GateFailure { + output: long_output(100), + failure_kind: Some(GateFailureKind::Test), + }, + worktree_cleaned_up: false, + story_archived: false, + report_path: Some(".huskies/merge_reports/42_story-123.log".to_string()), + }; + let bounded = bound_report_for_storage(&report); + assert!(bounded.result.output().len() < report.result.output().len()); + assert_eq!( + bounded.report_path.as_deref(), + report.report_path.as_deref() + ); + assert_eq!(bounded.story_id, report.story_id); + + // Round-trips through the same JSON shape `get_merge_status` expects. + let json = serde_json::to_string(&bounded).unwrap(); + let decoded: crate::agents::merge::MergeReport = serde_json::from_str(&json).unwrap(); + assert!(decoded.result.output().contains("line100")); + assert_eq!( + decoded.report_path.as_deref(), + Some(".huskies/merge_reports/42_story-123.log") + ); + } + + // ── bound_plain_error ──────────────────────────────────────────────────────── + + #[test] + fn bound_plain_error_appends_pointer_when_path_given() { + let bounded = bound_plain_error("boom", Some(".huskies/merge_reports/1_story-1.log")); + assert!(bounded.contains("boom")); + assert!(bounded.contains("Full output: .huskies/merge_reports/1_story-1.log")); + } + + #[test] + fn bound_plain_error_no_pointer_when_no_path() { + let bounded = bound_plain_error("boom", None); + assert_eq!(bounded, "boom"); + } + + #[test] + fn bound_plain_error_truncates_long_error() { + let error = long_output(100); + let bounded = bound_plain_error(&error, None); + assert!(bounded.len() < error.len()); + assert!(bounded.contains("line100")); + } }