fix(groups): cadence claim had no expiry; a NATS publish failure could drop a turn

docs/changelog.md

@@ -3,6 +3,26 @@
 > **Purpose:** Living document tracking all changes, decisions, and reasoning during implementation. Updated as work progresses for easy reference and review.
 
 
+---
+
+## 🔎 fix(groups): cadence claim had no expiry; NATS publish could drop a turn (2026-08-09)
+
+**Repo:** EDDI (`fix/team-cadence-claim-lease`)
+
+1. **A crashed pod wedged the cadence permanently.** `TeamCadenceService`'s Javadoc promised the run protocol was "crash-proof by construction: a pod crash mid-discussion loses nothing, because the next fire finds the terminal state and reconciles it". That only holds if something moves the discussion to a terminal state, and on a pod crash nothing does — no startup sweep touches an IN_PROGRESS `GroupConversation`, and `HitlCrashRecoveryObserver` handles only the AWAITING_* states. `reconcile` read IN_PROGRESS, answered "still running", and did so on every subsequent fire forever; `casRunningDiscussion` was never released. The cadence stayed wedged until a human cancelled the discussion by hand. A non-terminal, non-paused discussion whose progress heartbeat has not advanced within `eddi.groups.cadence.abandoned-run-lease` (default 6h) is now cancelled, its tasks returned to the backlog, and the claim released.
+   - **The lease is measured on `gc.getLastModified()`, not on claim age.** A claim-age lease cannot tell a dead pod from a healthy long-running discussion, and reclaiming a live one would orphan its outcomes *and* double-schedule its tasks — recreating, by design, the bug fixed in #2. The discussion loop persists the conversation at every phase boundary, so `lastModified` is a free liveness heartbeat.
+   - **AWAITING_* never expires, at any age.** A discussion may legitimately wait on a human for days, and every surface that resolves one (resume, the cross-pod cancel CAS, the timeout policies re-armed at startup) works cross-pod — so a paused discussion on a dead pod still progresses. `CREATED` *is* reclaimable: it is the crash window between the claim CAS and the executor starting.
+2. **A transient read error released the claim.** `reconcile` caught every exception from `conversationStore.read` and treated it as "discussion gone", returning the pulled tasks to PENDING — so a read failure while the discussion was genuinely running let the next fire start a **second** discussion on the same backlog. Narrowed to `ResourceNotFoundException` (provable absence), which is the distinction `AgentDeploymentManagement.isAgentConfigMissing` already makes one package over; anything else skips the fire and keeps the claim.
+3. **A NATS publish failure could silently drop a conversation turn.** `publishAndExecute` caught only `IOException | JetStreamApiException`, so any unchecked failure — the NATS client throws `IllegalStateException` on a closed or draining connection — escaped the method and skipped the `submitCallable` below it entirely. The turn was dropped with no execution, no callback and no dead-letter, while that very catch block promised "executing locally". The publish is an ordering marker, never the work, so no publish failure may cost a turn.
+
+**Two invariants pinned that previously rested on prose:**
+- `NatsCoordinatorInProcessInvariantTest` asserts the NATS coordinator hands the callable to the local `IRuntime` rather than serializing it onto the stream. `LiveDiscussionRegistry` is per-node, and if a member turn could ever run off-node the group task/artifact/recruit tools would not fail — they would silently not be assembled. That invariant was documented as verified in this changelog; it is now enforced.
+- `StoreIfFieldEqualsContractTest` pins that neither backend inherits the throwing default and that both declare both CAS outcomes. Every CAS-based safety property in `executeDiscussion` and the cadence claim protocol is implemented once over `IResourceStorage.storeIfFieldEquals`, so it is exactly as good as that method is on the active backend. (Reviewed in full: the Mongo and Postgres implementations do agree — conditional update, `rows == 0` → existence probe → 404-vs-409 — and the `long` overload correctly degrades to text on Postgres, where `data->>` renders a JSON number canonically, while Mongo needs typed BSON equality. Nothing tested it.)
+
+**Disproved during review, not changed:** `Cadence.maxBacklogTasksPerRun` was flagged as an unvalidated `.limit()` argument. It is not — the record's compact constructor clamps any non-positive value to the default, so `.limit()` can never receive a negative.
+
+Suites: 151 across TeamCadence / NATS / GroupWorkspace, all green, +14 new.
+
 ---
 
 ## 🔎 fix(groups): pre-merge deep review — facilitator HITL bypass, CALL_VOTE guards, metric cardinality, template honesty (2026-08-08)

