Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
216 changes: 216 additions & 0 deletions src/herdr.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,216 @@
//! Report agent state to herdr when running inside a herdr pane.
//!
//! Herdr injects `HERDR_PANE_ID` and `HERDR_SOCKET_PATH` into pane processes.
//! When present, crabcode posts `pane.report_agent` over the Unix socket so the
//! pane shows up under Agents (grouped). On exit it calls `pane.release_agent`
//! so the row is removed. See herdr's socket-api docs.

use crate::session::types::SessionStatus;
use std::io::{Read, Write};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

const SOURCE: &str = "crabcode";
const AGENT: &str = "crabcode";
const CONNECT_TIMEOUT: Duration = Duration::from_millis(80);
const IO_TIMEOUT: Duration = Duration::from_millis(120);

static ENV: OnceLock<Option<HerdrEnv>> = OnceLock::new();
static LAST_STATE: OnceLock<std::sync::Mutex<Option<&'static str>>> = OnceLock::new();
static SEQ: AtomicU64 = AtomicU64::new(0);

#[derive(Clone)]
struct HerdrEnv {
pane_id: String,
socket_path: String,
}

fn env() -> Option<&'static HerdrEnv> {
ENV.get_or_init(|| {
let pane_id = std::env::var("HERDR_PANE_ID").ok()?;
let socket_path = std::env::var("HERDR_SOCKET_PATH").ok()?;
if pane_id.is_empty() || socket_path.is_empty() {
return None;
}
Some(HerdrEnv {
pane_id,
socket_path,
})
})
.as_ref()
}

/// Whether crabcode is running inside a herdr pane.
pub fn is_active() -> bool {
env().is_some()
}

/// Map crabcode session status → herdr agent state.
fn herdr_state(status: SessionStatus) -> &'static str {
match status {
SessionStatus::Streaming => "working",
SessionStatus::Waiting => "blocked",
SessionStatus::Idle | SessionStatus::Failed | SessionStatus::Interrupted => "idle",
}
}

/// Report the current session status to herdr (no-op outside herdr).
pub fn report_session_status(status: SessionStatus) {
report_state(herdr_state(status), None);
}

/// Report idle on startup so the pane is classified as crabcode immediately.
pub fn report_startup() {
report_state("idle", Some("ready"));
}

/// Drop crabcode from herdr's agents panel. Custom (non-registry) agents are
/// not auto-cleared on process exit — callers must release explicitly.
pub fn report_shutdown() {
let Some(env) = env() else {
return;
};

if let Ok(mut guard) = LAST_STATE
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
{
*guard = None;
}

let seq = next_seq();
let payload = serde_json::json!({
"id": format!("crabcode:release:{seq}"),
"method": "pane.release_agent",
"params": {
"pane_id": env.pane_id,
"source": SOURCE,
"agent": AGENT,
"seq": seq,
},
});

let _ = send_rpc(&env.socket_path, &payload);
}

/// RAII guard: reports startup on create, release on drop (incl. panic unwind).
pub struct Session {
active: bool,
}

impl Session {
pub fn start() -> Self {
if is_active() {
report_startup();
Self { active: true }
} else {
Self { active: false }
}
}
}

impl Drop for Session {
fn drop(&mut self) {
if self.active {
report_shutdown();
}
}
}

fn report_state(state: &'static str, message: Option<&str>) {
let Some(env) = env() else {
return;
};

let last = LAST_STATE.get_or_init(|| std::sync::Mutex::new(None));
if let Ok(mut guard) = last.lock() {
if *guard == Some(state) && message.is_none() {
return;
}
*guard = Some(state);
}

let seq = next_seq();
let id = format!("crabcode:{seq}");
let mut params = serde_json::json!({
"pane_id": env.pane_id,
"source": SOURCE,
"agent": AGENT,
"state": state,
"seq": seq,
});
if let Some(message) = message {
params["message"] = serde_json::Value::String(message.to_string());
}

let payload = serde_json::json!({
"id": id,
"method": "pane.report_agent",
"params": params,
});

let _ = send_rpc(&env.socket_path, &payload);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Cache a state only after its report succeeds

If Herdr is restarting or the socket otherwise fails during connect/write, this result is discarded after LAST_STATE has already been updated. Every later report of the same state is then treated as a duplicate and suppressed, so a transient socket failure can leave the panel stale for the entire working or blocked interval; update the cache only after a successful RPC, or clear it on failure.

Useful? React with 👍 / 👎.

}

