fix(office): recover lost children_completed wakes for stalled parents ↗

Stuck-parent query and receipt store

The query finds parents whose non-archived children are all terminal and whose wake has not been delivered for the current child set.

type StuckParentCandidate struct {
  ParentTaskID           string `db:"parent_task_id"`
  AssigneeAgentProfileID string `db:"assignee_agent_profile_id"`
  WorkflowStepID         string `db:"workflow_step_id"`
  ChildSetKey            string `db:"child_set_key"`
}

WITH stuck AS (
  SELECT
    p.id AS parent_task_id,
    COALESCE((SELECT GROUP_CONCAT(c.id || ':' || c.state, ',')
      FROM (SELECT id, state FROM tasks WHERE parent_id = p.id AND archived_at IS NULL ORDER BY id) c), '') AS child_set_key,
    (SELECT MAX(c.updated_at) FROM tasks c WHERE c.parent_id = p.id AND c.archived_at IS NULL) AS newest_child_updated_at
  FROM tasks p
  WHERE p.archived_at IS NULL
    AND p.is_ephemeral = 0
    AND p.state NOT IN ('COMPLETED', 'CANCELLED')
    AND EXISTS (SELECT 1 FROM tasks c WHERE c.parent_id = p.id AND c.archived_at IS NULL)
    AND NOT EXISTS (SELECT 1 FROM tasks c WHERE c.parent_id = p.id AND c.archived_at IS NULL AND c.state NOT IN ('COMPLETED', 'CANCELLED'))
)
SELECT s.parent_task_id, s.assignee_agent_profile_id, s.workflow_step_id, s.child_set_key
FROM stuck s
LEFT JOIN parent_child_wake_receipts r ON r.parent_task_id = s.parent_task_id
INNER JOIN agent_profiles ap ON ap.id = s.assignee_agent_profile_id
WHERE s.assignee_agent_profile_id != ''
  AND ap.status NOT IN ('paused', 'stopped', 'pending_approval')
  AND (r.child_set_key IS NOT s.child_set_key OR (NOT EXISTS (SELECT 1 FROM runs delivered WHERE delivered.id = r.delivered_run_id) AND COALESCE(r.delivery_operation_id, '') = ''))
  AND NOT EXISTS (SELECT 1 FROM runs w WHERE json_extract(w.payload, '$.task_id') = s.parent_task_id AND w.reason = ? AND (w.status IN ('queued', 'claimed') OR (w.status IN ('finished', 'failed', 'cancelled') AND w.requested_at >= s.newest_child_updated_at)))
ORDER BY s.parent_task_id LIMIT ?

func (r *Repository) UpsertWakeReceiptTx(ctx context.Context, tx *sqlx.Tx, parentTaskID, childSetKey, deliveredRunID, deliveryOperationID string, deliveredAt time.Time) error {
  _, err := tx.ExecContext(ctx, tx.Rebind(`
    INSERT INTO parent_child_wake_receipts (parent_task_id, child_set_key, delivered_run_id, delivery_operation_id, delivered_at)
    VALUES (?, ?, ?, ?, ?)
    ON CONFLICT (parent_task_id) DO UPDATE SET
      child_set_key = excluded.child_set_key,
      delivered_run_id = excluded.delivered_run_id,
      delivery_operation_id = excluded.delivery_operation_id,
      delivered_at = excluded.delivered_at
  `), parentTaskID, childSetKey, deliveredRunID, deliveryOperationID, deliveredAt)
  return err
}

Level-triggered reconciler

The handler sweeps for stuck parents each tick and re-delivers the wake through the workflow engine with a deterministic operation id.

func (h *ParentWakeReconciler) Tick(ctx context.Context) error {
  adopted, err := h.scheduler.svc.repo.HasOfficeAdoption(ctx)
  if err != nil {
    if ctx.Err() != nil {
      return nil
    }
    return fmt.Errorf("check Office adoption: %w", err)
  }
  if !adopted {
    return nil
  }
  h.reconcile(ctx)
  return nil
}

func (h *ParentWakeReconciler) reconcileOne(ctx context.Context, svc *Service, c sqlite.StuckParentCandidate) {
  if err := svc.guardAgentStatus(ctx, c.AssigneeAgentProfileID); err != nil {
    svc.recordWakeAssigneeUnresolved(c.ParentTaskID, err.Error())
    return
  }
  payload, err := h.buildPayload(ctx, svc, c.ParentTaskID)
  if err != nil {
    return
  }
  currentKey, err := svc.repo.GetChildSetKey(ctx, c.ParentTaskID)
  if err != nil || currentKey != c.ChildSetKey {
    return
  }
  operationID := wakeOperationID(c.ParentTaskID, c.ChildSetKey)
  accepted, err := svc.dispatchEngineTriggerForRecovery(ctx, c.ParentTaskID, engine.TriggerOnChildrenCompleted, payload, operationID)
  if err != nil || !accepted {
    return
  }
  h.recordReceipt(ctx, svc, c, operationID)
}