src/main/java/ai/labs/eddi/engine/runtime/internal/NatsConversationCoordinator.java

@@ -333,6 +333,15 @@ private void publishAndExecute(String conversationId, BlockingQueue<RetryableCal
             });
         } catch (IOException | JetStreamApiException e) {
             log.warnf(e, "Failed to publish to NATS for conversation %s, executing locally", sanitize(conversationId));
+        } catch (RuntimeException e) {
+            // The publish is an ORDERING marker, never the work itself, so no publish
+            // failure may cost a conversation its turn. Catching only the two checked
+            // types left every unchecked one — the NATS client throws
+            // IllegalStateException on a closed or draining connection — to escape
+            // this method, skipping the submitCallable below entirely: the turn was
+            // then silently dropped, with no execution, no callback and no
+            // dead-letter, while this very block promised "executing locally".
+            log.warnf(e, "NATS publish failed unexpectedly for conversation %s, executing locally", sanitize(conversationId));
         }
 
         // Execute the callable via the runtime thread pool

src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java

@@ -12,6 +12,7 @@
 import ai.labs.eddi.configs.groups.model.GroupWorkspace;
 import ai.labs.eddi.configs.groups.model.GroupWorkspace.Cadence;
 import ai.labs.eddi.configs.groups.model.GroupWorkspace.MemberStats;
+import ai.labs.eddi.datastore.IResourceStore;
 import ai.labs.eddi.configs.groups.model.SharedTaskList;
 import ai.labs.eddi.configs.groups.model.SharedTaskList.TaskItem;
 import ai.labs.eddi.configs.groups.model.SharedTaskList.TaskStatus;
@@ -23,13 +24,17 @@
 import jakarta.annotation.PostConstruct;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.jboss.logging.Logger;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Executes team cadences (I13): scheduled pulls from a {@link GroupWorkspace}'s
