Repository navigation
fix(groups): a paused cadence discussion no longer wedges a standing team forever #650
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -7,10 +7,12 @@ | |||||||||||||
| import ai.labs.eddi.configs.groups.IAgentGroupStore; | ||||||||||||||
| import ai.labs.eddi.configs.groups.IGroupWorkspaceStore; | ||||||||||||||
| import ai.labs.eddi.configs.groups.IRestGroupWorkspace; | ||||||||||||||
| import ai.labs.eddi.configs.groups.model.AgentGroupConfiguration.DiscussionPhase; | ||||||||||||||
| import ai.labs.eddi.configs.groups.model.GroupWorkspace; | ||||||||||||||
| import ai.labs.eddi.configs.groups.model.GroupWorkspace.Cadence; | ||||||||||||||
| import ai.labs.eddi.configs.groups.model.SharedTaskList; | ||||||||||||||
| import ai.labs.eddi.configs.groups.model.SharedTaskList.TaskItem; | ||||||||||||||
| import ai.labs.eddi.configs.hitl.HitlTimeoutPolicy; | ||||||||||||||
| import ai.labs.eddi.datastore.IResourceStore; | ||||||||||||||
| import ai.labs.eddi.engine.runtime.internal.CronParser; | ||||||||||||||
| import ai.labs.eddi.engine.runtime.internal.TeamCadenceService; | ||||||||||||||
|
|
@@ -188,6 +190,11 @@ public Response addCadence(String groupId, CadenceRequest request) { | |||||||||||||
| .entity(Map.of("error", "inputTemplate exceeds " + MAX_INPUT_TEMPLATE_LENGTH + " characters")) | ||||||||||||||
| .build(); | ||||||||||||||
| } | ||||||||||||||
| String indefinitePause = indefinitePauseWarning(groupId); | ||||||||||||||
| if (indefinitePause != null) { | ||||||||||||||
| LOG.warnf("Cadence added to group %s which pauses for approval under WAIT_INDEFINITELY: %s", | ||||||||||||||
| sanitize(groupId), indefinitePause); | ||||||||||||||
| } | ||||||||||||||
| String cadenceId = UUID.randomUUID().toString(); | ||||||||||||||
| String principal = identity != null && identity.getPrincipal() != null | ||||||||||||||
| && identity.getPrincipal().getName() != null && !identity.getPrincipal().getName().isBlank() | ||||||||||||||
|
|
@@ -283,4 +290,49 @@ private void requireGroupExists(String groupId) throws IResourceStore.ResourceNo | |||||||||||||
| throw new IResourceStore.ResourceNotFoundException("Group not found."); | ||||||||||||||
| } | ||||||||||||||
| } | ||||||||||||||
|
|
||||||||||||||
| /** | ||||||||||||||
| * Warns when a group combines automated cadences with an approval gate that | ||||||||||||||
| * never times out. | ||||||||||||||
| * <p> | ||||||||||||||
| * A cadence discussion that pauses for an approval nobody gives is not a | ||||||||||||||
| * terminal state, so writeback never releases the workspace claim. The claim | ||||||||||||||
| * TTL in {@code TeamCadenceService} reclaims it eventually, but "eventually, | ||||||||||||||
| * after a day, by abandoning the run" is a backstop — the operator almost | ||||||||||||||
| * certainly wanted either a finite {@code hitlConfig.timeoutPolicy} or no | ||||||||||||||
| * approval gate on an unattended recurring run. Warn rather than reject: the | ||||||||||||||
| * combination is legitimate for a team whose approver genuinely is always | ||||||||||||||
| * available, and refusing it would break existing configs. | ||||||||||||||
| * | ||||||||||||||
| * @return a human-readable reason, or {@code null} when the combination is not | ||||||||||||||
| * present (or the config cannot be read — never fail an otherwise valid | ||||||||||||||
| * cadence over a diagnostic) | ||||||||||||||
| */ | ||||||||||||||
| private String indefinitePauseWarning(String groupId) { | ||||||||||||||
| try { | ||||||||||||||
| var resourceId = groupStore.getCurrentResourceId(groupId); | ||||||||||||||
| if (resourceId == null) { | ||||||||||||||
| return null; | ||||||||||||||
| } | ||||||||||||||
| var config = groupStore.read(groupId, resourceId.getVersion()); | ||||||||||||||
| if (config == null || config.getPhases() == null) { | ||||||||||||||
| return null; | ||||||||||||||
| } | ||||||||||||||
| boolean gatesForApproval = config.getPhases().stream().anyMatch(DiscussionPhase::requiresApproval); | ||||||||||||||
| if (!gatesForApproval) { | ||||||||||||||
| return null; | ||||||||||||||
| } | ||||||||||||||
| var hitl = config.getHitlConfig(); | ||||||||||||||
| boolean waitsForever = hitl == null || hitl.getTimeoutPolicy() == null | ||||||||||||||
| || hitl.getTimeoutPolicy() == HitlTimeoutPolicy.WAIT_INDEFINITELY; | ||||||||||||||
| if (!waitsForever) { | ||||||||||||||
| return null; | ||||||||||||||
| } | ||||||||||||||
| return "the group has requiresApproval phase(s) and hitlConfig.timeoutPolicy=WAIT_INDEFINITELY, so an " | ||||||||||||||
| + "unapproved run holds this team's cadence claim until the claim TTL reclaims it; set a finite " | ||||||||||||||
| + "timeoutPolicy to resolve such pauses properly"; | ||||||||||||||
|
Comment on lines
+331
to
+333
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win Make the warning accurate when claim reclamation is disabled. Non-positive Make the message conditional on an enabled positive TTL, or remove the unconditional reclamation promise. Proposed wording- + "unapproved run holds this team's cadence claim until the claim TTL reclaims it; set a finite "
- + "timeoutPolicy to resolve such pauses properly";
+ + "unapproved run can hold this team's cadence claim indefinitely; set a finite timeoutPolicy "
+ + "and configure a positive eddi.groups.cadence.claim-ttl if reclamation is desired";📝 Committable suggestion
Suggested change
🤖 Prompt for AI Agents |
||||||||||||||
| } catch (Exception e) { | ||||||||||||||
| return null; | ||||||||||||||
| } | ||||||||||||||
| } | ||||||||||||||
| } | ||||||||||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,15 +16,18 @@ | |
| import ai.labs.eddi.configs.groups.model.SharedTaskList.TaskItem; | ||
| import ai.labs.eddi.configs.groups.model.SharedTaskList.TaskStatus; | ||
| import ai.labs.eddi.engine.internal.GroupConversationService; | ||
| import ai.labs.eddi.engine.lifecycle.model.ControlSignal; | ||
| import ai.labs.eddi.modules.templating.ITemplatingEngine; | ||
| import ai.labs.eddi.utils.LogSanitizer; | ||
| import io.micrometer.core.instrument.Counter; | ||
| import io.micrometer.core.instrument.MeterRegistry; | ||
| 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; | ||
|
|
@@ -79,32 +82,60 @@ public class TeamCadenceService { | |
| /** Fallback identity when a cadence predates {@code createdBy}. */ | ||
| public static final String FALLBACK_USER_ID = "system:team-cadence"; | ||
|
|
||
| /** | ||
| * How long a cadence run may hold its workspace claim before {@link #reconcile} | ||
| * treats it as stale and reclaims it. | ||
| * <p> | ||
| * A claim is released by writeback when its discussion reaches a terminal | ||
| * state, and an {@code AWAITING_APPROVAL} / {@code AWAITING_HUMAN_INPUT} pause | ||
| * is not one. The default group HITL timeout policy is | ||
| * {@code WAIT_INDEFINITELY}, so one unapproved cadence discussion held the | ||
| * claim forever: every later fire for that group was skipped as "still | ||
| * running", and the tasks it pulled stayed IN_PROGRESS on the backlog. Nothing | ||
| * anywhere reaped it. | ||
| * <p> | ||
| * 24 hours is deliberately generous — an approval that arrives the next | ||
| * business morning must still land on the discussion it belongs to, not on a | ||
| * reclaimed corpse. This is a liveness backstop for a wedged team, not an SLA; | ||
| * an operator who wants a tighter bound should set a finite | ||
| * {@code hitlConfig.timeoutPolicy} on the group, which resolves the pause | ||
| * properly instead of abandoning it. | ||
| */ | ||
| static final Duration DEFAULT_CLAIM_TTL = Duration.ofHours(24); | ||
|
|
||
| private final IGroupWorkspaceStore workspaceStore; | ||
| private final IGroupConversationStore conversationStore; | ||
| private final GroupConversationService groupConversationService; | ||
| private final ITemplatingEngine templatingEngine; | ||
| private final MeterRegistry meterRegistry; | ||
| private final Duration claimTtl; | ||
|
|
||
| private Counter cadenceRunsStarted; | ||
| private Counter cadenceRunsSkipped; | ||
| private Counter cadenceWritebacks; | ||
| private Counter cadenceClaimsReclaimed; | ||
|
|
||
| @Inject | ||
| public TeamCadenceService(IGroupWorkspaceStore workspaceStore, IGroupConversationStore conversationStore, | ||
| GroupConversationService groupConversationService, ITemplatingEngine templatingEngine, | ||
| MeterRegistry meterRegistry) { | ||
| MeterRegistry meterRegistry, | ||
| @ConfigProperty(name = "eddi.groups.cadence.claim-ttl", defaultValue = "PT24H") Duration claimTtl) { | ||
| this.workspaceStore = workspaceStore; | ||
| this.conversationStore = conversationStore; | ||
| this.groupConversationService = groupConversationService; | ||
| this.templatingEngine = templatingEngine; | ||
| this.meterRegistry = meterRegistry; | ||
| // Non-positive disables reclaiming entirely, for an operator who would rather | ||
| // wedge than risk abandoning a pause. | ||
| this.claimTtl = claimTtl != null ? claimTtl : DEFAULT_CLAIM_TTL; | ||
| } | ||
|
|
||
| @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"); | ||
| cadenceClaimsReclaimed = meterRegistry.counter("eddi_team_cadence_claims_reclaimed_total"); | ||
| } | ||
|
|
||
| /** True if the given schedule metadata marks a team-cadence schedule. */ | ||
|
|
@@ -193,6 +224,10 @@ public CadenceResult processScheduledFire(Map<String, Object> metadata) { | |
| // 4. Claim: conditional on the persisted idle marker. | ||
| String before = workspace.getRunningDiscussionId(); | ||
| workspace.setRunningDiscussionId(gc.getId()); | ||
| // Stamped with the claim so reconcile can tell a long-running discussion | ||
| // from a wedged one. Set here, next to the id it belongs to, and cleared | ||
| // in settle() alongside it — the two are one fact. | ||
| workspace.setClaimedAt(Instant.now()); | ||
| workspace.setPulledTaskIds(pulled.stream().map(TaskItem::id).toList()); | ||
| for (TaskItem task : pulled) { | ||
| workspace.getBacklog().updateTask(withStatus(task, TaskStatus.IN_PROGRESS)); | ||
|
|
@@ -259,10 +294,55 @@ 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 | ||
| // IN_PROGRESS, SYNTHESIZING, AWAITING_* — still running, unless the claim | ||
| // has been held past its TTL. See reclaimIfStale. | ||
| default -> reclaimIfStale(workspace, gc); | ||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Releases a claim held past {@link #claimTtl}, cancelling the discussion that | ||
| * held it. | ||
| * <p> | ||
| * Without this a cadence discussion that pauses for an approval nobody gives | ||
| * wedges its team permanently: the pause is not a terminal state, so writeback | ||
| * never runs; the claim is never released; every subsequent fire is skipped as | ||
| * "still running"; and the tasks that run pulled stay IN_PROGRESS on the | ||
| * backlog. The default group HITL policy is {@code WAIT_INDEFINITELY}, so this | ||
| * needs no unusual configuration to happen — one unapproved run is enough. | ||
| * <p> | ||
| * The discussion is cancelled before the claim is released, so a run cannot | ||
| * keep spending against a budget nobody is tracking any more, and its tasks go | ||
| * back to PENDING through the ordinary failure writeback rather than a bespoke | ||
| * path. | ||
| * | ||
| * @return {@code true} when the claim was released and the workspace is idle | ||
| * again, {@code false} while the run is still legitimately in flight | ||
| */ | ||
| private boolean reclaimIfStale(GroupWorkspace workspace, GroupConversation gc) { | ||
| if (claimTtl == null || claimTtl.isZero() || claimTtl.isNegative()) { | ||
| return false; | ||
| } | ||
| Instant claimedAt = workspace.getClaimedAt(); | ||
| if (claimedAt == null) { | ||
| // A workspace written before the stamp existed. Reclaiming on a missing | ||
| // timestamp would be a guess; it gets a stamp on its next claim. | ||
| return false; | ||
| } | ||
| if (Instant.now().isBefore(claimedAt.plus(claimTtl))) { | ||
| return false; | ||
| } | ||
|
|
||
| LOGGER.warnf("Cadence claim on group %s has been held since %s (past the %s TTL) by discussion %s in state %s — " | ||
| + "cancelling it and releasing the claim so the team's cadences can fire again", | ||
| LogSanitizer.sanitize(workspace.getGroupId()), claimedAt, claimTtl, gc.getId(), gc.getState()); | ||
| cancelQuietly(gc.getId()); | ||
| cadenceClaimsReclaimed.increment(); | ||
| // The ordinary failure writeback: returns the pulled tasks to PENDING and | ||
| // clears the claim, exactly as a FAILED/CANCELLED discussion would. | ||
| return writebackFailure(workspace, gc); | ||
|
Comment on lines
+339
to
+343
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win Do not release a claim when cancellation is unconfirmed. Line 339 calls Only run failure writeback after cancellation succeeds. If cancellation fails, keep the claim or re-read the discussion and use the normal terminal-state path. Increment Proposed fix- cancelQuietly(gc.getId());
- cadenceClaimsReclaimed.increment();
- // The ordinary failure writeback: returns the pulled tasks to PENDING and
- // clears the claim, exactly as a FAILED/CANCELLED discussion would.
- return writebackFailure(workspace, gc);
+ if (!cancelQuietly(gc.getId())) {
+ return false;
+ }
+ // The ordinary failure writeback: returns the pulled tasks to PENDING and
+ // clears the claim, exactly as a FAILED/CANCELLED discussion would.
+ boolean released = writebackFailure(workspace, gc);
+ if (released) {
+ cadenceClaimsReclaimed.increment();
+ }
+ return released;-private void cancelQuietly(String discussionId) {
+private boolean cancelQuietly(String discussionId) {
try {
- groupConversationService.cancelDiscussion(discussionId, ControlSignal.CANCEL_GRACEFUL);
+ return groupConversationService.cancelDiscussion(discussionId, ControlSignal.CANCEL_GRACEFUL);
} catch (Exception e) {
LOGGER.warnf("Could not cancel discussion %s after a lost or expired cadence claim: %s", discussionId, e.getMessage());
+ return false;
}
}Also applies to: 517-526 🤖 Prompt for AI Agents |
||
| } | ||
|
|
||
| /** | ||
| * Writeback for a COMPLETED cadence discussion: match every pulled backlog task | ||
| * against the discussion's task list by subject — VERIFIED outcomes stay | ||
|
|
@@ -343,6 +423,7 @@ private boolean settle(GroupWorkspace workspace, GroupConversation gc, String se | |
| metrics.setTotalCost(metrics.getTotalCost() + gc.getTotalCost()); | ||
| } | ||
| workspace.setRunningDiscussionId(GroupWorkspace.NO_RUNNING_DISCUSSION); | ||
| workspace.setClaimedAt(null); | ||
| workspace.setPulledTaskIds(List.of()); | ||
| try { | ||
| if (!workspaceStore.casRunningDiscussion(workspace, settledDiscussionId)) { | ||
|
|
@@ -435,9 +516,14 @@ private String renderQuestion(Cadence cadence, GroupWorkspace workspace, List<Ta | |
|
|
||
| private void cancelQuietly(String discussionId) { | ||
| try { | ||
| groupConversationService.cancelDiscussion(discussionId, null); | ||
| // Explicitly graceful. A null mode already resolved to CANCEL_GRACEFUL — | ||
| // only CANCEL_IMMEDIATE takes the other branch — but this method now has a | ||
| // second caller (the stale-claim reclaim), and "which cancel is this?" is | ||
| // not a question either call site should have to answer by reading | ||
| // GroupHitlCoordinator. | ||
| groupConversationService.cancelDiscussion(discussionId, ControlSignal.CANCEL_GRACEFUL); | ||
| } catch (Exception e) { | ||
| LOGGER.warnf("Could not cancel discussion %s after a lost cadence claim: %s", discussionId, e.getMessage()); | ||
| LOGGER.warnf("Could not cancel discussion %s after a lost or expired cadence claim: %s", discussionId, e.getMessage()); | ||
| } | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Log the warning only after cadence persistence succeeds.
This block runs before
scheduleStore.createScheduleandworkspaceStore.update. If either operation fails, the log states that a cadence was added even though no cadence was persisted. Move the warning after the successful workspace update, or use preflight wording.Proposed fix
🤖 Prompt for AI Agents