fix(nomad): bound control requests with durable registration recovery

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Bound Nomad's finite control-plane requests while retaining durable recovery identity for uncertain registration, preserving caller cancellation and keeping established exec streams outside the request ceiling. [PR 1916](https://github.com/openclaw/crabbox/pull/1916). Thanks @SebTardif.
 - Preserve GCP capacity fallback when a bounded error summary omits retry evidence, while keeping user-visible diagnostics redacted and bounded. [PR 1987](https://github.com/openclaw/crabbox/pull/1987). Thanks @steipete.
 - AWS: add administrator-only legacy cleanup recovery backed by authenticated original CloudTrail allocation evidence, preserving remaining key and access cleanup and recording an atomic scope-recovery audit without force-success. [PR 1975](https://github.com/openclaw/crabbox/pull/1975).
 - AWS: complete cleanup after a verified empty instance response without skipping owned keys, bind new leases to the original account and Region, and retain unresolved historical cleanup when that authority is missing. [PR 1904](https://github.com/openclaw/crabbox/pull/1904). Thanks @vincentkoc.

docs/providers/nomad.md

@@ -70,6 +70,29 @@ region and namespace values. A reachable ACL-disabled cluster passes without a
 token; an anonymous `401`/`403` reports the missing token environment variable.
 Every check prints `mutation=false`.
 
+Finite JSON calls (`agent.self`, regions, namespace information, job registration,
+job information, job allocations, evaluation information and deregistration)
+have a two-minute request ceiling, preserving earlier caller deadlines and
+cancellation. Regions queries carry the caller context and retain sorted results.
+Internally created HTTP transports also bound response-header waits to 30 seconds;
+injected clients retain their settings. No whole-request client timeout is added
+to established allocation exec streams. Exec startup's HTTP node discovery can
+encounter the header deadline before the WebSocket connection is established.
+
+Before registration, Crabbox durably records the exact job and lease identity.
+If registration remains uncertain after reconciliation, Crabbox retains that
+recovery claim; a single missing-job response does not prove registration was
+rejected. `status` and `list` show
+`registration-pending` while a submitted job is absent, and `stop`/`cleanup`
+retain the claim with an unknown-outcome diagnostic. A prepared attempt that was
+never submitted can be removed locally. Matching observed jobs use the existing
+ownership-checked removal path; unexpected state is not adopted or deleted.
+Run failures expose a kept recovery session only for the exact retained claim,
+and warmup failures retain recovery identifiers in their diagnostic. Registration
+is not automatically retried. Existing claims without registration markers keep
+their confirmed-job behavior. Setup-failure rollback remains adapter-owned and
+does not become a retained successful allocation merely because `--keep` is set.
+
 `warmup` creates a Nomad job and local Crabbox claim. The job stays running
 until explicit `stop` or `cleanup`, even if `--keep` is omitted. A `run` without
 `--id` creates a fresh job and deletes it after the command unless `--keep` or

internal/cli/claim.go

@@ -1636,6 +1636,10 @@ func replaceLeaseClaimIfUnchangedDurableReturning(leaseID string, current, repla
 	return replaceLeaseClaimIfUnchangedWithWrite(leaseID, current, replacement, writeLeaseClaimAtomicDurable)
 }
 
+func replaceLeaseClaimIfUnchangedDurableReturningContext(ctx context.Context, leaseID string, current, replacement leaseClaim) (leaseClaim, error) {
+	return replaceLeaseClaimTransactionContext(ctx, leaseID, current, replacement, nil, writeLeaseClaimAtomicDurable)
+}
+
 func replaceLeaseClaimIfUnchangedDurableAfter(leaseID string, current, replacement leaseClaim, action func() error) (leaseClaim, error) {
 	// This entrypoint binds replacement identity; restore/replace retain the
 	// supplied payload, including incomplete claims used by rollback.
@@ -1648,7 +1652,12 @@ func replaceLeaseClaimIfUnchangedWithWrite(leaseID string, current, replacement
 }
 
 func replaceLeaseClaimTransaction(leaseID string, current, replacement leaseClaim, action func() error, write func(string, leaseClaim) error) (leaseClaim, error) {
+	return replaceLeaseClaimTransactionContext(context.Background(), leaseID, current, replacement, action, write)
+}
+
+func replaceLeaseClaimTransactionContext(ctx context.Context, leaseID string, current, replacement leaseClaim, action func() error, write func(string, leaseClaim) error) (leaseClaim, error) {
 	return transactLeaseClaim(leaseID, leaseClaimTransaction{
+		context:  ctx,
 		guard:    unchangedLeaseClaimGuard(leaseID, current, true),
 		action:   claimTransactionAction(action),
 		revision: claimRevisionAfterMutation,

internal/cli/claim_lock_test.go

@@ -269,7 +269,7 @@ func TestClaimSharedFinalizationPreservesReplacement(t *testing.T) {
 }
 
 func TestClaimFenceContextCancelsPublicationWithoutMutation(t *testing.T) {
-	for _, operation := range []string{"reuse", "publish", "finalize", "shared"} {
+	for _, operation := range []string{"reuse", "publish", "finalize", "shared", "durable replacement"} {
 		t.Run(operation, func(t *testing.T) {
 			t.Setenv("XDG_STATE_HOME", t.TempDir())
 			const id = "cbx_shared_publication"
@@ -300,6 +300,11 @@ func TestClaimFenceContextCancelsPublicationWithoutMutation(t *testing.T) {
 					return WithDurableLeaseClaimLockContext(ctx, id, func(*LeaseClaim, bool, func() error) error { return action() })
 				case "finalize":
 					return CleanupLeaseClaimIfUnchangedAfterContext(ctx, id, claim, true, action)
+				case "durable replacement":
+					replacement := cloneLeaseClaim(claim)
+					replacement.Labels = map[string]string{"state": "submitting"}
+					_, err := ReplaceLeaseClaimIfUnchangedDurableReturningContext(ctx, id, claim, replacement)
+					return err
 				default:
 					return WithLeaseClaimUnchangedShared(ctx, id, claim, action)
 				}

internal/cli/claim_transaction_test.go

@@ -3,6 +3,7 @@ package cli
 import (
 	"context"
 	"errors"
+	"fmt"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -154,6 +155,44 @@ func TestClaimTransactionContractActionFailureAndRevision(t *testing.T) {
 	}
 }
 
+func TestClaimTransactionDurableReplacementContext(t *testing.T) {
+	for _, contextual := range []bool{false, true} {
+		for _, stale := range []bool{false, true} {
+			t.Run(fmt.Sprintf("context=%t/stale=%t", contextual, stale), func(t *testing.T) {
+				stored := seedClaimContract(t)
+				expected := cloneLeaseClaim(stored)
+				if stale {
+					expected.Revision = "older"
+				}
+				replacement := cloneLeaseClaim(stored)
+				replacement.Labels = map[string]string{"state": "submitting"}
+				var updated leaseClaim
+				var err error
+				if contextual {
+					updated, err = ReplaceLeaseClaimIfUnchangedDurableReturningContext(t.Context(), stored.LeaseID, expected, replacement)
+				} else {
+					updated, err = ReplaceLeaseClaimIfUnchangedDurableReturning(stored.LeaseID, expected, replacement)
+				}
+				if stale {
+					if err == nil || !strings.Contains(err.Error(), "claim changed") {
+						t.Fatalf("stale replacement err=%v", err)
+					}
+					assertClaimContractStored(t, stored.LeaseID, stored)
+					return
+				}
+				if err != nil || updated.Revision == "" || updated.Revision == stored.Revision {
+					t.Fatalf("updated=%#v err=%v", updated, err)
+				}
+				replacement.Revision = updated.Revision
+				if !reflect.DeepEqual(updated, replacement) {
+					t.Fatalf("replacement policy changed: got=%#v want=%#v", updated, replacement)
+				}
+				assertClaimContractStored(t, stored.LeaseID, updated)
+			})
+		}
+	}
+}
+
 // The provider hook observes different revision phases for input endpoints and
 // action-produced endpoints. It must always run after the exact-claim guard.
 type claimContractProvider struct {

internal/cli/provider_exports.go

@@ -290,6 +290,12 @@ func ReplaceLeaseClaimIfUnchangedDurableReturning(leaseID string, current, repla
 	return replaceLeaseClaimIfUnchangedDurableReturning(leaseID, current, replacement)
 }
 
+// ReplaceLeaseClaimIfUnchangedDurableReturningContext also bounds waiting for
+// the claim fence. Existing context-free replacement APIs remain detached.
+func ReplaceLeaseClaimIfUnchangedDurableReturningContext(ctx context.Context, leaseID string, current, replacement LeaseClaim) (LeaseClaim, error) {
+	return replaceLeaseClaimIfUnchangedDurableReturningContext(ctx, leaseID, current, replacement)
+}
+
 func ReplaceLeaseClaimIfUnchangedDurableAfter(leaseID string, current, replacement LeaseClaim, action func() error) (LeaseClaim, error) {
 	return replaceLeaseClaimIfUnchangedDurableAfter(leaseID, current, replacement, action)
 }

internal/providers/nomad/claim.go

@@ -41,17 +41,6 @@ func claimScope(cfg Config) string {
 	}, "|")
 }
 
-func writeNomadClaim(cfg Config, leaseID, slug string, repo Repo, reclaim bool, ready allocationReadiness, expiresAt time.Time) (LeaseClaim, error) {
-	if err := claimLeaseForRepoProviderScopePond(leaseID, slug, providerName, claimScope(cfg), cfg.Pond, repo.Root, cfg.IdleTimeout, reclaim); err != nil {
-		return LeaseClaim{}, err
-	}
-	claim, err := readLeaseClaim(leaseID)
-	if err != nil {
-		return LeaseClaim{}, err
-	}
-	return updateLeaseClaimLabelsIfUnchanged(leaseID, claim, claimLabels(cfg, leaseID, slug, ready, expiresAt))
-}
-
 func claimLabels(cfg Config, leaseID, slug string, ready allocationReadiness, expiresAt time.Time) map[string]string {
 	labels := map[string]string{
 		"provider":                providerName,
@@ -96,6 +85,9 @@ func resolveNomadClaim(cfg Config, id string) (LeaseClaim, error) {
 }
 
 func authorizeClaimScope(cfg Config, claim LeaseClaim) error {
+	if _, err := registrationState(claim); err != nil {
+		return err
+	}
 	if claim.Provider != "" && claim.Provider != providerName {
 		return exit(2, "lease %s belongs to provider=%s, not %s", claim.LeaseID, claim.Provider, providerName)
 	}
@@ -118,6 +110,9 @@ func listNomadLeaseClaims() ([]LeaseClaim, error) {
 }
 
 func validateRemoteOwnership(cfg Config, claim LeaseClaim, job *nomadapi.Job) error {
+	if _, err := registrationState(claim); err != nil {
+		return err
+	}
 	if job == nil {
 		return exit(4, "nomad job for lease %s is missing or inaccessible", claim.LeaseID)
 	}

internal/providers/nomad/cleanup_ownership_test.go

@@ -64,11 +64,8 @@ func TestNomadDestructionFencesValidationPurgeAndAbsence(t *testing.T) {
 				expireClaim(t, claim, core.ClockNow(b.rt.Clock).Add(-time.Hour))
 				claim, _ = readLeaseClaim(claim.LeaseID)
 			}
-			expectedJob := cloneJob(fake.jobs[claim.Labels[claimLabelJobID]])
 			if operation == "rollback" {
-				if err := core.RemoveLeaseClaimIfUnchanged(claim.LeaseID, claim); err != nil {
-					t.Fatal(err)
-				}
+				claim = markRegistrationClaim(t, claim, fake.jobs[claim.Labels[claimLabelJobID]], registrationConfirmed)
 			}
 			lockPath := filepath.Join(os.Getenv("XDG_STATE_HOME"), "crabbox", "claim-locks", claim.LeaseID+".json.lock")
 			if err := os.MkdirAll(filepath.Dir(lockPath), 0o700); err != nil {
@@ -111,9 +108,9 @@ func TestNomadDestructionFencesValidationPurgeAndAbsence(t *testing.T) {
 				err = b.deleteOwnedRunJob(context.Background(), client, claim)
 			case "rollback":
 				cause := errors.New("setup failed")
-				err = b.cleanupUnclaimedJob(context.Background(), client, expectedJob, cause)
-				if err == cause {
-					err = nil
+				recovery, failure := b.rollbackRegistration(context.Background(), client, claim, cause)
+				if recovery != nil || !errors.Is(failure, cause) || !strings.Contains(failure.Error(), "rolled back") {
+					t.Fatalf("rollback result recovery=%#v err=%v", recovery, failure)
 				}
 			}
 			if err != nil || purges != 1 || reads < 2 {
@@ -320,23 +317,26 @@ func TestNomadKeptRunRefreshFencesClaimChangedDuringExecution(t *testing.T) {
 	}
 }
 
-func TestNomadSetupRollbackRetainsPublishedClaim(t *testing.T) {
+func TestNomadSetupRollbackRetainsChangedClaim(t *testing.T) {
 	for _, partial := range []bool{false, true} {
 		t.Run(strconv.FormatBool(partial), func(t *testing.T) {
 			fake := newLifecycleFakeClient()
 			b, _, _ := testBackend(t, fake)
 			claim := createClaim(t, b, "cbx_a44444444444", "setup-crab", "crabbox-a44444444444", "alloc-a")
-			expected := cloneJob(fake.jobs[claim.Labels[claimLabelJobID]])
+			claim = markRegistrationClaim(t, claim, fake.jobs[claim.Labels[claimLabelJobID]], registrationConfirmed)
+			expected := claim
+			labels := maps.Clone(claim.Labels)
+			labels["owner_revision"] = "replacement"
 			if partial {
-				var err error
-				claim, err = core.UpdateLeaseClaimLabelsIfUnchanged(claim.LeaseID, claim, nil)
-				if err != nil {
-					t.Fatal(err)
-				}
+				labels = nil
+			}
+			claim, err := core.UpdateLeaseClaimLabelsIfUnchanged(claim.LeaseID, claim, labels)
+			if err != nil {
+				t.Fatal(err)
 			}
 			cause := errors.New("publication failed")
-			err := b.cleanupUnclaimedJob(context.Background(), fake, expected, cause)
-			if !errors.Is(err, cause) || err == cause || len(fake.deregisters) != 0 {
+			recovery, err := b.rollbackRegistration(context.Background(), fake, expected, cause)
+			if recovery != nil || !errors.Is(err, cause) || err == cause || len(fake.deregisters) != 0 {
 				t.Fatalf("err=%v purges=%v", err, fake.deregisters)
 			}
 			assertNomadClaimRetained(t, claim)

internal/providers/nomad/client.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
 	"time"
 
@@ -35,12 +36,29 @@ type liveClient struct {
 	cfg    Config
 }
 
+const (
+	defaultNomadControlRequestTimeout = 2 * time.Minute
+	nomadDefaultResponseHeaderTimeout = 30 * time.Second
+)
+
 var (
 	errNomadCrossOriginRedirect = errors.New("nomad refused cross-origin redirect")
 	errNomadInvalidRedirect     = errors.New("nomad refused invalid redirect")
 	errNomadRedirectLimit       = errors.New("nomad redirect stopped after 10 redirects")
+	nomadControlRequestTimeout  = defaultNomadControlRequestTimeout
 )
 
+// controlRequestContext bounds finite JSON calls, not allocation exec.
+func controlRequestContext(ctx context.Context) (context.Context, context.CancelFunc) {
+	if nomadControlRequestTimeout <= 0 {
+		return ctx, func() {}
+	}
+	if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) <= nomadControlRequestTimeout {
+		return ctx, func() {}
+	}
+	return context.WithTimeout(ctx, nomadControlRequestTimeout)
+}
+
 func newNomadClient(cfg Config, rt Runtime) (Client, error) {
 	apiConfig, err := newNomadAPIConfig(cfg, os.Getenv)
 	if err != nil {
@@ -76,6 +94,7 @@ func configureNomadHTTPClient(apiConfig *nomadapi.Config, source *http.Client) e
 			source = cleanhttp.DefaultPooledClient()
 			transport = source.Transport.(*http.Transport)
 		}
+		transport.ResponseHeaderTimeout = nomadDefaultResponseHeaderTimeout
 		transport.TLSHandshakeTimeout = 10 * time.Second
 		transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12}
 		transport.ForceAttemptHTTP2 = false
@@ -160,6 +179,8 @@ func newNomadAPIConfig(cfg Config, lookup func(string) string) (*nomadapi.Config
 }
 
 func (c liveClient) AgentSelf(ctx context.Context) (*nomadapi.AgentSelf, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
 	var value nomadapi.AgentSelf
 	query := (&nomadapi.QueryOptions{}).WithContext(ctx)
 	if _, err := c.client.Raw().Query("/v1/agent/self", &value, query); err != nil {
@@ -168,17 +189,28 @@ func (c liveClient) AgentSelf(ctx context.Context) (*nomadapi.AgentSelf, error)
 	return &value, nil
 }
 
-func (c liveClient) Regions(context.Context) ([]string, error) {
-	value, err := c.client.Regions().List()
-	return value, sanitizeNomadClientError(err)
+func (c liveClient) Regions(ctx context.Context) ([]string, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
+	var value []string
+	query := (&nomadapi.QueryOptions{}).WithContext(ctx)
+	if _, err := c.client.Raw().Query("/v1/regions", &value, query); err != nil {
+		return nil, sanitizeNomadClientError(err)
+	}
+	sort.Strings(value)
+	return value, nil
 }
 
 func (c liveClient) NamespaceInfo(ctx context.Context, namespace string) (*nomadapi.Namespace, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
 	ns, _, err := c.client.Namespaces().Info(namespace, c.queryOptions(ctx))
 	return ns, sanitizeNomadClientError(err)
 }
 
 func (c liveClient) RegisterJob(ctx context.Context, job *nomadapi.Job) (string, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
 	resp, _, err := c.client.Jobs().RegisterOpts(job, &nomadapi.RegisterOptions{
 		EnforceIndex: true,
 		ModifyIndex:  0,
@@ -193,21 +225,29 @@ func (c liveClient) RegisterJob(ctx context.Context, job *nomadapi.Job) (string,
 }
 
 func (c liveClient) JobInfo(ctx context.Context, jobID string) (*nomadapi.Job, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
 	job, _, err := c.client.Jobs().Info(jobID, c.queryOptions(ctx))
 	return job, sanitizeNomadClientError(err)
 }
 
 func (c liveClient) JobAllocations(ctx context.Context, jobID string, all bool) ([]*nomadapi.AllocationListStub, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
 	allocs, _, err := c.client.Jobs().Allocations(jobID, all, c.queryOptions(ctx))
 	return allocs, sanitizeNomadClientError(err)
 }
 
 func (c liveClient) EvaluationInfo(ctx context.Context, evalID string) (*nomadapi.Evaluation, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
 	eval, _, err := c.client.Evaluations().Info(evalID, c.queryOptions(ctx))
 	return eval, sanitizeNomadClientError(err)
 }
 
 func (c liveClient) DeregisterJob(ctx context.Context, jobID string, purge bool) (string, error) {
+	ctx, cancel := controlRequestContext(ctx)
+	defer cancel()
 	evalID, _, err := c.client.Jobs().Deregister(jobID, purge, c.writeOptions(ctx))
 	return evalID, sanitizeNomadClientError(err)
 }