exercise

var BUILDER = Exerciser{Name: "builder", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, _ Conn) error {
	rec.Note(QF("For"))
	n := atomic.AddInt64(&builderSeq, 1)

	owner := &domain.Account{Email: fmt.Sprintf("builder%d@superapp.test", n), Name: "b", Role: "member", Active: true}
	if err := quark.For[domain.Account](ctx, client).Create(owner); err != nil {
		return fmt.Errorf("seed owner: %w", err)
	}
	proj := &domain.Project{OwnerID: owner.ID, Name: "builder-proj", Status: "active"}
	if err := quark.For[domain.Project](ctx, client).Create(proj); err != nil {
		return fmt.Errorf("seed project: %w", err)
	}
	for p := 1; p <= 5; p++ {
		t := &domain.Task{ProjectID: proj.ID, Title: fmt.Sprintf("t%d", p), Priority: p, Done: p%2 == 0}
		if err := quark.For[domain.Task](ctx, client).Create(t); err != nil {
			return fmt.Errorf("seed task: %w", err)
		}
	}
	scope := func() *quark.Query[domain.Task] {
		return quark.For[domain.Task](ctx, client).Where("project_id", "=", proj.ID)
	}

	if sum, err := quark.For[domain.Task](rec.Mark(ctx, QM("Sum")), client).Where("project_id", "=", proj.ID).Sum("priority"); err != nil || sum != 15 {
		return fmt.Errorf("Sum(priority)=%v err=%v, esperaba 15", sum, err)
	}
	if avg, err := quark.For[domain.Task](rec.Mark(ctx, QM("Avg")), client).Where("project_id", "=", proj.ID).Avg("priority"); err != nil || avg != 3 {
		return fmt.Errorf("Avg=%v err=%v, esperaba 3", avg, err)
	}
	if mn, err := quark.For[domain.Task](rec.Mark(ctx, QM("Min")), client).Where("project_id", "=", proj.ID).Min("priority"); err != nil || mn != 1 {
		return fmt.Errorf("Min=%v err=%v, esperaba 1", mn, err)
	}
	if mx, err := quark.For[domain.Task](rec.Mark(ctx, QM("Max")), client).Where("project_id", "=", proj.ID).Max("priority"); err != nil || mx != 5 {
		return fmt.Errorf("Max=%v err=%v, esperaba 5", mx, err)
	}

	rec.Note(QM("Select"), QM("GroupBy"), QM("Having"))
	groups, err := scope().Select("done").GroupBy("done").Having("done", "=", true).List()
	if err != nil {
		return fmt.Errorf("Select/GroupBy/Having: %w", err)
	}
	if len(groups) == 0 {
		return fmt.Errorf("GroupBy no devolvió grupos")
	}

	if c, err := quark.For[domain.Task](rec.Mark(ctx, QM("WhereIn")), client).Where("project_id", "=", proj.ID).WhereIn("priority", []any{1, 2, 3}).Count(); err != nil || c != 3 {
		return fmt.Errorf("WhereIn count=%d err=%v, esperaba 3", c, err)
	}

	if c, err := scope().Where("priority", "=", 5).Or(func(q *quark.Query[domain.Task]) *quark.Query[domain.Task] {
		return q.Where("project_id", "=", proj.ID).Where("priority", "=", 1)
	}).Count(); err != nil || c < 1 {
		return fmt.Errorf("Or count=%d err=%v, esperaba >=1", c, err)
	}
	rec.Note(QM("Or"))

	rec.Note(QM("OrderBy"), QM("Offset"))
	top, err := scope().OrderBy("priority", "DESC").Offset(1).Limit(1).List()
	if err != nil || len(top) != 1 || top[0].Priority != 4 {
		return fmt.Errorf("OrderBy/Offset: %+v err=%v, esperaba priority=4", top, err)
	}

	rec.Note(QM("Distinct"))
	if _, err := scope().Distinct().List(); err != nil {
		return fmt.Errorf("Distinct: %w", err)
	}

	want := top[0].ID
	found, err := quark.For[domain.Task](rec.Mark(ctx, QM("Find")), client).Find(want)
	if err != nil || found.ID != want {
		return fmt.Errorf("Find(%d)=%+v err=%v", want, found, err)
	}

	count := 0
	if err := scope().Iter(func(domain.Task) error { count++; return nil }); err != nil {
		return fmt.Errorf("Iter: %w", err)
	}
	if count != 5 {
		return fmt.Errorf("Iter contó %d, esperaba 5", count)
	}
	rec.Note(QM("Iter"))

	cur, err := scope().Cursor()
	if err != nil {
		return fmt.Errorf("Cursor: %w", err)
	}
	cn := 0
	for cur.Next() {
		var t domain.Task
		if err := cur.Scan(&t); err != nil {
			_ = cur.Close()
			return fmt.Errorf("Cursor.Scan: %w", err)
		}
		cn++
	}
	if err := cur.Err(); err != nil {
		_ = cur.Close()
		return fmt.Errorf("Cursor.Err: %w", err)
	}
	if err := cur.Close(); err != nil {
		return fmt.Errorf("Cursor.Close: %w", err)
	}
	if cn != 5 {
		return fmt.Errorf("Cursor contó %d, esperaba 5", cn)
	}
	rec.Note(QM("Cursor"))

	page, err := scope().Paginate(2, 0)
	if err != nil {
		return fmt.Errorf("Paginate: %w", err)
	}
	if page.Total != 5 || len(page.Items) != 2 {
		return fmt.Errorf("Paginate total=%d items=%d, esperaba 5/2", page.Total, len(page.Items))
	}
	rec.Note(QM("Paginate"))

	return nil
}}

