Agent-work Queue (in-operator) Implementation Plan¶

// api/v1alpha1/queuedevent_types_test.go
package v1alpha1

import "testing"

func TestValidateQueuedEventSpec(t *testing.T) {
    tests := []struct {
        name    string
        spec    QueuedEventSpec
        wantErr bool
    }{
        {"valid normal repo-scoped", QueuedEventSpec{Seq: 1, Class: QueueClassNormal, Kind: "issueLifecycle", ProjectRef: "p", RepositoryRef: "r", Payload: QueuedEventPayload{Kind: "issueLifecycle"}}, false},
        {"valid alert project-scoped", QueuedEventSpec{Seq: 2, Class: QueueClassAlert, Kind: "incident", ProjectRef: "p", Payload: QueuedEventPayload{Kind: "incident"}}, false},
        {"bad class", QueuedEventSpec{Seq: 1, Class: "urgent", Kind: "incident", ProjectRef: "p"}, true},
        {"bad kind", QueuedEventSpec{Seq: 1, Class: QueueClassNormal, Kind: "nope", ProjectRef: "p"}, true},
        {"missing projectRef", QueuedEventSpec{Seq: 1, Class: QueueClassNormal, Kind: "review"}, true},
        {"project-scoped kind with repoRef", QueuedEventSpec{Seq: 1, Class: QueueClassAlert, Kind: "incident", ProjectRef: "p", RepositoryRef: "r"}, true},
        {"repo-scoped kind without repoRef", QueuedEventSpec{Seq: 1, Class: QueueClassNormal, Kind: "issueLifecycle", ProjectRef: "p"}, true},
        {"zero seq", QueuedEventSpec{Seq: 0, Class: QueueClassNormal, Kind: "review", ProjectRef: "p", RepositoryRef: "r"}, true},
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            err := ValidateQueuedEventSpec(tt.spec)
            if (err != nil) != tt.wantErr {
                t.Fatalf("err=%v wantErr=%v", err, tt.wantErr)
            }
        })
    }
}

// api/v1alpha1/queuedevent_types.go
package v1alpha1

import (
    "fmt"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
    QueueClassNormal = "normal"
    QueueClassAlert  = "alert"

    QueueStateQueued   = "Queued"
    QueueStateAdmitted = "Admitted"
    QueueStateDone     = "Done"
)

// QueuedEventPayload is the Task blueprint a producer stashes; the dispatcher
// rebuilds the Task from it verbatim on admission. Producers keep ownership of
// label/goal/source construction so the dispatcher stays generic.
type QueuedEventPayload struct {
    Goal          string            `json:"goal,omitempty"`
    Kind          string            `json:"kind"`
    RepositoryRef string            `json:"repositoryRef,omitempty"`
    Source        *TaskSource       `json:"source,omitempty"`
    Labels        map[string]string `json:"labels,omitempty"`
    Annotations   map[string]string `json:"annotations,omitempty"`
    // GenerateName is used when Name is empty; Name is a fixed deterministic
    // Task name (issueLifecycle) that makes admission idempotent.
    GenerateName string `json:"generateName,omitempty"`
    Name         string `json:"name,omitempty"`
    // Provider + PodRepo feed agent.StampPodName on the rebuilt Task.
    Provider string `json:"provider,omitempty"`
    PodRepo  string `json:"podRepo,omitempty"`
}

type QueuedEventSpec struct {
    Seq           int64              `json:"seq"`
    Class         string             `json:"class"`
    Kind          string             `json:"kind"`
    Autonomous    bool               `json:"autonomous,omitempty"`
    ProjectRef    string             `json:"projectRef"`
    RepositoryRef string             `json:"repositoryRef,omitempty"`
    DedupKey      string             `json:"dedupKey,omitempty"`
    Payload       QueuedEventPayload `json:"payload"`
}

type QueuedEventStatus struct {
    State      string       `json:"state,omitempty"`
    TaskRef    string       `json:"taskRef,omitempty"`
    AdmittedAt *metav1.Time `json:"admittedAt,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Seq",type=integer,JSONPath=`.spec.seq`
// +kubebuilder:printcolumn:name="Class",type=string,JSONPath=`.spec.class`
// +kubebuilder:printcolumn:name="Kind",type=string,JSONPath=`.spec.kind`
// +kubebuilder:printcolumn:name="State",type=string,JSONPath=`.status.state`
type QueuedEvent struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`
    Spec              QueuedEventSpec   `json:"spec,omitempty"`
    Status            QueuedEventStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true
type QueuedEventList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []QueuedEvent `json:"items"`
}

