tatara-operator M0 (scaffold) Implementation Plan¶

package version_test

import (
    "testing"

    "github.com/szymonrychu/tatara-operator/internal/version"
)

func TestString(t *testing.T) {
    tests := []struct {
        name    string
        version string
        commit  string
        date    string
        want    string
    }{
        {name: "defaults", version: "dev", commit: "unknown", date: "unknown", want: "dev (unknown, unknown)"},
        {name: "release", version: "v1.2.3", commit: "abc123", date: "2026-06-06T00:00:00Z", want: "v1.2.3 (abc123, 2026-06-06T00:00:00Z)"},
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            version.Version = tt.version
            version.Commit = tt.commit
            version.Date = tt.date
            if got := version.String(); got != tt.want {
                t.Fatalf("String() = %q, want %q", got, tt.want)
            }
        })
    }
}

package version

import "fmt"

var (
    Version = "dev"
    Commit  = "unknown"
    Date    = "unknown"
)

func String() string {
    return fmt.Sprintf("%s (%s, %s)", Version, Commit, Date)
}

package v1alpha1_test

import (
    "testing"

    "github.com/szymonrychu/tatara-operator/api/v1alpha1"
)

func TestGroupVersion(t *testing.T) {
    if v1alpha1.GroupVersion.Group != "tatara.dev" {
        t.Fatalf("Group = %q, want tatara.dev", v1alpha1.GroupVersion.Group)
    }
    if v1alpha1.GroupVersion.Version != "v1alpha1" {
        t.Fatalf("Version = %q, want v1alpha1", v1alpha1.GroupVersion.Version)
    }
}

func TestAddToScheme(t *testing.T) {
    if v1alpha1.SchemeBuilder.GroupVersion.String() != "tatara.dev/v1alpha1" {
        t.Fatalf("SchemeBuilder.GroupVersion = %q, want tatara.dev/v1alpha1", v1alpha1.SchemeBuilder.GroupVersion.String())
    }
    if v1alpha1.AddToScheme == nil {
        t.Fatal("AddToScheme is nil")
    }
}

// Package v1alpha1 contains API Schema definitions for the tatara.dev v1alpha1 API group.
// +kubebuilder:object:generate=true
// +groupName=tatara.dev
package v1alpha1

import (
    "k8s.io/apimachinery/pkg/runtime/schema"
    "sigs.k8s.io/controller-runtime/pkg/scheme"
)

var (
    // GroupVersion is group version used to register these objects.
    GroupVersion = schema.GroupVersion{Group: "tatara.dev", Version: "v1alpha1"}

    // SchemeBuilder is used to add go types to the GroupVersionKind scheme.
    SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion}

    // AddToScheme adds the types in this group-version to the given scheme.
    AddToScheme = SchemeBuilder.AddToScheme
)

// +kubebuilder:object:generate=true
// +groupName=tatara.dev

// Package v1alpha1 holds the Project, Repository, Task, and Subtask custom
// resources for the tatara-operator. All kinds are namespaced.
package v1alpha1

package v1alpha1_test

import (
    "testing"

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

    "github.com/szymonrychu/tatara-operator/api/v1alpha1"
)

func TestProjectFields(t *testing.T) {
    p := v1alpha1.Project{
        Spec: v1alpha1.ProjectSpec{
            ScmSecretRef:       "scm-secret",
            TriggerLabel:       "tatara",
            MaxConcurrentTasks: 3,
            Agent: v1alpha1.AgentSpec{
                Model:              "claude-sonnet-4-6",
                Image:              "wrapper:latest",
                PermissionMode:     "bypassPermissions",
                MaxTurnsPerTask:    50,
                TurnTimeoutSeconds: 1800,
            },
        },
        Status: v1alpha1.ProjectStatus{
            WebhookURL: "https://example/operator/webhooks/p",
            Conditions: []metav1.Condition{{Type: "Ready", Status: metav1.ConditionTrue}},
        },
    }
    if p.Spec.Agent.MaxTurnsPerTask != 50 {
        t.Fatalf("MaxTurnsPerTask = %d, want 50", p.Spec.Agent.MaxTurnsPerTask)
    }
    if p.Status.WebhookURL == "" {
        t.Fatal("WebhookURL empty")
    }
}