var BUILDERADV = Exerciser{Name: "builder-advanced", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, conn Conn) error {
	mail := func(i int) string { return fmt.Sprintf("badv-%02d@superapp.test", i) }
	scopedAcc := func(c context.Context) *quark.Query[domain.Account] {
		return quark.For[domain.Account](c, client).Where("email", "LIKE", "badv-%")
	}

	accs := make([]*domain.Account, 4)
	for i := range accs {
		accs[i] = &domain.Account{Email: mail(i), Name: fmt.Sprintf("badv%d", i), Role: "member", Active: true}
	}
	accs[3].Role = "admin"
	if err := quark.For[domain.Account](ctx, client).CreateBatch(accs); err != nil {
		return fmt.Errorf("seed cuentas: %w", err)
	}
	cleanup := func() {
		cctx := context.Background()
		_, _ = quark.For[domain.Project](cctx, client).Where("name", "LIKE", "badv-%").DeleteBy()
		rows, _ := quark.For[domain.Account](cctx, client).Where("email", "LIKE", "badv-%").WithTrashed().List()
		for i := range rows {
			_, _ = quark.For[domain.Account](cctx, client).HardDelete(&rows[i])
		}
	}
	cleanupRegistered := false
	defer func() {
		if !cleanupRegistered {
			cleanup()
		}
	}()
	for _, p := range []*domain.Project{
		{OwnerID: accs[0].ID, Name: "badv-p0", Status: "draft"},
		{OwnerID: accs[1].ID, Name: "badv-p1", Status: "active"},
	} {
		if err := quark.For[domain.Project](ctx, client).Create(p); err != nil {
			return fmt.Errorf("seed proyecto: %w", err)
		}
	}

	if n, err := scopedAcc(rec.Mark(ctx, QM("WhereBetween"))).WhereBetween("id", accs[0].ID, accs[3].ID).Count(); err != nil || n != 4 {
		return fmt.Errorf("WhereBetween: n=%d err=%v, esperaba 4", n, err)
	}
	if n, err := scopedAcc(rec.Mark(ctx, QM("WhereNot"))).WhereNot("role", "=", "admin").Count(); err != nil || n != 3 {
		return fmt.Errorf("WhereNot: n=%d err=%v, esperaba 3", n, err)
	}

	rec.Note(QF("NewTypedColumn"), QF("Predicate"))
	role := quark.NewTypedColumn[string]("role")
	if n, err := scopedAcc(rec.Mark(ctx, QM("WhereP"))).WhereP(role.Eq("admin")).Count(); err != nil || n != 1 {
		return fmt.Errorf("WhereP: n=%d err=%v, esperaba 1", n, err)
	}

	rec.Note(QF("Col"), QF("Lit"), QF("Eq"))
	if n, err := scopedAcc(rec.Mark(ctx, QM("WhereExpr"))).WhereExpr(quark.Eq(quark.Col("role"), quark.Lit("admin"))).Count(); err != nil || n != 1 {
		return fmt.Errorf("WhereExpr: n=%d err=%v, esperaba 1", n, err)
	}

	rec.Note(QF("Scope"))
	admins := func(q *quark.Query[domain.Account]) *quark.Query[domain.Account] {
		return q.Where("role", "=", "admin")
	}
	if n, err := scopedAcc(rec.Mark(ctx, QM("Apply"))).Apply(admins).Count(); err != nil || n != 1 {
		return fmt.Errorf("Apply: n=%d err=%v, esperaba 1", n, err)
	}

	accs[2].Active = false
	if rows, err := quark.For[domain.Account](rec.Mark(ctx, QM("UpdateFields")), client).UpdateFields(accs[2], "active"); err != nil || rows != 1 {
		return fmt.Errorf("UpdateFields: rows=%d err=%v", rows, err)
	}
	if got, err := scopedAcc(ctx).Where("id", "=", accs[2].ID).First(); err != nil || got.Active {
		return fmt.Errorf("UpdateFields no escribió el zero-value Active=false (got %+v, err=%v)", got.Active, err)
	}

	rec.Note(QF("(*TrackedQuery[T]).Find"), QF("(*Tracked[T]).Save"), QF("(*Tracked[T]).Changed"))
	tracked, err := quark.For[domain.Account](rec.Mark(ctx, QM("Track")), client).Track().Find(accs[1].ID)
	if err != nil {
		return fmt.Errorf("Track.Find: %w", err)
	}
	tracked.Entity.Name = "badv-tracked"
	if len(tracked.Changed()) == 0 {
		return fmt.Errorf("Tracked.Changed vacío tras mutar")
	}
	if rows, err := tracked.Save(ctx); err != nil || rows != 1 {
		return fmt.Errorf("Tracked.Save: rows=%d err=%v", rows, err)
	}

	if rows, err := tracked.Save(ctx); err != nil || rows != 1 {
		return fmt.Errorf("re-Save: rows=%d err=%v (el hook garantiza delta)", rows, err)
	}

	rec.Note(QF("(*JoinBuilder[T]).On"))
	withProj, err := quark.For[domain.Account](rec.Mark(ctx, QM("LeftJoin")), client).
		Where("email", "LIKE", "badv-%").
		LeftJoin("projects").On("projects.owner_id", "=", "accounts.id").
		Limit(10).List()
	if err != nil {
		return fmt.Errorf("LeftJoin: %w", err)
	}
	if len(withProj) != 4 {
		return fmt.Errorf("LeftJoin: %d filas, esperaba 4 (conserva cuentas sin proyecto; a0/a1 tienen 1 cada una)", len(withProj))
	}

	rec.Note(QF("(*JoinBuilder[T]).OnRaw"))
	rj, err := quark.For[domain.Account](rec.Mark(ctx, QM("RightJoin")), client).
		Where("email", "LIKE", "badv-%").
		RightJoin("projects").OnRaw("projects.owner_id = accounts.id").
		Limit(10).List()
	if err != nil {
		return fmt.Errorf("RightJoin: %w", err)
	}
	if len(rj) != 2 {
		return fmt.Errorf("RightJoin: %d filas, esperaba 2 (los proyectos badv con su owner)", len(rj))
	}

	if _, err := scopedAcc(rec.Mark(ctx, QM("HavingAggregate"))).
		Select("role").GroupBy("role").HavingAggregate("COUNT", "id", ">=", 1).Limit(10).List(); err != nil {
		return fmt.Errorf("HavingAggregate: %w", err)
	}
	rec.Note(QF("Gte"))
	if _, err := scopedAcc(rec.Mark(ctx, QM("HavingExpr"))).
		Select("role").GroupBy("role").HavingExpr(quark.Gte(quark.Col("role"), quark.Lit("a"))).Limit(10).List(); err != nil {
		return fmt.Errorf("HavingExpr: %w", err)
	}
	rec.Note(QF("NewWindow"), QF("Over"), QF("RowNumber"), QF("(*Window).PartitionBy"), QF("(*Window).OrderBy"))
	w := quark.NewWindow().PartitionBy(quark.Col("role")).OrderBy(quark.Col("id"), false)
	if _, err := scopedAcc(rec.Mark(ctx, QM("SelectExpr"))).
		Select("id").SelectExpr("rn", quark.Over(quark.RowNumber(), w)).Limit(10).List(); err != nil {
		return fmt.Errorf("SelectExpr+window: %w", err)
	}

	rec.Note(QF("Subquery"))
	sub, err := quark.For[domain.Account](rec.Mark(ctx, QM("AsSubquery")), client).
		Select("id").Where("email", "LIKE", "badv-%").AsSubquery()
	if err != nil {
		return fmt.Errorf("AsSubquery: %w", err)
	}
	_ = quark.For[domain.Account](rec.Mark(ctx, QM("MustAsSubquery")), client).
		Select("id").Where("role", "=", "admin").MustAsSubquery()

	cte, err := quark.For[domain.Account](rec.Mark(ctx, QM("With")), client).
		With("badv_ids", sub).
		Join("badv_ids").On("accounts.id", "=", "badv_ids.id").
		Limit(10).List()
	if err != nil {
		return fmt.Errorf("With (CTE): %w", err)
	}
	if len(cte) != 4 {
		return fmt.Errorf("CTE: %d filas, esperaba 4", len(cte))
	}

	if _, err := quark.For[domain.Account](rec.Mark(ctx, QM("WithRecursive")), client).
		WithRecursive("badv_rec", sub).
		Join("badv_rec").On("accounts.id", "=", "badv_rec.id").
		Limit(10).List(); err != nil {
		return fmt.Errorf("WithRecursive: %w", err)
	}

	if _, err := scopedAcc(rec.Mark(ctx, QM("WhereSubquery"))).
		WhereSubquery("id", "IN", "SELECT id FROM accounts").Count(); !errors.Is(err, quark.ErrInvalidQuery) {
		return fmt.Errorf("WhereSubquery sin AllowRawQueries: esperaba ErrInvalidQuery, got %v", err)
	}

	lraw := quark.DefaultLimits()
	lraw.AllowRawQueries = true
	rawClient, err := quark.New(conn.Driver, conn.DSN, quark.WithLimits(lraw))
	if err != nil {
		return fmt.Errorf("client raw: %w", err)
	}
	defer rawClient.Close()
	if n, err := quark.For[domain.Account](ctx, rawClient).Where("email", "LIKE", "badv-%").
		WhereSubquery("id", "IN", "SELECT id FROM accounts WHERE role = 'admin'").Count(); err != nil || n != 1 {
		return fmt.Errorf("WhereSubquery con flag: n=%d err=%v, esperaba 1", n, err)
	}

	admins2 := quark.For[domain.Account](ctx, client).Select("email").Where("email", "LIKE", "badv-%").Where("role", "=", "admin")
	members := quark.For[domain.Account](ctx, client).Select("email").Where("email", "LIKE", "badv-%").Where("role", "=", "member")
	all := quark.For[domain.Account](ctx, client).Select("email").Where("email", "LIKE", "badv-%")
	if rows, err := admins2.Union(members).Limit(10).List(); err != nil {
		return fmt.Errorf("Union: %w", err)
	} else if len(rows) != 4 {
		return fmt.Errorf("Union: %d filas, esperaba 4 (1 admin + 3 members)", len(rows))
	}
	rec.Note(QM("Union"))
	if rows, err := admins2.UnionAll(admins2).Limit(10).List(); err != nil {
		return fmt.Errorf("UnionAll: %w", err)
	} else if len(rows) != 2 {
		return fmt.Errorf("UnionAll: %d filas, esperaba 2 (duplicados conservados)", len(rows))
	}
	rec.Note(QM("UnionAll"))

	if control.Supports(control.FeatIntersectExcept, conn.Engine) {
		if rows, err := all.Intersect(admins2).Limit(10).List(); err != nil {
			return fmt.Errorf("Intersect: %w", err)
		} else if len(rows) != 1 {
			return fmt.Errorf("Intersect: %d filas, esperaba 1", len(rows))
		}
		if rows, err := all.Except(admins2).Limit(10).List(); err != nil {
			return fmt.Errorf("Except: %w", err)
		} else if len(rows) != 3 {
			return fmt.Errorf("Except: %d filas, esperaba 3", len(rows))
		}
	} else {
		if _, err := all.Intersect(admins2).Limit(10).List(); !errors.Is(err, quark.ErrUnsupportedFeature) {
			return fmt.Errorf("Intersect en %s: esperaba ErrUnsupportedFeature, got %v", conn.Engine, err)
		}
		if _, err := all.Except(admins2).Limit(10).List(); !errors.Is(err, quark.ErrUnsupportedFeature) {
			return fmt.Errorf("Except en %s: esperaba ErrUnsupportedFeature, got %v", conn.Engine, err)
		}
	}
	rec.Note(QM("Intersect"), QM("Except"))

	if control.Supports(control.FeatSkipLocked, conn.Engine) {

		lockLimit := func(q *quark.Query[domain.Account]) *quark.Query[domain.Account] {
			if conn.Engine == control.Oracle {
				return q
			}
			return q.Limit(2)
		}

		if err := client.Tx(ctx, func(tx *quark.Tx) error {
			if _, err := lockLimit(quark.ForTx[domain.Account](rec.Mark(ctx, QM("ForUpdate")), tx).
				Where("email", "LIKE", "badv-%").ForUpdate()).List(); err != nil {
				return fmt.Errorf("ForUpdate: %w", err)
			}
			if _, err := lockLimit(quark.ForTx[domain.Account](rec.Mark(ctx, QM("SkipLocked")), tx).
				Where("email", "LIKE", "badv-%").ForUpdate().SkipLocked()).List(); err != nil {
				return fmt.Errorf("SkipLocked: %w", err)
			}
			if _, err := lockLimit(quark.ForTx[domain.Account](rec.Mark(ctx, QM("NoWait")), tx).
				Where("email", "LIKE", "badv-%").ForUpdate().NoWait()).List(); err != nil {

				if !errors.Is(err, quark.ErrUnsupportedFeature) {
					return fmt.Errorf("NoWait: %w", err)
				}
			}
			if _, err := lockLimit(quark.ForTx[domain.Account](rec.Mark(ctx, QM("ForShare")), tx).
				Where("email", "LIKE", "badv-%").ForShare()).List(); err != nil {

				if errors.Is(err, quark.ErrUnsupportedFeature) {
					return nil
				}
				return fmt.Errorf("ForShare: %w", err)
			}
			return nil
		}); err != nil {
			return fmt.Errorf("locking tx: %w", err)
		}
	} else {

		if _, err := scopedAcc(rec.Mark(ctx, QM("ForUpdate"))).ForUpdate().Limit(1).List(); !errors.Is(err, quark.ErrUnsupportedFeature) {
			return fmt.Errorf("ForUpdate en %s: esperaba ErrUnsupportedFeature, got %v", conn.Engine, err)
		}
		rec.Note(QM("ForShare"), QM("SkipLocked"), QM("NoWait"))
	}

	victim, err := scopedAcc(ctx).Where("id", "=", accs[0].ID).First()
	if err != nil {
		return fmt.Errorf("first pre-soft-delete: %w", err)
	}
	if _, err := quark.For[domain.Account](ctx, client).Delete(&victim); err != nil {
		return fmt.Errorf("soft-delete: %w", err)
	}
	if n, _ := scopedAcc(ctx).Count(); n != 3 {
		return fmt.Errorf("default scope: %d, esperaba 3 (excluye la borrada)", n)
	}
	if n, _ := scopedAcc(rec.Mark(ctx, QM("WithTrashed"))).WithTrashed().Count(); n != 4 {
		return fmt.Errorf("WithTrashed: esperaba 4")
	}
	if n, _ := scopedAcc(rec.Mark(ctx, QM("OnlyTrashed"))).OnlyTrashed().Count(); n != 1 {
		return fmt.Errorf("OnlyTrashed: esperaba 1")
	}
	if n, _ := scopedAcc(rec.Mark(ctx, QM("Unscoped"))).Unscoped().Count(); n != 4 {
		return fmt.Errorf("Unscoped: esperaba 4")
	}

	trashed, err := scopedAcc(ctx).OnlyTrashed().First()
	if err != nil {
		return fmt.Errorf("first trashed: %w", err)
	}
	if rows, err := quark.For[domain.Account](rec.Mark(ctx, QM("Restore")), client).Restore(&trashed); err != nil || rows != 1 {
		return fmt.Errorf("Restore: rows=%d err=%v", rows, err)
	}
	if n, _ := scopedAcc(ctx).Count(); n != 4 {
		return fmt.Errorf("post-Restore: esperaba 4 visibles")
	}

	up := &domain.Account{Email: mail(0), Name: "badv-upserted", Role: "member", Active: true}
	if err := quark.For[domain.Account](rec.Mark(ctx, QM("Upsert")), client).Upsert(up, []string{"email"}, []string{"name"}); err != nil {
		return fmt.Errorf("Upsert: %w", err)
	}
	if got, err := scopedAcc(ctx).Where("email", "=", mail(0)).First(); err != nil || got.Name != "badv-upserted" {
		return fmt.Errorf("Upsert no actualizó por conflicto: name=%q err=%v", got.Name, err)
	}

	batch := []*domain.Account{
		{Email: mail(1), Name: "badv-ub1", Role: "member", Active: true},

		{Email: mail(9), Name: "badv-new", Role: "viewer", Active: true},
	}
	if err := quark.For[domain.Account](rec.Mark(ctx, QM("UpsertBatch")), client).UpsertBatch(batch, []string{"email"}, []string{"name"}); err != nil {
		return fmt.Errorf("UpsertBatch: %w", err)
	}
	if n, _ := scopedAcc(ctx).Count(); n != 5 {
		return fmt.Errorf("post-UpsertBatch: esperaba 5 (1 update + 1 insert)")
	}

	rows, err := scopedAcc(ctx).Where("role", "=", "member").OrderBy("id", "ASC").Limit(2).List()
	if err != nil || len(rows) != 2 {
		return fmt.Errorf("pre-UpdateBatch: %d filas err=%v", len(rows), err)
	}
	ub := []*domain.Account{&rows[0], &rows[1]}
	ub[0].Name, ub[1].Name = "badv-batch0", "badv-batch1"
	if err := quark.For[domain.Account](rec.Mark(ctx, QM("UpdateBatch")), client).UpdateBatch(ub); err != nil {
		return fmt.Errorf("UpdateBatch: %w", err)
	}

	if n, err := quark.For[domain.Account](rec.Mark(ctx, QM("DeleteBatch")), client).DeleteBatch([]any{ub[0].ID, ub[1].ID}); err != nil || n != 2 {
		return fmt.Errorf("DeleteBatch: n=%d err=%v", n, err)
	}
	if n, err := scopedAcc(rec.Mark(ctx, QM("DeleteBy"))).Where("role", "=", "viewer").DeleteBy(); err != nil || n != 1 {
		return fmt.Errorf("DeleteBy: n=%d err=%v", n, err)
	}

	all2, err := scopedAcc(ctx).WithTrashed().List()
	if err != nil {
		return fmt.Errorf("list final: %w", err)
	}
	for i := range all2 {
		if _, err := quark.For[domain.Account](rec.Mark(ctx, QM("HardDelete")), client).HardDelete(&all2[i]); err != nil {
			return fmt.Errorf("HardDelete: %w", err)
		}
	}
	if n, _ := scopedAcc(ctx).Unscoped().Count(); n != 0 {
		return fmt.Errorf("post-HardDelete: quedan %d filas badv-", n)
	}
	if _, err := quark.For[domain.Project](ctx, client).Where("name", "LIKE", "badv-%").DeleteBy(); err != nil {
		return fmt.Errorf("cleanup proyectos: %w", err)
	}
	cleanupRegistered = true
	return nil
}}