func init() {
    SchemeBuilder.Register(&QueuedEvent{}, &QueuedEventList{})
}

// ValidateQueuedEventSpec mirrors ValidateTaskSpec's kind/repo-scoping rules.
func ValidateQueuedEventSpec(spec QueuedEventSpec) error {
    if spec.Seq <= 0 {
        return fmt.Errorf("queuedevent: seq must be positive")
    }
    if spec.Class != QueueClassNormal && spec.Class != QueueClassAlert {
        return fmt.Errorf("queuedevent: invalid class %q", spec.Class)
    }
    if spec.ProjectRef == "" {
        return fmt.Errorf("queuedevent: projectRef required")
    }
    if !repoScopedKinds[spec.Kind] && !projectScopedKinds[spec.Kind] {
        return fmt.Errorf("queuedevent: invalid kind %q", spec.Kind)
    }
    if projectScopedKinds[spec.Kind] && spec.RepositoryRef != "" {
        return fmt.Errorf("queuedevent: kind %q is project-scoped, repositoryRef must be empty", spec.Kind)
    }
    if repoScopedKinds[spec.Kind] && spec.RepositoryRef == "" {
        return fmt.Errorf("queuedevent: kind %q requires repositoryRef", spec.Kind)
    }
    return nil
}

// api/v1alpha1/project_types_test.go (add)
func TestQueueDefaults(t *testing.T) {
    // nil Queue: capacity falls back to MaxConcurrentTasks, cap to MaxOpenTasks, alert to 1.
    p := &Project{Spec: ProjectSpec{MaxConcurrentTasks: 5, MaxOpenTasks: 6}}
    if got := p.QueueCapacity(); got != 5 {
        t.Fatalf("QueueCapacity nil-queue = %d, want 5", got)
    }
    if got := p.QueuedAutonomousCap(); got != 6 {
        t.Fatalf("QueuedAutonomousCap nil-queue = %d, want 6", got)
    }
    if got := p.AlertCapacity(); got != 1 {
        t.Fatalf("AlertCapacity nil-queue = %d, want 1", got)
    }
    // explicit Queue overrides.
    p2 := &Project{Spec: ProjectSpec{MaxConcurrentTasks: 5, Queue: &QueueSpec{Capacity: 2, AlertCapacity: 3, QueuedAutonomousCap: 10}}}
    if p2.QueueCapacity() != 2 || p2.AlertCapacity() != 3 || p2.QueuedAutonomousCap() != 10 {
        t.Fatalf("explicit queue not honoured: %d/%d/%d", p2.QueueCapacity(), p2.AlertCapacity(), p2.QueuedAutonomousCap())
    }
    // hard floor when nothing set anywhere.
    p3 := &Project{}
    if p3.QueueCapacity() != 3 || p3.QueuedAutonomousCap() != 3 || p3.AlertCapacity() != 1 {
        t.Fatalf("hard floors wrong: %d/%d/%d", p3.QueueCapacity(), p3.QueuedAutonomousCap(), p3.AlertCapacity())
    }
}

// api/v1alpha1/project_types.go

// in ProjectSpec, sibling of Grafana:
//   // +optional
//   Queue *QueueSpec `json:"queue,omitempty"`

// QueueSpec configures the in-operator agent-work admission queue.
type QueueSpec struct {
    // Capacity N: max concurrently-admitted normal-class events (defaults to
    // MaxConcurrentTasks, else 3).
    // +optional
    Capacity int `json:"capacity,omitempty"`
    // AlertCapacity M: reserved concurrent slots for alert-class events (default 1).
    // +optional
    AlertCapacity int `json:"alertCapacity,omitempty"`
    // QueuedAutonomousCap K: max Queued autonomous (cron) events; crons stop
    // enqueuing past it (defaults to MaxOpenTasks, else 3). Webhooks/alerts exempt.
    // +optional
    QueuedAutonomousCap int `json:"queuedAutonomousCap,omitempty"`
}