fn next_seq() -> u64 {
let from_time = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let n = SEQ.fetch_add(1, Ordering::Relaxed);
from_time.saturating_add(n)
}

fn send_rpc(socket_path: &str, payload: &serde_json::Value) -> std::io::Result<()> {
#[cfg(unix)]
{
use std::os::unix::net::UnixStream;

let started = Instant::now();
let mut stream = UnixStream::connect(socket_path)?;
Comment on lines +170 to +171

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Enforce the connect timeout before blocking

When the Unix socket's accept queue is stalled or full, UnixStream::connect is a blocking call and can exceed the intended 80 ms limit because elapsed time is checked only after it returns. Since reporting runs synchronously from session-status handling, an unhealthy Herdr daemon can stall Crabcode's event loop well beyond the configured timeout; the connection itself needs to be nonblocking or moved off the event-loop thread.

Useful? React with 👍 / 👎.

stream.set_read_timeout(Some(IO_TIMEOUT))?;
stream.set_write_timeout(Some(IO_TIMEOUT))?;

if started.elapsed() > CONNECT_TIMEOUT {
return Ok(());
}

let mut body = serde_json::to_vec(payload)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
body.push(b'\n');
stream.write_all(&body)?;
stream.flush()?;

// Drain one response line so herdr does not see a reset mid-write.
let mut buf = [0u8; 512];
let _ = stream.read(&mut buf);
Ok(())
}

#[cfg(not(unix))]
{
let _ = (socket_path, payload);
Ok(())
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn maps_session_status_to_herdr_state() {
assert_eq!(herdr_state(SessionStatus::Streaming), "working");
assert_eq!(herdr_state(SessionStatus::Waiting), "blocked");
assert_eq!(herdr_state(SessionStatus::Idle), "idle");
assert_eq!(herdr_state(SessionStatus::Failed), "idle");
assert_eq!(herdr_state(SessionStatus::Interrupted), "idle");
}

#[test]
fn inactive_without_env() {
// Tests run outside herdr; env should be unset.
assert!(!is_active() || env().is_some());
}
}
3 changes: 3 additions & 0 deletions src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ mod auth;
mod autocomplete;
mod command;
mod config;
mod herdr;
mod llm;
mod logging;
mod mcp;
Expand Down Expand Up @@ -853,6 +854,8 @@ async fn main() -> Result<()> {
}

let mut app = App::new_with_model_override(args.model.as_deref())?;
// Keep herdr authority until this guard drops (normal exit or panic).
let _herdr = crate::herdr::Session::start();

if let Some(ref session_id) = args.session {
if app.session_manager.ensure_session_loaded(session_id) {
Expand Down
11 changes: 11 additions & 0 deletions src/session/manager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -503,6 +503,12 @@ impl SessionManager {
if self.sessions.contains_key(id) {
let _ = self.hydrate_session(id);
self.current_session_id = Some(id.to_string());
let status = self
.sessions
.get(id)
.map(|s| s.status)
.unwrap_or(SessionStatus::Idle);
crate::herdr::report_session_status(status);
true
} else {
false
Expand Down Expand Up @@ -777,6 +783,11 @@ impl SessionManager {
}
}

// Only the active pane session drives herdr's agent state.
if self.current_session_id.as_deref() == Some(id) {
crate::herdr::report_session_status(status);
}
Comment on lines +787 to +789

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Reset the report when the active session is cleared

When a user chooses New/Home or archives/deletes the current session while it is Streaming or Waiting, clear_current_session sets the current ID to None without reporting idle. Subsequent completion of that background session also skips this conditional because it is no longer current, so Herdr can continue displaying working or blocked until another session changes state or Crabcode exits.

Useful? React with 👍 / 👎.


Ok(())
}

Expand Down
Loading