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

refactor: PodControllerの構造のリファクタリング

Browse Source
This commit is contained in:
Keisuke Hirata 2026-05-13 06:07:38 +09:00
parent cb24586362
commit 8e50a9583a
No known key found for this signature in database
3 changed files with 525 additions and 592 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
TODO.md
View File

@ -9,7 +9,6 @@
- Pod: 任意ターンからの Fork（複数ターン巻き戻しを汎用化） → [tickets/pod-session-fork.md](tickets/pod-session-fork.md)
- Pod: 子→親の TurnEnded/Errored callback を親由来ターンのみに絞る → [tickets/pod-parent-turn-callback.md](tickets/pod-parent-turn-callback.md)
- Pod: セッションログをバックエンドにした Pod 単位の永続化 → [tickets/pod-persistent-state.md](tickets/pod-persistent-state.md)
- Pod: イベントハンドラからターン起動を分離（描画ループ式 dispatch） → [tickets/pod-event-loop-dispatch.md](tickets/pod-event-loop-dispatch.md)
- 永続化層のセマンティック整理 → [tickets/persistence-semantics.md](tickets/persistence-semantics.md)
- Exchange / Turn / Call セマンティクス整理 → [tickets/exchange-turn-call-semantics.md](tickets/exchange-turn-call-semantics.md)
- llm-worker のエラー耐性

655
crates/pod/src/controller.rs
View File

