codewhale-hq/Codewhale
1
0 Fork 0
mirror of https://github.com/codewhale-hq/Codewhale.git synced 2026-11-07 10:36:05 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

fix(engine): serialize worker admission and persist terminal receipts [skip ci]

Browse Source 
CW-6944 D02-01/D02-10: publish one revisioned format3 worker authority under a short OS transaction lock, preserve foreign records/claims and fence stale per-writer checkpoints. Keep exact legacy recovery bytes and a tombstone against the inspected old publisher.

Retain pending terminal results and live ownership until durable publication; retry the exact winner and report failed delivery explicitly. Fleet reservations retain previous-attempt uncertainty until exact teardown or an authoritative receipt. Final stream tails, usage and the terminal receipt share the existing atomic ledger record; failed sync retries do not duplicate accounting. Carry the same executor and manager lock through bounded owner work, and migrate all callers.

Update existing fixtures for the real transaction/provenance contract; no new repository test cases. Correct architecture and both changelogs. Read-only peer review checked the final source and ownership boundaries.

Verified at base b8c5417927 plus patch 116d80bad7a2bb58d68b6ecda05c5566f9673c14de970d1da623ff86db7f54a7; 2312 source inputs unchanged through builds and observations. Ordinary CLI build and Engine/TUI unit-binary build passed; 77 existing Engine + 4 TUI checks passed. 24 ordinary public-API native macOS process/fault/restart observations passed, plus 2 ordinary CLI RLM/handle_read synthetic-model journeys. Prior failed revisions remain in the local receipts. Existing blocking budget and changed-file formatting/diff checks passed.