@@ -55,8 +60,22 @@
  * </ol>
  * Writeback therefore happens on the fire AFTER the discussion ends (or on a
  * workspace read — see the REST layer's read-repair), never from inside the
- * discussion thread: a pod crash mid-discussion loses nothing, because the next
- * fire finds the terminal state and reconciles it.
+ * discussion thread.
+ * <p>
+ * <b>Crash recovery rests on the lease, not on the terminal state.</b> This
+ * used to claim that "a pod crash mid-discussion loses nothing, because the
+ * next fire finds the terminal state and reconciles it". It does not: on a pod
+ * crash nothing moves the discussion to a terminal state — no startup sweep
+ * touches an IN_PROGRESS {@code GroupConversation}, and
+ * {@code HitlCrashRecoveryObserver} handles only the AWAITING_* states — so
+ * {@code reconcile} answered "still running" on every subsequent fire and the
+ * cadence stayed wedged until a human cancelled the discussion by hand. A
+ * non-terminal, non-paused discussion whose own progress heartbeat
+ * ({@code lastModified}) has not advanced within
+ * {@code eddi.groups.cadence.abandoned-run-lease} is now treated as abandoned:
+ * cancelled, its tasks returned to the backlog, and the claim released. See
+ * {@code reclaimIfAbandoned} for why the lease is measured on progress rather
+ * than on claim age, and why AWAITING_* never expires.
  *
  * @author ginccc
  */
@@ -79,32 +98,70 @@ public class TeamCadenceService {
     /** Fallback identity when a cadence predates {@code createdBy}. */
     public static final String FALLBACK_USER_ID = "system:team-cadence";
 
+    /**
+     * States in which a discussion is legitimately waiting on a human and must
+     * never be treated as abandoned, however old the claim is.
+     */
+    private static final Set<GroupConversationState> AWAITING_STATES = EnumSet.of(GroupConversationState.AWAITING_APPROVAL,
+            GroupConversationState.AWAITING_HUMAN_INPUT);
+
+    /**
+     * Default lease for a non-terminal, non-paused cadence discussion that has
+     * stopped advancing. Generous on purpose — it is measured against the
+     * discussion's own progress heartbeat, so it only has to outlast the slowest
+     * legitimate gap between two phase boundaries, not a whole discussion.
+     */
+    static final String DEFAULT_ABANDONED_RUN_LEASE = "PT6H";
+
     private final IGroupWorkspaceStore workspaceStore;
     private final IGroupConversationStore conversationStore;
     private final GroupConversationService groupConversationService;
     private final ITemplatingEngine templatingEngine;
     private final MeterRegistry meterRegistry;
+    private final Duration abandonedRunLease;
 
     private Counter cadenceRunsStarted;
     private Counter cadenceRunsSkipped;
     private Counter cadenceWritebacks;
+    private Counter cadenceAbandonedRuns;
 
     @Inject
     public TeamCadenceService(IGroupWorkspaceStore workspaceStore, IGroupConversationStore conversationStore,
             GroupConversationService groupConversationService, ITemplatingEngine templatingEngine,
-            MeterRegistry meterRegistry) {
+            MeterRegistry meterRegistry,
+            @ConfigProperty(name = "eddi.groups.cadence.abandoned-run-lease", defaultValue = DEFAULT_ABANDONED_RUN_LEASE) String abandonedRunLease) {
         this.workspaceStore = workspaceStore;
         this.conversationStore = conversationStore;
         this.groupConversationService = groupConversationService;
         this.templatingEngine = templatingEngine;
         this.meterRegistry = meterRegistry;
+        this.abandonedRunLease = parseLease(abandonedRunLease);
+    }
+
+    /**
+     * An unparseable lease falls back to the default rather than failing startup.
+     */
+    private static Duration parseLease(String value) {
+        try {
+            Duration parsed = Duration.parse(value);
+            if (!parsed.isNegative() && !parsed.isZero()) {
+                return parsed;
+            }
+            LOGGER.warnf("eddi.groups.cadence.abandoned-run-lease must be positive (was '%s') — using %s",
+                    LogSanitizer.sanitize(value), DEFAULT_ABANDONED_RUN_LEASE);
+        } catch (Exception e) {
+            LOGGER.warnf("eddi.groups.cadence.abandoned-run-lease is not an ISO-8601 duration ('%s') — using %s",
+                    LogSanitizer.sanitize(value), DEFAULT_ABANDONED_RUN_LEASE);
+        }
+        return Duration.parse(DEFAULT_ABANDONED_RUN_LEASE);
     }
 
     @PostConstruct
     void initMetrics() {
         cadenceRunsStarted = meterRegistry.counter("eddi_team_cadence_runs_started_total");
         cadenceRunsSkipped = meterRegistry.counter("eddi_team_cadence_runs_skipped_total");
         cadenceWritebacks = meterRegistry.counter("eddi_team_cadence_writebacks_total");
+        cadenceAbandonedRuns = meterRegistry.counter("eddi_team_cadence_abandoned_runs_total");
     }
 
     /** True if the given schedule metadata marks a team-cadence schedule. */
@@ -245,24 +302,88 @@ public boolean reconcile(GroupWorkspace workspace) {
         GroupConversation gc;
         try {
             gc = conversationStore.read(runningId);
-        } catch (Exception e) {
-            // Deleted or unreadable — release the claim and return the pulled
-            // tasks; holding the workspace hostage to a vanished discussion would
-            // stall every future fire.
-            LOGGER.warnf("Cadence discussion %s for group %s is gone (%s) — releasing the claim",
-                    runningId, LogSanitizer.sanitize(workspace.getGroupId()), e.getClass().getSimpleName());
+        } catch (IResourceStore.ResourceNotFoundException e) {
+            // PROVABLY gone — release the claim and return the pulled tasks; holding
+            // the workspace hostage to a deleted discussion would stall every future
+            // fire.
+            LOGGER.warnf("Cadence discussion %s for group %s no longer exists — releasing the claim",
+                    LogSanitizer.sanitize(runningId), LogSanitizer.sanitize(workspace.getGroupId()));
             return writebackFailure(workspace, null);
+        } catch (Exception e) {
+            // NOT proof of absence. This used to catch everything and release the
+            // claim, so a read failure while the discussion was genuinely running
+            // returned its tasks to PENDING and let the next fire start a SECOND
+            // discussion on the same backlog. Same distinction
+            // AgentDeploymentManagement#isAgentConfigMissing already makes: skip this
+            // fire and leave the claim for the next one.
+            LOGGER.warnf("Could not read cadence discussion %s for group %s (%s) — skipping this fire and keeping the claim",
+                    LogSanitizer.sanitize(runningId), LogSanitizer.sanitize(workspace.getGroupId()), e.getClass().getSimpleName());
+            return false;
         }
         return switch (gc.getState()) {
             // A lost settle race means another reconciler (or a fresh claim) got
             // there first — this caller must treat the workspace as busy and let
             // the next fire read fresh state.
             case COMPLETED -> writebackCompleted(workspace, gc);
             case FAILED, CANCELLED -> writebackFailure(workspace, gc);
-            default -> false; // IN_PROGRESS, SYNTHESIZING, AWAITING_* — still running
+            // Non-terminal. Still running — unless nothing has touched it for longer
+            // than the lease, in which case the pod that owned it is gone.
+            default -> reclaimIfAbandoned(workspace, gc);
         };
     }
 
+    /**
+     * Releases the claim of a non-terminal discussion that has stopped making
+     * progress, or reports "still running" if it has not.
+     * <p>
+     * <b>Why a lease is needed at all.</b> This class's protocol was documented as
+     * crash-proof — "a pod crash mid-discussion loses nothing, because the next
+     * fire finds the terminal state and reconciles it". That only holds if
+     * something moves the discussion to a terminal state, and on a pod crash
+     * nothing does: no startup sweep touches an IN_PROGRESS
+     * {@code GroupConversation} ({@code HitlCrashRecoveryObserver} handles only the
+     * AWAITING_* states). {@code reconcile} then read IN_PROGRESS and answered
+     * "still running" on every future fire, forever. The cadence was wedged until a
+     * human cancelled the discussion by hand.
+     * <p>
+     * <b>Why the lease is measured on {@code lastModified}, not on claim time.</b>
+     * A claim-age lease cannot tell a dead pod from a healthy long-running
+     * discussion, and reclaiming a live one would orphan its outcomes AND
+     * double-schedule its tasks — recreating, by design, the bug fixed above. The
+     * discussion loop persists the conversation at every phase boundary, so
+     * {@code lastModified} is a free liveness heartbeat: only a discussion that has
+     * genuinely stopped advancing expires.
+     * <p>
+     * <b>AWAITING_* is never reclaimed, at any age.</b> A discussion may
+     * legitimately wait on a human for days, and every surface that resolves one —
+     * resume, cancel, the timeout policies re-armed at startup — works cross-pod,
+     * so a paused discussion on a dead pod still progresses. Only the states that
+     * require a live in-process loop can go stale.
+     */
+    private boolean reclaimIfAbandoned(GroupWorkspace workspace, GroupConversation gc) {
+        if (AWAITING_STATES.contains(gc.getState())) {
+            return false; // Legitimately waiting on a human — never expires.
+        }
+        // A missing progress stamp must NOT read as "active". Treating it that way
+        // reinstates exactly the deadlock this lease exists to break, for any record
+        // whose lastModified was never written. Fall back to the creation stamp,
+        // which the discussion-creation path always sets, and if neither exists there
+        // is no evidence of progress at all — which is the definition of abandoned.
+        Instant lastProgress = gc.getLastModified() != null ? gc.getLastModified() : gc.getCreated();
+        if (lastProgress != null && lastProgress.isAfter(Instant.now().minus(abandonedRunLease))) {
+            return false; // Still advancing.
+        }
+
+        LOGGER.warnf("Cadence discussion %s for group %s has not advanced since %s (lease %s) — treating it as abandoned, "
+                + "releasing the claim and returning its tasks to the backlog (a null timestamp means no progress was ever recorded)",
+                LogSanitizer.sanitize(gc.getId()), LogSanitizer.sanitize(workspace.getGroupId()), lastProgress, abandonedRunLease);
+        cadenceAbandonedRuns.increment();
+        // Cancel before releasing: a zombie loop that somehow survives must not keep
+        // spending the cadence's budget on work nobody will collect.
+        cancelQuietly(gc.getId());
+        return writebackFailure(workspace, gc);
+    }
+
     /**
      * Writeback for a COMPLETED cadence discussion: match every pulled backlog task
      * against the discussion's task list by subject — VERIFIED outcomes stay