fix(office): stop parent tasks from stalling when a reopened child finishes again ↗

Portable second-precision timestamp text

Normalizes timestamp rendering so SQLite whole-second text and Postgres microsecond text compare as equal strings.

func SecondPrecisionText(driver, expr string) string {
	if IsPostgres(driver) {
		return fmt.Sprintf("to_char(%s, 'YYYY-MM-DD HH24:MI:SS')", expr)
	}
	return fmt.Sprintf("strftime('%%Y-%%m-%%d %%H:%%M:%%S', %s)", expr)
}

Receipt now stores child generation

Stores the newest child updated_at at delivery time so a same-key reopen is still distinguishable.

type WakeReceipt struct {
	ParentTaskID        string    `db:"parent_task_id"`
	ChildSetKey         string    `db:"child_set_key"`
	DeliveredRunID      string    `db:"delivered_run_id"`
	DeliveryOperationID string    `db:"delivery_operation_id"`
	DeliveredAt         time.Time `db:"delivered_at"`
	ChildGeneration string `db:"child_generation"`
}

func (r *Repository) UpsertWakeReceiptTx(
	ctx context.Context, tx *sqlx.Tx,
	parentTaskID, childSetKey, deliveredRunID, deliveryOperationID, childGeneration 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, child_generation
		) 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,
			child_generation = excluded.child_generation
	`), parentTaskID, childSetKey, deliveredRunID, deliveryOperationID, deliveredAt, childGeneration)
	return err
}

func (r *Repository) GetChildSetKeyAndGeneration(ctx context.Context, parentTaskID string) (string, string, error) {
	var rows []childSetKeyRow
	if err := r.ro.SelectContext(ctx, &rows, r.ro.Rebind(`
		SELECT id, state FROM tasks WHERE parent_id = ? AND archived_at IS NULL ORDER BY id
	`), parentTaskID); err != nil {
		return "", "", err
	}
	driver := r.ro.DriverName()
	generationText := dialect.SecondPrecisionText(driver, "MAX(updated_at)")
	var generation sql.NullString
	if err := r.ro.GetContext(ctx, &generation, r.ro.Rebind(`
		SELECT `+generationText+` FROM tasks WHERE parent_id = ? AND archived_at IS NULL
	`), parentTaskID); err != nil {
		return "", "", err
	}
	return formatChildSetKey(rows), generation.String, nil
}

Sweep re-admits same-key reopen

Adds a generation equality check so a byte-identical child_set_key with a newer updated_at is swept again.

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"`
	NewestChildUpdatedAt string `db:"newest_child_updated_at"`
}

driver := r.ro.DriverName()
newestChildUpdatedAtText := dialect.SecondPrecisionText(driver, "MAX(c.updated_at)")
requestedAtText := dialect.SecondPrecisionText(driver, "w.requested_at")
// in CTE:
// SELECT strftime('%Y-%m-%d %H:%M:%S', MAX(c.updated_at)) AS newest_child_updated_at
// WHERE clause:
// AND (
//   r.child_set_key IS DISTINCT FROM 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, '') = '')
//   OR s.newest_child_updated_at != COALESCE(r.child_generation, '')
// )
// 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 to_char(w.requested_at,'YYYY-MM-DD HH24:MI:SS') >= s.newest_child_updated_at))
// )

Generation-aware operation id and receipt

Hashes parent, child ids, and generation so a new generation mints a new id and the engine does not swallow it as already applied.

func wakeOperationID(parentTaskID, childSetKey, generation string) string {
	childIDs := make([]string, 0)
	for _, child := range strings.Split(childSetKey, ",") {
		if child == "" { continue }
		if separator := strings.LastIndexByte(child, ':'); separator >= 0 {
			child = child[:separator]
		}
		childIDs = append(childIDs, child)
	}
	canonicalChildSet := strings.Join(childIDs, ",")
	sum := sha256.Sum256([]byte(parentTaskID + "\x00" + canonicalChildSet + "\x00" + generation))
	return fmt.Sprintf("task_children_completed:%s:%s", parentTaskID, hex.EncodeToString(sum[:]))
}

currentKey, err := svc.repo.GetChildSetKey(ctx, c.ParentTaskID)
if currentKey != c.ChildSetKey { return }
operationID := wakeOperationID(c.ParentTaskID, c.ChildSetKey, c.NewestChildUpdatedAt)
accepted, err := svc.dispatchEngineTriggerForRecovery(ctx, c.ParentTaskID, engine.TriggerOnChildrenCompleted, payload, operationID)
if !accepted { return }
h.recordReceipt(ctx, svc, c, operationID)
// recordReceipt persists c.NewestChildUpdatedAt as child_generation
if err := svc.repo.UpsertWakeReceiptTx(ctx, tx, c.ParentTaskID, c.ChildSetKey, "", operationID, c.NewestChildUpdatedAt, deliveredAt); err != nil { return }

Edge path uses same generation and operation id

Makes the edge-triggered dispatch and the reconciler derive the identical id for the same wave, so idempotency collapses the race.

func (s *Service) queueChildrenCompletedRun(ctx context.Context, parentID string) error {
	allDone, err := s.repo.AreAllChildrenTerminal(ctx, parentID)
	if err != nil || !allDone { return err }
	childSetKey, generation, err := s.repo.GetChildSetKeyAndGeneration(ctx, parentID)
	if err != nil { return fmt.Errorf("get child set key: %w", err) }
	children, _, err := s.repo.GetChildSummaries(ctx, parentID)
	key := wakeOperationID(parentID, childSetKey, generation)
	summaries := make([]engine.ChildSummary, 0, len(children))
	prsByTask := s.lookupChildPRLinks(ctx, children)
	for _, c := range children {
		summaries = append(summaries, engine.ChildSummary{TaskID: c.TaskID, Status: c.State, Summary: c.LastComment, PRLinks: prsByTask[c.TaskID]})
	}
	return s.dispatchEngineTrigger(ctx, parentID, engine.TriggerOnChildrenCompleted, engine.OnChildrenCompletedPayload{ChildSummaries: summaries}, key)
}

Schema adds child_generation column

Creates the column on fresh databases and migrates existing ones so the generation check has storage.

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,
	child_generation      TEXT NOT NULL DEFAULT ''
);

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 ''`)
	r.migrate.Apply("parent_child_wake_receipts.child_generation", `ALTER TABLE parent_child_wake_receipts ADD COLUMN child_generation TEXT NOT NULL DEFAULT ''`)
}