var CACHE = Exerciser{Name: "cache", Fn: runCache}

var CRUD = Exerciser{Name: "crud", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, _ Conn) error {
	rec.Note(QF("For"))
	email := fmt.Sprintf("crud%d@superapp.test", atomic.AddInt64(&crudSeq, 1))

	a := &domain.Account{Email: email, Name: "crud", Role: "member", Active: true}
	if err := quark.For[domain.Account](rec.Mark(ctx, QM("Create")), client).Create(a); err != nil {
		return fmt.Errorf("create: %w", err)
	}
	if a.ID == 0 {
		return fmt.Errorf("create no asignó ID")
	}

	rec.Note(QM("Where"))
	got, err := quark.For[domain.Account](rec.Mark(ctx, QM("First")), client).Where("email", "=", email).First()
	if err != nil {
		return fmt.Errorf("first: %w", err)
	}
	if got.ID != a.ID || got.Email != email {
		return fmt.Errorf("first round-trip roto: got id=%d email=%q", got.ID, got.Email)
	}

	n, err := quark.For[domain.Account](rec.Mark(ctx, QM("Count")), client).Where("email", "=", email).Count()
	if err != nil {
		return fmt.Errorf("count: %w", err)
	}
	if n != 1 {
		return fmt.Errorf("count=%d, esperaba 1", n)
	}

	got.Name = "crud-updated"
	rows, err := quark.For[domain.Account](rec.Mark(ctx, QM("Update")), client).Update(&got)
	if err != nil {
		return fmt.Errorf("update: %w", err)
	}
	if rows != 1 {
		return fmt.Errorf("update afectó %d filas, esperaba 1", rows)
	}

	fresh, err := quark.For[domain.Account](ctx, client).Where("id", "=", a.ID).First()
	if err != nil {
		return fmt.Errorf("reread: %w", err)
	}
	if fresh.Name != "crud-updated" {
		return fmt.Errorf("update no persistió: name=%q", fresh.Name)
	}

	if _, err := quark.For[domain.Account](rec.Mark(ctx, QM("Delete")), client).Delete(&fresh); err != nil {
		return fmt.Errorf("delete: %w", err)
	}
	after, err := quark.For[domain.Account](ctx, client).Where("id", "=", a.ID).Count()
	if err != nil {
		return fmt.Errorf("count post-delete: %w", err)
	}
	if after != 0 {
		return fmt.Errorf("soft-delete no excluyó la fila: count=%d", after)
	}

	rec.Note(QM("Limit"))
	if _, err := quark.For[domain.Account](rec.Mark(ctx, QM("List")), client).Limit(10).List(); err != nil {
		return fmt.Errorf("list: %w", err)
	}
	return nil
}}

var DBPERTENANT = Exerciser{Name: "tenant-db-per", Fn: runDBPerTenant}

var DEADLOCK = Exerciser{Name: "ha-deadlock", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, conn Conn) error {
	rec.Note(QF("WithDeadlockRetry"))

	dl, err := quark.New(conn.Driver, conn.DSN, append(rec.Options(),
		quark.WithDeadlockRetry(6), quark.WithMaxOpenConns(4))...)
	if err != nil {
		return fmt.Errorf("client deadlock-retry: %w", err)
	}
	defer dl.Close()

	if err := dl.Migrate(ctx, &dlProbe{}); err != nil {
		return fmt.Errorf("migrate dl_probes: %w", err)
	}
	defer func() { _, _ = dl.Raw().ExecContext(context.Background(), "DROP TABLE dl_probes") }()
	_, _ = dl.Raw().ExecContext(ctx, "DELETE FROM dl_probes")

	if err := quark.For[dlProbe](ctx, dl).CreateBatch([]*dlProbe{{Name: "dlA"}, {Name: "dlB"}}); err != nil {
		return fmt.Errorf("seed: %w", err)
	}
	rowA, err := quark.For[dlProbe](ctx, dl).Where("name", "=", "dlA").First()
	if err != nil {
		return fmt.Errorf("first dlA: %w", err)
	}
	rowB, err := quark.For[dlProbe](ctx, dl).Where("name", "=", "dlB").First()
	if err != nil {
		return fmt.Errorf("first dlB: %w", err)
	}

	update := func(tx *quark.Tx, id int64, bal int64) error {
		_, err := quark.ForTx[dlProbe](ctx, tx).Where("id", "=", id).
			UpdateMap(map[string]any{"balance": bal})
		return err
	}
	rec.Note(QM("UpdateMap"))

	if !control.Supports(control.FeatDeadlock, conn.Engine) {

		if err := dl.Tx(ctx, func(tx *quark.Tx) error {
			if err := update(tx, rowA.ID, 10); err != nil {
				return err
			}
			return update(tx, rowB.ID, 10)
		}); err != nil {
			return fmt.Errorf("tx multi-update (camino feliz): %w", err)
		}
		return nil
	}

	g1, g2 := make(chan struct{}, 1), make(chan struct{}, 1)
	barrier := func(self chan<- struct{}, other <-chan struct{}) {
		self <- struct{}{}
		select {
		case <-other:
		case <-time.After(10 * time.Second):
		}
	}
	var wg sync.WaitGroup
	var err1, err2 error
	wg.Add(2)
	go func() {
		defer wg.Done()
		first := true
		err1 = dl.Tx(ctx, func(tx *quark.Tx) error {
			if err := update(tx, rowA.ID, 1); err != nil {
				return err
			}
			if first {
				first = false
				barrier(g1, g2)
			}
			return update(tx, rowB.ID, 1)
		})
	}()
	go func() {
		defer wg.Done()
		first := true
		err2 = dl.Tx(ctx, func(tx *quark.Tx) error {
			if err := update(tx, rowB.ID, 2); err != nil {
				return err
			}
			if first {
				first = false
				barrier(g2, g1)
			}
			return update(tx, rowA.ID, 2)
		})
	}()
	wg.Wait()
	if err1 != nil || err2 != nil {
		return fmt.Errorf("WithDeadlockRetry no recuperó a la víctima: err1=%v err2=%v", err1, err2)
	}

	a, err := quark.For[dlProbe](ctx, dl).Where("id", "=", rowA.ID).First()
	if err != nil {
		return fmt.Errorf("reread A: %w", err)
	}
	b, err := quark.For[dlProbe](ctx, dl).Where("id", "=", rowB.ID).First()
	if err != nil {
		return fmt.Errorf("reread B: %w", err)
	}
	if a.Balance != b.Balance {
		return fmt.Errorf("balances divergentes tras el retry: A=%d B=%d (atomicidad rota)", a.Balance, b.Balance)
	}
	return nil
}}