func wakeOperationID(parentTaskID, childSetKey string) string {
  sum := sha256.Sum256([]byte(parentTaskID + "\x00" + childSetKey))
  return fmt.Sprintf("task_children_completed:%s:%s", parentTaskID, hex.EncodeToString(sum[:]))
}

Cron wiring

The shared cron loop now includes the parent-wake reconciler as a fifth handler.

	repos *Repositories,
	dispatcher *officeenginedispatcher.Dispatcher,
	routineSvc *officeroutines.RoutineService,
	officeRecovery schedulercron.Handler,
	parentWakeReconciler schedulercron.Handler,
	log *logger.Logger,
) *schedulercron.Loop {
	routines := schedulercron.NewRoutinesHandler(routineTicker, nil, log)
	loop := schedulercron.NewLoop(schedulercron.DefaultTickInterval, log,
		heartbeat, budget, routines, officeRecovery)
		heartbeat, budget, routines, officeRecovery, parentWakeReconciler)
	loop.Start(ctx)

var officeRecovery schedulercron.Handler
var parentWakeReconciler schedulercron.Handler
if services.Office != nil {
  officeRecovery = officeservice.NewOfficeRecoveryHandler(orchScheduler)
  parentWakeReconciler = officeservice.NewParentWakeReconciler(orchScheduler)
}
cronLoop := startCronScheduler(ctx, repos, engineDispatcher, officeRoutines, officeRecovery, parentWakeReconciler, log)

Receipt table and indexes

The new table stores one receipt per parent for the last delivered child set, so steady state costs one lookup per tick.

CREATE TABLE IF NOT EXISTS parent_child_wake_receipts (
  parent_task_id        TEXT PRIMARY KEY,
  child_set_key         TEXT NOT NULL,
  delivered_run_id      TEXT NOT NULL DEFAULT '',
  delivery_operation_id TEXT NOT NULL DEFAULT '',
  delivered_at          TIMESTAMP NOT NULL
);

func (r *Repository) migrateParentWakeIndexes() {
  r.migrate.Apply("idx_tasks_parent_id", `CREATE INDEX IF NOT EXISTS idx_tasks_parent_id ON tasks(parent_id)`)
}
func (r *Repository) migrateParentWakeReceiptColumns() {
  r.migrate.Apply("parent_child_wake_receipts.delivery_operation_id", `ALTER TABLE parent_child_wake_receipts ADD COLUMN delivery_operation_id TEXT NOT NULL DEFAULT ''`)
}

Observability and race guards

Counters and logs make the sweep observable and guard the receipt race between SELECT and write.

var (
  parentWakeCandidatesTotal         = expvar.NewInt("parent_wake_candidates_total")
  parentWakeEmittedTotal            = expvar.NewInt("parent_wake_emitted_total")
  parentWakeAssigneeUnresolvedTotal = expvar.NewInt("parent_wake_assignee_unresolved_total")
)
func (s *Service) recordWakeCandidate(parentTaskID string) {
  parentWakeCandidatesTotal.Add(1)
}
func (s *Service) recordWakeEmitted(parentTaskID, operationID string) {
  parentWakeEmittedTotal.Add(1)
}

func (r *Repository) GetTaskAssigneeTx(ctx context.Context, tx *sqlx.Tx, taskID string) (string, error) {
  var assignee string
  err := tx.QueryRowxContext(ctx, tx.Rebind(`SELECT `+RunnerProjection("tasks")+` FROM tasks WHERE id = ?`), taskID).Scan(&assignee)
  return assignee, err
}
// recordReceipt re-reads child set inside the tx before upsert
func (h *ParentWakeReconciler) recordReceipt(ctx context.Context, svc *Service, c sqlite.StuckParentCandidate, operationID string) {
  tx, _ := svc.repo.Writer().BeginTxx(ctx, nil)
  defer func() { _ = tx.Rollback() }()
  currentKey, _ := svc.repo.GetChildSetKeyTx(ctx, tx, c.ParentTaskID)
  if currentKey != c.ChildSetKey {
    return
  }
  svc.repo.UpsertWakeReceiptTx(ctx, tx, c.ParentTaskID, c.ChildSetKey, "", operationID, time.Now().UTC())
  tx.Commit()
}

func (s *Service) dispatchEngineTriggerForRecovery(ctx context.Context, taskID string, trigger engine.Trigger, payload any, opID string) (bool, error) {
  if s.engineDispatcher == nil {
    return false, nil
  }
  if d, ok := s.engineDispatcher.(interface{ HandleTriggerHandled(context.Context, string, engine.Trigger, any, string) (bool, error) }); ok {
    _, err := d.HandleTriggerHandled(ctx, taskID, trigger, payload, opID)
    if errors.Is(err, shared.ErrEngineNoSession) {
      return false, nil
    }
    return err == nil, err
  }
  return true, s.dispatchEngineTrigger(ctx, taskID, trigger, payload, opID)
}