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

feat: client-crateの実装

Browse Source
This commit is contained in:
Hare
2026-05-10 00:57:50 +09:00
parent df0be1cd6b
commit 3def5edbdf
11 changed files with 412 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tui/Cargo.toml
+1
View File
@@ -5,6 +5,7 @@ edition.workspace = true
license.workspace = true
[dependencies]
client = { workspace = true }
protocol = { workspace = true }
ratatui = { version = "0.30.0", features = ["scrolling-regions"] }
crossterm = "0.28"
crates/tui/src/client.rs
-45
View File
@@ -1,45 +0,0 @@
use std::io;
use std::path::Path;
use protocol::stream::{JsonLineReader, JsonLineWriter};
use protocol::{Event, Method};
use tokio::net::UnixStream;
use tokio::sync::mpsc;
pub struct PodClient {
writer: JsonLineWriter<tokio::io::WriteHalf<UnixStream>>,
event_rx: mpsc::Receiver<Event>,
}
impl PodClient {
pub async fn connect(path: &Path) -> Result<Self, io::Error> {
let stream = UnixStream::connect(path).await?;
let (reader, writer) = tokio::io::split(stream);
let writer = JsonLineWriter::new(writer);
let (event_tx, event_rx) = mpsc::channel::<Event>(256);
tokio::spawn(async move {
let mut reader = JsonLineReader::new(reader);
while let Ok(Some(event)) = reader.next::<Event>().await {
if event_tx.send(event).await.is_err() {
break;
}
}
});
Ok(Self { writer, event_rx })
}
pub async fn send(&mut self, method: &Method) -> Result<(), io::Error> {
self.writer.write(method).await
}
pub fn try_next_event(&mut self) -> Option<Event> {
self.event_rx.try_recv().ok()
}
pub async fn next_event(&mut self) -> Option<Event> {
self.event_rx.recv().await
}
}
crates/tui/src/main.rs
+2 -2
View File
@@ -1,7 +1,6 @@
mod app;
mod block;
mod cache;
mod client;
mod input;
mod markdown;
mod picker;
@@ -28,8 +27,9 @@ use ratatui::Terminal;
use ratatui::backend::CrosstermBackend;
use session_store::SessionId;
use client::PodClient;
use crate::app::App;
use crate::client::PodClient;
use crate::picker::PickerOutcome;
use crate::spawn::{SpawnOutcome, SpawnReady};
crates/tui/src/spawn.rs
+23 -221
View File
@@ -14,9 +14,9 @@
use std::io;
use std::path::PathBuf;
use std::process::Stdio;
use std::time::Duration;
use client::{SpawnConfig, spawn_pod};
use crossterm::event::{self, Event as TermEvent, KeyCode, KeyEventKind, KeyModifiers};
use manifest::{
PodManifestConfig, ScopeConfig, find_project_manifest_from, load_layer, user_manifest_path,
@@ -29,11 +29,8 @@ use ratatui::text::{Line, Span};
use ratatui::widgets::Paragraph;
use ratatui::{Frame, TerminalOptions, Viewport};
use session_store::SessionId;
use tokio::process::Command;
const READY_PREFIX: &str = "INSOMNIA-READY\t";
const VIEWPORT_LINES: u16 = 6;
const READY_TIMEOUT: Duration = Duration::from_secs(20);
pub struct SpawnReady {
pub pod_name: String,
@@ -50,9 +47,7 @@ pub enum SpawnError {
Io(io::Error),
Store(session_store::StoreError),
MissingResumeScope { session_id: SessionId },
PodLaunchFailed(io::Error),
PodExitedEarly { stderr_tail: String },
Timeout,
Spawn(client::SpawnError),
}
impl std::fmt::Display for SpawnError {
@@ -64,19 +59,7 @@ impl std::fmt::Display for SpawnError {
f,
"session {session_id} has no persisted scope snapshot; refusing resume without explicit scope"
),
Self::PodLaunchFailed(e) => write!(f, "failed to launch pod: {e}"),
Self::PodExitedEarly { stderr_tail } => {
if stderr_tail.is_empty() {
write!(f, "pod exited before becoming ready")
} else {
write!(f, "pod exited before becoming ready: {stderr_tail}")
}
}
Self::Timeout => write!(
f,
"pod did not become ready within {}s",
READY_TIMEOUT.as_secs()
),
Self::Spawn(e) => write!(f, "{e}"),
}
}
}
@@ -95,6 +78,12 @@ impl From<session_store::StoreError> for SpawnError {
}
}
impl From<client::SpawnError> for SpawnError {
fn from(e: client::SpawnError) -> Self {
Self::Spawn(e)
}
}
type InlineTerminal = Terminal<CrosstermBackend<io::Stdout>>;
/// Source session for a resume run. `None` = fresh spawn (current
@@ -283,169 +272,23 @@ async fn wait_for_ready(
form: &mut Form,
overlay_toml: &str,
) -> Result<SpawnReady, SpawnError> {
let pod_bin = resolve_pod_command();
let cwd = std::env::current_dir().map_err(SpawnError::Io)?;
let pod_runtime_dir = manifest::paths::pod_runtime_dir(&form.name).ok_or_else(|| {
io::Error::new(
io::ErrorKind::NotFound,
"could not resolve runtime directory (set INSOMNIA_HOME, INSOMNIA_RUNTIME_DIR, XDG_RUNTIME_DIR, or HOME)",
)
})?;
std::fs::create_dir_all(&pod_runtime_dir).map_err(SpawnError::Io)?;
let stderr_path = pod_runtime_dir.join("stderr.log");
let stderr_file = std::fs::File::create(&stderr_path).map_err(SpawnError::Io)?;
let mut command = Command::new(&pod_bin);
command
.arg("--overlay")
.arg(overlay_toml)
.current_dir(&cwd)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::from(stderr_file))
.process_group(0);
if let Some(id) = form.resume_from {
command.arg("--session").arg(id.to_string());
}
let mut child = command.spawn().map_err(SpawnError::PodLaunchFailed)?;
// Default `kill_on_drop = false` plus `process_group(0)` makes this
// a detached Pod for TUI lifecycle purposes once startup succeeds:
// dropping the handle does not terminate it, and terminal-generated
// signals for the TUI's process group do not hit the Pod. Runtime
// state/socket files are the source of truth after that point.
let ready = match wait_for_ready_file(terminal, form, &stderr_path, &mut child).await {
Ok(ready) => ready,
Err(e) => {
let _ = child.start_kill();
let _ = child.wait().await;
return Err(e);
}
let config = SpawnConfig {
pod_name: form.name.clone(),
overlay_toml: overlay_toml.to_string(),
cwd,
resume_from: form.resume_from,
};
tokio::spawn(async move {
let _ = child.wait().await;
});
Ok(ready)
}
async fn wait_for_ready_file(
terminal: &mut InlineTerminal,
form: &mut Form,
stderr_path: &std::path::Path,
child: &mut tokio::process::Child,
) -> Result<SpawnReady, SpawnError> {
let mut tail = StderrTail::new();
let deadline = tokio::time::Instant::now() + READY_TIMEOUT;
let mut offset = 0usize;
loop {
let content = match tokio::fs::read_to_string(stderr_path).await {
Ok(content) => content,
Err(e) if e.kind() == io::ErrorKind::NotFound => String::new(),
Err(e) => return Err(SpawnError::Io(e)),
};
if content.len() > offset {
for line in content[offset..].lines() {
if let Some(rest) = line.strip_prefix(READY_PREFIX) {
let mut parts = rest.splitn(2, '\t');
let pod_name = parts.next().unwrap_or("").to_string();
let socket_str = parts.next().unwrap_or("").to_string();
if pod_name.is_empty() || socket_str.is_empty() {
return Err(SpawnError::PodExitedEarly {
stderr_tail: format!("malformed ready line: {line}"),
});
}
let socket_path = PathBuf::from(socket_str);
wait_for_socket(
&socket_path,
deadline,
child,
stderr_path,
&mut tail,
&mut offset,
)
.await?;
return Ok(SpawnReady {
pod_name,
socket_path,
});
}
tail.push(line);
form.message = Some((line.to_string(), MessageKind::Progress));
let _ = terminal.draw(|f| draw_form(f, form));
}
offset = content.len();
}
if tokio::time::Instant::now() >= deadline {
return Err(SpawnError::Timeout);
}
tokio::select! {
status = child.wait() => {
let _ = status;
// Pod は exit 直前に最終 stderr 行を flush することがある。
// child.wait() が解決した後に再読みして、原因行を取りこ
// ぼさず PodExitedEarly に載せる。
drain_stderr_into_tail(stderr_path, &mut tail, &mut offset).await;
return Err(SpawnError::PodExitedEarly {
stderr_tail: tail.into_string(),
});
}
_ = tokio::time::sleep(Duration::from_millis(100)) => {}
}
}
}
async fn wait_for_socket(
socket_path: &std::path::Path,
deadline: tokio::time::Instant,
child: &mut tokio::process::Child,
stderr_path: &std::path::Path,
tail: &mut StderrTail,
offset: &mut usize,
) -> Result<(), SpawnError> {
loop {
match tokio::net::UnixStream::connect(socket_path).await {
Ok(_) => return Ok(()),
Err(e)
if e.kind() == io::ErrorKind::NotFound
|| e.kind() == io::ErrorKind::ConnectionRefused => {}
Err(e) => return Err(SpawnError::Io(e)),
}
if tokio::time::Instant::now() >= deadline {
return Err(SpawnError::Timeout);
}
tokio::select! {
status = child.wait() => {
let _ = status;
drain_stderr_into_tail(stderr_path, tail, offset).await;
return Err(SpawnError::PodExitedEarly {
stderr_tail: tail.as_string(),
});
}
_ = tokio::time::sleep(Duration::from_millis(50)) => {}
}
}
}
async fn drain_stderr_into_tail(
stderr_path: &std::path::Path,
tail: &mut StderrTail,
offset: &mut usize,
) {
let Ok(content) = tokio::fs::read_to_string(stderr_path).await else {
return;
};
if content.len() <= *offset {
return;
}
for line in content[*offset..].lines() {
if !line.starts_with(READY_PREFIX) {
tail.push(line);
}
}
*offset = content.len();
let ready = spawn_pod(config, |line| {
form.message = Some((line.to_string(), MessageKind::Progress));
let _ = terminal.draw(|f| draw_form(f, form));
})
.await?;
Ok(SpawnReady {
pod_name: ready.pod_name,
socket_path: ready.socket_path,
})
}
fn build_overlay_toml(form: &Form) -> String {
@@ -496,47 +339,6 @@ async fn load_resume_scope(session_id: SessionId) -> Result<ScopeConfig, SpawnEr
})
}
/// Resolves the binary used to launch a child Pod. Must point at a
/// `pod`-compatible executable — the parent reads the child's stderr
/// directly looking for `INSOMNIA-READY`, so any wrapper that emits
/// extra lines on stderr will pollute that handshake.
///
/// `INSOMNIA_POD_COMMAND` overrides the lookup (used by tests to inject
/// a mock binary). Otherwise we defer to `PATH` — missing binary
/// surfaces as the spawn `io::Error`.
fn resolve_pod_command() -> PathBuf {
if let Ok(cmd) = std::env::var("INSOMNIA_POD_COMMAND") {
if !cmd.is_empty() {
return PathBuf::from(cmd);
}
}
PathBuf::from("pod")
}
struct StderrTail {
lines: std::collections::VecDeque<String>,
}
impl StderrTail {
fn new() -> Self {
Self {
lines: std::collections::VecDeque::with_capacity(8),
}
}
fn push(&mut self, line: &str) {
if self.lines.len() == 8 {
self.lines.pop_front();
}
self.lines.push_back(line.to_string());
}
fn as_string(&self) -> String {
self.lines.iter().cloned().collect::<Vec<_>>().join(" | ")
}
fn into_string(self) -> String {
self.lines.into_iter().collect::<Vec<_>>().join(" | ")
}
}
enum MessageKind {
Info,
Ok,