var MIGRATE = Exerciser{Name: "migrate", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, conn Conn) error {
	eng := rec.Engine()

	p0, err := client.PlanMigration(rec.Mark(ctx, CM("PlanMigration")), domain.AllModels()...)
	if err != nil {
		return fmt.Errorf("plan converge: %w", err)
	}
	if !p0.IsEmpty() {
		if err := client.ApplyPlan(rec.Mark(ctx, CM("ApplyPlan")), p0); err != nil {
			return fmt.Errorf("converge (restos de un run anterior): %w", err)
		}
	}
	rec.Note(CM("Dialect"), CM("Raw"))
	_, _ = client.Raw().ExecContext(ctx,
		"DELETE FROM quark_backfill_state WHERE name = "+client.Dialect().Placeholder(1), backfillName)
	_, _ = client.Raw().ExecContext(ctx, fmt.Sprintf(
		"DELETE FROM quark_migrations WHERE id IN (%s, %s)",
		client.Dialect().Placeholder(1), client.Dialect().Placeholder(2)), migVNotesID, migVSeedID)

	p1, err := client.PlanMigration(ctx, domain.AllModels()...)
	if err != nil {
		return fmt.Errorf("plan round-trip: %w", err)
	}
	rec.Note(QF("(Plan).IsEmpty"), QF("(Plan).String"), QF("Plan"))
	if !p1.IsEmpty() {
		return fmt.Errorf("round-trip roto: el plan post-Migrate no es vacío:\n%s", p1.String())
	}

	if got := p1.String(); got != "(no changes)" {
		return fmt.Errorf("Plan.String() de plan vacío = %q, esperaba \"(no changes)\"", got)
	}

	models2 := append(domain.AllModels(), &migrateLedger{})
	p2, err := client.PlanMigration(ctx, models2...)
	if err != nil {
		return fmt.Errorf("plan con ledger: %w", err)
	}
	if p2.IsEmpty() {
		return fmt.Errorf("el plan con el modelo nuevo debería contener su CREATE TABLE")
	}
	if !strings.Contains(p2.String(), ledgerTable) {
		return fmt.Errorf("el plan no menciona %s:\n%s", ledgerTable, p2.String())
	}
	rec.Note(QF("(Plan).Hash"))
	if h := p2.Hash(); h == "" || h != p2.Hash() {
		return fmt.Errorf("Plan.Hash() debe ser determinista y no-vacío (got %q)", h)
	}
	if err := client.ApplyPlan(rec.Mark(ctx, CM("ApplyPlan")), p2); err != nil {
		return fmt.Errorf("apply ledger: %w", err)
	}
	p3, err := client.PlanMigration(ctx, models2...)
	if err != nil {
		return fmt.Errorf("plan post-apply: %w", err)
	}
	if !p3.IsEmpty() {
		return fmt.Errorf("round-trip post-ApplyPlan del ledger roto:\n%s", p3.String())
	}

	if err := quark.For[migrateLedger](rec.Mark(ctx, QM("Create")), client).Create(&migrateLedger{Ref: "r-1", Amount: 10}); err != nil {
		return fmt.Errorf("create en tabla creada por plan: %w", err)
	}
	if n, err := quark.For[migrateLedger](ctx, client).Count(); err != nil || n != 1 {
		return fmt.Errorf("count en ledger: n=%d err=%v", n, err)
	}

	if err := client.CreateIndex(rec.Mark(ctx, CM("CreateIndex")), ledgerTable, "idx_superapp_migrate_ref", []string{"ref"}, false); err != nil {
		return fmt.Errorf("create index: %w", err)
	}
	p4, err := client.PlanMigration(ctx, models2...)
	if err != nil {
		return fmt.Errorf("plan post-índice: %w", err)
	}
	if !p4.IsEmpty() {
		return fmt.Errorf("el índice manual generó drift en el plan (mergeNonColumnSurface roto):\n%s", p4.String())
	}

	rec.Note(CM("RegisteredModels"))
	if len(client.RegisteredModels()) == 0 {

		if err := client.MigrateRegistered(rec.Mark(ctx, CM("MigrateRegistered"))); err != nil {
			return fmt.Errorf("MigrateRegistered sin modelos debía ser no-op nil, got %v", err)
		}
	}
	rec.Note(CM("RegisterModel"))
	if err := client.RegisterModel(models2...); err != nil {
		return fmt.Errorf("RegisterModel: %w", err)
	}
	if got := len(client.RegisteredModels()); got != len(models2) {
		return fmt.Errorf("RegisteredModels()=%d, esperaba %d", got, len(models2))
	}
	if err := client.MigrateRegistered(rec.Mark(ctx, CM("MigrateRegistered"))); err != nil {
		return fmt.Errorf("MigrateRegistered (todo existente): %w", err)
	}
	pr, err := client.PlanMigrationRegistered(rec.Mark(ctx, CM("PlanMigrationRegistered")))
	if err != nil {
		return fmt.Errorf("PlanMigrationRegistered: %w", err)
	}
	if !pr.IsEmpty() {
		return fmt.Errorf("PlanMigrationRegistered no-vacío con todo migrado:\n%s", pr.String())
	}

	rec.Note(QF("SyncOptions"))
	if has, err := hasColumn(rec.Mark(ctx, CM("IntrospectSchema")), client, ledgerTable, "note"); err != nil {
		return fmt.Errorf("introspect pre-sync: %w", err)
	} else if has {
		return fmt.Errorf("la columna note no debería existir antes del Sync")
	}
	if err := client.Sync(rec.Mark(ctx, CM("Sync")), quark.SyncOptions{DryRun: true}, &migrateLedgerV2{}); err != nil {
		return fmt.Errorf("sync dry-run: %w", err)
	}
	if has, err := hasColumn(ctx, client, ledgerTable, "note"); err != nil {
		return fmt.Errorf("introspect post-dry-run: %w", err)
	} else if has {
		return fmt.Errorf("Sync con DryRun ejecutó DDL (la columna note existe)")
	}
	if err := client.Sync(ctx, quark.SyncOptions{}, &migrateLedgerV2{}); err != nil {
		return fmt.Errorf("sync add column: %w", err)
	}
	if has, err := hasColumn(ctx, client, ledgerTable, "note"); err != nil {
		return fmt.Errorf("introspect post-sync: %w", err)
	} else if !has {
		return fmt.Errorf("Sync no añadió la columna note")
	}

	got, err := quark.For[migrateLedgerV2](ctx, client).Where("ref", "=", "r-1").First()
	if err != nil {
		return fmt.Errorf("first V2: %w", err)
	}
	got.Note = quark.Nullable[string]{V: syncedNoteVal, Valid: true}
	if rows, err := quark.For[migrateLedgerV2](rec.Mark(ctx, QM("Update")), client).Update(&got); err != nil || rows != 1 {
		return fmt.Errorf("update V2: rows=%d err=%v", rows, err)
	}
	if re, err := quark.For[migrateLedgerV2](ctx, client).Where("ref", "=", "r-1").First(); err != nil || !re.Note.Valid || re.Note.V != syncedNoteVal {
		return fmt.Errorf("la columna añadida por Sync no hizo round-trip: note=%+v err=%v", re.Note, err)
	}

	if err := client.Sync(ctx, quark.SyncOptions{NoTransaction: true}, &migrateLedger{}); err != nil {
		return fmt.Errorf("sync drop column: %w", err)
	}
	if has, err := hasColumn(ctx, client, ledgerTable, "note"); err != nil {
		return fmt.Errorf("introspect post-drop: %w", err)
	} else if has {
		return fmt.Errorf("Sync de vuelta a V1 no dropeó la columna note (SafeMigrations=false)")
	}

	modelsV2 := append(domain.AllModels(), &migrateLedgerV2{})
	pAdd, err := client.PlanMigration(ctx, modelsV2...)
	if err != nil {
		return fmt.Errorf("plan add-column: %w", err)
	}
	if pAdd.IsEmpty() || !strings.Contains(pAdd.String(), "note") {
		return fmt.Errorf("el plan V2 debía proponer la columna note:\n%s", pAdd.String())
	}
	if err := client.ApplyPlan(rec.Mark(ctx, CM("ApplyPlan")), pAdd); err != nil {
		return fmt.Errorf("apply add-column: %w", err)
	}
	if has, err := hasColumn(ctx, client, ledgerTable, "note"); err != nil {
		return fmt.Errorf("introspect post-apply-add: %w", err)
	} else if !has {
		return fmt.Errorf("ApplyPlan no añadió la columna note")
	}
	pDropCol, err := client.PlanMigration(ctx, models2...)
	if err != nil {
		return fmt.Errorf("plan drop-column: %w", err)
	}
	if pDropCol.IsEmpty() {
		return fmt.Errorf("el plan de vuelta a V1 debía proponer el drop de note")
	}
	if err := client.ApplyPlan(rec.Mark(ctx, CM("ApplyPlan")), pDropCol); err != nil {
		return fmt.Errorf("apply drop-column: %w", err)
	}
	if has, err := hasColumn(ctx, client, ledgerTable, "note"); err != nil {
		return fmt.Errorf("introspect post-apply-drop: %w", err)
	} else if has {
		return fmt.Errorf("ApplyPlan no dropeó la columna note")
	}

	seed := make([]*migrateLedger, 0, 24)
	for i := 0; i < 24; i++ {
		seed = append(seed, &migrateLedger{Ref: fmt.Sprintf("bf-%02d", i), Amount: int64(i)})
	}
	if err := quark.For[migrateLedger](rec.Mark(ctx, QM("CreateBatch")), client).CreateBatch(seed); err != nil {
		return fmt.Errorf("seed backfill: %w", err)
	}

	rec.Note(QF("BackfillSpec"))
	sentinel := errors.New("fallo inyectado en el lote 2")
	var run1 [][]int64
	err = client.Backfill(rec.Mark(ctx, CM("Backfill")), quark.BackfillSpec{
		Name: backfillName, Table: ledgerTable, BatchSize: 10,
		Process: func(_ context.Context, pks []int64) error {
			if len(run1) == 1 {
				return sentinel
			}
			run1 = append(run1, pks)
			return nil
		},
	})
	if !errors.Is(err, sentinel) {
		return fmt.Errorf("backfill run1: esperaba el sentinel del lote 2, got %v", err)
	}
	if len(run1) != 1 {
		return fmt.Errorf("backfill run1 procesó %d lotes antes del fallo, esperaba exactamente 1", len(run1))
	}
	if len(run1[0]) != 10 {
		return fmt.Errorf("backfill run1: el lote 1 trajo %d PKs, esperaba 10", len(run1[0]))
	}
	maxSeen := run1[0][len(run1[0])-1]
	// Re-invocación con el mismo Name: resume DESPUÉS del último PK persistido,
	// sin reprocesar el lote 1.
	var run2 []int64
	err = client.Backfill(rec.Mark(ctx, CM("Backfill")), quark.BackfillSpec{
		Name: backfillName, Table: ledgerTable, BatchSize: 10,
		Process: func(_ context.Context, pks []int64) error {
			run2 = append(run2, pks...)
			return nil
		},
	})
	if err != nil {
		return fmt.Errorf("backfill run2 (resume): %w", err)
	}
	if len(run2) != 15 {
		return fmt.Errorf("backfill run2 procesó %d PKs, esperaba los 15 restantes", len(run2))
	}

	if run2[0] <= maxSeen {
		return fmt.Errorf("backfill run2 reprocesó PKs del lote 1 (primero=%d, estado=%d)", run2[0], maxSeen)
	}

	calls := 0
	err = client.Backfill(rec.Mark(ctx, CM("Backfill")), quark.BackfillSpec{
		Name: backfillName, Table: ledgerTable, BatchSize: 10,
		Process: func(_ context.Context, _ []int64) error { calls++; return nil },
	})
	if err != nil || calls != 0 {
		return fmt.Errorf("backfill run3 (completo): calls=%d err=%v, esperaba 0 y nil", calls, err)
	}

	rec.Note(QF("MigrationLock"))
	if control.Supports(control.FeatMigrationLock, eng) {
		lock1, err := client.AcquireMigrationLock(rec.Mark(ctx, CM("AcquireMigrationLock")), migLockName, 5*time.Second)
		if err != nil {
			return fmt.Errorf("acquire lock: %w", err)
		}

		if _, err := client.AcquireMigrationLock(ctx, migLockName, time.Second); !errors.Is(err, quark.ErrLockTimeout) {
			_ = lock1.Release(ctx)
			return fmt.Errorf("acquire concurrente: esperaba ErrLockTimeout, got %v", err)
		}
		rec.Note(QF("ErrLockTimeout"))
		if err := lock1.Release(ctx); err != nil {
			return fmt.Errorf("release: %w", err)
		}
		rec.Note(QF("(MigrationLock).Release"))

		lock2, err := client.AcquireMigrationLock(ctx, migLockName, 5*time.Second)
		if err != nil {
			return fmt.Errorf("re-acquire tras release: %w", err)
		}
		if err := lock2.Release(ctx); err != nil {
			return fmt.Errorf("release 2: %w", err)
		}
	} else {

		_, err := client.AcquireMigrationLock(rec.Mark(ctx, CM("AcquireMigrationLock")), migLockName, time.Second)
		if !errors.Is(err, quark.ErrUnsupportedFeature) {
			return fmt.Errorf("lock en %s: esperaba ErrUnsupportedFeature, got %v", eng, err)
		}
		rec.Note(QF("ErrUnsupportedFeature"))
	}

	migLimits := quark.DefaultLimits()
	migLimits.AllowRawQueries = true
	admin, err := quark.New(conn.Driver, conn.DSN, append(rec.Options(), quark.WithLimits(migLimits))...)
	if err != nil {
		return fmt.Errorf("client de migración (AllowRawQueries): %w", err)
	}
	defer admin.Close()

	migrate.Reset()
	defer migrate.Reset()
	rec.Note(MIG("Reset"), MIG("Register"), MIG("Migration"), MIG("Migrator"), MIG("NewMigrator"))
	migrate.Register(&migrate.Migration{
		ID: migVNotesID, Name: "create v_notes",
		Up:   func(ctx context.Context, c *quark.Client) error { return c.Migrate(ctx, &migrateVNote{}) },
		Down: func(ctx context.Context, c *quark.Client) error { return execRaw(ctx, c, "DROP TABLE "+vNotesTable) },
	})
	migrate.Register(&migrate.Migration{
		ID: migVSeedID, Name: "seed v_notes",
		Up: func(ctx context.Context, c *quark.Client) error {
			return quark.For[migrateVNote](ctx, c).Create(&migrateVNote{Body: "seeded"})
		},
		Down: func(ctx context.Context, c *quark.Client) error { return execRaw(ctx, c, "DELETE FROM "+vNotesTable) },
	})

	if n := migrate.RegisteredCount(); n != 2 {
		return fmt.Errorf("RegisteredCount: %d, esperaba 2", n)
	}
	rec.Note(MIG("RegisteredCount"))
	m := migrate.NewMigrator(admin)
	if err := m.Init(rec.Mark(ctx, MIG("(*Migrator).Init"))); err != nil {
		return fmt.Errorf("migrator init: %w", err)
	}

	if err := m.UpDryRun(rec.Mark(ctx, MIG("(*Migrator).UpDryRun")), 0); err != nil {
		return fmt.Errorf("up dry-run: %w", err)
	}
	if has, err := hasTable(ctx, client, vNotesTable); err != nil {
		return fmt.Errorf("introspect post-dry-run versionado: %w", err)
	} else if has {
		return fmt.Errorf("UpDryRun ejecutó la migración (existe %s)", vNotesTable)
	}

	if err := m.Up(rec.Mark(ctx, MIG("(*Migrator).Up")), 0); err != nil {
		return fmt.Errorf("up: %w", err)
	}
	if n, err := quark.For[migrateVNote](ctx, admin).Count(); err != nil || n != 1 {
		return fmt.Errorf("v_notes tras Up: n=%d err=%v, esperaba la fila seeded", n, err)
	}
	applied, err := m.GetApplied(rec.Mark(ctx, MIG("(*Migrator).GetApplied")))
	if err != nil {
		return fmt.Errorf("get applied: %w", err)
	}
	if !applied[migVNotesID] || !applied[migVSeedID] {
		return fmt.Errorf("GetApplied no registra las dos migraciones: %v", applied)
	}

	if err := m.Down(rec.Mark(ctx, MIG("(*Migrator).Down")), 1); err != nil {
		return fmt.Errorf("down(1): %w", err)
	}
	if n, err := quark.For[migrateVNote](ctx, admin).Count(); err != nil || n != 0 {
		return fmt.Errorf("v_notes tras Down(1): n=%d err=%v, esperaba 0 (seed revertido)", n, err)
	}
	applied, err = m.GetApplied(ctx)
	if err != nil {
		return fmt.Errorf("get applied post-down: %w", err)
	}
	if !applied[migVNotesID] || applied[migVSeedID] {
		return fmt.Errorf("Down(1) debía revertir sólo %s: %v", migVSeedID, applied)
	}

	if err := m.Up(ctx, 1); err != nil {
		return fmt.Errorf("re-up: %w", err)
	}
	if n, err := quark.For[migrateVNote](ctx, admin).Count(); err != nil || n != 1 {
		return fmt.Errorf("v_notes tras re-Up: n=%d err=%v", n, err)
	}
	if err := m.Down(ctx, 0); err != nil {
		return fmt.Errorf("down(0): %w", err)
	}
	if has, err := hasTable(ctx, client, vNotesTable); err != nil {
		return fmt.Errorf("introspect post-down-all: %w", err)
	} else if has {
		return fmt.Errorf("Down(0) no dropeó %s", vNotesTable)
	}

	pDrop, err := client.PlanMigration(ctx, domain.AllModels()...)
	if err != nil {
		return fmt.Errorf("plan drop: %w", err)
	}
	if pDrop.IsEmpty() || !strings.Contains(pDrop.String(), ledgerTable) {
		return fmt.Errorf("el plan de cleanup debía proponer el drop de %s:\n%s", ledgerTable, pDrop.String())
	}
	if err := client.ApplyPlan(ctx, pDrop); err != nil {
		return fmt.Errorf("apply drop: %w", err)
	}
	pFinal, err := client.PlanMigration(ctx, domain.AllModels()...)
	if err != nil {
		return fmt.Errorf("plan final: %w", err)
	}
	if !pFinal.IsEmpty() {
		return fmt.Errorf("la BD no quedó canónica tras el cleanup:\n%s", pFinal.String())
	}

	_, _ = client.Raw().ExecContext(ctx,
		"DELETE FROM quark_backfill_state WHERE name = "+client.Dialect().Placeholder(1), backfillName)
	return nil
}}

