Shared dedup key builders
apps/backend/internal/runs/dedupkeys/dedupkeys.go ↗Centralizes the generation-aware key so both assignment producers converge structurally, not by coincidence.
AssignmentKey
func AssignmentKey(taskID, agentProfileID string, generation int64) string {
return fmt.Sprintf("task_assigned:%s:%s:%d", taskID, agentProfileID, generation)
}
BlockerDigest
func BlockerDigest(blockerTaskIDs []string) string {
sorted := make([]string, len(blockerTaskIDs))
copy(sorted, blockerTaskIDs)
sort.Strings(sorted)
sum := sha256.Sum256([]byte(strings.Join(sorted, ",")))
return fmt.Sprintf("%x", sum)
}
Observable dedup in the runs queue
apps/backend/internal/runs/service/dedup.go ↗Classifies every suppression as windowed or durable, counts it, and logs it at a visible level.
Windowed vs durable
func ReportWindowedDedup(q QueueSource, reason, key string) QueueOutcome {
incRunDedup(q, reason, "windowed")
dedupLogger().Info("run deduplicated (windowed)",
zap.String("queue", string(q)),
zap.String("reason", reason),
zap.String("key", key))
return QueueOutcomeDeduped
}
func ReportInsertResult(q QueueSource, reason, key, agentProfileID string, err error) (QueueOutcome, error) {
if err == nil {
return QueueOutcomeQueued, nil
}
if runssqlite.IsIdempotencyKeyUniqueViolation(err) {
return ReportDurableDedup(q, reason, key, agentProfileID), nil
}
return QueueOutcomeNone, err
}
func ReportDurableDedup(q QueueSource, reason, key, agentProfileID string) QueueOutcome {
incRunDedup(q, reason, "durable")
dedupLogger().Warn("run deduplicated (durable index)",
zap.String("queue", string(q)),
zap.String("reason", reason),
zap.String("key", key),
zap.String("agent_profile_id", agentProfileID))
return QueueOutcomeDeduped
}
Keyless fallback
func ReportKeylessEnqueue(reason string, cause KeylessCause, detail string) {
incRunDedupKeyless(reason, cause)
if cause != KeylessCauseUnresolved {
return
}
dedupLogger().Info("run enqueued with no dedup key",
zap.String("reason", reason),
zap.String("cause", string(cause)),
zap.String("detail", detail))
}
Generation column and atomic bump
apps/backend/internal/office/repository/sqlite/tasks.go ↗Bumps assignment_generation on every assignment and returns the new value inside the same transaction.
UpdateTaskAssignee
func (r *Repository) UpdateTaskAssignee(ctx context.Context, taskID, assigneeID string) (int64, error) {
var stepID string
err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(
`SELECT COALESCE(workflow_step_id, '') FROM tasks WHERE id = ?`),
taskID).Scan(&stepID)
if err != nil {
if err == sql.ErrNoRows {
return 0, fmt.Errorf("task not found: %s", taskID)
}
return 0, err
}
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return 0, err
}
defer func() { _ = tx.Rollback() }()
// ... upsert runner participant ...
if _, err := tx.ExecContext(ctx, tx.Rebind(`
UPDATE tasks SET assignment_generation = assignment_generation + 1, updated_at = CURRENT_TIMESTAMP WHERE id = ?
`), taskID); err != nil {
return 0, err
}
var generation int64
if err := tx.QueryRowxContext(ctx, tx.Rebind(
`SELECT assignment_generation FROM tasks WHERE id = ?`),
taskID).Scan(&generation); err != nil {
return 0, err
}
if err := tx.Commit(); err != nil {
return 0, err
}
return generation, nil
}
Create path
if task.AssigneeAgentProfileID != "" && task.WorkflowStepID != "" {
if err := upsertRunnerInTx(ctx, tx, r.db.Rebind, task.WorkflowStepID, task.ID, task.AssigneeAgentProfileID); err != nil {
return "", err
}
if _, err := tx.ExecContext(ctx, r.db.Rebind(
`UPDATE tasks SET assignment_generation = 1 WHERE id = ?`), task.ID); err != nil {
return "", err
}
}
Dashboard carries the generation
apps/backend/internal/office/dashboard/service_tasks.go ↗Reads the generation from the committing transaction and passes it to the reactivity pipeline instead of re-reading later.
SetTaskAssigneeAsAgent
generation, err := s.repo.UpdateTaskAssignee(ctx, taskID, assigneeID)
if err != nil {
return err
}
s.publishTaskUpdated(ctx, taskID, []string{"assignee_agent_profile_id"})
s.runReactivityForAssigneeChange(ctx, taskID, prevAssignee, assigneeID, callerAgentID, generation)
return nil
Reactivity change
change := TaskReactivityChange{
NewAssigneeID: &newAssigneeID,
AssignmentGeneration: &generation,
PrevAssigneeID: prevAssigneeID,
ActorID: callerAgentID,
ActorType: actorType,
}
result, err := s.reactivity.ApplyTaskMutation(ctx, taskID, "", change)
Reactivity builds the generational key
apps/backend/internal/office/scheduler/reactivity.go ↗Uses the carried generation to build task_assigned:<task>:<agent>:<gen> and falls back to keyless enqueue when generation is nil.
reactToAssigneeChange
func (ss *SchedulerService) reactToAssigneeChange(
task *TaskSnapshot,
newAssigneeID string,
change TaskMutation,
queue func(string, RunContext),
res *ApplyTaskMutationResult,
) {
if task.AssigneeAgentProfileID != "" && newAssigneeID != task.AssigneeAgentProfileID {
res.InterruptSessionID = task.ID
}
if newAssigneeID == "" {
return
}
var key string
if change.AssignmentGeneration != nil {
key = dedupkeys.AssignmentKey(task.ID, newAssigneeID, *change.AssignmentGeneration)
} else {
runsservice.ReportKeylessEnqueue(RunReasonTaskAssigned, runsservice.KeylessCauseUnresolved, "nil_mutation_generation")
}
queue(newAssigneeID, RunContext{
Reason: RunReasonTaskAssigned,
TaskID: task.ID,
WorkspaceID: task.WorkspaceID,
ActorID: change.ActorID,
ActorType: change.ActorType,
IdempotencyKey: key,
})
}
Event subscriber converges on the same key
apps/backend/internal/office/service/event_subscribers.go ↗The second producer derives the identical key from the same generation, or enqueues keyless when the event lacks a generation.
queueTaskAssignedRun
func (s *Service) queueTaskAssignedRun(
ctx context.Context,
taskID string,
agentProfileID string,
assignmentGeneration *int64,
fallbackToStoredRunner bool,
) error {
if taskID == "" {
return nil
}
fields, err := s.repo.GetTaskExecutionFields(ctx, taskID)
if err != nil {
if errors.Is(err, sqlite.ErrTaskNotFound) {
return nil
}
return fmt.Errorf("get task execution fields for assignment: %w", err)
}
if fields == nil || !fields.IsFromOffice {
return nil
}
fellBackToStoredRunner := false
if agentProfileID == "" && fallbackToStoredRunner {
agentProfileID = fields.AssigneeAgentProfileID
fellBackToStoredRunner = agentProfileID != ""
}
if agentProfileID == "" {
return nil
}
payload := mustJSON(map[string]string{"task_id": taskID})
var key string
if fellBackToStoredRunner || assignmentGeneration == nil {
runsservice.ReportKeylessEnqueue(RunReasonTaskAssigned, runsservice.KeylessCauseUnresolved, "event_missing_generation")
} else {
key = dedupkeys.AssignmentKey(taskID, agentProfileID, *assignmentGeneration)
}
_, err = s.QueueRun(ctx, agentProfileID, RunReasonTaskAssigned, payload, key)
return err
}
Blocker convergence
key := fmt.Sprintf("blockers_resolved:%s:%s", blockedTaskID, dedupkeys.BlockerDigest(blockerIDs))
return s.dispatchEngineTrigger(ctx, blockedTaskID, engine.TriggerOnBlockerResolved,
engine.OnBlockerResolvedPayload{
ResolvedBlockerIDs: []string{resolvedBlockerID},
}, key)