func (p *Project) QueueCapacity() int {
    if p.Spec.Queue != nil && p.Spec.Queue.Capacity > 0 {
        return p.Spec.Queue.Capacity
    }
    if p.Spec.MaxConcurrentTasks > 0 {
        return p.Spec.MaxConcurrentTasks
    }
    return 3
}

func (p *Project) AlertCapacity() int {
    if p.Spec.Queue != nil && p.Spec.Queue.AlertCapacity > 0 {
        return p.Spec.Queue.AlertCapacity
    }
    return 1
}

func (p *Project) QueuedAutonomousCap() int {
    if p.Spec.Queue != nil && p.Spec.Queue.QueuedAutonomousCap > 0 {
        return p.Spec.Queue.QueuedAutonomousCap
    }
    if p.Spec.MaxOpenTasks > 0 {
        return p.Spec.MaxOpenTasks
    }
    return 3
}

// internal/queue/allocator_test.go
package queue

import (
    "sync"
    "testing"
)

func TestSeqAllocator_MonotonicFromZero(t *testing.T) {
    a := NewSeqAllocator()
    if a.Next() != 1 || a.Next() != 2 || a.Next() != 3 {
        t.Fatal("expected 1,2,3 from a fresh allocator")
    }
}

func TestSeqAllocator_RecoverMaxPlusOne(t *testing.T) {
    a := NewSeqAllocator()
    a.Recover(41)
    if got := a.Next(); got != 42 {
        t.Fatalf("Next after Recover(41) = %d, want 42", got)
    }
}

func TestSeqAllocator_ConcurrentUnique(t *testing.T) {
    a := NewSeqAllocator()
    const n = 1000
    seen := make([]int64, n)
    var wg sync.WaitGroup
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func(i int) { defer wg.Done(); seen[i] = a.Next() }(i)
    }
    wg.Wait()
    set := map[int64]bool{}
    for _, v := range seen {
        if set[v] {
            t.Fatalf("duplicate seq %d", v)
        }
        set[v] = true
    }
}

// internal/queue/allocator.go
package queue

import (
    "context"
    "sync"

    tatarav1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"
)

// SeqAllocator hands out a strictly increasing int64 sequence. Correctness
// relies on a single leader-elected active operator (one allocator instance).
type SeqAllocator struct {
    mu   sync.Mutex
    next int64
}

func NewSeqAllocator() *SeqAllocator { return &SeqAllocator{next: 0} }

// Recover sets the counter so the next allocation is maxSeq+1. Call once at boot
// with the max Seq of existing QueuedEvents.
func (a *SeqAllocator) Recover(maxSeq int64) {
    a.mu.Lock()
    defer a.mu.Unlock()
    if maxSeq > a.next {
        a.next = maxSeq
    }
}

func (a *SeqAllocator) Next() int64 {
    a.mu.Lock()
    defer a.mu.Unlock()
    a.next++
    return a.next
}

// SeqRecoverer is a manager.Runnable that recovers the allocator high-water mark
// from existing QueuedEvents after the cache syncs.
type SeqRecoverer struct {
    Client    client.Client
    Alloc     *SeqAllocator
    Namespace string
}

func (s *SeqRecoverer) Start(ctx context.Context) error {
    var list tatarav1alpha1.QueuedEventList
    if err := s.Client.List(ctx, &list, client.InNamespace(s.Namespace)); err != nil {
        return err
    }
    var max int64
    for i := range list.Items {
        if list.Items[i].Spec.Seq > max {
            max = list.Items[i].Spec.Seq
        }
    }
    s.Alloc.Recover(max)
    log.FromContext(ctx).Info("queue: seq recovered", "action", "seq_recover", "max", max)
    return nil
}

// internal/controller/queue_enqueue_test.go
package controller

import (
    "context"
    "testing"

    tatarav1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "github.com/szymonrychu/tatara-operator/internal/queue"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func testProject(name, ns string) *tatarav1alpha1.Project {
    return &tatarav1alpha1.Project{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}}
}