var OBSERVABILITY = Exerciser{Name: "observability", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, conn Conn) error {

	prevTP := otelapi.GetTracerProvider()
	sr := tracetest.NewSpanRecorder()
	tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
	otelapi.SetTracerProvider(tp)
	defer func() {

		otelapi.SetTracerProvider(prevTP)
		_ = tp.Shutdown(context.Background())
	}()
	prevMP := otelapi.GetMeterProvider()
	mreader := sdkmetric.NewManualReader()
	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(mreader))
	otelapi.SetMeterProvider(mp)
	defer func() {
		otelapi.SetMeterProvider(prevMP)
		_ = mp.Shutdown(context.Background())
	}()

	rec.Note(OTL("New"), OTL("WithDBSystem"), OTL("Middleware"), OTL("Option"),
		OTL("RedactionMode"), OTL("RedactArgs"),
		QF("WithMiddleware"), QF("WithLogger"), QF("WithSlowQueryThreshold"))
	var logBuf bytes.Buffer
	logger := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelDebug}))
	l := quark.DefaultLimits()
	l.SafeMigrations = false
	obsClient, err := quark.New(conn.Driver, conn.DSN, append(rec.Options(),
		quark.WithMiddleware(quarkotel.New(quarkotel.WithDBSystem(string(conn.Engine)))),
		quark.WithLogger(logger),
		quark.WithSlowQueryThreshold(time.Nanosecond),
		quark.WithLimits(l))...)
	if err != nil {
		return fmt.Errorf("client observado: %w", err)
	}
	defer obsClient.Close()

	if err := obsClient.Migrate(ctx, &obsProbe{}); err != nil {
		return fmt.Errorf("migrate obs_probes: %w", err)
	}
	defer func() { _, _ = obsClient.Raw().ExecContext(context.Background(), "DROP TABLE obs_probes") }()
	_, _ = obsClient.Raw().ExecContext(ctx, "DELETE FROM obs_probes")

	// Operaciones sobre las TRES vías del pipeline (exec / query / query_row),
	// con un bind secreto que la redacción debe retener.
	const secret = "obs-SECRET-93c1"
	rec.Note(OTL("(*Middleware).WrapExec"), OTL("(*Middleware).WrapQuery"), OTL("(*Middleware).WrapQueryRow"))
	row := &obsProbe{Name: secret, Val: 1}
	if err := quark.For[obsProbe](rec.Mark(ctx, QM("Create")), obsClient).Create(row); err != nil {
		return fmt.Errorf("create: %w", err)
	}
	got, err := quark.For[obsProbe](rec.Mark(ctx, QM("First")), obsClient).Where("name", "=", secret).First()
	if err != nil {
		return fmt.Errorf("first: %w", err)
	}
	got.Val = 2
	if _, err := quark.For[obsProbe](rec.Mark(ctx, QM("Update")), obsClient).Update(&got); err != nil {
		return fmt.Errorf("update: %w", err)
	}
	if _, err := quark.For[obsProbe](rec.Mark(ctx, QM("List")), obsClient).Limit(5).List(); err != nil {
		return fmt.Errorf("list: %w", err)
	}

	if _, err := quark.For[obsMissing](ctx, obsClient).Limit(1).List(); err == nil {
		return fmt.Errorf("esperaba error del motor por tabla inexistente")
	}

	spans := sr.Ended()
	if len(spans) < 5 {
		return fmt.Errorf("esperaba ≥5 spans (create/first/update/list/error), got %d", len(spans))
	}

	validNames := map[string]bool{"quark.exec": true, "quark.query": true, "quark.query_row": true}
	var sawStatement, sawError bool
	for _, s := range spans {
		if !validNames[s.Name()] {
			return fmt.Errorf("span con nombre inesperado %q", s.Name())
		}
		var hasSystem bool
		for _, kv := range s.Attributes() {
			switch string(kv.Key) {
			case "db.statement":
				sawStatement = true
				if strings.Contains(kv.Value.AsString(), secret) {
					return fmt.Errorf("FUGA: el bind secreto apareció en db.statement del span %s", s.Name())
				}
			case "db.statement.args":
				return fmt.Errorf("FUGA: db.statement.args presente bajo RedactArgs (span %s)", s.Name())
			case "db.system":
				hasSystem = true
				if kv.Value.AsString() != string(conn.Engine) {
					return fmt.Errorf("db.system=%q, esperaba %q", kv.Value.AsString(), conn.Engine)
				}
			}
		}
		if !hasSystem {
			return fmt.Errorf("span %s sin atributo db.system", s.Name())
		}
		if s.Status().Code == codes.Error {
			sawError = true
		}
	}
	if !sawStatement {
		return fmt.Errorf("ningún span llevó db.statement")
	}
	if !sawError {
		return fmt.Errorf("el error del motor no marcó ningún span con codes.Error")
	}

	// --- Asserts de métricas: quark.queries.total suma las operaciones. ------
	var rm metricdata.ResourceMetrics
	if err := mreader.Collect(ctx, &rm); err != nil {
		return fmt.Errorf("collect métricas: %w", err)
	}
	var total int64
	for _, sm := range rm.ScopeMetrics {
		for _, m := range sm.Metrics {
			if m.Name != "quark.queries.total" {
				continue
			}
			if sum, ok := m.Data.(metricdata.Sum[int64]); ok {
				for _, dp := range sum.DataPoints {
					total += dp.Value
				}
			}
		}
	}
	if total < int64(len(spans)) {
		return fmt.Errorf("quark.queries.total=%d, esperaba ≥ %d (un punto por operación)", total, len(spans))
	}

	logs := logBuf.String()
	if logs == "" {
		return fmt.Errorf("el logger de Quark no narró nada con SlowQueryThreshold=1ns")
	}
	if strings.Contains(logs, secret) {
		return fmt.Errorf("FUGA DE REDACCIÓN: el bind secreto apareció en el log de Quark")
	}
	if !strings.Contains(logs, "obs_probes") {
		return fmt.Errorf("el log no menciona la tabla consultada (¿narró las queries?)")
	}

	rec.Note(OTL("WithSpanRedaction"), OTL("IncludeArgs"))

	dbgClient, err := quark.New(conn.Driver, conn.DSN,
		quark.WithMiddleware(quarkotel.New(quarkotel.WithSpanRedaction(quarkotel.IncludeArgs))),
		quark.WithLimits(l))
	if err != nil {
		return fmt.Errorf("client debug: %w", err)
	}
	defer dbgClient.Close()
	const secret2 = "obs-DEBUG-41aa"
	before := len(sr.Ended())
	if _, err := quark.For[obsProbe](ctx, dbgClient).Where("name", "=", secret2).Count(); err != nil {
		return fmt.Errorf("count debug: %w", err)
	}
	var sawArgs bool
	for _, s := range sr.Ended()[before:] {
		for _, kv := range s.Attributes() {
			if string(kv.Key) == "db.statement.args" {
				sawArgs = true
				if !strings.Contains(fmt.Sprintf("%v", kv.Value.AsStringSlice()), secret2) {
					return fmt.Errorf("IncludeArgs activo pero el arg no está en db.statement.args")
				}
			}
		}
	}
	if !sawArgs {
		return fmt.Errorf("IncludeArgs no expuso db.statement.args en ningún span nuevo")
	}
	return nil
}}

