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

TUIからPauseする実装

Browse Source
This commit is contained in:
Hare
2026-04-19 14:27:53 +09:00
parent 605e78468c
commit 223d06c77e
10 changed files with 823 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/pod/tests/controller_test.rs
+358 -6
View File
@@ -3,10 +3,11 @@ use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
use async_trait::async_trait;
use futures::Stream;
use futures::{Stream, StreamExt};
use llm_worker::Worker;
use llm_worker::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
use llm_worker::llm_client::{ClientError, LlmClient, Request};
use llm_worker::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use session_store::FsStore;
use pod::{Event, Method, Pod, PodController, PodManifest, PodStatus};
@@ -15,17 +16,34 @@ use pod::{Event, Method, Pod, PodController, PodManifest, PodStatus};
// Mock LLM Client
// ---------------------------------------------------------------------------
/// One scripted mock response.
#[derive(Clone)]
enum MockResponse {
/// Emit the events and let the stream terminate naturally.
Complete(Vec<LlmEvent>),
/// Emit the events and then pend forever so the Worker blocks on
/// `stream.next()` — used to exercise the Cancel/Pause path while a
/// turn is actively in flight.
Hang(Vec<LlmEvent>),
}
#[derive(Clone)]
struct MockClient {
responses: Arc<Vec<Vec<LlmEvent>>>,
responses: Arc<Vec<MockResponse>>,
call_count: Arc<AtomicUsize>,
captured: Arc<Mutex<Vec<Request>>>,
}
impl MockClient {
fn new(events: Vec<LlmEvent>) -> Self {
Self::sequential(vec![MockResponse::Complete(events)])
}
/// Script multiple sequential responses. The Nth call to `stream()`
/// returns the Nth entry.
fn sequential(responses: Vec<MockResponse>) -> Self {
Self {
responses: Arc::new(vec![events]),
responses: Arc::new(responses),
call_count: Arc::new(AtomicUsize::new(0)),
captured: Arc::new(Mutex::new(Vec::new())),
}
@@ -56,9 +74,18 @@ impl LlmClient for MockClient {
message: "No more responses".into(),
});
}
let events = self.responses[count].clone();
let stream = futures::stream::iter(events.into_iter().map(Ok));
Ok(Box::pin(stream))
let response = self.responses[count].clone();
let (events, hang) = match response {
MockResponse::Complete(e) => (e, false),
MockResponse::Hang(e) => (e, true),
};
let iter = futures::stream::iter(events.into_iter().map(Ok));
if hang {
let pending = futures::stream::pending::<Result<LlmEvent, ClientError>>();
Ok(Box::pin(iter.chain(pending)))
} else {
Ok(Box::pin(iter))
}
}
}
@@ -539,3 +566,328 @@ async fn socket_invalid_method_returns_error() {
assert!(saw_error, "should see error for invalid method");
}
// ---------------------------------------------------------------------------
// Pause / Resume / Paused→Run
// ---------------------------------------------------------------------------
/// Tool that pends forever when called. Used to park a turn between
/// the ToolCall being committed to history and its ToolResult being
/// produced, so a `Method::Pause` leaves an orphan `tool_use` behind.
struct HangingTool;
#[async_trait]
impl Tool for HangingTool {
async fn execute(&self, _input: &str) -> Result<ToolOutput, ToolError> {
std::future::pending::<()>().await;
unreachable!()
}
}
fn hanging_tool_definition(name: &'static str) -> ToolDefinition {
Arc::new(move || {
(
ToolMeta::new(name)
.description("test-only tool that pends forever")
.input_schema(serde_json::json!({"type": "object"})),
Arc::new(HangingTool) as Arc<dyn Tool>,
)
})
}
async fn drain_until<F: FnMut(&Event) -> bool>(
rx: &mut tokio::sync::broadcast::Receiver<Event>,
timeout: std::time::Duration,
mut done: F,
) -> bool {
let deadline = tokio::time::Instant::now() + timeout;
loop {
tokio::select! {
ev = rx.recv() => {
match ev {
Ok(e) => { if done(&e) { return true; } }
Err(_) => return false,
}
}
_ = tokio::time::sleep_until(deadline) => return false,
}
}
}
/// Pause mid-stream, then Resume: status round-trips Running →
/// Paused → Running → Idle, and the final history contains exactly
/// one user turn plus the assistant reply produced by the resume call.
#[tokio::test]
async fn pause_then_resume_transitions_and_preserves_history_consistency() {
// Response 1: hang after opening a text block (no stop / completed),
// so the Worker is parked inside the stream read and `cancel_rx`
// races it cleanly on Method::Pause.
let hang = MockResponse::Hang(vec![
LlmEvent::text_block_start(0),
LlmEvent::text_delta(0, "partial..."),
]);
// Response 2: a clean assistant reply delivered on Resume.
let ok = MockResponse::Complete(vec![
LlmEvent::text_block_start(0),
LlmEvent::text_delta(0, "resumed output"),
LlmEvent::text_block_stop(0, None),
LlmEvent::Status(StatusEvent {
status: ResponseStatus::Completed,
}),
]);
let client = MockClient::sequential(vec![hang, ok]);
let pod = make_pod(client).await;
let handle = spawn_controller(pod).await;
let mut rx = handle.subscribe();
handle
.send(Method::Run {
input: "hello".into(),
})
.await
.unwrap();
// Wait for the partial text_delta to confirm the first stream is
// live before we pause.
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |e| matches!(
e,
Event::TextDelta { .. }
))
.await,
"text_delta should arrive before pause"
);
handle.send(Method::Pause).await.unwrap();
// The controller emits RunEnd { Paused } when the
// WorkerError::Cancelled is translated under pause_requested.
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |e| matches!(
e,
Event::RunEnd {
result: protocol::RunResult::Paused
}
))
.await,
"expected RunEnd::Paused after Pause"
);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
assert_eq!(handle.shared_state.get_status(), PodStatus::Paused);
handle.send(Method::Resume).await.unwrap();
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |e| matches!(
e,
Event::RunEnd {
result: protocol::RunResult::Finished
}
))
.await,
"expected RunEnd::Finished after Resume"
);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
assert_eq!(handle.shared_state.get_status(), PodStatus::Idle);
// History consistency: exactly [user "hello", assistant
// "resumed output"]. No artifacts from the aborted stream
// (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();
assert_eq!(
roles,
vec!["user", "assistant"],
"history = user + assistant only; got {items:?}"
);
let assistant_text = items[1]["content"]
.as_array()
.and_then(|parts| parts.iter().filter_map(|p| p["text"].as_str()).next())
.unwrap_or("");
assert_eq!(assistant_text, "resumed output");
let has_tool_call = items
.iter()
.any(|i| i["type"].as_str() == Some("tool_call"));
assert!(!has_tool_call, "no orphan tool_call in history");
}
/// Paused with an orphan `tool_use` in history + a fresh `Method::Run`
/// must produce a wire-valid next LLM request: the orphan is closed
/// with a synthetic `tool_result`, a system note is inserted, and the
/// new user input is appended.
#[tokio::test]
async fn paused_then_run_closes_orphan_tool_use_for_next_request() {
// Response 1: emit a tool_use block (complete with stop) targeting
// our hanging tool. The Worker commits the ToolCall to history,
// then parks inside `execute_tools` waiting on the tool — which is
// where Method::Pause catches it.
let tool_name = "HangyTool";
let first = MockResponse::Complete(vec![
LlmEvent::tool_use_start(0, "call_orphan", tool_name),
LlmEvent::tool_input_delta(0, "{}"),
LlmEvent::tool_use_stop(0),
LlmEvent::Status(StatusEvent {
status: ResponseStatus::Completed,
}),
]);
// Response 2: ordinary completion after the Paused→Run transition.
let second = MockResponse::Complete(vec![
LlmEvent::text_block_start(0),
LlmEvent::text_delta(0, "ok"),
LlmEvent::text_block_stop(0, None),
LlmEvent::Status(StatusEvent {
status: ResponseStatus::Completed,
}),
]);
let client = MockClient::sequential(vec![first, second]);
let client_for_assert = client.clone();
let mut pod = make_pod(client).await;
pod.worker_mut()
.register_tool(hanging_tool_definition(tool_name));
let handle = spawn_controller(pod).await;
let mut rx = handle.subscribe();
handle
.send(Method::Run {
input: "first".into(),
})
.await
.unwrap();
// Wait for ToolCallDone — the ToolCall is committed to history
// right before the Worker enters tool execution and pends.
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |e| matches!(
e,
Event::ToolCallDone { .. }
))
.await,
"tool_call_done should arrive before pause"
);
handle.send(Method::Pause).await.unwrap();
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |e| matches!(
e,
Event::RunEnd {
result: protocol::RunResult::Paused
}
))
.await,
"expected RunEnd::Paused"
);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
assert_eq!(handle.shared_state.get_status(), PodStatus::Paused);
// 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 {
input: "new request".into(),
})
.await
.unwrap();
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |e| matches!(
e,
Event::RunEnd {
result: protocol::RunResult::Finished
}
))
.await,
"expected RunEnd::Finished after Paused→Run"
);
// The second LLM request carries the closure chain. Walk its items
// and assert the invariants — order matters for wire correctness.
let requests = client_for_assert.captured_requests();
assert_eq!(requests.len(), 2, "two LLM calls expected");
let items = &requests[1].items;
// Find the ToolCall and ensure the immediately-subsequent
// ToolResult (if any) carries the synthetic summary.
let mut saw_synthetic_tool_result = false;
let mut saw_interruption_note = false;
let mut saw_new_user = false;
for item in items {
match item {
llm_worker::Item::ToolResult {
call_id, summary, ..
} if call_id == "call_orphan" => {
assert_eq!(summary, "[Interrupted by user]");
saw_synthetic_tool_result = true;
}
llm_worker::Item::Message { role, content, .. }
if *role == llm_worker::Role::System =>
{
let text: String = content.iter().map(|p| p.as_text()).collect();
if text.contains("interrupted by the user") {
saw_interruption_note = true;
}
}
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;
}
}
_ => {}
}
}
assert!(
saw_synthetic_tool_result,
"synthetic tool_result for orphan missing in 2nd request items: {items:?}"
);
assert!(
saw_interruption_note,
"system interruption note missing in 2nd request items: {items:?}"
);
assert!(
saw_new_user,
"new user message missing in 2nd request items: {items:?}"
);
// 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 sys_idx = idx(&|i| match i {
llm_worker::Item::Message {
role: llm_worker::Role::System,
content,
..
} => content
.iter()
.map(|p| p.as_text())
.collect::<String>()
.contains("interrupted by the user"),
_ => false,
});
let user_idx = idx(&|i| match i {
llm_worker::Item::Message {
role: llm_worker::Role::User,
content,
..
} => content
.iter()
.map(|p| p.as_text())
.collect::<String>()
.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");
}