Skip to content

Commit cb87dfc

Browse files
Merge pull request #16 from New1Direction/feat/swarm-campaign-live
fix(swarm): make the multi-persona campaign actually work (stdout was corrupting ACP)
2 parents b24a9a7 + 54ab4ae commit cb87dfc

7 files changed

Lines changed: 191 additions & 26 deletions

File tree

crates/korg-core/src/telemetry.rs

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,10 @@ pub fn init_tracing() {
3333
// Machine-readable JSON for log shipping / structured analysis
3434
let json_layer = fmt::layer()
3535
.json()
36+
// Logs go to STDERR. STDOUT is reserved for program output — and, in
37+
// `korg worker` subprocesses, for the ACP envelope stream the leader
38+
// parses as JSON. A log line on stdout corrupts that protocol.
39+
.with_writer(std::io::stderr)
3640
.with_current_span(true)
3741
.with_span_list(true)
3842
.with_target(true)
@@ -46,6 +50,9 @@ pub fn init_tracing() {
4650
} else {
4751
// Human-readable pretty output for development
4852
let pretty_layer = fmt::layer()
53+
// Logs go to STDERR (see note above): stdout is the ACP channel for
54+
// worker subprocesses and the program-output channel for the leader.
55+
.with_writer(std::io::stderr)
4956
.with_target(true)
5057
.with_thread_ids(false)
5158
.with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)

crates/korg-runtime/src/harness.rs

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,11 @@ impl SingleWorkerHarness {
3333
}
3434

3535
/// Main worker loop (legacy stub path).
36+
///
37+
/// Diagnostics go to STDERR: if this path is ever reached from a worker
38+
/// subprocess, stdout must stay a clean ACP envelope channel.
3639
pub async fn run(&mut self, client: &mut AcpClient) -> Result<()> {
37-
println!(
40+
eprintln!(
3841
"[Harness] Worker {} entering main loop (legacy client path)",
3942
self.worker_id
4043
);
@@ -49,7 +52,7 @@ impl SingleWorkerHarness {
4952
permissions,
5053
..
5154
} => {
52-
println!(
55+
eprintln!(
5356
"[Harness] Received RouteWork with base_snapshot: {}",
5457
base_snapshot
5558
);
@@ -64,12 +67,12 @@ impl SingleWorkerHarness {
6467
.await?;
6568
}
6669
_ => {
67-
println!("[Harness] Received unhandled message: {:?}", msg);
70+
eprintln!("[Harness] Received unhandled message: {:?}", msg);
6871
}
6972
}
7073
}
7174

72-
println!("[Harness] Worker {} exiting after task", self.worker_id);
75+
eprintln!("[Harness] Worker {} exiting after task", self.worker_id);
7376
Ok(())
7477
}
7578

@@ -142,6 +145,10 @@ impl SingleWorkerHarness {
142145
eprintln!("[Harness] Sent signed tool result back to leader");
143146
}
144147
}
148+
// handle_route_work already sent the authoritative
149+
// TerminationReport (carrying the real success/doom_loop
150+
// status + terminal_tx_id). The leader reads it now that
151+
// stdout is a clean ACP channel — no extra report needed.
145152
}
146153

