kaizen/daemon/
client.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2//! Daemon IPC client calls.
3
4use crate::ipc::{
5    CaptureStatus, ClientHello, ClientKind, DaemonRequest, DaemonResponse, ObservedSession,
6    PROTO_VERSION, ProxyEndpoint, ServerHello, read_frame, write_frame,
7};
8use anyhow::{Context, Result, anyhow};
9use std::path::Path;
10use std::time::Duration;
11use tokio::net::UnixStream;
12
13const LIFECYCLE_TIMEOUT_MS: u64 = 500;
14
15pub fn request_blocking(request: DaemonRequest) -> Result<DaemonResponse> {
16    super::ensure_running()?;
17    tokio::runtime::Runtime::new()?.block_on(request_async(request))
18}
19
20fn request_blocking_for(workspace: &str, request: DaemonRequest) -> Result<DaemonResponse> {
21    super::ensure_running_for(Path::new(workspace))?;
22    tokio::runtime::Runtime::new()?.block_on(request_async(request))
23}
24
25pub fn hello_blocking(client: ClientKind, workspace: Option<String>) -> Result<ServerHello> {
26    let request = DaemonRequest::Hello(ClientHello {
27        proto_version: PROTO_VERSION,
28        client,
29        workspace: workspace.clone(),
30    });
31    let response = match workspace.as_deref() {
32        Some(workspace) => request_blocking_for(workspace, request),
33        None => request_blocking(request),
34    }?;
35    match response {
36        DaemonResponse::Hello(hello) => Ok(hello),
37        DaemonResponse::Error { message, .. } => Err(anyhow!(message)),
38        _ => Err(anyhow!("unexpected daemon hello response")),
39    }
40}
41
42pub fn ensure_capture_blocking(workspace: String, deep: bool) -> Result<CaptureStatus> {
43    let request = DaemonRequest::EnsureWorkspaceCapture {
44        workspace: workspace.clone(),
45        deep,
46    };
47    match request_blocking_for(&workspace, request)? {
48        DaemonResponse::CaptureStatus(status) => Ok(*status),
49        DaemonResponse::Error { message, .. } => Err(anyhow!(message)),
50        _ => Err(anyhow!("unexpected daemon capture response")),
51    }
52}
53
54pub fn ensure_proxy_blocking(workspace: String, provider: String) -> Result<ProxyEndpoint> {
55    let request = DaemonRequest::EnsureProxy {
56        workspace: workspace.clone(),
57        provider,
58    };
59    match request_blocking_for(&workspace, request)? {
60        DaemonResponse::ProxyEndpoint(endpoint) => Ok(endpoint),
61        DaemonResponse::Error { message, .. } => Err(anyhow!(message)),
62        _ => Err(anyhow!("unexpected daemon proxy response")),
63    }
64}
65
66pub fn begin_observed_session_blocking(
67    workspace: String,
68    agent: String,
69) -> Result<ObservedSession> {
70    let request = DaemonRequest::BeginObservedSession {
71        workspace: workspace.clone(),
72        agent,
73    };
74    match request_blocking_for(&workspace, request)? {
75        DaemonResponse::ObservedSession(session) => Ok(session),
76        DaemonResponse::Error { message, .. } => Err(anyhow!(message)),
77        _ => Err(anyhow!("unexpected daemon observe response")),
78    }
79}
80
81pub(super) async fn request_async(request: DaemonRequest) -> Result<DaemonResponse> {
82    let paths = super::runtime_paths()?;
83    let mut stream = UnixStream::connect(&paths.sock)
84        .await
85        .with_context(|| format!("connect daemon socket: {}", paths.sock.display()))?;
86    write_frame(&mut stream, &request).await?;
87    read_frame(&mut stream).await
88}
89
90pub(super) async fn request_lifecycle_async(request: DaemonRequest) -> Result<DaemonResponse> {
91    let timeout = Duration::from_millis(LIFECYCLE_TIMEOUT_MS);
92    tokio::time::timeout(timeout, request_async(request))
93        .await
94        .map_err(|_| anyhow!("daemon IPC timed out after {LIFECYCLE_TIMEOUT_MS}ms"))?
95}
kaizen/daemon/client.rs

kaizen/daemon/
client.rs