var RELATIONS = Exerciser{Name: "relations", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, _ Conn) error {
	rec.Note(QF("For"), QM("Preload"), QM("Where"))
	n := atomic.AddInt64(&relSeq, 1)

	owner := &domain.Account{Email: fmt.Sprintf("rel%d@superapp.test", n), Name: "rel", Role: "member", Active: true}
	if err := quark.For[domain.Account](ctx, client).Create(owner); err != nil {
		return fmt.Errorf("seed owner: %w", err)
	}
	proj := &domain.Project{OwnerID: owner.ID, Name: "rel-proj", Status: "active"}
	if err := quark.For[domain.Project](ctx, client).Create(proj); err != nil {
		return fmt.Errorf("seed project: %w", err)
	}
	aid := owner.ID
	assigned := &domain.Task{ProjectID: proj.ID, Title: "assigned", AssigneeID: &aid}
	if err := quark.For[domain.Task](ctx, client).Create(assigned); err != nil {
		return fmt.Errorf("seed assigned task: %w", err)
	}
	unassigned := &domain.Task{ProjectID: proj.ID, Title: "unassigned"}
	if err := quark.For[domain.Task](ctx, client).Create(unassigned); err != nil {
		return fmt.Errorf("seed unassigned task: %w", err)
	}

	t1, err := quark.For[domain.Task](rec.Mark(ctx, QM("First")), client).
		Preload("Project").Preload("Assignee").Where("id", "=", assigned.ID).First()
	if err != nil {
		return fmt.Errorf("preload belongs_to: %w", err)
	}
	if t1.Project == nil || t1.Project.ID != proj.ID {
		return fmt.Errorf("belongs_to Project no cargado: %+v", t1.Project)
	}
	if t1.Assignee == nil || t1.Assignee.ID != owner.ID {
		return fmt.Errorf("belongs_to Assignee no cargado: %+v", t1.Assignee)
	}

	t2, err := quark.For[domain.Task](ctx, client).Preload("Assignee").Where("id", "=", unassigned.ID).First()
	if err != nil {
		return fmt.Errorf("preload nullable FK: %w", err)
	}
	if t2.Assignee != nil {
		return fmt.Errorf("BB-5: Assignee debía quedar nil, cargó %+v", t2.Assignee)
	}

	acc, err := quark.For[domain.Account](ctx, client).Preload("Projects").Where("id", "=", owner.ID).First()
	if err != nil {
		return fmt.Errorf("preload has_many Projects: %w", err)
	}
	if len(acc.Projects) < 1 {
		return fmt.Errorf("has_many Projects vacío")
	}
	p, err := quark.For[domain.Project](ctx, client).Preload("Tasks").Where("id", "=", proj.ID).First()
	if err != nil {
		return fmt.Errorf("preload has_many Tasks: %w", err)
	}
	if len(p.Tasks) != 2 {
		return fmt.Errorf("has_many Tasks=%d, esperaba 2", len(p.Tasks))
	}

	mProj := &domain.Project{
		OwnerID: owner.ID, Name: "m2m-proj", Status: "active",
		Tags: []domain.Tag{
			{Slug: fmt.Sprintf("rel-tag-a-%d", n)},
			{Slug: fmt.Sprintf("rel-tag-b-%d", n)},
		},
	}
	if err := quark.For[domain.Project](rec.Mark(ctx, QM("Create")), client).Create(mProj); err != nil {
		return fmt.Errorf("m2m create con tags: %w", err)
	}
	mp, err := quark.For[domain.Project](ctx, client).Preload("Tags").Where("id", "=", mProj.ID).First()
	if err != nil {
		return fmt.Errorf("preload m2m Tags: %w", err)
	}
	if len(mp.Tags) != 2 {
		return fmt.Errorf("m2m Tags=%d, esperaba 2", len(mp.Tags))
	}

	return nil
}}