The macOS linker retains its __eh_frame size warning. Native observations use synthetic state/fake children with no provider/network; SIGKILL is not power-loss proof. Legacy admitted attempts without actual settlement fail closed. Native Windows, installed packages, hosted final CI, providers and release qualification remain separate. CW-6945 retry/replay policy and the wider 0.10.3 plan stay open; no publication.
This commit is contained in:
CodeWhale Bot committed 2026-11-01 22:45:39 -08:00
1 parent b8c5417927
commit 382e648da1
21 files changed
+4079 -1245
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG.md
+2
View File
@@ -160,6 +160,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- Worker coordination serializes durable admissions across processes, preserves foreign workers and winning completion receipts, and fences stale checkpoints. Fleet restart reservations retain uncertain prior processes until exact teardown is confirmed; final output, usage and the receipt publish together without duplicate accounting on retry. Retained transcripts follow committed ownership.
- JSONPath projections bound path length, traversal steps and intermediate matches before cloning results. Direct fetches share the same parser and enforce one aggregate serialized-field budget; oversized selections refuse with narrower-selection guidance.
- Long tool output and diffs use bounded display windows and a byte-limited row cache. Retained details load verified byte ranges and label what search and copy include; explicit source copy keeps stored text with control-sequence cleanup. Oversized Unicode is shown safely without changing question answers; long answers retain their exact bytes in the existing owned output artifact before bounded model-context projection. New pager notices are translated across all 15 complete language packs.
crates/engine/src/core/engine.rs
+147 -58
View File
@@ -198,8 +198,7 @@ impl StructuredState {
working_set.summary_block(&workspace, &crate::tools::file::working_set_read_allowed);
let subagent_snapshots = if let Some(handle) = subagents {
let mut guard = handle.write().await;
guard.cleanup_for_session(active_session_id, Duration::from_secs(60 * 60));
let guard = handle.read().await;
guard
.list_for_session(active_session_id)
.into_iter()
@@ -4051,12 +4050,28 @@ impl Engine {
)
};
let event = if due {
let mut manager = self.subagent_manager.write().await;
manager.cleanup_for_session(
&active_session_id,
Duration::from_secs(60 * 60),
);
agent_list_event(&manager, &active_session_id)
let manager = self.subagent_manager.clone();
let session_id = active_session_id.clone();
match codewhale_runtime::runtime_control::owner_work(move || {
let mut manager = manager.blocking_write();
manager.cleanup_for_session(
&session_id,
Duration::from_secs(60 * 60),
)?;
Ok(agent_list_event(&manager, &session_id))
})
.await
{
Ok(event) => event,
Err(error) => {
let _ = self
.send_event(Event::error(ErrorEnvelope::transient(
format!("Failed to clean up sub-agents: {error}"),
)))
.await;
continue;
}
}
} else {
let manager = self.subagent_manager.read().await;
agent_list_event(&manager, &active_session_id)
@@ -4082,11 +4097,16 @@ impl Engine {
}
Op::CancelSubAgent { agent_id } => {
let active_session_id = self.session.id.clone();
let cancelled = self
.subagent_manager
.write()
.await
.cancel_agent_for_session(&active_session_id, &agent_id);
let manager = self.subagent_manager.clone();
let session_id = active_session_id.clone();
let target = agent_id.clone();
let cancelled =
codewhale_runtime::runtime_control::owner_work(move || {
manager
.blocking_write()
.cancel_agent_for_session(&session_id, &target)
})
.await;
let result = match cancelled {
Ok(snapshot) => {
let snapshot = crate::tools::subagent::settle_requested_child(
@@ -4159,18 +4179,27 @@ impl Engine {
let active_session_id = self.session.id.clone();
let runtime = self.off_turn_subagent_runtime();
let manager_handle = Arc::clone(&self.subagent_manager);
let (outcome, refresh) = {
let mut manager = self.subagent_manager.write().await;
let outcome = manager
.continue_child_from_user_for_session(
&active_session_id,
manager_handle,
runtime,
&agent_id,
&text,
)
.map_err(|err| err.to_string());
(outcome, agent_list_event(&manager, &active_session_id))
let manager = Arc::clone(&manager_handle);
let session = active_session_id.clone();
let target = agent_id.clone();
let result =
codewhale_runtime::runtime_control::owner_work(move || {
let mut manager = manager.blocking_write();
let outcome = manager
.continue_child_from_user_for_session(
&session,
manager_handle,
runtime,
&target,
&text,
)
.map_err(|err| err.to_string());
Ok((outcome, agent_list_event(&manager, &session)))
})
.await;
let (outcome, refresh) = match result {
Ok((outcome, refresh)) => (outcome, Some(refresh)),
Err(error) => (Err(error.to_string()), None),
};
let _ = self
.send_event(Event::SubAgentFollowUp {
@@ -4179,7 +4208,9 @@ impl Engine {
outcome,
})
.await;
if let Err(_e) = self.tx_event.try_send(refresh) {
if let Some(refresh) = refresh
&& self.tx_event.try_send(refresh).is_err()
{
tracing::debug!(
"Event channel full; dropping FollowUpSubAgent refresh"
);
@@ -4322,12 +4353,6 @@ impl Engine {
workspace,
mode,
} => {
// A fresh conversation arrives with its own
// sources: re-arm the guard and rebuild the notice
// from the assembly that installs its header.
self.instruction_guard_key = None;
self.pending_instruction_notices.clear();
self.session.instruction_assembly = None;
let plugin_workspace_changed =
self.plugin_registry.workspace() != workspace.as_path();
let previous_session_id = self.session.id.clone();
@@ -4338,7 +4363,8 @@ impl Engine {
} else {
previous_session_id.clone()
};
let closed_session_id = self.install_synced_session_id(next_session_id);
let closed_session_id = (next_session_id != previous_session_id)
.then_some(previous_session_id);
// SyncSession installs a conversation's identity; an id
// change IS a conversation boundary in this runtime —
// callers must pass their own conversation id for
@@ -4349,11 +4375,28 @@ impl Engine {
// the new conversation (#5372). Same-session reloads
// keep their id and are deliberately left untouched.
if let Some(closed_session_id) = closed_session_id {
let finalized = self
.subagent_manager
.write()
.await
.finalize_session_close_for_session(&closed_session_id);
let manager = self.subagent_manager.clone();
let session_id = closed_session_id.clone();
let finalized =
codewhale_runtime::runtime_control::owner_work(move || {
manager
.blocking_write()
.finalize_session_close_for_session(&session_id)
})
.await;
let finalized = match finalized {
Ok(finalized) => finalized,
Err(error) => {
let _ = self
.send_event(Event::error(ErrorEnvelope::transient(
format!(
"Failed to close sub-agent session {closed_session_id}; session was not switched: {error}"
),
)))
.await;
continue;
}
};
crate::tools::subagent::settle_requested_children_for_session(
&self.subagent_manager,
&closed_session_id,
@@ -4368,6 +4411,13 @@ impl Engine {
);
}
}
// A fresh conversation arrives with its own
// sources: re-arm the guard and rebuild the notice
// from the assembly that installs its header.
self.instruction_guard_key = None;
self.pending_instruction_notices.clear();
self.session.instruction_assembly = None;
let _ = self.install_synced_session_id(next_session_id);
self.restore_session_history(
messages,
system_prompt,
@@ -4705,19 +4755,41 @@ impl Engine {
// #freeze: flush any sub-agent checkpoint that the hot-path debounce
// coalesced away, so a graceful shutdown keeps the latest progress.
{
let mut manager = self.subagent_manager.write().await;
let children = manager.list_for_session(&self.session.id);
let manager = self.subagent_manager.clone();
let session_id = self.session.id.clone();
let owner_agent_id = self
.child_host
.as_ref()
.map(|owner| owner.authority.owner_agent_id.clone());
let shutdown = codewhale_runtime::runtime_control::owner_work(move || {
let mut manager = manager.blocking_write();
let children = manager.list_for_session(&session_id);
let mut failure = None;
for child in children {
let owned = self.child_host.as_ref().is_none_or(|owner| {
child.parent_run_id.as_deref()
== Some(owner.authority.owner_agent_id.as_str())
});
if owned && child.status == SubAgentStatus::Running {
let _ = manager.cancel_agent_for_session(&self.session.id, &child.agent_id);
let owned = owner_agent_id
.as_ref()
.is_none_or(|owner| child.parent_run_id.as_deref() == Some(owner.as_str()));
if owned
&& child.status == SubAgentStatus::Running
&& let Err(error) =
manager.cancel_agent_for_session(&session_id, &child.agent_id)
{
failure.get_or_insert(error);
}
}
manager.flush_pending_persist();
if let Err(error) = manager.flush_pending_persist() {
failure.get_or_insert(error);
}
failure.map_or(Ok(()), Err)
})
.await;
if let Err(error) = shutdown {
let _ = self
.send_event(Event::error(ErrorEnvelope::transient(format!(
"Sub-agent shutdown is uncertain: {error}"
))))
.await;
child_result = Some(Err(error));
}
crate::tools::subagent::settle_requested_children_for_session(
&self.subagent_manager,
@@ -4727,7 +4799,19 @@ impl Engine {
.map(|child| child.authority.owner_agent_id.as_str()),
)
.await;
self.subagent_manager.write().await.flush_pending_persist();
let manager = self.subagent_manager.clone();
if let Err(error) = codewhale_runtime::runtime_control::owner_work(move || {
manager.blocking_write().flush_pending_persist()
})
.await
{
let _ = self
.send_event(Event::error(ErrorEnvelope::transient(format!(
"Sub-agent shutdown checkpoint is uncertain: {error}"
))))
.await;
child_result = Some(Err(error));
}
// #420: graceful MCP shutdown — send SIGTERM and give stdio servers
// a brief window to exit before drop fires SIGKILL via kill_on_drop.
@@ -10393,27 +10477,32 @@ fn apply_patch_permission_paths(input: &Value) -> Vec<String> {
}
/// Spawn the engine in a background task
pub fn spawn_engine(
pub async fn spawn_engine(
config: EngineConfig,
api_config: &Config,
mcp_handoff: Option<crate::mcp::McpHandoff>,
) -> EngineHandle {
let (engine, handle) = Engine::new(config, api_config);
engine
.mcp_pool_slot
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.handoff = mcp_handoff;
let api_config = api_config.clone();
let (engine, handle) = codewhale_runtime::runtime_control::owner_work(move || {
let (engine, handle) = Engine::new(config, &api_config);
engine
.mcp_pool_slot
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.handoff = mcp_handoff;
Ok((Box::new(engine), handle))
})
.await
.expect("Engine initialization owner failed");
// Box the run future before supervision. An extra async wrapper embeds
// the large engine state again in both its own and the supervisor's poll
// frames, which can overflow an ordinary worker-thread stack.
spawn_supervised(
"engine-event-loop",
std::panic::Location::caller(),
Box::pin(engine.run()),
Box::pin((*engine).run()),
);
handle
}
crates/engine/src/fleet/executor.rs
+104 -14
View File
@@ -22,7 +22,9 @@
use anyhow::Result;
use codewhale_config::FleetExecConfig;
use codewhale_protocol::fleet::{FleetHostSpec, FleetTaskSpec, FleetWorkerEventPayload};
use codewhale_protocol::fleet::{
FleetHostSpec, FleetReceipt, FleetTaskSpec, FleetWorkerEventPayload,
};
use super::host::{FleetHostAdapter, FleetWorkerCommand};
use super::profile::AgentProfile;
@@ -595,6 +597,8 @@ struct WorkerStream {
// corrupts valid UTF-8 when a multibyte code point crosses a read boundary.
pending: Vec<u8>,
terminal: bool,
confirmed_stopped: bool,
pending_terminal: Option<PendingFleetTerminal>,
terminal_route: TerminalRouteEvidence,
/// When this worker process was started, for per-task wall-clock limits (R5).
started_at: std::time::Instant,
@@ -720,20 +724,25 @@ pub struct FleetWorkerTerminalEvent {
pub requires_reported_route: bool,
}
#[derive(Debug, Clone)]
pub(crate) struct FleetPreparedTaskOutcome {
pub payload: FleetWorkerEventPayload,
pub tail_payloads: Vec<FleetWorkerEventPayload>,
pub final_status: Option<super::ledger::FleetTaskLedgerStatus>,
pub receipt: FleetReceipt,
}
#[derive(Debug, Clone)]
enum PendingFleetTerminal {
Observed(FleetWorkerTerminalEvent),
Prepared(FleetPreparedTaskOutcome),
}
impl FleetExecutor {
pub async fn refresh_tailnet_inventory(&mut self) {
let snapshot = tokio::task::spawn_blocking(|| {
super::tailnet::TailnetSnapshot::new(
super::tailnet::inventory().map_err(|error| error.to_string()),
)
})
.await
.unwrap_or_else(|error| {
super::tailnet::TailnetSnapshot::new(Err(format!(
"Tailnet inventory task failed: {error}"
)))
});
self.tailnet_snapshot = Some(std::sync::Arc::new(snapshot));
pub fn refresh_tailnet_inventory(&mut self) {
self.tailnet_snapshot = Some(std::sync::Arc::new(super::tailnet::TailnetSnapshot::new(
super::tailnet::inventory().map_err(|error| error.to_string()),
)));
}
pub fn new(workspace: impl AsRef<std::path::Path>) -> Self {
@@ -855,6 +864,8 @@ impl FleetExecutor {
offset: 0,
pending: Vec::new(),
terminal: false,
confirmed_stopped: false,
pending_terminal: None,
terminal_route: TerminalRouteEvidence::default(),
started_at: std::time::Instant::now(),
final_answer: None,
@@ -879,6 +890,79 @@ impl FleetExecutor {
.and_then(|stream| stream.attempt.clone())
}
pub(crate) fn confirmed_stopped_attempts(
&self,
) -> std::collections::BTreeMap<String, FleetExecutorAttempt> {
self.streams
.iter()
.filter_map(|(id, stream)| {
stream
.confirmed_stopped
.then(|| stream.attempt.clone())
.flatten()
.map(|attempt| (id.clone(), attempt))
})
.collect()
}
pub(crate) fn is_confirmed_stopped(&self, worker_id: &str) -> bool {
self.streams
.get(worker_id)
.is_some_and(|stream| stream.confirmed_stopped)
}
pub(crate) fn prepared_terminal_outcome(
&self,
worker_id: &str,
) -> Option<FleetPreparedTaskOutcome> {
match self.streams.get(worker_id)?.pending_terminal.as_ref()? {
PendingFleetTerminal::Prepared(outcome) => Some(outcome.clone()),
PendingFleetTerminal::Observed(_) => None,
}
}
pub(crate) fn retain_prepared_terminal_outcome(
&mut self,
worker_id: &str,
outcome: FleetPreparedTaskOutcome,
) -> Result<()> {
let stream = self
.streams
.get_mut(worker_id)
.ok_or_else(|| anyhow::anyhow!("Fleet terminal stream is absent"))?;
anyhow::ensure!(
stream.confirmed_stopped,
"Fleet terminal process is not confirmed stopped"
);
if let Some(attempt) = &stream.attempt {
anyhow::ensure!(
outcome.receipt.run_id == attempt.run_id
&& outcome.receipt.task_id == attempt.task_id
&& outcome.receipt.attempt == Some(attempt.attempt)
&& outcome.receipt.worker_id == worker_id,
"Fleet terminal candidate does not match its tracked attempt"
);
}
stream.pending_terminal = Some(PendingFleetTerminal::Prepared(outcome));
Ok(())
}
pub(crate) fn poll_terminal_for_commit(
&mut self,
worker_id: &str,
) -> Option<FleetWorkerTerminalEvent> {
if let Some(pending) = self.streams.get(worker_id)?.pending_terminal.as_ref() {
return match pending {
PendingFleetTerminal::Observed(event) => Some(event.clone()),
PendingFleetTerminal::Prepared(_) => None,
};
}
let event = self.poll_terminal_with_status(worker_id)?;
self.streams.get_mut(worker_id)?.pending_terminal =
Some(PendingFleetTerminal::Observed(event.clone()));
Some(event)
}
/// Wall-clock time this worker process has been running (R5). `None` when
/// the worker is not tracked or its exit was already observed: a process
/// that has exited is not running, so a deadline must not rewrite the
@@ -912,6 +996,9 @@ impl FleetExecutor {
} else {
self.adapter.stop_worker(worker_id)?;
}
if let Some(stream) = self.streams.get_mut(worker_id) {
stream.confirmed_stopped = true;
}
Ok(())
}
@@ -1027,6 +1114,7 @@ impl FleetExecutor {
tail_payloads.push(payload);
}
stream.terminal = true;
stream.confirmed_stopped = true;
let final_answer = stream.final_answer.take();
// Surface the visible final answer on a successful completion so
// report/summary tasks (no scorer, no file artifact) show their
@@ -1198,6 +1286,8 @@ mod tests {
offset: 0,
pending: Vec::new(),
terminal: false,
confirmed_stopped: false,
pending_terminal: None,
terminal_route: TerminalRouteEvidence::default(),
started_at: std::time::Instant::now(),
final_answer: None,
crates/engine/src/fleet/ledger.rs
+179 -17
View File
@@ -153,6 +153,8 @@ pub enum FleetLedgerRecord {
/// publishing a terminal task without the receipt that explains it.
TaskAttemptFinalized {
event: FleetWorkerEvent,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
tail_events: Vec<FleetWorkerEvent>,
#[serde(default, skip_serializing_if = "Option::is_none")]
final_status: Option<FleetTaskLedgerStatus>,
receipt: Box<FleetReceipt>,
@@ -340,6 +342,7 @@ pub struct FleetLedger {
ledger_file: WorkspaceFile,
lock_file: WorkspaceFile,
original_lock: std::fs::File,
durable_barrier: parking_lot::Mutex<Option<(std::fs::File, u64)>>,
/// Stable advisory-lock inode shared by every manager for this workspace.
///
/// The ledger itself is replaced during compaction, so locking
@@ -369,6 +372,7 @@ impl FleetLedger {
ledger_file,
lock_file,
original_lock,
durable_barrier: parking_lot::Mutex::new(None),
ledger_path,
lock_path,
#[cfg(any(test, feature = "test-support"))]
@@ -444,6 +448,7 @@ impl FleetLedger {
.ledger_file
.open_update(true, true)
.with_context(|| format!("opening fleet ledger {}", self.ledger_path.display()))?;
self.durable_barrier.lock().take();
// A process can die after writing only part of its final JSON record.
// O_APPEND would otherwise concatenate the next valid record directly
// onto that unterminated tail, causing replay to discard both. Preserve
@@ -480,6 +485,8 @@ impl FleetLedger {
.with_context(|| format!("flushing fleet ledger {}", self.ledger_path.display()))?;
file.sync_data()
.with_context(|| format!("syncing fleet ledger {}", self.ledger_path.display()))?;
let length = file.metadata()?.len();
*self.durable_barrier.lock() = Some((file, length));
// Wake SSE subscribers after the bytes are durable (#6211 R7b). Every
// record kind funnels through here, so a wake only promises "something
// changed" — subscribers re-poll with their cursor.
@@ -954,16 +961,40 @@ impl FleetLedger {
timestamp: &str,
payload: FleetWorkerEventPayload,
final_status: Option<FleetTaskLedgerStatus>,
receipt: FleetReceipt,
) -> Result<Option<FleetWorkerEvent>> {
self.finalize_task_attempt_with_tail_if_leased(
run_id,
worker_id,
task_id,
expected_attempts,
timestamp,
payload,
Vec::new(),
final_status,
receipt,
)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn finalize_task_attempt_with_tail_if_leased(
&self,
run_id: &FleetRunId,
worker_id: &str,
task_id: &str,
expected_attempts: u32,
timestamp: &str,
payload: FleetWorkerEventPayload,
tail_payloads: Vec<FleetWorkerEventPayload>,
final_status: Option<FleetTaskLedgerStatus>,
mut receipt: FleetReceipt,
) -> Result<Option<FleetWorkerEvent>> {
if !matches!(
&payload,
FleetWorkerEventPayload::Completed { .. }
| FleetWorkerEventPayload::Failed { .. }
| FleetWorkerEventPayload::Cancelled { .. }
) {
bail!("attempt finalization requires a terminal worker event");
}
let payload_status = match &payload {
FleetWorkerEventPayload::Completed { .. } => FleetTaskLedgerStatus::Completed,
FleetWorkerEventPayload::Failed { .. } => FleetTaskLedgerStatus::Failed,
FleetWorkerEventPayload::Cancelled { .. } => FleetTaskLedgerStatus::Cancelled,
_ => bail!("attempt finalization requires a terminal worker event"),
};
if final_status.is_some_and(|status| {
!matches!(
status,
@@ -984,28 +1015,126 @@ impl FleetLedger {
{
bail!("attempt receipt generation does not match its lease");
}
self.with_write_lock(|| {
for tail in &tail_payloads {
if matches!(
tail,
FleetWorkerEventPayload::Completed { .. }
| FleetWorkerEventPayload::Failed { .. }
| FleetWorkerEventPayload::Cancelled { .. }
) {
bail!("terminal tail contains another terminal outcome");
}
}
let had_usage = tail_payloads
.iter()
.any(|payload| matches!(payload, FleetWorkerEventPayload::UsageReport { .. }));
let tail_digest = (!tail_payloads.is_empty())
.then(|| serde_json::to_vec(&tail_payloads))
.transpose()?
.map(|bytes| codewhale_runtime::hashing::sha256_hex(&bytes));
let appended = self.with_write_lock(|| {
let state = self.rebuild_state_unlocked()?;
let key = task_key(&run_id.0, task_id);
let Some(task) = state.tasks.get(&key) else {
return Ok(None);
};
if task.status != FleetTaskLedgerStatus::Leased
|| task.leased_to.as_deref() != Some(worker_id)
if task.leased_to.as_deref() != Some(worker_id)
|| task.entry.attempts != expected_attempts
{
return Ok(None);
}
let event = next_worker_event(&state, run_id, worker_id, task_id, timestamp, payload);
let lifecycle_key = event_key(worker_id, &run_id.0, task_id);
if task.status != FleetTaskLedgerStatus::Leased {
let Some(existing_receipt) = state.receipts.get(&key) else {
return Ok(None);
};
let Some(existing_event) = state.latest_events.get(&lifecycle_key) else {
return Ok(None);
};
receipt.attempt = Some(expected_attempts);
receipt.terminal_seq = Some(existing_event.seq);
let mut expected_event = next_worker_event(
&state,
run_id,
worker_id,
task_id,
timestamp,
payload.clone(),
);
expected_event.seq = existing_event.seq;
if let Some(digest) = &tail_digest {
expected_event
.extra
.insert("terminal_tail_sha256".into(), Value::String(digest.clone()));
}
if task.status != final_status.unwrap_or(payload_status)
|| serde_json::to_value(existing_receipt)? != serde_json::to_value(&receipt)?
|| serde_json::to_value(existing_event)?
!= serde_json::to_value(&expected_event)?
{
return Ok(None);
}
self.sync_durable_unlocked()
.context("retrying exact Fleet terminal durability")?;
return Ok(Some(existing_event.clone()));
}
let mut sequence = state.latest_seq.get(&lifecycle_key).copied().unwrap_or(0);
let mut tail_events = Vec::with_capacity(tail_payloads.len());
for tail in tail_payloads {
let mut event =
next_worker_event(&state, run_id, worker_id, task_id, timestamp, tail);
sequence = sequence
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("Fleet event sequence exhausted"))?;
event.seq = sequence;
tail_events.push(event);
}
let mut event =
next_worker_event(&state, run_id, worker_id, task_id, timestamp, payload);
event.seq = sequence
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("Fleet event sequence exhausted"))?;
if let Some(digest) = tail_digest {
event
.extra
.insert("terminal_tail_sha256".into(), Value::String(digest));
}
receipt.attempt = Some(expected_attempts);
receipt.terminal_seq = Some(event.seq);
self.append_record_unlocked(&FleetLedgerRecord::TaskAttemptFinalized {
event: event.clone(),
tail_events,
final_status,
receipt: Box::new(receipt),
})?;
Ok(Some(event))
})
});
if matches!(&appended, Ok(Some(_))) && had_usage {
self.record_budget_exceeded_alert_once(run_id, timestamp)?;
}
appended
}
pub(crate) fn ensure_durable(&self) -> Result<()> {
self.with_write_lock(|| self.sync_durable_unlocked())
}
fn sync_durable_unlocked(&self) -> Result<()> {
let file = self.ledger_file.open_update(false, false)?;
let length = file.metadata()?.len();
let mut cached = self.durable_barrier.lock();
if let Some((previous, durable_length)) = cached.take()
&& durable_length == length
&& same_file(&previous, &file)?
{
*cached = Some((previous, durable_length));
return Ok(());
}
file.sync_data()
.context("acknowledging Fleet ledger durability")?;
let length = file.metadata()?.len();
*cached = Some((file, length));
Ok(())
}
/// Append a scheduler-owned terminal event only if the stale lease version
@@ -1770,6 +1899,7 @@ impl FleetLedger {
if !contents.is_empty() {
contents.push('\n');
}
self.durable_barrier.lock().take();
self.ledger_file
.replace(contents.as_bytes())
.with_context(|| {
@@ -1871,10 +2001,38 @@ fn runtime_events_from_record(
worker_seq: None,
payload: json!({ "status": status }),
}],
FleetLedgerRecord::TaskAttemptFinalized { event, receipt, .. } => vec![
runtime_worker_event("worker", ledger_ordinal, replay_epoch, event),
runtime_receipt_event("receipt", ledger_ordinal, replay_epoch, receipt),
],
FleetLedgerRecord::TaskAttemptFinalized {
event,
tail_events,
receipt,
..
} => {
let mut events = tail_events
.iter()
.enumerate()
.map(|(index, event)| {
runtime_worker_event(
&format!("tail-{index}"),
ledger_ordinal,
replay_epoch,
event,
)
})
.collect::<Vec<_>>();
events.push(runtime_worker_event(
"worker",
ledger_ordinal,
replay_epoch,
event,
));
events.push(runtime_receipt_event(
"receipt",
ledger_ordinal,
replay_epoch,
receipt,
));
events
}
FleetLedgerRecord::EventAppended { event } => {
vec![runtime_worker_event(
"worker",
@@ -2287,9 +2445,13 @@ fn apply_record(state: &mut FleetLedgerState, record: FleetLedgerRecord) {
}
FleetLedgerRecord::TaskAttemptFinalized {
event,
tail_events,
final_status,
receipt,
} => {
for tail in tail_events {
apply_record(state, FleetLedgerRecord::EventAppended { event: tail });
}
let run_id = event.run_id.clone();
let task_id = event.task_id.clone();
let worker_id = event.worker_id.clone();
crates/engine/src/fleet/manager.rs
+541 -319
View File
File diff suppressed because it is too large. Load diff
crates/engine/src/tools/subagent/completion_usage_tests.rs
+15 -3
View File
@@ -94,7 +94,11 @@ async fn completion_usage_live_terminal_counts_grandchildren_and_continuations_o
result.status = status;
// The locked ledger wins over an earlier result snapshot.
result.usage.as_mut().unwrap().total_tokens = Some(123_456);
assert!(manager.finish_terminal_result(&root, result, false, false));
assert!(
manager
.finish_terminal_result(&root, result, false, false)
.expect("durable worker operation")
);
let completion = rx.try_recv().unwrap();
let payload = sentinel(&completion);
assert_eq!(payload["usage"]["own"]["total_tokens"], 10);
@@ -138,7 +142,11 @@ async fn completion_usage_recovery_restores_measured_lineage_without_recounting(
let mut manager = SubAgentManager::new(base.clone(), 8).with_state_path(path.clone());
let (root, _, grandchild, continued) = family(&mut manager);
let result = terminal_result(&manager, &root);
assert!(manager.finish_terminal_result(&root, result, false, false));
assert!(
manager
.finish_terminal_result(&root, result, false, false)
.expect("durable worker operation")
);
let before = receipt(&manager, &root);
manager.persist_state_synchronously().unwrap();
let mut loaded = SubAgentManager::new(base, 8).with_state_path(path);
@@ -340,7 +348,11 @@ async fn completion_usage_counts_real_manifest_only_root_fork() {
.unwrap();
let source = child(&mut guard, "source", None, Some(1));
let completed = terminal_result(&guard, &source);
assert!(guard.finish_terminal_result(&source, completed, true, false));
assert!(
guard
.finish_terminal_result(&source, completed, true, false)
.expect("durable worker operation")
);
let fork = guard
.spawn_background_with_assignment_options(
Arc::clone(&manager),
crates/engine/src/tools/subagent/coord.rs
+247 -220
View File
@@ -122,25 +122,27 @@ impl ToolSpec for AgentsListTool {
.unwrap_or(false);
let agent_ref = parse_agent_ref(&input)?;
let mut manager = self.manager.write().await;
manager.cleanup_for_session(&context.state_namespace, COMPLETED_AGENT_RETENTION);
let summaries = if let Some(agent_ref) = agent_ref {
let summary = manager
.coordination_summary_for_session(
&context.state_namespace,
&agent_ref,
let manager = self.manager.clone();
let session = context.state_namespace.clone();
let summaries = codewhale_runtime::runtime_control::owner_work(move || {
let mut manager = manager.blocking_write();
manager.cleanup_for_session(&session, COMPLETED_AGENT_RETENTION)?;
let summaries = if let Some(agent_ref) = agent_ref {
manager
.coordination_summary_for_session(&session, &agent_ref, RECENT_PROGRESS_LIMIT)
.map(|summary| vec![summary])
.map_err(|err| ToolError::invalid_input(err.to_string()))
} else {
Ok(manager.list_coordination_summaries_for_session(
&session,
include_archived,
RECENT_PROGRESS_LIMIT,
)
.map_err(|err| ToolError::invalid_input(err.to_string()))?;
vec![summary]
} else {
manager.list_coordination_summaries_for_session(
&context.state_namespace,
include_archived,
RECENT_PROGRESS_LIMIT,
)
};
drop(manager);
))
};
Ok(summaries)
})
.await
.map_err(|error| ToolError::execution_failed(error.to_string()))??;
let payload = json!({
"action": "list",
@@ -316,66 +318,80 @@ impl AgentsFollowupTool {
message: &str,
context: &ToolContext,
) -> Result<Value, ToolError> {
let mut manager = self.manager.write().await;
let (source, target) = manager
.continuation_target_for_caller(
&context.state_namespace,
agent_ref,
self.caller_agent_id.as_deref(),
"agents/followup",
)
.map_err(|error| ToolError::invalid_input(error.to_string()))?;
let snapshot = manager
.get_result(&target)
.map_err(|error| ToolError::invalid_input(error.to_string()))?;
let resumed_already = source != target;
let receipt = if super::subagent_checkpoint_is_continuable(&snapshot)
&& self.runtime.is_some()
{
let snapshot = manager
.resume_from_checkpoint_for_session(
&context.state_namespace,
Arc::clone(&self.manager),
self.runtime.clone().expect("runtime checked"),
&target,
message,
)
.map_err(|error| ToolError::execution_failed(error.to_string()))?;
ParentMailReceipt {
agent_id: snapshot.agent_id.clone(),
status: subagent_status_name(&snapshot.status).to_string(),
queue_depth: 0,
woke: true,
continued_from_checkpoint: true,
continuation_handle: None,
note: format!(
"resumed from checkpoint {source} as {}; original receipt retained",
snapshot.agent_id
),
}
} else if resumed_already && snapshot.status != SubAgentStatus::Running {
ParentMailReceipt {
agent_id: target, status: subagent_status_name(&snapshot.status).to_string(),
queue_depth: 0, woke: false, continued_from_checkpoint: true,
continuation_handle: None,
note: "Existing continuation has settled; no duplicate worker was started and no message was delivered.".to_string(),
}
} else {
manager
.followup_child_for_session(&context.state_namespace, &target, message.to_string())
.map_err(|error| ToolError::invalid_input(error.to_string()))?
};
let child_route = manager
.get_worker_record_for_session(&context.state_namespace, &receipt.agent_id)
.and_then(|record| record.spec.child_route);
Ok(json!({
"action": "followup", "from": source, "to": receipt.agent_id,
"agent_id": receipt.agent_id, "queued": receipt.woke || receipt.queue_depth > 0,
"woke": receipt.woke, "queue_depth": receipt.queue_depth, "status": receipt.status,
"continued_from_checkpoint": receipt.continued_from_checkpoint || resumed_already,
"continuation_handle": receipt.continuation_handle, "note": receipt.note,
"child_route": child_route,
}))
let manager_handle = self.manager.clone();
let manager = manager_handle.clone();
let session = context.state_namespace.clone();
let caller = self.caller_agent_id.clone();
let runtime = self.runtime.clone();
let agent_ref = agent_ref.to_owned();
let message = message.to_owned();
codewhale_runtime::runtime_control::owner_work(move || {
let mut manager = manager.blocking_write();
let result = (|| {
let (source, target) = manager
.continuation_target_for_caller(
&session,
&agent_ref,
caller.as_deref(),
"agents/followup",
)
.map_err(|error| ToolError::invalid_input(error.to_string()))?;
let snapshot = manager
.get_result(&target)
.map_err(|error| ToolError::invalid_input(error.to_string()))?;
let resumed_already = source != target;
let receipt = if super::subagent_checkpoint_is_continuable(&snapshot)
&& runtime.is_some()
{
let snapshot = manager
.resume_from_checkpoint_for_session(
&session,
Arc::clone(&manager_handle),
runtime.expect("runtime checked"),
&target,
&message,
)
.map_err(|error| ToolError::execution_failed(error.to_string()))?;
ParentMailReceipt {
agent_id: snapshot.agent_id.clone(),
status: subagent_status_name(&snapshot.status).to_string(),
queue_depth: 0,
woke: true,
continued_from_checkpoint: true,
continuation_handle: None,
note: format!(
"resumed from checkpoint {source} as {}; original receipt retained",
snapshot.agent_id
),
}
} else if resumed_already && snapshot.status != SubAgentStatus::Running {
ParentMailReceipt {
agent_id: target, status: subagent_status_name(&snapshot.status).to_string(),
queue_depth: 0, woke: false, continued_from_checkpoint: true,
continuation_handle: None,
note: "Existing continuation has settled; no duplicate worker was started and no message was delivered.".to_string(),
}
} else {
manager
.followup_child_for_session(&session, &target, message)
.map_err(|error| ToolError::invalid_input(error.to_string()))?
};
let child_route = manager
.get_worker_record_for_session(&session, &receipt.agent_id)
.and_then(|record| record.spec.child_route);
Ok(json!({
"action": "followup", "from": source, "to": receipt.agent_id,
"agent_id": receipt.agent_id, "queued": receipt.woke || receipt.queue_depth > 0,
"woke": receipt.woke, "queue_depth": receipt.queue_depth, "status": receipt.status,
"continued_from_checkpoint": receipt.continued_from_checkpoint || resumed_already,
"continuation_handle": receipt.continuation_handle, "note": receipt.note,
"child_route": child_route,
}))
})();
Ok(result)
})
.await
.map_err(|error| ToolError::execution_failed(error.to_string()))?
}
}
@@ -602,17 +618,19 @@ impl ToolSpec for AgentsInterruptTool {
.unwrap_or("interrupted by parent via agents/interrupt")
.to_string();
let (prior, snapshot) = {
let mut manager = self.manager.write().await;
manager
.interrupt_child_for_session(
&context.state_namespace,
&agent_ref,
self.caller_agent_id.as_deref(),
reason,
)
.map_err(|err| ToolError::invalid_input(err.to_string()))?
};
let manager = self.manager.clone();
let session = context.state_namespace.clone();
let caller = self.caller_agent_id.clone();
let (prior, snapshot) = codewhale_runtime::runtime_control::owner_work(move || {
manager.blocking_write().interrupt_child_for_session(
&session,
&agent_ref,
caller.as_deref(),
reason,
)
})
.await
.map_err(|error| ToolError::execution_failed(error.to_string()))?;
let snapshot = super::settle_requested_child(&self.manager, snapshot).await;
let worker_record = {
@@ -1445,30 +1463,22 @@ impl ToolSpec for AgentsCoordinateTool {
)));
}
let mut manager = self.manager.clone().write_owned().await;
if matches!(action, "claim" | "propose") {
let caller = self.caller.clone();
let session = context.state_namespace.clone();
let workspace = context.workspace.clone();
return codewhale_runtime::runtime_control::owner_work(move || {
Ok(Self::mutate_coordination(
caller.as_deref(),
&session,
&workspace,
&mut manager,
input,
))
let manager = self.manager.clone();
let caller = self.caller.clone();
let session = context.state_namespace.clone();
let workspace = context.workspace.clone();
codewhale_runtime::runtime_control::owner_work(move || {
let mut manager = manager.blocking_write();
manager.with_store_transaction(|manager| {
Self::mutate_coordination(caller.as_deref(), &session, &workspace, manager, input)
.map_err(anyhow::Error::new)
})
.await
.map_err(|error| ToolError::execution_failed(error.to_string()))?;
}
Self::mutate_coordination(
self.caller.as_deref(),
&context.state_namespace,
&context.workspace,
&mut manager,
input,
)
})
.await
.map_err(|error| match error.downcast::<ToolError>() {
Ok(error) => error,
Err(error) => ToolError::execution_failed(error.to_string()),
})
}
}
@@ -1710,7 +1720,10 @@ mod tests {
.await
.expect_err("mutation must fail when its receipt cannot persist")
.to_string();
assert!(error.contains("failed to persist"), "{error}");
assert!(
error.contains("Runtime coordination was opened without its writer lock"),
"{error}"
);
assert!(
blocked_manager
.read()
@@ -1825,31 +1838,35 @@ mod tests {
));
let (claimant, owner) = {
let mut guard = manager.write().await;
let claimant = guard.insert_test_running_agent("claimant", tmp.path());
let owner = guard.insert_test_running_agent("owner", tmp.path());
let active = [claimant.clone(), owner.clone()]
.into_iter()
.collect::<BTreeSet<_>>();
for claim in [
WriteScopeClaim {
owner: claimant.clone(),
roots: vec!["src/claimant".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
WriteScopeClaim {
owner: owner.clone(),
roots: vec!["src/shared".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
] {
guard
.coordination
.register_claim(claim, false, |candidate| active.contains(candidate))
.expect("initial non-overlapping claim");
}
(claimant, owner)
guard
.with_store_transaction(|guard| {
let claimant = guard.insert_test_running_agent("claimant", tmp.path());
let owner = guard.insert_test_running_agent("owner", tmp.path());
let active = [claimant.clone(), owner.clone()]
.into_iter()
.collect::<BTreeSet<_>>();
for claim in [
WriteScopeClaim {
owner: claimant.clone(),
roots: vec!["src/claimant".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
WriteScopeClaim {
owner: owner.clone(),
roots: vec!["src/shared".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
] {
guard
.coordination
.register_claim(claim, false, |candidate| active.contains(candidate))
.expect("initial non-overlapping claim");
}
Ok((claimant, owner))
})
.unwrap()
};
let error = AgentsCoordinateTool::new(Arc::clone(&manager), Some(claimant.clone()))
@@ -1886,31 +1903,35 @@ mod tests {
));
let claimant = {
let mut guard = manager.write().await;
let claimant = guard.insert_test_running_agent("claimant", tmp.path());
let owner = guard.insert_test_running_agent("owner", tmp.path());
let active = [claimant.clone(), owner.clone()]
.into_iter()
.collect::<BTreeSet<_>>();
for claim in [
WriteScopeClaim {
owner: claimant.clone(),
roots: vec!["src/claimant".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
WriteScopeClaim {
owner: owner.clone(),
roots: vec!["src/shared".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
] {
guard
.coordination
.register_claim(claim, false, |candidate| active.contains(candidate))
.expect("initial non-overlapping claim");
}
claimant
guard
.with_store_transaction(|guard| {
let claimant = guard.insert_test_running_agent("claimant", tmp.path());
let owner = guard.insert_test_running_agent("owner", tmp.path());
let active = [claimant.clone(), owner.clone()]
.into_iter()
.collect::<BTreeSet<_>>();
for claim in [
WriteScopeClaim {
owner: claimant.clone(),
roots: vec!["src/claimant".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
WriteScopeClaim {
owner: owner.clone(),
roots: vec!["src/shared".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
] {
guard
.coordination
.register_claim(claim, false, |candidate| active.contains(candidate))
.expect("initial non-overlapping claim");
}
Ok(claimant)
})
.unwrap()
};
AgentsCoordinateTool::new(Arc::clone(&manager), Some(claimant.clone()))
@@ -1923,23 +1944,27 @@ mod tests {
let resolution_sequence = {
let mut guard = manager.write().await;
let record = guard
.coordination
.register_claim(
WriteScopeClaim {
owner: claimant.clone(),
roots: vec!["src/isolated".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
true,
|_| true,
)
.expect("later isolated claim resolves contention");
guard
.persist_state_synchronously()
.expect("persist resolved contention");
record.sequence
.with_store_transaction(|guard| {
let record = guard
.coordination
.register_claim(
WriteScopeClaim {
owner: claimant.clone(),
roots: vec!["src/isolated".into()],
exact_files: Vec::new(),
contracts: Vec::new(),
},
true,
|_| true,
)
.expect("later isolated claim resolves contention");
guard
.persist_state_synchronously()
.expect("persist resolved contention");
Ok(record.sequence)
})
.unwrap()
};
let mut replayed = super::super::SubAgentManager::new(tmp.path().to_path_buf(), 4)
@@ -1955,48 +1980,50 @@ mod tests {
Some(resolution_sequence)
);
let slots = COORDINATION_RECORD_LIMIT - replayed.coordination.write_claims.len();
for index in 0..slots {
replayed
.coordination
.register_claim(
WriteScopeClaim {
owner: format!("inactive-fill-{index:03}"),
roots: vec![format!("pkg/fill-{index:03}")],
exact_files: Vec::new(),
contracts: Vec::new(),
},
true,
|_| false,
)
.expect("fill inactive claim capacity");
}
for index in 0..2 {
replayed
.coordination
.register_claim(
WriteScopeClaim {
owner: format!("inactive-overflow-{index}"),
roots: vec![format!("pkg/overflow-{index}")],
exact_files: Vec::new(),
contracts: Vec::new(),
},
true,
|_| false,
)
.expect("evict oldest inactive claim at capacity");
}
assert!(
!replayed
.coordination
.write_claims
.iter()
.any(|claim| claim.claim.owner == claimant),
"the resolving claimant claim must be evicted for the durability regression"
);
replayed
.persist_state_synchronously()
.expect("persist after inactive claim eviction");
.with_store_transaction(|held| {
let slots = COORDINATION_RECORD_LIMIT - held.coordination.write_claims.len();
for index in 0..slots {
held.coordination
.register_claim(
WriteScopeClaim {
owner: format!("inactive-fill-{index:03}"),
roots: vec![format!("pkg/fill-{index:03}")],
exact_files: Vec::new(),
contracts: Vec::new(),
},
true,
|_| false,
)
.expect("fill inactive claim capacity");
}
for index in 0..2 {
held.coordination
.register_claim(
WriteScopeClaim {
owner: format!("inactive-overflow-{index}"),
roots: vec![format!("pkg/overflow-{index}")],
exact_files: Vec::new(),
contracts: Vec::new(),
},
true,
|_| false,
)
.expect("evict oldest inactive claim at capacity");
}
assert!(
!held
.coordination
.write_claims
.iter()
.any(|claim| claim.claim.owner == claimant),
"the resolving claimant claim must be evicted for the durability regression"
);
held.persist_state_synchronously()
.expect("persist after inactive claim eviction");
Ok(())
})
.unwrap();
let mut final_replay = super::super::SubAgentManager::new(tmp.path().to_path_buf(), 4)
.with_state_path(state_path);
crates/engine/src/tools/subagent/coord/ledger.rs
+1 -1
View File
@@ -388,7 +388,7 @@ fn projected_claim(claim: &WriteScopeClaim, root: &Path) -> WriteScopeClaim {
}
/// Durable, bounded coordination state owned by `SubAgentManager`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct CoordinationLedger {
#[serde(default = "coordination_schema_version")]
pub schema_version: u32,
crates/engine/src/tools/subagent/lifecycle_tests.rs
+21 -9
View File
@@ -171,8 +171,7 @@ async fn lifecycle_resume_lineage_survives_persist_and_rejects_cycles_and_foreig
manager.insert_test_interrupted_continuable_agent("latest", &base, prior_messages());
manager.resume_targets.insert(a.clone(), b.clone());
manager.resume_targets.insert(b.clone(), c.clone());
let (path, payload) = manager.build_persist_payload().unwrap().unwrap();
write_json_atomic(&base, &path, &payload).unwrap();
manager.persist_state().unwrap().join().unwrap();
let mut loaded = SubAgentManager::new(base.clone(), 6).with_state_path(state_path);
loaded.load_state().unwrap();
assert_eq!(loaded.continuation_target(&a).unwrap(), c);
@@ -614,8 +613,9 @@ async fn lifecycle_continuation_link_is_durable_before_the_child_can_run() {
// Holding the manager lock keeps run_subagent_task_inner at its first
// await. This is the earliest published snapshot, before any child step.
assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 0);
let persisted: PersistedSubAgentState =
let persisted: DurableSubAgentStore =
serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
let persisted = persisted.state;
assert_eq!(
persisted.resume_targets.get(&source),
Some(&successor.agent_id)
@@ -635,7 +635,7 @@ async fn lifecycle_continuation_persist_failure_rolls_back_worker_and_link() {
let base = dir.path().canonicalize().unwrap();
let path = base.join(".codewhale/subagents/state.json");
let manager = Arc::new(RwLock::new(
SubAgentManager::new(base.clone(), 4).with_state_path(path),
SubAgentManager::new(base.clone(), 4).with_state_path(path.clone()),
));
let mut runtime = super::tests::stub_runtime();
runtime.manager = Arc::clone(&manager);
@@ -657,9 +657,17 @@ async fn lifecycle_continuation_persist_failure_rolls_back_worker_and_link() {
.permissions
.write
);
std::fs::create_dir_all(base.join(".codewhale")).unwrap();
std::fs::write(base.join(".codewhale/subagents"), "not a directory").unwrap();
let result = guard.resume_from_checkpoint(Arc::clone(&manager), runtime, &source, "Continue");
guard.persist_state_synchronously().unwrap();
let retained_path = path.with_extension("retained-before-failure");
let result = guard.with_store_transaction(|held| {
std::fs::rename(&path, &retained_path)?;
std::fs::create_dir(&path)?;
let resumed =
held.resume_from_checkpoint(Arc::clone(&manager), runtime, &source, "Continue");
std::fs::remove_dir(&path)?;
std::fs::rename(&retained_path, &path)?;
resumed
});
assert!(result.is_err());
assert_eq!(guard.agents.len(), 1);
assert_eq!(guard.worker_records.len(), 1);
@@ -824,7 +832,9 @@ async fn lifecycle_cleanup_keeps_source_identity_while_descendants_still_need_it
record.status = AgentWorkerStatus::Completed;
record.completed_at_ms = Some(epoch_millis_now().saturating_sub(120_000));
manager.cleanup_for_session("workspace", Duration::from_secs(60));
manager
.cleanup_for_session("workspace", Duration::from_secs(60))
.expect("durable worker operation");
assert_eq!(manager.continuation_target(&source).unwrap(), current);
assert!(
manager
@@ -847,7 +857,9 @@ async fn lifecycle_cleanup_keeps_source_identity_while_descendants_still_need_it
for record in manager.worker_records.values_mut() {
record.completed_at_ms = Some(epoch_millis_now().saturating_sub(120_000));
}
manager.cleanup_for_session("workspace", Duration::from_secs(60));
manager
.cleanup_for_session("workspace", Duration::from_secs(60))
.expect("durable worker operation");
assert!(manager.agents.is_empty());
assert!(manager.worker_records.is_empty());
assert!(manager.resume_targets.is_empty());
crates/engine/src/tools/subagent/mod.rs
+2383 -432
View File
File diff suppressed because it is too large. Load diff
crates/engine/src/tools/subagent/tests.rs
+136 -57
View File
@@ -1008,7 +1008,9 @@ fn persisted_subagent_state_has_bounded_serialized_size() {
// Sanity-check: file is valid JSON and contains expected records.
let parsed: serde_json::Value =
serde_json::from_slice(&serialized).expect("persisted state is valid JSON");
let workers = parsed["workers"].as_array().expect("workers array");
let workers = parsed["state"]["workers"]
.as_array()
.expect("workers array");
assert_eq!(
workers.len(),
10,
@@ -1023,31 +1025,36 @@ fn coordination_ledger_persists_and_replays_bounded_contracts() {
let mut manager =
SubAgentManager::new(tmp.path().to_path_buf(), 4).with_state_path(state_path.clone());
manager
.coordination
.register_claim(
WriteScopeClaim {
owner: "agent_writer".into(),
roots: vec!["src".into()],
exact_files: vec!["Cargo.toml".into()],
contracts: vec!["public-api".into()],
},
false,
|_| false,
)
.expect("claim");
manager
.record_coordination_decision(DecisionRecord {
decision_id: "decision_storage".into(),
subject: "storage".into(),
status: DecisionStatus::Accepted,
owner: "agent_writer".into(),
scope: vec!["router".into()],
constraints: vec!["bounded".into()],
evidence_handles: vec!["test:coordination".into()],
version: 1,
sequence: 0,
.with_store_transaction(|manager| {
manager
.coordination
.register_claim(
WriteScopeClaim {
owner: "agent_writer".into(),
roots: vec!["src".into()],
exact_files: vec!["Cargo.toml".into()],
contracts: vec!["public-api".into()],
},
false,
|_| false,
)
.expect("claim");
manager
.record_coordination_decision(DecisionRecord {
decision_id: "decision_storage".into(),
subject: "storage".into(),
status: DecisionStatus::Accepted,
owner: "agent_writer".into(),
scope: vec!["router".into()],
constraints: vec!["bounded".into()],
evidence_handles: vec!["test:coordination".into()],
version: 1,
sequence: 0,
})
.expect("decision");
Ok(())
})
.expect("decision");
.unwrap();
manager.persist_state().unwrap().join().unwrap();
let mut loaded = SubAgentManager::new(tmp.path().to_path_buf(), 4).with_state_path(state_path);
@@ -1411,7 +1418,12 @@ fn session_close_finalizes_live_fleet_and_releases_write_claims() {
assert!(overlap.contains("worker-a"), "{overlap}");
// Session close finalizes both owners and releases their claims.
assert!(manager.finalize_session_close() > 0);
assert!(
manager
.finalize_session_close()
.expect("durable worker operation")
> 0
);
assert!(!manager.active_coordination_owners().contains(&legacy_id));
assert!(!manager.active_coordination_owners().contains("worker-a"));
assert!(manager.coordination.write_claims.is_empty());
@@ -1427,7 +1439,12 @@ fn session_close_finalizes_live_fleet_and_releases_write_claims() {
// Idempotent: a second close pass has no live fleet to finalize and
// must not release (or re-own) anything.
assert_eq!(manager.finalize_session_close(), 0);
assert_eq!(
manager
.finalize_session_close()
.expect("durable worker operation"),
0
);
assert!(manager.coordination.write_claims.is_empty());
manager
.preflight_worker_coordination(&make_write_worker_spec(
@@ -1650,7 +1667,9 @@ fn a_second_session_without_the_lock_keeps_its_running_agents_live() {
let agent_id = agent.id.clone();
contender.agents.insert(agent_id.clone(), agent);
let cancelled = contender.cleanup(Duration::from_secs(3600));
let cancelled = contender
.cleanup(Duration::from_secs(3600))
.expect("durable worker operation");
assert_eq!(
cancelled, 0,
"lock loss alone must not terminalize anything"
@@ -1780,7 +1799,9 @@ fn cleanup_terminalizes_orphans_on_heartbeat_evidence_not_on_lock_ownership() {
let fresh = insert_orphan(&mut contender, "fresh-orphan", Duration::from_secs(1));
let stale = insert_orphan(&mut contender, "stale-orphan", Duration::from_secs(300));
let cancelled = contender.cleanup(Duration::from_secs(3600));
let cancelled = contender
.cleanup(Duration::from_secs(3600))
.expect("durable worker operation");
assert_eq!(
cancelled, 1,
"only the agent with stale heartbeat evidence terminalizes"
@@ -7420,7 +7441,9 @@ async fn late_completion_does_not_overwrite_cancelled_outcome() {
late.status = SubAgentStatus::Completed;
late.result = Some("late success".to_string());
assert!(
!manager.update_from_result(&agent_id, late),
!manager
.update_from_result(&agent_id, late)
.expect("durable worker operation"),
"late completion must lose the terminal transition"
);
@@ -7504,7 +7527,11 @@ async fn completion_claim_preserves_running_gate_and_excludes_late_cancel() {
let mut result = manager.get_result(&agent_id).unwrap();
result.status = SubAgentStatus::Completed;
result.result = Some("done".to_string());
assert!(manager.update_from_result(&agent_id, result));
assert!(
manager
.update_from_result(&agent_id, result)
.expect("durable worker operation")
);
assert_eq!(
manager.get_result(&agent_id).unwrap().status,
SubAgentStatus::Completed
@@ -11188,7 +11215,12 @@ async fn cleanup_auto_cancels_stale_running_agent_and_releases_slot() {
0,
"stale running agents must not keep the concurrency slot occupied"
);
assert_eq!(manager.cleanup(Duration::from_secs(60 * 60)), 1);
assert_eq!(
manager
.cleanup(Duration::from_secs(60 * 60))
.expect("durable worker operation"),
1
);
let snapshot = manager
.get_result(&agent_id)
@@ -11302,7 +11334,12 @@ async fn cleanup_keeps_recent_running_agent() {
manager.agents.insert(agent_id.clone(), agent);
assert_eq!(manager.running_count(), 1);
assert_eq!(manager.cleanup(Duration::from_secs(60 * 60)), 0);
assert_eq!(
manager
.cleanup(Duration::from_secs(60 * 60))
.expect("durable worker operation"),
0
);
assert_eq!(
manager.get_result(&agent_id).expect("agent").status,
SubAgentStatus::Running
@@ -11351,7 +11388,12 @@ async fn touch_refreshes_stale_running_agent_heartbeat() {
assert_eq!(manager.running_count(), 0);
assert!(manager.touch(&agent_id));
assert_eq!(manager.running_count(), 1);
assert_eq!(manager.cleanup(Duration::from_secs(60 * 60)), 0);
assert_eq!(
manager
.cleanup(Duration::from_secs(60 * 60))
.expect("durable worker operation"),
0
);
manager
.agents
.get_mut(&agent_id)
@@ -11388,6 +11430,8 @@ fn test_persist_and_reload_marks_running_agent_as_interrupted() {
.join()
.expect("persist thread");
drop(manager);
let mut reloaded = SubAgentManager::new(workspace, 2)
.with_state_path(default_state_path(tmp.path()).expect("default state path"));
reloaded.load_state().expect("load state");
@@ -11435,13 +11479,13 @@ fn generated_whale_name_is_not_persisted_or_replayed_on_load() {
serde_json::from_str(&std::fs::read_to_string(&state_path).expect("read persisted state"))
.expect("parse persisted state");
assert!(
persisted["agents"][0].get("nickname").is_none(),
persisted["state"]["agents"][0].get("nickname").is_none(),
"generated locale text is not durable identity"
);
// Recreate a pre-fix state file whose generated display came from a
// Japanese session. Loading under a later session must discard it.
persisted["agents"][0]["nickname"] = json!(generated);
persisted["state"]["agents"][0]["nickname"] = json!(generated);
std::fs::write(
&state_path,
serde_json::to_string_pretty(&persisted).expect("serialize legacy state"),
@@ -11493,7 +11537,7 @@ fn explicit_nonmatching_whale_word_is_persisted_and_loaded() {
serde_json::from_str(&std::fs::read_to_string(&state_path).expect("read persisted state"))
.expect("parse persisted state");
assert_eq!(
persisted["agents"][0]["nickname"],
persisted["state"]["agents"][0]["nickname"],
json!(explicit_whale),
"the explicit whale-word nickname remains durable"
);
@@ -14257,7 +14301,9 @@ fn park_child_at_turn_end(manager: &mut SubAgentManager, agent_id: &str) {
terminal.checkpoint = checkpoint;
terminal.needs_input = needs_input;
assert!(
manager.finish_terminal_result(agent_id, terminal, true, false),
manager
.finish_terminal_result(agent_id, terminal, true, false)
.expect("durable worker operation"),
"the parked projection must own the terminal transition"
);
}
@@ -15636,10 +15682,13 @@ fn persisted_advisory_assignment_roles_replay_and_repersist_as_consultant() {
)
.expect("parse canonical state");
assert_eq!(
repersisted["agents"][0]["assignment"]["role"],
repersisted["state"]["agents"][0]["assignment"]["role"],
json!("advisor")
);
assert_eq!(
repersisted["state"]["agents"][0]["agent_type"],
json!("advisor")
);
assert_eq!(repersisted["agents"][0]["agent_type"], json!("advisor"));
}
}
@@ -16342,7 +16391,9 @@ fn ownerless_legacy_rows_fail_closed_and_scoped_close_preserves_foreign_workers(
assign_test_session_owner(&mut manager, &agent_b, "session-b");
assert_eq!(
manager.finalize_session_close_for_session("session-a"),
manager
.finalize_session_close_for_session("session-a")
.expect("durable worker operation"),
1,
"paired agent and worker record are one finalized worker"
);
@@ -17110,12 +17161,14 @@ async fn run_subagent_task_claims_before_delivery_and_then_finalizes() {
.denied_tools
.contains(&"write".to_string())
);
held.cleanup(Duration::ZERO);
held.cleanup(Duration::ZERO)
.expect("durable worker operation");
assert!(held.worker_records.contains_key(&identity.child_id));
drop(held);
let persisted: PersistedSubAgentState =
let persisted: DurableSubAgentStore =
serde_json::from_slice(&std::fs::read(&state_path).expect("persisted file"))
.expect("persisted worker state");
let persisted = persisted.state;
assert_eq!(
persisted
.workers
@@ -17155,9 +17208,9 @@ async fn run_subagent_task_claims_before_delivery_and_then_finalizes() {
.expect("activation state")
.clone(),
};
let blocked_path = workspace.path().join("blocked-worker-store");
std::fs::create_dir(&blocked_path).expect("block publication with a directory");
manager.write().await.state_path = Some(blocked_path);
let retained_path = state_path.with_extension("retained-before-failure");
std::fs::rename(&state_path, &retained_path).expect("retain admitted store bytes");
std::fs::create_dir(&state_path).expect("block the admitted publication destination");
drop(prepared);
let error = tokio::time::timeout(Duration::from_secs(30), failed.settled())
.await
@@ -17167,10 +17220,12 @@ async fn run_subagent_task_claims_before_delivery_and_then_finalizes() {
assert!(failed.accepted().await.is_err());
{
let mut held = manager.write().await;
held.cleanup(Duration::ZERO);
held.cleanup(Duration::ZERO)
.expect_err("blocked durable worker store");
assert!(held.agents.contains_key(&identity.child_id));
assert!(held.worker_records.contains_key(&identity.child_id));
held.state_path = Some(state_path);
std::fs::remove_dir(&state_path).expect("remove the fixture obstruction");
std::fs::rename(&retained_path, &state_path).expect("restore the admitted store bytes");
}
let retried = persist_child_settlement(&manager, &identity)
.await
@@ -19866,7 +19921,9 @@ fn cleanup_evicts_stale_terminal_worker_records_and_keeps_live_ones() {
}
// One-hour retention matches COMPLETED_AGENT_RETENTION used by cleanup callers.
let auto_cancelled = manager.cleanup(Duration::from_secs(60 * 60));
let auto_cancelled = manager
.cleanup(Duration::from_secs(60 * 60))
.expect("durable worker operation");
assert_eq!(auto_cancelled, 0);
assert!(
@@ -19919,7 +19976,9 @@ fn cleanup_removes_complete_transcript_after_worker_retention_expires() {
.expect("write transcript artifact");
assert!(artifact.exists());
manager.cleanup(Duration::from_secs(60 * 60));
manager
.cleanup(Duration::from_secs(60 * 60))
.expect("durable worker operation");
assert!(manager.get_worker_record(agent_id).is_none());
assert!(
@@ -19942,7 +20001,9 @@ fn cleanup_due_gates_write_locked_cleanup_to_a_bounded_cadence() {
"a never-cleaned manager should be due"
);
manager.cleanup(Duration::from_secs(3600));
manager
.cleanup(Duration::from_secs(3600))
.expect("durable worker operation");
assert!(
!manager.cleanup_due(Duration::from_secs(3600)),
"immediately after cleanup it should not be due again within the interval"
@@ -20329,7 +20390,7 @@ fn write_json_atomic_survives_concurrent_writers() {
snapshot_sequence: i + 1,
..PersistedSubAgentState::default()
};
let _ = write_json_atomic(&ws, &p, &payload);
let _ = write_json_atomic(&ws, &p, &DurableSubAgentStore::from_legacy(payload));
}));
}
for h in handles {
@@ -20339,7 +20400,7 @@ fn write_json_atomic_survives_concurrent_writers() {
let contents = std::fs::read_to_string(&*path).expect("read state.json");
let parsed: serde_json::Value =
serde_json::from_str(&contents).expect("state.json must be complete/valid JSON");
assert!(parsed.get("snapshot_sequence").is_some());
assert!(parsed["state"].get("snapshot_sequence").is_some());
// No stray temp files left behind.
let leftover: Vec<_> = std::fs::read_dir(path.parent().unwrap())
.expect("read subagents dir")
@@ -21298,7 +21359,12 @@ async fn tracked_running_background_shell_keeps_its_owner_off_the_heartbeat_reap
1,
"a worker waiting on a tracked background shell must survive the heartbeat"
);
assert_eq!(manager.cleanup(Duration::from_secs(60 * 60)), 0);
assert_eq!(
manager
.cleanup(Duration::from_secs(60 * 60))
.expect("durable worker operation"),
0
);
// The converse: once the job stops running, nothing extends the heartbeat.
{
@@ -24232,7 +24298,11 @@ fn user_follow_up_to_completed_child_requires_a_runtime_to_resume() {
let mut terminal = manager.get_result(&agent_id).expect("snapshot");
terminal.status = SubAgentStatus::Completed;
terminal.result = Some("done".to_string());
assert!(manager.finish_terminal_result(&agent_id, terminal, true, false));
assert!(
manager
.finish_terminal_result(&agent_id, terminal, true, false)
.expect("durable worker operation")
);
let err = manager
.continue_child_from_user(handle, None, &agent_id, "one more")
.expect_err("no runtime, no resume");
@@ -25302,7 +25372,11 @@ mod child_permission_gate {
let mut terminal = manager.get_result(&agent_id).expect("owned agent");
terminal.status = SubAgentStatus::Cancelled;
terminal.needs_input = None;
assert!(manager.finish_terminal_result(&agent_id, terminal, false, true));
assert!(
manager
.finish_terminal_result(&agent_id, terminal, false, true)
.expect("durable worker operation")
);
}
});
let approval_id = next_child_approval_id(&mut rx).await;
@@ -25463,7 +25537,12 @@ mod child_permission_gate {
"held",
)
.unwrap();
assert!(manager.finalize_session_close_for_session("workspace") > 0);
assert!(
manager
.finalize_session_close_for_session("workspace")
.expect("durable worker operation")
> 0
);
assert_eq!(
manager.pending_child_approvals(),
0,
crates/engine/src/tools/workflow/mod.rs
+65 -13
View File
@@ -3909,6 +3909,7 @@ struct SubAgentWorkflowDriver {
owner_session_id: String,
manager: SharedSubAgentManager,
runtime: SubAgentRuntime,
owner_runtime: Option<tokio::runtime::Handle>,
parent_cancel_token: CancellationToken,
state: Arc<WorkflowWorkspaceState>,
completion_tx: mpsc::Sender<SubAgentCompletion>,
@@ -4007,6 +4008,7 @@ impl SubAgentWorkflowDriver {
owner_session_id,
manager,
runtime,
owner_runtime: tokio::runtime::Handle::try_current().ok(),
parent_cancel_token,
state,
completion_tx,
@@ -4046,7 +4048,13 @@ impl SubAgentWorkflowDriver {
if let Ok(mut permits) = self.spawn_permits.lock() {
permits.clear();
}
cancel_child_agents(self.manager.clone(), ids);
cancel_child_agents(
self.manager.clone(),
self.owner_session_id.clone(),
ids,
self.owner_runtime.clone(),
self.runtime.event_tx.clone(),
);
if let Ok(mut state) = self.completion_state.lock() {
for (_, waiter) in state.waiters.drain() {
let _ = waiter.send(TaskCompletion::Cancelled);
@@ -4892,7 +4900,22 @@ impl SubAgentWorkflowDriver {
};
let task_id = result.result.agent_id.clone();
if let Err(err) = self.ensure_admission_open() {
let _ = self.manager.write().await.cancel_agent(&task_id);
let manager = self.manager.clone();
let session = self.owner_session_id.clone();
let target = task_id.clone();
let cancelled = codewhale_runtime::runtime_control::owner_work(move || {
manager
.blocking_write()
.cancel_agent_for_session(&session, &target)
.map(|_| ())
})
.await;
let err = match cancelled {
Ok(()) => err,
Err(error) => DriverError::Rejected(format!(
"{err}; cancellation of child {task_id} is uncertain: {error}"
)),
};
if let Some(receipt) = &fleet_receipt {
self.record_orphaned_fleet_receipt(receipt, &err.to_string());
}
@@ -5700,27 +5723,56 @@ fn provider_usage_was_reported(
.any(Option::is_some)
}
fn cancel_child_agents(manager: SharedSubAgentManager, ids: Vec<String>) {
fn cancel_child_agents(
manager: SharedSubAgentManager,
session: String,
ids: Vec<String>,
owner_runtime: Option<tokio::runtime::Handle>,
event_tx: Option<mpsc::Sender<Event>>,
) {
if ids.is_empty() {
return;
}
if let Ok(mut manager_guard) = manager.try_write() {
for id in ids {
let _ = manager_guard.cancel_agent(&id);
}
return;
}
if tokio::runtime::Handle::try_current().is_ok() {
if let Some(owner_runtime) =
owner_runtime.or_else(|| tokio::runtime::Handle::try_current().ok())
{
let _runtime = owner_runtime.enter();
spawn_supervised(
"workflow-cancel-children",
std::panic::Location::caller(),
async move {
let mut manager_guard = manager.write().await;
for id in ids {
let _ = manager_guard.cancel_agent(&id);
let cancelled = codewhale_runtime::runtime_control::owner_work(move || {
let mut manager = manager.blocking_write();
let mut failure = None;
for id in ids {
if let Err(error) = manager.cancel_agent_for_session(&session, &id) {
failure.get_or_insert(error);
}
}
failure.map_or(Ok(()), Err)
})
.await;
if let Err(error) = cancelled {
let message = format!("Workflow child cancellation is uncertain: {error}");
tracing::error!(%message);
if let Some(event_tx) = event_tx {
let _ = event_tx
.send(Event::error(
codewhale_error_taxonomy::ErrorEnvelope::transient(message),
))
.await;
}
}
},
);
} else {
let message = "Workflow child cancellation is uncertain: no runtime is available";
tracing::error!(%message);
if let Some(event_tx) = event_tx {
let _ = event_tx.try_send(Event::error(
codewhale_error_taxonomy::ErrorEnvelope::transient(message),
));
}
}
}
crates/runtime/src/runtime_control.rs
+4
View File
@@ -15,6 +15,8 @@ where
T: Send + 'static,
F: FnOnce() -> anyhow::Result<T> + Send + 'static,
{
#[cfg(any(test, feature = "test-support"))]
let env_ticket = crate::test_support::env_scope_ticket();
static SLOTS: std::sync::OnceLock<std::sync::Arc<tokio::sync::Semaphore>> =
std::sync::OnceLock::new();
let permit = SLOTS
@@ -23,6 +25,8 @@ where
.acquire_owned()
.await?;
tokio::task::spawn_blocking(move || {
#[cfg(any(test, feature = "test-support"))]
let _membership = crate::test_support::join_env_scope(env_ticket);
let _permit = permit;
work()
})
crates/tui/CHANGELOG.md
+2
View File
@@ -160,6 +160,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- Worker coordination serializes durable admissions across processes, preserves foreign workers and winning completion receipts, and fences stale checkpoints. Fleet restart reservations retain uncertain prior processes until exact teardown is confirmed; final output, usage and the receipt publish together without duplicate accounting on retry. Retained transcripts follow committed ownership.
- JSONPath projections bound path length, traversal steps and intermediate matches before cloning results. Direct fetches share the same parser and enforce one aggregate serialized-field budget; oversized selections refuse with narrower-selection guidance.
- Long tool output and diffs use bounded display windows and a byte-limited row cache. Retained details load verified byte ranges and label what search and copy include; explicit source copy keeps stored text with control-sequence cleanup. Oversized Unicode is shown safely without changing question answers; long answers retain their exact bytes in the existing owned output artifact before bounded model-context projection. New pager notices are translated across all 15 complete language packs.
crates/tui/src/exec_agent.rs
+1 -1
View File
@@ -665,7 +665,7 @@ pub(crate) async fn run_exec_agent(
.unwrap_or_else(codewhale_engine::tools::subagent::AdvisorConfig::disabled),
};
let engine_handle = spawn_engine(engine_config, &execution_config, None);
let engine_handle = spawn_engine(engine_config, &execution_config, None).await;
// A restricted posture that resolves to no enforcing sandbox must reach
// the person running exec, not only the model's posture line: policy-only
// is accepted (read-only already fails closed in the shell tool), but it
crates/tui/src/lib.rs
+75 -43
View File
@@ -3793,17 +3793,6 @@ async fn run_fleet_command(workspace: &Path, config: &Config, args: FleetArgs) -
.active_provider_identity()
.map_err(anyhow::Error::msg)?;
let max_subagents = config.max_subagents_for_provider(&identity);
let coordination_manager =
codewhale_engine::tools::subagent::new_shared_subagent_manager_with_timeout(
workspace.to_path_buf(),
max_subagents,
config
.max_admitted_subagents_for_provider(&identity)
.max(max_subagents),
Duration::from_secs(config.subagent_heartbeat_timeout_secs_for_provider(&identity)),
config.launch_concurrency_for_provider(&identity),
);
// Probe the durable ledger *before* opening the manager: FleetManager::open
// creates `.codewhale/fleet.jsonl` as a side effect, so a later probe would
// always find a ledger and the CLI would report availability differently
@@ -3837,12 +3826,30 @@ async fn run_fleet_command(workspace: &Path, config: &Config, args: FleetArgs) -
// The configured route is the operator: fleet workers without a
// task/profile model pin inherit the session's active model.
let manager = FleetManager::open(workspace)?
.with_exec_config(fleet_config.exec.clone())
.with_fleet_config(fleet_config)
.with_sub_agent_manager(coordination_manager)
.with_session_model(config.default_model())
.with_route_config(config.clone());
let owned_workspace = workspace.to_path_buf();
let owned_config = config.clone();
let manager = codewhale_runtime::runtime_control::owner_work(move || {
let config = &owned_config;
let max_subagents = config.max_subagents_for_provider(&identity);
let coordination_manager =
codewhale_engine::tools::subagent::new_shared_subagent_manager_with_timeout(
owned_workspace.clone(),
max_subagents,
config
.max_admitted_subagents_for_provider(&identity)
.max(max_subagents),
Duration::from_secs(config.subagent_heartbeat_timeout_secs_for_provider(&identity)),
config.launch_concurrency_for_provider(&identity),
);
let manager = FleetManager::open(&owned_workspace)?
.with_exec_config(fleet_config.exec.clone())
.with_fleet_config(fleet_config)
.with_sub_agent_manager(coordination_manager)
.with_session_model(config.default_model())
.with_route_config(config.clone());
Ok(manager)
})
.await?;
match args.command {
FleetCommand::Init => {
println!("Fleet ledger: {}", manager.ledger_path().display());
@@ -3853,7 +3860,12 @@ async fn run_fleet_command(workspace: &Path, config: &Config, args: FleetArgs) -
let max_workers = args.max_workers.clamp(1, 128);
let manager =
manager.with_stale_after(Duration::from_secs(args.stale_after_seconds.max(1)));
let report = manager.create_run_from_task_spec_path(&args.task_spec, max_workers)?;
let launch_manager = manager.clone();
let task_spec = args.task_spec.clone();
let report = codewhale_runtime::runtime_control::owner_work(move || {
launch_manager.create_run_from_task_spec_path(&task_spec, max_workers)
})
.await?;
println!(
"Fleet run: {} tasks={} leased={} queued={}",
report.run_id.0, report.task_count, report.leased, report.queued
@@ -3872,19 +3884,20 @@ async fn run_fleet_command(workspace: &Path, config: &Config, args: FleetArgs) -
println!(
"manager loop running; use `codewhale fleet status`, `inspect`, `interrupt`, or `stop --all` from another terminal."
);
let mut executor = FleetExecutor::new(workspace)
let executor = FleetExecutor::new(workspace)
.with_sessions_dir(session_manager::default_sessions_dir()?);
let codewhale_binary = fleet::executor::configured_codewhale_binary();
let status = manager
.run_to_completion(
&report.run_id,
max_workers,
&mut executor,
executor,
&codewhale_binary,
None,
Duration::from_secs(2),
)
.await?;
.await?
.0;
print_status(&status);
Ok(())
}
@@ -3928,25 +3941,30 @@ async fn run_fleet_command(workspace: &Path, config: &Config, args: FleetArgs) -
))
}
FleetCommand::Restart { worker_id } => {
let report = manager.restart_worker(&worker_id)?;
let restart_manager = manager.clone();
let report = codewhale_runtime::runtime_control::owner_work(move || {
restart_manager.restart_worker(&worker_id)
})
.await?;
print_inspection(&report.inspection);
println!(
"manager loop running for restarted run {}; use `codewhale fleet status`, `inspect`, `interrupt`, or `stop --all` from another terminal.",
report.run_id.0
);
let mut executor = FleetExecutor::new(workspace)
let executor = FleetExecutor::new(workspace)
.with_sessions_dir(session_manager::default_sessions_dir()?);
let codewhale_binary = fleet::executor::configured_codewhale_binary();
let status = manager
.run_to_completion(
&report.run_id,
report.max_workers,
&mut executor,
executor,
&codewhale_binary,
None,
Duration::from_secs(2),
)
.await?;
.await?
.0;
print_status(&status);
Ok(())
}
@@ -3955,14 +3973,19 @@ async fn run_fleet_command(workspace: &Path, config: &Config, args: FleetArgs) -
stale_after_seconds,
} => {
let manager = manager.with_stale_after(Duration::from_secs(stale_after_seconds.max(1)));
emit_fleet_receipt(&fleet_control::execute_fleet_control_with(
ControlSurface::Cli,
workspace,
fleet_context,
&manager,
ControlOperation::FleetResume,
Some(&run_id),
))
let owned_workspace = workspace.to_path_buf();
let receipt = codewhale_runtime::runtime_control::owner_work(move || {
Ok(fleet_control::execute_fleet_control_with(
ControlSurface::Cli,
&owned_workspace,
fleet_context,
&manager,
ControlOperation::FleetResume,
Some(&run_id),
))
})
.await?;
emit_fleet_receipt(&receipt)
}
FleetCommand::Stop { all } => {
if !all {
@@ -14933,16 +14956,25 @@ async fn build_direct_workflow_tool(
context = context.with_sandbox_backend(Arc::from(backend));
}
let max_subagents = config.max_subagents_for_provider(&identity);
let manager = new_shared_subagent_manager_with_timeout(
workspace.to_path_buf(),
max_subagents,
config
.max_admitted_subagents_for_provider(&identity)
.max(max_subagents),
Duration::from_secs(config.subagent_heartbeat_timeout_secs_for_provider(&identity)),
config.launch_concurrency_for_provider(&identity),
);
let owned_workspace = workspace.to_path_buf();
let owned_config = config.clone();
let owned_identity = identity.clone();
let manager = codewhale_runtime::runtime_control::owner_work(move || {
let config = &owned_config;
let identity = owned_identity;
let max_subagents = config.max_subagents_for_provider(&identity);
let manager = new_shared_subagent_manager_with_timeout(
owned_workspace,
max_subagents,
config
.max_admitted_subagents_for_provider(&identity)
.max(max_subagents),
Duration::from_secs(config.subagent_heartbeat_timeout_secs_for_provider(&identity)),
config.launch_concurrency_for_provider(&identity),
);
Ok(manager)
})
.await?;
let roster = Arc::new(codewhale_engine::fleet::identity::load_effective_roster(
&config.fleet_config(),
workspace,
crates/tui/src/runtime_api.rs
+24 -7
View File
@@ -2222,7 +2222,12 @@ pub async fn run_http_server(
existing_binding.as_ref(),
)?;
let sessions_dir = runtime_threads.sessions_dir().to_path_buf();
let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, options.workers)?;
let sub_agent_manager = codewhale_runtime::runtime_control::owner_work({
let workspace = workspace.clone();
let workers = options.workers;
move || runtime_api_sub_agent_manager(&workspace, workers)
})
.await?;
runtime_threads
.attach_subagent_manager(sub_agent_manager.clone())
.await?;
@@ -4496,7 +4501,13 @@ async fn start_fleet_run(
));
}
let run_id = FleetRunId::from(run_id);
let report = manager.activate_run(&run_id).map_err(|error| {
let activation_manager = manager.clone();
let activation_run = run_id.clone();
let report = codewhale_runtime::runtime_control::owner_work(move || {
activation_manager.activate_run(&activation_run)
})
.await
.map_err(|error| {
let message = format!("Failed to start Fleet run '{}': {error}", run_id.0);
if message.contains("already terminal") {
ApiError::conflict(message)
@@ -4516,12 +4527,12 @@ async fn start_fleet_run(
let workspace_scope = state.workspace_scope.clone();
tokio::spawn(async move {
let _workspace_scope = workspace_scope;
let mut executor = FleetExecutor::new(&workspace).with_sessions_dir(sessions_dir);
let executor = FleetExecutor::new(&workspace).with_sessions_dir(sessions_dir);
if let Err(error) = manager
.run_to_completion(
&execution_run_id,
max_workers,
&mut executor,
executor,
&codewhale_binary,
None,
Duration::from_millis(250),
@@ -4848,7 +4859,13 @@ async fn restart_fleet_worker(
Path(worker_id): Path<String>,
) -> Result<Json<Value>, ApiError> {
let manager = open_fleet_manager(&state)?;
let report = manager.restart_worker(&worker_id).map_err(|err| {
let restart_manager = manager.clone();
let restart_worker = worker_id.clone();
let report = codewhale_runtime::runtime_control::owner_work(move || {
restart_manager.restart_worker(&restart_worker)
})
.await
.map_err(|err| {
ApiError::bad_request(format!(
"Failed to restart Fleet worker '{worker_id}': {err}"
))
@@ -4862,12 +4879,12 @@ async fn restart_fleet_worker(
let workspace_scope = state.workspace_scope.clone();
tokio::spawn(async move {
let _workspace_scope = workspace_scope;
let mut executor = FleetExecutor::new(&workspace).with_sessions_dir(sessions_dir);
let executor = FleetExecutor::new(&workspace).with_sessions_dir(sessions_dir);
if let Err(err) = manager
.run_to_completion(
&run_id,
max_workers,
&mut executor,
executor,
&codewhale_binary,
None,
Duration::from_millis(250),
crates/tui/src/runtime_api/tests.rs
+6 -3
View File
@@ -3113,6 +3113,10 @@ async fn agent_runs_runtime_api_exposes_persisted_worker_receipts() -> Result<()
},
pending_request: None,
status: AgentWorkerStatus::Completed,
prepared_generation: None,
prepared_previous_attempt_unsettled: false,
launch_admission_version: 1,
recovery_interrupted: false,
created_at_ms: 1,
updated_at_ms: 2,
started_at_ms: Some(1),
@@ -3379,9 +3383,8 @@ async fn agent_run_cancel_refuses_a_run_owned_by_a_session_it_does_not_host() ->
// A child parked on a question in another terminal session: in flight on
// disk, but no engine in this runtime owns it.
let mut record = {
let manager =
codewhale_engine::tools::subagent::new_shared_subagent_manager(workspace.clone(), 1);
let mut guard = manager.write().await;
let mut guard =
codewhale_engine::tools::subagent::SubAgentManager::new(workspace.clone(), 1);
let id = guard.insert_test_running_agent("elsewhere", &workspace);
guard.assign_test_session_owner(&id, "terminal-session");
guard
crates/tui/src/tui/ui.rs
+5 -4
View File
@@ -496,7 +496,7 @@ mod memory_quick_add_tests {
}
}
fn spawn_tui_engine(
async fn spawn_tui_engine(
app: &mut App,
api_config: &Config,
mcp_handoff: Option<codewhale_engine::mcp::McpHandoff>,
@@ -505,7 +505,8 @@ fn spawn_tui_engine(
build_engine_config(app, api_config),
api_config,
mcp_handoff,
);
)
.await;
app.mcp_snapshot_generation = 0;
app.mcp_snapshot_generation_invalidated = false;
// Prime durable agent + coordination state through the same engine event
@@ -551,7 +552,7 @@ async fn retire_tui_engine(
async fn replace_tui_engine(app: &mut App, api_config: &Config, engine_handle: &mut EngineHandle) {
let mcp_handoff = retire_tui_engine(app, engine_handle).await;
*engine_handle = spawn_tui_engine(app, api_config, mcp_handoff);
*engine_handle = spawn_tui_engine(app, api_config, mcp_handoff).await;
}
/// Startup and consent-triggered replacement restore the same conversation
@@ -562,7 +563,7 @@ async fn spawn_tui_engine_with_session(
config: &Config,
mcp_handoff: Option<codewhale_engine::mcp::McpHandoff>,
) -> Result<EngineHandle> {
let handle = spawn_tui_engine(app, config, mcp_handoff);
let handle = spawn_tui_engine(app, config, mcp_handoff).await;
let restored = async {
if !app.api_messages.is_empty() {
handle
docs/architecture/delegated-coordination.md
+119 -43
View File
@@ -107,52 +107,128 @@ they did is reachable through an `agent` action — `status`/`peek` for `list`,
`message`, `followup`, `interrupt`, `wait`, and `claim` for the one capability
that had no equivalent, write-scope expansion.
## The workspace lock, and what losing it does and does not mean
## Durable workspace transactions and writer ownership
The ledger lives in one file, `.codewhale/state/subagents.v1.json`, written as a
whole-document atomic replace. One per-workspace advisory flock
(`subagents.v1.lock`) decides who may *write* the file, in one of two modes:
The current ledger is `.codewhale/state/subagents.v3.json`: one format3
envelope around the existing agent, worker and coordination schemas. It records
writer provenance, publication sequences and a durable revision. It is the sole
current state authority. Transcripts remain their existing separate artifacts;
legacy recovery bytes are retained backups, not another writable ledger.
- **Shared, warn on failure.** Interactive and `exec` sessions, `codewhale
fleet`, `codewhale resume`/`fork` mounts and the TUI's background task runtime
(`/task`, `task_create`, automations) take the flock shared, so several
sessions in one folder hold it at once. Failing to get it only logs a warning.
- **Exclusive, degrade read-only.** `codewhale serve` and the other standalone
Runtime hosts take it exclusively and refuse every durable ledger append when
another process already holds it; they log the refusal and pick the ledger back
up when the process restarts after that owner exits. Beside such an owner,
shared holders cannot append.
A short OS transaction lock serializes store initialization, migration and every
read-current/check/mutate/publish operation. Claim overlap, decisions and their
sequence/session/scope receipts are checked against that current state before
acknowledgement. Unrelated writers' records are preserved. A stale snapshot
cannot replace a newer publication from its writer, an older launch generation
cannot replace a newer one, and a terminal receipt cannot be resurrected as
Running by a checkpoint. An explicit terminal retry accepts only the same
winning outcome; authoritative newer accounting metadata is retained.
Every shared holder may append, and each append replaces the whole file, so the
last rename wins. The ledger is loaded once at boot; agents another process
started reload as non-live, so no other process's write claims are enforced
here. A resume or fork mount only reads the ledger.
Normal interactive/exec/Fleet/background sessions continue to share the
existing admission flock. The explicit exclusive Runtime constructor retains
its host admission policy; it is not the mechanism that conserves shared
snapshots. No manager holds the transaction lock while a child executes or
while it waits for a provider or channel capacity. Conflicting lock contention
returns a bounded retry error. Unsupported locking or failed store admission
refuses durable writes rather than silently creating an empty writable ledger.
A shared holder beside an exclusive owner, or one whose acquisition failed, does
not get the flock. That is an ordinary state, not a failure:
Each writer holds its own boot lease. Other managers preserve its live records
and use its active claims during admission. An existing unlocked lease can
establish that its writer ended; a missing, invalid or unreadable lease is
uncertain and fails closed. Local task handles and a failed admission flock do
not establish that another process's work stopped. Parked and headless
WaitingForUser claim rules retain their existing lifecycle meaning; they do not
transfer a still-live writer's record ownership to a different boot.
- **It does not affect liveness.** A session that cannot write the ledger runs
its own agents normally. Whether an agent is alive is decided by heartbeat
evidence, never by lock ownership. (Before v0.9.4 the cleanup pass
terminalized every running agent with no live task handle purely because this
process lacked the flock; that coupling is gone.)
- **It does not affect reads.** The boot-time load is unconditional. Such a
session sees the workspace's decisions and write claims even though it cannot
append to them. Gating the load on the write flock previously left a session
without the flock holding an empty default ledger, which it would write straight over
the real one the moment the first session exited and the flock became
acquirable.
- **It does mean no durable ledger appends.** Decision, claim, contention, and
reconciliation mutations still require the flock, and so does any
shared-workspace write-capable launch, because such a launch must be durably
replayable before it executes. A session without the flock can therefore
delegate only read-only and isolated-worktree work. Two shared holders can
both launch shared-workspace writers, and neither enforces the other's claims
(see the known gap below).
Shared coordination mutations run through the existing manager transaction.
Failed operations do not automatically publish partial state. A deliberately
persisted contention receipt remains inspectable even though admission returns
an error. Fleet compensation is fenced by the exact store revision and the
same writer's successful publication; a stale snapshot cannot undo another
writer's work. Worker retention removes its owned records durably before
removing their transcript artifacts or evicting handles.
Known gap: every session holds the ledger as of its own boot, and shared holders
rewrite the whole file. Records another session appended after this one booted
can be lost when this one writes, and write claims are only enforced within one
process. Closing that needs per-session ledger segments unioned on read, so that
no two processes ever write the same file and the claim-overlap check runs
against the union. That work is not done.
Taking over a proven-dead writer refreshes its exact durable worker and agent
projection and records that recovery before the requested mutation. The newer
snapshot sequence fences any older captured checkpoint, including for preflight.
Failed admission restores ownership from the last committed recovery image.
Synthetic recovery interruption is recorded as uncertainty and can be resolved
by the exact authoritative terminal outcome; it cannot authorize replay of an
admitted attempt. Unresolved recovery records remain retained. Local heartbeat
cleanup and retention act only on owned
records; transcript deletion also checks the committed retained identities.
Fleet restart preparation records the next exact generation as an unstarted
reservation. Cold recovery preserves that reservation until the matching Fleet
lease consumes it. The executor durably admits the full registered launch spec
before starting a process; an admitted or settled attempt cannot launch again
under the same generation. Current ledger cancellation can settle an unstarted
reservation only after any previous attempt has a confirmed outcome; an old
ledger attempt cannot consume its prepared successor.
An explicit controller may prepare that exact next generation for a live
foreign headless worker under the existing Fleet lease comparison, preserving
the execution owner's writer identity and every unrelated receipt. The owner
refreshes its durable generation before its next mutation; old-generation
checkpoints cannot undo the preparation.
The existing Fleet ledger and executor move through the bounded blocking owner
with the same manager-lock file. Successful completion reconciles the worker
projection after processing the executor's terminal event.
The per-worker `launch_admission_version` distinguishes the current durable
pre-spawn protocol from legacy rows. A missing legacy start timestamp or a
legacy terminal projection does not prove process teardown. Preparation retains
`prepared_previous_attempt_unsettled` until the owning executor confirms the
exact prior run, task, generation and full launch spec, or an authoritative
attempt receipt supplies that proof. A cold owner without either proof refuses
the reserved launch and retains its uncertainty. It does not invent a Failed
outcome or release a live writer's claim. Busy coordination yields without
leasing; unrelated foreign workers remain untouched.
The existing Fleet executor retains observed terminal output, the prepared
receipt and stopped host until publication is acknowledged. Final stream tails,
including usage, and the terminal receipt form one `TaskAttemptFinalized` ledger
record. An exact retry synchronizes the same candidate instead of appending
usage again. A visible record after a failed sync is not an acknowledgment.
The ledger remembers successful durability only for the same owned file and
length; appends, replacement and compaction invalidate that receipt.
Terminal completion keeps the public Running gate and its ownership until the
exact terminal result is durably recorded and parent/mailbox/host fan-in is
queued. The admitted bounded blocking owner performs that complete sequence,
so dropping its waiter cannot discard an admitted commit. A filesystem or
publication failure produces uncertainty, retains the winning pending result
and claims, and sends no successful completion. A later retry uses that same
winner. Failed live delivery after a successful commit leaves the durable
terminal fact intact and reports the delivery failure. Explicit in-memory
managers remain in-memory; their results are not durable receipts.
Current store publication propagates file-write and Unix directory-sync errors.
Opening or observing a new store revision resolves its directory durability
before acknowledging idempotent state, and exact terminal retries repeat that
barrier. Unchanged preflight does not add a directory sync.
An error after rename can still mean the new image is visible, so recovery reads
the committed image and never announces success solely from a failed waiter.
Native filesystem and power-loss qualification remain separate evidence.
Canceling a Fleet waiter retains an admitted blocking batch and its lock until
the batch returns. Dropped executor ownership uses the existing parent-death
pipe or Windows Job containment; that fallback does not create a successful
terminal receipt or claim that a remote effect was undone.
### Legacy cutover and recovery
The first current-format admission takes the legacy admission lock exclusively
only for migration, after previous holders have left. It retains the original
legacy bytes and a validated recovery copy, then leaves
`subagents.v1.json` as a private directory containing that recovery material.
The current format3 file is published under the same short transaction lock.
Interrupted cutover states are validated and resumed without replacing a later
current-format revision with the old backup. Future or malformed envelopes
cannot fall back to parsing as legacy state.
The retained directory prevents the inspected older atomic file publisher from
replacing that pathname with its stale snapshot. It does not prevent an old
binary from executing tools or changing a workspace. Concurrent arbitrary old
binaries are not a supported current coordination domain; close them before
cutover. Recovery data is not silently deleted, and migration qualification
uses isolated synthetic state rather than existing user sessions. Different
explicit state roots remain separate coordination domains as described above.
scripts/check-blocking-calls-budget.json
+2 -1
View File
@@ -211,7 +211,8 @@
},
"crates/engine/src/tools/subagent/mod.rs": {
"path_canonicalize": 4,
"std_fs": 4
"std_fs": 4,
"thread_sleep": 1
},
"crates/engine/src/tools/subagent/worktree.rs": {
"path_canonicalize": 6