// Package tasksws is a reconnecting WebSocket client for l-ai-tasks board events. package tasksws import ( "context" "encoding/json" "fmt" "log/slog" "math/rand" "net/http" "net/url" "strconv" "time" "github.com/gorilla/websocket" ) // Event mirrors l-ai-tasks realtime.Event JSON. type Event struct { Type string `json:"type"` Seq int64 `json:"seq"` At time.Time `json:"at"` Project *Ref `json:"project,omitempty"` Task *TaskRef `json:"task,omitempty"` Comment *Comment `json:"comment,omitempty"` Mentions []Mention `json:"mentions,omitempty"` } type Ref struct { ID string `json:"id"` Slug string `json:"slug"` Name string `json:"name"` } type TaskRef struct { ID string `json:"id"` Title string `json:"title"` Status string `json:"status"` AssigneeID string `json:"assignee_id,omitempty"` } type Comment struct { ID string `json:"id"` Body string `json:"body"` AuthorType string `json:"author_type"` AuthorID string `json:"author_id"` } type Mention struct { Type string `json:"type"` ID string `json:"id"` } // Handler is called for each board event (after ready). type Handler func(ctx context.Context, ev Event) error // Client reconnects forever until ctx cancel. type Client struct { URL string Token string Agent string // logical id for filtering (nora, coordinator, kaya) Log *slog.Logger OnEvent Handler } // Run blocks until ctx is done. func (c *Client) Run(ctx context.Context) { log := c.Log if log == nil { log = slog.Default() } var since int64 attempt := 0 for { if ctx.Err() != nil { return } err := c.session(ctx, &since) if ctx.Err() != nil { return } attempt++ delay := backoff(attempt) log.Warn("tasks ws disconnected; reconnecting", "error", err, "attempt", attempt, "delay", delay, "since", since) select { case <-ctx.Done(): return case <-time.After(delay): } } } func (c *Client) session(ctx context.Context, since *int64) error { u, err := url.Parse(c.URL) if err != nil { return err } q := u.Query() q.Set("since", strconv.FormatInt(*since, 10)) u.RawQuery = q.Encode() hdr := http.Header{} if c.Token != "" { hdr.Set("Authorization", "Bearer "+c.Token) } dialer := websocket.Dialer{HandshakeTimeout: 15 * time.Second} conn, _, err := dialer.DialContext(ctx, u.String(), hdr) if err != nil { return err } defer conn.Close() _ = conn.SetReadDeadline(time.Now().Add(90 * time.Second)) // Server sends Ping; gorilla delivers it as Ping (not Pong). Reset deadline or // idle boards die every ~90s with "i/o timeout" despite healthy pings. conn.SetPingHandler(func(appData string) error { _ = conn.SetReadDeadline(time.Now().Add(90 * time.Second)) return conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(10*time.Second)) }) conn.SetPongHandler(func(string) error { _ = conn.SetReadDeadline(time.Now().Add(90 * time.Second)) return nil }) for { select { case <-ctx.Done(): return ctx.Err() default: } _, data, err := conn.ReadMessage() if err != nil { return err } _ = conn.SetReadDeadline(time.Now().Add(90 * time.Second)) var probe struct { Type string `json:"type"` Seq int64 `json:"seq"` } if err := json.Unmarshal(data, &probe); err != nil { continue } if probe.Type == "ready" { continue } var ev Event if err := json.Unmarshal(data, &ev); err != nil { continue } if ev.Seq > *since { *since = ev.Seq } if c.OnEvent != nil { ev := ev hctx := context.WithoutCancel(ctx) go func() { if err := c.OnEvent(hctx, ev); err != nil && c.Log != nil { c.Log.Warn("tasks ws handler", "error", err, "type", ev.Type) } }() } } } func backoff(attempt int) time.Duration { if attempt < 1 { attempt = 1 } if attempt > 6 { attempt = 6 } base := 200 * time.Millisecond d := base * time.Duration(1< 5*time.Second { d = 5 * time.Second } jitter := time.Duration(rand.Int63n(int64(d/2) + 1)) return d/2 + jitter } // MentionsSelf reports whether ev.mentions includes agentID. func MentionsSelf(ev Event, agentID string) bool { for _, m := range ev.Mentions { if m.Type == "agent" && m.ID == agentID { return true } } return false } // ErrAuth is returned conceptually for logging; dial errors wrap status. func FormatDial(url, agent string) string { return fmt.Sprintf("tasksws agent=%s url=%s", agent, url) }