var REPLICAS = Exerciser{Name: "ha-replicas", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, conn Conn) error {
	rec.Note(QF("ReplicaStrategy"), QF("ReplicaRoundRobin"), QF("ReplicaRandom"), QF("ReplicaLeastConn"))
	if !control.Supports(control.FeatDBPerTenantProvision, conn.Engine) {

		rec.Note(QF("WithReplicas"), QF("WithReplicaStrategy"), QF("WithReplicaDownCooldown"), QF("Sticky"))
		routed, err := quark.New(conn.Driver, conn.DSN,
			quark.WithReplicas(conn.DSN),
			quark.WithReplicaStrategy(quark.ReplicaRoundRobin),
			quark.WithReplicaDownCooldown(30*time.Second))
		if err != nil {
			return fmt.Errorf("réplica degenerada (%s): %w", conn.Engine, err)
		}
		defer routed.Close()
		_ = quark.Sticky(ctx)
		return nil
	}

	dsns, cleanup, err := provisionHADBs(ctx, conn, "r", 2)
	if err != nil {
		return fmt.Errorf("provision réplicas: %w", err)
	}
	defer cleanup()

	for i, dsn := range dsns {
		rc, err := quark.New(conn.Driver, dsn)
		if err != nil {
			return fmt.Errorf("replica client %d: %w", i+1, err)
		}
		if err := rc.Migrate(ctx, &haProbe{}); err != nil {
			_ = rc.Close()
			return fmt.Errorf("migrate réplica %d: %w", i+1, err)
		}
		if err := quark.For[haProbe](ctx, rc).Create(&haProbe{Marker: fmt.Sprintf("r%d", i+1)}); err != nil {
			_ = rc.Close()
			return fmt.Errorf("seed réplica %d: %w", i+1, err)
		}
		if err := rc.Close(); err != nil {
			return fmt.Errorf("close réplica %d: %w", i+1, err)
		}
	}

	if err := client.Migrate(ctx, &haProbe{}); err != nil {
		return fmt.Errorf("migrate primary: %w", err)
	}
	defer func() { _, _ = client.Raw().ExecContext(context.Background(), "DROP TABLE ha_probes") }()
	_, _ = client.Raw().ExecContext(ctx, "DELETE FROM ha_probes")
	for _, m := range []string{"primary", "primary2"} {
		if err := quark.For[haProbe](ctx, client).Create(&haProbe{Marker: m}); err != nil {
			return fmt.Errorf("seed primary: %w", err)
		}
	}

	rec.Note(QF("WithReplicas"), QF("WithReplicaStrategy"), QF("WithReplicaDownCooldown"))
	routed, err := quark.New(conn.Driver, conn.DSN, append(rec.Options(),
		quark.WithReplicas(dsns...),
		quark.WithReplicaStrategy(quark.ReplicaRoundRobin),
		quark.WithReplicaDownCooldown(30*time.Second))...)
	if err != nil {
		return fmt.Errorf("routed client: %w", err)
	}
	defer routed.Close()

	seen := map[string]bool{}
	for i := 0; i < 4; i++ {
		got, err := quark.For[haProbe](rec.Mark(ctx, QM("First")), routed).OrderBy("id", "ASC").First()
		if err != nil {
			return fmt.Errorf("read no-sticky %d: %w", i, err)
		}
		seen[got.Marker] = true
	}
	if seen["primary"] || seen["primary2"] {
		return fmt.Errorf("una lectura no-sticky llegó al primary: %v", seen)
	}
	if !seen["r1"] || !seen["r2"] {
		return fmt.Errorf("round-robin no repartió entre las 2 réplicas en 4 lecturas: %v", seen)
	}

	if n, err := quark.For[haProbe](ctx, routed).Count(); err != nil || n != 1 {
		return fmt.Errorf("count no-sticky: n=%d err=%v, esperaba 1 (réplica)", n, err)
	}

	sctx := quark.Sticky(ctx)
	rec.Note(QF("Sticky"))
	if n, err := quark.For[haProbe](sctx, routed).Count(); err != nil || n != 2 {
		return fmt.Errorf("count sticky: n=%d err=%v, esperaba 2 (primary)", n, err)
	}

	if err := routed.Tx(ctx, func(tx *quark.Tx) error {
		n, err := quark.ForTx[haProbe](ctx, tx).Count()
		if err != nil {
			return err
		}
		if n != 2 {
			return fmt.Errorf("count en tx: n=%d, esperaba 2 (primary)", n)
		}
		return nil
	}); err != nil {
		return fmt.Errorf("tx read: %w", err)
	}

	if err := quark.For[haProbe](ctx, routed).Create(&haProbe{Marker: "fresh"}); err != nil {
		return fmt.Errorf("write ruteado: %w", err)
	}
	if n, err := quark.For[haProbe](ctx, routed).Count(); err != nil || n != 1 {
		return fmt.Errorf("count no-sticky post-write: n=%d err=%v, esperaba 1 (la réplica no ve el write)", n, err)
	}
	if n, err := quark.For[haProbe](sctx, routed).Count(); err != nil || n != 3 {
		return fmt.Errorf("count sticky post-write: n=%d err=%v, esperaba 3 (write aterrizó en primary)", n, err)
	}

	for _, st := range []quark.ReplicaStrategy{quark.ReplicaRandom, quark.ReplicaLeastConn} {
		c, err := quark.New(conn.Driver, conn.DSN, quark.WithReplicas(dsns...), quark.WithReplicaStrategy(st))
		if err != nil {
			return fmt.Errorf("client estrategia %v: %w", st, err)
		}
		n, err := quark.For[haProbe](ctx, c).Count()
		cerr := c.Close()
		if err != nil || n != 1 {
			return fmt.Errorf("estrategia %v: n=%d err=%v, esperaba 1 (réplica)", st, n, err)
		}
		if cerr != nil {
			return fmt.Errorf("close estrategia %v: %w", st, cerr)
		}
	}
	return nil
}}

var RLSNATIVE = Exerciser{Name: "tenant-rls-native", Fn: runRLSNative}

var SCHEMAPERTENANT = Exerciser{Name: "tenant-schema-per", Fn: runSchemaPerTenant}

var SECURITY = Exerciser{Name: "security", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, _ Conn) error {
	rec.Note(QF("For"), QM("Where"), QM("WhereJSON"), QM("Join"))

	hostile := []string{
		`id; DROP TABLE accounts;--`,
		`id) OR 1=1 --`,
		`name'`,
		`1=1`,
		`id'); DELETE FROM accounts;--`,
	}
	for _, h := range hostile {
		if _, err := quark.For[domain.Account](ctx, client).Where(h, "=", 1).List(); err == nil {
			return fmt.Errorf("identificador hostil %q NO fue rechazado", h)
		} else if !strings.Contains(strings.ToLower(err.Error()), "identifier") {
			return fmt.Errorf("identificador hostil %q: error inesperado: %v", h, err)
		}
	}

	if _, err := quark.For[domain.Account](ctx, client).OrderBy("name; DROP TABLE--", "ASC").Limit(1).List(); err == nil {
		return fmt.Errorf("OrderBy con columna hostil NO fue rechazado")
	}

	if _, err := quark.For[domain.Account](ctx, client).WhereJSON("settings", `theme'; DROP--`, "=", "x").List(); !errors.Is(err, quark.ErrInvalidJSONPath) {
		return fmt.Errorf("JSON-path hostil: esperaba ErrInvalidJSONPath, got %v", err)
	}

	if _, err := quark.For[domain.Task](ctx, client).Join("projects").On(`id; DROP TABLE projects`, "=", "x").List(); !errors.Is(err, quark.ErrInvalidJoin) {
		return fmt.Errorf("JOIN ON hostil: esperaba ErrInvalidJoin, got %v", err)
	}

	return nil
}}

