Merge pull request #176 from BaaRai/feat/claim-heartbeat-fencing

feat(engine): intent claim heartbeat, TTL and fencing token (S3c)

agent/cancelcause.go

@@ -47,6 +47,8 @@ var (
 		"The user paused a running Worker. This call was cancelled and the intent turned paused; all recorded intents, facts, vulnerabilities, and activity records are kept, and execution restarts from scratch after resuming")
 	AbortWorkCancelledByUser = cause("work_cancelled_by_user", "The user deleted this Worker intent",
 		"The user deleted a running Worker. This call was cancelled; after the Worker exits the write area, the server handles the intent per the user-chosen deletion mode — soft delete only marks it deleted and keeps all outputs, hard delete cascades to remove the intent and the downstream nodes supported only by it")
+	AbortWorkClaimLost = cause("work_claim_lost", "Another Worker took this intent over",
+		"This Worker's claim on the intent expired (no heartbeat reached the database within the claim TTL) and another Worker took the intent over, so its heartbeat was rejected. This run was cancelled and its result is discarded; the intent now belongs to the new claimer")
 	AbortWorkFinished = cause("work_finished", "Worker finished normally, context freed",
 		"The Worker finished normally, and the engine released its context resources in detachWork. This is not an interruption; if it appears in an interruption message, the cancel and finish events raced")
 	AbortPausedRaceGuard = cause("paused_race_guard", "Refused a new run while task is paused",

agent/terminalreason_test.go

@@ -141,7 +141,7 @@ func TestAbortCausesAreWellFormed(t *testing.T) {
 	all := []*AbortCause{
 		AbortPausedByUser, AbortPausedByOrchestrator, AbortTaskDeleted, AbortPausedOnReload,
 		AbortGoalMet, AbortSettleDrainTimeout, AbortKilledByPlanner, AbortWorkPausedByUser,
-		AbortWorkCancelledByUser, AbortWorkFinished, AbortPausedRaceGuard,
+		AbortWorkCancelledByUser, AbortWorkClaimLost, AbortWorkFinished, AbortPausedRaceGuard,
 		AbortChatStoppedByUser, AbortChatPausedWithTask, AbortChatTurnFinished,
 		AbortShutdown, AbortRunHardTimeout,
 	}

db/exploration.go

@@ -2,7 +2,9 @@ package db
 
 import (
 	"bytes"
+	"crypto/rand"
 	"database/sql"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -16,6 +18,33 @@ import (
 // it to distinguish an expected control race (HTTP 409) from a storage failure.
 var ErrIntentStateConflict = errors.New("intent state changed concurrently")
 
+// ErrIntentClaimLost marks a heartbeat or a result write rejected by the claim
+// fence: the caller's unclaim token is no longer the intent's current one (its
+// claim expired and another worker took the intent over, or it was already
+// settled). Nothing was written; the caller must stop working on that intent.
+var ErrIntentClaimLost = errors.New("intent claim lost")
+
+// IntentClaimTTL is how long an intent claim stays valid without a heartbeat. A
+// running intent whose last_heartbeat_at is older than this (per the database
+// clock) is considered abandoned — its worker crashed or its process is gone —
+// and may be taken over; a younger claim is live and is never taken over. It is
+// six heartbeat intervals, so a live worker loses its claim only after five
+// consecutive missed heartbeats (about 75s of lost database connectivity).
+const IntentClaimTTL = 90 * time.Second
+
+// IntentClaimHeartbeatInterval is how often a busy worker renews its claim, and
+// how often the engine looks for expired claims to take over.
+const IntentClaimHeartbeatInterval = 15 * time.Second
+
+// newUnclaimToken returns a fresh random claim token (128 bits, hex).
+func newUnclaimToken() (string, error) {
+	var b [16]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(b[:]), nil
+}
+
 // utf8Clean makes a string safe for a PostgreSQL text column. It (1) replaces
 // invalid UTF-8 byte sequences with U+FFFD and (2) strips NUL (0x00) bytes.
 // Tool output (nmap/curl/… stdout) can carry raw/truncated bytes that PostgreSQL's
@@ -309,9 +338,19 @@ func (s *ExplorationStore) DeleteGoal(id int64) error {
 // is re-claimed. Called on startup: a 'running' intent with no live worker (a
 // backend restart or crashed worker goroutine left it stuck) would otherwise spin
 // forever in the UI. Workers resume from their transcript, so reopening is safe.
+//
+// Only an expired claim is taken over: last_heartbeat_at older than
+// IntentClaimTTL, or NULL (a legacy row claimed before the lease columns
+// existed, whose claimer never heartbeats). A live claim — including one held by
+// another instance on the same database — is left alone. The expiry check and
+// the takeover are one atomic UPDATE on the database clock, and clearing
+// unclaim_token fences the previous claimer out. The engine also calls this
+// periodically, so a claim that was still live at startup is taken over once it
+// expires.
 func (s *ExplorationStore) ResetRunningIntents() (int64, error) {
-	res, err := s.db.Exec(`UPDATE exploration_nodes SET state='open', completed_at=NULL, blocked_reason=NULL
-WHERE exploration_id=$1 AND kind='intent' AND state='running'`, s.expID)
+	res, err := s.db.Exec(`UPDATE exploration_nodes SET state='open', completed_at=NULL, blocked_reason=NULL, unclaim_token=NULL
+WHERE exploration_id=$1 AND kind='intent' AND state='running'
+  AND (last_heartbeat_at IS NULL OR last_heartbeat_at < now() - $2 * INTERVAL '1 second')`, s.expID, IntentClaimTTL.Seconds())
 	if err != nil {
 		return 0, err
 	}
@@ -1216,13 +1255,107 @@ WHERE exploration_id=$1 AND kind='goal' AND state='open' ORDER BY id`, s.expID)
 
 // ClaimIntent atomically moves an open intent to running. Returns true if claimed.
 func (s *ExplorationStore) ClaimIntent(id int64, owner string) (bool, error) {
-	res, err := s.db.Exec(`UPDATE exploration_nodes SET state='running', owner=$1
-WHERE id=$2 AND exploration_id=$3 AND kind='intent' AND state='open'`, owner, id, s.expID)
+	token, err := s.ClaimIntentToken(id, owner)
+	return token != "", err
+}
+
+// ClaimIntentToken is ClaimIntent returning the claim's unclaim token ("" when
+// the intent was not claimable). Every claim gets a fresh random token and
+// starts its lease (last_heartbeat_at) in the same statement; the claimer must
+// present the token to HeartbeatIntentClaim and to the SettleIntentClaim* writes.
+func (s *ExplorationStore) ClaimIntentToken(id int64, owner string) (string, error) {
+	token, err := newUnclaimToken()
 	if err != nil {
-		return false, err
+		return "", err
 	}
-	n, _ := res.RowsAffected()
-	return n == 1, nil
+	res, err := s.db.Exec(`UPDATE exploration_nodes SET state='running', owner=$1, unclaim_token=$4, last_heartbeat_at=now()
+WHERE id=$2 AND exploration_id=$3 AND kind='intent' AND state='open'`, owner, id, s.expID, token)
+	if err != nil {
+		return "", err
+	}
+	if n, _ := res.RowsAffected(); n != 1 {
+		return "", nil
+	}
+	return token, nil
+}
+
+// ClaimPausedIntentToken is the paused->running counterpart of ClaimIntentToken
+// (the human-message path, which never goes through 'open'). It applies the same
+// column updates as CompareAndSetIntentState(id, "paused", "running") and starts
+// a claim with a fresh token. Returns "" when the intent is no longer paused.
+func (s *ExplorationStore) ClaimPausedIntentToken(id int64) (string, error) {
+	token, err := newUnclaimToken()
+	if err != nil {
+		return "", err
+	}
+	res, err := s.db.Exec(`UPDATE exploration_nodes
+SET state='running', blocked_reason=NULL, content_version=content_version+1, completed_at=NULL, unclaim_token=$3, last_heartbeat_at=now()
+WHERE id=$1 AND exploration_id=$2 AND kind='intent' AND state='paused'`, id, s.expID, token)
+	if err != nil {
+		return "", err
+	}
+	if n, _ := res.RowsAffected(); n != 1 {
+		return "", nil
+	}
+	return token, nil
+}
+
+// HeartbeatIntentClaim renews the lease of the claim identified by token. It
+// touches the row only while token is the running intent's current unclaim
+// token; otherwise it returns ErrIntentClaimLost and writes nothing.
+func (s *ExplorationStore) HeartbeatIntentClaim(id int64, token string) error {
+	res, err := s.db.Exec(`UPDATE exploration_nodes SET last_heartbeat_at=now()
+WHERE id=$1 AND exploration_id=$2 AND kind='intent' AND state='running' AND unclaim_token=$3`, id, s.expID, token)
+	if err != nil {
+		return err
+	}
+	if n, _ := res.RowsAffected(); n != 1 {
+		return fmt.Errorf("%w: heartbeat for intent %d rejected", ErrIntentClaimLost, id)
+	}
+	return nil
+}
+
+// SettleIntentClaim is the fenced write of a worker's intent result: it moves
+// the running intent to state and releases the claim, but only while token is
+// still the intent's current unclaim token. A worker whose claim was taken over
+// gets ErrIntentClaimLost and nothing is written. ErrIntentStateConflict means
+// the claim is still the caller's but the intent left 'running' concurrently.
+func (s *ExplorationStore) SettleIntentClaim(id int64, token, state string) error {
+	terminal := state == "done" || state == "blocked" || state == "exhausted" || state == "stopped"
+	res, err := s.db.Exec(`UPDATE exploration_nodes
+SET state=$1, blocked_reason=NULL, content_version=content_version+1, completed_at = CASE WHEN $5 THEN now() ELSE NULL END, unclaim_token=NULL
+WHERE id=$2 AND exploration_id=$3 AND kind='intent' AND state='running' AND unclaim_token=$4`, state, id, s.expID, token, terminal)
+	return s.settleIntentClaimResult(id, token, res, err)
+}
+
+// SettleIntentClaimBlocked is the fenced counterpart of SetIntentBlockedReason:
+// it settles the running intent as blocked with reason under the same token
+// check as SettleIntentClaim.
+func (s *ExplorationStore) SettleIntentClaimBlocked(id int64, token, reason string) error {
+	res, err := s.db.Exec(`UPDATE exploration_nodes
+SET state='blocked', blocked_reason=NULLIF($1,''), completed_at=now(), unclaim_token=NULL
+WHERE id=$2 AND exploration_id=$3 AND kind='intent' AND state='running' AND unclaim_token=$4`, reason, id, s.expID, token)
+	return s.settleIntentClaimResult(id, token, res, err)
+}
+
+// settleIntentClaimResult turns a fenced settle that matched no row into the
+// explicit rejection: claim lost unless the token is still the current one.
+func (s *ExplorationStore) settleIntentClaimResult(id int64, token string, res sql.Result, err error) error {
+	if err != nil {
+		return err
+	}
+	if n, _ := res.RowsAffected(); n == 1 {
+		return nil
+	}
+	var current sql.NullString
+	err = s.db.QueryRow(`SELECT unclaim_token FROM exploration_nodes WHERE id=$1 AND exploration_id=$2 AND kind='intent'`, id, s.expID).Scan(&current)
+	if err != nil && err != sql.ErrNoRows {
+		return err
+	}
+	if current.Valid && current.String == token {
+		return fmt.Errorf("%w: intent %d is no longer in the running state", ErrIntentStateConflict, id)
+	}
+	return fmt.Errorf("%w: result write for intent %d rejected", ErrIntentClaimLost, id)
 }
 
 // Stats returns node counts grouped by kind (for dashboard).

db/intent_claim_test.go

@@ -0,0 +1,398 @@
+package db
+
+import (
+	"database/sql"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+// openClaimInstances opens the test database twice — two application instances
+// sharing one database — and returns each instance's store for a fresh
+// exploration.
+func openClaimInstances(t *testing.T) (*DB, *ExplorationStore, *ExplorationStore) {
+	t.Helper()
+	dsn := testDSN(t)
+	a, err := Open(dsn)
+	if err != nil {
+		t.Skipf("postgres unavailable (%v) - skipping", err)
+	}
+	t.Cleanup(func() { a.Close() })
+	b, err := Open(dsn)
+	if err != nil {
+		t.Fatalf("open second instance: %v", err)
+	}
+	t.Cleanup(func() { b.Close() })
+	expID, err := a.CreateExploration("intent claim lease", "heartbeat, expiry and fencing")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { a.Exec(`DELETE FROM explorations WHERE id=$1`, expID) })
+	return a, a.Exploration(expID), b.Exploration(expID)
+}
+
+func addClaimIntent(t *testing.T, store *ExplorationStore) int64 {
+	t.Helper()
+	id, err := store.AddIntent(map[string]any{"summary": "leased"}, 1, nil, "planner")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return id
+}
+
+func claimIntentToken(t *testing.T, store *ExplorationStore, id int64, owner string) string {
+	t.Helper()
+	token, err := store.ClaimIntentToken(id, owner)
+	if err != nil || token == "" {
+		t.Fatalf("claim by %s: token=%q err=%v", owner, token, err)
+	}
+	return token
+}
+
+// ageClaimHeartbeat moves the intent's last heartbeat age into the past, on the
+// database clock the expiry check uses.
+func ageClaimHeartbeat(t *testing.T, d *DB, id int64, age time.Duration) {
+	t.Helper()
+	if _, err := d.Exec(`UPDATE exploration_nodes SET last_heartbeat_at = now() - $2 * INTERVAL '1 second' WHERE id=$1`, id, age.Seconds()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+type claimRow struct {
+	state         string
+	token         sql.NullString
+	blockedReason sql.NullString
+	heartbeatAge  sql.NullFloat64 // seconds since last_heartbeat_at; NULL when never set
+}
+
+func readClaimRow(t *testing.T, d *DB, id int64) claimRow {
+	t.Helper()
+	var row claimRow
+	if err := d.QueryRow(`SELECT state, unclaim_token, blocked_reason, EXTRACT(EPOCH FROM now() - last_heartbeat_at)::float8
+FROM exploration_nodes WHERE id=$1`, id).Scan(&row.state, &row.token, &row.blockedReason, &row.heartbeatAge); err != nil {
+		t.Fatal(err)
+	}
+	return row
+}
+
+func mustResetRunning(t *testing.T, store *ExplorationStore, want int64) {
+	t.Helper()
+	n, err := store.ResetRunningIntents()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != want {
+		t.Fatalf("ResetRunningIntents reclaimed %d intents, want %d", n, want)
+	}
+}
+
+func TestIntentClaimLiveClaimIsNotReclaimed(t *testing.T) {
+	d, a, b := openClaimInstances(t)
+	id := addClaimIntent(t, a)
+	token := claimIntentToken(t, a, id, "work#1")
+
+	row := readClaimRow(t, d, id)
+	if row.state != "running" || row.token.String != token || !row.heartbeatAge.Valid {
+		t.Fatalf("claim did not start a lease: %+v", row)
+	}
+
+	// Neither the claiming instance nor a second instance on the same database
+	// may take a live claim over, by reset or by a fresh claim.
+	mustResetRunning(t, a, 0)
+	mustResetRunning(t, b, 0)
+	if other, err := b.ClaimIntentToken(id, "work#2"); err != nil || other != "" {
+		t.Fatalf("second instance claimed a live intent: token=%q err=%v", other, err)
+	}
+
+	// Still live just inside the TTL.
+	ageClaimHeartbeat(t, d, id, IntentClaimTTL-10*time.Second)
+	mustResetRunning(t, b, 0)
+
+	row = readClaimRow(t, d, id)
+	if row.state != "running" || row.token.String != token {
+		t.Fatalf("live claim changed: %+v, want running with token %q", row, token)
+	}
+	if err := a.HeartbeatIntentClaim(id, token); err != nil {
+		t.Fatalf("heartbeat of the live claimer: %v", err)
+	}
+}
+
+func TestIntentClaimExpiredClaimIsReclaimed(t *testing.T) {
+	d, a, b := openClaimInstances(t)
+	id := addClaimIntent(t, a)
+	old := claimIntentToken(t, a, id, "work#1")
+
+	ageClaimHeartbeat(t, d, id, IntentClaimTTL+time.Second)
+	mustResetRunning(t, b, 1)
+
+	row := readClaimRow(t, d, id)
+	if row.state != "open" || row.token.Valid {
+		t.Fatalf("expired claim not released: %+v, want open with no token", row)
+	}
+	fresh := claimIntentToken(t, b, id, "work#2")
+	if fresh == old {
+		t.Fatal("re-claim reused the previous unclaim token")
+	}
+	mustResetRunning(t, a, 0) // the new claim is live
+}
+
+// A row claimed before the lease columns existed has neither a heartbeat nor a
+// token; SetIntentState produces exactly that shape. It counts as expired.
+func TestIntentClaimLegacyRunningRowIsReclaimed(t *testing.T) {
+	d, a, b := openClaimInstances(t)
+	id := addClaimIntent(t, a)
+	if err := a.SetIntentState(id, "running"); err != nil {
+		t.Fatal(err)
+	}
+	row := readClaimRow(t, d, id)
+	if row.heartbeatAge.Valid || row.token.Valid {
+		t.Fatalf("legacy fixture carries lease data: %+v", row)
+	}
+
+	mustResetRunning(t, b, 1)
+	if row = readClaimRow(t, d, id); row.state != "open" {
+		t.Fatalf("legacy running row state=%q, want open", row.state)
+	}
+	claimIntentToken(t, b, id, "work#1")
+}
+
+func TestIntentClaimHeartbeatExtendsLease(t *testing.T) {
+	d, a, b := openClaimInstances(t)
+	id := addClaimIntent(t, a)
+	token := claimIntentToken(t, a, id, "work#1")
+
+	// Past the TTL but not yet taken over: the claimer's heartbeat renews it.
+	ageClaimHeartbeat(t, d, id, IntentClaimTTL+time.Second)
+	if err := a.HeartbeatIntentClaim(id, token); err != nil {
+		t.Fatalf("heartbeat: %v", err)
+	}
+	mustResetRunning(t, b, 0)
+
+	row := readClaimRow(t, d, id)
+	if row.state != "running" || row.token.String != token {
+		t.Fatalf("heartbeat did not keep the claim: %+v", row)
+	}
+	if !row.heartbeatAge.Valid || row.heartbeatAge.Float64 > 30 {
+		t.Fatalf("last_heartbeat_at not renewed: age=%+v", row.heartbeatAge)
+	}
+
+	// Without a further heartbeat the same claim does expire.
+	ageClaimHeartbeat(t, d, id, IntentClaimTTL+time.Second)
+	mustResetRunning(t, b, 1)
+}
+
+func TestIntentClaimStaleWorkerIsFencedAfterReclaim(t *testing.T) {
+	d, a, b := openClaimInstances(t)
+	id := addClaimIntent(t, a)
+	stale := claimIntentToken(t, a, id, "work#1")
+
+	ageClaimHeartbeat(t, d, id, IntentClaimTTL+time.Second)
+	mustResetRunning(t, b, 1)
+
+	assertFenced := func(when string) {
+		t.Helper()
+		before := readClaimRow(t, d, id)
+		if err := a.HeartbeatIntentClaim(id, stale); !errors.Is(err, ErrIntentClaimLost) {
+			t.Fatalf("%s: stale heartbeat err=%v, want ErrIntentClaimLost", when, err)
+		}
+		if err := a.SettleIntentClaim(id, stale, "done"); !errors.Is(err, ErrIntentClaimLost) {
+			t.Fatalf("%s: stale result write err=%v, want ErrIntentClaimLost", when, err)
+		}
+		if err := a.SettleIntentClaimBlocked(id, stale, IntentBlockedLLMQuota); !errors.Is(err, ErrIntentClaimLost) {
+			t.Fatalf("%s: stale blocked write err=%v, want ErrIntentClaimLost", when, err)
+		}
+		after := readClaimRow(t, d, id)
+		if after.state != before.state || after.token != before.token || after.blockedReason != before.blockedReason {
+			t.Fatalf("%s: a rejected stale write changed the intent: %+v -> %+v", when, before, after)
+		}
+	}
+	assertFenced("after takeover, before the re-claim")
+
+	current := claimIntentToken(t, b, id, "work#2")
+	ageClaimHeartbeat(t, d, id, time.Minute)
+	assertFenced("after the re-claim")
+	if row := readClaimRow(t, d, id); row.state != "running" || row.token.String != current || row.heartbeatAge.Float64 < 50 {
+		t.Fatalf("stale worker touched the new claim: %+v", row)
+	}
+
+	// The current claimer is unaffected and its result write releases the claim.
+	if err := b.HeartbeatIntentClaim(id, current); err != nil {
+		t.Fatalf("current claimer heartbeat: %v", err)
+	}
+	if err := b.SettleIntentClaim(id, current, "done"); err != nil {
+		t.Fatalf("current claimer result write: %v", err)
+	}
+	if row := readClaimRow(t, d, id); row.state != "done" || row.token.Valid {
+		t.Fatalf("settled intent: %+v, want done with no token", row)
+	}
+	if err := b.SettleIntentClaim(id, current, "blocked"); !errors.Is(err, ErrIntentClaimLost) {
+		t.Fatalf("second settle with a released token err=%v, want ErrIntentClaimLost", err)
+	}
+}
+
+func TestSettleIntentClaimBlockedAndStateConflict(t *testing.T) {
+	d, a, _ := openClaimInstances(t)
+
+	blocked := addClaimIntent(t, a)
+	token := claimIntentToken(t, a, blocked, "work#1")
+	if err := a.SettleIntentClaimBlocked(blocked, token, IntentBlockedLLMQuota); err != nil {
+		t.Fatal(err)
+	}
+	if row := readClaimRow(t, d, blocked); row.state != "blocked" || row.blockedReason.String != IntentBlockedLLMQuota || row.token.Valid {
+		t.Fatalf("blocked settle: %+v", row)
+	}
+
+	// The token is still current but the intent left 'running' through an
+	// unfenced path: that is a state conflict, not a lost claim.
+	moved := addClaimIntent(t, a)
+	token = claimIntentToken(t, a, moved, "work#1")
+	if err := a.SetIntentState(moved, "paused"); err != nil {
+		t.Fatal(err)
+	}
+	err := a.SettleIntentClaim(moved, token, "done")
+	if !errors.Is(err, ErrIntentStateConflict) || errors.Is(err, ErrIntentClaimLost) {
+		t.Fatalf("settle of a non-running intent err=%v, want ErrIntentStateConflict", err)
+	}
+	if row := readClaimRow(t, d, moved); row.state != "paused" {
+		t.Fatalf("rejected settle changed the state to %q", row.state)
+	}
+}
+
+func TestClaimPausedIntentTokenStartsLease(t *testing.T) {
+	d, a, b := openClaimInstances(t)
+	id := addClaimIntent(t, a)
+	first := claimIntentToken(t, a, id, "work#1")
+	if err := a.SettleIntentClaim(id, first, "paused"); err != nil {
+		t.Fatal(err)
+	}
+	ageClaimHeartbeat(t, d, id, IntentClaimTTL+time.Second) // the paused intent's old heartbeat is stale
+
+	token, err := a.ClaimPausedIntentToken(id)
+	if err != nil || token == "" || token == first {
+		t.Fatalf("claim paused: token=%q first=%q err=%v", token, first, err)
+	}
+	if again, err := b.ClaimPausedIntentToken(id); err != nil || again != "" {
+		t.Fatalf("second paused claim: token=%q err=%v, want none", again, err)
+	}
+	mustResetRunning(t, b, 0)
+	if row := readClaimRow(t, d, id); row.state != "running" || row.token.String != token {
+		t.Fatalf("paused claim: %+v", row)
+	}
+	if err := a.SettleIntentClaim(id, first, "done"); !errors.Is(err, ErrIntentClaimLost) {
+		t.Fatalf("previous run's token err=%v, want ErrIntentClaimLost", err)
+	}
+}
+
+// A heartbeat racing a takeover of the same expired claim must have exactly one
+// winner: either the claim is renewed and stays, or it is taken over and the
+// heartbeat is rejected. Both succeeding would be a stolen live claim.
+func TestIntentClaimTakeoverRacesHeartbeatAtomically(t *testing.T) {
+	d, a, b := openClaimInstances(t)
+	id := addClaimIntent(t, a)
+	token := claimIntentToken(t, a, id, "work#1")
+
+	var renewed, takenOver int
+	for round := 0; round < 40; round++ {
+		ageClaimHeartbeat(t, d, id, IntentClaimTTL+time.Second)
+		var heartbeatErr, resetErr error
+		var reclaimed int64
+		var wg sync.WaitGroup
+		start := make(chan struct{})
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			<-start
+			heartbeatErr = a.HeartbeatIntentClaim(id, token)
+		}()
+		go func() {
+			defer wg.Done()
+			<-start
+			reclaimed, resetErr = b.ResetRunningIntents()
+		}()
+		close(start)
+		wg.Wait()
+		if resetErr != nil {
+			t.Fatal(resetErr)
+		}
+		row := readClaimRow(t, d, id)
+		switch {
+		case heartbeatErr == nil && reclaimed == 0:
+			renewed++
+			if row.state != "running" || row.token.String != token {
+				t.Fatalf("round %d: renewed claim is %+v", round, row)
+			}
+		case errors.Is(heartbeatErr, ErrIntentClaimLost) && reclaimed == 1:
+			takenOver++
+			if row.state != "open" || row.token.Valid {
+				t.Fatalf("round %d: taken-over claim is %+v", round, row)
+			}
+			token = claimIntentToken(t, a, id, "work#1")
+		default:
+			t.Fatalf("round %d: heartbeat err=%v with %d reclaimed, want exactly one winner", round, heartbeatErr, reclaimed)
+		}
+	}
+	t.Logf("heartbeat won %d rounds, takeover won %d rounds", renewed, takenOver)
+}
+
+// The lease columns reach existing installs through schema.sql, which runs on
+// every startup. This replays it on a table stripped back to its pre-upgrade
+// shape, inside a transaction that is always rolled back.
+func TestIntentClaimColumnsMigrateIdempotently(t *testing.T) {
+	d, err := Open(testDSN(t))
+	if err != nil {
+		t.Skipf("postgres unavailable (%v) - skipping", err)
+	}
+	defer d.Close()
+
+	tx, err := d.Begin()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tx.Rollback() //nolint:errcheck // the test never commits
+
+	if _, err := tx.Exec(`ALTER TABLE exploration_nodes DROP COLUMN IF EXISTS last_heartbeat_at, DROP COLUMN IF EXISTS unclaim_token`); err != nil {
+		t.Fatalf("simulate old table: %v", err)
+	}
+	var legacy int64
+	if err := tx.QueryRow(`
+WITH exp AS (INSERT INTO explorations(description, goal) VALUES ('legacy', 'pre-lease install') RETURNING id)
+INSERT INTO exploration_nodes(exploration_id, kind, state, owner)
+SELECT id, 'intent', 'running', 'work#1' FROM exp RETURNING id`).Scan(&legacy); err != nil {
+		t.Fatalf("insert legacy running intent: %v", err)
+	}
+
+	for run := 1; run <= 2; run++ {
+		if _, err := tx.Exec(schemaSQL); err != nil {
+			t.Fatalf("schema run %d: %v", run, err)
+		}
+		for col, typ := range map[string]string{"last_heartbeat_at": "timestamp with time zone", "unclaim_token": "text"} {
+			var dataType, nullable string
+			if err := tx.QueryRow(`SELECT data_type, is_nullable FROM information_schema.columns
+WHERE table_schema=current_schema() AND table_name='exploration_nodes' AND column_name=$1`, col).Scan(&dataType, &nullable); err != nil {
+				t.Fatalf("schema run %d: column %s: %v", run, col, err)
+			}
+			if dataType != typ || nullable != "YES" {
+				t.Fatalf("schema run %d: column %s is %s nullable=%s, want %s nullable", run, col, dataType, nullable, typ)
+			}
+		}
+		var checks int
+		if err := tx.QueryRow(`SELECT count(*) FROM pg_constraint
+WHERE conrelid='exploration_nodes'::regclass AND contype='c'
+  AND (pg_get_constraintdef(oid) LIKE '%last_heartbeat_at%' OR pg_get_constraintdef(oid) LIKE '%unclaim_token%')`).Scan(&checks); err != nil {
+			t.Fatal(err)
+		}
+		if checks != 0 {
+			t.Fatalf("schema run %d: %d CHECK constraints on the lease columns, want none", run, checks)
+		}
+		var state string
+		var heartbeat sql.NullTime
+		var token sql.NullString
+		if err := tx.QueryRow(`SELECT state, last_heartbeat_at, unclaim_token FROM exploration_nodes WHERE id=$1`, legacy).Scan(&state, &heartbeat, &token); err != nil {
+			t.Fatal(err)
+		}
+		if state != "running" || heartbeat.Valid || token.Valid {
+			t.Fatalf("schema run %d: legacy row state=%q heartbeat=%v token=%v, want running with NULL lease columns", run, state, heartbeat, token)
+		}
+	}
+}

db/schema.sql

@@ -201,6 +201,13 @@ ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS delete_reason TEXT;
 -- a node most recently went from "has a live downstream branch" to none (NULL = hot).
 ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS content_version  INT    NOT NULL DEFAULT 0;
 ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS cold_since_round BIGINT;
+-- Intent claim lease: unclaim_token is the fencing token of the current claim (fresh
+-- per claim, cleared when the claim is settled or taken over); last_heartbeat_at is
+-- the claimer's latest heartbeat. A running intent whose heartbeat is older than
+-- db.IntentClaimTTL may be taken over. Both are NULL on rows claimed before these
+-- columns existed, which counts as expired.
+ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS last_heartbeat_at TIMESTAMPTZ;
+ALTER TABLE exploration_nodes ADD COLUMN IF NOT EXISTS unclaim_token     TEXT;
 -- ck_node_kind: existing installs predate the 'digest' kind — recreate to allow it.
 DO $$
 BEGIN

server/claim_heartbeat_test.go

@@ -0,0 +1,102 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/Autumn-27/artex/agent"
+	"github.com/Autumn-27/artex/db"
+)
+
+// The worker-side heartbeat must keep a claim alive while it runs and, once
+// another worker has taken the intent over, cancel the stale run with the named
+// claim-lost cause instead of carrying on.
+func TestClaimHeartbeatRenewsThenCancelsOnTakeover(t *testing.T) {
+	skipWithoutDSN(t)
+	dsn, _, err := db.DSN()
+	if err != nil {
+		t.Fatal(err)
+	}
+	d, err := db.Open(dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	expID, err := d.CreateExploration("claim heartbeat", "renew, then stop on takeover")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Exec(`DELETE FROM explorations WHERE id=$1`, expID)
+	store := d.Exploration(expID)
+	id, err := store.AddIntent(map[string]any{"summary": "leased"}, 1, nil, "planner")
+	if err != nil {
+		t.Fatal(err)
+	}
+	token, err := store.ClaimIntentToken(id, "work#1")
+	if err != nil || token == "" {
+		t.Fatalf("claim: token=%q err=%v", token, err)
+	}
+	expire := func() {
+		t.Helper()
+		if _, err := d.Exec(`UPDATE exploration_nodes SET last_heartbeat_at = now() - $2 * INTERVAL '1 second' WHERE id=$1`,
+			id, (db.IntentClaimTTL + time.Second).Seconds()); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+	stop := startClaimHeartbeat(store, 20*time.Millisecond, "work#1", id, token, cancel)
+	defer stop()
+
+	// Renewal: an expired-looking heartbeat is refreshed by the next ticks.
+	expire()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		var fresh bool
+		if err := d.QueryRow(`SELECT last_heartbeat_at > now() - INTERVAL '30 seconds' FROM exploration_nodes WHERE id=$1`, id).Scan(&fresh); err != nil {
+			t.Fatal(err)
+		}
+		if fresh {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("heartbeat did not renew the claim")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if ctx.Err() != nil {
+		t.Fatalf("run cancelled while its claim was live: %v", context.Cause(ctx))
+	}
+
+	// Takeover: expire the claim and reclaim it before the next heartbeat lands
+	// (retried, because a heartbeat may win the race and renew the claim first).
+	var reclaimed int64
+	for attempt := 0; attempt < 200 && reclaimed == 0; attempt++ {
+		expire()
+		if reclaimed, err = store.ResetRunningIntents(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if reclaimed != 1 {
+		t.Fatalf("could not take the claim over: reclaimed=%d", reclaimed)
+	}
+	other, err := store.ClaimIntentToken(id, "work#2")
+	if err != nil || other == "" {
+		t.Fatalf("re-claim: token=%q err=%v", other, err)
+	}
+
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("stale run was not cancelled after the takeover")
+	}
+	if cause := context.Cause(ctx); !errors.Is(cause, agent.AbortWorkClaimLost) {
+		t.Fatalf("cancel cause=%v, want AbortWorkClaimLost", cause)
+	}
+	if err := store.HeartbeatIntentClaim(id, other); err != nil {
+		t.Fatalf("new claimer lost its claim: %v", err)
+	}
+}

server/engine.go

@@ -480,17 +480,6 @@ func (e *Engine) releaseWorkControl(intentID int64, run *workExecution, action s
 	e.workMu.Unlock()
 }
 
-func transitionIntentState(store *db.ExplorationStore, intentID int64, expected, state string) error {
-	changed, err := store.CompareAndSetIntentState(intentID, expected, state)
-	if err != nil {
-		return err
-	}
-	if !changed {
-		return fmt.Errorf("%w: intent %d is no longer in the %s state", db.ErrIntentStateConflict, intentID, expected)
-	}
-	return nil
-}
-
 // SteerWork queues a mid-run course-correction for the work running intentID (the
 // planner's steer_work tool). The worker delivers it before its next tool call and
 // re-plans — no kill. Errors if no work is currently running that intent.
@@ -761,10 +750,11 @@ func (e *Engine) Run(ctx context.Context, t *Task) {
 		t.Notify() // already running — just nudge a planning round
 		return
 	}
-	rt := e.registerTaskRoutines(ctx, t.ID, 1+workers)
+	rt := e.registerTaskRoutines(ctx, t.ID, 2+workers)
 	e.deleteMu.RUnlock()
 	e.touch(t.ID)
 	runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) })
+	runTaskRoutine(rt, func(loopCtx context.Context) { e.claimSweepLoop(loopCtx, t) })
 	for i := 0; i < workers; i++ {
 		name := fmt.Sprintf("work#%d", i+1)
 		runTaskRoutine(rt, func(loopCtx context.Context) { e.workerLoop(loopCtx, t, name) })
@@ -950,6 +940,64 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
 	}
 }
 
+// claimSweepLoop periodically returns this task's expired intent claims (no
+// heartbeat within db.IntentClaimTTL: the claiming worker crashed or its process
+// is gone) to the frontier, where the worker loops re-claim them. The startup
+// reset alone would miss a claim that was still live when this process started.
+func (e *Engine) claimSweepLoop(ctx context.Context, t *Task) {
+	ticker := time.NewTicker(db.IntentClaimHeartbeatInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+		if !e.beginTaskOperation(t.ID) {
+			return
+		}
+		n, err := t.Store.ResetRunningIntents()
+		e.decInflight(t.ID)
+		if err != nil {
+			log.Printf("[engine] task %s expired-claim sweep failed: %v", t.ID, err)
+		} else if n > 0 {
+			log.Printf("[engine] task %s returned %d expired intent claims to open", t.ID, n)
+		}
+	}
+}
+
+// startClaimHeartbeat renews an intent claim every interval while its worker is
+// busy. A heartbeat rejected with db.ErrIntentClaimLost means the claim expired
+// and another worker took the intent over: the run is cancelled with
+// AbortWorkClaimLost so the stale worker stops. Any other store error is logged
+// and retried on the next tick; the claim survives until db.IntentClaimTTL
+// passes without a heartbeat. stop blocks until the heartbeat has ended.
+func startClaimHeartbeat(store *db.ExplorationStore, interval time.Duration, label string, intentID int64, token string, cancel context.CancelCauseFunc) (stop func()) {
+	quit, done := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(done)
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-quit:
+				return
+			case <-ticker.C:
+			}
+			err := store.HeartbeatIntentClaim(intentID, token)
+			if errors.Is(err, db.ErrIntentClaimLost) {
+				log.Printf("[worker %s] intent #%d heartbeat rejected, cancelling this run: %v", label, intentID, err)
+				cancel(agent.AbortWorkClaimLost)
+				return
+			}
+			if err != nil {
+				log.Printf("[worker %s] intent #%d heartbeat failed, retrying: %v", label, intentID, err)
+			}
+		}
+	}()
+	return func() { close(quit); <-done }
+}
+
 func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) {
 	for {
 		select {
@@ -1000,12 +1048,12 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) {
 // runIntent. Returns false when nothing was claimable. The pool worker loop is its
 // only caller.
 func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker *agent.Worker) bool {
-	intent := e.claimNext(t, name)
+	intent, token := e.claimNext(t, name)
 	if intent == nil {
 		return false
 	}
 	log.Printf("[worker %s] task %s claimed intent #%d", name, t.ID, intent.ID)
-	return e.runIntent(ctx, t, name, worker, intent, "", "")
+	return e.runIntent(ctx, t, name, worker, intent, token, "", "")
 }
 
 // runIntent executes and fully settles one already-claimed (state=running) intent.
@@ -1016,7 +1064,10 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker
 // transcript marker that dedups re-injection across model_error retries. The caller
 // must already hold one task-operation admission for the whole sequence so a delete
 // cannot observe quiescence between the LLM return and the final DB writes.
-func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *agent.Worker, intent *db.Node, requestID, message string) bool {
+// token is the unclaim token of the caller's claim: a heartbeat renews the claim
+// while the worker is busy, and every intent state write below is fenced on it,
+// so a worker whose claim was taken over cannot overwrite the new claimer.
+func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *agent.Worker, intent *db.Node, token, requestID, message string) bool {
 	// Per-intent LLM rotation (#85): each claimed worker intent runs on the
 	// next chain head. The pin travels through the run context into the
 	// provider resolver; non-worker roles and failover keep the active head.
@@ -1031,7 +1082,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 	emit := func(r db.Activity) { e.emitActivity(t, r) }
 	ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; carries the task deadline
 	if ectx.Err() != nil || e.IsDeleting(t.ID) {
-		if err := transitionIntentState(t.Store, intent.ID, "running", "open"); err != nil {
+		if err := t.Store.SettleIntentClaim(intent.ID, token, "open"); err != nil {
 			log.Printf("[worker %s] task %s intent #%d claim rollback failed: %v", name, t.ID, intent.ID, err)
 		}
 		return true
@@ -1039,6 +1090,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 	// per-work child context so the planner's kill_work can stop just this work.
 	workCtx, workCancel := context.WithCancelCause(ectx)
 	e.registerWork(intent.ID, workCancel)
+	stopHeartbeat := startClaimHeartbeat(t.Store, db.IntentClaimHeartbeatInterval, name, intent.ID, token, workCancel)
 	// wrap the guard hooks so steer_work can inject a mid-run course-correction
 	// for THIS intent (drained before the worker's next tool call).
 	iid := intent.ID
@@ -1089,6 +1141,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 		}
 		e.EndLLMCall(t.ID)
 	}
+	stopHeartbeat() // no longer busy: the state writes below follow within the just-renewed claim
 	// Capture kill state before detachWork cancels workCtx. kill = this work's
 	// ctx was cancelled (planner kill_work) while the TASK ctx kept running; a
 	// pause cancels the task ctx (ectx) instead. Checking workCtx.Err() AFTER
@@ -1111,7 +1164,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 	var controlErr error
 	defer func() { completeWork(controlErr) }()
 	if action == "pause" {
-		controlErr = transitionIntentState(t.Store, intent.ID, "running", "paused")
+		controlErr = t.Store.SettleIntentClaim(intent.ID, token, "paused")
 		if controlErr != nil {
 			log.Printf("[worker %s] task %s intent #%d failed to persist the paused state: %v", name, t.ID, intent.ID, controlErr)
 			return true
@@ -1124,7 +1177,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 		// Park the stopped run in paused before handing cleanup to the API. If the
 		// request disconnects after cancellation, the intent remains recoverable and
 		// a later cancel can finish cleanup instead of leaving a phantom running row.
-		controlErr = transitionIntentState(t.Store, intent.ID, "running", "paused")
+		controlErr = t.Store.SettleIntentClaim(intent.ID, token, "paused")
 		if controlErr != nil {
 			log.Printf("[worker %s] task %s intent #%d failed to persist the cancel barrier: %v", name, t.ID, intent.ID, controlErr)
 			return true
@@ -1133,11 +1186,18 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 		e.touch(t.ID)
 		return true
 	}
+	// The heartbeat was rejected: the claim expired and another worker took the
+	// intent over. This run's result is void and the intent state is not ours to
+	// write (the fence would reject every write below).
+	if errors.Is(workCause, agent.AbortWorkClaimLost) {
+		log.Printf("[worker %s] task %s intent #%d claim lost to another worker, result discarded", name, t.ID, intent.ID)
+		return true
+	}
 	// if a pause cancelled this run mid-flight, return the intent to the frontier
 	// so it is re-claimed on resume — the worker will resume the prior LLM
 	// conversation from its transcript instead of restarting from scratch.
 	if ectx.Err() != nil && taskExecutionPaused(context.Cause(ectx)) {
-		if err := transitionIntentState(t.Store, intent.ID, "running", "open"); err != nil {
+		if err := t.Store.SettleIntentClaim(intent.ID, token, "open"); err != nil {
 			log.Printf("[worker %s] task %s intent #%d task-pause rollback failed: %v", name, t.ID, intent.ID, err)
 		}
 		return true
@@ -1145,7 +1205,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 	// The hard-fallback cancel of task-timeout settlement (not pause, not kill) cancelled
 	// this run → classify it as exhausted (settled), not mislabeled blocked. By then the worker has usually written its result back during the settlement phase.
 	if ectx.Err() != nil && e.isSettling(t.ID) {
-		if err := transitionIntentState(t.Store, intent.ID, "running", "exhausted"); err != nil {
+		if err := t.Store.SettleIntentClaim(intent.ID, token, "exhausted"); err != nil {
 			log.Printf("[worker %s] task %s intent #%d failed to persist the timeout-settlement state: %v", name, t.ID, intent.ID, err)
 		}
 		log.Printf("[worker %s] task %s intent #%d ended by task-timeout settlement (exhausted), wrote back %s", name, t.ID, intent.ID, wrote)
@@ -1155,7 +1215,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 	// Task judged complete (done via the regular path) → cancelExec above cancelled this
 	// run. The intent's result is meaningless now; mark it stopped (not blocked) and do not pollute the intent states of a completed task.
 	if ectx.Err() != nil && isTerminalStatus(e.m.TaskStatus(t.ID)) {
-		if err := transitionIntentState(t.Store, intent.ID, "running", "stopped"); err != nil {
+		if err := t.Store.SettleIntentClaim(intent.ID, token, "stopped"); err != nil {
 			log.Printf("[worker %s] task %s intent #%d failed to persist the terminal stop: %v", name, t.ID, intent.ID, err)
 		}
 		log.Printf("[worker %s] task %s intent #%d cancelled because the task completed (stopped)", name, t.ID, intent.ID)
@@ -1164,7 +1224,7 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 	}
 	// killed by the planner: mark stopped (don't write back results, don't auto-reclaim).
 	if killed {
-		if err := transitionIntentState(t.Store, intent.ID, "running", "stopped"); err != nil {
+		if err := t.Store.SettleIntentClaim(intent.ID, token, "stopped"); err != nil {
 			log.Printf("[worker %s] task %s intent #%d failed to persist the planner stop: %v", name, t.ID, intent.ID, err)
 		}
 		log.Printf("[worker %s] task %s intent #%d killed (stopped)", name, t.ID, intent.ID)
@@ -1188,11 +1248,16 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
 		state = "exhausted"
 		log.Printf("[worker %s] intent %d ran out of time (exhausted), wrote back %s after settling", name, intent.ID, wrote)
 	}
+	var stateErr error
 	if state == "blocked" && isTaskLLMChainExhausted(err) {
-		_ = t.Store.SetIntentBlockedReason(intent.ID, db.IntentBlockedLLMQuota)
+		stateErr = t.Store.SettleIntentClaimBlocked(intent.ID, token, db.IntentBlockedLLMQuota)
 	} else {
-		if stateErr := transitionIntentState(t.Store, intent.ID, "running", state); stateErr != nil {
-			log.Printf("[worker %s] task %s intent #%d failed to persist terminal state %s: %v", name, t.ID, intent.ID, state, stateErr)
+		stateErr = t.Store.SettleIntentClaim(intent.ID, token, state)
+	}
+	if stateErr != nil {
+		log.Printf("[worker %s] task %s intent #%d failed to persist terminal state %s: %v", name, t.ID, intent.ID, state, stateErr)
+		if errors.Is(stateErr, db.ErrIntentClaimLost) {
+			return true // taken over: the intent is not finished, its new claimer reports the result
 		}
 	}
 	log.Printf("[worker %s] task %s intent #%d finished: %s (wrote back %s)", name, t.ID, intent.ID, state, wrote)
@@ -1234,11 +1299,11 @@ func (e *Engine) runDetachedIntent(ctx context.Context, t *Task, intentID int64,
 	if node == nil || node.Kind != db.KindIntent {
 		return fmt.Errorf("intent not found")
 	}
-	changed, err := t.Store.CompareAndSetIntentState(intentID, "paused", "running")
+	token, err := t.Store.ClaimPausedIntentToken(intentID)
 	if err != nil {
 		return err
 	}
-	if !changed {
+	if token == "" {
 		return fmt.Errorf("%w: intent is no longer in the paused state", db.ErrIntentStateConflict)
 	}
 	node.State, node.Owner = "running", "chat"
@@ -1251,7 +1316,7 @@ func (e *Engine) runDetachedIntent(ctx context.Context, t *Task, intentID int64,
 	release = false // ownership of the admission passes to the goroutine
 	go func() {
 		defer e.decInflight(t.ID)
-		e.runIntent(ctx, t, "chat", worker, node, requestID, agentMessage)
+		e.runIntent(ctx, t, "chat", worker, node, token, requestID, agentMessage)
 	}()
 	return nil
 }
@@ -1279,12 +1344,12 @@ func sleepCtx(ctx context.Context, d time.Duration) (done bool) {
 	}
 }
 
-func (e *Engine) claimNext(t *Task, name string) *db.Node {
+func (e *Engine) claimNext(t *Task, name string) (*db.Node, string) {
 	fr, _ := t.Store.Frontier(20)
 	for _, in := range fr {
-		if ok, _ := t.Store.ClaimIntent(in.ID, name); ok {
-			return in
+		if token, _ := t.Store.ClaimIntentToken(in.ID, name); token != "" {
+			return in, token
 		}
 	}
-	return nil
+	return nil, ""
 }