func TestRepositoryFields(t *testing.T) {
    r := v1alpha1.Repository{
        Spec: v1alpha1.RepositorySpec{
            ProjectRef:    "p",
            URL:           "https://example/repo.git",
            DefaultBranch: "main",
            IngestEnabled: true,
        },
        Status: v1alpha1.RepositoryStatus{
            Phase:              "Ingested",
            LastIngestedCommit: "abc123",
            JobName:            "ingest-1",
        },
    }
    if r.Spec.DefaultBranch != "main" {
        t.Fatalf("DefaultBranch = %q, want main", r.Spec.DefaultBranch)
    }
    if r.Status.Phase != "Ingested" {
        t.Fatalf("Phase = %q, want Ingested", r.Status.Phase)
    }
}

func TestProjectRegisteredInScheme(t *testing.T) {
    s := runtime.NewScheme()
    if err := v1alpha1.AddToScheme(s); err != nil {
        t.Fatalf("AddToScheme: %v", err)
    }
    if !s.Recognizes(v1alpha1.GroupVersion.WithKind("Project")) {
        t.Fatal("Project kind not recognized by scheme")
    }
    if !s.Recognizes(v1alpha1.GroupVersion.WithKind("Repository")) {
        t.Fatal("Repository kind not recognized by scheme")
    }
}

package v1alpha1

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

// AgentSpec configures the wrapper agent session a Task runs.
type AgentSpec struct {
    // +optional
    Model string `json:"model,omitempty"`
    // +optional
    Image string `json:"image,omitempty"`
    // +kubebuilder:default="bypassPermissions"
    // +optional
    PermissionMode string `json:"permissionMode,omitempty"`
    // +kubebuilder:default=50
    // +optional
    MaxTurnsPerTask int `json:"maxTurnsPerTask,omitempty"`
    // +kubebuilder:default=1800
    // +optional
    TurnTimeoutSeconds int `json:"turnTimeoutSeconds,omitempty"`
}

// ProjectSpec defines the desired state of a Project.
type ProjectSpec struct {
    ScmSecretRef string `json:"scmSecretRef"`
    // +kubebuilder:default="tatara"
    // +optional
    TriggerLabel string `json:"triggerLabel,omitempty"`
    // +kubebuilder:default=3
    // +optional
    MaxConcurrentTasks int `json:"maxConcurrentTasks,omitempty"`
    // +optional
    Agent AgentSpec `json:"agent,omitempty"`
}

// ProjectStatus defines the observed state of a Project.
type ProjectStatus struct {
    // +optional
    WebhookURL string `json:"webhookURL,omitempty"`
    // +optional
    // +listType=map
    // +listMapKey=type
    Conditions []metav1.Condition `json:"conditions,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:scope=Namespaced
// +kubebuilder:printcolumn:name="Webhook",type=string,JSONPath=`.status.webhookURL`

// Project is the top-level grouping for repositories and tasks.
type Project struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   ProjectSpec   `json:"spec,omitempty"`
    Status ProjectStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// ProjectList contains a list of Project.
type ProjectList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []Project `json:"items"`
}

func init() {
    SchemeBuilder.Register(&Project{}, &ProjectList{})
}

package v1alpha1

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

// RepositorySpec defines the desired state of a Repository.
type RepositorySpec struct {
    ProjectRef string `json:"projectRef"`
    URL        string `json:"url"`
    // +kubebuilder:default="main"
    // +optional
    DefaultBranch string `json:"defaultBranch,omitempty"`
    // +kubebuilder:default=true
    // +optional
    IngestEnabled bool `json:"ingestEnabled,omitempty"`
}

// RepositoryStatus defines the observed state of a Repository.
type RepositoryStatus struct {
    // +kubebuilder:validation:Enum=Pending;Ingesting;Ingested;Failed
    // +optional
    Phase string `json:"phase,omitempty"`
    // +optional
    LastIngestedCommit string `json:"lastIngestedCommit,omitempty"`
    // +optional
    LastIngestTime *metav1.Time `json:"lastIngestTime,omitempty"`
    // +optional
    JobName string `json:"jobName,omitempty"`
    // +optional
    // +listType=map
    // +listMapKey=type
    Conditions []metav1.Condition `json:"conditions,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:scope=Namespaced
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
// +kubebuilder:printcolumn:name="Commit",type=string,JSONPath=`.status.lastIngestedCommit`

// Repository is a git remote ingested into tatara-memory for a Project.
type Repository struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   RepositorySpec   `json:"spec,omitempty"`
    Status RepositoryStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// RepositoryList contains a list of Repository.
type RepositoryList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []Repository `json:"items"`
}