147154
// === Direct tool request (if worker is sent a tool as first message) ===
@@ -346,7 +353,8 @@ impl SingleWorkerHarness {
346353
loop {
347354
ticker.tick().await;
348355
tick += 1;
349-
println!(
356+
// stderr, not stdout — stdout is the worker's ACP envelope channel.
357+
eprintln!(
350358
"[TelemetryEmitter] {} – live pulse #{} (continuous real-time telemetry)",
351359
rid, tick
352360
);

crates/korg-runtime/src/personas.rs

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -276,7 +276,15 @@ impl LlmPersona {
276276
top_p: self.top_p,
277277
presence_penalty: self.presence_penalty,
278278
frequency_penalty: self.frequency_penalty,
279-
response_format: None,
279+
// Implementer personas (Benjamin/Lucas) must emit a JSON mutations
280+
// block; ask OpenAI-compatible live providers for strictly valid JSON
281+
// so the patch reliably parses. Planner/researcher/evaluator personas
282+
// produce prose and must NOT be forced to JSON. The deterministic stub
283+
// ignores this field, so the hermetic path is unchanged.
284+
response_format: match self.persona {
285+
Persona::Benjamin | Persona::Lucas => Some("json_object".to_string()),
286+
_ => None,
287+
},
280288
};
281289

282290
let response = self.provider.complete(request).await?;

crates/korg-runtime/src/session.rs

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -358,6 +358,17 @@ impl SessionBackend for SubprocessBackend {
358358
cmd.env(k, v);
359359
}
360360

361+
// The worker builds its own LLM provider from `KorgConfig::load()`, which
362+
// reads these env vars. The child inherits the parent env by default, but
363+
// forward them explicitly so the worker's provider selection (e.g.
364+
// `--provider ollama` exported by the CLI) is visible here and survives
365+
// even if the inherited environment is ever cleared.
366+
for key in ["KORG_DEFAULT_LLM", "KORG_MODEL", "OLLAMA_BASE_URL"] {
367+
if let Ok(val) = std::env::var(key) {
368+
cmd.env(key, val);
369+
}
370+
}
371+
361372
#[cfg(unix)]
362373
{
363374
use std::os::unix::process::CommandExt;

src/main.rs

Lines changed: 47 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,21 @@ struct Cli {
106106
#[arg(long)]
107107
speculative: bool,
108108

109+
/// LLM provider for the whole swarm: `deterministic` (default, hermetic) or
110+
/// `ollama` (live local model — every persona does real, measured work).
111+
/// Exported as `KORG_DEFAULT_LLM` so each worker subprocess builds it too.
112+
#[arg(long, global = true)]
113+
provider: Option<String>,
114+
115+
/// Model name for a live provider (e.g. `qwen2.5:7b`). Exported as `KORG_MODEL`.
116+
#[arg(long, global = true)]
117+
model: Option<String>,
118+
119+
/// Base URL for a live provider (ollama default http://localhost:11434/v1).
120+
/// Exported as `OLLAMA_BASE_URL`.
121+
#[arg(long, global = true)]
122+
base_url: Option<String>,
123+
109124
#[command(subcommand)]
110125
command: Option<Commands>,
111126
}
@@ -260,18 +275,6 @@ enum Commands {
260275
/// Target repo. Defaults to a temp git-inited copy of the bundled fixture.
261276
#[arg(long)]
262277
repo: Option<std::path::PathBuf>,
263-
264-
/// Provider: `deterministic` (default, hermetic) or `ollama` (live local model).
265-
#[arg(long, default_value = "deterministic")]
266-
provider: String,
267-
268-
/// Model name for live providers (e.g. `qwen2.5:7b` for ollama).
269-
#[arg(long)]
270-
model: Option<String>,
271-
272-
/// Base URL override for the live provider (ollama default: http://localhost:11434/v1).
273-
#[arg(long)]
274-
base_url: Option<String>,
275278
},
276279

277280
/// Run the premium Claude Code cooperative session replay and speculative rewind demo
@@ -494,6 +497,27 @@ async fn main() -> Result<()> {
494497
korg_registry::IS_PREVIEW_MODE.store(true, std::sync::atomic::Ordering::Relaxed);
495498
}
496499

500+
// Export the swarm's LLM provider selection so every worker SUBPROCESS — each
501+
// a separate `korg worker` OS process that builds its own provider via
502+
// `KorgConfig::load()` (which reads these env vars) — uses it. Set once here at
503+
// startup; the children inherit the environment on spawn. This makes the full
504+
// multi-persona campaign run on a real model (`--provider ollama`) with no
505+
// config threading. `run-once` reads the flags directly and ignores this.
506+
//
507+
// `set_var` is unsound if another thread reads the environment concurrently.
508+
// These writes run once at the very top of `main()`, before any campaign task,
509+
// worker spawn, or `KorgConfig::load()` is reached — so no concurrent env
510+
// access is in flight here. (Edition 2021; `set_var` is still a safe fn.)
511+
if let Some(provider) = &cli.provider {
512+
std::env::set_var("KORG_DEFAULT_LLM", provider);
513+
if let Some(model) = &cli.model {
514+
std::env::set_var("KORG_MODEL", model);
515+
}
516+
if let Some(base_url) = &cli.base_url {
517+
std::env::set_var("OLLAMA_BASE_URL", base_url);
518+
}
519+
}
520+
497521
if let Some(prompt) = &cli.prompt {
498522
if cli.web {
499523
println!(
@@ -836,14 +860,17 @@ async fn main() -> Result<()> {
836860
}
837861
}
838862

839-
Commands::RunOnce {
840-
task,
841-
repo,
842-
provider,
843-
model,
844-
base_url,
845-
} => {
846-
run_once_command(task, repo, provider, model, base_url).await?;
863+
Commands::RunOnce { task, repo } => {
864+
run_once_command(
865+
task,
866+
repo,
867+
cli.provider
868+
.clone()
869+
.unwrap_or_else(|| "deterministic".to_string()),
870+
cli.model.clone(),
871+
cli.base_url.clone(),
872+
)
873+
.await?;
847874
}
848875

849876
Commands::Demo => {

tests/campaign_e2e.rs

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
//! Gated end-to-end test for the multi-persona swarm campaign.
2+
//!
3+
//! Proves the campaign actually runs: each persona executes as a real `korg
4+
//! worker` subprocess in a git worktree, completes its work, and sends a
5+
//! `TerminationReport`, so the leader records it as DONE — not the false
6+
//! `exit_code=-1` "crash" that stdout pollution used to cause for *every*
7+
//! worker. Guards the swarm-real fix (worker sends TerminationReport; all logs
8+
//! go to stderr so stdout stays a clean ACP channel).
9+
//!
10+
//! GATED: spawns the `korg` binary plus worker subprocesses in git worktrees —
11+
//! CI-hostile and slow (~60-90s). Run locally:
12+
//! cargo test --test campaign_e2e -- --ignored --nocapture
13+
14+
use std::process::Command;
15+
16+
fn git(dir: &std::path::Path, args: &[&str]) {
17+
let out = Command::new("git")
18+
.args(args)
19+
.current_dir(dir)
20+
.output()
21+
.expect("git spawn failed");
22+
assert!(
23+
out.status.success(),
24+
"git {:?} failed: {}",
25+
args,
26+
String::from_utf8_lossy(&out.stderr)
27+
);
28+
}
29+
30+
/// Last N lines of a log, for readable assertion failures.
31+
fn tail(s: &str) -> String {
32+
let lines: Vec<&str> = s.lines().collect();
33+
let start = lines.len().saturating_sub(40);
34+
lines[start..].join("\n")
35+
}
36+
37+
#[test]
38+
#[ignore = "spawns korg + worker subprocesses in git worktrees; CI-hostile/slow — run locally with --ignored"]
39+
fn campaign_workers_complete_and_attest_real_work() {
40+
// A fixture crate with the canonical add-bug (`a - b`). The deterministic
41+
// provider produces Benjamin's real fix for this task, applied + measured.
42+
let dir = std::env::temp_dir().join(format!("korg-campaign-e2e-{}", std::process::id()));
43+
let _ = std::fs::remove_dir_all(&dir);
44+
std::fs::create_dir_all(dir.join("src")).unwrap();
45+
std::fs::write(
46+
dir.join("Cargo.toml"),
47+
"[package]\nname = \"e2e\"\nversion = \"0.1.0\"\nedition = \"2021\"\n",
48+
)
49+
.unwrap();
50+
std::fs::write(
51+
dir.join("src/lib.rs"),
52+
"pub fn add(a: i64, b: i64) -> i64 { a - b }\n",
53+
)
54+
.unwrap();
55+
git(&dir, &["init", "-q"]);
56+
git(&dir, &["add", "-A"]);
57+
git(
58+
&dir,
59+
&[
60+
"-c",
61+
"user.email=t@t",
62+
"-c",
63+
"user.name=t",
64+
"commit",
65+
"-qm",
66+
"base",
67+
],
68+
);
69+
70+
let out = Command::new(env!("CARGO_BIN_EXE_korg"))
71+
.args([
72+
"Fix the add function in src/lib.rs so it adds",
73+
"--goal",
74+
"--provider",
75+
"deterministic",
76+
])
77+
.current_dir(&dir)
78+
.output()
79+
.expect("run korg campaign");
80+
let log = String::from_utf8_lossy(&out.stderr);
81+
82+
let _ = std::fs::remove_dir_all(&dir);
83+
84+
// The workers must terminate SUCCESS — before the fix, stdout pollution
85+
// corrupted the ACP stream and the leader stamped every worker crashed.
86+
assert!(
87+
log.contains("exit_status=success"),
88+
"expected workers to terminate success (TerminationReport received); got:\n{}",
89+
tail(&log)
90+
);
91+
assert!(
92+
!log.contains("worker_crashed"),
93+
"no worker should be falsely marked crashed; got:\n{}",
94+
tail(&log)
95+
);
96+
// Benjamin (the implementer) attests exactly one REAL measured mutation —
97+
// the applied fix on the fixture — proving real per-persona work flows
98+
// through the campaign, not theater.
99+
assert!(
100+
log.contains("persona=\"Benjamin\" mutations=1"),
101+
"Benjamin should attest one real measured mutation; got:\n{}",
102+
tail(&log)
103+
);
104+
}

0 commit comments

Comments
 (0)