fix(office): stop the stuck-parent wake sweep from failing every tick on Postgres ↗

Dialect-aware child-set aggregation

Returns STRING_AGG with ORDER BY on Postgres and GROUP_CONCAT on SQLite so the SQL key matches the Go key byte for byte.

// childSetKeyAggregate renders the "id:state" concatenation ListStuckParents
// compares against a stored receipt. Its output must match formatChildSetKey
// byte for byte: receipts are written from the Go form and compared against
// this SQL form, so any difference in separator or ordering makes every
// receipt look stale and re-wakes the parent on every tick. Postgres does not
// inherit input ordering from the subquery's ORDER BY, so the ordering is
// restated inside the aggregate.
func childSetKeyAggregate(driver string) string {
	if dialect.IsPostgres(driver) {
		return `STRING_AGG(c.id || ':' || c.state, ',' ORDER BY c.id)`
	}
	return `GROUP_CONCAT(c.id || ':' || c.state, ',')`
}

				COALESCE((
					SELECT GROUP_CONCAT(c.id || ':' || c.state, ',')
					SELECT `+childSetKeyAggregate(driver)+`
					FROM (
						SELECT id, state FROM tasks
						WHERE parent_id = p.id AND archived_at IS NULL
						ORDER BY id
					) c
				), '') AS child_set_key,

Postgres-safe receipt and run filters

Wires driver-aware SQL for receipt comparison and JSON extraction so the sweep parses on Postgres.

func (r *Repository) ListStuckParents(ctx context.Context, reason string, limit int) ([]StuckParentCandidate, error) {
	driver := r.ro.DriverName()
	var rows []StuckParentCandidate
	err := r.ro.SelectContext(ctx, &rows, r.ro.Rebind(`

		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
		      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, '') = ''
		      )
		  )

		  AND NOT EXISTS (
		      SELECT 1 FROM runs w
		      WHERE json_extract(w.payload, '$.task_id') = s.parent_task_id
		      WHERE `+dialect.JSONExtract(driver, "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
		            )
		        )
		  )

Postgres test suite for the sweep

Proves the query runs on Postgres, the SQL aggregate matches Go, and covered parents are excluded.

func TestPostgresListStuckParents(t *testing.T) {
	repo, ctx := newPostgresWakeRepo(t)
	wantKey := seedPostgresStuckParent(t, ctx, repo, "pg-parent-1", "pg-ws-1")
	rows, err := repo.ListStuckParents(ctx, "task_children_completed", 5)
	if err != nil {
		t.Fatalf("ListStuckParents: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("len(rows) = %d, want 1", len(rows))
	}
	if rows[0].ChildSetKey != wantKey {
		t.Errorf("ChildSetKey = %q, want %q", rows[0].ChildSetKey, wantKey)
	}
}

func TestPostgresListStuckParentsChildSetKeyMatchesGo(t *testing.T) {
	repo, ctx := newPostgresWakeRepo(t)
	seedPostgresTask(t, ctx, repo, "pg-parent-2", "pg-ws-2")
	execPostgres(t, ctx, repo,
		`UPDATE tasks SET project_id = 'office-project' WHERE id = ?`, "pg-parent-2")
	for _, childID := range []string{"pg-parent-2-child-c", "pg-parent-2-child-a", "pg-parent-2-child-b"} {
		seedPostgresTask(t, ctx, repo, childID, "pg-ws-2")
		execPostgres(t, ctx, repo,
			`UPDATE tasks SET parent_id = ?, state = 'COMPLETED' WHERE id = ?`, "pg-parent-2", childID)
	}
	seedPostgresRunner(t, ctx, repo, "pg-parent-2")
	goKey, _ := repo.GetChildSetKey(ctx, "pg-parent-2")
	rows, _ := repo.ListStuckParents(ctx, "task_children_completed", 5)
	if rows[0].ChildSetKey != goKey {
		t.Errorf("ChildSetKey = %q, want %q (GetChildSetKey)", rows[0].ChildSetKey, goKey)
	}
}

func TestPostgresListStuckParentsExcludesCoveredParent(t *testing.T) {
	repo, ctx := newPostgresWakeRepo(t)
	key := seedPostgresStuckParent(t, ctx, repo, "pg-parent-3", "pg-ws-3")
	execPostgres(t, ctx, repo, `
		INSERT INTO parent_child_wake_receipts (parent_task_id, child_set_key, delivery_operation_id, delivered_at)
		VALUES (?, ?, 'op-1', ?)
	`, "pg-parent-3", key, time.Now().UTC())
	rows, _ := repo.ListStuckParents(ctx, "task_children_completed", 5)
	if len(rows) != 0 {
		t.Fatalf("len(rows) = %d, want 0 — receipt already covers this child set", len(rows))
	}
}