aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docs/next/website/src/content/docs/ja/socket-api.mdx10
-rw-r--r--docs/next/website/src/content/docs/socket-api.mdx40
-rw-r--r--docs/next/website/src/content/docs/zh-cn/socket-api.mdx10
-rw-r--r--src/api/event_hub.rs81
-rw-r--r--src/api/server.rs23
-rw-r--r--src/api/server/subscription_socket_tests.rs355
-rw-r--r--src/api/subscriptions.rs255
7 files changed, 717 insertions, 57 deletions
diff --git a/docs/next/website/src/content/docs/ja/socket-api.mdx b/docs/next/website/src/content/docs/ja/socket-api.mdx
index a556100c..09532d5e 100644
--- a/docs/next/website/src/content/docs/ja/socket-api.mdx
+++ b/docs/next/website/src/content/docs/ja/socket-api.mdx
@@ -111,7 +111,7 @@ herdr pane read w1:p2 --source recent --lines 50
`workspace.move_block` は、順序付きの `workspace_ids` を `before_workspace_id` の前へアトミックに移動します。アンカーを省略するとブロックを末尾へ移動します。id は一意である必要があり、アンカーをブロックに含めることはできません。レスポンスにはサーバーが確定した順序付きワークスペース一覧が含まれます。
-`session.snapshot` は、独自のローカルランタイムキャッシュを持つクライアント向けに、一度限りのブートストラップスナップショットを返します。レスポンスには、バージョン/プロトコルメタデータ、フォーカス中のワークスペース/タブ/ペイン id、ワークスペース、タブ、ペイン、タブレイアウト、エージェントの各レコードが含まれます。これは購読ではありません。ブートストラップ中のイベントを取りこぼさないよう、まず別の接続で `events.subscribe` を開いて確認応答を待ち、そのストリームをバッファしながら `session.snapshot` を呼び出します。スナップショットを適用してから、バッファしたイベントを順番に適用し、そのままストリーミングを続けます。再接続後やキャッシュが古い可能性がある場合は、`session.snapshot` を再度呼び出します。ワークスペースレコードには、関連付けられた worktree の出自情報も含まれます。リポジトリ全体の worktree 検出には引き続き `worktree.list` を使います。
+`session.snapshot` は、独自のローカルランタイムキャッシュを持つクライアント向けに、一度限りのブートストラップスナップショットを返します。レスポンスには、バージョン/プロトコルメタデータ、フォーカス中のワークスペース/タブ/ペイン id、ワークスペース、タブ、ペイン、タブレイアウト、エージェントの各レコードが含まれます。これは購読ではありません。キャッシュの初期化と復旧では、先に購読を開始し、後述のイベント購読セクションの手順でサーバーから現在の状態を読み取り、キャッシュを照合・更新します。イベントはキャッシュの無効化を知らせるものであり、スナップショットに無条件で再適用できるものではありません。再接続後やキャッシュが古い可能性がある場合は、`session.snapshot` を再度呼び出します。ワークスペースレコードには、関連付けられた worktree の出自情報も含まれます。リポジトリ全体の worktree 検出には引き続き `worktree.list` を使います。
CLI の `herdr api snapshot` は、クライアントやエージェントが簡単にブートストラップできるよう、現在の `session.snapshot` レスポンスを JSON で出力します。
@@ -639,10 +639,16 @@ workspace の get/list 応答は結果の `tokens` マップを公開し、ス
最初のレスポンスは購読の確認応答です。以降の行はプッシュされるイベントです。ライフサイクルイベントの購読はリクエストが受理された時点から始まり、それ以前に保持されていたイベントは再生しません。
+ライフサイクルとエージェント状態のイベントは、各購読内の順序を保った上限付きのバッチで送信されます。イベント履歴は永続ログではありません。購読の初期化中も含め、購読者が保持範囲より遅れた場合、サーバーは元のリクエスト `id` と `error.code: "events_lost"` を含むエラーレスポンスを送り、その購読接続を閉じます。欠落したイベントを通知せずに配信を続けることはありません。他のクライアント接続には影響しません。履歴はイベント種別間で共有されるため、失われたイベントが購読条件に一致しなかった可能性があっても、保持範囲を超えたことを通知します。
+
+`events_lost` を受け取ったら、キャッシュを古い状態として扱います。新しい購読を開始して `subscription_started` を待ち、読み取りを続けながら別の接続で `session.snapshot` を取得してください。サーバーの現在の状態でキャッシュを照合し、スナップショットでキャッシュ全体を置き換えます。イベントはキャッシュの無効化通知として扱い、再取得の契機にします。取得は直列に行い、取得中にイベントが届いた場合は完了後に再取得してください。
+
+スナップショットとイベントに共通のシーケンス境界はありません。順序が保証されるのは各購読項目内のみで、項目間やスナップショットとの前後関係は保証されません。バッファしたイベントの内容をスナップショットへ無条件に再適用しないでください。復旧できるのは現在の状態であり、失われたイベント履歴ではありません。再び `events_lost` を受け取った場合は、再購読して状態を照合し直します。読み取りが遅いクライアントでは、読み取り速度を上げるかイベントの負荷を減らす必要があります。
+
ワークスペースのイベント購読には `workspace.created`、`workspace.updated`、`workspace.metadata_updated`、`workspace.renamed`、`workspace.moved`、`workspace.reordered`、`workspace.closed`、`workspace.focused` があります。`workspace.metadata_updated` はプラグインイベントフックを実行せずにトークン変更と TTL 失効を報告します。その他のワークスペースイベントは Herdr の UI/ランタイムのライフサイクルを記述します。ワークスペースが worktree グループに属している場合、`workspace.created` は任意の `workspace.worktree` 出自情報を含みます。`workspace.moved` は、移動した `workspace_id`、要求された `insert_index`、更新済みの順序付き `workspaces` リストを含みます。`workspace.reordered` は、アトミックに移動した `workspace_ids`、省略可能な `before_workspace_id`、サーバーが確定した順序付き `workspaces` リストを含みます。削除前に Herdr がまだ識別できる場合、`workspace.closed` は最終的な `workspace` スナップショットを含みます。
タブのイベント購読には `tab.created`、`tab.closed`、`tab.focused`、`tab.renamed`、`tab.moved` があります。`tab.moved` は、移動した `tab_id`、`workspace_id`、要求された `insert_index`、そのワークスペースの更新済みの順序付き `tabs` リストを含みます。
ペインのイベント購読には `pane.created`、`pane.updated`、`pane.closed`、`pane.focused`、`pane.moved`、`pane.exited`、`pane.agent_detected`、`pane.output_matched`、`pane.agent_status_changed`、`pane.scroll_changed` があります。`pane.focused` は、このサーバーに接続した任意のクライアントでの手動ペイン選択も報告します。他のクライアントの表示は移動させず、ペイロードにクライアントの識別情報は含みません。選択済みのペインを再度選んでも発行しません。ターミナルタイトルの変更は `pane.updated` を発行することがありますが、生のタイトルでスピナーだけが変化し、`terminal_title_stripped` が変わらない場合は発行しません。`pane.scroll_changed` は 1 つの `pane_id` を対象とし、Herdr がスクロールスナップショットの変化を検出するたびに `pane_id`、`workspace_id`、現在の `scroll` 情報を発行します。
-レイアウトのイベント購読には `layout.updated` があります。このイベントは 1 つのタブについて更新済みの `PaneLayoutSnapshot` を運びます。`session.snapshot` でブートストラップするクライアントは、同じ `workspace_id` と `tab_id` のキャッシュ済みレイアウトを置き換えてください。
+レイアウトのイベント購読には `layout.updated` があります。このイベントは 1 つのタブについて更新済みの `PaneLayoutSnapshot` を運びます。`session.snapshot` でキャッシュを管理するクライアントは、このイベントを契機にサーバーから対象のレイアウトを再取得してください。遅れて届いたペイロードは、最新のスナップショットより古い可能性があります。
worktree のイベント購読には `worktree.created`、`worktree.opened`、`worktree.removed` があります。worktree イベントは Git チェックアウトのライフサイクルを記述します。`worktree.created` は開かれた `workspace` と作成された `worktree` を含みます。`worktree.opened` は対象の `workspace`、開かれた `worktree`、`already_open` を含みます。`worktree.removed` は `workspace_id`、削除された `worktree`、`forced` を含みます。
diff --git a/docs/next/website/src/content/docs/socket-api.mdx b/docs/next/website/src/content/docs/socket-api.mdx
index 84068cea..e6ae6a57 100644
--- a/docs/next/website/src/content/docs/socket-api.mdx
+++ b/docs/next/website/src/content/docs/socket-api.mdx
@@ -118,12 +118,11 @@ Raw socket method names use dot notation:
`session.snapshot` returns a one-time bootstrap snapshot for clients that keep
their own local runtime cache. The response includes version/protocol metadata,
focused workspace/tab/pane ids, workspace records, tab records, pane records,
-tab layout snapshots, and agent records. It is not a subscription. To avoid a
-bootstrap gap, first open `events.subscribe` on another connection and wait for
-its acknowledgement. Buffer that stream while calling `session.snapshot`,
-install the snapshot, then apply the buffered events in order and continue
-streaming. Call `session.snapshot` again after reconnecting or when the local
-cache may be stale. Attached worktree provenance is included on workspace
+tab layout snapshots, and agent records. It is not a subscription. For cache
+bootstrap and recovery, subscribe first, then reconcile through authoritative
+reads as described under Event subscriptions below. Events invalidate cached state; they
+cannot be unconditionally replayed over a snapshot. Call `session.snapshot`
+again after reconnecting or when the local cache may be stale. Attached worktree provenance is included on workspace
records. Full repo worktree discovery remains `worktree.list`.
From the CLI, `herdr api snapshot` prints the live `session.snapshot` response
@@ -825,6 +824,29 @@ and retry rather than assuming the other entries were subscribed.
Lifecycle subscriptions start when the request is accepted and do not replay
events retained before that point.
+Lifecycle and agent-status events are drained in bounded batches, in order within
+each subscription. Event history is not durable. If a subscriber falls behind
+the retained history, including while its subscriptions are being initialized,
+the server sends an error response with the original request `id` and
+`error.code: "events_lost"`, then closes that subscription connection instead of
+silently continuing with missing events. Other clients remain connected.
+The history is shared across event types, so an overrun is reported even when
+the evicted events might not have matched this subscription's filters.
+
+On `events_lost`, treat the cached state as stale. Open a new subscription,
+wait for `subscription_started`, and keep reading while requesting
+`session.snapshot` on another connection. Use authoritative reads to reconcile
+current state: replace the cache with the snapshot, and treat incoming events
+as invalidation signals that trigger another read. Serialize refreshes and
+refresh again if events arrive while a read is in flight.
+
+Snapshots and events have no shared sequence boundary. Ordering is guaranteed
+within each subscription entry, not across entries or relative to a snapshot,
+so do not unconditionally replay buffered event payloads onto the snapshot.
+Recovery restores current state, not the missing event history. Another
+`events_lost` requires resubscribing and reconciling again. Slow readers may
+need to increase read throughput or reduce event load.
+
Workspace event subscriptions include `workspace.created`, `workspace.updated`, `workspace.metadata_updated`, `workspace.renamed`, `workspace.moved`, `workspace.reordered`, `workspace.closed`, and `workspace.focused`. `workspace.metadata_updated` reports token changes and TTL expiry without invoking plugin event hooks. Other workspace events describe Herdr UI/runtime lifecycle. `workspace.created` includes optional `workspace.worktree` provenance when the workspace belongs to a worktree group. `workspace.moved` includes the moved `workspace_id`, requested `insert_index`, and updated ordered `workspaces` list. `workspace.reordered` includes the atomically moved `workspace_ids`, optional `before_workspace_id`, and authoritative ordered `workspaces` list. `workspace.closed` includes a final `workspace` snapshot when Herdr can still identify it before removal.
Tab event subscriptions include `tab.created`, `tab.closed`, `tab.focused`,
`tab.renamed`, and `tab.moved`. `tab.moved` includes the moved `tab_id`,
@@ -841,9 +863,9 @@ Terminal-title changes can emit `pane.updated`, but spinner-only raw-title chang
`workspace_id`, and the current `scroll` metrics whenever Herdr observes a
changed scroll snapshot.
Layout event subscriptions include `layout.updated`. The event carries the
-updated `PaneLayoutSnapshot` for one tab. Clients that bootstrap with
-`session.snapshot` should replace the cached layout with the same
-`workspace_id` and `tab_id`.
+updated `PaneLayoutSnapshot` for one tab. Clients maintaining a cache with
+`session.snapshot` should use this event to refresh the affected layout through
+an authoritative read; a delayed payload may predate their latest snapshot.
Worktree event subscriptions include `worktree.created`, `worktree.opened`, and `worktree.removed`. Worktree events describe Git checkout lifecycle. `worktree.created` includes the opened `workspace` and created `worktree`. `worktree.opened` includes the target `workspace`, opened `worktree`, and `already_open`. `worktree.removed` includes the `workspace_id`, removed `worktree`, and `forced`.
diff --git a/docs/next/website/src/content/docs/zh-cn/socket-api.mdx b/docs/next/website/src/content/docs/zh-cn/socket-api.mdx
index 43400eaa..3aae7131 100644
--- a/docs/next/website/src/content/docs/zh-cn/socket-api.mdx
+++ b/docs/next/website/src/content/docs/zh-cn/socket-api.mdx
@@ -111,7 +111,7 @@ herdr pane read w1:p2 --source recent --lines 50
`workspace.move_block` 会将有序的 `workspace_ids` 原子地移动到 `before_workspace_id` 之前;省略锚点则将该块移动到末尾。id 必须唯一,且锚点不能属于被移动的块。响应包含服务器确认的有序工作区列表。
-`session.snapshot` 为维护本地运行时缓存的客户端返回一次性引导快照。响应包含版本/协议元数据、当前聚焦的工作区/标签页/窗格 id、工作区记录、标签页记录、窗格记录、标签页布局快照和智能体记录。它不是订阅。为避免引导期间漏掉事件,应先在另一个连接上打开 `events.subscribe` 并等待确认,在调用 `session.snapshot` 时缓冲该事件流,然后安装快照、按顺序应用缓冲的事件并继续接收事件。重新连接后或本地缓存可能已过期时,再次调用 `session.snapshot`。工作区记录也包含关联 worktree 的来源信息。完整的仓库 worktree 发现仍使用 `worktree.list`。
+`session.snapshot` 为维护本地运行时缓存的客户端返回一次性引导快照。响应包含版本/协议元数据、当前聚焦的工作区/标签页/窗格 id、工作区记录、标签页记录、窗格记录、标签页布局快照和智能体记录。它不是订阅。初始化或恢复缓存时,应先建立订阅,再按下文事件订阅部分的说明读取服务器当前状态并核对、更新缓存。事件表示缓存可能过期,不能无条件重放到快照上。重新连接后或本地缓存可能已过期时,再次调用 `session.snapshot`。工作区记录也包含关联 worktree 的来源信息。完整的仓库 worktree 发现仍使用 `worktree.list`。
CLI 的 `herdr api snapshot` 会把当前 `session.snapshot` 响应输出为 JSON,让客户端和智能体可以简单地完成引导。
@@ -635,10 +635,16 @@ workspace 的 get/list 响应会公开生成的 `tokens` 映射,空间侧边栏
第一个响应确认订阅。之后的行是推送的事件。生命周期事件订阅从请求被接受时开始,不会重放在此之前保留的事件。
+生命周期和智能体状态事件按有界批次发送,每个订阅内保持事件顺序。事件历史不是持久日志。如果订阅者落后到保留范围之外,包括订阅初始化期间,服务器会发送包含原请求 `id` 和 `error.code: "events_lost"` 的错误响应,然后关闭该订阅连接,而不是静默跳过缺失事件后继续发送。其他客户端的连接不受影响。各类事件共享历史缓冲,因此即使被淘汰的事件可能不匹配此订阅的过滤条件,仍会报告溢出。
+
+收到 `events_lost` 后,将缓存状态视为过期。建立新订阅并等待 `subscription_started`,然后持续读取,同时在另一条连接上请求 `session.snapshot`。通过服务器返回的当前状态重新核对缓存：用快照整体替换缓存,将收到的事件作为缓存失效通知并触发再次读取。串行执行刷新；如果读取期间又收到事件,完成后再次刷新。
+
+快照和事件没有共同的序列边界。顺序仅在每个订阅条目内保证,不保证跨条目或相对于快照的顺序,因此不要把缓冲的事件载荷无条件重放到快照上。恢复得到的是当前状态,无法补回丢失的事件历史。再次收到 `events_lost` 时,需要重新订阅并核对状态。读取过慢的客户端可能需要提高读取速度或减少事件负载。
+
工作区事件订阅包括 `workspace.created`、`workspace.updated`、`workspace.metadata_updated`、`workspace.renamed`、`workspace.moved`、`workspace.reordered`、`workspace.closed` 和 `workspace.focused`。`workspace.metadata_updated` 报告令牌变更和 TTL 到期,但不会调用插件事件钩子。其他工作区事件描述 Herdr UI/运行时的生命周期。当工作区属于 worktree 组时,`workspace.created` 包含可选的 `workspace.worktree` 来源信息。`workspace.moved` 包含被移动的 `workspace_id`、请求的 `insert_index` 和更新后的有序 `workspaces` 列表。`workspace.reordered` 包含原子移动的 `workspace_ids`、可选的 `before_workspace_id` 和服务器确认的有序 `workspaces` 列表。在移除前 Herdr 仍能识别时,`workspace.closed` 包含最终的 `workspace` 快照。
标签页事件订阅包括 `tab.created`、`tab.closed`、`tab.focused`、`tab.renamed` 和 `tab.moved`。`tab.moved` 包含被移动的 `tab_id`、`workspace_id`、请求的 `insert_index` 和该工作区更新后的有序 `tabs` 列表。
窗格事件订阅包括 `pane.created`、`pane.updated`、`pane.closed`、`pane.focused`、`pane.moved`、`pane.exited`、`pane.agent_detected`、`pane.output_matched`、`pane.agent_status_changed` 和 `pane.scroll_changed`。`pane.focused` 也会上报任意连接到此服务器的客户端手动选择窗格的变化，不会移动其他客户端的视图，且载荷不包含客户端身份。重复选择已选中的窗格不会再次发出事件。终端标题变化可能发出 `pane.updated`,但如果原始标题只有旋转指示符发生变化且 `terminal_title_stripped` 不变,则不会发出。`pane.scroll_changed` 只针对一个 `pane_id`,每当 Herdr 观察到滚动快照变化时,都会发出 `pane_id`、`workspace_id` 和当前 `scroll` 指标。
-布局事件订阅包括 `layout.updated`。该事件携带一个标签页更新后的 `PaneLayoutSnapshot`。使用 `session.snapshot` 引导的客户端应替换具有相同 `workspace_id` 和 `tab_id` 的缓存布局。
+布局事件订阅包括 `layout.updated`。该事件携带一个标签页更新后的 `PaneLayoutSnapshot`。使用 `session.snapshot` 维护缓存的客户端应将该事件作为重新读取受影响布局的通知；延迟到达的载荷可能早于最新快照。
Worktree 事件订阅包括 `worktree.created`、`worktree.opened` 和 `worktree.removed`。Worktree 事件描述 Git 检出的生命周期。`worktree.created` 包含打开的 `workspace` 和创建的 `worktree`。`worktree.opened` 包含目标 `workspace`、打开的 `worktree` 和 `already_open`。`worktree.removed` 包含 `workspace_id`、被移除的 `worktree` 和 `forced`。
diff --git a/src/api/event_hub.rs b/src/api/event_hub.rs
index 42cf47a2..29843313 100644
--- a/src/api/event_hub.rs
+++ b/src/api/event_hub.rs
@@ -9,6 +9,12 @@ struct EventHubState {
events: Vec<(u64, crate::api::schema::EventEnvelope)>,
}
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum EventHistoryError {
+ Lost,
+ Unavailable,
+}
+
impl EventHub {
const MAX_EVENTS: usize = 512;
@@ -37,6 +43,29 @@ impl EventHub {
.collect()
}
+ pub(super) fn events_after_checked(
+ &self,
+ sequence: u64,
+ ) -> Result<Vec<(u64, crate::api::schema::EventEnvelope)>, EventHistoryError> {
+ let state = self
+ .inner
+ .lock()
+ .map_err(|_| EventHistoryError::Unavailable)?;
+ if state
+ .events
+ .first()
+ .is_some_and(|(first, _)| sequence < first.saturating_sub(1))
+ {
+ return Err(EventHistoryError::Lost);
+ }
+ Ok(state
+ .events
+ .iter()
+ .filter(|(event_sequence, _)| *event_sequence > sequence)
+ .cloned()
+ .collect())
+ }
+
pub fn current_sequence(&self) -> u64 {
let Ok(state) = self.inner.lock() else {
return 0;
@@ -44,3 +73,55 @@ impl EventHub {
state.next_sequence
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::api::schema::{EventData, EventEnvelope, EventKind};
+
+ fn event() -> EventEnvelope {
+ EventEnvelope {
+ event: EventKind::WorkspaceFocused,
+ data: EventData::WorkspaceFocused {
+ workspace_id: "workspace_1".into(),
+ },
+ }
+ }
+
+ #[test]
+ fn checked_history_distinguishes_retained_boundary_from_lost_events() {
+ let hub = EventHub::default();
+ assert!(hub.events_after_checked(0).unwrap().is_empty());
+ for _ in 0..EventHub::MAX_EVENTS {
+ hub.push(event());
+ }
+ assert_eq!(
+ hub.events_after_checked(0).unwrap().len(),
+ EventHub::MAX_EVENTS
+ );
+ hub.push(event());
+ assert_eq!(hub.events_after_checked(0), Err(EventHistoryError::Lost));
+ let retained = hub.events_after_checked(1).unwrap();
+ assert_eq!(retained.len(), EventHub::MAX_EVENTS);
+ assert_eq!(retained.first().unwrap().0, 2);
+ assert_eq!(retained.last().unwrap().0, hub.current_sequence());
+ assert!(hub
+ .events_after_checked(hub.current_sequence())
+ .unwrap()
+ .is_empty());
+ }
+
+ #[test]
+ fn checked_history_reports_unavailable_instead_of_empty_after_poison() {
+ let hub = EventHub::default();
+ assert!(std::panic::catch_unwind(|| {
+ let _guard = hub.inner.lock().unwrap();
+ panic!("poison the test event history");
+ })
+ .is_err());
+ assert_eq!(
+ hub.events_after_checked(0),
+ Err(EventHistoryError::Unavailable)
+ );
+ }
+}
diff --git a/src/api/server.rs b/src/api/server.rs
index 03724813..44a2a0df 100644
--- a/src/api/server.rs
+++ b/src/api/server.rs
@@ -24,6 +24,9 @@ use crate::ipc::{
mod pane_graphics_stream;
+#[cfg(test)]
+mod subscription_socket_tests;
+
const SOCKET_PERMISSION_MODE: u32 = 0o600;
pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100);
pub(super) const APP_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
@@ -749,7 +752,7 @@ fn stream_subscriptions(
if let Err(err) = write_json_line(
&mut stream,
&SuccessResponse {
- id: request_id,
+ id: request_id.clone(),
result: ResponseResult::SubscriptionStarted {},
},
) {
@@ -765,7 +768,23 @@ fn stream_subscriptions(
}
for subscription in &mut subscriptions {
- if let Some(event) = subscription.poll(api_tx, event_hub) {
+ let events = match subscription.poll_batch(api_tx, event_hub) {
+ Ok(events) => events,
+ Err(error) => {
+ write_json_line_allow_disconnect(
+ &mut stream,
+ &ErrorResponse {
+ id: request_id,
+ error,
+ },
+ )?;
+ return Ok(());
+ }
+ };
+ for event in events {
+ if should_stop_connection(&mut stream, running)? {
+ return Ok(());
+ }
if let Err(err) = write_json_line(&mut stream, &event) {
if is_connection_closed_error(&err) {
return Ok(());
diff --git a/src/api/server/subscription_socket_tests.rs b/src/api/server/subscription_socket_tests.rs
new file mode 100644
index 00000000..5cfbc706
--- /dev/null
+++ b/src/api/server/subscription_socket_tests.rs
@@ -0,0 +1,355 @@
+use super::*;
+use crate::api::schema::{
+ AgentStatus, EventData, EventEnvelope, EventKind, PaneInfo, PaneReadResult, ReadFormat,
+ ReadSource,
+};
+use crate::ipc::{poll_local_stream_read_count, LocalStreamReadCount};
+use interprocess::local_socket::traits::Listener as _;
+use serde_json::{json, Value};
+use std::sync::atomic::AtomicU64;
+use tokio::sync::mpsc;
+
+const RESPONSE_TIMEOUT: Duration = Duration::from_secs(2);
+
+struct SocketTest {
+ hub: EventHub,
+ running: Arc<AtomicBool>,
+ api_tx: ApiRequestSender,
+ api_rx: mpsc::UnboundedReceiver<ApiRequestMessage>,
+ workers: Vec<std::thread::JoinHandle<io::Result<()>>>,
+ paths: Vec<PathBuf>,
+}
+
+impl SocketTest {
+ fn new() -> Self {
+ let (api_tx, api_rx) = mpsc::unbounded_channel();
+ Self {
+ hub: EventHub::default(),
+ running: Arc::new(AtomicBool::new(true)),
+ api_tx,
+ api_rx,
+ workers: Vec::new(),
+ paths: Vec::new(),
+ }
+ }
+
+ fn connect(&mut self) -> Client {
+ static NEXT_SOCKET: AtomicU64 = AtomicU64::new(0);
+ let path = std::env::temp_dir().join(format!(
+ "herdr-sub-{}-{}",
+ std::process::id(),
+ NEXT_SOCKET.fetch_add(1, Ordering::Relaxed)
+ ));
+ let listener = bind_local_listener(&path).unwrap();
+ self.paths.push(path.clone());
+ let mut stream = crate::ipc::connect_local_stream(&path).unwrap();
+ let server = listener.accept().unwrap();
+ set_local_stream_polling(&mut stream, true).unwrap();
+ let api_tx = self.api_tx.clone();
+ let hub = self.hub.clone();
+ let running = Arc::clone(&self.running);
+ let worker =
+ std::thread::spawn(move || handle_connection(server, &api_tx, &hub, &running, None));
+ self.workers.push(worker);
+ Client {
+ stream,
+ buffered: Vec::new(),
+ }
+ }
+
+ fn app_request(&mut self) -> ApiRequestMessage {
+ let deadline = Instant::now() + RESPONSE_TIMEOUT;
+ loop {
+ match self.api_rx.try_recv() {
+ Ok(request) => return request,
+ Err(mpsc::error::TryRecvError::Empty) => {
+ assert!(
+ Instant::now() < deadline,
+ "timed out waiting for app request"
+ );
+ std::thread::sleep(Duration::from_millis(1));
+ }
+ Err(error) => panic!("app request channel closed: {error}"),
+ }
+ }
+ }
+}
+
+impl Drop for SocketTest {
+ fn drop(&mut self) {
+ self.running.store(false, Ordering::Relaxed);
+ self.api_rx.close();
+ while self.api_rx.try_recv().is_ok() {}
+ let deadline = Instant::now() + APP_RESPONSE_TIMEOUT + Duration::from_secs(1);
+ let mut failures = Vec::new();
+ for worker in self.workers.drain(..) {
+ while !worker.is_finished() && Instant::now() < deadline {
+ std::thread::sleep(Duration::from_millis(1));
+ }
+ if worker.is_finished() {
+ match worker.join() {
+ Ok(Ok(())) => {}
+ Ok(Err(error)) => failures.push(format!("connection failed: {error}")),
+ Err(_) => failures.push("connection panicked".into()),
+ }
+ } else {
+ failures.push("subscription connection did not stop".into());
+ }
+ }
+ // Also remove Windows listener marker files, including after a setup failure.
+ for path in self.paths.drain(..) {
+ if let Err(error) = std::fs::remove_file(path) {
+ if error.kind() != io::ErrorKind::NotFound {
+ failures.push(format!("socket cleanup failed: {error}"));
+ }
+ }
+ }
+ if !std::thread::panicking() {
+ assert!(failures.is_empty(), "{failures:?}");
+ }
+ }
+}
+
+struct Client {
+ stream: LocalStream,
+ buffered: Vec<u8>,
+}
+
+impl Client {
+ fn send(&mut self, request: Value) {
+ writeln!(self.stream, "{request}").unwrap();
+ }
+
+ fn subscribe(&mut self, id: &str, subscriptions: Value) {
+ self.send(json!({
+ "id": id,
+ "method": "events.subscribe",
+ "params": {"subscriptions": subscriptions}
+ }));
+ }
+
+ fn next_line(&mut self, deadline: Instant) -> Option<Value> {
+ // LocalStream recv timeouts are unsupported on Windows. Use the same bounded
+ // nonblocking/PeekNamedPipe reads as the API client, retaining partial JSON lines.
+ loop {
+ if let Some(end) = self.buffered.iter().position(|byte| *byte == b'\n') {
+ let line: Vec<_> = self.buffered.drain(..=end).collect();
+ return Some(serde_json::from_slice(&line).expect("subscription JSON"));
+ }
+ assert!(
+ Instant::now() < deadline,
+ "timed out waiting for socket response"
+ );
+ let mut bytes = [0; 4096];
+ match poll_local_stream_read_count(&mut self.stream, &mut bytes).unwrap() {
+ LocalStreamReadCount::Data(count) => {
+ self.buffered.extend_from_slice(&bytes[..count])
+ }
+ LocalStreamReadCount::Pending => std::thread::sleep(Duration::from_millis(1)),
+ LocalStreamReadCount::Closed => {
+ assert!(self.buffered.is_empty(), "incomplete JSON at EOF");
+ return None;
+ }
+ }
+ }
+ }
+
+ fn response(&mut self) -> Value {
+ self.next_line(Instant::now() + RESPONSE_TIMEOUT)
+ .expect("socket response before EOF")
+ }
+
+ fn assert_started(&mut self, id: &str) {
+ let response = self.response();
+ assert_eq!(response["id"], id);
+ assert_eq!(response["result"]["type"], "subscription_started");
+ }
+
+ fn assert_renames(&mut self, indices: std::ops::Range<usize>, deadline: Instant) {
+ for index in indices {
+ let event = self.next_line(deadline).expect("rename before EOF");
+ assert_eq!(event["event"], "workspace_renamed");
+ assert_eq!(event["data"]["label"], format!("flood-{index}"));
+ }
+ }
+
+ fn assert_history_lost(&mut self, id: &str) {
+ let response = self.response();
+ assert_eq!(response["id"], id);
+ assert_eq!(response["error"]["code"], "events_lost", "{response}");
+ assert_eq!(self.next_line(Instant::now() + RESPONSE_TIMEOUT), None);
+ }
+}
+
+fn renamed_event(index: usize) -> EventEnvelope {
+ EventEnvelope {
+ event: EventKind::WorkspaceRenamed,
+ data: EventData::WorkspaceRenamed {
+ workspace_id: "workspace_1".into(),
+ label: format!("flood-{index}"),
+ },
+ }
+}
+
+fn output_subscription() -> Value {
+ json!({
+ "type": "pane.output_matched",
+ "pane_id": "pane_1",
+ "source": "recent",
+ "match": {"type": "substring", "value": "never"}
+ })
+}
+
+fn reply(request: ApiRequestMessage, result: ResponseResult) {
+ request
+ .respond_to
+ .send(
+ serde_json::to_string(&SuccessResponse {
+ id: request.request.id,
+ result,
+ })
+ .unwrap(),
+ )
+ .unwrap();
+}
+
+fn reply_to_probe(request: ApiRequestMessage) {
+ let result = match request.request.method {
+ Method::PaneGet(_) => ResponseResult::PaneInfo {
+ pane: PaneInfo {
+ pane_id: "pane_1".into(),
+ terminal_id: "term_1".into(),
+ workspace_id: "workspace_1".into(),
+ tab_id: "tab_1".into(),
+ focused: true,
+ cwd: None,
+ foreground_cwd: None,
+ label: None,
+ agent: Some("pi".into()),
+ title: None,
+ terminal_title: None,
+ terminal_title_stripped: None,
+ display_agent: None,
+ agent_status: AgentStatus::Working,
+ state_labels: Default::default(),
+ tokens: Default::default(),
+ agent_session: None,
+ scroll: None,
+ revision: 0,
+ },
+ },
+ Method::PaneRead(_) => ResponseResult::PaneRead {
+ read: PaneReadResult {
+ pane_id: "pane_1".into(),
+ workspace_id: "workspace_1".into(),
+ tab_id: "tab_1".into(),
+ source: ReadSource::RecentUnwrapped,
+ format: ReadFormat::Text,
+ text: String::new(),
+ revision: 0,
+ truncated: false,
+ },
+ },
+ ref other => panic!("unexpected subscription probe: {other:?}"),
+ };
+ reply(request, result);
+}
+
+#[test]
+fn subscriptions_drain_retained_bursts_without_per_event_poll_delay() {
+ let mut test = SocketTest::new();
+ let mut client = test.connect();
+ client.subscribe("burst", json!([{"type": "workspace.renamed"}]));
+ client.assert_started("burst");
+ for index in 0..128 {
+ test.hub.push(renamed_event(index));
+ }
+ // One deadline for the entire batch detects a 100 ms delay per event.
+ client.assert_renames(0..128, Instant::now() + RESPONSE_TIMEOUT);
+}
+
+#[test]
+fn subscriptions_report_history_loss_before_sending_a_partial_stream() {
+ assert_subscription_history_loss(false);
+}
+
+#[test]
+fn subscriptions_report_history_loss_before_initial_agent_status() {
+ assert_subscription_history_loss(true);
+}
+
+fn assert_subscription_history_loss(agent_status: bool) {
+ let mut test = SocketTest::new();
+ let mut client = test.connect();
+ let subscriptions = if agent_status {
+ json!([{
+ "type": "pane.agent_status_changed",
+ "pane_id": "pane_1",
+ "agent_status": "working"
+ }])
+ } else {
+ json!([{"type": "workspace.renamed"}, output_subscription()])
+ };
+ client.subscribe("history-gap", subscriptions);
+ // Hold the setup probe after the server pins its subscription cursor.
+ let probe = test.app_request();
+ assert!(probe.request.id.ends_with(":probe"));
+ for index in 0..600 {
+ test.hub.push(renamed_event(index));
+ }
+ reply_to_probe(probe);
+ client.assert_started("history-gap");
+ client.assert_history_lost("history-gap");
+}
+
+#[test]
+fn lagging_subscription_closes_without_interrupting_other_clients() {
+ let mut test = SocketTest::new();
+ let mut healthy = test.connect();
+ healthy.subscribe("healthy", json!([{"type": "workspace.renamed"}]));
+ healthy.assert_started("healthy");
+
+ let mut slow = test.connect();
+ slow.subscribe(
+ "slow",
+ json!([{"type": "workspace.renamed"}, output_subscription()]),
+ );
+ let probe = test.app_request();
+ assert_eq!(probe.request.id, "slow:sub:1:probe");
+ reply_to_probe(probe);
+ slow.assert_started("slow");
+ // Pause only this connection in an existing app request, rather than depending
+ // on OS socket buffer sizes or sleeping to make its event cursor fall behind.
+ let paused_read = test.app_request();
+ assert_eq!(paused_read.request.id, "slow:sub:1:read");
+ assert!(matches!(paused_read.request.method, Method::PaneRead(_)));
+ // All five batches must finish before the held app request can time out.
+ let deadline = Instant::now() + RESPONSE_TIMEOUT;
+ for batch in 0..5 {
+ let indices = batch * 128..(batch + 1) * 128;
+ for index in indices.clone() {
+ test.hub.push(renamed_event(index));
+ }
+ healthy.assert_renames(indices, deadline);
+ }
+
+ reply_to_probe(paused_read);
+ slow.assert_history_lost("slow");
+ test.hub.push(renamed_event(640));
+ healthy.assert_renames(640..641, Instant::now() + RESPONSE_TIMEOUT);
+
+ let mut ordinary = test.connect();
+ ordinary.send(json!({"id": "ordinary", "method": "workspace.list", "params": {}}));
+ let request = test.app_request();
+ assert_eq!(request.request.id, "ordinary");
+ assert!(matches!(request.request.method, Method::WorkspaceList(_)));
+ reply(
+ request,
+ ResponseResult::WorkspaceList {
+ workspaces: Vec::new(),
+ },
+ );
+ let response = ordinary.response();
+ assert_eq!(response["id"], "ordinary");
+ assert_eq!(response["result"]["type"], "workspace_list");
+}
diff --git a/src/api/subscriptions.rs b/src/api/subscriptions.rs
index 0888250e..5be2d523 100644
--- a/src/api/subscriptions.rs
+++ b/src/api/subscriptions.rs
@@ -1,5 +1,6 @@
use regex::Regex;
+use crate::api::event_hub::EventHistoryError;
use crate::api::schema::{
ErrorBody, ErrorResponse, EventKind, Method, PaneAgentStatusChangedEvent,
PaneOutputMatchedEvent, PaneScrollChangedEvent, PaneScrollInfo, Request, Subscription,
@@ -277,6 +278,72 @@ impl ActiveSubscription {
_ => Ok(self.poll(api_tx, event_hub)),
}
}
+
+ pub(super) fn poll_batch(
+ &mut self,
+ api_tx: &ApiRequestSender,
+ event_hub: &EventHub,
+ ) -> Result<Vec<serde_json::Value>, ErrorBody> {
+ match self {
+ Self::Event(subscription) => {
+ let events = subscription_events_after(event_hub, subscription.last_sequence)?;
+ let mut matching = Vec::new();
+ for (sequence, event) in events {
+ subscription.last_sequence = sequence;
+ if event.event == subscription.event_kind {
+ matching.push(serde_json::to_value(event).map_err(event_encoding_error)?);
+ }
+ }
+ Ok(matching)
+ }
+ Self::AgentStatusChanged(subscription) => {
+ let events = subscription_events_after(event_hub, subscription.last_sequence)?;
+ let mut matching = Vec::new();
+ for (sequence, event) in events {
+ subscription.last_sequence = sequence;
+ if let Some(event) = subscription.event_from_history(event) {
+ matching.push(serde_json::to_value(event).map_err(event_encoding_error)?);
+ }
+ }
+ if matching.is_empty() {
+ if let Some(event) =
+ subscription.poll_snapshot(api_tx, event_hub).ok().flatten()
+ {
+ matching.push(serde_json::to_value(event).map_err(event_encoding_error)?);
+ }
+ }
+ Ok(matching)
+ }
+ // These subscriptions sample current state, not retained event history.
+ // Keep their existing cadence even when a lifecycle batch was nonempty.
+ Self::OutputMatched(_) | Self::ScrollChanged(_) => {
+ Ok(self.poll(api_tx, event_hub).into_iter().collect())
+ }
+ }
+ }
+}
+
+fn subscription_events_after(
+ event_hub: &EventHub,
+ sequence: u64,
+) -> Result<Vec<(u64, crate::api::schema::EventEnvelope)>, ErrorBody> {
+ event_hub.events_after_checked(sequence).map_err(|error| match error {
+ EventHistoryError::Lost => ErrorBody {
+ code: "events_lost".into(),
+ message: "event subscription fell behind retained history; resubscribe and resync with session.snapshot".into(),
+ },
+ EventHistoryError::Unavailable => ErrorBody {
+ code: "server_unavailable".into(),
+ message: "event history is unavailable".into(),
+ },
+ })
+}
+
+fn event_encoding_error(error: serde_json::Error) -> ErrorBody {
+ ErrorBody {
+ code: "internal_error".into(),
+ message: format!("failed to encode subscription event: {error}"),
+ }
}
impl ActiveEventSubscription {
@@ -341,10 +408,55 @@ impl ActiveAgentStatusChangedSubscription {
api_tx: &ApiRequestSender,
event_hub: &EventHub,
) -> Result<Option<SubscriptionEventEnvelope>, ErrorResponse> {
- let mut saw_status_event = false;
for (sequence, event) in event_hub.events_after(self.last_sequence) {
self.last_sequence = sequence;
- let crate::api::schema::EventData::PaneAgentStatusChanged {
+ if let Some(event) = self.event_from_history(event) {
+ return Ok(Some(event));
+ }
+ }
+
+ self.poll_snapshot(api_tx, event_hub)
+ }
+
+ fn event_from_history(
+ &mut self,
+ event: crate::api::schema::EventEnvelope,
+ ) -> Option<SubscriptionEventEnvelope> {
+ if event.event != EventKind::PaneAgentStatusChanged {
+ return None;
+ }
+ let crate::api::schema::EventData::PaneAgentStatusChanged {
+ pane_id,
+ workspace_id,
+ agent_status,
+ agent,
+ title,
+ display_agent,
+ state_labels,
+ } = event.data
+ else {
+ return None;
+ };
+ if pane_id != self.pane_id {
+ return None;
+ }
+ self.last_status = Some(agent_status);
+ self.last_presentation = Some(PanePresentationSnapshot::from_event(
+ &title,
+ &display_agent,
+ &state_labels,
+ ));
+ self.initial_event = None;
+ if self
+ .status_filter
+ .is_some_and(|wanted| wanted != agent_status)
+ {
+ return None;
+ }
+
+ Some(SubscriptionEventEnvelope {
+ event: SubscriptionEventKind::PaneAgentStatusChanged,
+ data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent {
pane_id,
workspace_id,
agent_status,
@@ -352,47 +464,16 @@ impl ActiveAgentStatusChangedSubscription {
title,
display_agent,
state_labels,
- } = event.data
- else {
- continue;
- };
- if event.event != crate::api::schema::EventKind::PaneAgentStatusChanged {
- continue;
- }
- if pane_id != self.pane_id {
- continue;
- }
- saw_status_event = true;
-
- let current_presentation =
- PanePresentationSnapshot::from_event(&title, &display_agent, &state_labels);
- self.last_status = Some(agent_status);
- self.last_presentation = Some(current_presentation);
- if self
- .status_filter
- .is_some_and(|wanted| wanted != agent_status)
- {
- continue;
- }
-
- self.initial_event = None;
- return Ok(Some(SubscriptionEventEnvelope {
- event: SubscriptionEventKind::PaneAgentStatusChanged,
- data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent {
- pane_id,
- workspace_id,
- agent_status,
- agent,
- title,
- display_agent,
- state_labels,
- }),
- }));
- }
+ }),
+ })
+ }
- if saw_status_event {
- self.initial_event = None;
- } else if event_hub.current_sequence() != self.last_sequence {
+ fn poll_snapshot(
+ &mut self,
+ api_tx: &ApiRequestSender,
+ event_hub: &EventHub,
+ ) -> Result<Option<SubscriptionEventEnvelope>, ErrorResponse> {
+ if event_hub.current_sequence() != self.last_sequence {
return Ok(None);
} else if let Some(event) = self.initial_event.take() {
return Ok(Some(SubscriptionEventEnvelope {
@@ -688,6 +769,96 @@ mod tests {
}
#[test]
+ fn lifecycle_batch_drains_in_order_and_advances_past_unmatched_events() {
+ let event_hub = EventHub::default();
+ event_hub.push(workspace_focused_event("old"));
+ let start = event_hub.current_sequence();
+ event_hub.push(workspace_focused_event("setup"));
+ let (api_tx, _api_rx) = tokio::sync::mpsc::unbounded_channel();
+ let mut subscription = ActiveSubscription::new(
+ Subscription::WorkspaceFocused {},
+ "batch",
+ 0,
+ &api_tx,
+ &event_hub,
+ start,
+ )
+ .unwrap();
+ event_hub.push(presentation_event(None));
+ event_hub.push(workspace_focused_event("live"));
+ let events = subscription.poll_batch(&api_tx, &event_hub).unwrap();
+ assert_eq!(events.len(), 2);
+ assert_eq!(events[0]["data"]["workspace_id"], "setup");
+ assert_eq!(events[1]["data"]["workspace_id"], "live");
+ assert!(subscription
+ .poll_batch(&api_tx, &event_hub)
+ .unwrap()
+ .is_empty());
+ let ActiveSubscription::Event(subscription) = subscription else {
+ panic!("expected lifecycle subscription");
+ };
+ assert_eq!(subscription.last_sequence, event_hub.current_sequence());
+ }
+
+ #[test]
+ fn agent_status_batch_preserves_transitions_filters_and_initial_state_ordering() {
+ for filtered in [false, true] {
+ let event_hub = EventHub::default();
+ let mut subscription = ActiveSubscription::AgentStatusChanged(Box::new(
+ ActiveAgentStatusChangedSubscription {
+ pane_id: "pane_1".into(),
+ status_filter: filtered.then_some(AgentStatus::Working),
+ last_status: Some(AgentStatus::Working),
+ last_presentation: None,
+ last_sequence: event_hub.current_sequence(),
+ initial_event: Some(PaneAgentStatusChangedEvent {
+ pane_id: "pane_1".into(),
+ workspace_id: "workspace_1".into(),
+ agent_status: AgentStatus::Working,
+ agent: Some("pi".into()),
+ title: Some("stale initial snapshot".into()),
+ display_agent: None,
+ state_labels: HashMap::new(),
+ }),
+ request_prefix: "batch".into(),
+ },
+ ));
+ for (status, title) in [
+ (AgentStatus::Working, "started"),
+ (AgentStatus::Blocked, "approval"),
+ (AgentStatus::Idle, "finished"),
+ (AgentStatus::Working, "restarted"),
+ ] {
+ let mut event = presentation_event(Some(title));
+ let EventData::PaneAgentStatusChanged { agent_status, .. } = &mut event.data else {
+ panic!("expected status data");
+ };
+ *agent_status = status;
+ event_hub.push(event);
+ }
+ let (api_tx, _api_rx) = tokio::sync::mpsc::unbounded_channel();
+ let events = subscription.poll_batch(&api_tx, &event_hub).unwrap();
+ let titles = events
+ .iter()
+ .map(|event| event["data"]["title"].as_str().unwrap())
+ .collect::<Vec<_>>();
+ assert_eq!(
+ titles,
+ if filtered {
+ vec!["started", "restarted"]
+ } else {
+ vec!["started", "approval", "finished", "restarted"]
+ }
+ );
+ let ActiveSubscription::AgentStatusChanged(subscription) = subscription else {
+ panic!("expected agent subscription");
+ };
+ assert_eq!(subscription.last_sequence, event_hub.current_sequence());
+ assert!(subscription.initial_event.is_none());
+ }
+ }
+
+ #[test]
fn scroll_subscription_emits_when_scroll_snapshot_changes() {
let at_bottom = PaneScrollInfo {
offset_from_bottom: 0,
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-11 14:41:09 +0000