func TestEnqueueEvent_AssignsSeqAndFields(t *testing.T) {
    scheme := newTestScheme(t) // helper already present in suite_test.go / export_test.go
    c := fake.NewClientBuilder().WithScheme(scheme).Build()
    alloc := queue.NewSeqAllocator()
    proj := testProject("p", "tatara")
    pl := tatarav1alpha1.QueuedEventPayload{Kind: "incident", GenerateName: "incident-"}
    qe, created, err := EnqueueEvent(context.Background(), c, alloc, proj, tatarav1alpha1.QueueClassAlert, false, "grp1", pl)
    if err != nil || !created {
        t.Fatalf("created=%v err=%v", created, err)
    }
    if qe.Spec.Seq != 1 || qe.Spec.Class != tatarav1alpha1.QueueClassAlert || qe.Spec.Kind != "incident" {
        t.Fatalf("bad spec: %+v", qe.Spec)
    }
    if qe.Labels[LabelDedupKey] != "grp1" || qe.Status.State != tatarav1alpha1.QueueStateQueued {
        t.Fatalf("bad labels/state: %v %q", qe.Labels, qe.Status.State)
    }
}

func TestEnqueueEvent_DedupSkips(t *testing.T) {
    scheme := newTestScheme(t)
    c := fake.NewClientBuilder().WithScheme(scheme).Build()
    alloc := queue.NewSeqAllocator()
    proj := testProject("p", "tatara")
    pl := tatarav1alpha1.QueuedEventPayload{Kind: "incident", GenerateName: "incident-"}
    if _, created, _ := EnqueueEvent(context.Background(), c, alloc, proj, tatarav1alpha1.QueueClassAlert, false, "grp1", pl); !created {
        t.Fatal("first enqueue should create")
    }
    _, created, err := EnqueueEvent(context.Background(), c, alloc, proj, tatarav1alpha1.QueueClassAlert, false, "grp1", pl)
    if err != nil {
        t.Fatal(err)
    }
    if created {
        t.Fatal("duplicate dedupKey should skip")
    }
}

func TestBuildTaskFromQueuedEvent(t *testing.T) {
    scheme := newTestScheme(t)
    proj := testProject("p", "tatara")
    qe := &tatarav1alpha1.QueuedEvent{
        ObjectMeta: metav1.ObjectMeta{Name: "qe-1", Namespace: "tatara"},
        Spec: tatarav1alpha1.QueuedEventSpec{
            Seq: 1, Class: tatarav1alpha1.QueueClassNormal, Kind: "review", ProjectRef: "p", RepositoryRef: "r",
            Payload: tatarav1alpha1.QueuedEventPayload{
                Kind: "review", RepositoryRef: "r", Goal: "g", GenerateName: "scan-",
                Labels: map[string]string{"x": "y"}, Provider: "github", PodRepo: "r",
            },
        },
    }
    task, err := buildTaskFromQueuedEvent(qe, proj, scheme)
    if err != nil {
        t.Fatal(err)
    }
    if task.Spec.Kind != "review" || task.Spec.Goal != "g" || task.Spec.RepositoryRef != "r" {
        t.Fatalf("bad task spec: %+v", task.Spec)
    }
    if task.Labels[LabelQueuedEvent] != "qe-1" || task.Labels["x"] != "y" {
        t.Fatalf("missing labels: %v", task.Labels)
    }
    if task.GenerateName != "scan-" {
        t.Fatalf("bad generateName: %q", task.GenerateName)
    }
}

// internal/controller/queue_enqueue.go
package controller

import (
    "context"
    "fmt"

    tatarav1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "github.com/szymonrychu/tatara-operator/internal/agent"
    "github.com/szymonrychu/tatara-operator/internal/queue"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)

const (
    LabelQueuedEvent = "tatara.dev/queued-event"
    LabelDedupKey    = "tatara.dev/dedup-key"
)

// dedupExists reports whether a non-Done QueuedEvent or a non-terminal Task with
// dedupKey already exists for the project.
func dedupExists(ctx context.Context, c client.Client, ns, projectRef, dedupKey string) (bool, error) {
    if dedupKey == "" {
        return false, nil
    }
    var qel tatarav1alpha1.QueuedEventList
    if err := c.List(ctx, &qel, client.InNamespace(ns), client.MatchingLabels{LabelDedupKey: dedupKey}); err != nil {
        return false, err
    }
    for i := range qel.Items {
        if qel.Items[i].Spec.ProjectRef == projectRef && qel.Items[i].Status.State != tatarav1alpha1.QueueStateDone {
            return true, nil
        }
    }
    var tl tatarav1alpha1.TaskList
    if err := c.List(ctx, &tl, client.InNamespace(ns), client.MatchingLabels{LabelDedupKey: dedupKey}); err != nil {
        return false, err
    }
    for i := range tl.Items {
        if tl.Items[i].Spec.ProjectRef == projectRef && !tatarav1alpha1.TaskTerminal(&tl.Items[i]) {
            return true, nil
        }
    }
    return false, nil
}

