From a80ad86f635d5deb3a32c05e19ebca06cc986a6e Mon Sep 17 00:00:00 2001 From: Gregor Jarisch Date: Sun, 9 Aug 2026 15:56:13 +0200 Subject: [PATCH 1/3] fix(groups): give the cadence claim a lease; stop a NATS publish failure dropping a turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The claim protocol had no expiry. Its Javadoc promised crash-proofness via 'the next fire finds the terminal state', but nothing moves a discussion to a terminal state when its pod dies, so reconcile answered 'still running' forever and the cadence was wedged until a human cancelled by hand. A non-terminal, non-paused discussion whose progress heartbeat has not advanced within the lease is now cancelled, its tasks returned, and the claim released. The lease is measured on gc.getLastModified(), not 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. AWAITING_* never expires — those resolve cross-pod. CREATED does, being the crash window between the claim CAS and the executor starting. reconcile also caught every read exception as 'discussion gone', so a transient error double-scheduled the backlog; narrowed to ResourceNotFoundException. NatsConversationCoordinator.publishAndExecute caught only the two checked types, so an unchecked publish failure escaped and skipped the execution below it — dropping the turn with no callback and no dead-letter, while the catch block promised 'executing locally'. +14 tests, including two that pin invariants previously resting on prose: NATS callables execute in-process, and neither storage backend inherits the throwing storeIfFieldEquals default. --- docs/changelog.md | 20 ++ .../internal/NatsConversationCoordinator.java | 9 + .../runtime/internal/TeamCadenceService.java | 133 ++++++++++- .../StoreIfFieldEqualsContractTest.java | 131 +++++++++++ ...NatsCoordinatorInProcessInvariantTest.java | 94 ++++++++ .../internal/TeamCadenceClaimLeaseTest.java | 221 ++++++++++++++++++ .../internal/TeamCadenceServiceTest.java | 2 +- 7 files changed, 600 insertions(+), 10 deletions(-) create mode 100644 src/test/java/ai/labs/eddi/datastore/StoreIfFieldEqualsContractTest.java create mode 100644 src/test/java/ai/labs/eddi/engine/runtime/internal/NatsCoordinatorInProcessInvariantTest.java create mode 100644 src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java diff --git a/docs/changelog.md b/docs/changelog.md index d2f80a1af0..644137e303 100644 --- a/docs/changelog.md +++ b/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) diff --git a/src/main/java/ai/labs/eddi/engine/runtime/internal/NatsConversationCoordinator.java b/src/main/java/ai/labs/eddi/engine/runtime/internal/NatsConversationCoordinator.java index bea4db5e07..91ad139b19 100644 --- a/src/main/java/ai/labs/eddi/engine/runtime/internal/NatsConversationCoordinator.java +++ b/src/main/java/ai/labs/eddi/engine/runtime/internal/NatsConversationCoordinator.java @@ -333,6 +333,15 @@ private void publishAndExecute(String conversationId, BlockingQueue * 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. + *

+ * Crash recovery rests on the lease, not on the terminal state. 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,25 +98,61 @@ 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 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", 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", value, + DEFAULT_ABANDONED_RUN_LEASE); + } + return Duration.parse(DEFAULT_ABANDONED_RUN_LEASE); } @PostConstruct @@ -105,6 +160,7 @@ 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,13 +301,23 @@ public boolean reconcile(GroupWorkspace workspace) { GroupConversation gc; try { gc = conversationStore.read(runningId); + } 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", + runningId, LogSanitizer.sanitize(workspace.getGroupId())); + return writebackFailure(workspace, null); } 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", + // 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", runningId, LogSanitizer.sanitize(workspace.getGroupId()), e.getClass().getSimpleName()); - return writebackFailure(workspace, null); + return false; } return switch (gc.getState()) { // A lost settle race means another reconciler (or a fresh claim) got @@ -259,10 +325,59 @@ public boolean reconcile(GroupWorkspace workspace) { // 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. + *

+ * Why a lease is needed at all. 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. + *

+ * Why the lease is measured on {@code lastModified}, not on claim time. + * 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. + *

+ * AWAITING_* is never reclaimed, at any age. 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. + } + Instant lastProgress = gc.getLastModified(); + 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", + 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 diff --git a/src/test/java/ai/labs/eddi/datastore/StoreIfFieldEqualsContractTest.java b/src/test/java/ai/labs/eddi/datastore/StoreIfFieldEqualsContractTest.java new file mode 100644 index 0000000000..0b693a4a17 --- /dev/null +++ b/src/test/java/ai/labs/eddi/datastore/StoreIfFieldEqualsContractTest.java @@ -0,0 +1,131 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.datastore; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.mock; + +/** + * The compare-and-swap contract every group-concurrency safety property rests + * on. + *

