Settle failed jobs on chain (abandonJob), skip stale session requests, remove-models CLI

cmd/cli/main.go

@@ -78,6 +78,8 @@ func main() {
 		runRegister()
 	case "add-models":
 		runAddModels()
+	case "remove-models":
+		runRemoveModels()
 	case "reinstate":
 		runReinstate()
 	case "top-up-stake":
@@ -316,6 +318,35 @@ func runAddModels() {
 	}
 }
 
+// runRemoveModels takes SUPPORTED_MODELS as the list of models to REMOVE from
+// the worker's on-chain set, e.g.
+//   SUPPORTED_MODELS=deepseek-r1:32b lightchain-worker remove-models
+func runRemoveModels() {
+	cfg, logger := loadAndValidateCfg()
+	signingKey, workerAddr := loadSigningKey(cfg, logger)
+	modelIDs, modelNames := parseModels(cfg, logger)
+
+	chainClient := dialChain(cfg, signingKey, logger)
+	defer chainClient.Close()
+
+	h := &cli.Handler{
+		Client:     chainClient,
+		WorkerAddr: workerAddr,
+		ModelIDs:   modelIDs,
+		ModelNames: modelNames,
+		Out:        os.Stdout,
+		Logger:     logger,
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), txTimeout)
+	defer cancel()
+
+	if err := h.RemoveModels(ctx); err != nil {
+		logger.Error("remove-models failed", "error", err)
+		os.Exit(1)
+	}
+}
+
 func runDeregister() {
 	cfg, logger := loadAndValidateCfg()
 	signingKey, workerAddr := loadSigningKey(cfg, logger)

internal/chain/client.go

@@ -224,6 +224,13 @@ func (c *ChainClient) AddSupportedModel(ctx context.Context, modelID [32]byte) e
 	})
 }
 