// EnqueueEvent writes a QueuedEvent (seq-assigned, owned by Project, state=Queued).
// Returns created=false when dedupKey already has live work.
func EnqueueEvent(ctx context.Context, c client.Client, alloc *queue.SeqAllocator, proj *tatarav1alpha1.Project,
    class string, autonomous bool, dedupKey string, payload tatarav1alpha1.QueuedEventPayload) (*tatarav1alpha1.QueuedEvent, bool, error) {

    dup, err := dedupExists(ctx, c, proj.Namespace, proj.Name, dedupKey)
    if err != nil {
        return nil, false, err
    }
    if dup {
        return nil, false, nil
    }
    labels := map[string]string{}
    if dedupKey != "" {
        labels[LabelDedupKey] = dedupKey
    }
    qe := &tatarav1alpha1.QueuedEvent{
        ObjectMeta: metav1.ObjectMeta{
            GenerateName: "qe-",
            Namespace:    proj.Namespace,
            Labels:       labels,
        },
        Spec: tatarav1alpha1.QueuedEventSpec{
            Seq:           alloc.Next(),
            Class:         class,
            Kind:          payload.Kind,
            Autonomous:    autonomous,
            ProjectRef:    proj.Name,
            RepositoryRef: payload.RepositoryRef,
            DedupKey:      dedupKey,
            Payload:       payload,
        },
    }
    if err := controllerutil.SetControllerReference(proj, qe, c.Scheme()); err != nil {
        return nil, false, fmt.Errorf("enqueue: set ownerref: %w", err)
    }
    if err := c.Create(ctx, qe); err != nil {
        return nil, false, fmt.Errorf("enqueue: create queuedevent: %w", err)
    }
    qe.Status.State = tatarav1alpha1.QueueStateQueued
    if err := c.Status().Update(ctx, qe); err != nil {
        return nil, false, fmt.Errorf("enqueue: set state: %w", err)
    }
    return qe, true, nil
}

// buildTaskFromQueuedEvent reconstructs the Task the producer described, labelled
// with the QueuedEvent name (dispatcher completion mapping) and dedup key.
func buildTaskFromQueuedEvent(qe *tatarav1alpha1.QueuedEvent, proj *tatarav1alpha1.Project, scheme *runtime.Scheme) (*tatarav1alpha1.Task, error) {
    p := qe.Spec.Payload
    labels := map[string]string{}
    for k, v := range p.Labels {
        labels[k] = v
    }
    labels[LabelQueuedEvent] = qe.Name
    if qe.Spec.DedupKey != "" {
        labels[LabelDedupKey] = qe.Spec.DedupKey
    }
    om := metav1.ObjectMeta{
        Namespace:   qe.Namespace,
        Labels:      labels,
        Annotations: p.Annotations,
    }
    if p.Name != "" {
        om.Name = p.Name
    } else {
        om.GenerateName = p.GenerateName
    }
    task := &tatarav1alpha1.Task{
        ObjectMeta: om,
        Spec: tatarav1alpha1.TaskSpec{
            ProjectRef:    proj.Name,
            RepositoryRef: p.RepositoryRef,
            Goal:          p.Goal,
            Kind:          p.Kind,
            Source:        p.Source,
        },
    }
    agent.StampPodName(task, proj.Name, p.Provider, p.PodRepo)
    if err := controllerutil.SetControllerReference(proj, task, scheme); err != nil {
        return nil, fmt.Errorf("build task: set ownerref: %w", err)
    }
    return task, nil
}

// internal/controller/queue_controller_test.go
package controller

import (
    "testing"

    tatarav1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func qe(name, class, state, taskRef string) tatarav1alpha1.QueuedEvent {
    return tatarav1alpha1.QueuedEvent{
        ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "tatara"},
        Spec:       tatarav1alpha1.QueuedEventSpec{Class: class, ProjectRef: "p"},
        Status:     tatarav1alpha1.QueuedEventStatus{State: state, TaskRef: taskRef},
    }
}

