feat(lip): push-driven health via long-lived subscribe

CKB used to re-probe IndexStatus on a 60 s TTL every time the rerank
path checked whether the LIP index was mixed-models. The LIP daemon has
pushed IndexChanged frames to all active sessions since v1.5.0, so the
polling was pure debt.

New internal/lip/subscribe.go opens a long-lived connection, pings
IndexStatus every 3 s to flush the daemon's broadcast channel (the
session loop drains queued pushes only after writing a response), reads
every frame in a loop, and routes index_changed and index_status by
type tag. Reconnects with exponential backoff to 30 s on daemon drop.

Engine owns one subscriber, started in NewEngine and cancelled in
Close. The cached availability/mixed flags are now written by the
subscriber, not the query path — lipSemanticAvailable is lock-free
RLock+read, no RPC. Worst-case staleness for the rerank gate drops
from 60 s to ~3 s.

Tests rewritten: the fake daemon now serves multiple requests per
connection and tests wait for the first health frame before asserting,
plus a new TestLipSubscriber_ReusesSingleConnection verifies the hot
path issues zero requests.

Also: CHANGELOG entries for #208 and #209 which landed on develop
without one.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: SimplyLiz

CHANGELOG.md

@@ -4,6 +4,17 @@ All notable changes to CKB will be documented in this file.
 
 ## [Unreleased]
 
+### Changed
+
+- **LIP health: push-driven, not polled** — the Engine now opens a long-lived
+  connection to the LIP daemon at startup (`internal/lip/subscribe.go`) and
+  receives `index_changed` frames plus per-ping `index_status` snapshots
+  instead of issuing a fresh `IndexStatus` RPC on a 60 s TTL. Worst-case
+  staleness for the mixed-models gate drops from 60 s to ~3 s, the hot query
+  path is lock-free (no RPC, no dial, no TTL check), and the subscriber
+  reconnects with exponential backoff when the daemon restarts. The old
+  `lipHealthTTL` constant is gone; callers read the cached flag directly.
+
 ### Added
 
 - Troubleshooting section in `docs/plans/review-cicd.md` covering shallow

internal/lip/subscribe.go

