Concurrency gate self-heals and closes atomically
apps/backend/internal/office/routines/service.go ↗The gate now repairs a stale active run before it decides to skip or coalesce, and closes runs with an atomic conditional update.
Self-heal before policy
func (s *RoutineService) selfHealIfTaskTerminal(ctx context.Context, routine *Routine, active *RoutineRun) *RoutineRun {
if active.LinkedTaskID == "" {
return active
}
terminalStatus, err := s.repo.GetTaskTerminalStatus(ctx, active.LinkedTaskID)
if err != nil {
s.logger.Warn("check linked task terminal state", zap.String("run_id", active.ID), zap.Error(err))
return active
}
if terminalStatus == "" {
return active
}
if _, err := s.closeOutRun(ctx, active, terminalStatus); err != nil {
s.logger.Warn("close out stale active run", zap.String("run_id", active.ID), zap.Error(err))
return active
}
return nil
}
Loop with repair
func (s *RoutineService) applyConcurrencyPolicy(ctx context.Context, routine *Routine, run *RoutineRun, fingerprint string) (models.RoutineRunStatus, error) {
if routine.ConcurrencyPolicy == models.ConcurrencyPolicyAlwaysCreate {
return "", nil
}
for {
active, err := s.repo.GetActiveRunForFingerprint(ctx, routine.ID, fingerprint)
if err != nil {
return "", fmt.Errorf("check active run: %w", err)
}
if active == nil {
return "", nil
}
if repaired := s.selfHealIfTaskTerminal(ctx, routine, active); repaired == nil {
continue
} else {
active = repaired
}
switch routine.ConcurrencyPolicy {
case models.ConcurrencyPolicySkipIfActive:
_ = s.repo.UpdateRunStatus(ctx, run.ID, models.RoutineRunStatusSkipped, "")
run.Status = models.RoutineRunStatusSkipped
return models.RoutineRunStatusSkipped, nil
case models.ConcurrencyPolicyCoalesceIfActive:
_ = s.repo.UpdateRunCoalesced(ctx, run.ID, active.ID)
run.Status = models.RoutineRunStatusCoalesced
return models.RoutineRunStatusCoalesced, nil
}
}
}
Atomic close
func (s *RoutineService) closeOutRun(ctx context.Context, run *RoutineRun, terminalStatus string) (bool, error) {
status := models.RoutineRunStatusDone
switch terminalStatus {
case "cancelled":
status = models.RoutineRunStatusCancelled
case "failed", "missing":
status = models.RoutineRunStatusFailed
}
closed, err := s.repo.UpdateRunStatusIfTaskCreated(ctx, run.ID, status, run.LinkedTaskID)
if err != nil {
return false, fmt.Errorf("update run status: %w", err)
}
return closed, nil
}
func (s *RoutineService) SyncRunStatus(ctx context.Context, taskID, terminalStatus string) error {
if taskID == "" {
return nil
}
run, err := s.repo.GetRoutineRunByLinkedTaskID(ctx, taskID)
if err != nil {
return fmt.Errorf("get routine run by linked task: %w", err)
}
if run == nil {
return nil
}
_, err = s.closeOutRun(ctx, run, terminalStatus)
return err
}
Repository adds terminal-state readers and conditional close
apps/backend/internal/office/repository/sqlite/routines.go ↗The repo now reads the shared tasks table directly and closes a run only while it is still task_created.
Terminal status reader
func (r *Repository) GetTaskTerminalStatus(ctx context.Context, taskID string) (string, error) {
var state string
var archivedAt sql.NullTime
err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(
`SELECT COALESCE(state, ''), archived_at FROM tasks WHERE id = ?`), taskID).Scan(&state, &archivedAt)
if err == sql.ErrNoRows {
return "missing", nil
}
if err != nil {
return "", err
}
if archivedAt.Valid {
return "cancelled", nil
}
switch state {
case taskStateCompleted:
return "done", nil
case taskStateFailed:
return "failed", nil
case taskStateCancelled:
return "cancelled", nil
default:
return "", nil
}
}
Conditional close
func (r *Repository) UpdateRunStatusIfTaskCreated(ctx context.Context, runID string, status models.RoutineRunStatus, linkedTaskID string) (bool, error) {
now := time.Now().UTC()
res, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE office_routine_runs
SET status = ?, linked_task_id = ?, completed_at = ?
WHERE id = ? AND status = 'task_created'
`), status, linkedTaskID, now, runID)
if err != nil {
return false, err
}
rows, err := res.RowsAffected()
return rows > 0, err
}
Active means task_created only
func (r *Repository) GetActiveRunForFingerprint(ctx context.Context, routineID, fingerprint string) (*models.RoutineRun, error) {
var run models.RoutineRun
err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(`
SELECT * FROM office_routine_runs
WHERE routine_id = ? AND dispatch_fingerprint = ?
AND status = 'task_created'
ORDER BY created_at DESC LIMIT 1
`), routineID, fingerprint).StructScan(&run)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
return &run, nil
}
Narrow last_run_at touch
func (r *Repository) TouchRoutineLastRun(ctx context.Context, routineID string, at time.Time) error {
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE office_routines SET last_run_at = ?, updated_at = ? WHERE id = ?
`), at, time.Now().UTC(), routineID)
return err
}
TaskMoved now clears the routine gate
apps/backend/internal/office/service/event_subscribers.go ↗When a task enters Done or Cancelled, the service closes the linked heavy routine run before it wakes blockers and parents.
finalizeDone
func (s *Service) finalizeDone(ctx context.Context, data *TaskMovedData) error {
if s.routineRunSyncer != nil {
terminal := "done"
if strings.EqualFold(data.ToStepName, "cancelled") {
terminal = "cancelled"
}
if err := s.routineRunSyncer.SyncRunStatus(ctx, data.TaskID, terminal); err != nil {
s.logger.Warn("sync routine run status", zap.Error(err))
}
}
if err := s.queueBlockersResolvedRuns(ctx, data.TaskID); err != nil {
s.logger.Error("blocker resolution runs failed", zap.Error(err))
}
if data.ParentID != "" {
if err := s.queueChildrenCompletedRun(ctx, data.ParentID); err != nil {
s.logger.Warn("children completed run failed", zap.Error(err))
}
}
return nil
}
Wiring and wakeup idempotency handling
apps/backend/internal/backendapp/adapters_office.go ↗The adapter translates the SQLite idempotency conflict into the routines sentinel and adds a fail path for lightweight dispatches.
Adapter translation
func (a *routineWakeupAdapter) CreateWakeupRequest(ctx context.Context, req *officeroutines.WakeupRequest) error {
row := &officesqlite.WakeupRequest{
ID: req.ID, AgentProfileID: req.AgentProfileID, Source: req.Source,
Reason: req.Reason, Payload: req.Payload, RequestedAt: req.RequestedAt,
}
if req.IdempotencyKey != "" {
row.IdempotencyKey = sql.NullString{String: req.IdempotencyKey, Valid: true}
}
if err := a.repo.CreateWakeupRequest(ctx, row); err != nil {
if errors.Is(err, officesqlite.ErrWakeupIdempotencyConflict) {
return officeroutines.ErrWakeupAlreadyRequested
}
return err
}
return nil
}
func (a *routineWakeupAdapter) FailWakeupRequest(ctx context.Context, requestID, reason string) error {
return a.repo.MarkWakeupRequestFailed(ctx, requestID, reason)
}
Lightweight dispatch handles conflict and failure
if err := s.wakeup.CreateWakeupRequest(ctx, req); err != nil {
if errors.Is(err, ErrWakeupAlreadyRequested) {
return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusDone)
}
s.logger.Warn("create routine wakeup request", zap.String("routine", routine.Name), zap.Error(err))
return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusFailed)
}
if err := s.wakeup.Dispatch(ctx, req.ID); err != nil {
s.logger.Warn("dispatch routine wakeup request", zap.String("wakeup_id", req.ID), zap.Error(err))
if failErr := s.wakeup.FailWakeupRequest(ctx, req.ID, "dispatch failed"); failErr != nil {
s.logger.Warn("mark failed routine wakeup request", zap.Error(failErr))
}
if finalizeErr := s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusFailed); finalizeErr != nil {
return finalizeErr
}
return fmt.Errorf("dispatch routine wakeup request: %w", err)
}
return s.finalizeLightweightRun(ctx, run, models.RoutineRunStatusDone)
Wire syncer in composition root
routineSvc.SetWakeupEnqueuer(&routineWakeupAdapter{repo: repo, dispatcher: routineWakeupDispatcher})
routineSvc.SetWorkflowEnsurer(&workflowEnsurerAdapter{repo: taskRepo})
routineSvc.SetTaskCreator(&taskCreatorAdapter{taskSvc: services.Task})
if services.Office != nil {
services.Office.SetRoutineRunSyncer(routineSvc)
}
Terminal state now includes FAILED and new indexes
apps/backend/internal/office/repository/sqlite/blockers.go ↗FAILED was missing from the terminal check, so a failed task still blocked dependents and routine gates.
Shared constants and fix
const (
taskStateCompleted = "COMPLETED"
taskStateFailed = "FAILED"
taskStateCancelled = "CANCELLED"
)
func (r *Repository) IsTaskInTerminalStep(ctx context.Context, taskID string) (bool, error) {
var state string
err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(
`SELECT COALESCE(state, '') FROM tasks WHERE id = ?`), taskID).Scan(&state)
if err != nil {
return false, err
}
return state == taskStateCompleted || state == taskStateFailed || state == taskStateCancelled, nil
}
Partial indexes for hot paths
CREATE INDEX IF NOT EXISTS idx_office_routine_runs_active_fingerprint
ON office_routine_runs(routine_id, dispatch_fingerprint, created_at DESC)
WHERE status = 'task_created';
CREATE INDEX IF NOT EXISTS idx_office_routine_runs_linked_task
ON office_routine_runs(linked_task_id, created_at DESC)
WHERE linked_task_id != '';
Failed wakeup becomes terminal
func (r *Repository) MarkWakeupRequestFailed(ctx context.Context, id, reason string) error {
now := time.Now().UTC()
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE agent_wakeup_requests
SET status = ?, reason = ?, finished_at = ?
WHERE id = ? AND status = ?
`), WakeupStatusFailed, reason, now, id, WakeupStatusQueued)
return err
}