@ -1,4 +1,4 @@
use std::path::Path;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use llm_worker::WorkerError;
@ -19,7 +19,7 @@ use crate::spawn::comm_tools::{
use crate::spawn::registry::SpawnedPodRegistry;
use crate::spawn::tool::spawn_pod_tool;
use protocol::{
AlertLevel, AlertSource, ErrorCode, Event, Method, PodStatus, RunResult, TurnResult,
AlertLevel, AlertSource, ErrorCode, Event, Method, PodStatus, RunResult, Segment, TurnResult,
};
fn is_system_message_item(item: &Item) -> bool {
@ -94,6 +94,17 @@ async fn finish_controller_run<C, St>(
pod.spawn_post_run_memory_jobs();
}
/// Pending turn launch staged by an event handler for the next outer-loop
/// iteration. Each variant carries the input needed by the corresponding
/// `Pod::*` entry point — `RunForNotification` carries none because
/// `pod.run_for_notification()` drains the NotifyBuffer on its own.
enum PendingRun {
Run(Vec<Segment>),
InterruptAndRun(Vec<Segment>),
RunForNotification,
Resume,
}
// ---------------------------------------------------------------------------
// PodController — actor that owns a Pod
// ---------------------------------------------------------------------------
@ -111,35 +122,23 @@ impl PodController {
C: LlmClient + Clone + 'static,
St: Store + Clone + 'static,
{
// === 1. Initialization (channels / RuntimeDir / pod-immutable
// snapshots / SpawnedPodRegistry / alerter attach /
// bash-output scope) ===
let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
let (method_tx, mut method_rx) = mpsc::channel::<Method>(32);
let (method_tx, method_rx) = mpsc::channel::<Method>(32);
let (event_tx, _) = broadcast::channel::<Event>(256);
let alerter = Alerter::new(event_tx.clone());
// Runtime directory is created before tool registration because
// the spawn-tool factories need its socket path, and before the
// initial status/history writes because those writes consume the
// greeting we build after registration is complete.
// initial status/history writes consume the greeting we build
// after registration is complete.
let runtime_dir =
Arc::new(RuntimeDir::create(runtime_base, &pod.manifest().pod.name).await?);
// Snapshot pod-immutable values needed for tool factories so the
// mutable worker borrow below doesn't conflict with reads on `pod`.
let scope_handle = pod.scope().clone();
let pwd_for_tools = pod.pwd().to_path_buf();
let spawner_name = pod.manifest().pod.name.clone();
let spawner_model = pod.manifest().model.clone();
let memory_config = pod.manifest().memory.clone();
// Parent callback socket (this Pod's own parent, used for
// `PodEvent` upward reports). `None` for top-level Pods.
let self_parent_socket = pod.callback_socket().cloned();
// `SpawnedPodRegistry` is shared between the Pod-orchestration
// tools (registered below) and the main loop's `PodEvent`
// handler (added later in this function), so hoist its creation
// above the worker-borrow block.
let spawner_socket = runtime_dir.socket_path();
let spawned_registry = SpawnedPodRegistry::new(runtime_dir.clone());
// Hand the alerter to the Pod so internal operations (compaction,
@ -153,11 +152,7 @@ impl PodController {
// Bash spills long outputs to a per-pod subdir under the runtime
// dir. Push a recursive `allow(Read)` for that path into the
// Pod's runtime scope so the agent can `Read` saved files
// without polluting the workspace. The Pod's SharedScope is the
// single source of truth — every ScopedFs (builtin tools,
// fs_view, compact worker) reads from it, and any future scope
// mutation (SpawnPod-style revoke, future GrantScope)
// propagates through it.
// without polluting the workspace.
let bash_output_dir = runtime_dir.path().join("bash-output");
std::fs::create_dir_all(&bash_output_dir).map_err(|e| {
std::io::Error::other(format!(
@ -172,15 +167,97 @@ impl PodController {
}])
.map_err(std::io::Error::other)?;
// Stashed during tool registration below so we can attach a
// `PodFsView` to the shared state once the latter exists.
let fs_for_view: tools::ScopedFs;
let task_store = pod.task_store();
let session_id_for_usage = pod.session_id().to_string();
// === 2. Worker event bridge wiring ===
wire_event_bridges_on_worker(&mut pod, &event_tx, &alerter);
let scope_change_sink = pod.scope_change_sink();
// === 3. Tool registration (builtin / memory / spawn-orchestration) ===
let fs_for_view = register_pod_tools(
&mut pod,
bash_output_dir,
runtime_dir.socket_path(),
runtime_base.to_path_buf(),
spawned_registry.clone(),
);
// Register event bridge callbacks on the worker
// Materialise pending tool factories so the greeting reflects
// the actual registered set instead of a hand-maintained mirror.
pod.worker().tool_server_handle().flush_pending();
// === 4. Initial runtime files + PodSharedState + PodHandle +
// SocketServer ===
let manifest_toml = toml::to_string_pretty(pod.manifest()).unwrap_or_default();
let greeting = build_greeting(&pod);
let shared_state = Arc::new(PodSharedState::new(
pod.manifest().pod.name.clone(),
pod.session_id(),
manifest_toml.clone(),
greeting,
));
shared_state.update_history(pod.worker().history().to_vec());
shared_state.set_user_segments(pod.user_segments().to_vec());
shared_state.set_fs_view(crate::fs_view::PodFsView::new(fs_for_view));
shared_state.set_workflows(
pod.workflow_completions()
.into_iter()
.map(|slug| crate::shared_state::WorkflowCandidate { slug })
.collect(),
);
shared_state.set_knowledge(
pod.knowledge_completions()
.into_iter()
.map(|slug| crate::shared_state::KnowledgeCandidate { slug })
.collect(),
);
runtime_dir.write_manifest(&manifest_toml).await?;
runtime_dir.write_status(&shared_state).await?;
runtime_dir.write_history(&shared_state).await?;
let handle = PodHandle {
method_tx,
event_tx: event_tx.clone(),
shared_state: shared_state.clone(),
runtime_dir: runtime_dir.clone(),
alerter: alerter.clone(),
};
let socket_server = SocketServer::start(&handle).await?;
// === 5. controller_loop ===
// Clone cancel sender and notification buffer before moving pod
// into the controller task so the in-flight turn can be reached
// via these handles while pod itself is borrowed by drive_turn.
let cancel_tx = pod.worker_mut().cancel_sender();
let notify_buffer = pod.notify_buffer_handle();
tokio::spawn(controller_loop(
pod,
method_rx,
event_tx,
shared_state,
runtime_dir,
cancel_tx,
notify_buffer,
self_parent_socket,
spawner_name,
spawned_registry,
shutdown_tx,
socket_server,
));
Ok((handle, shutdown_rx))
}
}
/// Wire the per-event broadcast bridges on the Pod's Worker. Each callback
/// re-publishes a worker-level signal as a `protocol::Event` on `event_tx`
/// so subscribers (TUI, socket clients) get a single typed stream.
fn wire_event_bridges_on_worker<C, St>(
pod: &mut Pod<C, St>,
event_tx: &broadcast::Sender<Event>,
alerter: &Alerter,
) where
C: LlmClient + Clone + 'static,
St: Store + Clone + 'static,
{
let worker = pod.worker_mut();
@ -215,9 +292,8 @@ impl PodController {
let tx = event_tx.clone();
worker.on_thinking_block(move |block| {
// Start fires unconditionally so the TUI can show
// "Thinking..." even when the provider doesn't emit
// plaintext deltas.
// Start fires unconditionally so the TUI can show "Thinking..."
// even when the provider doesn't emit plaintext deltas.
let _ = tx.send(Event::ThinkingStart);
let tx_d = tx.clone();
block.on_delta(move |text| {
@ -296,48 +372,66 @@ impl PodController {
let _ = tx.send(Event::SystemMessage { item: value });
}
});
}
// Register the builtin file-manipulation tools (Read / Write /
// Edit / Glob / Grep / Bash). `ScopedFs` carries the pod-
// lifetime scope/pwd; `Tracker` is session-scoped — a fresh
// instance per controller spawn ensures state from a previous
// process lifetime cannot be reused after a resume. The tracker
// is also handed to the Pod itself so Pod-level operations (e.g.
// context compaction) can ask which files the agent has been
// touching.
//
// The Pod's SharedScope (already augmented with the
// bash-output Read rule above) is the single source of
// truth — every ScopedFs (builtin tools, fs_view, compact
// worker) reads from it, and any future scope mutation
// (SpawnPod-style revoke, future GrantScope) propagates
// through it.
let fs =
tools::ScopedFs::with_shared_scope(scope_handle.clone(), pwd_for_tools.clone());
/// Register the builtin file-manipulation tools, optional memory tools,
/// and the Pod-orchestration tools (SpawnPod + comm) on the Pod's
/// Worker. Returns the `ScopedFs` clone used to attach a `PodFsView` to
/// the shared state.
fn register_pod_tools<C, St>(
pod: &mut Pod<C, St>,
bash_output_dir: PathBuf,
spawner_socket: PathBuf,
runtime_base: PathBuf,
spawned_registry: Arc<SpawnedPodRegistry>,
) -> tools::ScopedFs
where
C: LlmClient + Clone + 'static,
St: Store + Clone + 'static,
{
// Pod-immutable snapshots taken before the mutable worker borrow
// below so the worker borrow doesn't conflict with reads on `pod`.
let scope_handle = pod.scope().clone();
let pwd = pod.pwd().to_path_buf();
let task_store = pod.task_store();
let session_id_for_usage = pod.session_id().to_string();
let scope_change_sink = pod.scope_change_sink();
let memory_config = pod.manifest().memory.clone();
let spawner_name = pod.manifest().pod.name.clone();
let spawner_model = pod.manifest().model.clone();
let self_parent_socket = pod.callback_socket().cloned();
let worker = pod.worker_mut();
// The Pod's SharedScope (already augmented with the bash-output
// Read rule by the caller) is the single source of truth — every
// ScopedFs (builtin tools, fs_view, compact worker) reads from it,
// and any future scope mutation (SpawnPod-style revoke, future
// GrantScope) propagates through it.
let fs = tools::ScopedFs::with_shared_scope(scope_handle.clone(), pwd.clone());
let tracker = tools::Tracker::new();
// The same ScopedFs also powers the IPC `ListCompletions`
// query — keep a clone for the FS view we attach below,
// since the tools consume `fs` itself.
fs_for_view = fs.clone();
// Same ScopedFs also powers the IPC `ListCompletions` query — keep
// a clone for the FS view we attach below, since the tools consume
// `fs` itself.
let fs_for_view = fs.clone();
worker.register_tools(tools::builtin_tools(
fs,
tracker.clone(),
task_store.clone(),
task_store,
bash_output_dir,
));
// Memory subsystem opt-in. When `[memory]` is present in
// the manifest, register the memory-specific Read/Write/Edit
// tools that target `<workspace>/memory/` and
// `<workspace>/knowledge/` with their built-in linter. The
// companion deny rules on the generic CRUD scope were
// already applied during `Pod::from_manifest`.
// Memory subsystem opt-in. When `[memory]` is present in the
// manifest, register the memory-specific Read/Write/Edit tools that
// target `<workspace>/memory/` and `<workspace>/knowledge/` with
// their built-in linter. Companion deny rules on the generic CRUD
// scope were already applied during `Pod::from_manifest`.
if let Some(mem) = memory_config.as_ref() {
let layout = memory::WorkspaceLayout::resolve(mem, &pwd_for_tools);
let layout = memory::WorkspaceLayout::resolve(mem, &pwd);
let query_cfg = memory::tool::QueryConfig::from(mem);
worker.register_tool(memory::tool::read_tool_with_usage(
layout.clone(),
session_id_for_usage.clone(),
session_id_for_usage,
));
worker.register_tool(memory::tool::write_tool(layout.clone()));
worker.register_tool(memory::tool::edit_tool(layout.clone()));
@ -345,133 +439,67 @@ impl PodController {
worker.register_tool(memory::tool::knowledge_query_tool(layout, query_cfg));
}
// Pod-orchestration tools (SpawnPod + the four comm tools)
// share the Pod-scoped `SpawnedPodRegistry` hoisted above
// (also consumed by the main loop's `PodEvent` handler).
// Pod-orchestration tools (SpawnPod + the four comm tools) share
// the Pod-scoped `SpawnedPodRegistry` (also consumed by the main
// loop's `PodEvent` handler).
worker.register_tool(spawn_pod_tool(
spawner_name.clone(),
spawner_socket.clone(),
runtime_base.to_path_buf(),
pwd_for_tools,
spawner_name,
spawner_socket,
runtime_base,
pwd,
spawned_registry.clone(),
self_parent_socket.clone(),
spawner_model.clone(),
scope_handle.clone(),
scope_change_sink.clone(),
self_parent_socket,
spawner_model,
scope_handle,
scope_change_sink,
));
worker.register_tool(send_to_pod_tool(spawned_registry.clone()));
worker.register_tool(read_pod_output_tool(spawned_registry.clone()));
worker.register_tool(stop_pod_tool(spawned_registry.clone()));
worker.register_tool(list_pods_tool(spawned_registry.clone()));
worker.register_tool(list_pods_tool(spawned_registry));
pod.attach_tracker(tracker);
fs_for_view
}
// Materialise pending tool factories so the greeting reflects
// the actual registered set instead of a hand-maintained mirror.
pod.worker().tool_server_handle().flush_pending();
/// Idle/Paused event loop. Each iteration either fires a staged
/// `PendingRun` (delegating to [`drive_turn`] for the Running phase) or
/// waits for the next `Method`. Method handlers stop at "update state +
/// stage `pending`"; the loop's top-of-iteration block owns the
/// status-flip → run → finish sequence so it lives in exactly one
/// place.
#[allow(clippy::too_many_arguments)]
async fn controller_loop<C, St>(
mut pod: Pod<C, St>,
mut method_rx: mpsc::Receiver<Method>,
event_tx: broadcast::Sender<Event>,
shared_state: Arc<PodSharedState>,
runtime_dir: Arc<RuntimeDir>,
cancel_tx: mpsc::Sender<()>,
notify_buffer: NotifyBuffer,
self_parent_socket: Option<PathBuf>,
spawner_name: String,
spawned_registry: Arc<SpawnedPodRegistry>,
shutdown_tx: oneshot::Sender<()>,
socket_server: SocketServer,
) where
C: LlmClient + Clone + 'static,
St: Store + Clone + 'static,
{
// Hold socket server alive for the lifetime of the controller task.
let _socket_server = socket_server;
// Greeting + initial runtime files now that the tool list is final.
let manifest_toml = toml::to_string_pretty(pod.manifest()).unwrap_or_default();
let greeting = build_greeting(&pod);
let shared_state = Arc::new(PodSharedState::new(
pod.manifest().pod.name.clone(),
pod.session_id(),
manifest_toml.clone(),
greeting,
));
shared_state.update_history(pod.worker().history().to_vec());
shared_state.set_user_segments(pod.user_segments().to_vec());
shared_state.set_fs_view(crate::fs_view::PodFsView::new(fs_for_view));
shared_state.set_workflows(
pod.workflow_completions()
.into_iter()
.map(|slug| crate::shared_state::WorkflowCandidate { slug })
.collect(),
);
shared_state.set_knowledge(
pod.knowledge_completions()
.into_iter()
.map(|slug| crate::shared_state::KnowledgeCandidate { slug })
.collect(),
);
runtime_dir.write_manifest(&manifest_toml).await?;
runtime_dir.write_status(&shared_state).await?;
runtime_dir.write_history(&shared_state).await?;
let handle = PodHandle {
method_tx,
event_tx: event_tx.clone(),
shared_state: shared_state.clone(),
runtime_dir: runtime_dir.clone(),
alerter: alerter.clone(),
};
// Start socket server (lives as a background task, cleaned up on
// drop via RuntimeDir). Kept alive by moving it into the
// controller task so it drops when that task ends.
let _socket_server = SocketServer::start(&handle).await?;
// Clone cancel sender and notification buffer before moving pod
// into the controller task so the main loop can route
// `Method::Notify` into the buffer even while `pod` is held by
// an in-flight `run_for_notification` / `run` future.
let cancel_tx = pod.worker_mut().cancel_sender();
let notify_buffer = pod.notify_buffer_handle();
tokio::spawn(async move {
// Hold socket server alive for the lifetime of the controller task
let _socket_server = _socket_server;
let mut pending: Option<PendingRun> = None;
loop {
let method = match method_rx.recv().await {
Some(m) => m,
None => break,
};
match method {
Method::Run { input } => {
let status_before = shared_state.get_status();
if status_before == PodStatus::Running {
let _ = event_tx.send(Event::Error {
code: ErrorCode::AlreadyRunning,
message: "Pod is already executing a turn".into(),
});
continue;
}
let was_paused = status_before == PodStatus::Paused;
if let Err(e) = pod.validate_workflow_invocations(&input) {
let _ = event_tx.send(Event::Error {
code: ErrorCode::InvalidRequest,
message: e.to_string(),
});
continue;
}
// Broadcast the accepted user message so every
// subscriber (including the submitter) can
// render the turn header + user line from a
// single source of truth. shared_state's
// `user_segments` is re-synced from `pod` after
// the run completes, so we don't push here.
let _ = event_tx.send(Event::UserMessage {
segments: input.clone(),
});
set_controller_status(
&shared_state,
&runtime_dir,
&event_tx,
PodStatus::Running,
)
.await;
let run_future = async {
if was_paused {
pod.interrupt_and_run(input).await
} else {
pod.run(input).await
}
};
let (new_status, shutdown) = run_with_cancel_support(
run_future,
// Top-of-iteration: if an event handler staged a run, fire it
// here so the status flip → drive_turn → finish sequence lives
// in one place, regardless of which Method caused it.
if let Some(run) = pending.take() {
set_controller_status(&shared_state, &runtime_dir, &event_tx, PodStatus::Running).await;
let (new_status, shutdown) = match run {
PendingRun::Run(input) => {
drive_turn(
pod.run(input),
&mut method_rx,
&event_tx,
&cancel_tx,
@ -481,47 +509,24 @@ impl PodController {
&spawner_name,
&spawned_registry,
)
.await;
finish_controller_run(
&mut pod,
&shared_state,
&runtime_dir,
.await
}
PendingRun::InterruptAndRun(input) => {
drive_turn(
pod.interrupt_and_run(input),
&mut method_rx,
&event_tx,
new_status,
)
.await;
if shutdown {
let _ = event_tx.send(Event::Shutdown);
break;
}
}
Method::Notify { message } => {
let _ = event_tx.send(Event::Notify {
message: message.clone(),
});
pod.push_notify(message);
let status = shared_state.get_status();
if status != PodStatus::Idle {
// RUNNING / Paused: the buffer push is the
// entire operation; an in-flight turn (or the
// next Resume/Run) will drain the buffer
// at its next pre_llm_request.
continue;
}
// IDLE: auto-start a turn so the LLM sees the
// buffered notification(s) without a human Run.
set_controller_status(
&cancel_tx,
&shared_state,
&runtime_dir,
&event_tx,
PodStatus::Running,
&notify_buffer,
self_parent_socket.as_ref(),
&spawner_name,
&spawned_registry,
)
.await;
let (new_status, shutdown) = run_with_cancel_support(
.await
}
PendingRun::RunForNotification => {
drive_turn(
pod.run_for_notification(),
&mut method_rx,
&event_tx,
@ -532,21 +537,88 @@ impl PodController {
&spawner_name,
&spawned_registry,
)
.await;
finish_controller_run(
&mut pod,
&shared_state,
&runtime_dir,
.await
}
PendingRun::Resume => {
drive_turn(
pod.resume(),
&mut method_rx,
&event_tx,
new_status,
&cancel_tx,
&shared_state,
&notify_buffer,
self_parent_socket.as_ref(),
&spawner_name,
&spawned_registry,
)
.await
}
};
finish_controller_run(&mut pod, &shared_state, &runtime_dir, &event_tx, new_status)
.await;
if shutdown {
let _ = event_tx.send(Event::Shutdown);
break;
}
continue;
}
let method = match method_rx.recv().await {
Some(m) => m,
None => break,
};
match method {
Method::Run { input } => {
let status_before = shared_state.get_status();
if status_before == PodStatus::Running {
// Defensive: the inner select! inside drive_turn
// already rejects `Run` while a turn is live, so
// this branch is only reachable across a race window
// around status flips.
let _ = event_tx.send(Event::Error {
code: ErrorCode::AlreadyRunning,
message: "Pod is already executing a turn".into(),
});
continue;
}
if let Err(e) = pod.validate_workflow_invocations(&input) {
let _ = event_tx.send(Event::Error {
code: ErrorCode::InvalidRequest,
message: e.to_string(),
});
continue;
}
// Broadcast the accepted user message so every
// subscriber (including the submitter) can render the
// turn header + user line from a single source of
// truth. shared_state's `user_segments` is re-synced
// from `pod` after the run completes, so we don't push
// here.
let _ = event_tx.send(Event::UserMessage {
segments: input.clone(),
});
pending = Some(if status_before == PodStatus::Paused {
PendingRun::InterruptAndRun(input)
} else {
PendingRun::Run(input)
});
}
Method::Notify { message } => {
let _ = event_tx.send(Event::Notify {
message: message.clone(),
});
pod.push_notify(message);
// RUNNING / Paused: the buffer push is the entire
// operation; an in-flight turn (or the next
// Resume/Run) will drain it at its next
// pre_llm_request. IDLE: auto-start a turn so the LLM
// sees the buffered notification(s) without a human
// Run.
if shared_state.get_status() == PodStatus::Idle {
pending = Some(PendingRun::RunForNotification);
}
}
Method::Resume => {
@ -557,40 +629,7 @@ impl PodController {
});
continue;
}
set_controller_status(
&shared_state,
&runtime_dir,
&event_tx,
PodStatus::Running,
)
.await;
let (new_status, shutdown) = run_with_cancel_support(
pod.resume(),
&mut method_rx,
&event_tx,
&cancel_tx,
&shared_state,
&notify_buffer,
self_parent_socket.as_ref(),
&spawner_name,
&spawned_registry,
)
.await;
finish_controller_run(
&mut pod,
&shared_state,
&runtime_dir,
&event_tx,
new_status,
)
.await;
if shutdown {
let _ = event_tx.send(Event::Shutdown);
break;
}
pending = Some(PendingRun::Resume);
}
Method::Cancel => {
@ -602,11 +641,9 @@ impl PodController {
Method::Pause => {
// Already paused → idempotent no-op. Otherwise the
// Pod is Idle (Running turns go through
// `run_with_cancel_support`, not this outer match), so
// there is nothing to pause.
let status = shared_state.get_status();
if status != PodStatus::Paused {
// Pod is Idle (Running turns go through `drive_turn`,
// not this outer match), so there is nothing to pause.
if shared_state.get_status() != PodStatus::Paused {
let _ = event_tx.send(Event::Error {
code: ErrorCode::NotRunning,
message: "Pod is not running".into(),
@ -620,18 +657,18 @@ impl PodController {
}
// GetHistory / ListCompletions are handled at the socket
// layer (direct response). If they somehow reach the
// controller, ignore them.
// layer (direct response). If they reach the controller,
// ignore them.
Method::GetHistory | Method::ListCompletions { .. } => {}
Method::PodEvent(event) => {
// Echo the received event to all subscribers so
// every client sees the input that drove any
// following auto-kicked turn.
// Echo the received event to all subscribers so every
// client sees the input that drove any following
// auto-kicked turn.
let _ = event_tx.send(Event::PodEvent(event.clone()));
// (1) system side effects — idempotent and
// tolerant of out-of-order delivery (e.g.
// `TurnEnded` arriving after `ShutDown`).
// (1) system side effects — idempotent and tolerant of
// out-of-order delivery (e.g. `TurnEnded` arriving
// after `ShutDown`).
crate::ipc::event::apply_event_side_effects(
&event,
&spawned_registry,
@ -639,50 +676,16 @@ impl PodController {
&self_parent_socket,
)
.await;
// (2) render a one-line summary and push it
// into the notification buffer; the next LLM
// request will inject it as a system message
// via `PodInterceptor::pre_llm_request`.
let text = crate::ipc::event::render_event(&event);
pod.push_notify(text);
// (2) render a one-line summary and push it into the
// notification buffer; the next LLM request will
// inject it as a system message via
// `PodInterceptor::pre_llm_request`.
pod.push_notify(crate::ipc::event::render_event(&event));
// Auto-kick a turn if the Pod is idle so the
// notification is not stranded. Matches the
// `Method::Notify` idle path.
if shared_state.get_status() == PodStatus::Idle {
set_controller_status(
&shared_state,
&runtime_dir,
&event_tx,
PodStatus::Running,
)
.await;
let (new_status, shutdown) = run_with_cancel_support(
pod.run_for_notification(),
&mut method_rx,
&event_tx,
&cancel_tx,
&shared_state,
&notify_buffer,
self_parent_socket.as_ref(),
&spawner_name,
&spawned_registry,
)
.await;
finish_controller_run(
&mut pod,
&shared_state,
&runtime_dir,
&event_tx,
new_status,
)
.await;
if shutdown {
let _ = event_tx.send(Event::Shutdown);
break;
}
pending = Some(PendingRun::RunForNotification);
}
}
}
@ -693,12 +696,11 @@ impl PodController {
// staging writes and consolidation cleanups are not abandoned.
pod.wait_for_memory_jobs().await;
// Report upward that this Pod is stopping before the
// controller task exits. Awaited (not fire-and-forget):
// after `shutdown_tx.send` the process may exit quickly,
// and a spawned task would be killed mid-send. The
// `connect_and_send` helper enforces a 5 s timeout so a
// stuck parent cannot block process exit indefinitely.
// Report upward that this Pod is stopping before the controller
// task exits. Awaited (not fire-and-forget): after `shutdown_tx.send`
// the process may exit quickly, and a spawned task would be killed
// mid-send. The `connect_and_send` helper enforces a 5 s timeout so
// a stuck parent cannot block process exit indefinitely.
if let Some(parent) = self_parent_socket.as_ref() {
if let Err(e) = crate::ipc::event::send_pod_event(
parent,
@ -713,29 +715,26 @@ impl PodController {
}
let _ = shutdown_tx.send(());
});
Ok((handle, shutdown_rx))
}
}
/// Runs a Pod future while concurrently processing incoming methods.
///
/// Returns `(final_status, shutdown_requested)`.
/// Drives a Pod future (one in-flight turn) while concurrently
/// processing incoming methods through an inner select! arm. Returns
/// `(final_status, shutdown_requested)`.
///
/// `parent_socket` / `self_name` drive upward `PodEvent` reports
/// (`TurnEnded` on a clean Finished, `Errored` on a worker failure).
/// `None` parent skips the send (top-level Pod). Transient method
/// rejections such as `AlreadyRunning` are intentionally NOT reported
/// as `Errored` — only the worker-execution `Err` branch below fires.
async fn run_with_cancel_support<F>(
#[allow(clippy::too_many_arguments)]
async fn drive_turn<F>(
pod_future: F,
method_rx: &mut mpsc::Receiver<Method>,
event_tx: &broadcast::Sender<Event>,
cancel_tx: &mpsc::Sender<()>,
shared_state: &Arc<PodSharedState>,
notify_buffer: &NotifyBuffer,
parent_socket: Option<&std::path::PathBuf>,
parent_socket: Option<&PathBuf>,
self_name: &str,
spawned_registry: &Arc<SpawnedPodRegistry>,
) -> (PodStatus, bool)

65
tickets/pod-event-loop-dispatch.md
View File

@ -1,65 +0,0 @@
# Pod: イベントハンドラからターン起動を分離する
## 背景
`crates/pod/src/controller.rs` の controller タスクは、 outer loop の Method ハンドラ内に **イベント処理とターン起動を同居** させている。
該当する箇所:
- `Method::Run` → outer arm 内で `run_with_cancel_support(pod.run(...)).await` と `finish_controller_run.await` を呼ぶ
- `Method::Notify` (Idle) → outer arm 内で `set_controller_status(Running).await` のあと `run_with_cancel_support(pod.run_for_notification()).await` / `finish_controller_run.await`
- `Method::PodEvent` (Idle) → 同上 (副作用処理 `apply_event_side_effects.await` も同じ arm 内)
- `Method::Resume` → 同上
ハンドラ自身が「ターンを丸ごと await する長時間処理」 を抱えているため、 以下が起きる:
- 同じターン起動コード (`set_controller_status` → `run_with_cancel_support` → `finish_controller_run`) が 4 箇所に重複する
- ターン起動の発火元 (Run / Notify / PodEvent / Resume) と起動本体が分離されていないため、 起動条件の変更が複数 arm に波及する
- ターン起動に必要な前処理 (history への user message append / NotifyBuffer への push / status 遷移) も各 arm に分散している
ターン起動を outer loop の周回トップに集約し、 各イベントハンドラは「次のターンに渡す入力をその場で確定 (history append / NotifyBuffer push) + `needs_run` フラグセット」 までに留めることで、 起動コードの重複と入力経路の分散をなくす。
加えて、 `PodController::spawn` 自体が約 615 行に膨らんでおり、「組み立て (channel/runtime_dir/Worker hook 配線/ツール登録/PodHandle 構築)」と「実行ループ (controller タスク)」が単一関数に同居している。 outer loop を上記の形に書き直すタイミングで、 spawn の構造分解も同時に行う方が、 重複コードの統合先が見やすくなる。
なお、 別途観測されている「auto-kick されたターンの内部で controller がデッドロックする現象」 はこの整理では解決しない (inner select! のアーム内 await が pending する経路は構造上残る)。 根本原因の特定は別チケットの対象。
## 要件
- イベントハンドラは **「状態更新と `needs_run` フラグ立て」 まで** にとどめる。ハンドラ内で `run_with_cancel_support.await` / `finish_controller_run.await` を呼ばない
- outer loop の各周回はまず `needs_run` を評価し、 立っていればターン起動 (`run_with_cancel_support` → `finish_controller_run`) を実行してフラグを降ろす。 立っていなければ `method_rx.recv().await` で次の Method を待つ
- 既存の inner select! によるターン中の Method 並行受信は維持する。 ターン本体の借用構造 (`&mut Pod` を Worker が抱える) も変更しない
- `needs_run` を立てる契機は最低限以下を含める (= 現状 auto-kick している経路):
- `Method::Run`
- `Method::Resume`
- `Method::Notify` (Idle 時のみ)
- `Method::PodEvent` (Idle 時のみ)
- 「次のターンの起動意図と入力」 は `needs_run` を起動意図を持つ enum として表現する:
- `Run { input }` / `InterruptAndRun { input }` / `RunForNotification` / `Resume`
- `Method::Notify` / `Method::PodEvent` の Idle 経路は NotifyBuffer に push したうえで `RunForNotification` を立てる (現状の `pod.run_for_notification()` が NotifyBuffer から自動取得する挙動に乗るため、 enum に入力を載せる必要はない)
- `Worker::run` / `Pod::run` の入力受け取り API には触らない (interceptor の `on_prompt_submit` cancel / 書き換え / extras ordering の invariant に踏み込まない)
- Pause / Shutdown / Cancel はハンドラ内で完結する (フラグ化しない、 既存通り即時処理)
- Running 中に来た `Method::Notify` / `Method::PodEvent` の挙動 (NotifyBuffer に積むだけ、 副作用は実行) は変えない
- 上記改修に合わせて `PodController::spawn` を以下の責務単位に分解する:
- 初期化 (channel 群 / `RuntimeDir` / pod-immutable snapshot / `SpawnedPodRegistry` / `alerter` 装着 / bash-output scope)
- Worker への event bridge コールバック配線 (`on_turn_*` / `on_*_block` / `on_tool_result` / `on_usage` / `on_warning` / `on_history_append` 等)
- ツール登録 (builtin / memory / spawn orchestration)
- 初期ファイル書き出し + `PodSharedState` / `PodHandle` 構築 + `SocketServer` 起動
- `controller_loop` — Idle/Paused 状態の event loop + 後処理 (現 `tokio::spawn(async move { loop { ... } })` の本体)
- `drive_turn` — Running 状態の event loop。 現 `run_with_cancel_support` を改名し、 `controller_loop` と同格の関数として並べる (「cancel support」 という実装詳細名から役割ベースの名前に改める)
- 分解後の `spawn` はこの順序で各責務を呼び出す薄いフローになる。 Idle/Paused 側 (`controller_loop`) と Running 側 (`drive_turn`) の event loop が同格の関数として並ぶ形に揃える
## 完了条件
- `Method::Run` / `Method::Resume` / `Method::Notify(Idle)` / `Method::PodEvent(Idle)` のいずれも、 ターン起動が outer loop の周回トップ 1 箇所に集約されており、 ハンドラ側にはターン起動コードが残っていない
- 各イベントハンドラの async body 内に「ターン丸ごと」 や `finish_controller_run` 等の長時間 await が無い
- 既存挙動が変わらない (どの Method で auto-kick されるか、 ターン中の Cancel / Pause / Shutdown が効くか、 NotifyBuffer に積まれた内容が次ターンで反映されるか、 ターン起動順序)
- `PodController::spawn` が責務単位の関数列を順に呼ぶ薄いフローになっており、 単一関数の中に組み立てと実行ループが同居していない
- Idle/Paused 状態の event loop (`controller_loop`) と Running 状態の event loop (`drive_turn`) が同格の関数として並んでおり、 名前が役割を表している (`run_with_cancel_support` の名前は残らない)
- `crates/pod/tests/controller_test.rs` の既存テストが通る
## 範囲外
- 観測されたデッドロックの根本原因特定 (別チケット予定)
- `Pod` 構造体の `&mut self` borrow を分解して outer loop を完全 dispatcher 化する大改修 (将来検討)
- Cancel / Shutdown の専用チャネル化 (controller が固まっている時には別経路でも効かないため、 メリット薄と判断して見送り)
- `Worker::run` / `Pod::run` の入力なし API 化 (controller の dispatch を bool に圧縮するメリットに対し、 worker interceptor invariant を壊すコストが見合わない)