+// RemoveSupportedModel drops a model from this worker's on-chain list (WorkerRegistry.removeSupportedModel).
+func (c *ChainClient) RemoveSupportedModel(ctx context.Context, modelID [32]byte) error {
+	return c.submitPreparedTx(ctx, "RemoveSupportedModel", nil, func(opts *bind.TransactOpts) (*types.Transaction, error) {
+		return c.registry.RemoveSupportedModel(opts, modelID)
+	})
+}
+
 // SetCapabilities declares this worker's full on-chain capability mask (overwrite, not merge)..
 func (c *ChainClient) SetCapabilities(ctx context.Context, mask *big.Int) error {
 	return c.submitPreparedTx(ctx, "SetCapabilities", nil, func(opts *bind.TransactOpts) (*types.Transaction, error) {
@@ -675,6 +682,17 @@ func (c *ChainClient) AcknowledgeJobAsync(
 	return pending.Hash(), pending.Wait, nil
 }
 
+// AbandonJob hands a job the worker cannot finish back to the registry
+// (JobRegistry.abandonJob): the consumer's fee is refunded and the job leaves
+// this worker's in-flight count at once instead of after deadline +
+// resolutionTimeout. Without it a worker that fails a few jobs an hour fills
+// the maxConcurrentJobs cap and the chain refuses its session claims.
+func (c *ChainClient) AbandonJob(ctx context.Context, jobID uint64) error {
+	return c.submitPreparedTx(ctx, "AbandonJob", nil, func(opts *bind.TransactOpts) (*types.Transaction, error) {
+		return c.jobRegistry.AbandonJob(opts, new(big.Int).SetUint64(jobID))
+	})
+}
+
 // CompleteJob submits a completeJob transaction with a single bytes32 response
 // blob hash. Post-audit the contract takes one bytes32 and enforces
 // `blobhash(0) == responseBlobHash`, so the blob-carrying TX must contain

internal/chain/client_test.go

@@ -29,6 +29,7 @@ type MockRegistrationClient struct {
 	IsWorkerRegisteredFn     func(ctx context.Context, worker common.Address) (bool, error)
 	RegisterWorkerFn         func(ctx context.Context, encryptionPubKey []byte, stake *big.Int) error
 	AddSupportedModelFn      func(ctx context.Context, modelID [32]byte) error
+	RemoveSupportedModelFn   func(ctx context.Context, modelID [32]byte) error
 	DeregisterWorkerFn       func(ctx context.Context) error
 	GetMinWorkerStakeFn      func(ctx context.Context) (*big.Int, error)
 	GetWorkerEncryptionKeyFn func(ctx context.Context, worker common.Address) ([]byte, error)
@@ -60,6 +61,12 @@ func (m *MockRegistrationClient) RegisterWorker(ctx context.Context, encryptionP
 func (m *MockRegistrationClient) AddSupportedModel(ctx context.Context, modelID [32]byte) error {
 	return m.AddSupportedModelFn(ctx, modelID)
 }
+func (m *MockRegistrationClient) RemoveSupportedModel(ctx context.Context, modelID [32]byte) error {
+	if m.RemoveSupportedModelFn == nil {
+		return nil
+	}
+	return m.RemoveSupportedModelFn(ctx, modelID)
+}
 
 func (m *MockRegistrationClient) DeregisterWorker(ctx context.Context) error {
 	return m.DeregisterWorkerFn(ctx)

internal/chain/interfaces.go

@@ -89,6 +89,10 @@ type RegistrationClient interface {
 	// Per Amendment A-2, this is non-payable — no additional ETH is required per model.
 	AddSupportedModel(ctx context.Context, modelID [32]byte) error
 
+	// RemoveSupportedModel drops a model ID from the worker's supported model set on-chain,
+	// so sortition stops routing that model's session requests to this worker.
+	RemoveSupportedModel(ctx context.Context, modelID [32]byte) error
+
 	// DeregisterWorker removes the worker from the registry and withdraws all stake.
 	DeregisterWorker(ctx context.Context) error
 

internal/cli/handler.go

@@ -155,6 +155,39 @@ func (h *Handler) AddModels(ctx context.Context) error {
 	return nil
 }
 
+// RemoveModels removes models from an already-registered worker's on-chain
+// list. SUPPORTED_MODELS in the environment only decides what the worker
+// serves; sortition routes by the registry, so a model the box can no longer
+// serve has to leave the registry too or its session requests keep landing
+// here and timing out.
+func (h *Handler) RemoveModels(ctx context.Context) error {
+	if len(h.ModelIDs) == 0 {
+		return fmt.Errorf("SUPPORTED_MODELS is required for remove-models: list the models to remove")
+	}
+	if len(h.ModelNames) != len(h.ModelIDs) {
+		return fmt.Errorf("ModelNames length %d does not match ModelIDs length %d", len(h.ModelNames), len(h.ModelIDs))
+	}
+
+	var failed []string
+	removed := 0
+	for i, modelID := range h.ModelIDs {
+		if err := h.Client.RemoveSupportedModel(ctx, modelID); err != nil {
+			h.Logger.Error("remove model failed", "model", h.ModelNames[i], "index", i, "error", err)
+			failed = append(failed, fmt.Sprintf("%q: %v", h.ModelNames[i], err))
+			continue
+		}
+		removed++
+		h.Logger.Info("model removed", "model", h.ModelNames[i])
+	}
+
+	_, _ = fmt.Fprintf(h.Out, "Removed %d of %d models for worker %s\n", removed, len(h.ModelIDs), h.WorkerAddr.Hex())
+	if len(failed) > 0 {
+		return fmt.Errorf("%d model(s) not removed (a model the worker never supported reverts): %s",
+			len(failed), strings.Join(failed, "; "))
+	}
+	return nil
+}
+
 // Deregister removes the worker from the registry and withdraws stake.
 func (h *Handler) Deregister(ctx context.Context) error {
 	mgr := registration.NewManager(h.Client, h.WorkerAddr, nil, h.Logger)

internal/cli/handler_test.go

@@ -26,6 +26,7 @@ type mockRegClient struct {
 	isWorkerRegisteredFn     func(ctx context.Context, worker common.Address) (bool, error)
 	registerWorkerFn         func(ctx context.Context, encryptionPubKey []byte, stake *big.Int) error
 	addSupportedModelFn      func(ctx context.Context, modelID [32]byte) error
+	removeSupportedModelFn   func(ctx context.Context, modelID [32]byte) error
 	deregisterWorkerFn       func(ctx context.Context) error
 	getMinWorkerStakeFn      func(ctx context.Context) (*big.Int, error)
 	getWorkerEncryptionKeyFn func(ctx context.Context, worker common.Address) ([]byte, error)
@@ -55,6 +56,12 @@ func (m *mockRegClient) RegisterWorker(ctx context.Context, encryptionPubKey []b
 func (m *mockRegClient) AddSupportedModel(ctx context.Context, modelID [32]byte) error {
 	return m.addSupportedModelFn(ctx, modelID)
 }
+func (m *mockRegClient) RemoveSupportedModel(ctx context.Context, modelID [32]byte) error {
+	if m.removeSupportedModelFn == nil {
+		return nil
+	}
+	return m.removeSupportedModelFn(ctx, modelID)
+}
 func (m *mockRegClient) DeregisterWorker(ctx context.Context) error {
 	return m.deregisterWorkerFn(ctx)
 }

internal/cli/init_test.go

@@ -198,6 +198,8 @@ func (c *initChain) RegisterWorker(_ context.Context, encKey []byte, stake *big.
 	return nil
 }
 
+func (c *initChain) RemoveSupportedModel(context.Context, [32]byte) error { return nil }
+
 func (c *initChain) AddSupportedModel(_ context.Context, id [32]byte) error {
 	c.addCalls = append(c.addCalls, id)
 	if c.addErr != nil {

internal/config/config.go

@@ -57,11 +57,18 @@ type Config struct {
 	// 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
+	SearchEnabled bool
+	// PublicOutputEnabled serves envelope v4 public-output jobs and
+	// advertises the "public-output" capability. Default on; turn it off
+	// on an engine without JSON-schema `format` support. ExactProfile
+	// additionally advertises "exact-profile": the operator asserts the
+	// pinned deterministic profile for exact-match action jobs.
+	PublicOutputEnabled bool
+	ExactProfile        bool
+	TavilyAPIKey        string
+	TavilyURL           string
+	SearchTimeout       time.Duration
+	SearchMaxResults    int
 	// OllamaStream selects stream=true inference. When true the worker
 	// publishes incremental `chunk` frames as tokens arrive, collapsing
 	// time-to-first-token from full generation time to ~1s. The terminal
@@ -312,6 +319,11 @@ type Config struct {
 
 	// Sortition mode (Phase 3 — worker self-claims sessions, no dispatcher).
 	SortitionEnabled       bool
+	// AbandonFailedJobs makes the worker call JobRegistry.abandonJob for a job
+	// it failed, refunding the consumer and freeing its in-flight slot at once
+	// instead of after deadline + resolutionTimeout. Needs the abandonJob
+	// contract upgrade; off by default until it is live (ABANDON_FAILED_JOBS).
+	AbandonFailedJobs bool
 	SessionManagerAddress  common.Address
 	SortitionStateDir      string
 	SortitionChunkSize     uint64
@@ -326,6 +338,12 @@ type Config struct {
 	// (1800 blocks at 2 s slots, 600 at 6 s); raise it where callers open
 	// requests with longer expiries. Defaults to 2000; 0 disables.
 	SortitionSessionLookbackBlocks uint64
+	// SortitionMaxRequestAgeBlocks drops a pending session request once it is
+	// this many blocks old without a claim. The consumer-api stops waiting 60 s
+	// after it requested the session and never provides keys for it, so a
+	// claim after that only spends gas on a session nobody uses. Defaults to
+	// 150 (5 min at 2 s slots, 15 min at 6 s); 0 disables.
+	SortitionMaxRequestAgeBlocks uint64
 	// SortitionSessionRetryLimit bounds how many passes the JobWatcher retries
 	// a job whose session is not Active before giving up and skipping it, so
 	// one session that never returns to Active cannot wedge the watcher and
@@ -461,6 +479,12 @@ func Load() (*Config, error) {
 	}
 	cfg.OllamaTimeout = parseDuration("OLLAMA_TIMEOUT", "120s", &errs)
 	cfg.SearchEnabled = parseBool("SEARCH_ENABLED", false, &errs)
+	cfg.AbandonFailedJobs = parseBool("ABANDON_FAILED_JOBS", false, &errs)
+	cfg.PublicOutputEnabled = parseBool("PUBLIC_OUTPUT", true, &errs)
+	cfg.ExactProfile = parseBool("EXACT_PROFILE", false, &errs)
+	if cfg.ExactProfile && !cfg.PublicOutputEnabled {
+		errs = append(errs, "EXACT_PROFILE requires PUBLIC_OUTPUT")
+	}
 	cfg.SearchTimeout = parseDuration("SEARCH_TIMEOUT", "10s", &errs)
 	cfg.SearchMaxResults = parseInt("SEARCH_MAX_RESULTS", 5, &errs)
 	cfg.AckTxTimeout = parseDuration("ACK_TX_TIMEOUT", "15s", &errs)
@@ -564,6 +588,7 @@ func Load() (*Config, error) {
 	cfg.SortitionPollInterval = parseDuration("SORTITION_POLL_INTERVAL", "1s", &errs)
 	cfg.SortitionConfirmations = parseUint64("SORTITION_CONFIRMATIONS", 0, &errs)
 	cfg.SortitionSessionLookbackBlocks = parseUint64("SORTITION_SESSION_LOOKBACK_BLOCKS", 2000, &errs)
+	cfg.SortitionMaxRequestAgeBlocks = parseUint64("SORTITION_MAX_REQUEST_AGE_BLOCKS", 150, &errs)
 	cfg.SortitionSessionRetryLimit = parseInt("SORTITION_SESSION_RETRY_LIMIT", 10, &errs)
 
 	if len(errs) > 0 {

internal/ollama/client.go

@@ -31,6 +31,10 @@ const (
 // (re-execution) rely on greedy decoding being reproducible across services.
 const deterministicSeed = 42
 
+// DeterministicSeed is deterministicSeed for callers that record the seed a
+// job ran under (the public response R echoes it).
+const DeterministicSeed = deterministicSeed
+
 // Options mirrors the Ollama request "options" object. Only the knobs the
 // worker needs are modelled. Every field is omitempty so an unset option
 // leaves the server-side default untouched.