var SHARDING = Exerciser{Name: "ha-sharding", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, conn Conn) error {
	rec.Note(QF("ShardRouter"), QF("ShardFunc"), QF("ShardResolver"))
	if !control.Supports(control.FeatDBPerTenantProvision, conn.Engine) {

		rec.Note(QF("NewShardRouter"), QF("HashShardFunc"), QF("DefaultShardResolver"),
			SRM("ShardNames"), QF("WithShardKey"), QF("WithShardKeyOf"), QF("ShardKeyFromContext"), SRM("GetClient"))
		router, err := quark.NewShardRouter(
			map[string]*quark.Client{"s1": client},
			quark.DefaultShardResolver, quark.HashShardFunc([]string{"s1"}))
		if err != nil {
			return fmt.Errorf("shard router degenerado (%s): %w", conn.Engine, err)
		}
		_ = router.ShardNames()
		kctx := quark.WithShardKey(ctx, "acct-007")
		_ = quark.ShardKeyFromContext(kctx)
		if _, err := router.GetClient(kctx); err != nil {
			return fmt.Errorf("GetClient degenerado (%s): %w", conn.Engine, err)
		}

		if _, err := router.GetClient(quark.WithShardKeyOf(ctx, shardProbe{Skey: "acct-007"})); err != nil {
			return fmt.Errorf("GetClient via WithShardKeyOf degenerado (%s): %w", conn.Engine, err)
		}

		rec.Note(QF("ScatterCount"), QF("ScatterGather"))
		if err := client.Migrate(ctx, &shardProbe{}); err != nil {
			return fmt.Errorf("migrate shardProbe degenerado (%s): %w", conn.Engine, err)
		}
		if _, err := quark.ScatterCount[shardProbe](ctx, router, nil); err != nil {
			return fmt.Errorf("ScatterCount degenerado (%s): %w", conn.Engine, err)
		}
		if _, err := quark.ScatterGather(ctx, router,
			func(q *quark.Query[shardProbe]) *quark.Query[shardProbe] { return q.Limit(1) },
			quark.ScatterMerge[shardProbe]{}); err != nil {
			return fmt.Errorf("ScatterGather degenerado (%s): %w", conn.Engine, err)
		}
		return nil
	}

	dsns, cleanup, err := provisionHADBs(ctx, conn, "s", 3)
	if err != nil {
		return fmt.Errorf("provision shards: %w", err)
	}
	defer cleanup()

	names := []string{"s1", "s2", "s3"}
	shards := make(map[string]*quark.Client, 3)
	defer func() {
		for _, c := range shards {
			_ = c.Close()
		}
	}()
	for i, name := range names {
		c, err := quark.New(conn.Driver, dsns[i], rec.Options()...)
		if err != nil {
			return fmt.Errorf("shard client %s: %w", name, err)
		}
		shards[name] = c
		if err := c.Migrate(ctx, &shardProbe{}); err != nil {
			return fmt.Errorf("migrate shard %s: %w", name, err)
		}
	}

	rec.Note(QF("NewShardRouter"), QF("HashShardFunc"), QF("DefaultShardResolver"))
	router, err := quark.NewShardRouter(shards, quark.DefaultShardResolver, quark.HashShardFunc(names))
	if err != nil {
		return fmt.Errorf("NewShardRouter: %w", err)
	}
	rec.Note(SRM("ShardNames"))
	if got := router.ShardNames(); len(got) != 3 {
		return fmt.Errorf("ShardNames()=%v, esperaba 3", got)
	}

	if _, err := quark.For[shardProbe](ctx, router).Limit(1).List(); !errors.Is(err, quark.ErrInvalidQuery) {
		return fmt.Errorf("sin shard key esperaba ErrInvalidQuery, got %v", err)
	}

	rec.Note(QF("WithShardKey"), QF("WithShardKeyOf"), QF("ShardKeyFromContext"), SRM("GetClient"))
	kctx := quark.WithShardKey(ctx, "acct-007")
	if got := quark.ShardKeyFromContext(kctx); got != "acct-007" {
		return fmt.Errorf("ShardKeyFromContext=%q", got)
	}
	c1, err := router.GetClient(kctx)
	if err != nil {
		return fmt.Errorf("GetClient: %w", err)
	}
	if c2, _ := router.GetClient(kctx); c2 != c1 {
		return fmt.Errorf("la misma key resolvió a clients distintos")
	}

	// Distribución: 30 keys → todas las filas aterrizan, cada shard recibe
	// alguna (FNV-1a reparte; el chi-square fino ya lo pinneó el bug-bash F10).
	const nKeys = 30
	for i := 0; i < nKeys; i++ {
		k := fmt.Sprintf("cust-%02d", i)
		p := &shardProbe{Skey: k}

		if err := quark.For[shardProbe](rec.Mark(quark.WithShardKeyOf(ctx, p), QM("Create")), router).Create(p); err != nil {
			return fmt.Errorf("create %s: %w", k, err)
		}
	}
	total := int64(0)
	for name, c := range shards {
		n, err := quark.For[shardProbe](ctx, c).Count()
		if err != nil {
			return fmt.Errorf("count shard %s: %w", name, err)
		}
		if n == 0 {
			return fmt.Errorf("el shard %s no recibió ninguna key (distribución rota)", name)
		}
		total += n
	}
	if total != nKeys {
		return fmt.Errorf("filas totales=%d, esperaba %d (¿fan-out o pérdida?)", total, nKeys)
	}

	probe, err := quark.For[shardProbe](quark.WithShardKey(ctx, "cust-00"), router).Where("skey", "=", "cust-00").First()
	if err != nil {
		return fmt.Errorf("first cust-00: %w", err)
	}
	owner, _ := router.GetClient(quark.WithShardKey(ctx, "cust-00"))
	for name, c := range shards {
		n, err := quark.For[shardProbe](ctx, c).Where("skey", "=", "cust-00").Count()
		if err != nil {
			return fmt.Errorf("isolation count %s: %w", name, err)
		}
		if c == owner && n != 1 {
			return fmt.Errorf("el shard dueño %s tiene %d filas de cust-00 (id=%d), esperaba 1", name, n, probe.ID)
		}
		if c != owner && n != 0 {
			return fmt.Errorf("fuga cross-shard: %s tiene %d filas de cust-00", name, n)
		}
	}

	rec.Note(QF("ScatterCount"), QF("ScatterGather"))
	scN, err := quark.ScatterCount[shardProbe](ctx, router, nil)
	if err != nil {
		return fmt.Errorf("ScatterCount: %w", err)
	}
	if scN != nKeys {
		return fmt.Errorf("ScatterCount=%d, esperaba %d (suma cross-shard)", scN, nKeys)
	}
	top, err := quark.ScatterGather(ctx, router,
		func(q *quark.Query[shardProbe]) *quark.Query[shardProbe] {
			return q.OrderBy("skey", "DESC").Limit(5)
		},
		quark.ScatterMerge[shardProbe]{
			Less:  func(a, b shardProbe) bool { return a.Skey > b.Skey },
			Limit: 5,
		})
	if err != nil {
		return fmt.Errorf("ScatterGather: %w", err)
	}
	if len(top) != 5 || top[0].Skey != "cust-29" {
		return fmt.Errorf("ScatterGather top-5 por skey DESC = %v, esperaba 5 filas con cust-29 primero", top)
	}

	if err := c1.Tx(ctx, func(tx *quark.Tx) error {
		return quark.ForTx[shardProbe](ctx, tx).Create(&shardProbe{Skey: "tx-row"})
	}); err != nil {
		return fmt.Errorf("tx por shard: %w", err)
	}

	dsns4, cleanup4, err := provisionHADBs(ctx, conn, "x", 1)
	if err != nil {
		return fmt.Errorf("provision 4º shard: %w", err)
	}
	defer cleanup4()
	c4, err := quark.New(conn.Driver, dsns4[0])
	if err != nil {
		return fmt.Errorf("client 4º shard: %w", err)
	}
	defer c4.Close()
	if err := c4.Migrate(ctx, &shardProbe{}); err != nil {
		return fmt.Errorf("migrate 4º shard: %w", err)
	}
	shards4 := map[string]*quark.Client{"s1": shards["s1"], "s2": shards["s2"], "s3": shards["s3"], "s4": c4}
	router4, err := quark.NewShardRouter(shards4, quark.DefaultShardResolver, quark.HashShardFunc([]string{"s1", "s2", "s3", "s4"}))
	if err != nil {
		return fmt.Errorf("router resharded: %w", err)
	}
	if err := quark.For[shardProbe](quark.WithShardKey(ctx, "post-reshard"), router4).Create(&shardProbe{Skey: "post-reshard"}); err != nil {
		return fmt.Errorf("create post-reshard: %w", err)
	}
	return nil
}}

var SURFACE = Exerciser{Name: "surface-funcs", Fn: runSurfaceFuncs}

var SURFACECLIENT = Exerciser{Name: "surface-client", Fn: runSurfaceClient}

var SURFACEMETHODS = Exerciser{Name: "surface-methods", Fn: runSurfaceMethods}

var TENANT = Exerciser{Name: "tenant", Fn: runTenant}

var TX = Exerciser{Name: "tx", Fn: func(ctx context.Context, client *quark.Client, rec *recorder.Recorder, _ Conn) error {
	rec.Note(CM("Tx"), QF("ForTx"))
	n := atomic.AddInt64(&txSeq, 1)

	email := fmt.Sprintf("tx%d@superapp.test", n)
	var accID int64
	err := client.Tx(rec.Mark(ctx, CM("Tx")), func(tx *quark.Tx) error {
		a := &domain.Account{Email: email, Name: "tx", Role: "member", Active: true}
		if err := quark.ForTx[domain.Account](ctx, tx).Create(a); err != nil {
			return err
		}
		accID = a.ID
		p := &domain.Project{OwnerID: a.ID, Name: "tx-proj", Status: "active"}
		return quark.ForTx[domain.Project](ctx, tx).Create(p)
	})
	if err != nil {
		return fmt.Errorf("commit: %w", err)
	}
	if c, err := quark.For[domain.Account](ctx, client).Where("email", "=", email).Count(); err != nil || c != 1 {
		return fmt.Errorf("commit no persistió account (count=%d err=%v)", c, err)
	}
	if c, err := quark.For[domain.Project](ctx, client).Where("owner_id", "=", accID).Count(); err != nil || c != 1 {
		return fmt.Errorf("commit no persistió project (count=%d err=%v)", c, err)
	}

	rbEmail := fmt.Sprintf("txrb%d@superapp.test", n)
	sentinel := errors.New("rollback intencional")
	err = client.Tx(ctx, func(tx *quark.Tx) error {
		a := &domain.Account{Email: rbEmail, Name: "rb", Role: "member", Active: true}
		if cerr := quark.ForTx[domain.Account](ctx, tx).Create(a); cerr != nil {
			return cerr
		}
		return sentinel
	})
	if !errors.Is(err, sentinel) {
		return fmt.Errorf("tx con error no propagó el sentinel: %v", err)
	}
	if c, err := quark.For[domain.Account](ctx, client).Where("email", "=", rbEmail).Count(); err != nil || c != 0 {
		return fmt.Errorf("rollback no revirtió (count=%d err=%v)", c, err)
	}
	return nil
}}