+ * {@code compareAndSetState}, {@code updateIfState} and + * {@code casRunningDiscussion} are all implemented once, over + * {@link IResourceStorage#storeIfFieldEquals}, so their cross-process atomicity + * is exactly as good as that method is on the active backend. If the two + * backends disagree even slightly, every CAS-based guarantee in + * {@code executeDiscussion} and the cadence claim protocol is theoretical on + * one of them. + *

+ * The behavioural halves of that contract need a live database and are covered + * by the store integration tests. What is pinned here is the part that can + * silently rot without one: that a backend cannot forget to implement the + * method, and that both overloads exist on both backends with the same shape. + */ +@DisplayName("storeIfFieldEquals — cross-backend contract") +class StoreIfFieldEqualsContractTest { + + private static final Set BACKENDS = Set.of("ai.labs.eddi.datastore.mongo.MongoResourceStorage", + "ai.labs.eddi.datastore.postgres.PostgresResourceStorage"); + + @Nested + @DisplayName("the default must never silently degrade a CAS") + class NoSilentDegradation { + + /** + * A default that stored unconditionally would turn every CAS in the group + * subsystem into a last-writer-wins update, with no error and no test failure — + * the single most dangerous shape this interface could have. + */ + @Test + @DisplayName("the interface default throws rather than storing unconditionally") + void defaultThrows() throws Exception { + // A backend that implements nothing but the interface defaults. + IResourceStorage unimplemented = mock(IResourceStorage.class, CALLS_REAL_METHODS); + + var thrown = assertThrows(UnsupportedOperationException.class, + () -> unimplemented.storeIfFieldEquals(null, "state", "IN_PROGRESS")); + assertTrue(thrown.getMessage().contains("must never silently degrade"), thrown.getMessage()); + + var thrownNumeric = assertThrows(UnsupportedOperationException.class, + () -> unimplemented.storeIfFieldEquals(null, "version", 3L)); + assertTrue(thrownNumeric.getMessage().contains("must never silently degrade"), thrownNumeric.getMessage()); + } + } + + @Nested + @DisplayName("both backends implement both overloads") + class BackendParity { + + /** + * A backend that inherited the default would throw at the first conditional + * write — in production, on a code path exercised only under concurrency. + */ + @Test + @DisplayName("neither backend inherits the throwing default") + void bothBackendsOverrideBothOverloads() throws Exception { + for (String backend : BACKENDS) { + Class type = Class.forName(backend); + + Method stringOverload = type.getMethod("storeIfFieldEquals", IResourceStorage.IResource.class, String.class, String.class); + Method longOverload = type.getMethod("storeIfFieldEquals", IResourceStorage.IResource.class, String.class, long.class); + + assertEquals(type, stringOverload.getDeclaringClass(), + backend + " must declare the String overload — inheriting the default throws at the first conditional write"); + assertEquals(type, longOverload.getDeclaringClass(), + backend + " must declare the long overload; the two backends disagree about text-comparing numbers, " + + "which is the whole reason it exists"); + } + } + + /** + * Both overloads must be able to report the deleted-vs-mismatch distinction: + * callers map {@code ResourceNotFoundException} to 404 and + * {@code ResourceModifiedException} to 409, and + * {@code GroupConversationStore.updateIfState} converts the former into + * {@code GroupConversationGoneException} so a deletion is never reported as a + * state conflict. + */ + @Test + @DisplayName("both backends declare both CAS outcomes") + void bothBackendsDeclareBothOutcomes() throws Exception { + for (String backend : BACKENDS) { + Class type = Class.forName(backend); + for (Method method : new Method[]{ + type.getMethod("storeIfFieldEquals", IResourceStorage.IResource.class, String.class, String.class), + type.getMethod("storeIfFieldEquals", IResourceStorage.IResource.class, String.class, long.class)}) { + + Set> declared = Set.of(method.getExceptionTypes()); + assertTrue(declared.contains(IResourceStore.ResourceModifiedException.class), + backend + "." + method.getName() + " must declare ResourceModifiedException (the 409 case)"); + assertTrue(declared.contains(IResourceStore.ResourceNotFoundException.class), + backend + "." + method.getName() + " must declare ResourceNotFoundException (the 404 case) — " + + "conflating it with a mismatch reports a deleted discussion as a state conflict"); + } + } + } + } + + @Test + @DisplayName("the contract is documented on the interface, not only in the implementations") + void contractIsOnTheInterface() { + assertDoesNotThrow(() -> IResourceStorage.class.getMethod("storeIfFieldEquals", IResourceStorage.IResource.class, String.class, + String.class)); + assertDoesNotThrow( + () -> IResourceStorage.class.getMethod("storeIfFieldEquals", IResourceStorage.IResource.class, String.class, long.class)); + } + +} diff --git a/src/test/java/ai/labs/eddi/engine/runtime/internal/NatsCoordinatorInProcessInvariantTest.java b/src/test/java/ai/labs/eddi/engine/runtime/internal/NatsCoordinatorInProcessInvariantTest.java new file mode 100644 index 0000000000..dac23cdd3c --- /dev/null +++ b/src/test/java/ai/labs/eddi/engine/runtime/internal/NatsCoordinatorInProcessInvariantTest.java @@ -0,0 +1,94 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.engine.runtime.internal; + +import ai.labs.eddi.engine.runtime.IRuntime; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import jakarta.enterprise.inject.Instance; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Pins the node-affinity invariant that {@code LiveDiscussionRegistry} — and + * with it every group task / artifact / recruit tool — silently depends on. + *

+ * The registry is per-node and holds the live {@code GroupConversation} the + * discussion loop is mutating. Its Javadoc states the dependency explicitly: + * {@code MemberTurnExecutor#executeAgentTurn} always runs a member's turn + * in-process, so a tool invoked during that turn can hold a live reference into + * the registry. If a member turn could ever be routed to another node, those + * tools would not fail — they would silently not be assembled, and the model + * would lose them mid-discussion with nothing in the logs. + *

+ * That invariant currently holds because the NATS coordinator uses JetStream + * purely as a distributed ordering primitive: it publishes the + * conversation id as a marker and then executes the callable through the local + * runtime. Nothing enforces that, and the payload is the only thing standing + * between "ordering primitive" and "work queue" — so this test enforces it, + * rather than leaving a changelog note to be trusted. + */ +@DisplayName("NATS coordinator — callables execute in-process") +class NatsCoordinatorInProcessInvariantTest { + + @SuppressWarnings("unchecked") + private static NatsConversationCoordinator coordinator(IRuntime runtime) { + Instance metrics = mock(Instance.class); + when(metrics.isResolvable()).thenReturn(false); + return new NatsConversationCoordinator(runtime, metrics, new SimpleMeterRegistry(), "nats://localhost:4222", "eddi-conversations", + "eddi-deadletter", 3, 100); + } + + /** + * Two invariants in one, because the same line carries both. + *

    + *
  1. The callable is handed to {@code IRuntime}, i.e. executed on THIS JVM — + * the node-affinity invariant the group tools depend on.
    2. + *
  2. It is handed over even when the publish fails. The publish is an ordering + * marker, never the work, so no publish failure may cost a conversation its + * turn. {@code publishAndExecute} used to catch only {@code IOException} and + * {@code JetStreamApiException}, so an unchecked failure — the NATS client + * throws {@code IllegalStateException} on a closed or draining connection — + * escaped the method and skipped the execution entirely, silently dropping the + * turn while the catch block promised "executing locally".
    3. + *
+ */ + @Test + @DisplayName("the submitted callable is handed to the local runtime, never to a remote consumer") + void callableRunsThroughLocalRuntime() { + var runtime = mock(IRuntime.class); + var executed = new AtomicReference>(); + + doAnswer(invocation -> { + Callable callable = invocation.getArgument(0); + executed.set(callable); + callable.call(); + return CompletableFuture.completedFuture(null); + }).when(runtime).submitCallable(any(), any(IRuntime.IFinishedExecution.class), any()); + + var ranInProcess = new AtomicReference(); + Callable work = () -> { + ranInProcess.set(Thread.currentThread().getName()); + return null; + }; + + coordinator(runtime).submitInOrder("conversation-1", work); + + assertSame(work, executed.get(), "the coordinator must hand the original callable to the local runtime — " + + "if it were ever serialized onto the stream instead, a member turn could run on another node " + + "and LiveDiscussionRegistry would silently withhold every group tool"); + assertEquals(Thread.currentThread().getName(), ranInProcess.get(), "the work must have run in this process"); + } +} diff --git a/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java b/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java new file mode 100644 index 0000000000..d98336d7a6 --- /dev/null +++ b/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java @@ -0,0 +1,221 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.engine.runtime.internal; + +import ai.labs.eddi.configs.groups.IGroupConversationStore; +import ai.labs.eddi.configs.groups.IGroupWorkspaceStore; +import ai.labs.eddi.configs.groups.model.GroupConversation; +import ai.labs.eddi.configs.groups.model.GroupConversation.GroupConversationState; +import ai.labs.eddi.configs.groups.model.GroupWorkspace; +import ai.labs.eddi.datastore.IResourceStore; +import ai.labs.eddi.engine.internal.GroupConversationService; +import ai.labs.eddi.modules.templating.ITemplatingEngine; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * The cadence claim protocol had no expiry. + *

+ * Its Javadoc promised that "a pod crash mid-discussion loses nothing, because + * the next fire finds the terminal state and reconciles it" — but on a pod + * crash nothing moves the discussion to a terminal state, so {@code reconcile} + * answered "still running" forever and the cadence was wedged until a human + * cancelled the discussion by hand. + */ +@DisplayName("TeamCadenceService — claim lease") +class TeamCadenceClaimLeaseTest { + + private IGroupWorkspaceStore workspaceStore; + private IGroupConversationStore conversationStore; + private GroupConversationService groupConversationService; + private TeamCadenceService service; + + @BeforeEach + void setUp() { + workspaceStore = mock(IGroupWorkspaceStore.class); + conversationStore = mock(IGroupConversationStore.class); + groupConversationService = mock(GroupConversationService.class); + service = new TeamCadenceService(workspaceStore, conversationStore, groupConversationService, mock(ITemplatingEngine.class), + new SimpleMeterRegistry(), "PT6H"); + service.initMetrics(); + } + + private GroupWorkspace claimedWorkspace() { + var workspace = new GroupWorkspace(); + workspace.setGroupId("group-1"); + workspace.setId("ws-1"); + workspace.setRunningDiscussionId("gc-1"); + workspace.setPulledTaskIds(List.of()); + return workspace; + } + + private GroupConversation discussion(GroupConversationState state, Instant lastModified) { + var gc = new GroupConversation(); + gc.setId("gc-1"); + gc.setGroupId("group-1"); + gc.setState(state); + gc.setLastModified(lastModified); + return gc; + } + + @Nested + @DisplayName("an abandoned run") + class AbandonedRun { + + @Test + @DisplayName("an IN_PROGRESS discussion that stopped advancing past the lease is reclaimed") + void staleInProgressIsReclaimed() throws Exception { + when(conversationStore.read("gc-1")).thenReturn(discussion(GroupConversationState.IN_PROGRESS, Instant.now().minus(7, ChronoUnit.HOURS))); + when(workspaceStore.casRunningDiscussion(any(), anyString())).thenReturn(true); + + var workspace = claimedWorkspace(); + assertTrue(service.reconcile(workspace), "a wedged cadence must be reclaimable without human intervention"); + assertEquals(GroupWorkspace.NO_RUNNING_DISCUSSION, workspace.getRunningDiscussionId()); + } + + /** + * A zombie loop that somehow survives must not keep spending the cadence's + * budget on work whose outcomes nobody will collect. + */ + @Test + @DisplayName("the abandoned discussion is cancelled before the claim is released") + void abandonedDiscussionIsCancelled() throws Exception { + when(conversationStore.read("gc-1")).thenReturn(discussion(GroupConversationState.IN_PROGRESS, Instant.now().minus(7, ChronoUnit.HOURS))); + when(workspaceStore.casRunningDiscussion(any(), anyString())).thenReturn(true); + + service.reconcile(claimedWorkspace()); + + verify(groupConversationService).cancelDiscussion("gc-1", null); + } + + @Test + @DisplayName("CREATED counts too — a crash between the claim and the first turn") + void staleCreatedIsReclaimed() throws Exception { + when(conversationStore.read("gc-1")).thenReturn(discussion(GroupConversationState.CREATED, Instant.now().minus(7, ChronoUnit.HOURS))); + when(workspaceStore.casRunningDiscussion(any(), anyString())).thenReturn(true); + + assertTrue(service.reconcile(claimedWorkspace())); + } + } + + @Nested + @DisplayName("a run that is still alive") + class StillAlive { + + /** + * The lease is measured on the discussion's own progress heartbeat, not on + * claim age. A claim-age lease could not tell a dead pod from a healthy + * long-running discussion, and reclaiming a live one would orphan its outcomes + * AND double-schedule its tasks. + */ + @Test + @DisplayName("a recently-advanced discussion is left alone however old the claim is") + void recentlyAdvancedIsNotReclaimed() throws Exception { + when(conversationStore.read("gc-1")) + .thenReturn(discussion(GroupConversationState.IN_PROGRESS, Instant.now().minus(2, ChronoUnit.MINUTES))); + + assertFalse(service.reconcile(claimedWorkspace()), "a discussion that is still advancing must not be reclaimed"); + verify(groupConversationService, never()).cancelDiscussion(anyString(), any()); + verify(workspaceStore, never()).casRunningDiscussion(any(), anyString()); + } + + /** + * A discussion may legitimately wait on a human for days, and every surface + * that resolves one works cross-pod, so a paused discussion on a dead pod still + * progresses. Expiring it would destroy a live pending approval. + */ + @Test + @DisplayName("AWAITING_APPROVAL never expires, however stale") + void awaitingApprovalNeverExpires() throws Exception { + when(conversationStore.read("gc-1")) + .thenReturn(discussion(GroupConversationState.AWAITING_APPROVAL, Instant.now().minus(30, ChronoUnit.DAYS))); + + assertFalse(service.reconcile(claimedWorkspace())); + verify(groupConversationService, never()).cancelDiscussion(anyString(), any()); + } + + @Test + @DisplayName("AWAITING_HUMAN_INPUT never expires either") + void awaitingHumanInputNeverExpires() throws Exception { + when(conversationStore.read("gc-1")) + .thenReturn(discussion(GroupConversationState.AWAITING_HUMAN_INPUT, Instant.now().minus(30, ChronoUnit.DAYS))); + + assertFalse(service.reconcile(claimedWorkspace())); + verify(groupConversationService, never()).cancelDiscussion(anyString(), any()); + } + } + + @Nested + @DisplayName("read failures") + class ReadFailures { + + /** + * A transient read failure is not proof of absence. This used to catch every + * exception and release the claim, so a network blip while the discussion was + * genuinely running returned its tasks to PENDING and let the next fire start a + * SECOND discussion on the same backlog. + */ + @Test + @DisplayName("a transient read error keeps the claim rather than double-scheduling the work") + void transientErrorKeepsClaim() throws Exception { + when(conversationStore.read("gc-1")).thenThrow(new IResourceStore.ResourceStoreException("connection reset", null)); + + var workspace = claimedWorkspace(); + assertFalse(service.reconcile(workspace)); + assertEquals("gc-1", workspace.getRunningDiscussionId(), "the claim must survive a read failure"); + verify(workspaceStore, never()).casRunningDiscussion(any(), anyString()); + } + + /** A provably deleted discussion is a different matter: release the claim. */ + @Test + @DisplayName("a provably deleted discussion releases the claim") + void deletedReleasesClaim() throws Exception { + when(conversationStore.read("gc-1")).thenThrow(new IResourceStore.ResourceNotFoundException("gone")); + when(workspaceStore.casRunningDiscussion(any(), anyString())).thenReturn(true); + + var workspace = claimedWorkspace(); + assertTrue(service.reconcile(workspace)); + assertEquals(GroupWorkspace.NO_RUNNING_DISCUSSION, workspace.getRunningDiscussionId()); + } + } + + @Nested + @DisplayName("lease configuration") + class LeaseConfiguration { + + @Test + @DisplayName("an unparseable or non-positive lease falls back to the default instead of failing startup") + void invalidLeaseFallsBack() throws Exception { + for (String invalid : new String[]{"not-a-duration", "PT0S", "-PT1H"}) { + var fallback = new TeamCadenceService(workspaceStore, conversationStore, groupConversationService, + mock(ITemplatingEngine.class), new SimpleMeterRegistry(), invalid); + fallback.initMetrics(); + + when(conversationStore.read("gc-1")) + .thenReturn(discussion(GroupConversationState.IN_PROGRESS, Instant.now().minus(2, ChronoUnit.MINUTES))); + + assertFalse(fallback.reconcile(claimedWorkspace()), + "lease '" + invalid + "' must fall back to the 6h default, not to an instant expiry"); + } + } + } +} diff --git a/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceServiceTest.java b/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceServiceTest.java index 4845f25f47..80fa2da7bd 100644 --- a/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceServiceTest.java +++ b/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceServiceTest.java @@ -64,7 +64,7 @@ void setUp() { groupConversationService = mock(GroupConversationService.class); templatingEngine = mock(ITemplatingEngine.class); service = new TeamCadenceService(workspaceStore, conversationStore, groupConversationService, - templatingEngine, new SimpleMeterRegistry()); + templatingEngine, new SimpleMeterRegistry(), TeamCadenceService.DEFAULT_ABANDONED_RUN_LEASE); service.initMetrics(); } From 0303b82cb6ca33dbebcdd3194264f724f18ce3c6 Mon Sep 17 00:00:00 2001 From: Gregor Jarisch Date: Sun, 9 Aug 2026 16:06:22 +0200 Subject: [PATCH 2/3] fix(groups): sanitize ids and the configured lease in the new cadence log statements Matches the file's existing convention (groupId was already sanitized) and pre-empts the log-injection class CodeQL flagged on the sibling branches. --- .../engine/runtime/internal/TeamCadenceService.java | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java b/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java index 52a0154ed4..ca70eb13a1 100644 --- a/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java +++ b/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java @@ -147,10 +147,11 @@ private static Duration parseLease(String value) { if (!parsed.isNegative() && !parsed.isZero()) { return parsed; } - LOGGER.warnf("eddi.groups.cadence.abandoned-run-lease must be positive (was '%s') — using %s", value, DEFAULT_ABANDONED_RUN_LEASE); + 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", value, - DEFAULT_ABANDONED_RUN_LEASE); + 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); } @@ -306,7 +307,7 @@ public boolean reconcile(GroupWorkspace workspace) { // 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", - runningId, LogSanitizer.sanitize(workspace.getGroupId())); + 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 @@ -316,7 +317,7 @@ public boolean reconcile(GroupWorkspace workspace) { // 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", - runningId, LogSanitizer.sanitize(workspace.getGroupId()), e.getClass().getSimpleName()); + LogSanitizer.sanitize(runningId), LogSanitizer.sanitize(workspace.getGroupId()), e.getClass().getSimpleName()); return false; } return switch (gc.getState()) { @@ -370,7 +371,7 @@ private boolean reclaimIfAbandoned(GroupWorkspace workspace, GroupConversation g 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", - gc.getId(), LogSanitizer.sanitize(workspace.getGroupId()), lastProgress, abandonedRunLease); + 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. From 7da9fdc9c6c510a52652653d22575aa2349831a6 Mon Sep 17 00:00:00 2001 From: Gregor Jarisch Date: Sun, 9 Aug 2026 19:22:32 +0200 Subject: [PATCH 3/3] fix(groups): a missing progress stamp must not read as an active discussion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit on #657 (Major): reclaimIfAbandoned treated a null lastModified as 'still advancing', which reinstates exactly the deadlock the lease exists to break for any record whose lastModified was never written. Falls back to the creation stamp — which the discussion-creation path always sets, so a discussion created moments ago is not mistaken for an abandoned one — and reclaims when neither timestamp exists, since that is the definition of no progress ever recorded. +2 tests: null-everything reclaims, null-lastModified-but-recently-created does not. --- .../runtime/internal/TeamCadenceService.java | 11 ++++++-- .../internal/TeamCadenceClaimLeaseTest.java | 28 +++++++++++++++++++ 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java b/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java index ca70eb13a1..ff0a30db79 100644 --- a/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java +++ b/src/main/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceService.java @@ -364,13 +364,18 @@ private boolean reclaimIfAbandoned(GroupWorkspace workspace, GroupConversation g if (AWAITING_STATES.contains(gc.getState())) { return false; // Legitimately waiting on a human — never expires. } - Instant lastProgress = gc.getLastModified(); - if (lastProgress == null || lastProgress.isAfter(Instant.now().minus(abandonedRunLease))) { + // 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", + + "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 diff --git a/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java b/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java index d98336d7a6..5e786827b9 100644 --- a/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java +++ b/src/test/java/ai/labs/eddi/engine/runtime/internal/TeamCadenceClaimLeaseTest.java @@ -107,6 +107,34 @@ void abandonedDiscussionIsCancelled() throws Exception { verify(groupConversationService).cancelDiscussion("gc-1", null); } + /** + * A missing progress stamp is not evidence of activity. Reading it as "active" + * would reinstate the deadlock for any record whose lastModified was never + * written. + */ + @Test + @DisplayName("a discussion with no progress timestamp at all is reclaimed, not treated as active") + void nullTimestampsAreReclaimed() throws Exception { + var gc = discussion(GroupConversationState.IN_PROGRESS, null); + gc.setCreated(null); + when(conversationStore.read("gc-1")).thenReturn(gc); + when(workspaceStore.casRunningDiscussion(any(), anyString())).thenReturn(true); + + assertTrue(service.reconcile(claimedWorkspace()), "a null progress stamp must not wedge the cadence forever"); + } + + @Test + @DisplayName("a null lastModified falls back to the creation stamp before expiring") + void nullLastModifiedFallsBackToCreated() throws Exception { + var recentlyCreated = discussion(GroupConversationState.IN_PROGRESS, null); + recentlyCreated.setCreated(Instant.now().minus(3, ChronoUnit.MINUTES)); + when(conversationStore.read("gc-1")).thenReturn(recentlyCreated); + + assertFalse(service.reconcile(claimedWorkspace()), + "a discussion created minutes ago has simply not persisted progress yet"); + verify(groupConversationService, never()).cancelDiscussion(anyString(), any()); + } + @Test @DisplayName("CREATED counts too — a crash between the claim and the first turn") void staleCreatedIsReclaimed() throws Exception {