feat(web-search): worker-side search — capability, Tavily, streaming, v2 capture, greedy inference

internal/config/config.go

@@ -48,6 +48,15 @@ type Config struct {
 	OllamaURL     string
 	OllamaTimeout time.Duration
 
+	// Web search (Tavily). SearchEnabled gates advertisement of the "search"
+	// capability and the per-job search stage. When false the worker behaves
+	// exactly as before.
+	SearchEnabled    bool
+	TavilyAPIKey     string
+	TavilyURL        string
+	SearchTimeout    time.Duration
+	SearchMaxResults int
+
 	// Beacon API (CL node)
 	BeaconAPIURL string
 
@@ -166,6 +175,8 @@ func Load() (*Config, error) {
 		RedisURL:               envOrDefault("REDIS_URL", "redis://localhost:6379"),
 		RedisPassword:          os.Getenv("REDIS_PASSWORD"),
 		OllamaURL:              envOrDefault("OLLAMA_URL", "http://localhost:11434"),
+		TavilyAPIKey:           os.Getenv("TAVILY_API_KEY"),
+		TavilyURL:              envOrDefault("TAVILY_URL", "https://api.tavily.com"),
 		BeaconAPIURL:           envOrDefault("BEACON_API_URL", "http://localhost:3500"),
 		SessionKeyFile:         envOrDefault("SESSION_KEY_FILE", "data/session-keys.enc"),
 		WorkerGatewayURL:       os.Getenv("WORKER_GATEWAY_URL"),
@@ -275,6 +286,9 @@ func Load() (*Config, error) {
 		errs = append(errs, fmt.Sprintf("LIGHTCHAIN_DRAIN_SLACK: must be >= 0, got %s", cfg.DrainSlack))
 	}
 	cfg.OllamaTimeout = parseDuration("OLLAMA_TIMEOUT", "120s", &errs)
+	cfg.SearchEnabled = parseBool("SEARCH_ENABLED", false, &errs)
+	cfg.SearchTimeout = parseDuration("SEARCH_TIMEOUT", "10s", &errs)
+	cfg.SearchMaxResults = parseInt("SEARCH_MAX_RESULTS", 5, &errs)
 	cfg.AckTxTimeout = parseDuration("ACK_TX_TIMEOUT", "15s", &errs)
 	// BlobTxTimeout bounds stage 8 (submit_blob): slot wait + SendTransaction
 	// + WaitMined. Default 90s = ~15 blocks at 6s block time, generous buffer
@@ -367,6 +381,17 @@ func (c *Config) Validate() []string {
 	if c.OllamaTimeout <= 0 {
 		errs = append(errs, "OLLAMA_TIMEOUT must be positive")
 	}
+	if c.SearchEnabled {
+		if c.TavilyAPIKey == "" {
+			errs = append(errs, "TAVILY_API_KEY is required when SEARCH_ENABLED=true")
+		}
+		if c.SearchTimeout <= 0 {
+			errs = append(errs, "SEARCH_TIMEOUT must be positive")
+		}
+		if c.SearchMaxResults <= 0 {
+			errs = append(errs, "SEARCH_MAX_RESULTS must be positive")
+		}
+	}
 	if c.AckTxTimeout <= 0 {
 		errs = append(errs, "ACK_TX_TIMEOUT must be positive")
 	}

internal/config/config_test.go

@@ -490,6 +490,31 @@ func TestLoad_JobExecutionOverrides(t *testing.T) {
 	assert.Equal(t, 5*time.Second, cfg.ReceiptPollInterval)
 }
 
+func TestLoad_SearchDefaults(t *testing.T) {
+	validEnv(t)
+	t.Setenv("WORKER_KEYSTORE_PATH", "/tmp/k")
+	t.Setenv("WORKER_KEYSTORE_PASSWORD", "p")
+	// SEARCH_* unset → search disabled, sane defaults.
+	cfg, err := Load()
+	require.NoError(t, err)
+	assert.False(t, cfg.SearchEnabled)
+	assert.Equal(t, "https://api.tavily.com", cfg.TavilyURL)
+	assert.Equal(t, 10*time.Second, cfg.SearchTimeout)
+	assert.Equal(t, 5, cfg.SearchMaxResults)
+}
+
+func TestLoad_SearchEnabled(t *testing.T) {
+	validEnv(t)
+	t.Setenv("WORKER_KEYSTORE_PATH", "/tmp/k")
+	t.Setenv("WORKER_KEYSTORE_PASSWORD", "p")
+	t.Setenv("SEARCH_ENABLED", "true")
+	t.Setenv("TAVILY_API_KEY", "tvly-xxx")
+	cfg, err := Load()
+	require.NoError(t, err)
+	assert.True(t, cfg.SearchEnabled)
+	assert.Equal(t, "tvly-xxx", cfg.TavilyAPIKey)
+}
+
 func TestValidate_JobRegistryZeroAddress(t *testing.T) {
 	validEnv(t)
 	t.Setenv("WORKER_KEYSTORE_PATH", "/tmp/keystore.json")

internal/heartbeat/monitor.go

@@ -48,15 +48,16 @@ type MonitorConfig struct {
 
 // Monitor publishes periodic heartbeat payloads to Redis.
 type Monitor struct {
-	redisClient *redis.Client
-	cfg         MonitorConfig
-	workerAddr  string   // EIP-55 checksummed hex, no 0x prefix
-	modelIDs    []string // 0x-prefixed lowercase hex bytes32
-	startedAt   time.Time
-	httpClient  *http.Client
-	logger      *slog.Logger
-	jobCounter  *atomic.Int32
-	maxJobs     int
+	redisClient  *redis.Client
+	cfg          MonitorConfig
+	workerAddr   string   // EIP-55 checksummed hex, no 0x prefix
+	modelIDs     []string // 0x-prefixed lowercase hex bytes32
+	capabilities []string // advertised capability tokens, e.g. ["search"]
+	startedAt    time.Time
+	httpClient   *http.Client
+	logger       *slog.Logger
+	jobCounter   *atomic.Int32
+	maxJobs      int
 	// metrics is optional — when nil the monitor still writes to Redis but
 	// does not update OllamaUp / HeartbeatLastEmit. Production constructs
 	// always pass non-nil; some tests pass nil to keep them focused.
@@ -78,23 +79,25 @@ func NewMonitor(
 	cfg MonitorConfig,
 	workerAddr string,
 	modelIDs []string,
+	capabilities []string,
 	jobCounter *atomic.Int32,
 	maxJobs int,
 	logger *slog.Logger,
 	metricsCollector *metrics.Metrics,
 ) *Monitor {
 	return &Monitor{
-		redisClient: redisClient,
-		cfg:         cfg,
-		workerAddr:  workerAddr,
-		modelIDs:    modelIDs,
-		startedAt:   time.Now(),
-		httpClient:  &http.Client{Timeout: 2 * time.Second},
-		logger:      logger,
-		jobCounter:  jobCounter,
-		maxJobs:     maxJobs,
-		metrics:     metricsCollector,
-		done:        make(chan struct{}),
+		redisClient:  redisClient,
+		cfg:          cfg,
+		workerAddr:   workerAddr,
+		modelIDs:     modelIDs,
+		capabilities: capabilities,
+		startedAt:    time.Now(),
+		httpClient:   &http.Client{Timeout: 2 * time.Second},
+		logger:       logger,
+		jobCounter:   jobCounter,
+		maxJobs:      maxJobs,
+		metrics:      metricsCollector,
+		done:         make(chan struct{}),
 	}
 }
 
@@ -162,20 +165,31 @@ func (m *Monitor) emit(ctx context.Context) error {
 		return fmt.Errorf("marshal model IDs: %w", err)
 	}
 
+	caps := m.capabilities
+	if caps == nil {
+		caps = []string{}
+	}
+	capsJSON, err := json.Marshal(caps)
+	if err != nil {
+		return fmt.Errorf("marshal capabilities: %w", err)
+	}
+
 	ttl := 3 * m.cfg.Interval
 	key := pkgtypes.HeartbeatRedisKey(m.workerAddr)
 
 	pipe := m.redisClient.TxPipeline()
 	pipe.HSet(ctx, key, map[string]interface{}{
-		pkgtypes.HBFieldLastHeartbeat: time.Now().Unix(),
-		pkgtypes.HBFieldActiveJobs:    activeJobs,
-		pkgtypes.HBFieldMaxJobs:       m.maxJobs,
-		pkgtypes.HBFieldLatencyMs:     0,
-		pkgtypes.HBFieldGPUUtil:       strconv.FormatFloat(0, 'f', -1, 64),
-		pkgtypes.HBFieldStatus:        pkgtypes.HeartbeatStatusActive,
-		pkgtypes.HBFieldModels:        string(modelsJSON),
-		pkgtypes.HBFieldOllamaStatus:  ollamaStatus,
-		pkgtypes.HBFieldUptime:        int64(time.Since(m.startedAt).Seconds()),
+		pkgtypes.HBFieldLastHeartbeat:   time.Now().Unix(),
+		pkgtypes.HBFieldActiveJobs:      activeJobs,
+		pkgtypes.HBFieldMaxJobs:         m.maxJobs,
+		pkgtypes.HBFieldLatencyMs:       0,
+		pkgtypes.HBFieldGPUUtil:         strconv.FormatFloat(0, 'f', -1, 64),
+		pkgtypes.HBFieldStatus:          pkgtypes.HeartbeatStatusActive,
+		pkgtypes.HBFieldModels:          string(modelsJSON),
+		pkgtypes.HBFieldOllamaStatus:    ollamaStatus,
+		pkgtypes.HBFieldUptime:          int64(time.Since(m.startedAt).Seconds()),
+		pkgtypes.HBFieldCapabilities:    string(capsJSON),
+		pkgtypes.HBFieldProtocolVersion: pkgtypes.WorkerProtocolVersion,
 	})
 	pipe.PExpire(ctx, key, ttl)
 

internal/heartbeat/monitor_test.go

@@ -5,6 +5,7 @@ import (
 	"log/slog"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -34,7 +35,7 @@ func newTestMonitor(t *testing.T, mr *miniredis.Miniredis, ollamaURL string) (*M
 
 	logger := slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError}))
 	counter := &atomic.Int32{}
-	m := NewMonitor(redisClient, cfg, testWorkerAddr, []string{"0xmodel1"}, counter, 0, logger, nil)
+	m := NewMonitor(redisClient, cfg, testWorkerAddr, []string{"0xmodel1"}, nil, counter, 0, logger, nil)
 	return m, redisClient
 }
 
@@ -240,6 +241,41 @@ func TestEmitOnce_RedisUnreachable(t *testing.T) {
 	require.Error(t, err, "EmitOnce must return error when Redis is stopped")
 }
 
+func TestEmit_WritesCapabilities(t *testing.T) {
+	t.Parallel()
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = rdb.Close() })
+
+	logger := slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError}))
+	counter := &atomic.Int32{}
+	m := NewMonitor(rdb, MonitorConfig{Interval: time.Second}, testWorkerAddr,
+		[]string{"0xmodel"}, []string{"search"}, counter, 4, logger, nil)
+	require.NoError(t, m.EmitOnce(t.Context()))
+
+	key := pkgtypes.HeartbeatRedisKey(testWorkerAddr)
+	got, err := rdb.HGet(t.Context(), key, pkgtypes.HBFieldCapabilities).Result()
+	require.NoError(t, err)
+	assert.JSONEq(t, `["search"]`, got)
+}
+
+func TestEmitWritesProtocolVersion(t *testing.T) {
+	t.Parallel()
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = rdb.Close() })
+
+	logger := slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError}))
+	counter := &atomic.Int32{}
+	m := NewMonitor(rdb, MonitorConfig{Interval: time.Second}, testWorkerAddr,
+		[]string{"0xmodel"}, nil, counter, 0, logger, nil)
+	require.NoError(t, m.EmitOnce(t.Context()))
+
+	got, err := rdb.HGet(t.Context(), pkgtypes.HeartbeatRedisKey(m.workerAddr), pkgtypes.HBFieldProtocolVersion).Result()
+	require.NoError(t, err)
+	require.Equal(t, strconv.Itoa(pkgtypes.WorkerProtocolVersion), got)
+}
+
 func TestMonitor_emit_DynamicJobCounts(t *testing.T) {
 	t.Parallel()
 	mr := miniredis.RunT(t)
@@ -259,7 +295,7 @@ func TestMonitor_emit_DynamicJobCounts(t *testing.T) {
 		OllamaURL: ollamaSrv.URL,
 	}
 	logger := slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError}))
-	m := NewMonitor(redisClient, cfg, testWorkerAddr, []string{"0xmodel1"}, counter, 5, logger, nil)
+	m := NewMonitor(redisClient, cfg, testWorkerAddr, []string{"0xmodel1"}, nil, counter, 5, logger, nil)
 
 	ctx := t.Context()
 	require.NoError(t, m.emit(ctx))