xiaoxue_admin
/
herdr
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(api): add socket subscriptions and lifecycle events

This commit is contained in:
Ogulcan Celik 2026-03-29 17:20:53 +03:00
parent fd1e3b8b05
commit 8b7848aac3
7 changed files with 1662 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
README.md
View File

@ -7,7 +7,7 @@
<p align="center">herd your agents.</p>
<p align="center">
<a href="https://herdr.dev">herdr.dev</a> · <a href="#install">install</a> · <a href="#usage">usage</a> · <a href="./CONFIGURATION.md">configuration</a>
<a href="https://herdr.dev">herdr.dev</a> · <a href="#install">install</a> · <a href="#usage">usage</a> · <a href="./CONFIGURATION.md">configuration</a> · <a href="./SOCKET_API.md">socket api</a>
</p>
---
@ -174,12 +174,25 @@ this means detection works with any supported agent, installed any way, with zer
the heuristics are pattern-matched against each agent's actual terminal output: prompt boxes, spinners, "waiting for input" messages, tool execution indicators. detection runs on a separate async task per pane, polled every 300-500ms, decoupled from terminal rendering.
## socket api
herdr now has a local unix socket API for scripts, tools, and coding agents.
you can:
- create, focus, rename, and close workspaces
- list, inspect, read, split, and close panes
- send text / keys into panes
- wait for output matches
- subscribe to lifecycle, agent, and output-match events over a single long-lived connection
see [`SOCKET_API.md`](./SOCKET_API.md) for request shapes, examples, and subscription behavior.
## what's coming
- **notification hooks**: a socket API so any agent or script can report its state to herdr. for agents without built-in detection, wire up a simple hook.
- **notification hooks**: richer agent/script-side state reporting on top of the socket foundation, so unsupported tools can report status directly to herdr.
- **in-app preferences**: rerun onboarding and adjust things like sound and toast notifications without editing config by hand.
- **native notifications**: OS-level notifications when an agent needs attention and herdr isn't in focus.
- **agent API**: `herdr create`, `herdr split`, `herdr send`, so agents and scripts can manage herdr workspaces programmatically.
- **agent cli wrapper**: `herdr pane ...`, `herdr wait ...`, and similar shell-friendly commands layered on top of the socket API.
## built with agents
@ -191,6 +204,8 @@ there will be rough edges. if you hit one, [open an issue](https://github.com/og
## cli
current built-in commands:
```
herdr launch herdr
herdr update download and install the latest version
@ -200,6 +215,11 @@ herdr --no-session start without restoring or saving sessions
herdr --help show help
```
programmatic control today lives in the socket API:
- [`SOCKET_API.md`](./SOCKET_API.md)
shell-friendly wrapper commands like `herdr pane ...` and `herdr wait ...` are planned on top of that API.
## building from source
```bash
@ -212,7 +232,7 @@ cargo build --release
## testing
```bash
just test # unit tests (157 tests)
just test # unit tests
just test-integration # LLM-based integration tests
just test-all # both
```

360
SOCKET_API.md Normal file
View File

@ -0,0 +1,360 @@
# herdr socket api
herdr exposes a local unix socket API for scripts, tools, and coding agents that want to control a running herdr instance or subscribe to pane/workspace events.
this is the low-level integration surface.
a CLI wrapper on top of it is planned, but the socket API is the foundation.
## transport
- transport: unix domain socket
- encoding: newline-delimited JSON
- request/response: one JSON request per line, one JSON response per line
- subscriptions: send `events.subscribe`, receive an ack, then keep the same connection open for pushed events
socket path resolution:
1. `HERDR_SOCKET_PATH`
2. `$XDG_RUNTIME_DIR/herdr.sock`
3. `$XDG_CONFIG_HOME/herdr/herdr.sock`
4. `$HOME/.config/herdr/herdr.sock`
5. `/tmp/herdr.sock`
## request shape
all requests use this envelope:
```json
{
"id": "req_1",
"method": "ping",
"params": {}
}
```
success responses:
```json
{
"id": "req_1",
"result": {
"type": "pong",
"version": "0.1.2"
}
}
```
error responses:
```json
{
"id": "req_1",
"error": {
"code": "pane_not_found",
"message": "pane p_1_99 not found"
}
}
```
## ids
workspace ids look like:
- `w_1`
- `w_2`
pane ids look like:
- `p_1_1`
- `p_1_2`
- `p_2_1`
that means:
- first number = workspace number
- second number = pane id inside that workspace
## core request methods
currently useful methods include:
### basic
- `ping`
### workspace
- `workspace.list`
- `workspace.get`
- `workspace.create`
- `workspace.focus`
- `workspace.rename`
- `workspace.close`
### pane
- `pane.list`
- `pane.get`
- `pane.read`
- `pane.send_text`
- `pane.send_keys`
- `pane.split`
- `pane.close`
### waits / events
- `pane.wait_for_output`
- `events.subscribe`
## example: create a workspace
```json
{
"id": "req_create",
"method": "workspace.create",
"params": {
"cwd": "/home/can/Projects/herdr",
"focus": true
}
}
```
example response:
```json
{
"id": "req_create",
"result": {
"type": "workspace_info",
"workspace": {
"workspace_id": "w_1",
"number": 1,
"label": "herdr",
"focused": true,
"pane_count": 1,
"agent_state": "unknown"
}
}
}
```
## example: read pane output
```json
{
"id": "req_read",
"method": "pane.read",
"params": {
"pane_id": "p_1_1",
"source": "recent",
"lines": 80
}
}
```
`source` can be:
- `visible`
- `recent`
## example: send text and press enter
low-level input is intentionally explicit:
```json
{
"id": "req_send_text",
"method": "pane.send_text",
"params": {
"pane_id": "p_1_1",
"text": "bun run dev"
}
}
```
then:
```json
{
"id": "req_send_keys",
"method": "pane.send_keys",
"params": {
"pane_id": "p_1_1",
"keys": ["Enter"]
}
}
```
this is kept separate on purpose. sending text is not always the same thing as submitting it.
a future CLI wrapper will likely offer a more ergonomic `pane run` style command on top of this.
## example: one-shot wait for output
```json
{
"id": "req_wait",
"method": "pane.wait_for_output",
"params": {
"pane_id": "p_1_1",
"source": "recent",
"lines": 200,
"match": { "type": "substring", "value": "ready" },
"timeout_ms": 30000
}
}
```
regex matching is also supported:
```json
{
"type": "regex",
"value": "server.*ready"
}
```
## subscriptions
`events.subscribe` is the long-lived pubsub entrypoint.
you send a subscribe request once, get an ack on the same connection, and then keep reading newline-delimited JSON events from that same socket.
### subscription ack
```json
{
"id": "sub_1",
"result": {
"type": "subscription_started"
}
}
```
## supported subscriptions
### lifecycle / base events
- `workspace.created`
- `workspace.closed`
- `workspace.focused`
- `pane.created`
- `pane.closed`
- `pane.focused`
- `pane.exited`
- `pane.agent_detected`
- `pane.agent_state_changed`
### parameterized event
- `pane.output_matched`
## example: subscribe to lifecycle events
```json
{
"id": "sub_life",
"method": "events.subscribe",
"params": {
"subscriptions": [
{ "type": "workspace.created" },
{ "type": "workspace.focused" },
{ "type": "pane.created" },
{ "type": "pane.focused" },
{ "type": "pane.agent_detected" },
{ "type": "pane.closed" },
{ "type": "workspace.closed" }
]
}
}
```
example pushed event:
```json
{
"event": "workspace_created",
"data": {
"workspace": {
"workspace_id": "w_1",
"number": 1,
"label": "herdr",
"focused": true,
"pane_count": 1,
"agent_state": "unknown"
}
}
}
```
## example: subscribe to output matches and agent state changes
```json
{
"id": "sub_1",
"method": "events.subscribe",
"params": {
"subscriptions": [
{
"type": "pane.output_matched",
"pane_id": "p_1_1",
"source": "recent",
"lines": 200,
"match": { "type": "substring", "value": "ready" }
},
{
"type": "pane.agent_state_changed",
"pane_id": "p_1_1",
"state": "idle"
}
]
}
}
```
example pushed `pane.output_matched` event:
```json
{
"event": "pane.output_matched",
"data": {
"pane_id": "p_1_1",
"matched_line": "server ready",
"read": {
"pane_id": "p_1_1",
"workspace_id": "w_1",
"source": "recent",
"text": "...server ready...",
"revision": 0,
"truncated": false
}
}
}
```
example pushed `pane.agent_state_changed` event:
```json
{
"event": "pane.agent_state_changed",
"data": {
"pane_id": "p_1_1",
"workspace_id": "w_1",
"state": "idle",
"agent": "pi"
}
}
```
## behavior notes
- `pane.output_matched` emits when a subscription transitions into a matching state. it does not repeatedly spam the same visible match on every poll.
- closing the socket connection ends the subscription.
- there is no separate transport for events.
- the same herdr process can serve regular request/response calls and long-lived subscription connections at the same time.
## intended layering
recommended architecture:
- socket api = foundational integration protocol
- future `herdr ...` commands = ergonomic wrapper for humans and coding agents
for agent workflows, the future CLI should likely expose blocking commands like:
- `herdr pane read ...`
- `herdr pane run ...`
- `herdr wait output ...`
- `herdr wait agent-state ...`
but those should sit on top of this socket surface rather than replacing it.

504
src/api/mod.rs
View File

@ -4,13 +4,16 @@ use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::time::Duration;
use tracing::{debug, error, info, warn};
use regex::Regex;
use crate::api::schema::{
ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse,
ErrorBody, ErrorResponse, Method, PaneAgentStateChangedEvent, PaneOutputMatchedEvent, Request,
ResponseResult, Subscription, SubscriptionEventData, SubscriptionEventEnvelope,
SubscriptionEventKind, SuccessResponse,
};
pub const SOCKET_PATH_ENV_VAR: &str = "HERDR_SOCKET_PATH";
@ -20,6 +23,46 @@ pub struct ApiRequestMessage {
pub respond_to: std::sync::mpsc::Sender<String>,
}
#[derive(Clone, Default)]
pub struct EventHub {
inner: std::sync::Arc<std::sync::Mutex<EventHubState>>,
}
#[derive(Default)]
struct EventHubState {
next_sequence: u64,
events: Vec<(u64, crate::api::schema::EventEnvelope)>,
}
impl EventHub {
const MAX_EVENTS: usize = 512;
pub fn push(&self, event: crate::api::schema::EventEnvelope) {
let Ok(mut state) = self.inner.lock() else {
return;
};
state.next_sequence += 1;
let sequence = state.next_sequence;
state.events.push((sequence, event));
let overflow = state.events.len().saturating_sub(Self::MAX_EVENTS);
if overflow > 0 {
state.events.drain(0..overflow);
}
}
pub fn events_after(&self, sequence: u64) -> Vec<(u64, crate::api::schema::EventEnvelope)> {
let Ok(state) = self.inner.lock() else {
return Vec::new();
};
state
.events
.iter()
.filter(|(event_sequence, _)| *event_sequence > sequence)
.cloned()
.collect()
}
}
pub fn socket_path() -> PathBuf {
if let Ok(path) = std::env::var(SOCKET_PATH_ENV_VAR) {
return PathBuf::from(path);
@ -57,6 +100,7 @@ impl Drop for ServerHandle {
pub fn start_server(
api_tx: std::sync::mpsc::Sender<ApiRequestMessage>,
event_hub: EventHub,
) -> std::io::Result<ServerHandle> {
let path = socket_path();
prepare_socket_path(&path)?;
@ -68,9 +112,13 @@ pub fn start_server(
for stream in listener.incoming() {
match stream {
Ok(stream) => {
if let Err(err) = handle_connection(stream, &api_tx) {
warn!(err = %err, "api connection failed");
}
let api_tx = api_tx.clone();
let event_hub = event_hub.clone();
std::thread::spawn(move || {
if let Err(err) = handle_connection(stream, &api_tx, &event_hub) {
warn!(err = %err, "api connection failed");
}
});
}
Err(err) => {
error!(err = %err, "api listener accept failed");
@ -104,6 +152,7 @@ fn prepare_socket_path(path: &Path) -> std::io::Result<()> {
fn handle_connection(
mut stream: UnixStream,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
event_hub: &EventHub,
) -> std::io::Result<()> {
let mut line = String::new();
{
@ -119,21 +168,41 @@ fn handle_connection(
return Ok(());
}
let response = match serde_json::from_str::<Request>(line) {
Ok(request) => handle_request(request, api_tx),
Err(err) => serde_json::to_string(&ErrorResponse {
id: String::new(),
error: ErrorBody {
code: "invalid_request".into(),
message: format!("invalid request: {err}"),
},
})?,
let request = match serde_json::from_str::<Request>(line) {
Ok(request) => request,
Err(err) => {
write_json_line(
&mut stream,
&ErrorResponse {
id: String::new(),
error: ErrorBody {
code: "invalid_request".into(),
message: format!("invalid request: {err}"),
},
},
)?;
return Ok(());
}
};
stream.write_all(response.as_bytes())?;
stream.write_all(b"\n")?;
stream.flush()?;
Ok(())
match request.method {
Method::EventsSubscribe(params) => {
stream_subscriptions(stream, request.id, params, api_tx, event_hub)
}
method => {
let response = handle_request(
Request {
id: request.id,
method,
},
api_tx,
);
stream.write_all(response.as_bytes())?;
stream.write_all(b"\n")?;
stream.flush()?;
Ok(())
}
}
}
fn handle_request(request: Request, api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>) -> String {
@ -213,19 +282,7 @@ fn wait_for_output(
.unwrap();
};
let matched_line = match &params.r#match {
crate::api::schema::OutputMatch::Substring { value } => read
.text
.lines()
.find(|line| line.contains(value))
.map(|line| line.to_string()),
crate::api::schema::OutputMatch::Regex { .. } => regex.as_ref().and_then(|re| {
read.text
.lines()
.find(|line| re.is_match(line))
.map(|line| line.to_string())
}),
};
let matched_line = match_output(&read.text, &params.r#match, regex.as_ref());
if matched_line.is_some() {
let revision = read.revision;
return serde_json::to_string(&SuccessResponse {
@ -255,6 +312,393 @@ fn wait_for_output(
}
}
fn stream_subscriptions(
mut stream: UnixStream,
request_id: String,
params: crate::api::schema::EventsSubscribeParams,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
event_hub: &EventHub,
) -> std::io::Result<()> {
let mut subscriptions = Vec::with_capacity(params.subscriptions.len());
for (index, subscription) in params.subscriptions.into_iter().enumerate() {
let active =
match ActiveSubscription::new(subscription, &request_id, index, api_tx, event_hub) {
Ok(active) => active,
Err(response) => {
write_json_line(&mut stream, &response)?;
return Ok(());
}
};
subscriptions.push(active);
}
write_json_line(
&mut stream,
&SuccessResponse {
id: request_id,
result: ResponseResult::SubscriptionStarted {},
},
)?;
loop {
for subscription in &mut subscriptions {
if let Some(event) = subscription.poll(api_tx, event_hub) {
write_json_line(&mut stream, &event)?;
}
}
std::thread::sleep(Duration::from_millis(100));
}
}
fn write_json_line<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> std::io::Result<()> {
let encoded = serde_json::to_string(value)
.map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?;
stream.write_all(encoded.as_bytes())?;
stream.write_all(b"\n")?;
stream.flush()?;
Ok(())
}
fn match_output(
text: &str,
matcher: &crate::api::schema::OutputMatch,
regex: Option<&Regex>,
) -> Option<String> {
match matcher {
crate::api::schema::OutputMatch::Substring { value } => text
.lines()
.find(|line| line.contains(value))
.map(|line| line.to_string()),
crate::api::schema::OutputMatch::Regex { .. } => regex.and_then(|re| {
text.lines()
.find(|line| re.is_match(line))
.map(|line| line.to_string())
}),
}
}
struct ActiveOutputMatchedSubscription {
pane_id: String,
source: crate::api::schema::ReadSource,
lines: Option<u32>,
matcher: crate::api::schema::OutputMatch,
regex: Option<Regex>,
strip_ansi: bool,
currently_matching: bool,
request_prefix: String,
}
struct ActiveAgentStateChangedSubscription {
pane_id: String,
state_filter: Option<crate::api::schema::PaneAgentState>,
last_state: Option<crate::api::schema::PaneAgentState>,
request_prefix: String,
}
struct ActiveEventSubscription {
event_kind: crate::api::schema::EventKind,
last_sequence: u64,
}
enum ActiveSubscription {
Event(ActiveEventSubscription),
OutputMatched(ActiveOutputMatchedSubscription),
AgentStateChanged(ActiveAgentStateChangedSubscription),
}
impl ActiveSubscription {
fn new(
subscription: Subscription,
request_id: &str,
index: usize,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
_event_hub: &EventHub,
) -> Result<Self, ErrorResponse> {
match subscription {
Subscription::WorkspaceCreated {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::WorkspaceCreated,
last_sequence: 0,
})),
Subscription::WorkspaceClosed {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::WorkspaceClosed,
last_sequence: 0,
})),
Subscription::WorkspaceFocused {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::WorkspaceFocused,
last_sequence: 0,
})),
Subscription::PaneCreated {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::PaneCreated,
last_sequence: 0,
})),
Subscription::PaneClosed {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::PaneClosed,
last_sequence: 0,
})),
Subscription::PaneFocused {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::PaneFocused,
last_sequence: 0,
})),
Subscription::PaneExited {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::PaneExited,
last_sequence: 0,
})),
Subscription::PaneAgentDetected {} => Ok(Self::Event(ActiveEventSubscription {
event_kind: crate::api::schema::EventKind::PaneAgentDetected,
last_sequence: 0,
})),
Subscription::PaneOutputMatched {
pane_id,
source,
lines,
r#match,
strip_ansi,
} => {
let regex = match &r#match {
crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) {
Ok(regex) => Some(regex),
Err(err) => {
return Err(ErrorResponse {
id: request_id.to_string(),
error: ErrorBody {
code: "invalid_regex".into(),
message: err.to_string(),
},
});
}
},
crate::api::schema::OutputMatch::Substring { .. } => None,
};
let probe = pane_read(
format!("{request_id}:sub:{index}:probe"),
&pane_id,
source.clone(),
lines,
strip_ansi,
api_tx,
);
if let Err(error) = probe {
return Err(error);
}
Ok(Self::OutputMatched(ActiveOutputMatchedSubscription {
pane_id,
source,
lines,
matcher: r#match,
regex,
strip_ansi,
currently_matching: false,
request_prefix: format!("{request_id}:sub:{index}"),
}))
}
Subscription::PaneAgentStateChanged { pane_id, state } => {
let probe =
match pane_get(format!("{request_id}:sub:{index}:probe"), &pane_id, api_tx) {
Ok(probe) => probe,
Err(error) => return Err(error),
};
Ok(Self::AgentStateChanged(
ActiveAgentStateChangedSubscription {
pane_id,
state_filter: state,
last_state: Some(probe.agent_state),
request_prefix: format!("{request_id}:sub:{index}"),
},
))
}
}
}
fn poll(
&mut self,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
event_hub: &EventHub,
) -> Option<serde_json::Value> {
match self {
Self::Event(subscription) => subscription.poll(event_hub),
Self::OutputMatched(subscription) => {
serde_json::to_value(subscription.poll(api_tx)?).ok()
}
Self::AgentStateChanged(subscription) => {
serde_json::to_value(subscription.poll(api_tx)?).ok()
}
}
}
}
impl ActiveEventSubscription {
fn poll(&mut self, event_hub: &EventHub) -> Option<serde_json::Value> {
for (sequence, event) in event_hub.events_after(self.last_sequence) {
self.last_sequence = sequence;
if event.event == self.event_kind {
return serde_json::to_value(event).ok();
}
}
None
}
}
impl ActiveOutputMatchedSubscription {
fn poll(
&mut self,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
) -> Option<SubscriptionEventEnvelope> {
let read = pane_read(
format!("{}:read", self.request_prefix),
&self.pane_id,
self.source.clone(),
self.lines,
self.strip_ansi,
api_tx,
)
.ok()?;
let matched_line = match_output(&read.text, &self.matcher, self.regex.as_ref());
match matched_line {
Some(matched_line) => {
if self.currently_matching {
return None;
}
self.currently_matching = true;
Some(SubscriptionEventEnvelope {
event: SubscriptionEventKind::PaneOutputMatched,
data: SubscriptionEventData::PaneOutputMatched(PaneOutputMatchedEvent {
pane_id: self.pane_id.clone(),
matched_line,
read,
}),
})
}
None => {
self.currently_matching = false;
None
}
}
}
}
impl ActiveAgentStateChangedSubscription {
fn poll(
&mut self,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
) -> Option<SubscriptionEventEnvelope> {
let pane = pane_get(
format!("{}:pane", self.request_prefix),
&self.pane_id,
api_tx,
)
.ok()?;
let current_state = pane.agent_state;
let previous_state = self.last_state.replace(current_state);
if previous_state.is_none() || previous_state == Some(current_state) {
return None;
}
if self
.state_filter
.is_some_and(|wanted| wanted != current_state)
{
return None;
}
Some(SubscriptionEventEnvelope {
event: SubscriptionEventKind::PaneAgentStateChanged,
data: SubscriptionEventData::PaneAgentStateChanged(PaneAgentStateChangedEvent {
pane_id: pane.pane_id,
workspace_id: pane.workspace_id,
state: current_state,
agent: pane.agent,
}),
})
}
}
fn pane_read(
request_id: String,
pane_id: &str,
source: crate::api::schema::ReadSource,
lines: Option<u32>,
strip_ansi: bool,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
) -> Result<crate::api::schema::PaneReadResult, ErrorResponse> {
let response = dispatch_to_app(
Request {
id: request_id.clone(),
method: Method::PaneRead(crate::api::schema::PaneReadParams {
pane_id: pane_id.to_string(),
source,
lines,
strip_ansi,
}),
},
api_tx,
);
let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse {
id: request_id.clone(),
error: ErrorBody {
code: "internal_error".into(),
message: "failed to decode pane read response".into(),
},
})?;
if value.get("error").is_some() {
return serde_json::from_value(value).map_err(|_| ErrorResponse {
id: request_id,
error: ErrorBody {
code: "internal_error".into(),
message: "failed to decode pane read error".into(),
},
});
}
serde_json::from_value(value["result"]["read"].clone()).map_err(|_| ErrorResponse {
id: request_id,
error: ErrorBody {
code: "internal_error".into(),
message: "failed to decode pane read result".into(),
},
})
}
fn pane_get(
request_id: String,
pane_id: &str,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
) -> Result<crate::api::schema::PaneInfo, ErrorResponse> {
let response = dispatch_to_app(
Request {
id: request_id.clone(),
method: Method::PaneGet(crate::api::schema::PaneTarget {
pane_id: pane_id.to_string(),
}),
},
api_tx,
);
let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse {
id: request_id.clone(),
error: ErrorBody {
code: "internal_error".into(),
message: "failed to decode pane get response".into(),
},
})?;
if value.get("error").is_some() {
return serde_json::from_value(value).map_err(|_| ErrorResponse {
id: request_id,
error: ErrorBody {
code: "internal_error".into(),
message: "failed to decode pane get error".into(),
},
});
}
serde_json::from_value(value["result"]["pane"].clone()).map_err(|_| ErrorResponse {
id: request_id,
error: ErrorBody {
code: "internal_error".into(),
message: "failed to decode pane get result".into(),
},
})
}
fn dispatch_to_app(
request: Request,
api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,

157
src/api/schema.rs
View File

@ -132,11 +132,44 @@ pub enum ReadSource {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EventsSubscribeParams {
pub events: Vec<EventKind>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pane_id: Option<String>,
pub subscriptions: Vec<Subscription>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum Subscription {
#[serde(rename = "workspace.created")]
WorkspaceCreated {},
#[serde(rename = "workspace.closed")]
WorkspaceClosed {},
#[serde(rename = "workspace.focused")]
WorkspaceFocused {},
#[serde(rename = "pane.created")]
PaneCreated {},
#[serde(rename = "pane.closed")]
PaneClosed {},
#[serde(rename = "pane.focused")]
PaneFocused {},
#[serde(rename = "pane.exited")]
PaneExited {},
#[serde(rename = "pane.agent_detected")]
PaneAgentDetected {},
#[serde(rename = "pane.output_matched")]
PaneOutputMatched {
pane_id: String,
source: ReadSource,
#[serde(default, skip_serializing_if = "Option::is_none")]
lines: Option<u32>,
r#match: OutputMatch,
#[serde(default = "default_true")]
strip_ansi: bool,
},
#[serde(rename = "pane.agent_state_changed")]
PaneAgentStateChanged {
pane_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
state: Option<PaneAgentState>,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@ -270,9 +303,7 @@ pub enum ResponseResult {
PaneRead {
read: PaneReadResult,
},
SubscriptionStarted {
events: Vec<EventKind>,
},
SubscriptionStarted {},
WaitMatched {
event: EventEnvelope,
},
@ -324,6 +355,43 @@ pub struct EventEnvelope {
pub data: EventData,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum SubscriptionEventKind {
#[serde(rename = "pane.output_matched")]
PaneOutputMatched,
#[serde(rename = "pane.agent_state_changed")]
PaneAgentStateChanged,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SubscriptionEventEnvelope {
pub event: SubscriptionEventKind,
pub data: SubscriptionEventData,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum SubscriptionEventData {
PaneOutputMatched(PaneOutputMatchedEvent),
PaneAgentStateChanged(PaneAgentStateChangedEvent),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PaneOutputMatchedEvent {
pub pane_id: String,
pub matched_line: String,
pub read: PaneReadResult,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PaneAgentStateChangedEvent {
pub pane_id: String,
pub workspace_id: String,
pub state: PaneAgentState,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub agent: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum EventData {
@ -476,6 +544,79 @@ mod tests {
assert_eq!(restored, event);
}
#[test]
fn subscribe_request_parses_parameterized_subscriptions() {
let json = r#"
{
"id": "sub_1",
"method": "events.subscribe",
"params": {
"subscriptions": [
{
"type": "pane.output_matched",
"pane_id": "p_1_1",
"source": "recent",
"lines": 200,
"match": { "type": "substring", "value": "auth: received" }
},
{
"type": "pane.agent_state_changed",
"pane_id": "p_1_1",
"state": "waiting"
}
]
}
}
"#;
let request: Request = serde_json::from_str(json).unwrap();
let Method::EventsSubscribe(params) = request.method else {
panic!("wrong method parsed");
};
assert_eq!(params.subscriptions.len(), 2);
assert!(matches!(
&params.subscriptions[0],
Subscription::PaneOutputMatched {
pane_id,
source: ReadSource::Recent,
lines: Some(200),
r#match: OutputMatch::Substring { value },
strip_ansi: true,
} if pane_id == "p_1_1" && value == "auth: received"
));
assert!(matches!(
&params.subscriptions[1],
Subscription::PaneAgentStateChanged {
pane_id,
state: Some(PaneAgentState::Waiting),
} if pane_id == "p_1_1"
));
}
#[test]
fn subscription_event_envelope_round_trips() {
let event = SubscriptionEventEnvelope {
event: SubscriptionEventKind::PaneOutputMatched,
data: SubscriptionEventData::PaneOutputMatched(PaneOutputMatchedEvent {
pane_id: "p_1_1".into(),
matched_line: "auth: received".into(),
read: PaneReadResult {
pane_id: "p_1_1".into(),
workspace_id: "w_1".into(),
source: ReadSource::Recent,
text: "auth: received\n".into(),
revision: 0,
truncated: false,
},
}),
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("\"event\":\"pane.output_matched\""));
let restored: SubscriptionEventEnvelope = serde_json::from_str(&json).unwrap();
assert_eq!(restored, event);
}
#[test]
fn success_response_round_trips() {
let response = SuccessResponse {

375
src/app/mod.rs
View File

@ -29,6 +29,8 @@ pub struct App {
pub event_tx: mpsc::Sender<AppEvent>,
event_rx: mpsc::Receiver<AppEvent>,
api_rx: std::sync::mpsc::Receiver<crate::api::ApiRequestMessage>,
event_hub: crate::api::EventHub,
last_focus: Option<(usize, crate::layout::PaneId)>,
no_session: bool,
config_diagnostic_deadline: Option<Instant>,
toast_deadline: Option<Instant>,
@ -40,6 +42,7 @@ impl App {
no_session: bool,
config_diagnostic: Option<String>,
api_rx: std::sync::mpsc::Receiver<crate::api::ApiRequestMessage>,
event_hub: crate::api::EventHub,
) -> Self {
let (prefix_code, prefix_mods) = config.prefix_key();
let (event_tx, event_rx) = mpsc::channel::<AppEvent>(64);
@ -111,6 +114,13 @@ impl App {
std::thread::spawn(move || crate::update::auto_update(update_tx));
}
let last_focus = state.active.and_then(|idx| {
state
.workspaces
.get(idx)
.map(|ws| (idx, ws.layout.focused()))
});
Self {
config_diagnostic_deadline: state
.config_diagnostic
@ -121,6 +131,8 @@ impl App {
event_tx,
event_rx,
api_rx,
event_hub,
last_focus,
no_session,
}
}
@ -148,25 +160,15 @@ impl App {
crate::ui::render(&self.state, frame);
})?;
// Drain internal events first so API reads observe fresh pane state.
self.drain_internal_events();
while let Ok(msg) = self.api_rx.try_recv() {
let response = self.handle_api_request(msg.request);
let _ = msg.respond_to.send(response);
}
// Drain internal events
while let Ok(ev) = self.event_rx.try_recv() {
let previous_toast = self.state.toast.clone();
self.state.handle_app_event(ev);
if self.state.toast != previous_toast {
self.toast_deadline = self.state.toast.as_ref().map(|toast| {
let duration = match toast.kind {
ToastKind::NeedsAttention => Duration::from_secs(8),
ToastKind::Finished => Duration::from_secs(5),
};
Instant::now() + duration
});
}
}
self.sync_focus_events();
if event::poll(Duration::from_millis(16))? {
match event::read()? {
@ -204,7 +206,114 @@ impl App {
Ok(())
}
fn drain_internal_events(&mut self) {
while let Ok(ev) = self.event_rx.try_recv() {
match &ev {
AppEvent::PaneDied { pane_id } => {
if let Some((ws_idx, _)) = self.find_pane(*pane_id) {
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneExited,
data: crate::api::schema::EventData::PaneExited {
pane_id: format!("p_{}_{}", ws_idx + 1, pane_id.raw()),
workspace_id: format!("w_{}", ws_idx + 1),
},
});
}
}
AppEvent::StateChanged {
pane_id,
agent,
state,
} => {
if let Some((ws_idx, pane)) = self.find_pane(*pane_id) {
let pane_id = format!("p_{}_{}", ws_idx + 1, pane_id.raw());
let workspace_id = format!("w_{}", ws_idx + 1);
if pane.detected_agent != *agent {
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneAgentDetected,
data: crate::api::schema::EventData::PaneAgentDetected {
pane_id: pane_id.clone(),
workspace_id: workspace_id.clone(),
agent: agent.map(agent_name),
},
});
}
if pane.state != *state {
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneAgentStateChanged,
data: crate::api::schema::EventData::PaneAgentStateChanged {
pane_id,
workspace_id,
state: pane_agent_state(*state),
},
});
}
}
}
AppEvent::UpdateReady { .. } => {}
}
let previous_toast = self.state.toast.clone();
self.state.handle_app_event(ev);
if self.state.toast != previous_toast {
self.toast_deadline = self.state.toast.as_ref().map(|toast| {
let duration = match toast.kind {
ToastKind::NeedsAttention => Duration::from_secs(8),
ToastKind::Finished => Duration::from_secs(5),
};
Instant::now() + duration
});
}
}
}
fn emit_event(&self, event: crate::api::schema::EventEnvelope) {
self.event_hub.push(event);
}
fn sync_focus_events(&mut self) {
let current_focus = self.state.active.and_then(|idx| {
self.state
.workspaces
.get(idx)
.map(|ws| (idx, ws.layout.focused()))
});
if current_focus == self.last_focus {
return;
}
if let Some((ws_idx, pane_id)) = current_focus {
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::WorkspaceFocused,
data: crate::api::schema::EventData::WorkspaceFocused {
workspace_id: format!("w_{}", ws_idx + 1),
},
});
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneFocused,
data: crate::api::schema::EventData::PaneFocused {
pane_id: format!("p_{}_{}", ws_idx + 1, pane_id.raw()),
workspace_id: format!("w_{}", ws_idx + 1),
},
});
}
self.last_focus = current_focus;
}
fn find_pane(
&self,
pane_id: crate::layout::PaneId,
) -> Option<(usize, &crate::pane::PaneState)> {
self.state
.workspaces
.iter()
.enumerate()
.find_map(|(ws_idx, ws)| ws.panes.get(&pane_id).map(|pane| (ws_idx, pane)))
}
fn handle_api_request(&mut self, request: crate::api::schema::Request) -> String {
self.drain_internal_events();
use bytes::Bytes;
use crate::api::schema::{
@ -260,12 +369,32 @@ impl App {
.or_else(|| std::env::current_dir().ok())
.unwrap_or_else(|| std::path::PathBuf::from("/"));
match self.create_workspace_with_options(cwd, params.focus) {
Ok(index) => SuccessResponse {
id: request.id,
result: ResponseResult::WorkspaceInfo {
workspace: workspace_info(&self.state, index),
},
},
Ok(index) => {
let workspace = workspace_info(&self.state, index);
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::WorkspaceCreated,
data: crate::api::schema::EventData::WorkspaceCreated {
workspace: workspace.clone(),
},
});
if let Some(pane_id) = self.state.workspaces[index]
.layout
.pane_ids()
.first()
.copied()
{
if let Some(pane) = self.pane_info(index, pane_id) {
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneCreated,
data: crate::api::schema::EventData::PaneCreated { pane },
});
}
}
SuccessResponse {
id: request.id,
result: ResponseResult::WorkspaceInfo { workspace },
}
}
Err(err) => {
return serde_json::to_string(&ErrorResponse {
id: request.id,
@ -278,6 +407,160 @@ impl App {
}
}
}
Method::WorkspaceFocus(target) => {
let Some(index) = parse_workspace_id(&target.workspace_id) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "workspace_not_found".into(),
message: format!("workspace {} not found", target.workspace_id),
},
})
.unwrap();
};
if self.state.workspaces.get(index).is_none() {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "workspace_not_found".into(),
message: format!("workspace {} not found", target.workspace_id),
},
})
.unwrap();
}
self.state.switch_workspace(index);
SuccessResponse {
id: request.id,
result: ResponseResult::WorkspaceInfo {
workspace: workspace_info(&self.state, index),
},
}
}
Method::WorkspaceRename(params) => {
let Some(index) = parse_workspace_id(&params.workspace_id) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "workspace_not_found".into(),
message: format!("workspace {} not found", params.workspace_id),
},
})
.unwrap();
};
let Some(ws) = self.state.workspaces.get_mut(index) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "workspace_not_found".into(),
message: format!("workspace {} not found", params.workspace_id),
},
})
.unwrap();
};
ws.set_custom_name(params.label.clone());
SuccessResponse {
id: request.id,
result: ResponseResult::WorkspaceInfo {
workspace: workspace_info(&self.state, index),
},
}
}
Method::WorkspaceClose(target) => {
let Some(index) = parse_workspace_id(&target.workspace_id) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "workspace_not_found".into(),
message: format!("workspace {} not found", target.workspace_id),
},
})
.unwrap();
};
if self.state.workspaces.get(index).is_none() {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "workspace_not_found".into(),
message: format!("workspace {} not found", target.workspace_id),
},
})
.unwrap();
}
self.state.selected = index;
self.state.close_selected_workspace();
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::WorkspaceClosed,
data: crate::api::schema::EventData::WorkspaceClosed {
workspace_id: target.workspace_id,
},
});
SuccessResponse {
id: request.id,
result: ResponseResult::Ok {},
}
}
Method::PaneSplit(params) => {
let Some((ws_idx, target_pane_id)) = parse_pane_id(&params.target_pane_id) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "pane_not_found".into(),
message: format!("pane {} not found", params.target_pane_id),
},
})
.unwrap();
};
let (rows, cols) = self.state.estimate_pane_size();
let Some(ws) = self.state.workspaces.get_mut(ws_idx) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "pane_not_found".into(),
message: format!("pane {} not found", params.target_pane_id),
},
})
.unwrap();
};
ws.layout.focus_pane(target_pane_id);
let direction = match params.direction {
crate::api::schema::SplitDirection::Right => {
ratatui::layout::Direction::Horizontal
}
crate::api::schema::SplitDirection::Down => {
ratatui::layout::Direction::Vertical
}
};
let new_pane_id = match ws.split_focused(
direction,
rows,
cols,
params.cwd.map(std::path::PathBuf::from),
) {
Ok(new_pane_id) => new_pane_id,
Err(err) => {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "pane_split_failed".into(),
message: err.to_string(),
},
})
.unwrap();
}
};
if !params.focus {
ws.layout.focus_pane(target_pane_id);
}
let pane = self.pane_info(ws_idx, new_pane_id).unwrap();
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneCreated,
data: crate::api::schema::EventData::PaneCreated { pane: pane.clone() },
});
SuccessResponse {
id: request.id,
result: ResponseResult::PaneInfo { pane },
}
}
Method::PaneList(PaneListParams { workspace_id }) => {
match self.collect_panes_for_workspace(workspace_id.as_deref()) {
Ok(panes) => SuccessResponse {
@ -395,6 +678,58 @@ impl App {
result: ResponseResult::Ok {},
}
}
Method::PaneClose(target) => {
let Some((ws_idx, pane_id)) = parse_pane_id(&target.pane_id) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "pane_not_found".into(),
message: format!("pane {} not found", target.pane_id),
},
})
.unwrap();
};
let Some(ws) = self.state.workspaces.get_mut(ws_idx) else {
return serde_json::to_string(&ErrorResponse {
id: request.id,
error: ErrorBody {
code: "pane_not_found".into(),
message: format!("pane {} not found", target.pane_id),
},
})
.unwrap();
};
let workspace_id = format!("w_{}", ws_idx + 1);
let pane_count = ws.layout.pane_count();
if pane_count <= 1 {
self.state.selected = ws_idx;
self.state.close_selected_workspace();
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneClosed,
data: crate::api::schema::EventData::PaneClosed {
pane_id: target.pane_id.clone(),
workspace_id: workspace_id.clone(),
},
});
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::WorkspaceClosed,
data: crate::api::schema::EventData::WorkspaceClosed { workspace_id },
});
} else {
ws.remove_pane(pane_id);
self.emit_event(crate::api::schema::EventEnvelope {
event: crate::api::schema::EventKind::PaneClosed,
data: crate::api::schema::EventData::PaneClosed {
pane_id: target.pane_id,
workspace_id,
},
});
}
SuccessResponse {
id: request.id,
result: ResponseResult::Ok {},
}
}
Method::PaneSendKeys(params) => {
let Some((ws_idx, pane_id)) = parse_pane_id(&params.pane_id) else {
return serde_json::to_string(&ErrorResponse {

5
src/main.rs
View File

@ -225,7 +225,8 @@ fn main() -> io::Result<()> {
init_logging();
let (api_tx, api_rx) = std::sync::mpsc::channel();
let _api_server = api::start_server(api_tx)?;
let event_hub = api::EventHub::default();
let _api_server = api::start_server(api_tx, event_hub.clone())?;
let no_session = std::env::args().any(|a| a == "--no-session");
let in_tmux = std::env::var("TMUX").is_ok();
@ -288,7 +289,7 @@ fn main() -> io::Result<()> {
std::io::stdout().flush()?;
}
let mut app = app::App::new(config, no_session, config_diagnostic, api_rx);
let mut app = app::App::new(config, no_session, config_diagnostic, api_rx, event_hub);
let result = app.run(&mut terminal).await;
// Reset modifyOtherKeys if we enabled it

297
tests/api_ping.rs
View File

@ -32,6 +32,15 @@ fn wait_for_socket(path: &Path, timeout: Duration) {
}
fn spawn_herdr(config_home: &Path, runtime_dir: &Path, socket_path: &Path) -> SpawnedHerdr {
spawn_herdr_with_path(config_home, runtime_dir, socket_path, None)
}
fn spawn_herdr_with_path(
config_home: &Path,
runtime_dir: &Path,
socket_path: &Path,
path_override: Option<&Path>,
) -> SpawnedHerdr {
fs::create_dir_all(config_home.join("herdr")).unwrap();
fs::create_dir_all(runtime_dir).unwrap();
fs::write(
@ -54,6 +63,9 @@ fn spawn_herdr(config_home: &Path, runtime_dir: &Path, socket_path: &Path) -> Sp
cmd.env("XDG_CONFIG_HOME", config_home);
cmd.env("XDG_RUNTIME_DIR", runtime_dir);
cmd.env("HERDR_SOCKET_PATH", socket_path);
if let Some(path) = path_override {
cmd.env("PATH", path);
}
let child = pair.slave.spawn_command(cmd).unwrap();
@ -75,6 +87,38 @@ fn send_request(socket_path: &Path, json: &str) -> serde_json::Value {
serde_json::from_str(&line).unwrap()
}
fn open_subscription(socket_path: &Path, json: &str) -> (UnixStream, BufReader<UnixStream>) {
let mut stream = UnixStream::connect(socket_path).unwrap();
stream.write_all(json.as_bytes()).unwrap();
stream.write_all(b"\n").unwrap();
stream.flush().unwrap();
let reader = BufReader::new(stream.try_clone().unwrap());
(stream, reader)
}
fn read_json_line(reader: &mut BufReader<UnixStream>, timeout: Duration) -> serde_json::Value {
reader.get_ref().set_read_timeout(Some(timeout)).unwrap();
let mut line = String::new();
reader.read_line(&mut line).unwrap();
serde_json::from_str(&line).unwrap()
}
fn wait_for_event(
reader: &mut BufReader<UnixStream>,
expected: &str,
timeout: Duration,
) -> serde_json::Value {
let deadline = Instant::now() + timeout;
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
let value = read_json_line(reader, remaining.max(Duration::from_millis(1)));
if value["event"] == expected {
return value;
}
}
}
#[test]
fn ping_over_socket_returns_version() {
let base = unique_test_dir();
@ -244,3 +288,256 @@ fn workspace_list_and_create_round_trip() {
let _ = child.child.wait();
let _ = fs::remove_dir_all(base);
}
#[test]
fn events_subscribe_streams_lifecycle_and_agent_events() {
let base = unique_test_dir();
let config_home = base.join("config");
let runtime_dir = base.join("runtime");
let socket_path = runtime_dir.join("herdr.sock");
let bin_dir = base.join("bin");
fs::create_dir_all(&bin_dir).unwrap();
let fake_pi = bin_dir.join("pi");
fs::write(
&fake_pi,
"#!/bin/sh\nprintf 'Working...\\n'\nsleep 1\nprintf '\\033[2J\\033[Hdone\\n'\n",
)
.unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&fake_pi).unwrap().permissions();
perms.set_mode(0o755);
fs::set_permissions(&fake_pi, perms).unwrap();
}
let inherited_path = std::env::var("PATH").unwrap_or_default();
let path_override = format!("{}:{}", bin_dir.display(), inherited_path);
let mut child = spawn_herdr_with_path(
&config_home,
&runtime_dir,
&socket_path,
Some(Path::new(&path_override)),
);
wait_for_socket(&socket_path, Duration::from_secs(5));
let (_stream, mut reader) = open_subscription(
&socket_path,
r#"{"id":"sub_life","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"},{"type":"workspace.focused"},{"type":"pane.created"},{"type":"pane.focused"},{"type":"pane.agent_detected"},{"type":"pane.closed"},{"type":"workspace.closed"}]}}"#,
);
let ack = read_json_line(&mut reader, Duration::from_secs(2));
assert_eq!(ack["id"], "sub_life");
assert_eq!(ack["result"]["type"], "subscription_started");
let created = send_request(
&socket_path,
&format!(
r#"{{"id":"req_l1","method":"workspace.create","params":{{"cwd":"{}","focus":true}}}}"#,
base.display()
),
);
let workspace_id = created["result"]["workspace"]["workspace_id"]
.as_str()
.unwrap()
.to_string();
let workspace_created =
wait_for_event(&mut reader, "workspace_created", Duration::from_secs(2));
assert_eq!(
workspace_created["data"]["workspace"]["workspace_id"],
workspace_id
);
let workspace_focused =
wait_for_event(&mut reader, "workspace_focused", Duration::from_secs(2));
assert_eq!(workspace_focused["data"]["workspace_id"], workspace_id);
let pane_created = wait_for_event(&mut reader, "pane_created", Duration::from_secs(2));
let pane_id = pane_created["data"]["pane"]["pane_id"]
.as_str()
.unwrap()
.to_string();
let pane_focused = wait_for_event(&mut reader, "pane_focused", Duration::from_secs(2));
assert_eq!(pane_focused["data"]["pane_id"], pane_id);
let send_pi = send_request(
&socket_path,
&format!(
r#"{{"id":"req_l2","method":"pane.send_text","params":{{"pane_id":"{}","text":"pi"}}}}"#,
pane_id
),
);
assert_eq!(send_pi["result"]["type"], "ok");
let send_enter = send_request(
&socket_path,
&format!(
r#"{{"id":"req_l3","method":"pane.send_keys","params":{{"pane_id":"{}","keys":["Enter"]}}}}"#,
pane_id
),
);
assert_eq!(send_enter["result"]["type"], "ok");
let agent_detected = wait_for_event(&mut reader, "pane_agent_detected", Duration::from_secs(3));
assert_eq!(agent_detected["data"]["pane_id"], pane_id);
assert_eq!(agent_detected["data"]["agent"], "pi");
let split = send_request(
&socket_path,
&format!(
r#"{{"id":"req_l4","method":"pane.split","params":{{"target_pane_id":"{}","direction":"right","focus":true}}}}"#,
pane_id
),
);
let split_pane_id = split["result"]["pane"]["pane_id"]
.as_str()
.unwrap()
.to_string();
let split_created = wait_for_event(&mut reader, "pane_created", Duration::from_secs(2));
assert_eq!(split_created["data"]["pane"]["pane_id"], split_pane_id);
let closed = send_request(
&socket_path,
&format!(
r#"{{"id":"req_l5","method":"pane.close","params":{{"pane_id":"{}"}}}}"#,
split_pane_id
),
);
assert_eq!(closed["result"]["type"], "ok");
let pane_closed = wait_for_event(&mut reader, "pane_closed", Duration::from_secs(2));
assert_eq!(pane_closed["data"]["pane_id"], split_pane_id);
let closed_ws = send_request(
&socket_path,
&format!(
r#"{{"id":"req_l6","method":"workspace.close","params":{{"workspace_id":"{}"}}}}"#,
workspace_id
),
);
assert_eq!(closed_ws["result"]["type"], "ok");
let workspace_closed = wait_for_event(&mut reader, "workspace_closed", Duration::from_secs(2));
assert_eq!(workspace_closed["data"]["workspace_id"], workspace_id);
let _ = child.child.kill();
let _ = child.child.wait();
let _ = fs::remove_dir_all(base);
}
#[test]
fn events_subscribe_streams_output_and_agent_state_events() {
let base = unique_test_dir();
let config_home = base.join("config");
let runtime_dir = base.join("runtime");
let socket_path = runtime_dir.join("herdr.sock");
let bin_dir = base.join("bin");
fs::create_dir_all(&bin_dir).unwrap();
let fake_pi = bin_dir.join("pi");
fs::write(
&fake_pi,
"#!/bin/sh\nprintf 'Working...\\n'\nsleep 1\nprintf '\\033[2J\\033[Hdone\\n'\n",
)
.unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&fake_pi).unwrap().permissions();
perms.set_mode(0o755);
fs::set_permissions(&fake_pi, perms).unwrap();
}
let inherited_path = std::env::var("PATH").unwrap_or_default();
let path_override = format!("{}:{}", bin_dir.display(), inherited_path);
let mut child = spawn_herdr_with_path(
&config_home,
&runtime_dir,
&socket_path,
Some(Path::new(&path_override)),
);
wait_for_socket(&socket_path, Duration::from_secs(5));
let created = send_request(
&socket_path,
&format!(
r#"{{"id":"req_20","method":"workspace.create","params":{{"cwd":"{}","focus":true}}}}"#,
base.display()
),
);
assert_eq!(created["result"]["workspace"]["workspace_id"], "w_1");
let panes = send_request(
&socket_path,
r#"{"id":"req_21","method":"pane.list","params":{}}"#,
);
let pane_id = panes["result"]["panes"][0]["pane_id"]
.as_str()
.unwrap()
.to_string();
let (_stream, mut reader) = open_subscription(
&socket_path,
&format!(
r#"{{"id":"sub_1","method":"events.subscribe","params":{{"subscriptions":[{{"type":"pane.output_matched","pane_id":"{}","source":"recent","lines":40,"match":{{"type":"substring","value":"hello from socket"}}}},{{"type":"pane.agent_state_changed","pane_id":"{}","state":"idle"}}]}}}}"#,
pane_id, pane_id,
),
);
let ack = read_json_line(&mut reader, Duration::from_secs(2));
assert_eq!(ack["id"], "sub_1");
assert_eq!(ack["result"]["type"], "subscription_started");
let send_text = send_request(
&socket_path,
&format!(
r#"{{"id":"req_22","method":"pane.send_text","params":{{"pane_id":"{}","text":"echo hello from socket"}}}}"#,
pane_id
),
);
assert_eq!(send_text["result"]["type"], "ok");
let send_enter = send_request(
&socket_path,
&format!(
r#"{{"id":"req_23","method":"pane.send_keys","params":{{"pane_id":"{}","keys":["Enter"]}}}}"#,
pane_id
),
);
assert_eq!(send_enter["result"]["type"], "ok");
let output_event = read_json_line(&mut reader, Duration::from_secs(3));
assert_eq!(output_event["event"], "pane.output_matched");
assert_eq!(output_event["data"]["pane_id"], pane_id);
assert!(output_event["data"]["matched_line"]
.as_str()
.unwrap()
.contains("hello from socket"));
assert!(output_event["data"]["read"]["text"]
.as_str()
.unwrap()
.contains("hello from socket"));
let send_pi = send_request(
&socket_path,
&format!(
r#"{{"id":"req_24","method":"pane.send_text","params":{{"pane_id":"{}","text":"pi"}}}}"#,
pane_id
),
);
assert_eq!(send_pi["result"]["type"], "ok");
let send_enter = send_request(
&socket_path,
&format!(
r#"{{"id":"req_25","method":"pane.send_keys","params":{{"pane_id":"{}","keys":["Enter"]}}}}"#,
pane_id
),
);
assert_eq!(send_enter["result"]["type"], "ok");
let agent_idle = read_json_line(&mut reader, Duration::from_secs(8));
assert_eq!(agent_idle["event"], "pane.agent_state_changed");
assert_eq!(agent_idle["data"]["pane_id"], pane_id);
assert_eq!(agent_idle["data"]["state"], "idle");
assert_eq!(agent_idle["data"]["agent"], "pi");
let _ = child.child.kill();
let _ = child.child.wait();
let _ = fs::remove_dir_all(base);
}