func init() {
    SchemeBuilder.Register(&Repository{}, &RepositoryList{})
}

package v1alpha1_test

import (
    "testing"

    "k8s.io/apimachinery/pkg/runtime"

    "github.com/szymonrychu/tatara-operator/api/v1alpha1"
)

func TestTaskFields(t *testing.T) {
    task := v1alpha1.Task{
        Spec: v1alpha1.TaskSpec{
            ProjectRef:    "p",
            RepositoryRef: "r",
            Goal:          "do the thing",
            Source: &v1alpha1.TaskSource{
                Provider: "github",
                IssueRef: "owner/repo#123",
                URL:      "https://github.com/owner/repo/issues/123",
            },
            MaxTurns: 25,
        },
        Status: v1alpha1.TaskStatus{
            Phase:          "Running",
            PodName:        "task-p-1",
            TurnsCompleted: 4,
            PrURL:          "https://github.com/owner/repo/pull/5",
            ResultSummary:  "opened PR",
        },
    }
    if task.Spec.Source.Provider != "github" {
        t.Fatalf("Source.Provider = %q, want github", task.Spec.Source.Provider)
    }
    if task.Status.TurnsCompleted != 4 {
        t.Fatalf("TurnsCompleted = %d, want 4", task.Status.TurnsCompleted)
    }
}

func TestSubtaskFields(t *testing.T) {
    s := v1alpha1.Subtask{
        Spec: v1alpha1.SubtaskSpec{
            TaskRef: "task-p-1",
            Title:   "write test",
            Detail:  "add the failing test",
            Order:   1,
        },
        Status: v1alpha1.SubtaskStatus{
            Phase:  "Done",
            TurnID: "turn-abc",
            Result: "test added",
        },
    }
    if s.Spec.Order != 1 {
        t.Fatalf("Order = %d, want 1", s.Spec.Order)
    }
    if s.Status.TurnID != "turn-abc" {
        t.Fatalf("TurnID = %q, want turn-abc", s.Status.TurnID)
    }
}

func TestTaskAndSubtaskRegisteredInScheme(t *testing.T) {
    sch := runtime.NewScheme()
    if err := v1alpha1.AddToScheme(sch); err != nil {
        t.Fatalf("AddToScheme: %v", err)
    }
    for _, kind := range []string{"Task", "Subtask"} {
        if !sch.Recognizes(v1alpha1.GroupVersion.WithKind(kind)) {
            t.Fatalf("%s kind not recognized by scheme", kind)
        }
    }
}

package v1alpha1

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

// TaskSource records the SCM work-item that originated a webhook-born Task.
type TaskSource struct {
    // +kubebuilder:validation:Enum=github;gitlab
    Provider string `json:"provider"`
    IssueRef string `json:"issueRef"`
    // +optional
    URL string `json:"url,omitempty"`
}

// TaskSpec defines the desired state of a Task.
type TaskSpec struct {
    ProjectRef    string `json:"projectRef"`
    RepositoryRef string `json:"repositoryRef"`
    Goal          string `json:"goal"`
    // +optional
    Source *TaskSource `json:"source,omitempty"`
    // +optional
    MaxTurns int `json:"maxTurns,omitempty"`
}

