@@ -0,0 +1,401 @@
1+// Package herdr provides native integration with the herdr terminal
2+// multiplexer. When Crush runs inside a herdr-managed pane it reports
3+// agent state (idle, working, blocked) and session identity over
4+// herdr's Unix socket API so herdr can display accurate status without
5+// screen scraping.
6+//
7+// The client consumes a small, herdr-specific event vocabulary rather
8+// than accepting raw proto or domain types. Callers translate their
9+// events into herdr.Event before forwarding. This keeps the client
10+// decoupled from both the proto and internal domain layers.
11+package herdr
12+
13+import (
14+ "context"
15+ "encoding/json"
16+ "flag"
17+ "fmt"
18+ "io"
19+ "log/slog"
20+ "net"
21+ "os"
22+ "sync"
23+ "time"
24+)
25+
26+// State values matching herdr's PaneAgentState enum.
27+const (
28+ stateIdle = "idle"
29+ stateWorking = "working"
30+ stateBlocked = "blocked"
31+)
32+
33+// Event is the herdr-specific event vocabulary. Each type maps to a
34+// distinct state transition in the agent lifecycle. Callers translate
35+// from proto or domain types into these before calling HandleEvent.
36+type Event interface {
37+ herdrEvent()
38+}
39+
40+// AssistantMessage indicates the agent produced output. Transitions
41+// to working if not already active.
42+type AssistantMessage struct {
43+ SessionID string
44+}
45+
46+func (AssistantMessage) herdrEvent() {}
47+
48+// RunComplete indicates the agent finished a turn. Transitions to idle.
49+type RunComplete struct {
50+ SessionID string
51+}
52+
53+func (RunComplete) herdrEvent() {}
54+
55+// PermissionRequested indicates the agent is waiting for user approval.
56+// Transitions to blocked.
57+type PermissionRequested struct{}
58+
59+func (PermissionRequested) herdrEvent() {}
60+
61+// PermissionResolved indicates a permission decision was made.
62+// Transitions back to working if a run is active, idle otherwise.
63+type PermissionResolved struct{}
64+
65+func (PermissionResolved) herdrEvent() {}
66+
67+// Summarizing indicates the agent is compacting context. Transitions
68+// to working if not already active.
69+type Summarizing struct{}
70+
71+func (Summarizing) herdrEvent() {}
72+
73+// sender abstracts the transport layer for reporting state to herdr.
74+// Production uses a Unix socket; tests use a recorder.
75+type sender interface {
76+ send(req reportRequest) error
77+ close()
78+}
79+
80+// Client reports Crush agent state to a running herdr instance.
81+type Client struct {
82+ socketPath string
83+ paneID string
84+
85+ mu sync.Mutex
86+ sessionID string
87+ state string
88+ runActive bool
89+ seq uint64
90+
91+ snd sender
92+}
93+
94+// defaultClient is the process-wide herdr client. Initialized once
95+// via Init(). All integration sites share this single instance so
96+// only one Unix socket connection exists per process.
97+var (
98+ defaultClient *Client
99+ initOnce sync.Once
100+)
101+
102+// Init returns the process-wide herdr Client, creating it on first
103+// call from environment variables. Returns nil when Crush is not
104+// running inside a herdr pane. Safe to call from any goroutine.
105+func Init() *Client {
106+ initOnce.Do(func() {
107+ defaultClient = newFromEnv()
108+ })
109+ return defaultClient
110+}
111+
112+func newFromEnv() *Client {
113+ if os.Getenv("HERDR_ENV") != "1" {
114+ return nil
115+ }
116+ // A test binary inherits the launching shell's HERDR_* env, so
117+ // without this it would attach to the developer's live pane and
118+ // release its agent on teardown. Skip herdr entirely under test.
119+ if flag.Lookup("test.v") != nil {
120+ slog.Debug("Herdr integration disabled: running under go test")
121+ return nil
122+ }
123+ socketPath := os.Getenv("HERDR_SOCKET_PATH")
124+ paneID := os.Getenv("HERDR_PANE_ID")
125+ if socketPath == "" || paneID == "" {
126+ slog.Debug("Herdr integration disabled: incomplete environment",
127+ "has_socket", socketPath != "",
128+ "has_pane_id", paneID != "",
129+ )
130+ return nil
131+ }
132+ c := &Client{
133+ socketPath: socketPath,
134+ paneID: paneID,
135+ state: stateIdle,
136+ seq: uint64(time.Now().UnixNano()),
137+ snd: newUnixSender(socketPath),
138+ }
139+ c.registerInitial()
140+ return c
141+}
142+
143+// registerInitial sends an initial idle-state report to herdr so the
144+// pane knows about the agent immediately, not just after the first
145+// event. Called once during client creation. Bypasses the dedup
146+// check since the initial state must always be reported regardless
147+// of redundancy.
148+//
149+// herdr remembers the highest seq it has seen per source for the
150+// lifetime of a pane and silently drops any report with a seq that
151+// is not strictly greater. Because crush seeds seq from the wall
152+// clock at startup (see newFromEnv), a restarted crush in the same
153+// pane always reports above the previous run's high-water mark, so
154+// the first report is accepted instead of being rejected as stale.
155+func (c *Client) registerInitial() {
156+ if c == nil {
157+ return
158+ }
159+ c.mu.Lock()
160+ defer c.mu.Unlock()
161+ c.snd.send(c.newRequestLocked("pane.report_agent", "init", stateIdle))
162+}
163+
164+// Close releases the agent's authority on the pane and shuts down
165+// the background writer. Safe to call on a nil client.
166+func (c *Client) Close() {
167+ if c == nil {
168+ return
169+ }
170+ c.releaseAgent()
171+ c.snd.close()
172+}
173+
174+// releaseAgent sends a pane.release_agent request to herdr so the
175+// pane is freed for a new agent to claim authority. This is the
176+// clean-shutdown protocol per herdr's socket API. Sends directly
177+// on the socket to ensure delivery even if the write loop is busy.
178+func (c *Client) releaseAgent() {
179+ c.mu.Lock()
180+ defer c.mu.Unlock()
181+ req := c.newRequestLocked("pane.release_agent", "release", "")
182+ if err := dialSend(c.socketPath, req); err != nil {
183+ slog.Debug("Herdr release_agent failed", "error", err)
184+ }
185+}
186+
187+// HandleEvent processes a single herdr event and reports state changes.
188+// Safe to call from any goroutine.
189+func (c *Client) HandleEvent(ev Event) {
190+ if c == nil {
191+ return
192+ }
193+ switch e := ev.(type) {
194+ case AssistantMessage:
195+ c.onAssistantMessage(e.SessionID)
196+ case RunComplete:
197+ c.onRunComplete(e.SessionID)
198+ case PermissionRequested:
199+ c.onPermissionRequest()
200+ case PermissionResolved:
201+ c.onPermissionResolved()
202+ case Summarizing:
203+ c.onSummarizing()
204+ }
205+}
206+
207+// SetSessionID sets the session ID for reporting. Call this when the
208+// session is created or resolved, before events start flowing.
209+func (c *Client) SetSessionID(id string) {
210+ if c == nil {
211+ return
212+ }
213+ c.mu.Lock()
214+ defer c.mu.Unlock()
215+ c.sessionID = id
216+}
217+
218+func (c *Client) onAssistantMessage(sessionID string) {
219+ c.mu.Lock()
220+ defer c.mu.Unlock()
221+ if sessionID != "" {
222+ c.sessionID = sessionID
223+ }
224+ if !c.runActive {
225+ c.runActive = true
226+ c.reportLocked(stateWorking)
227+ }
228+}
229+
230+func (c *Client) onRunComplete(sessionID string) {
231+ c.mu.Lock()
232+ defer c.mu.Unlock()
233+ c.runActive = false
234+ if sessionID != "" {
235+ c.sessionID = sessionID
236+ }
237+ c.reportLocked(stateIdle)
238+}
239+
240+func (c *Client) onPermissionRequest() {
241+ c.mu.Lock()
242+ defer c.mu.Unlock()
243+ // A permission request implies a run is active, even if no
244+ // assistant message has arrived yet (e.g. tool calls that fire
245+ // before any text output).
246+ if !c.runActive {
247+ c.runActive = true
248+ }
249+ c.reportLocked(stateBlocked)
250+}
251+
252+func (c *Client) onPermissionResolved() {
253+ c.mu.Lock()
254+ defer c.mu.Unlock()
255+ if c.runActive {
256+ c.reportLocked(stateWorking)
257+ } else {
258+ c.reportLocked(stateIdle)
259+ }
260+}
261+
262+func (c *Client) onSummarizing() {
263+ c.mu.Lock()
264+ defer c.mu.Unlock()
265+ if !c.runActive {
266+ c.runActive = true
267+ }
268+ c.reportLocked(stateWorking)
269+}
270+
271+// newRequestLocked builds a seq-stamped JSON-RPC request to herdr.
272+// Must be called with c.mu held. Every request increments c.seq so
273+// herdr accepts it as strictly newer than the last (see
274+// registerInitial for why monotonic seq matters). State is empty for
275+// requests that carry no agent state, such as pane.release_agent.
276+func (c *Client) newRequestLocked(method, idPrefix, state string) reportRequest {
277+ c.seq++
278+ return reportRequest{
279+ ID: fmt.Sprintf("crush:%s:%d", idPrefix, time.Now().UnixNano()),
280+ Method: method,
281+ Params: reportParams{
282+ PaneID: c.paneID,
283+ Source: "crush",
284+ Agent: "crush",
285+ State: state,
286+ Seq: c.seq,
287+ AgentSessionID: c.sessionID,
288+ },
289+ }
290+}
291+
292+// reportLocked sends a pane.report_agent request to herdr. Must be
293+// called with c.mu held. Skips redundant reports when the state has
294+// not changed.
295+func (c *Client) reportLocked(state string) {
296+ if state == c.state {
297+ return
298+ }
299+ c.state = state
300+ c.snd.send(c.newRequestLocked("pane.report_agent", "report", state))
301+}
302+
303+// reportRequest is the JSON-RPC envelope sent to herdr.
304+type reportRequest struct {
305+ ID string `json:"id"`
306+ Method string `json:"method"`
307+ Params reportParams `json:"params"`
308+}
309+
310+// reportParams carries the agent state payload.
311+type reportParams struct {
312+ PaneID string `json:"pane_id"`
313+ Source string `json:"source"`
314+ Agent string `json:"agent"`
315+ State string `json:"state"`
316+ Seq uint64 `json:"seq"`
317+ AgentSessionID string `json:"agent_session_id"`
318+}
319+
320+// unixSender sends JSON-RPC requests over a Unix domain socket using
321+// a single background writer goroutine and a buffered channel. This
322+// serializes writes and avoids spawning unbounded goroutines under
323+// high event throughput. Each report opens a short-lived connection.
324+type unixSender struct {
325+ socketPath string
326+ ch chan reportRequest
327+ cancel context.CancelFunc
328+}
329+
330+func newUnixSender(socketPath string) *unixSender {
331+ ctx, cancel := context.WithCancel(context.Background())
332+ s := &unixSender{
333+ socketPath: socketPath,
334+ ch: make(chan reportRequest, 16),
335+ cancel: cancel,
336+ }
337+ go s.writeLoop(ctx)
338+ return s
339+}
340+
341+func (s *unixSender) send(req reportRequest) error {
342+ select {
343+ case s.ch <- req:
344+ default:
345+ // Drop if the buffer is full. State reports are
346+ // best-effort; blocking the agent is worse than
347+ // missing a transition.
348+ }
349+ return nil
350+}
351+
352+func (s *unixSender) close() {
353+ s.cancel()
354+}
355+
356+func (s *unixSender) writeLoop(ctx context.Context) {
357+ for {
358+ select {
359+ case req, ok := <-s.ch:
360+ if !ok {
361+ return
362+ }
363+ if err := dialSend(s.socketPath, req); err != nil {
364+ slog.Debug("Herdr report failed", "error", err)
365+ }
366+ case <-ctx.Done():
367+ return
368+ }
369+ }
370+}
371+
372+// dialSend opens a short-lived Unix socket connection to herdr,
373+// sends a single JSON-RPC request, and drains the response.
374+func dialSend(socketPath string, req reportRequest) error {
375+ ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
376+ defer cancel()
377+
378+ dialer := net.Dialer{}
379+ conn, err := dialer.DialContext(ctx, "unix", socketPath)
380+ if err != nil {
381+ return err
382+ }
383+ defer conn.Close()
384+
385+ _ = conn.SetDeadline(time.Now().Add(500 * time.Millisecond))
386+
387+ data, err := json.Marshal(req)
388+ if err != nil {
389+ return err
390+ }
391+ data = append(data, '\n')
392+
393+ _, err = conn.Write(data)
394+ if err != nil {
395+ return err
396+ }
397+
398+ // Drain the response to complete the request cycle.
399+ _, _ = io.Copy(io.Discard, conn)
400+ return nil
401+}