Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

cargo fmt

Browse Source
This commit is contained in:
Hare
2026-04-27 22:51:07 +09:00
parent bcaa4645f7
commit 7a0ed7d744
62 changed files with 485 additions and 527 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/pod/build.rs
+1 -6
View File
@@ -28,12 +28,7 @@ fn main() {
let prompt_section = parsed
.get("prompt")
.and_then(|v| v.as_table())
.unwrap_or_else(|| {
panic!(
"{} must contain a `[prompt]` table",
toml_path.display()
)
});
.unwrap_or_else(|| panic!("{} must contain a `[prompt]` table", toml_path.display()));
let mut keys: Vec<String> = prompt_section.keys().cloned().collect();
keys.sort();
crates/pod/src/compact/worker.rs
+2 -1
View File
@@ -59,7 +59,8 @@ impl CompactWorkerContext {
}
fn remaining_budget(&self) -> u64 {
self.auto_read_budget.saturating_sub(self.auto_read_consumed)
self.auto_read_budget
.saturating_sub(self.auto_read_consumed)
}
}
crates/pod/src/controller.rs
+15 -13
View File
@@ -8,16 +8,16 @@ use tokio::sync::{broadcast, mpsc, oneshot};
use crate::ipc::alerter::Alerter;
use crate::ipc::notify_buffer::NotifyBuffer;
use crate::ipc::server::SocketServer;
use crate::pod::{Pod, PodError, PodRunResult};
use crate::runtime::dir::RuntimeDir;
use crate::shared_state::{PodSharedState, PodStatus};
use crate::spawn::comm_tools::{
list_pods_tool, read_pod_output_tool, send_to_pod_tool, stop_pod_tool,
};
use crate::runtime::dir::RuntimeDir;
use crate::shared_state::{PodSharedState, PodStatus};
use crate::ipc::server::SocketServer;
use crate::spawn::tool::spawn_pod_tool;
use crate::spawn::registry::SpawnedPodRegistry;
use protocol::{ErrorCode, Event, Method, AlertLevel, AlertSource, RunResult, TurnResult};
use crate::spawn::tool::spawn_pod_tool;
use protocol::{AlertLevel, AlertSource, ErrorCode, Event, Method, RunResult, TurnResult};
// ---------------------------------------------------------------------------
// PodHandle — client-facing, Clone-able
@@ -215,11 +215,7 @@ impl PodController {
let alerter_for_worker = alerter.clone();
worker.on_warning(move |message| {
alerter_for_worker.alert(
AlertLevel::Warn,
AlertSource::Worker,
message.to_owned(),
);
alerter_for_worker.alert(AlertLevel::Warn, AlertSource::Worker, message.to_owned());
});
// Register the builtin file-manipulation tools (Read / Write /
@@ -735,9 +731,15 @@ where
.map(|def| def().0.name)
.collect();
tool_names.extend(
["SpawnPod", "SendToPod", "ReadPodOutput", "StopPod", "ListPods"]
.iter()
.map(|s| (*s).into()),
[
"SpawnPod",
"SendToPod",
"ReadPodOutput",
"StopPod",
"ListPods",
]
.iter()
.map(|s| (*s).into()),
);
protocol::Greeting {
pod_name: manifest.pod.name.clone(),
crates/pod/src/hook.rs
+1 -1
View File
@@ -15,10 +15,10 @@
//! exposing the underlying mutable state.
use async_trait::async_trait;
use llm_worker::tool::ToolOutput;
use llm_worker::interceptor::{
PostToolAction, PreRequestAction, PreToolAction, PromptAction, TurnEndAction,
};
use llm_worker::tool::ToolOutput;
use serde_json::Value;
// =============================================================================
crates/pod/src/interrupt_and_run.rs
+3 -5
View File
@@ -42,7 +42,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
if !closures.is_empty() {
self.worker_mut().extend_history(closures);
}
self.worker_mut().push_item(Item::system_message(system_note));
self.worker_mut()
.push_item(Item::system_message(system_note));
self.run(input).await
}
}
@@ -84,10 +85,7 @@ mod tests {
#[test]
fn no_orphans_returns_empty() {
let history = vec![
Item::user_message("hi"),
Item::assistant_message("hello"),
];
let history = vec![Item::user_message("hi"), Item::assistant_message("hello")];
let summary = interrupt_tool_result_summary();
assert!(orphan_tool_result_closures(&history, &summary).is_empty());
}
crates/pod/src/ipc/event.rs
+1 -1
View File
@@ -26,9 +26,9 @@ use std::sync::Arc;
use protocol::{Method, PodEvent, ScopeRule};
use crate::spawn::comm_tools::connect_and_send;
use crate::runtime::dir::SpawnedPodRecord;
use crate::runtime::scope_lock::{self, ScopeLockError};
use crate::spawn::comm_tools::connect_and_send;
use crate::spawn::registry::SpawnedPodRegistry;
/// Connect to `socket`, send a single `Method::PodEvent(event)`, and
crates/pod/src/ipc/interceptor.rs
+1 -1
View File
@@ -21,13 +21,13 @@ use session_store::UsageRecord;
use tracing::info;
use crate::compact::state::CompactState;
use crate::compact::token_counter::total_tokens_impl;
use crate::hook::{
AbortInfo, HookRegistry, PreRequestInfo, PromptSubmitInfo, ToolCallSummary, ToolResultSummary,
TurnEndInfo,
};
use crate::ipc::notify_buffer::{NotifyBuffer, format_notify};
use crate::prompt::catalog::PromptCatalog;
use crate::compact::token_counter::total_tokens_impl;
use tracing::warn;
/// Maximum number of bytes copied into `TurnEndInfo::final_text_preview`.
crates/pod/src/ipc/notify_buffer.rs
+4 -1
View File
@@ -105,7 +105,10 @@ mod tests {
assert_eq!(drained.len(), CAPACITY);
// Oldest 5 were dropped; first retained is msg5.
assert_eq!(drained[0].message, "msg5");
assert_eq!(drained[CAPACITY - 1].message, format!("msg{}", CAPACITY + 4));
assert_eq!(
drained[CAPACITY - 1].message,
format!("msg{}", CAPACITY + 4)
);
}
#[test]
crates/pod/src/lib.rs
+1 -1
View File
@@ -18,7 +18,7 @@ pub use hook::{Hook, HookEventKind, HookRegistryBuilder};
pub use ipc::alerter::Alerter;
pub use ipc::server::SocketServer;
pub use manifest::{
AuthRef, ModelManifest, PodManifest, PodManifestConfig, PodMetaConfig, Scope, SchemeKind,
AuthRef, ModelManifest, PodManifest, PodManifestConfig, PodMetaConfig, SchemeKind, Scope,
};
pub use pod::{Pod, PodError, PodRunResult, apply_worker_manifest};
pub use prompt::catalog::{CatalogError, PodPrompt, PromptCatalog};
crates/pod/src/main.rs
+1 -4
View File
@@ -171,10 +171,7 @@ async fn main() -> ExitCode {
// (e.g. the TUI's interactive `spawn` flow). Tab-separated so a
// pod name with spaces still parses cleanly. Emit before the
// human line so a stderr-watching parent sees it first.
eprintln!(
"INSOMNIA-READY\t{pod_name}\t{}",
socket_path.display()
);
eprintln!("INSOMNIA-READY\t{pod_name}\t{}", socket_path.display());
eprintln!("pod: {pod_name} listening on {:?}", socket_path);
tokio::select! {
crates/pod/src/pod.rs
+23 -34
View File
@@ -13,25 +13,25 @@ use tracing::{info, warn};
use manifest::{PodManifest, PodManifestConfig, ResolveError, Scope, ScopeError, WorkerManifest};
use crate::prompt::agents_md::read_agents_md;
use crate::compact::state::CompactState;
use crate::compact::usage_tracker::UsageTracker;
use crate::hook::{
Hook, HookRegistryBuilder, OnAbort, OnPromptSubmit, OnTurnEnd, PostToolCall, PreLlmRequest,
PreRequestInfo, PreToolCall,
};
use crate::ipc::alerter::Alerter;
use crate::ipc::notify_buffer::NotifyBuffer;
use crate::ipc::interceptor::PodInterceptor;
use crate::prompt::loader::PromptLoader;
use crate::ipc::notify_buffer::NotifyBuffer;
use crate::prompt::agents_md::read_agents_md;
use crate::prompt::catalog::{CatalogError, PromptCatalog};
use crate::prompt::loader::PromptLoader;
use crate::prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTemplate};
use crate::runtime::dir;
use crate::runtime::scope_lock::{self, ScopeAllocationGuard, ScopeLockError};
use crate::prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTemplate};
use crate::compact::usage_tracker::UsageTracker;
use protocol::{AlertLevel, AlertSource, Event, Segment};
use tokio::sync::broadcast;
use async_trait::async_trait;
use llm_worker::interceptor::PreRequestAction;
use protocol::{AlertLevel, AlertSource, Event, Segment};
use tokio::sync::broadcast;
/// Pre-LLM-request hook that records `history.len()` at send time into a
/// shared `UsageTracker`. The on_usage callback later pairs this with the
@@ -511,9 +511,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
None
};
let usage_history_handle = compact_state
.as_ref()
.map(|_| self.usage_history.clone());
let usage_history_handle = compact_state.as_ref().map(|_| self.usage_history.clone());
let interceptor = PodInterceptor::new(
registry,
@@ -553,11 +551,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
let agents_md_read = read_agents_md(&self.pwd);
for warning in agents_md_read.warnings {
if let Some(n) = alerter.as_ref() {
n.alert(
AlertLevel::Warn,
AlertSource::AgentsMd,
warning,
);
n.alert(AlertLevel::Warn, AlertSource::AgentsMd, warning);
}
}
// Resident-injection collection: only when memory is enabled in
@@ -603,10 +597,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// Equivalent to `run(vec![Segment::text(s)])`. The dumb-client
/// counterpart of [`protocol::Method::run_text`]; primarily for
/// tests and tools that have only a string in hand.
pub async fn run_text(
&mut self,
s: impl Into<String>,
) -> Result<PodRunResult, PodError> {
pub async fn run_text(&mut self, s: impl Into<String>) -> Result<PodRunResult, PodError> {
self.run(vec![Segment::text(s)]).await
}
@@ -995,7 +986,12 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
.manifest
.compaction
.as_ref()
.map(|c| (c.compact_auto_read_budget, c.compact_worker_max_input_tokens))
.map(|c| {
(
c.compact_auto_read_budget,
c.compact_worker_max_input_tokens,
)
})
.unwrap_or((
manifest::defaults::COMPACT_AUTO_READ_BUDGET,
manifest::defaults::COMPACT_WORKER_MAX_INPUT_TOKENS,
@@ -1054,8 +1050,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// Tools: read_file (shared scope, fresh tracker) + the three
// compact-specific tools that populate `ctx`.
summary_worker.register_tool(tools::read_tool(scoped_fs.clone(), summary_tracker));
summary_worker
.register_tool(mark_read_required_tool(scoped_fs.clone(), ctx.clone()));
summary_worker.register_tool(mark_read_required_tool(scoped_fs.clone(), ctx.clone()));
summary_worker.register_tool(add_reference_tool(ctx.clone()));
summary_worker.register_tool(write_summary_tool(ctx.clone()));
@@ -1092,10 +1087,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
}
};
if let Some(prompt) = nudge {
let _ = locked_worker
.run(prompt)
.await
.map_err(PodError::Worker)?;
let _ = locked_worker.run(prompt).await.map_err(PodError::Worker)?;
}
let final_ctx = ctx.lock().expect("compact ctx poisoned").clone();
@@ -1154,7 +1146,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// Build new history: [summary, ...auto-read, references, ...retained].
let mut new_history = Vec::with_capacity(
1 + auto_read_messages.len() + reference_message.is_some() as usize
1 + auto_read_messages.len()
+ reference_message.is_some() as usize
+ retained_items.len(),
);
new_history.push(Item::system_message(format!(
@@ -1456,9 +1449,7 @@ fn build_summary_input(items: &[Item], default_refs: &[PathBuf]) -> String {
}
out.push_str("## Conversation\n");
out.push_str(&build_summary_prompt(items));
out.push_str(
"\n\nWhen you are done, call `write_summary` with the final 5-section text.",
);
out.push_str("\n\nWhen you are done, call `write_summary` with the final 5-section text.");
out
}
@@ -1579,10 +1570,8 @@ fn current_pwd() -> Result<PathBuf, PodError> {
pwd: PathBuf::from("."),
source,
})?;
cwd.canonicalize().map_err(|source| PodError::InvalidPwd {
pwd: cwd,
source,
})
cwd.canonicalize()
.map_err(|source| PodError::InvalidPwd { pwd: cwd, source })
}
#[cfg(test)]
crates/pod/src/prompt/catalog.rs
+2 -6
View File
@@ -310,10 +310,7 @@ impl PromptCatalog {
}
/// Render `PodPrompt::WorkingBoundariesSection` with `{{ scope_summary }}`.
pub fn working_boundaries_section(
&self,
scope_summary: &str,
) -> Result<String, CatalogError> {
pub fn working_boundaries_section(&self, scope_summary: &str) -> Result<String, CatalogError> {
self.render(
PodPrompt::WorkingBoundariesSection,
single("scope_summary", scope_summary),
@@ -343,8 +340,7 @@ fn single(key: &'static str, value: &str) -> Value {
}
fn parse_builtin_pack() -> Result<HashMap<String, String>, CatalogError> {
let parsed: PackFile =
toml::from_str(INTERNAL_TOML).map_err(CatalogError::ParseBuiltin)?;
let parsed: PackFile = toml::from_str(INTERNAL_TOML).map_err(CatalogError::ParseBuiltin)?;
Ok(parsed.prompt)
}
crates/pod/src/prompt/loader.rs
+14 -12
View File
@@ -23,8 +23,7 @@ use std::path::{Path, PathBuf};
use include_dir::{Dir, include_dir};
use thiserror::Error;
static BUILTIN_PROMPTS: Dir<'static> =
include_dir!("$CARGO_MANIFEST_DIR/../../resources/prompts");
static BUILTIN_PROMPTS: Dir<'static> = include_dir!("$CARGO_MANIFEST_DIR/../../resources/prompts");
const PREFIX_INSOMNIA: &str = "$insomnia";
const PREFIX_USER: &str = "$user";
@@ -190,10 +189,12 @@ impl PromptLoader {
}
if let Some(prefix) = trimmed.strip_prefix('$') {
let (prefix_name, rest) =
prefix.split_once('/').ok_or_else(|| LoaderError::InvalidRef {
raw: raw.to_string(),
reason: "prefix must be followed by '/'".into(),
})?;
prefix
.split_once('/')
.ok_or_else(|| LoaderError::InvalidRef {
raw: raw.to_string(),
reason: "prefix must be followed by '/'".into(),
})?;
let prefix = parse_prefix(raw, prefix_name)?;
let path = normalize_path(raw, rest)?;
Ok(PromptRef { prefix, path })
@@ -293,10 +294,7 @@ fn load_from_dir(dir: &Path, reference: &PromptRef) -> Result<String, LoaderErro
}
}
fn load_from_include_dir(
dir: &Dir<'static>,
reference: &PromptRef,
) -> Result<String, LoaderError> {
fn load_from_include_dir(dir: &Dir<'static>, reference: &PromptRef) -> Result<String, LoaderError> {
let path = format!("{}.md", reference.path);
dir.get_file(&path)
.and_then(|f| f.contents_utf8())
@@ -349,7 +347,9 @@ mod tests {
#[test]
fn missing_file_is_hard_error() {
let loader = PromptLoader::builtins_only();
let err = loader.resolve("$insomnia/definitely-missing", None).unwrap_err();
let err = loader
.resolve("$insomnia/definitely-missing", None)
.unwrap_err();
assert!(matches!(err, LoaderError::NotFound { .. }));
}
@@ -380,7 +380,9 @@ mod tests {
#[test]
fn unqualified_ref_resolves_relative_to_current() {
let loader = PromptLoader::builtins_only();
let current = loader.parse_ref("$insomnia/common/tool-usage", None).unwrap();
let current = loader
.parse_ref("$insomnia/common/tool-usage", None)
.unwrap();
// Sibling lookup under the same prefix and directory.
let sibling = loader.parse_ref("workspace", Some(&current)).unwrap();
assert_eq!(sibling.to_qualified_string(), "$insomnia/common/workspace");
crates/pod/src/prompt/system.rs
+10 -10
View File
@@ -23,8 +23,8 @@ use minijinja::value::Value;
use minijinja::{Environment, ErrorKind, UndefinedBehavior};
use thiserror::Error;
use crate::prompt::loader::{LoaderError, PromptLoader, PromptRef};
use crate::prompt::catalog::{CatalogError, PromptCatalog};
use crate::prompt::loader::{LoaderError, PromptLoader, PromptRef};
#[derive(Debug, Error)]
pub enum SystemPromptError {
@@ -55,10 +55,7 @@ impl SystemPromptTemplate {
/// Parse the instruction asset referenced by `instruction_ref`
/// using the supplied [`PromptLoader`]. The reference is resolved
/// at parse time so syntax errors surface immediately.
pub fn parse(
instruction_ref: &str,
loader: PromptLoader,
) -> Result<Self, SystemPromptError> {
pub fn parse(instruction_ref: &str, loader: PromptLoader) -> Result<Self, SystemPromptError> {
let root_ref = loader
.parse_ref(instruction_ref, None)
.map_err(SystemPromptError::LoaderResolve)?;
@@ -75,9 +72,7 @@ impl SystemPromptTemplate {
// The joined name is then looked up via `set_loader` below.
let loader_for_join = loader.clone();
env.set_path_join_callback(move |name, parent| {
let parent_ref = loader_for_join
.parse_ref(parent, None)
.ok();
let parent_ref = loader_for_join.parse_ref(parent, None).ok();
match loader_for_join.parse_ref(name, parent_ref.as_ref()) {
Ok(r) => r.to_qualified_string().into(),
// Propagate the raw name on error so set_loader surfaces
@@ -93,7 +88,10 @@ impl SystemPromptTemplate {
.map_err(|e| minijinja::Error::new(ErrorKind::TemplateNotFound, e.to_string()))?;
match loader_for_src.load(&reference) {
Ok(source) => Ok(Some(source)),
Err(e) => Err(minijinja::Error::new(ErrorKind::TemplateNotFound, e.to_string())),
Err(e) => Err(minijinja::Error::new(
ErrorKind::TemplateNotFound,
e.to_string(),
)),
}
});
@@ -459,7 +457,9 @@ mod tests {
let tmpl = SystemPromptTemplate::parse("$user/ghost", loader).unwrap();
let dir = TempDir::new().unwrap();
let scope = build_scope(dir.path());
let err = tmpl.render(&ctx(dir.path(), &scope, vec![], None)).unwrap_err();
let err = tmpl
.render(&ctx(dir.path(), &scope, vec![], None))
.unwrap_err();
assert!(matches!(err, SystemPromptError::Render(_)));
}
crates/pod/src/runtime/dir.rs
+1 -4
View File
@@ -82,10 +82,7 @@ impl RuntimeDir {
/// Write `spawned_pods.json` atomically. The entries are the full
/// set of spawned children known to this Pod — callers pass the
/// replacement list, no incremental merge.
pub async fn write_spawned_pods(
&self,
records: &[SpawnedPodRecord],
) -> Result<(), io::Error> {
pub async fn write_spawned_pods(&self, records: &[SpawnedPodRecord]) -> Result<(), io::Error> {
let json = serde_json::to_vec_pretty(records).map_err(io::Error::other)?;
atomic_write(&self.path.join("spawned_pods.json"), &json).await
}
crates/pod/src/runtime/scope_lock.rs
+8 -13
View File
@@ -206,10 +206,7 @@ pub fn is_within_effective_write(lock: &LockFile, parent: &str, rule: &ScopeRule
return false;
};
if rule.permission != Permission::Write {
return alloc
.scope_allow
.iter()
.any(|r| covers_fully(r, rule));
return alloc.scope_allow.iter().any(|r| covers_fully(r, rule));
}
let covered = alloc
.scope_allow
@@ -244,7 +241,11 @@ pub fn find_conflict_owner(
if rule.permission != Permission::Write {
return None;
}
for alloc in lock.allocations.iter().filter(|a| a.delegated_from.is_none()) {
for alloc in lock
.allocations
.iter()
.filter(|a| a.delegated_from.is_none())
{
if let Some(owner) = find_conflict_in_subtree(lock, alloc, rule) {
if Some(owner.as_str()) == exempt {
continue;
@@ -526,18 +527,12 @@ pub enum ScopeLockError {
#[error("pod name `{0}` is already registered")]
DuplicatePodName(String),
#[error("requested scope `{}` conflicts with pod `{competitor}`", .rule.target.display())]
WriteConflict {
competitor: String,
rule: ScopeRule,
},
WriteConflict { competitor: String, rule: ScopeRule },
#[error(
"requested scope `{}` is not within spawner `{spawner}`'s effective scope",
.rule.target.display()
)]
NotSubset {
spawner: String,
rule: ScopeRule,
},
NotSubset { spawner: String, rule: ScopeRule },
#[error("pod `{0}` is not registered")]
UnknownPod(String),
}
crates/pod/src/spawn/comm_tools.rs
+13 -15
View File
@@ -43,8 +43,7 @@ struct NameInput {
// SendToPod
// ---------------------------------------------------------------------------
const SEND_TO_POD_DESCRIPTION: &str =
"Send a text message to a previously spawned Pod. The spawned Pod \
const SEND_TO_POD_DESCRIPTION: &str = "Send a text message to a previously spawned Pod. The spawned Pod \
processes it as a user turn. Fails if the Pod is already executing a \
turn — retry after it finishes. Does not wait for the turn to complete; \
use `ReadPodOutput` to fetch results afterwards.";
@@ -109,8 +108,7 @@ pub fn send_to_pod_tool(registry: Arc<SpawnedPodRegistry>) -> ToolDefinition {
// ReadPodOutput
// ---------------------------------------------------------------------------
const READ_POD_OUTPUT_DESCRIPTION: &str =
"Fetch new assistant text from a spawned Pod since the last read. \
const READ_POD_OUTPUT_DESCRIPTION: &str = "Fetch new assistant text from a spawned Pod since the last read. \
Uses an internal cursor per-Pod so consecutive calls return only \
newly-produced output. Returns the Pod's current status and the new \
text, or reports `stopped` if the Pod can no longer be reached.";
@@ -122,9 +120,8 @@ struct ReadPodOutputTool {
#[async_trait]
impl Tool for ReadPodOutputTool {
async fn execute(&self, input_json: &str) -> Result<ToolOutput, ToolError> {
let input: NameInput = serde_json::from_str(input_json).map_err(|e| {
ToolError::InvalidArgument(format!("invalid ReadPodOutput input: {e}"))
})?;
let input: NameInput = serde_json::from_str(input_json)
.map_err(|e| ToolError::InvalidArgument(format!("invalid ReadPodOutput input: {e}")))?;
let record = self
.registry
.get(&input.name)
@@ -154,7 +151,10 @@ impl Tool for ReadPodOutputTool {
format!("pod `{}` running; no new assistant text", input.name)
} else {
let lines = new_text.lines().count();
format!("pod `{}`: {lines} new line(s) of assistant text", input.name)
format!(
"pod `{}`: {lines} new line(s) of assistant text",
input.name
)
};
let content = if new_text.is_empty() {
None
@@ -183,8 +183,7 @@ pub fn read_pod_output_tool(registry: Arc<SpawnedPodRegistry>) -> ToolDefinition
// StopPod
// ---------------------------------------------------------------------------
const STOP_POD_DESCRIPTION: &str =
"Terminate a spawned Pod and reclaim the delegated scope. The Pod \
const STOP_POD_DESCRIPTION: &str = "Terminate a spawned Pod and reclaim the delegated scope. The Pod \
receives `Shutdown`; its scope entry is released in the machine-wide \
registry so the spawner can spawn a new Pod over the same paths.";
@@ -247,8 +246,7 @@ pub fn stop_pod_tool(registry: Arc<SpawnedPodRegistry>) -> ToolDefinition {
// ListPods
// ---------------------------------------------------------------------------
const LIST_PODS_DESCRIPTION: &str =
"List all Pods spawned by this Pod along with their reachability \
const LIST_PODS_DESCRIPTION: &str = "List all Pods spawned by this Pod along with their reachability \
status (`alive` / `stopped`) and the scope each was granted.";
#[derive(Debug, Deserialize, schemars::JsonSchema)]
@@ -364,9 +362,9 @@ async fn send_run_and_confirm(socket: &Path, input: String) -> Result<(), SendRu
input: vec![protocol::Segment::text(input)],
}),
)
.await
.map_err(|_| SendRunError::Io("write timed out".into()))?
.map_err(|e| SendRunError::Io(format!("write: {e}")))?;
.await
.map_err(|_| SendRunError::Io("write timed out".into()))?
.map_err(|e| SendRunError::Io(format!("write: {e}")))?;
loop {
let event = tokio::time::timeout(SOCKET_OP_TIMEOUT, reader.next::<Event>())
.await
crates/pod/src/spawn/registry.rs
+6 -2
View File
@@ -43,7 +43,9 @@ impl SpawnedPodRegistry {
pub async fn add(&self, record: SpawnedPodRecord) -> io::Result<()> {
let mut records = self.records.lock().await;
records.push(record);
self.runtime_dir.write_spawned_pods(records.as_slice()).await
self.runtime_dir
.write_spawned_pods(records.as_slice())
.await
}
/// Look up a record by pod name. Cloned so callers can drop the lock.
@@ -67,7 +69,9 @@ impl SpawnedPodRegistry {
let mut records = self.records.lock().await;
let idx = records.iter().position(|r| r.pod_name == pod_name);
let removed = idx.map(|i| records.remove(i));
self.runtime_dir.write_spawned_pods(records.as_slice()).await?;
self.runtime_dir
.write_spawned_pods(records.as_slice())
.await?;
removed
};
self.cursors.lock().await.remove(pod_name);
crates/pod/src/spawn/tool.rs
+7 -4
View File
@@ -294,9 +294,9 @@ impl SpawnPodTool {
.stderr(Stdio::from(stderr_file))
.process_group(0);
let child = cmd
.spawn()
.map_err(|e| ToolError::ExecutionFailed(format!("failed to spawn `{pod_command}`: {e}")))?;
let child = cmd.spawn().map_err(|e| {
ToolError::ExecutionFailed(format!("failed to spawn `{pod_command}`: {e}"))
})?;
// Default `kill_on_drop = false` keeps the process alive after
// the `Child` is dropped. We intentionally do not `.wait()` —
@@ -498,7 +498,10 @@ mod tests {
assert_eq!(parsed.model.scheme, Some(SchemeKind::Anthropic));
assert_eq!(parsed.model.model_id.as_deref(), Some("claude-sonnet-4"));
assert_eq!(parsed.model.base_url.as_deref(), Some("https://example.test"));
assert_eq!(
parsed.model.base_url.as_deref(),
Some("https://example.test")
);
let file = match parsed.model.auth {
Some(AuthRef::ApiKey { file, .. }) => file,
_ => panic!("expected ApiKey"),
crates/pod/tests/controller_test.rs
+37 -62
View File
@@ -1,6 +1,6 @@
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use futures::{Stream, StreamExt};
@@ -169,10 +169,7 @@ async fn run_updates_shared_state_to_idle_after_completion() {
let pod = make_pod(client).await;
let handle = spawn_controller(pod).await;
handle
.send(Method::run_text("Hello"))
.await
.unwrap();
handle.send(Method::run_text("Hello")).await.unwrap();
// Wait for the run to complete
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -186,10 +183,7 @@ async fn run_populates_history() {
let pod = make_pod(client).await;
let handle = spawn_controller(pod).await;
handle
.send(Method::run_text("Hello"))
.await
.unwrap();
handle.send(Method::run_text("Hello")).await.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -207,10 +201,7 @@ async fn events_are_broadcast() {
let handle = spawn_controller(pod).await;
let mut rx = handle.subscribe();
handle
.send(Method::run_text("Hello"))
.await
.unwrap();
handle.send(Method::run_text("Hello")).await.unwrap();
let mut saw_turn_start = false;
let mut saw_text_delta = false;
@@ -258,16 +249,10 @@ async fn double_run_returns_error() {
let mut rx = handle.subscribe();
// Send first run
handle
.send(Method::run_text("first"))
.await
.unwrap();
handle.send(Method::run_text("first")).await.unwrap();
// Immediately send second run (should get error)
handle
.send(Method::run_text("second"))
.await
.unwrap();
handle.send(Method::run_text("second")).await.unwrap();
// Look for the error event
let mut saw_already_running = false;
@@ -410,8 +395,14 @@ async fn run_with_paste_segment_inlines_content_and_emits_typed_user_message() {
.iter()
.find_map(|i| i.as_text().map(|s| s.to_string()))
.unwrap_or_default();
assert!(user_text.contains("see line1\nline2 thanks"), "got: {user_text:?}");
assert!(!user_text.contains("[Clipboard"), "label must not leak: {user_text:?}");
assert!(
user_text.contains("see line1\nline2 thanks"),
"got: {user_text:?}"
);
assert!(
!user_text.contains("[Clipboard"),
"label must not leak: {user_text:?}"
);
}
#[tokio::test]
@@ -424,12 +415,11 @@ async fn run_with_unresolved_segment_emits_alert_and_placeholder() {
let segments = vec![
protocol::Segment::text("look at "),
protocol::Segment::FileRef { path: "src/lib.rs".into() },
protocol::Segment::FileRef {
path: "src/lib.rs".into(),
},
];
handle
.send(Method::Run { input: segments })
.await
.unwrap();
handle.send(Method::Run { input: segments }).await.unwrap();
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
let mut saw_alert_for_file_ref = false;
@@ -527,10 +517,7 @@ async fn notify_while_running_does_not_emit_already_running_error() {
let handle = spawn_controller(pod).await;
let mut rx = handle.subscribe();
handle
.send(Method::run_text("start"))
.await
.unwrap();
handle.send(Method::run_text("start")).await.unwrap();
handle
.send(Method::Notify {
message: "ping".into(),
@@ -591,10 +578,7 @@ async fn socket_run_receives_events() {
let mut writer = JsonLineWriter::new(writer);
// Send run method via socket
writer
.write(&Method::run_text("Hello"))
.await
.unwrap();
writer.write(&Method::run_text("Hello")).await.unwrap();
// Collect events
let mut saw_turn_start = false;
@@ -739,10 +723,7 @@ async fn pause_then_resume_transitions_and_preserves_history_consistency() {
let handle = spawn_controller(pod).await;
let mut rx = handle.subscribe();
handle
.send(Method::run_text("hello"))
.await
.unwrap();
handle.send(Method::run_text("hello")).await.unwrap();
// Wait for the partial text_delta to confirm the first stream is
// live before we pause.
@@ -794,10 +775,7 @@ async fn pause_then_resume_transitions_and_preserves_history_consistency() {
// (partial text is not committed), no orphan tool_use.
let history_json = handle.shared_state.history_json();
let items: Vec<serde_json::Value> = serde_json::from_str(&history_json).unwrap();
let roles: Vec<&str> = items
.iter()
.filter_map(|i| i["role"].as_str())
.collect();
let roles: Vec<&str> = items.iter().filter_map(|i| i["role"].as_str()).collect();
assert_eq!(
roles,
vec!["user", "assistant"],
@@ -850,10 +828,7 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() {
let handle = spawn_controller(pod).await;
let mut rx = handle.subscribe();
handle
.send(Method::run_text("first"))
.await
.unwrap();
handle.send(Method::run_text("first")).await.unwrap();
// Wait for ToolCallDone — the ToolCall is committed to history
// right before the Worker enters tool execution and pends.
@@ -883,10 +858,7 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() {
// New user input while Paused → controller routes to
// `Pod::interrupt_and_run`, which closes the orphan + injects a
// system note before the fresh user message.
handle
.send(Method::run_text("new request"))
.await
.unwrap();
handle.send(Method::run_text("new request")).await.unwrap();
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |e| matches!(
e,
@@ -925,9 +897,7 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() {
saw_interruption_note = true;
}
}
llm_worker::Item::Message { role, content, .. }
if *role == llm_worker::Role::User =>
{
llm_worker::Item::Message { role, content, .. } if *role == llm_worker::Role::User => {
let text: String = content.iter().map(|p| p.as_text()).collect();
if text.contains("new request") {
saw_new_user = true;
@@ -952,10 +922,10 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() {
// Also confirm the closure chain is ordered: tool_result for the
// orphan precedes the system note, which precedes the new user
// message.
let idx = |pred: &dyn Fn(&llm_worker::Item) -> bool| {
items.iter().position(pred).unwrap()
};
let tool_result_idx = idx(&|i| matches!(i, llm_worker::Item::ToolResult { call_id, .. } if call_id == "call_orphan"));
let idx = |pred: &dyn Fn(&llm_worker::Item) -> bool| items.iter().position(pred).unwrap();
let tool_result_idx = idx(
&|i| matches!(i, llm_worker::Item::ToolResult { call_id, .. } if call_id == "call_orphan"),
);
let sys_idx = idx(&|i| match i {
llm_worker::Item::Message {
role: llm_worker::Role::System,
@@ -980,7 +950,12 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() {
.contains("new request"),
_ => false,
});
assert!(tool_result_idx < sys_idx, "tool_result must precede system note");
assert!(sys_idx < user_idx, "system note must precede new user message");
assert!(
tool_result_idx < sys_idx,
"tool_result must precede system note"
);
assert!(
sys_idx < user_idx,
"system note must precede new user message"
);
}
crates/pod/tests/pod_comm_tools_test.rs
+17 -5
View File
@@ -14,11 +14,11 @@ use std::sync::{Arc, LazyLock, Mutex};
use llm_worker::llm_client::types::{ContentPart, Item, Role};
use llm_worker::tool::ToolOutput;
use manifest::{Permission, ScopeRule};
use pod::runtime::dir::{RuntimeDir, SpawnedPodRecord};
use pod::runtime::scope_lock::{self, LockFileGuard};
use pod::spawn::comm_tools::{
list_pods_tool, read_pod_output_tool, send_to_pod_tool, stop_pod_tool,
};
use pod::runtime::dir::{RuntimeDir, SpawnedPodRecord};
use pod::runtime::scope_lock::{self, LockFileGuard};
use pod::spawn::registry::SpawnedPodRegistry;
use protocol::stream::{JsonLineReader, JsonLineWriter};
use protocol::{ErrorCode, Event, Greeting, Method};
@@ -211,7 +211,11 @@ async fn send_to_pod_delivers_run_method() {
let (_meta, tool) = def();
let input = json!({ "name": "child", "message": "hello there" }).to_string();
let output: ToolOutput = tool.execute(&input).await.unwrap();
assert!(output.summary.contains("child"), "summary: {}", output.summary);
assert!(
output.summary.contains("child"),
"summary: {}",
output.summary
);
let method = received.await.unwrap().expect("expected a method");
match method {
@@ -292,7 +296,11 @@ async fn read_pod_output_returns_new_assistant_text_then_empty_on_second_call()
// Cursor now points past all items — second call returns no new text.
let second: ToolOutput = tool.execute(&input).await.unwrap();
assert!(second.content.is_none(), "unexpected content: {:?}", second.content);
assert!(
second.content.is_none(),
"unexpected content: {:?}",
second.content
);
assert!(
second.summary.contains("no new assistant text"),
"summary: {}",
@@ -451,6 +459,10 @@ async fn list_pods_empty_when_nothing_registered() {
let def = list_pods_tool(registry);
let (_meta, tool) = def();
let output: ToolOutput = tool.execute("{}").await.unwrap();
assert!(output.summary.contains("no spawned pods"), "{}", output.summary);
assert!(
output.summary.contains("no spawned pods"),
"{}",
output.summary
);
assert!(output.content.is_none());
}
crates/pod/tests/pod_events_test.rs
+1 -3
View File
@@ -77,9 +77,7 @@ fn clear_runtime_dir() {
}
/// Accept a single connection, read one `Method`, and return it.
fn accept_one_method(
listener: UnixListener,
) -> tokio::task::JoinHandle<Option<Method>> {
fn accept_one_method(listener: UnixListener) -> tokio::task::JoinHandle<Option<Method>> {
tokio::spawn(async move {
let (stream, _) = listener.accept().await.ok()?;
let (reader, _writer) = stream.into_split();
crates/pod/tests/spawn_pod_test.rs
+11 -6
View File
@@ -14,8 +14,8 @@ use llm_worker::tool::{ToolError, ToolOutput};
use manifest::{AuthRef, ModelManifest, Permission, SchemeKind, ScopeRule};
use pod::runtime::dir::{RuntimeDir, SpawnedPodRecord};
use pod::runtime::scope_lock::{self, LockFileGuard};
use pod::spawn::tool::spawn_pod_tool;
use pod::spawn::registry::SpawnedPodRegistry;
use pod::spawn::tool::spawn_pod_tool;
use protocol::Method;
use protocol::stream::JsonLineReader;
use serde_json::json;
@@ -99,9 +99,7 @@ async fn bind_mock_pod_socket(runtime_base: &Path, pod_name: &str) -> (PathBuf,
/// `Method` line, then returns it. `wait_for_socket` inside the tool
/// makes a probe connection that carries no data, so the task must
/// tolerate an empty connection and keep listening.
fn accept_one_method(
listener: UnixListener,
) -> tokio::task::JoinHandle<Option<Method>> {
fn accept_one_method(listener: UnixListener) -> tokio::task::JoinHandle<Option<Method>> {
tokio::spawn(async move {
loop {
let (stream, _) = listener.accept().await.ok()?;
@@ -192,7 +190,11 @@ async fn spawn_pod_delegates_scope_and_sends_run() {
.to_string();
let output: ToolOutput = tool.execute(&input).await.unwrap();
assert!(output.summary.contains("child"), "summary: {}", output.summary);
assert!(
output.summary.contains("child"),
"summary: {}",
output.summary
);
// Verify the tool delivered Method::Run to the socket.
let method = received.await.unwrap().expect("expected one Method line");
@@ -261,7 +263,10 @@ async fn spawn_pod_rejects_scope_outside_spawner() {
let err = tool.execute(&input).await.unwrap_err();
match err {
ToolError::InvalidArgument(msg) => {
assert!(msg.contains("not within"), "expected NotSubset wording: {msg}");
assert!(
msg.contains("not within"),
"expected NotSubset wording: {msg}"
);
}
other => panic!("expected InvalidArgument, got {other:?}"),
}
crates/pod/tests/system_prompt_template_test.rs
+8 -8
View File
@@ -246,11 +246,11 @@ async fn agents_md_absent_omits_trailing_section() {
#[tokio::test]
async fn agents_md_not_reread_after_compact() {
let client = MockClient::new(vec![
single_text_events("a"), // pod.run_text("first")
single_text_events("b"), // pod.run_text("second")
single_text_events("a"), // pod.run_text("first")
single_text_events("b"), // pod.run_text("second")
write_summary_tool_use_events("call-1", "compacted summary"), // compact worker: tool_use
single_text_events("done"), // compact worker: close
single_text_events("c"), // pod.run_text("third")
single_text_events("done"), // compact worker: close
single_text_events("c"), // pod.run_text("third")
]);
let (mut pod, pwd) = make_pod_with_body("BODY", client).await.unwrap();
let agents_path = pwd.join("AGENTS.md");
@@ -278,11 +278,11 @@ async fn agents_md_not_reread_after_compact() {
#[tokio::test]
async fn compact_preserves_system_prompt() {
let client = MockClient::new(vec![
single_text_events("a"), // pod.run_text("first")
single_text_events("b"), // pod.run_text("second")
single_text_events("a"), // pod.run_text("first")
single_text_events("b"), // pod.run_text("second")
write_summary_tool_use_events("call-1", "compacted summary"), // compact worker: tool_use
single_text_events("done"), // compact worker: close
single_text_events("c"), // pod.run_text("third")
single_text_events("done"), // compact worker: close
single_text_events("c"), // pod.run_text("third")
]);
let (mut pod, _pwd) = make_pod_with_body("SP cwd={{ cwd }}", client)
.await