// TaskStatus defines the observed state of a Task.
type TaskStatus struct {
    // +kubebuilder:validation:Enum=Pending;Planning;Running;Succeeded;Failed
    // +optional
    Phase string `json:"phase,omitempty"`
    // +optional
    PodName string `json:"podName,omitempty"`
    // +optional
    TurnsCompleted int `json:"turnsCompleted,omitempty"`
    // +optional
    PrURL string `json:"prURL,omitempty"`
    // +optional
    ResultSummary string `json:"resultSummary,omitempty"`
    // +optional
    // +listType=map
    // +listMapKey=type
    Conditions []metav1.Condition `json:"conditions,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:scope=Namespaced
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
// +kubebuilder:printcolumn:name="Turns",type=integer,JSONPath=`.status.turnsCompleted`

// Task is one agent session driving a Repository toward a goal.
type Task struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   TaskSpec   `json:"spec,omitempty"`
    Status TaskStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// TaskList contains a list of Task.
type TaskList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []Task `json:"items"`
}

func init() {
    SchemeBuilder.Register(&Task{}, &TaskList{})
}

package v1alpha1

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

// SubtaskSpec defines the desired state of a Subtask. Created/updated by the
// agent via the REST API and consumed by the Task reconciler's turn loop.
type SubtaskSpec struct {
    TaskRef string `json:"taskRef"`
    Title   string `json:"title"`
    // +optional
    Detail string `json:"detail,omitempty"`
    // +optional
    Order int `json:"order,omitempty"`
}

// SubtaskStatus defines the observed state of a Subtask.
type SubtaskStatus struct {
    // +kubebuilder:validation:Enum=Pending;Running;Done;Failed
    // +optional
    Phase string `json:"phase,omitempty"`
    // +optional
    TurnID string `json:"turnId,omitempty"`
    // +optional
    Result string `json:"result,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:scope=Namespaced
// +kubebuilder:printcolumn:name="Order",type=integer,JSONPath=`.spec.order`
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`

// Subtask is a unit of work fed to a Task's agent session one turn at a time.
type Subtask struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   SubtaskSpec   `json:"spec,omitempty"`
    Status SubtaskStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// SubtaskList contains a list of Subtask.
type SubtaskList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []Subtask `json:"items"`
}

func init() {
    SchemeBuilder.Register(&Subtask{}, &SubtaskList{})
}

package config_test

import (
    "testing"

    "github.com/szymonrychu/tatara-operator/internal/config"
)

func TestLoad(t *testing.T) {
    env := map[string]string{
        "HTTP_ADDR":                   ":8080",
        "METRICS_ADDR":                ":9090",
        "INTERNAL_ADDR":               ":8081",
        "OIDC_ISSUER":                 "https://kc/realms/tatara",
        "OIDC_AUDIENCE":               "tatara-operator",
        "MEMORY_BASE_URL":             "http://tatara-memory:8080",
        "INGESTER_IMAGE":              "harbor/ingester:1",
        "EXTERNAL_WEBHOOK_BASE":       "https://ops.example",
        "OPERATOR_OIDC_CLIENT_ID":     "tatara-operator",
        "OPERATOR_OIDC_CLIENT_SECRET": "shh",
        "ANTHROPIC_SECRET_NAME":       "anthropic",
        "CLI_OIDC_SECRET_NAME":        "cli-oidc",
        "LOG_LEVEL":                   "debug",
    }
    for k, v := range env {
        t.Setenv(k, v)
    }

    cfg, err := config.Load()
    if err != nil {
        t.Fatalf("Load: %v", err)
    }

    tests := []struct {
        name string
        got  string
        want string
    }{
        {"HTTPAddr", cfg.HTTPAddr, ":8080"},
        {"MetricsAddr", cfg.MetricsAddr, ":9090"},
        {"InternalAddr", cfg.InternalAddr, ":8081"},
        {"OIDCIssuer", cfg.OIDCIssuer, "https://kc/realms/tatara"},
        {"OIDCAudience", cfg.OIDCAudience, "tatara-operator"},
        {"MemoryBaseURL", cfg.MemoryBaseURL, "http://tatara-memory:8080"},
        {"IngesterImage", cfg.IngesterImage, "harbor/ingester:1"},
        {"ExternalWebhookBase", cfg.ExternalWebhookBase, "https://ops.example"},
        {"OperatorOIDCClientID", cfg.OperatorOIDCClientID, "tatara-operator"},
        {"OperatorOIDCClientSecret", cfg.OperatorOIDCClientSecret, "shh"},
        {"AnthropicSecretName", cfg.AnthropicSecretName, "anthropic"},
        {"CLIOIDCSecretName", cfg.CLIOIDCSecretName, "cli-oidc"},
        {"LogLevel", cfg.LogLevel, "debug"},
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            if tt.got != tt.want {
                t.Fatalf("%s = %q, want %q", tt.name, tt.got, tt.want)
            }
        })
    }
}

func TestLoad_MissingRequired(t *testing.T) {
    t.Setenv("OIDC_ISSUER", "")
    t.Setenv("OIDC_AUDIENCE", "")
    if _, err := config.Load(); err == nil {
        t.Fatal("expected error for missing required OIDC_ISSUER/OIDC_AUDIENCE")
    }
}

package config

import (
    "fmt"
    "os"
)

// Config holds the env-scalar configuration for the operator. Each field is
// populated from an env var injected via the chart ConfigMap/Secret (rule 6).
type Config struct {
    HTTPAddr                 string
    MetricsAddr              string
    InternalAddr             string
    OIDCIssuer               string
    OIDCAudience             string
    MemoryBaseURL            string
    IngesterImage            string
    ExternalWebhookBase      string
    OperatorOIDCClientID     string
    OperatorOIDCClientSecret string
    AnthropicSecretName      string
    CLIOIDCSecretName        string
    LogLevel                 string
}

func getDefault(key, def string) string {
    if v := os.Getenv(key); v != "" {
        return v
    }
    return def
}

// Load reads the operator configuration from the environment, applying
// defaults for the listener addresses and log level. OIDC issuer and
// audience are required.
func Load() (Config, error) {
    cfg := Config{
        HTTPAddr:                 getDefault("HTTP_ADDR", ":8080"),
        MetricsAddr:              getDefault("METRICS_ADDR", ":9090"),
        InternalAddr:             getDefault("INTERNAL_ADDR", ":8081"),
        OIDCIssuer:               os.Getenv("OIDC_ISSUER"),
        OIDCAudience:             os.Getenv("OIDC_AUDIENCE"),
        MemoryBaseURL:            os.Getenv("MEMORY_BASE_URL"),
        IngesterImage:            os.Getenv("INGESTER_IMAGE"),
        ExternalWebhookBase:      os.Getenv("EXTERNAL_WEBHOOK_BASE"),
        OperatorOIDCClientID:     os.Getenv("OPERATOR_OIDC_CLIENT_ID"),
        OperatorOIDCClientSecret: os.Getenv("OPERATOR_OIDC_CLIENT_SECRET"),
        AnthropicSecretName:      os.Getenv("ANTHROPIC_SECRET_NAME"),
        CLIOIDCSecretName:        os.Getenv("CLI_OIDC_SECRET_NAME"),
        LogLevel:                 getDefault("LOG_LEVEL", "info"),
    }
    if cfg.OIDCIssuer == "" {
        return Config{}, fmt.Errorf("config: OIDC_ISSUER is required")
    }
    if cfg.OIDCAudience == "" {
        return Config{}, fmt.Errorf("config: OIDC_AUDIENCE is required")
    }
    return cfg, nil
}

package obs_test

import (
    "bytes"
    "encoding/json"
    "log/slog"
    "testing"

    "github.com/szymonrychu/tatara-operator/internal/obs"
)

func TestNewLogger_EmitsJSON(t *testing.T) {
    var buf bytes.Buffer
    logger := obs.NewLogger(&buf, slog.LevelInfo)
    logger.Info("hello", slog.String("action", "test"))

    var entry map[string]any
    if err := json.Unmarshal(buf.Bytes(), &entry); err != nil {
        t.Fatalf("log line is not valid JSON: %v (%q)", err, buf.String())
    }
    if entry["msg"] != "hello" {
        t.Fatalf("msg = %v, want hello", entry["msg"])
    }
    if entry["action"] != "test" {
        t.Fatalf("action = %v, want test", entry["action"])
    }
}

func TestParseLevel(t *testing.T) {
    tests := []struct {
        in   string
        want slog.Level
    }{
        {"debug", slog.LevelDebug},
        {"info", slog.LevelInfo},
        {"warn", slog.LevelWarn},
        {"error", slog.LevelError},
        {"", slog.LevelInfo},
        {"bogus", slog.LevelInfo},
    }
    for _, tt := range tests {
        t.Run(tt.in, func(t *testing.T) {
            if got := obs.ParseLevel(tt.in); got != tt.want {
                t.Fatalf("ParseLevel(%q) = %v, want %v", tt.in, got, tt.want)
            }
        })
    }
}

package obs

import (
    "io"
    "log/slog"
    "strings"
)

// NewLogger returns a JSON-format slog.Logger writing to w at the given level.
func NewLogger(w io.Writer, level slog.Level) *slog.Logger {
    h := slog.NewJSONHandler(w, &slog.HandlerOptions{Level: level})
    return slog.New(h)
}

// ParseLevel maps a config string to an slog.Level, defaulting to Info.
func ParseLevel(s string) slog.Level {
    switch strings.ToLower(strings.TrimSpace(s)) {
    case "debug":
        return slog.LevelDebug
    case "warn", "warning":
        return slog.LevelWarn
    case "error":
        return slog.LevelError
    default:
        return slog.LevelInfo
    }
}

package obs_test

import (
    "testing"

    "github.com/prometheus/client_golang/prometheus/testutil"

    "github.com/szymonrychu/tatara-operator/internal/obs"
)

func TestNewMetrics_RegistersAll(t *testing.T) {
    m := obs.NewMetrics()
    if m.Registry == nil {
        t.Fatal("Registry is nil")
    }

    m.ReconcileTotal.WithLabelValues("Project", "success").Inc()
    m.IngestJobDuration.Observe(1.5)
    m.TurnDuration.Observe(2.5)
    m.WebhookEvents.WithLabelValues("github", "push", "accepted").Inc()
    m.TasksInflight.Set(3)

    want := []string{
        "operator_reconcile_total",
        "operator_ingest_job_duration_seconds",
        "operator_turn_duration_seconds",
        "operator_webhook_events_total",
        "operator_tasks_inflight",
    }
    for _, name := range want {
        t.Run(name, func(t *testing.T) {
            if testutil.CollectAndCount(m.Registry, name) == 0 {
                t.Fatalf("metric %q not registered/collected", name)
            }
        })
    }
}

package obs

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/collectors"
)

// Metrics holds the operator's Prometheus collectors and the registry they
// are registered against. One Metrics per process.
type Metrics struct {
    Registry          *prometheus.Registry
    ReconcileTotal    *prometheus.CounterVec
    IngestJobDuration prometheus.Histogram
    TurnDuration      prometheus.Histogram
    WebhookEvents     *prometheus.CounterVec
    TasksInflight     prometheus.Gauge
}

// NewMetrics constructs and registers all operator metrics on a fresh registry
// pre-populated with the Go and process collectors.
func NewMetrics() *Metrics {
    reg := prometheus.NewRegistry()
    reg.MustRegister(collectors.NewGoCollector())
    reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))

    m := &Metrics{
        Registry: reg,
        ReconcileTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
            Name: "operator_reconcile_total",
            Help: "Total reconcile invocations by kind and result.",
        }, []string{"kind", "result"}),
        IngestJobDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
            Name:    "operator_ingest_job_duration_seconds",
            Help:    "Duration of repository ingest Jobs in seconds.",
            Buckets: prometheus.DefBuckets,
        }),
        TurnDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
            Name:    "operator_turn_duration_seconds",
            Help:    "Duration of agent turns in seconds.",
            Buckets: prometheus.DefBuckets,
        }),
        WebhookEvents: prometheus.NewCounterVec(prometheus.CounterOpts{
            Name: "operator_webhook_events_total",
            Help: "Total webhook events by provider, kind and result.",
        }, []string{"provider", "kind", "result"}),
        TasksInflight: prometheus.NewGauge(prometheus.GaugeOpts{
            Name: "operator_tasks_inflight",
            Help: "Number of Tasks currently running.",
        }),
    }
    reg.MustRegister(m.ReconcileTotal, m.IngestJobDuration, m.TurnDuration, m.WebhookEvents, m.TasksInflight)
    return m
}

package auth_test

import (
    "context"
    "crypto/rand"
    "crypto/rsa"
    "testing"
    "time"

    "github.com/stretchr/testify/require"

    "github.com/szymonrychu/tatara-operator/internal/auth"
    "github.com/szymonrychu/tatara-operator/internal/auth/testjwks"
)

func TestVerifier_ValidToken(t *testing.T) {
    srv := testjwks.NewServer(t)
    ctx := context.Background()

    v, err := auth.NewVerifier(ctx, auth.Config{Issuer: srv.Issuer(), Audience: "tatara-operator"})
    require.NoError(t, err)

    tok := srv.SignTypedToken(t, testjwks.Claims{
        Issuer:   srv.Issuer(),
        Audience: []string{"tatara-operator"},
        Subject:  "agent-1",
        Extra:    map[string]any{"preferred_username": "agent"},
    })

    claims, err := v.Verify(ctx, tok)
    require.NoError(t, err)
    require.Equal(t, "agent-1", claims.Subject)
    require.Equal(t, "agent", claims.PreferredUsername)
}

func TestVerifier_Rejections(t *testing.T) {
    srv := testjwks.NewServer(t)
    ctx := context.Background()
    v, err := auth.NewVerifier(ctx, auth.Config{Issuer: srv.Issuer(), Audience: "tatara-operator"})
    require.NoError(t, err)

    foreign, err := rsa.GenerateKey(rand.Reader, 2048)
    require.NoError(t, err)

    tests := []struct {
        name string
        sign func() string
    }{
        {
            name: "expired",
            sign: func() string {
                return srv.SignTypedToken(t, testjwks.Claims{
                    Issuer:    srv.Issuer(),
                    Audience:  []string{"tatara-operator"},
                    Subject:   "agent-1",
                    IssuedAt:  time.Now().Add(-2 * time.Hour),
                    NotBefore: time.Now().Add(-2 * time.Hour),
                    ExpiresAt: time.Now().Add(-time.Hour),
                })
            },
        },
        {
            name: "wrong-issuer",
            sign: func() string {
                return srv.SignTypedToken(t, testjwks.Claims{
                    Issuer:   "https://evil.example/realms/master",
                    Audience: []string{"tatara-operator"},
                    Subject:  "agent-1",
                })
            },
        },
        {
            name: "wrong-audience",
            sign: func() string {
                return srv.SignTypedToken(t, testjwks.Claims{
                    Issuer:   srv.Issuer(),
                    Audience: []string{"some-other-app"},
                    Subject:  "agent-1",
                })
            },
        },
        {
            name: "bad-signature",
            sign: func() string {
                return srv.SignTokenWithKey(t, foreign, testjwks.Claims{
                    Issuer:   srv.Issuer(),
                    Audience: []string{"tatara-operator"},
                    Subject:  "agent-1",
                })
            },
        },
        {
            name: "missing-sub",
            sign: func() string {
                return srv.SignTypedToken(t, testjwks.Claims{
                    Issuer:   srv.Issuer(),
                    Audience: []string{"tatara-operator"},
                })
            },
        },
    }
    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            _, err := v.Verify(ctx, tt.sign())
            require.Error(t, err)
        })
    }
}

func TestConfig_Validate(t *testing.T) {
    require.Error(t, auth.Config{Audience: "x"}.Validate())
    require.Error(t, auth.Config{Issuer: "x"}.Validate())
    require.NoError(t, auth.Config{Issuer: "x", Audience: "y"}.Validate())
}