@@ -0,0 +1,237 @@
+package lip
+
+import (
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"io"
+	"net"
+	"sync"
+	"time"
+)
+
+// IndexChangedEvent carries the payload of a LIP `index_changed` push frame.
+// The daemon emits one per upsert with the total indexed-file count and the
+// URIs that were touched. `Pending` is populated opportunistically from the
+// piggybacked IndexStatus probe that keeps the push channel flushed — it may
+// be stale by a ping interval but is good enough for cache invalidation.
+type IndexChangedEvent struct {
+	IndexedFiles int
+	AffectedURIs []string
+}
+
+// HealthEvent carries the index-health view fetched on each keepalive ping.
+// It is the push-equivalent of polling `IndexStatus`: the subscriber pulls
+// it once per ping interval instead of every query, and callers key off
+// `MixedModels` for the rerank gate.
+type HealthEvent struct {
+	Available     bool
+	MixedModels   bool
+	IndexedFiles  int
+	ModelsInIndex []string
+}
+
+// SubscribeHandler receives push events.
+// Implementations must be non-blocking — work should be dispatched to a
+// separate goroutine or written to a buffered channel.
+type SubscribeHandler interface {
+	OnIndexChanged(IndexChangedEvent)
+	OnHealth(HealthEvent)
+}
+
+// pingInterval is how often the subscriber issues a cheap request to flush
+// the daemon's broadcast channel. The LIP daemon drains pending push frames
+// only after writing a response, so this caps the worst-case push latency.
+// Picked to give ~3 s freshness without generating meaningful daemon load.
+const pingInterval = 3 * time.Second
+
+// reconnectMaxBackoff bounds the reconnect backoff so a long outage still
+// recovers within ~minute once the daemon returns.
+const reconnectMaxBackoff = 30 * time.Second
+
+// Subscribe opens a long-lived connection to the LIP daemon, polls
+// `IndexStatus` on a short ticker as a keepalive, and dispatches every
+// `index_changed` frame the daemon pushes in between. It reconnects with
+// exponential backoff when the daemon is unavailable and returns when ctx
+// is cancelled.
+//
+// The daemon drains queued push frames only after writing a reply to a
+// client request (see `session.rs` run loop), so a purely passive reader
+// would never observe pushes. The ticker doubles as a health probe: every
+// response carries the current `IndexStatusInfo`, so callers get a
+// `HealthEvent` per interval for free — replacing the per-query
+// `IndexStatus` RPC the engine used to issue.
+func Subscribe(ctx context.Context, h SubscribeHandler) {
+	backoff := 500 * time.Millisecond
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		err := runSession(ctx, h)
+		if ctx.Err() != nil {
+			return
+		}
+		// Emit a final "unavailable" health event so consumers fail closed
+		// quickly when the daemon drops.
+		h.OnHealth(HealthEvent{Available: false})
+		if err == nil {
+			backoff = 500 * time.Millisecond
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(backoff):
+		}
+		backoff *= 2
+		if backoff > reconnectMaxBackoff {
+			backoff = reconnectMaxBackoff
+		}
+	}
+}
+
+// runSession holds a single connection open: it writes a keepalive on
+// `pingInterval`, reads every incoming frame, and routes them by `type`.
+// Returns on any I/O error so the caller can reconnect.
+func runSession(ctx context.Context, h SubscribeHandler) error {
+	conn, err := net.DialTimeout("unix", SocketPath(), 500*time.Millisecond)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	sessCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// Writer goroutine: pings until the session is torn down.
+	var writeMu sync.Mutex
+	ping := func() error {
+		writeMu.Lock()
+		defer writeMu.Unlock()
+		return writeFrame(conn, map[string]any{"type": "index_status"})
+	}
+	if err := ping(); err != nil {
+		return err
+	}
+
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		t := time.NewTicker(pingInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-sessCtx.Done():
+				return
+			case <-t.C:
+				if err := ping(); err != nil {
+					cancel()
+					return
+				}
+			}
+		}
+	})
+
+	// Reader loop: any I/O error tears down the session.
+	for {
+		if sessCtx.Err() != nil {
+			break
+		}
+		frame, err := readFrame(conn)
+		if err != nil {
+			cancel()
+			wg.Wait()
+			return err
+		}
+		dispatchFrame(frame, h)
+	}
+	wg.Wait()
+	return nil
+}
+
+// dispatchFrame routes a decoded server frame to the handler by `type` tag.
+// The LIP wire format nests ServerMessage inside a `ServerResponse` envelope
+// with `ok`/`error` fields.
+func dispatchFrame(frame map[string]json.RawMessage, h SubscribeHandler) {
+	// Drop envelope: ServerResponse { ok: ServerMessage, error: Option<String> }
+	var inner map[string]json.RawMessage
+	if raw, ok := frame["ok"]; ok && len(raw) > 0 && string(raw) != "null" {
+		_ = json.Unmarshal(raw, &inner)
+	} else {
+		inner = frame
+	}
+	var kind string
+	_ = json.Unmarshal(inner["type"], &kind)
+
+	switch kind {
+	case "index_changed":
+		var ev struct {
+			IndexedFiles int      `json:"indexed_files"`
+			AffectedURIs []string `json:"affected_uris"`
+		}
+		if b, ok := marshalInner(inner); ok {
+			_ = json.Unmarshal(b, &ev)
+		}
+		h.OnIndexChanged(IndexChangedEvent{
+			IndexedFiles: ev.IndexedFiles,
+			AffectedURIs: ev.AffectedURIs,
+		})
+	case "index_status":
+		var s indexStatusResp
+		if b, ok := marshalInner(inner); ok {
+			_ = json.Unmarshal(b, &s)
+		}
+		h.OnHealth(HealthEvent{
+			Available:     true,
+			MixedModels:   s.MixedModels,
+			IndexedFiles:  s.IndexedFiles,
+			ModelsInIndex: s.ModelsInIndex,
+		})
+	}
+}
+
+func marshalInner(m map[string]json.RawMessage) ([]byte, bool) {
+	b, err := json.Marshal(m)
+	if err != nil {
+		return nil, false
+	}
+	return b, true
+}
+
+// writeFrame encodes a single length-prefixed JSON frame.
+func writeFrame(conn net.Conn, payload any) error {
+	b, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	lenBuf := make([]byte, 4)
+	binary.BigEndian.PutUint32(lenBuf, uint32(len(b)))
+	_ = conn.SetWriteDeadline(time.Now().Add(2 * time.Second))
+	if _, err := conn.Write(append(lenBuf, b...)); err != nil {
+		return err
+	}
+	return nil
+}
+
+// readFrame reads one length-prefixed JSON frame with no overall deadline —
+// the ping loop detects dead connections via write failures instead, so the
+// reader can block indefinitely between pushes.
+func readFrame(conn net.Conn) (map[string]json.RawMessage, error) {
+	_ = conn.SetReadDeadline(time.Time{})
+	lenBuf := make([]byte, 4)
+	if _, err := io.ReadFull(conn, lenBuf); err != nil {
+		return nil, err
+	}
+	respLen := binary.BigEndian.Uint32(lenBuf)
+	if respLen == 0 || respLen > 64<<20 {
+		return nil, errors.New("lip: frame length out of range")
+	}
+	buf := make([]byte, respLen)
+	if _, err := io.ReadFull(conn, buf); err != nil {
+		return nil, err
+	}
+	var out map[string]json.RawMessage
+	if err := json.Unmarshal(buf, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}

internal/query/engine.go

@@ -64,11 +64,15 @@ type Engine struct {
 	cachedState     *RepoState
 	stateComputedAt time.Time
 
-	// LIP health (cached; refreshed on a short TTL to avoid per-query RPCs).
+	// LIP health, maintained by a background subscriber that keeps a long-lived
+	// connection open and receives `index_changed` pushes plus per-ping health
+	// snapshots. `lipHealthCheckedAt` is zero until the first frame arrives —
+	// callers check it before trusting the flags.
 	lipHealthMu        sync.RWMutex
 	cachedLipMixed     bool
 	cachedLipAvailable bool
 	lipHealthCheckedAt time.Time
+	lipSubCancel       context.CancelFunc
 
 	// Cache stats
 	cacheStatsMu sync.RWMutex
@@ -136,6 +140,8 @@ func NewEngine(repoRoot string, db *storage.DB, logger *slog.Logger, cfg *config
 		// Don't fail - async operations will be unavailable
 	}
 
+	engine.startLipSubscriber()
+
 	return engine, nil
 }
 
@@ -440,6 +446,10 @@ func (e *Engine) DB() *storage.DB {
 func (e *Engine) Close() error {
 	var lastErr error
 
+	if e.lipSubCancel != nil {
+		e.lipSubCancel()
+	}
+
 	// Stop job runner first
 	if e.jobRunner != nil {
 		if err := e.jobRunner.Stop(10 * time.Second); err != nil {

internal/query/lip_health.go

@@ -1,38 +1,58 @@
 package query
 
 import (
+	"context"
 	"time"
 
 	"github.com/SimplyLiz/CodeMCP/internal/lip"
 )
 
-// lipHealthTTL caps how often we re-probe the LIP daemon for index status.
-// IndexStatus is a 200 ms RPC, so we do not want this per-query.
-const lipHealthTTL = 60 * time.Second
-
-// lipSemanticAvailable reports whether LIP semantic operations (rerank, semantic
-// search) can be trusted. Returns false when the daemon is unavailable OR when
-// the index contains vectors from more than one embedding model — cosine
-// similarity across different vector spaces is mathematically meaningless, so a
-// mixed-model index silently produces garbage rankings.
+// lipSemanticAvailable reports whether LIP semantic operations (rerank,
+// semantic search) can be trusted. The flag is maintained by the background
+// subscriber started in `startLipSubscriber` — no RPC happens here, so this
+// is safe to call on the hot query path. Returns false when:
+//
+//   - the subscriber has not yet observed its first frame (daemon down at
+//     startup, or engine created less than a ping ago),
+//   - the most recent health frame reported the daemon unavailable,
+//   - the index contains vectors from more than one embedding model, which
+//     makes cross-vector cosine similarity mathematically meaningless.
 func (e *Engine) lipSemanticAvailable() bool {
 	e.lipHealthMu.RLock()
-	fresh := !e.lipHealthCheckedAt.IsZero() && time.Since(e.lipHealthCheckedAt) < lipHealthTTL
-	avail, mixed := e.cachedLipAvailable, e.cachedLipMixed
-	e.lipHealthMu.RUnlock()
-	if fresh {
-		return avail && !mixed
+	defer e.lipHealthMu.RUnlock()
+	if e.lipHealthCheckedAt.IsZero() {
+		return false
 	}
+	return e.cachedLipAvailable && !e.cachedLipMixed
+}
 
-	status, _ := lip.IndexStatus()
-	e.lipHealthMu.Lock()
-	e.lipHealthCheckedAt = time.Now()
-	if status == nil {
-		e.cachedLipAvailable, e.cachedLipMixed = false, false
-	} else {
-		e.cachedLipAvailable, e.cachedLipMixed = true, status.MixedModels
-	}
-	avail, mixed = e.cachedLipAvailable, e.cachedLipMixed
-	e.lipHealthMu.Unlock()
-	return avail && !mixed
+// engineLipSubscriber is the adapter between `lip.Subscribe` and the
+// Engine's cached health flags. A dedicated type (instead of binding methods
+// to Engine) keeps the handler interface invisible to the rest of the
+// package.
+type engineLipSubscriber struct{ e *Engine }
+
+func (s engineLipSubscriber) OnHealth(ev lip.HealthEvent) {
+	s.e.lipHealthMu.Lock()
+	defer s.e.lipHealthMu.Unlock()
+	s.e.lipHealthCheckedAt = time.Now()
+	s.e.cachedLipAvailable = ev.Available
+	s.e.cachedLipMixed = ev.MixedModels
+}
+
+func (s engineLipSubscriber) OnIndexChanged(_ lip.IndexChangedEvent) {
+	// Pushes are handled opportunistically: the ping that follows carries the
+	// refreshed IndexStatus, so we don't need to re-probe here. This hook
+	// exists so future consumers (e.g. cache invalidation keyed on
+	// AffectedURIs) can extend it without changing the transport.
+}
+
+// startLipSubscriber launches the background subscriber goroutine. It is a
+// no-op-safe — if the daemon is absent, Subscribe backs off and retries
+// until Close is called. The first health frame lands within `pingInterval`
+// of daemon availability.
+func (e *Engine) startLipSubscriber() {
+	ctx, cancel := context.WithCancel(context.Background())
+	e.lipSubCancel = cancel
+	go lip.Subscribe(ctx, engineLipSubscriber{e: e})
 }