func tk(name, phase, lifecycle, queuedEvent string) tatarav1alpha1.Task {
    return tatarav1alpha1.Task{
        ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "tatara", Labels: map[string]string{LabelQueuedEvent: queuedEvent}},
        Spec:       tatarav1alpha1.TaskSpec{ProjectRef: "p"},
        Status:     tatarav1alpha1.TaskStatus{Phase: phase, LifecycleState: lifecycle},
    }
}

func TestPoolInflight_CountsAdmittedNonTerminal(t *testing.T) {
    r := &DispatcherReconciler{}
    qes := []tatarav1alpha1.QueuedEvent{
        qe("a", tatarav1alpha1.QueueClassNormal, tatarav1alpha1.QueueStateAdmitted, "t-a"), // running -> counts
        qe("b", tatarav1alpha1.QueueClassNormal, tatarav1alpha1.QueueStateAdmitted, "t-b"), // terminal -> not
        qe("c", tatarav1alpha1.QueueClassAlert, tatarav1alpha1.QueueStateAdmitted, "t-c"),  // alert running
        qe("d", tatarav1alpha1.QueueClassNormal, tatarav1alpha1.QueueStateQueued, ""),      // queued -> not
    }
    tasks := []tatarav1alpha1.Task{
        tk("t-a", "Running", "", "a"),
        tk("t-b", "Succeeded", "", "b"),
        tk("t-c", "Running", "", "c"),
    }
    if got := r.poolInflight(qes, tasks, tatarav1alpha1.QueueClassNormal); got != 1 {
        t.Fatalf("normal inflight = %d, want 1", got)
    }
    if got := r.poolInflight(qes, tasks, tatarav1alpha1.QueueClassAlert); got != 1 {
        t.Fatalf("alert inflight = %d, want 1", got)
    }
}

// internal/controller/queue_controller.go
package controller

import (
    "context"

    tatarav1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "github.com/szymonrychu/tatara-operator/internal/obs"
    "github.com/szymonrychu/tatara-operator/internal/queue"
    "k8s.io/apimachinery/pkg/runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"
)

type DispatcherReconciler struct {
    client.Client
    Scheme  *runtime.Scheme
    Alloc   *queue.SeqAllocator
    Metrics *obs.OperatorMetrics
}

func taskByName(tasks []tatarav1alpha1.Task, name string) *tatarav1alpha1.Task {
    for i := range tasks {
        if tasks[i].Name == name {
            return &tasks[i]
        }
    }
    return nil
}

// poolInflight counts Admitted QueuedEvents of class whose Task is still
// non-terminal. (Migration of unlabelled pre-queue Tasks is added in Task 7.)
func (r *DispatcherReconciler) poolInflight(qes []tatarav1alpha1.QueuedEvent, tasks []tatarav1alpha1.Task, class string) int {
    n := 0
    for i := range qes {
        q := &qes[i]
        if q.Spec.Class != class || q.Status.State != tatarav1alpha1.QueueStateAdmitted {
            continue
        }
        t := taskByName(tasks, q.Status.TaskRef)
        if t != nil && !tatarav1alpha1.TaskTerminal(t) {
            n++
        }
    }
    return n
}

// reconcileDone flips Admitted events whose Task is terminal or gone to Done.
func (r *DispatcherReconciler) reconcileDone(ctx context.Context, qes []tatarav1alpha1.QueuedEvent, tasks []tatarav1alpha1.Task) (bool, error) {
    changed := false
    for i := range qes {
        q := &qes[i]
        if q.Status.State != tatarav1alpha1.QueueStateAdmitted {
            continue
        }
        t := taskByName(tasks, q.Status.TaskRef)
        if t == nil || tatarav1alpha1.TaskTerminal(t) {
            q.Status.State = tatarav1alpha1.QueueStateDone
            if err := r.Status().Update(ctx, q); err != nil {
                return changed, err
            }
            changed = true
            log.FromContext(ctx).Info("queue: event done", "action", "queue_done", "resource_id", q.Name, "class", q.Spec.Class)
        }
    }
    return changed, nil
}

// internal/controller/queue_controller_test.go (add; uses the envtest k8sClient from suite_test.go)
func TestAdmit_AlertBeforeNormal_AndCapacity(t *testing.T) {
    ctx := context.Background()
    ns := "tatara"
    proj := &tatarav1alpha1.Project{
        ObjectMeta: metav1.ObjectMeta{Name: "p-admit", Namespace: ns},
        Spec:       tatarav1alpha1.ProjectSpec{Queue: &tatarav1alpha1.QueueSpec{Capacity: 1, AlertCapacity: 1}},
    }
    mustCreate(t, ctx, proj) // helper: k8sClient.Create + cleanup

    mkQE := func(seq int64, class string) *tatarav1alpha1.QueuedEvent {
        q := &tatarav1alpha1.QueuedEvent{
            ObjectMeta: metav1.ObjectMeta{GenerateName: "qe-", Namespace: ns},
            Spec: tatarav1alpha1.QueuedEventSpec{
                Seq: seq, Class: class, Kind: "incident", ProjectRef: proj.Name,
                Payload: tatarav1alpha1.QueuedEventPayload{Kind: "incident", GenerateName: "x-"},
            },
        }
        mustCreate(t, ctx, q)
        q.Status.State = tatarav1alpha1.QueueStateQueued
        mustStatusUpdate(t, ctx, q)
        return q
    }
    normalQE := mkQE(1, tatarav1alpha1.QueueClassNormal) // older
    alertQE := mkQE(2, tatarav1alpha1.QueueClassAlert)   // newer but priority pool

    r := &DispatcherReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()}
    qes, tasks := listQEsTasks(t, ctx, proj.Name) // helper
    if err := r.admit(ctx, proj, qes, tasks); err != nil {
        t.Fatal(err)
    }

    // Both pools have capacity 1: one alert + one normal admitted.
    got := refreshQE(t, ctx, alertQE)
    if got.Status.State != tatarav1alpha1.QueueStateAdmitted || got.Status.TaskRef == "" {
        t.Fatalf("alert not admitted: %+v", got.Status)
    }
    gotN := refreshQE(t, ctx, normalQE)
    if gotN.Status.State != tatarav1alpha1.QueueStateAdmitted {
        t.Fatalf("normal not admitted: %+v", gotN.Status)
    }
}

// internal/controller/queue_controller.go (add)
import "sort" // add to the import block

// admit drains the alert pool to AlertCapacity, then the normal pool to
// QueueCapacity, each in strict ascending seq order (pure global FIFO within a
// pool; head-of-line blocking accepted).
func (r *DispatcherReconciler) admit(ctx context.Context, proj *tatarav1alpha1.Project,
    qes []tatarav1alpha1.QueuedEvent, tasks []tatarav1alpha1.Task) error {

    admitPool := func(class string, cap int) error {
        inflight := r.poolInflight(qes, tasks, class)
        queued := make([]*tatarav1alpha1.QueuedEvent, 0)
        for i := range qes {
            if qes[i].Spec.Class == class && qes[i].Status.State == tatarav1alpha1.QueueStateQueued {
                queued = append(queued, &qes[i])
            }
        }
        sort.Slice(queued, func(i, j int) bool { return queued[i].Spec.Seq < queued[j].Spec.Seq })
        for _, q := range queued {
            if inflight >= cap {
                break
            }
            task, err := buildTaskFromQueuedEvent(q, proj, r.Scheme)
            if err != nil {
                return err
            }
            if err := r.Create(ctx, task); err != nil {
                // Leave Queued; requeue. Slot not consumed (inflight derives from Admitted).
                return err
            }
            q.Status.State = tatarav1alpha1.QueueStateAdmitted
            q.Status.TaskRef = task.Name
            now := metav1.Now()
            q.Status.AdmittedAt = &now
            if err := r.Status().Update(ctx, q); err != nil {
                return err
            }
            inflight++
            if r.Metrics != nil {
                r.Metrics.QueueAdmitted(class, q.Spec.Kind)
            }
            log.FromContext(ctx).Info("queue: admitted",
                "action", "queue_admit", "resource_id", q.Name, "task", task.Name,
                "class", class, "seq", q.Spec.Seq, "kind", q.Spec.Kind)
        }
        return nil
    }

    if err := admitPool(tatarav1alpha1.QueueClassAlert, proj.AlertCapacity()); err != nil {
        return err
    }
    return admitPool(tatarav1alpha1.QueueClassNormal, proj.QueueCapacity())
}