laya: incident decision engine over alert + pod state

Adds internal/laya, a client and safety-gated decision engine for the
Laya decision API (https://laya.k6n.net). The API's OpenAPI document
leaves every response schema empty, so the wire format was pinned by
probing the live service: questions are typed (choice/score/noul) and
carry "criteria" as an object for a choice and an array for a score.

An alert is rendered into a compact factual state and asked four typed
questions: which action (restart/ignore/notify/escalate), severity and
how urgent. The state always includes the pods' last events - restart
count, readiness, last termination reason and exit code, log tail, and
any overlapping rollout - because a 503 burst from a pod being replaced
is otherwise indistinguishable from an outage.

The model proposes, the policy disposes. Every gate records why it
changed an answer: an executable-action allowlist, a probability and
lead-over-runner-up threshold, no restart while pods are crash-looping
(the kubelet is already doing it) or while a rollout is replacing every
pod (the 503s are the rollout's own), never ignoring a new error or a
high-priority spike without a human's mute, an hourly restart budget
per deployment, and a plain notification whenever the model is
unreachable. The margin, not the raw probability, is the useful signal:
label probabilities are diffuse and a correct answer rarely exceeds 0.5.

topology.PodLog now carries the pod's lifecycle facts alongside the log
tail. PodsNotReady/CrashLooping helpers build on them.

Verified end to end against the live API: a crash loop is withheld as
"already crash-looping", a rollout's 503 burst is held back at a 2%
margin, and crawler 404s are ignored at 13%. LAYA_URL=... go test
./internal/laya -run TestLiveModel replays it.

internal/laya/client.go

@@ -0,0 +1,289 @@
+// Package laya talks to the Laya decision API (https://laya.k6n.net) - a fast
+// non-autoregressive "System 1" model that answers typed questions about a
+// state in one forward pass. It is used here as an incident triage brain: give
+// it a compact textual state of a firing spike and it returns a calibrated
+// choice (restart / ignore / notify / escalate) with probabilities.
+//
+// The API is described by https://laya.k6n.net/openapi.json. Only the parts
+// this repo uses are modelled; the wire format is deliberately loose on the
+// response side (answers are a map) because the model decides which questions
+// get answers.
+package laya
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// Question kinds understood by the API.
+const (
+	KindChoice = "choice" // pick one label from Criteria
+	KindScore  = "score"  // pick a level index from Levels
+	KindNoul   = "noul"   // a single 0..1 non-choice answer
+)
+
+// Question is one typed question. Use the ChoiceQuestion, ScoreQuestion or
+// NoulQuestion constructors rather than building it by hand, because
+// "criteria" is a JSON object for choice questions and a JSON array for score
+// ones - both are rendered by MarshalJSON.
+type Question struct {
+	kind         string
+	instructions string
+	criteria     map[string]string // choice: label -> what that label means
+	levels       []string          // score: level descriptions, index 0 first
+}
+
+func (q Question) MarshalJSON() ([]byte, error) {
+	out := map[string]any{"type": q.kind, "instructions": q.instructions}
+	switch q.kind {
+	case KindChoice:
+		if len(q.criteria) == 0 {
+			return nil, errors.New("laya: choice question needs criteria")
+		}
+		out["criteria"] = q.criteria
+	case KindScore:
+		if len(q.levels) == 0 {
+			return nil, errors.New("laya: score question needs criteria levels")
+		}
+		out["criteria"] = q.levels
+	}
+	return json.Marshal(out)
+}
+
+func (q Question) Kind() string { return q.kind }
+
+// ChoiceQuestion asks the model to pick one of criteria. The label -> text map
+// matters: the model reads the descriptions, not just the labels, so "restart"
+// should say what restarting means here.
+func ChoiceQuestion(instructions string, criteria map[string]string) Question {
+	return Question{kind: KindChoice, instructions: instructions, criteria: criteria}
+}
+
+// ScoreQuestion asks for a level between 0 and len(levels)-1.
+func ScoreQuestion(instructions string, levels []string) Question {
+	return Question{kind: KindScore, instructions: instructions, levels: levels}
+}
+
+// NoulQuestion asks a single open question answered with one 0..1 value.
+func NoulQuestion(instructions string) Question {
+	return Question{kind: KindNoul, instructions: instructions}
+}
+
+// PredictRequest is the body of POST /v1/predict. State is a string, a map or
+// a list; this repo always sends a rendered text state.
+type PredictRequest struct {
+	State     any                 `json:"state"`
+	Questions map[string]Question `json:"questions"`
+	Lang      string              `json:"lang,omitempty"`
+	MaxLen    int                 `json:"max_len,omitempty"`
+	BatchSize int                 `json:"batch_size,omitempty"` // batch endpoint only
+}
+
+// BatchPredictRequest is the body of POST /v1/predict/batch: several states
+// evaluated against the same questions, packed into shared forward passes.
+type BatchPredictRequest struct {
+	States    []any               `json:"states"`
+	Questions map[string]Question `json:"questions"`
+	Lang      string              `json:"lang,omitempty"`
+	BatchSize int                 `json:"batch_size,omitempty"`
+	MaxLen    int                 `json:"max_len,omitempty"`
+}
+
+// Answer is one answered question. Only the fields matching Type are set.
+type Answer struct {
+	Type          string             `json:"type"`
+	Choice        string             `json:"choice,omitempty"`
+	Probabilities map[string]float64 `json:"probabilities,omitempty"`
+	Score         float64            `json:"score,omitempty"`
+	Legend        map[string]string  `json:"legend,omitempty"`
+	Noul          float64            `json:"noul,omitempty"`
+	Confidence    float64            `json:"confidence,omitempty"`
+	// AnswerConfidence is the model's confidence in the answer it picked,
+	// as opposed to Confidence, which measures how decisive the forward pass
+	// was overall.
+	AnswerConfidence float64 `json:"answer_confidence,omitempty"`
+}
+
+// Probability returns the model's probability for one choice label, 0 if the
+// label was not offered or not answered.
+func (a Answer) Probability(label string) float64 { return a.Probabilities[label] }
+
+// Margin is how far the chosen answer leads the next best one. The label
+// probabilities are diffuse - even a correct answer rarely exceeds 0.5 - so
+// the lead over the runner-up is a better confidence signal than the raw
+// probability.
+func (a Answer) Margin() float64 {
+	best, second := 0.0, 0.0
+	for _, p := range a.Probabilities {
+		switch {
+		case p > best:
+			second, best = best, p
+		case p > second:
+			second = p
+		}
+	}
+	return best - second
+}
+
+type Usage struct {
+	InputTokens  int `json:"input_tokens"`
+	OutputTokens int `json:"output_tokens"`
+}
+
+type Response struct {
+	Model   string            `json:"model"`
+	Answers map[string]Answer `json:"answers"`
+	Usage   Usage             `json:"usage"`
+}
+
+// Health is the body of GET /healthz.
+type Health struct {
+	Status  string `json:"status"`
+	Model   string `json:"model"`
+	Device  string `json:"device"`
+	Version string `json:"version"`
+}
+
+// Client is a Laya API client. BaseURL defaults to https://laya.k6n.net.
+type Client struct {
+	BaseURL   string
+	HTTP      *http.Client
+	Lang      string
+	MaxLen    int
+	BatchSize int // forward-pass chunking for the batch endpoint; 0 = server default
+}
+
+// NewClient returns a client for base ("https://laya.k6n.net" when empty).
+func NewClient(base string) *Client {
+	return &Client{BaseURL: strings.TrimRight(base, "/"), HTTP: &http.Client{Timeout: 30 * time.Second}, Lang: "en"}
+}
+
+func (c *Client) url(path string) string {
+	base := c.BaseURL
+	if base == "" {
+		base = "https://laya.k6n.net"
+	}
+	return base + path
+}
+
+func (c *Client) client() *http.Client {
+	if c.HTTP != nil {
+		return c.HTTP
+	}
+	return &http.Client{Timeout: 30 * time.Second}
+}
+
+// Health reports the model behind the API, for a setup probe.
+func (c *Client) Health(ctx context.Context) (Health, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url("/healthz"), nil)
+	if err != nil {
+		return Health{}, err
+	}
+	resp, err := c.client().Do(req)
+	if err != nil {
+		return Health{}, unwrap(err)
+	}
+	defer resp.Body.Close()
+	if err := checkStatus(resp); err != nil {
+		return Health{}, err
+	}
+	var h Health
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<16)).Decode(&h); err != nil {
+		return Health{}, fmt.Errorf("laya: decode health: %w", err)
+	}
+	return h, nil
+}
+
+// Predict evaluates one state against the questions in a single pass.
+func (c *Client) Predict(ctx context.Context, state any, questions map[string]Question) (Response, error) {
+	body := PredictRequest{State: state, Questions: questions, Lang: c.Lang, MaxLen: c.MaxLen}
+	return c.post(ctx, "/v1/predict", body)
+}
+
+// PredictBatch evaluates several states against the same questions. Results
+// are returned in request order, one entry per state.
+func (c *Client) PredictBatch(ctx context.Context, states []any, questions map[string]Question) ([]Response, error) {
+	body := BatchPredictRequest{States: states, Questions: questions, Lang: c.Lang, BatchSize: c.BatchSize, MaxLen: c.MaxLen}
+	var out struct {
+		Results []Response `json:"results"`
+	}
+	if err := c.postInto(ctx, "/v1/predict/batch", body, &out); err != nil {
+		return nil, err
+	}
+	if len(out.Results) != len(states) {
+		return nil, fmt.Errorf("laya: batch returned %d results for %d states", len(out.Results), len(states))
+	}
+	return out.Results, nil
+}
+
+func (c *Client) post(ctx context.Context, path string, body any) (Response, error) {
+	var out Response
+	if err := c.postInto(ctx, path, body, &out); err != nil {
+		return Response{}, err
+	}
+	return out, nil
+}
+
+func (c *Client) postInto(ctx context.Context, path string, body, out any) error {
+	raw, err := json.Marshal(body)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url(path), bytes.NewReader(raw))
+	if err != nil {
+		return errors.New("laya: invalid request")
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := c.client().Do(req)
+	if err != nil {
+		return unwrap(err)
+	}
+	defer resp.Body.Close()
+	if err := checkStatus(resp); err != nil {
+		return err
+	}
+	return json.NewDecoder(io.LimitReader(resp.Body, 1<<22)).Decode(out)
+}
+
+// unwrap drops the URL the transport puts in its error, which may carry a
+// token; the returned error must be safe to log.
+func unwrap(err error) error {
+	var ue interface{ Unwrap() error }
+	if errors.As(err, &ue) {
+		return errors.New(scrub(ue.Unwrap().Error()))
+	}
+	return errors.New(scrub(err.Error()))
+}
+
+// scrub removes anything that looks like a credential from a URL in an error.
+func scrub(msg string) string {
+	for _, part := range strings.Fields(msg) {
+		if i := strings.Index(part, "://"); i >= 0 {
+			rest := part[i+3:]
+			if at := strings.IndexByte(rest, '@'); at >= 0 {
+				host := rest[at+1:]
+				msg = strings.ReplaceAll(msg, part, "laya://"+host)
+			}
+		}
+	}
+	return msg
+}
+
+func checkStatus(resp *http.Response) error {
+	if resp.StatusCode/100 == 2 {
+		return nil
+	}
+	detail, _ := io.ReadAll(io.LimitReader(resp.Body, 500))
+	msg := strings.TrimSpace(string(detail))
+	if msg == "" {
+		msg = http.StatusText(resp.StatusCode)
+	}
+	return fmt.Errorf("laya: HTTP %d: %s", resp.StatusCode, scrub(msg))
+}

internal/laya/decision.go

@@ -0,0 +1,579 @@
+package laya
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+)
+
+// Action is what the decision engine wants done with a firing alert. These are
+// the labels offered to the model, so they must stay stable: a rename changes
+// what the model was trained to answer.
+type Action string
+
+const (
+	ActionRestart  Action = "restart"  // roll the Deployment's pods
+	ActionIgnore   Action = "ignore"   // known noise, do not page
+	ActionNotify   Action = "notify"   // tell the on-call
+	ActionEscalate Action = "escalate" // tell a human who owns the service
+)
+
+// Actions is the full ladder, least to most forceful.
+var Actions = []Action{ActionIgnore, ActionNotify, ActionEscalate, ActionRestart}
+
+// Pattern is one reduced error pattern of an alert, as the triage package
+// classifies it. It is deliberately plain data: the engine can be fed from
+// anywhere without importing the alerting pipeline.
+type Pattern struct {
+	What     string // "HTTP 500 | /api/cart | panic: ..."
+	Category string // triage category, e.g. service-defect
+	Status   int    // HTTP status, 0 when the error was not an HTTP one
+	Count    int
+	Baseline float64
+	New      bool     // never seen in the baseline windows
+	Clients  int      // distinct clients affected
+	Services []string // every service failing in these requests
+	Host     string   // ingress host, for edge errors
+	Path     string   // request path
+	Routes   []string // "host/path -> service ns/name -> deployment ns/name"
+}
+
+// Pod is a pod behind one of the alert's deployments, with the facts that
+// explain errors the log lines alone cannot: a pod being replaced right now
+// (so a 503 burst is expected) or a crash loop (so a restart changes nothing).
+type Pod struct {
+	Namespace, Deployment, Name string
+	Found                       bool
+	Ready                       bool
+	Phase                       string
+	Restarts                    int32  // restarts of this pod
+	PodRestarts                 int32  // restarts across the Deployment
+	Reason                      string // CrashLoopBackOff, OOMKilled, ...
+	ExitCode                    *int32
+	Message                     string
+	LastTerminatedAt            time.Time
+	LastEvents                  []string // last log lines, oldest first
+	Err                         string
+}
+
+// Deploy is a rollout that overlaps the alert window.
+type Deploy struct {
+	Namespace, Workload string
+	Release             string
+	Age                 time.Duration // how long ago the rollout started
+	Completed           bool
+	Ready, Desired      int32
+}
+
+// Input is everything known about a firing alert, rendered into the state text
+// the model reasons over.
+type Input struct {
+	AlertID      string
+	Service      string
+	Priority     string // high | medium | low
+	ErrorCount   int
+	Window       time.Duration
+	BaselineMean float64
+	NewErrors    bool
+	Clients      int
+	Patterns     []Pattern
+	Pods         []Pod
+	Deploys      []Deploy
+	// KnownNoise marks a pattern a human has already marked as ignorable, e.g.
+	// a scanner or a muted issue. The model is told so it need not guess.
+	KnownNoise bool
+}
+
+// maxStateChars keeps the rendered state inside the model's context. The tail
+// (pods, deploys) is what the model most often misses, so patterns are cut
+// first, never the pod events.
+const maxStateChars = 4000
+
+// Render turns an alert into a compact factual state. It states only what was
+// measured - the model supplies judgement, this supplies facts - and always
+// includes the pods' last events, because a 503 burst while a pod is being
+// replaced is expected and indistinguishable from an outage without them.
+func (in Input) Render(now time.Time) string {
+	var b strings.Builder
+	multiple := 0.0
+	if in.BaselineMean > 0 {
+		multiple = float64(in.ErrorCount) / in.BaselineMean
+	}
+	fmt.Fprintf(&b, "Firing error spike in service %s (priority %s).\n", or(in.Service, "unknown"), or(in.Priority, "unknown"))
+	fmt.Fprintf(&b, "%d errors in the last %s, normally about %.0f per window (%.0fx normal).\n",
+		in.ErrorCount, in.Window.Round(time.Second), in.BaselineMean, multiple)
+	if in.NewErrors {
+		b.WriteString("At least one error is new: it was never seen in the baseline windows.\n")
+	}
+	if in.Clients > 0 {
+		fmt.Fprintf(&b, "%d distinct clients affected.\n", in.Clients)
+	}
+	if in.KnownNoise {
+		b.WriteString("A human has already marked this pattern as ignorable noise.\n")
+	}
+	b.WriteString("\nError patterns, most urgent first:\n")
+	for i, p := range in.Patterns {
+		fmt.Fprintf(&b, "%d. [%s] %s\n", i+1, or(p.Category, "unclassified"), or(p.What, "no message"))
+		fmt.Fprintf(&b, "   %d in the window, normally %.0f. %s. clients affected: %d.",
+			p.Count, p.Baseline, newness(p.New), p.Clients)
+		if p.Status > 0 {
+			fmt.Fprintf(&b, " HTTP %d.", p.Status)
+		}
+		b.WriteString("\n")
+		if len(p.Services) > 1 {
+			fmt.Fprintf(&b, "   Failing together: %s.\n", strings.Join(p.Services, ", "))
+		}
+		for _, r := range p.Routes {
+			fmt.Fprintf(&b, "   Route: %s.\n", r)
+		}
+	}
+	if len(in.Pods) == 0 {
+		b.WriteString("\nPod state: unknown, the deployment behind these errors was not resolved.\n")
+	}
+	for _, p := range in.Pods {
+		b.WriteString("\nPod events:\n")
+		fmt.Fprintf(&b, "- %s/%s pod %s: %s\n", p.Namespace, p.Deployment, or(p.Name, "unknown"), podStateLine(p, now))
+		for _, line := range p.LastEvents {
+			fmt.Fprintf(&b, "  log: %s\n", clip(line, 160))
+		}
+	}
+	for _, d := range in.Deploys {
+		state := "rollout in progress"
+		if d.Completed {
+			state = "rollout completed"
+		}
+		fmt.Fprintf(&b, "\nDeploy: %s/%s %s started %s ago, %d/%d replicas ready",
+			d.Namespace, d.Workload, state, d.Age.Round(time.Second), d.Ready, d.Desired)
+		if d.Release != "" {
+			fmt.Fprintf(&b, " (release %s)", d.Release)
+		}
+		b.WriteString("\n")
+	}
+	return clip(b.String(), maxStateChars)
+}
+
+// podStateLine says what the pod is doing now, in the words an operator would
+// use: crash-looping, being replaced, up but not ready, or healthy.
+func podStateLine(p Pod, now time.Time) string {
+	switch {
+	case p.Err != "":
+		return "could not be inspected (" + clip(p.Err, 80) + ")"
+	case !p.Found:
+		return "no pod running, the Deployment is scaled to 0 or has nothing ready"
+	}
+	var parts []string
+	phase := or(p.Phase, "unknown")
+	if p.Ready {
+		parts = append(parts, "phase "+phase+", passing its readiness probe")
+	} else {
+		parts = append(parts, "phase "+phase+", NOT passing its readiness probe")
+	}
+	switch {
+	case p.Restarting(now):
+		parts = append(parts, fmt.Sprintf("its container restarted %s ago and is still coming up", since(p.LastTerminatedAt, now)))
+	case p.CrashLooping(now):
+		parts = append(parts, fmt.Sprintf("crash-looping: %d restarts", p.Restarts))
+	case p.Restarts > 0:
+		parts = append(parts, fmt.Sprintf("%d restarts, none recent", p.Restarts))
+	}
+	if p.PodRestarts > p.Restarts {
+		parts = append(parts, fmt.Sprintf("%d restarts across the Deployment", p.PodRestarts))
+	}
+	if p.Reason != "" {
+		parts = append(parts, "last container termination: "+p.Reason)
+	}
+	if p.ExitCode != nil {
+		parts = append(parts, fmt.Sprintf("exit code %d", *p.ExitCode))
+	}
+	if p.Message != "" {
+		parts = append(parts, clip(p.Message, 120))
+	}
+	return strings.Join(parts, "; ")
+}
+
+// Restarting reports a pod that is right now being replaced, so the errors
+// around it are the rollout's, not a new defect.
+func (p Pod) Restarting(now time.Time) bool {
+	return !p.LastTerminatedAt.IsZero() && !now.Before(p.LastTerminatedAt) && now.Sub(p.LastTerminatedAt) < rollingWindow
+}
+
+// CrashLooping reports a pod stuck restarting: a restart cannot fix it.
+func (p Pod) CrashLooping(now time.Time) bool {
+	if p.Reason == "CrashLoopBackOff" {
+		return true
+	}
+	return p.Restarts >= 2 && !p.LastTerminatedAt.IsZero() && now.Sub(p.LastTerminatedAt) < rollingWindow
+}
+
+func newness(isNew bool) string {
+	if isNew {
+		return "never seen before"
+	}
+	return "seen before"
+}
+
+func since(t, now time.Time) string {
+	d := now.Sub(t)
+	switch {
+	case d < time.Minute:
+		return fmt.Sprintf("%ds", int(d.Seconds()))
+	case d < time.Hour:
+		return fmt.Sprintf("%dm", int(d.Minutes()))
+	default:
+		return fmt.Sprintf("%.1fh", d.Hours())
+	}
+}
+
+func or(v, fallback string) string {
+	if strings.TrimSpace(v) == "" {
+		return fallback
+	}
+	return v
+}
+
+func clip(s string, n int) string {
+	s = strings.TrimSpace(s)
+	if len(s) <= n {
+		return s
+	}
+	return strings.TrimSpace(s[:n-1]) + "…"
+}
+
+// Policy decides which of the model's answers may be executed. The model
+// proposes; the policy disposes. Nothing is acted on unless it is in Actions,
+// clears MinProbability, and survives the gates below.
+type Policy struct {
+	// Actions allowed to be executed automatically. An empty list means
+	// "decide and report, never act", which is the safe default.
+	Actions []Action
+	// MinProbability is the model's probability for its chosen action below
+	// which the decision is downgraded to a notification.
+	MinProbability float64
+	// MinMargin is the lead over the runner-up required to act. The label
+	// probabilities are diffuse, so this is the more meaningful of the two.
+	MinMargin float64
+	// MinClients: a pattern affecting fewer clients than this is never
+	// auto-acted on, however confident the model is.
+	MinClients int
+	// MaxRestartsPerHour caps automatic restarts per deployment, so a flapping
+	// service cannot be restarted into a loop.
+	MaxRestartsPerHour int
+	// Now is the clock, overridable in tests.
+	Now func() time.Time
+}
+
+func (p Policy) now() time.Time {
+	if p.Now != nil {
+		return p.Now()
+	}
+	return time.Now()
+}
+
+func (p Policy) allows(a Action) bool {
+	for _, allowed := range p.Actions {
+		if allowed == a {
+			return true
+		}
+	}
+	return false
+}
+
+func (p Policy) minProbability() float64 {
+	if p.MinProbability <= 0 {
+		return 0.35 // below this the leading label barely beats the field
+	}
+	return p.MinProbability
+}
+
+func (p Policy) minMargin() float64 {
+	if p.MinMargin <= 0 {
+		return 0.05
+	}
+	return p.MinMargin
+}
+
+// Verdict is the outcome for one alert: what the model asked for, what was
+// allowed, and why it differs.
+type Verdict struct {
+	AlertID       string
+	Action        Action  // what is actually done
+	Requested     Action  // what the model chose
+	Probability   float64 // model's probability for Requested
+	Margin        float64 // how far Requested leads the runner-up
+	Confidence    float64 // how decisive the forward pass was
+	Probabilities map[string]float64
+	Severity      string
+	Urgency       int
+	Model         string
+	State         string   // the text the model was given, kept for the UI
+	Gates         []string // why the action was changed or withheld, in order
+	// Targets are the deployments a restart was cleared for, after the
+	// hourly budget. Executable means exactly this list may be acted on.
+	Targets []string
+	Tokens  int
+	At      time.Time
+	Error   string // set when the model could not be reached
+}
+
+// Executable reports whether the action may be carried out automatically.
+func (v Verdict) Executable() bool {
+	return v.Error == "" && v.Action != ActionNotify && v.Action != ActionEscalate
+}
+
+// Changed reports whether a gate altered the model's answer.
+func (v Verdict) Changed() bool { return v.Error == "" && v.Requested != "" && v.Action != v.Requested }
+
+// Explain renders the verdict as one line for a notification or a log.
+func (v Verdict) Explain() string {
+	if v.Error != "" {
+		return "decision unavailable: " + v.Error + " (notifying as usual)"
+	}
+	msg := fmt.Sprintf("laya %s: %s (%.0f%%, margin %.0f%%, decisiveness %.2f)", v.Action, v.Requested, v.Probability*100, v.Margin*100, v.Confidence)
+	if v.Severity != "" {
+		msg += ", severity " + v.Severity
+	}
+	for _, g := range v.Gates {
+		msg += "; " + g
+	}
+	return msg
+}
+
+// Decider is the part of an Engine the alerting pipeline talks to, so the
+// pipeline can be tested without a model.
+type Decider interface {
+	Decide(ctx context.Context, in Input) Verdict
+}
+
+// Engine renders an alert, asks Laya what to do, and applies the policy gates.
+type Engine struct {
+	Client *Client
+	Policy Policy
+	// restarts counts automatic restarts per deployment within the hour.
+	restarts map[string][]time.Time
+}
+
+func NewEngine(client *Client, policy Policy) *Engine {
+	return &Engine{Client: client, Policy: policy, restarts: map[string][]time.Time{}}
+}
+
+// Questions are the typed questions asked of the model. The criteria spell
+// out what each action means here, because the model reads the descriptions
+// rather than the labels.
+func Questions() map[string]Question {
+	return map[string]Question{
+		"action": ChoiceQuestion("What is the right operational response to this error spike right now?", map[string]string{
+			"restart":  "Restart the pods of the deployment: only right when the pod itself is wedged or crash-looping and a fresh process would fix it",
+			"ignore":   "Ignore: known noise, an expected consequence of a deploy or a scanner, nothing for a human to do",
+			"notify":   "Notify the on-call: a real problem that needs a human to look, but not self-fixable",
+			"escalate": "Escalate to the service owner: serious, ongoing, and likely to need a code change or a rollback",
+		}),
+		"severity": ChoiceQuestion("How severe is this for users of the service?", map[string]string{
+			"low":      "cosmetic, a tiny fraction of requests, no user-visible effect",
+			"medium":   "degraded, some users are affected but the service still works",
+			"high":     "users are visibly broken, a large share of requests fail",
+			"critical": "the service is down or unusable for most of its users",
+		}),
+		"urgency": ScoreQuestion("How soon does a human need to act? 0 = nobody needs to act at all, 4 = a human is needed right now.",
+			[]string{"0 nobody needs to act", "1 low priority", "2 today", "3 within the hour", "4 needed right now"}),
+	}
+}
+
+// Decide renders the state, asks the model, and gates the answer. It never
+// returns an error: a decision that could not be made degrades to a plain
+// notification, which is what the pipeline would have done anyway.
+func (e *Engine) Decide(ctx context.Context, in Input) Verdict {
+	now := e.Policy.now()
+	v := Verdict{AlertID: in.AlertID, Requested: ActionNotify, Action: ActionNotify, At: now, State: in.Render(now)}
+	if e == nil || e.Client == nil {
+		return v
+	}
+	resp, err := e.Client.Predict(ctx, v.State, Questions())
+	if err != nil {
+		v.Error = err.Error()
+		return v
+	}
+	v.Model, v.Tokens = resp.Model, resp.Usage.InputTokens
+	answer := resp.Answers["action"]
+	v.Probabilities = answer.Probabilities
+	v.Probability = answer.Probability(answer.Choice)
+	v.Margin = answer.Margin()
+	v.Confidence = answer.Confidence
+	v.Severity = resp.Answers["severity"].Choice
+	v.Urgency = int(resp.Answers["urgency"].Score + 0.5)
+
+	requested := Action(answer.Choice)
+	if !knownAction(requested) {
+		v.Error = "model answered with an unknown action " + answer.Choice
+		return v
+	}
+	v.Requested = requested
+	v.Action = requested
+	e.gate(ctx, &v, in, now)
+	return v
+}
+
+func knownAction(a Action) bool {
+	for _, k := range Actions {
+		if k == a {
+			return true
+		}
+	}
+	return false
+}
+
+// gate is the safety envelope. Each check either leaves the action alone or
+// downgrades it to a notification, always recording why.
+func (e *Engine) gate(ctx context.Context, v *Verdict, in Input, now time.Time) {
+	downgrade := func(to Action, why string) {
+		if v.Action != to {
+			v.Gates = append(v.Gates, why)
+			v.Action = to
+		}
+	}
+	// 1. The model has to lead the field by a real margin. Both the absolute
+	//    probability and the lead are checked: a diffuse answer that only just
+	//    wins is a guess, and must not move a cluster on its own.
+	if v.Probability < e.Policy.minProbability() {
+		downgrade(ActionNotify, fmt.Sprintf("model was only %.0f%% sure of %s", v.Probability*100, v.Requested))
+	}
+	if v.Margin < e.Policy.minMargin() {
+		downgrade(ActionNotify, fmt.Sprintf("%s only just beat the alternatives (margin %.0f%%)", v.Requested, v.Margin*100))
+	}
+	// 2. Restarts are the one action that can make things worse, so the pod
+	//    state has to support it rather than merely allow it.
+	if v.Action == ActionRestart {
+		switch reason := restartRefused(in, now); {
+		case reason != "":
+			downgrade(ActionNotify, reason)
+		case !e.Policy.allows(ActionRestart):
+			downgrade(ActionNotify, "automatic restarts are not enabled")
+		case !e.allowRestart(v, in, now):
+			downgrade(ActionNotify, "restart budget exhausted for this deployment")
+		}
+	}
+	// 3. Ignoring is only safe when the errors are explained: a human's mute,
+	//    a crawler, or a rollout that is still replacing pods.
+	if v.Action == ActionIgnore {
+		if reason := ignoreRefused(in); reason != "" {
+			downgrade(ActionNotify, reason)
+		}
+	}
+	// 4. Anything not on the allowlist is reported, never executed.
+	if v.Action != ActionNotify && v.Action != ActionEscalate && !e.Policy.allows(v.Action) {
+		downgrade(ActionNotify, "action "+string(v.Action)+" is not in the allowed set")
+	}
+}
+
+// restartRefused explains why a restart would not help, using the pods' last
+// events. A pod the kubelet is already restarting will just be restarted again,
+// and an expected 503 burst during a rollout is not an incident at all.
+func restartRefused(in Input, now time.Time) string {
+	var live, crashLooping, restarting int
+	for _, p := range in.Pods {
+		if p.Err != "" || !p.Found {
+			continue
+		}
+		live++
+		switch {
+		case p.CrashLooping(now):
+			crashLooping++
+		case p.Restarting(now):
+			restarting++
+		}
+	}
+	if live == 0 {
+		return "no pod state to justify a restart"
+	}
+	if crashLooping == live {
+		return "pods are already crash-looping, a restart repeats what the kubelet already does"
+	}
+	// Every pod died moments ago, or a rollout is replacing them: the 503s
+	// around it are the rollout's own, and restarting would restart the
+	// rollout as well.
+	if restarting == live || (notReady(in) && rolloutOngoing(in, now)) {
+		return "every pod is being replaced by a rollout, the 503s are the rollout's own"
+	}
+	return ""
+}
+
+// notReady reports whether no pod behind the alert is serving traffic.
+func notReady(in Input) bool {
+	ready := false
+	for _, p := range in.Pods {
+		if p.Err == "" && p.Found && p.Ready {
+			ready = true
+		}
+	}
+	return !ready
+}
+
+// ignoreRefused explains why a spike may not be dropped silently. A brand new
+// error is never ignored: nobody has seen it before, so nobody can know it is
+// harmless.
+func ignoreRefused(in Input) string {
+	if in.KnownNoise {
+		return ""
+	}
+	if in.NewErrors {
+		return "the error is new, so nobody has confirmed it is harmless"
+	}
+	if in.Priority == "high" && in.ErrorCount > 0 {
+		return "a high-priority spike is never ignored without a human's confirmation"
+	}
+	for _, p := range in.Patterns {
+		if p.New {
+			return "pattern " + p.What + " is new, so it cannot be assumed harmless"
+		}
+	}
+	return ""
+}
+
+func rolloutOngoing(in Input, now time.Time) bool {
+	for _, d := range in.Deploys {
+		if !d.Completed && d.Age > 0 && d.Age < rollingWindow {
+			return true
+		}
+	}
+	return false
+}
+
+// allowRestart enforces the hourly budget per deployment, so a service that
+// flaps cannot be restarted over and over.
+func (e *Engine) allowRestart(v *Verdict, in Input, now time.Time) bool {
+	limit := e.Policy.MaxRestartsPerHour
+	if limit <= 0 {
+		limit = 3
+	}
+	cleared := 0
+	for _, p := range in.Pods {
+		key := p.Namespace + "/" + p.Deployment
+		cut := now.Add(-time.Hour)
+		kept := e.restarts[key][:0]
+		for _, at := range e.restarts[key] {
+			if at.After(cut) {
+				kept = append(kept, at)
+			}
+		}
+		e.restarts[key] = kept
+		if len(kept) >= limit {
+			continue
+		}
+		e.restarts[key] = append(kept, now)
+		v.Targets = append(v.Targets, key)
+		cleared++
+	}
+	return cleared > 0
+}
+
+// RollingWindow is how recently a container must have died for the errors
+// around it to be attributed to the pod still coming up.
+const rollingWindow = 5 * time.Minute
+
+// sortedDeploys is used by the pipeline to feed Deploys newest first.
+func sortedDeploys(in []Deploy) []Deploy {
+	out := append([]Deploy(nil), in...)
+	sort.SliceStable(out, func(i, j int) bool { return out[i].Age < out[j].Age })
+	return out
+}

internal/laya/laya_test.go

@@ -0,0 +1,344 @@
+package laya
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+// fake serves the Laya API with a fixed answer per action label, and records
+// the last state and questions it was asked.
+type fake struct {
+	choice     string
+	prob       map[string]float64
+	lastState  string
+	lastQs     map[string]json.RawMessage
+	status     int
+	batchSizes int
+}
+
+func (f *fake) server(t *testing.T) *httptest.Server {
+	t.Helper()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
+		writeJSON(w, Health{Status: "ok", Model: "test-model"})
+	})
+	mux.HandleFunc("/v1/predict", func(w http.ResponseWriter, r *http.Request) {
+		var req struct {
+			State     json.RawMessage            `json:"state"`
+			Questions map[string]json.RawMessage `json:"questions"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			t.Errorf("decode request: %v", err)
+		}
+		f.lastState = string(req.State)
+		f.lastQs = req.Questions
+		f.respond(w, map[string]Answer{
+			"action":   {Type: KindChoice, Choice: f.choice, Probabilities: f.prob},
+			"severity": {Type: KindChoice, Choice: "high", Probabilities: map[string]float64{"high": 0.8}},
+			"urgency":  {Type: KindScore, Score: 3},
+		})
+	})
+	mux.HandleFunc("/v1/predict/batch", func(w http.ResponseWriter, r *http.Request) {
+		var req struct {
+			States []json.RawMessage `json:"states"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			t.Errorf("decode batch request: %v", err)
+		}
+		f.batchSizes = len(req.States)
+		results := make([]Response, 0, len(req.States))
+		for range req.States {
+			results = append(results, Response{Model: "test-model", Answers: map[string]Answer{
+				"action": {Type: KindChoice, Choice: f.choice, Probabilities: f.prob},
+			}})
+		}
+		writeJSON(w, map[string]any{"results": results})
+	})
+	mux.HandleFunc("/boom", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusBadGateway)
+		_, _ = w.Write([]byte("upstream exploded"))
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func (f *fake) respond(w http.ResponseWriter, answers map[string]Answer) {
+	if f.status != 0 {
+		w.WriteHeader(f.status)
+		return
+	}
+	writeJSON(w, Response{Model: "test-model", Answers: answers, Usage: Usage{InputTokens: 128}})
+}
+
+func writeJSON(w http.ResponseWriter, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+func TestPredictSendsTypedQuestionsAndReadsAnswers(t *testing.T) {
+	f := &fake{choice: "restart", prob: map[string]float64{"restart": 0.9, "notify": 0.1}}
+	srv := f.server(t)
+	c := NewClient(srv.URL)
+
+	resp, err := c.Predict(context.Background(), "checkout-api is 500ing", Questions())
+	if err != nil {
+		t.Fatalf("Predict: %v", err)
+	}
+	if got := resp.Answers["action"].Choice; got != "restart" {
+		t.Errorf("choice = %q, want restart", got)
+	}
+	if got := resp.Answers["urgency"].Score; got != 3 {
+		t.Errorf("urgency score = %v, want 3", got)
+	}
+	if got := resp.Usage.InputTokens; got != 128 {
+		t.Errorf("input tokens = %d, want 128", got)
+	}
+	// The choice question must carry a criteria object and the score question an
+	// array, from the same "criteria" key.
+	if q := string(f.lastQs["action"]); !strings.Contains(q, `"criteria":{"escalate"`) {
+		t.Errorf("action question = %s, want a criteria object", q)
+	}
+	if q := string(f.lastQs["urgency"]); !strings.Contains(q, `"criteria":[`) {
+		t.Errorf("urgency question = %s, want a criteria array", q)
+	}
+	if f.lastState != `"checkout-api is 500ing"` {
+		t.Errorf("state = %s, want the JSON-encoded state", f.lastState)
+	}
+}
+
+func TestPredictBatchKeepsOrder(t *testing.T) {
+	srv := (&fake{choice: "notify", prob: map[string]float64{"notify": 1}}).server(t)
+	got, err := NewClient(srv.URL).PredictBatch(context.Background(), []any{"a", "b", "c"}, Questions())
+	if err != nil {
+		t.Fatalf("PredictBatch: %v", err)
+	}
+	if len(got) != 3 {
+		t.Fatalf("got %d results, want 3", len(got))
+	}
+	for i, r := range got {
+		if r.Answers["action"].Choice != "notify" {
+			t.Errorf("result %d choice = %q", i, r.Answers["action"].Choice)
+		}
+	}
+}
+
+func TestPredictErrorsAreSafeToLog(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+	_, err := NewClient(srv.URL).Predict(context.Background(), "x", Questions())
+	if err == nil {
+		t.Fatal("expected an error")
+	}
+	if !strings.Contains(err.Error(), "503") {
+		t.Errorf("error = %v, want the status code", err)
+	}
+}
+
+func TestScrubRemovesCredentialsFromURL(t *testing.T) {
+	got := scrub(`Post "https://user:hunter2@laya.k6n.net/v1/predict": timeout`)
+	if strings.Contains(got, "hunter2") {
+		t.Errorf("scrub(%q) leaked the password: %s", "user:hunter2@", got)
+	}
+	if !strings.Contains(got, "laya.k6n.net") {
+		t.Errorf("scrub(%q) dropped the host: %s", "user:hunter2@", got)
+	}
+}
+
+func TestHealth(t *testing.T) {
+	srv := (&fake{}).server(t)
+	h, err := NewClient(srv.URL).Health(context.Background())
+	if err != nil {
+		t.Fatalf("Health: %v", err)
+	}
+	if h.Model != "test-model" || h.Status != "ok" {
+		t.Errorf("health = %+v, want status ok from test-model", h)
+	}
+}
+
+// --- state rendering and gates ---
+
+func nowFixture() time.Time { return time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) }
+
+func healthyPod() Pod {
+	return Pod{Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-7d9", Found: true, Ready: true, Phase: "Running"}
+}
+
+func engineFor(t *testing.T, policy Policy, choice string, prob float64) (*Engine, *fake) {
+	t.Helper()
+	f := &fake{choice: choice, prob: map[string]float64{choice: prob}}
+	srv := f.server(t)
+	policy.Now = nowFixture
+	return NewEngine(NewClient(srv.URL), policy), f
+}
+
+func TestRenderStatesPodEvents(t *testing.T) {
+	exit := int32(137)
+	in := Input{
+		AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 412, Window: 5 * time.Minute,
+		BaselineMean: 3, NewErrors: true, Clients: 42,
+		Patterns: []Pattern{{What: "HTTP 500 | /api/cart | panic", Category: "service-defect", Status: 500, Count: 300, Baseline: 2, New: true, Clients: 42}},
+		Pods: []Pod{{
+			Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-7d9", Found: true, Phase: "Running",
+			Restarts: 3, PodRestarts: 9, Reason: "OOMKilled", ExitCode: &exit, LastTerminatedAt: nowFixture().Add(-30 * time.Second),
+			LastEvents: []string{"panic: runtime error: invalid memory address"},
+		}},
+		Deploys: []Deploy{{Namespace: "shop", Workload: "checkout-api", Age: 4 * time.Minute, Ready: 1, Desired: 3}},
+	}
+	state := in.Render(nowFixture())
+	for _, want := range []string{
+		"shop/checkout-api pod checkout-api-7d9",
+		"OOMKilled",
+		"exit code 137",
+		"its container restarted 30s ago and is still coming up",
+		"9 restarts across the Deployment",
+		"panic: runtime error: invalid memory address",
+		"rollout in progress started 4m0s ago",
+		"137x normal",
+	} {
+		if !strings.Contains(state, want) {
+			t.Errorf("state is missing %q:\n%s", want, state)
+		}
+	}
+}
+
+func TestDecideExecutesRestartOnlyWhenPodsSupportIt(t *testing.T) {
+	e, f := engineFor(t, Policy{Actions: []Action{ActionRestart, ActionIgnore, ActionNotify, ActionEscalate}, MinProbability: 0.5}, "restart", 0.82)
+	v := e.Decide(context.Background(), Input{
+		AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 100, Window: 5 * time.Minute, BaselineMean: 2, Clients: 30,
+		Patterns: []Pattern{{What: "HTTP 500 | /api/cart", Category: "service-defect", Count: 90, Baseline: 1, New: true, Clients: 30}},
+		Pods:     []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "p1", Found: true, Ready: false, Phase: "Running"}},
+	})
+	if !v.Executable() || v.Action != ActionRestart {
+		t.Fatalf("action = %q executable=%v gates=%v, want an executed restart", v.Action, v.Executable(), v.Gates)
+	}
+	if len(v.Targets) != 1 || v.Targets[0] != "shop/checkout-api" {
+		t.Errorf("targets = %v, want shop/checkout-api", v.Targets)
+	}
+	if !strings.Contains(f.lastState, "pod p1") {
+		t.Errorf("model was not given the pod state:\n%s", f.lastState)
+	}
+}
+
+func TestDecideRefusesRestartDuringCrashLoop(t *testing.T) {
+	e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart}, MinProbability: 0.5}, "restart", 0.95)
+	v := e.Decide(context.Background(), Input{
+		AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 300, Window: 5 * time.Minute, BaselineMean: 2,
+		Patterns: []Pattern{{What: "HTTP 500 | panic", Category: "service-defect", Count: 300, New: true}},
+		Pods:     []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "p1", Found: true, Phase: "Running", Restarts: 5, LastTerminatedAt: nowFixture().Add(-20 * time.Second)}},
+	})
+	if v.Action != ActionNotify {
+		t.Errorf("action = %q, want notify: a crash loop needs a new release, not a restart", v.Action)
+	}
+	if len(v.Gates) == 0 || !strings.Contains(strings.Join(v.Gates, " "), "crash-looping") {
+		t.Errorf("gates = %v, want the crash-loop reason", v.Gates)
+	}
+}
+
+// A short 503 burst while every pod is being replaced is the rollout's own, so
+// a restart must never be fired into it.
+func TestDecideRefusesRestartDuringRollout(t *testing.T) {
+	e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart, ActionIgnore}, MinProbability: 0.5}, "restart", 0.9)
+	v := e.Decide(context.Background(), Input{
+		AlertID: "a1", Service: "checkout-api", Priority: "medium", ErrorCount: 40, Window: 5 * time.Minute, BaselineMean: 2,
+		Patterns: []Pattern{{What: "HTTP 503 | /api/cart", Category: "capacity-saturation", Status: 503, Count: 40}},
+		Pods:     []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "p1", Found: true, Phase: "Pending", LastTerminatedAt: nowFixture().Add(-40 * time.Second)}},
+		Deploys:  []Deploy{{Namespace: "shop", Workload: "checkout-api", Age: 2 * time.Minute, Ready: 1, Desired: 3}},
+	})
+	if v.Action == ActionRestart {
+		t.Fatalf("action = restart during a rollout, want it withheld (gates=%v)", v.Gates)
+	}
+	if !strings.Contains(strings.Join(v.Gates, " "), "replaced by a rollout") {
+		t.Errorf("gates = %v, want the rollout reason", v.Gates)
+	}
+}
+
+func TestDecideRefusesIgnoreOfANewError(t *testing.T) {
+	e, _ := engineFor(t, Policy{Actions: []Action{ActionIgnore}, MinProbability: 0.5}, "ignore", 0.9)
+	v := e.Decide(context.Background(), Input{
+		AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 200, Window: 5 * time.Minute, BaselineMean: 2, NewErrors: true,
+		Patterns: []Pattern{{What: "HTTP 500 | /api/cart", Count: 200, New: true}},
+		Pods:     []Pod{healthyPod()},
+	})
+	if v.Action != ActionNotify {
+		t.Errorf("action = %q, want notify: a new error was never confirmed harmless", v.Action)
+	}
+}
+
+func TestDecideAllowsIgnoreOfKnownNoise(t *testing.T) {
+	e, _ := engineFor(t, Policy{Actions: []Action{ActionIgnore}, MinProbability: 0.5}, "ignore", 0.88)
+	v := e.Decide(context.Background(), Input{
+		AlertID: "a1", Service: "blog", Priority: "low", ErrorCount: 1800, Window: 5 * time.Minute, BaselineMean: 20, KnownNoise: true,
+		Patterns: []Pattern{{What: "HTTP 404 | /old-post", Status: 404, Count: 1800}},
+	})
+	if v.Action != ActionIgnore || !v.Executable() {
+		t.Errorf("action = %q executable = %v, want an executable ignore", v.Action, v.Executable())
+	}
+}
+
+func TestDecideWithholdsActionsOutsideTheAllowlist(t *testing.T) {
+	e, _ := engineFor(t, Policy{Actions: []Action{ActionNotify}, MinProbability: 0.5}, "ignore", 0.99)
+	v := e.Decide(context.Background(), Input{
+		AlertID: "a1", Service: "blog", Priority: "low", ErrorCount: 100, Window: 5 * time.Minute, BaselineMean: 20,
+		Patterns: []Pattern{{What: "HTTP 404", Status: 404, Count: 100}},
+		Pods:     []Pod{healthyPod()},
+	})
+	if v.Action != ActionNotify || v.Requested != ActionIgnore || !v.Changed() {
+		t.Errorf("action = %q requested = %q changed = %v, want ignore downgraded to notify", v.Action, v.Requested, v.Changed())
+	}
+}
+
+func TestDecideDowngradesALowConfidenceChoice(t *testing.T) {
+	e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart, ActionIgnore}, MinProbability: 0.6}, "restart", 0.41)
+	v := e.Decide(context.Background(), Input{
+		AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 90, Window: 5 * time.Minute, BaselineMean: 2,
+		Patterns: []Pattern{{What: "HTTP 500", Count: 90, New: true}},
+		Pods:     []Pod{{Namespace: "shop", Deployment: "checkout-api", Found: true, Ready: false, Phase: "Running"}},
+	})
+	if v.Action != ActionNotify {
+		t.Errorf("action = %q, want notify: 41%% is a guess", v.Action)
+	}
+}
+
+func TestRestartBudgetIsEnforced(t *testing.T) {
+	e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart}, MinProbability: 0.5, MaxRestartsPerHour: 1}, "restart", 0.9)
+	in := Input{
+		AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 90, Window: 5 * time.Minute, BaselineMean: 2,
+		Patterns: []Pattern{{What: "HTTP 500", Count: 90, New: true}},
+		Pods:     []Pod{{Namespace: "shop", Deployment: "checkout-api", Found: true, Ready: false, Phase: "Running"}},
+	}
+	if v := e.Decide(context.Background(), in); v.Action != ActionRestart {
+		t.Fatalf("first decision = %q, want restart", v.Action)
+	}
+	second := e.Decide(context.Background(), in)
+	if second.Action != ActionNotify {
+		t.Errorf("second decision = %q, want notify: the hourly restart budget is spent", second.Action)
+	}
+}
+
+func TestDecideDegradesToNotifyWhenTheModelIsDown(t *testing.T) {
+	f := &fake{status: http.StatusServiceUnavailable}
+	srv := f.server(t)
+	e := NewEngine(NewClient(srv.URL), Policy{Actions: []Action{ActionRestart}, Now: nowFixture})
+	v := e.Decide(context.Background(), Input{AlertID: "a1", Service: "checkout-api", ErrorCount: 90, Window: 5 * time.Minute})
+	if v.Action != ActionNotify || v.Executable() || v.Error == "" {
+		t.Errorf("action = %q executable = %v error = %q, want a plain notification", v.Action, v.Executable(), v.Error)
+	}
+}
+
+func TestDecideRejectsAnUnknownAction(t *testing.T) {
+	f := &fake{choice: "delete-the-cluster", prob: map[string]float64{"delete-the-cluster": 1}}
+	srv := f.server(t)
+	e := NewEngine(NewClient(srv.URL), Policy{Actions: []Action{ActionRestart}, Now: nowFixture})
+	v := e.Decide(context.Background(), Input{AlertID: "a1", Service: "x", ErrorCount: 1, Window: time.Minute})
+	if v.Error == "" || v.Executable() {
+		t.Errorf("error = %q executable = %v, want an unknown action to be rejected", v.Error, v.Executable())
+	}
+}

internal/laya/live_test.go

@@ -0,0 +1,59 @@
+package laya
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+)
+
+// TestLiveModel is a smoke test against the real Laya API, so the state
+// rendering and the questions stay in step with what the model actually
+// understands. It is opt-in: LAYA_URL=https://laya.k6n.net go test ./internal/laya -run TestLiveModel
+func TestLiveModel(t *testing.T) {
+	base := os.Getenv("LAYA_URL")
+	if base == "" {
+		t.Skip("set LAYA_URL to run against the live model")
+	}
+	cases := []struct {
+		name string
+		in   Input
+	}{
+		{
+			name: "crash loop after a release",
+			in: Input{
+				Service: "checkout-api", Priority: "high", ErrorCount: 412, Window: 5 * time.Minute, BaselineMean: 3, NewErrors: true, Clients: 42,
+				Patterns: []Pattern{{What: "HTTP 500 | /api/cart | panic: runtime error: invalid memory address", Category: "service-defect", Status: 500, Count: 300, Baseline: 2, New: true, Clients: 42, Services: []string{"checkout-api", "cart-service"}}},
+				Pods:     []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-7d9", Found: true, Phase: "Running", Restarts: 3, PodRestarts: 9, Reason: "CrashLoopBackOff", LastTerminatedAt: time.Now().Add(-30 * time.Second), LastEvents: []string{"panic: runtime error: invalid memory address", "SIGTERM received"}}},
+				Deploys:  []Deploy{{Namespace: "shop", Workload: "checkout-api", Release: "v2.31", Age: 4 * time.Minute, Ready: 1, Desired: 3}},
+			},
+		},
+		{
+			name: "503 burst while a pod is being replaced",
+			in: Input{
+				Service: "checkout-api", Priority: "medium", ErrorCount: 40, Window: 5 * time.Minute, BaselineMean: 2, Clients: 8,
+				Patterns: []Pattern{{What: "HTTP 503 | /api/cart | no healthy upstream", Category: "capacity-saturation", Status: 503, Count: 40, Baseline: 1, Clients: 8}},
+				Pods:     []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-2k1", Found: true, Phase: "Pending", LastTerminatedAt: time.Now().Add(-40 * time.Second), LastEvents: []string{"terminated"}}},
+				Deploys:  []Deploy{{Namespace: "shop", Workload: "checkout-api", Age: 2 * time.Minute, Ready: 1, Desired: 3}},
+			},
+		},
+		{
+			name: "crawler 404s, nobody affected",
+			in: Input{
+				Service: "blog", Priority: "low", ErrorCount: 1800, Window: 5 * time.Minute, BaselineMean: 20, KnownNoise: true,
+				Patterns: []Pattern{{What: "HTTP 404 | /old-post-2019 | UA looks like a crawler", Status: 404, Count: 1800, Baseline: 15}},
+				Pods:     []Pod{{Namespace: "shop", Deployment: "blog", Name: "blog-6f2", Found: true, Ready: true, Phase: "Running", LastEvents: []string{"GET /old-post-2019 404"}}},
+			},
+		},
+	}
+	// Engine defaults, as deployed: the label probabilities are diffuse, so the
+	// gate is the lead over the runner-up.
+	engine := NewEngine(NewClient(base), Policy{Actions: Actions, Now: time.Now})
+	for _, tc := range cases {
+		v := engine.Decide(context.Background(), tc.in)
+		t.Logf("%s:\n%s\n  -> %s\n", tc.name, v.State, v.Explain())
+		if v.Error != "" {
+			t.Errorf("%s: %v", tc.name, v.Error)
+		}
+	}
+}

internal/topology/podlogs.go

@@ -16,6 +16,38 @@ type PodLog struct {
 	Lines []string // most recent log lines, oldest first
 	Found bool     // a pod exists and was queried (Lines may still be empty: a quiet pod)
 	Err   string   // set when the lookup itself failed; Found is false
+
+	// The pod's own lifecycle facts, which explain errors the log lines alone
+	// cannot: a 503 burst while a replica is being replaced, a crash loop the
+	// kubelet is already retrying, or a pod that runs but never becomes ready.
+	Ready            bool      // container passing its readiness probe
+	Phase            string    // Pending | Running | Succeeded | Failed | Unknown
+	Restarts         int32     // restarts of the selected pod's containers
+	PodRestarts      int32     // restarts summed over every pod of the Deployment
+	Reason           string    // CrashLoopBackOff, OOMKilled, ImagePullBackOff, ... ("" while healthy)
+	ExitCode         *int32    // exit code of the last container termination
+	Message          string    // message of the last container termination
+	LastTerminatedAt time.Time // when the last container terminated
+}
+
+// rollingWindow is how recently a container must have died for the resulting
+// connection errors to be attributed to its replacement still coming up.
+const rollingWindow = 5 * time.Minute
+
+// CrashLooping reports whether the pod is stuck restarting instead of serving.
+// Restarting the Deployment then only repeats what the kubelet is already
+// doing, so the answer is a new release, not a restart.
+func (p PodLog) CrashLooping() bool {
+	if p.Reason == "CrashLoopBackOff" {
+		return true
+	}
+	return p.Restarts > 0 && !p.LastTerminatedAt.IsZero()
+}
+
+// Restarting reports whether a pod of this Deployment is right now being
+// replaced, which is what makes a short 503 burst expected rather than new.
+func (p PodLog) Restarting() bool {
+	return !p.LastTerminatedAt.IsZero() && time.Since(p.LastTerminatedAt) < rollingWindow
 }
 
 // podLogTimeout bounds one lookup so a slow or unreachable API server never
@@ -64,6 +96,7 @@ func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, ta
 		return live[i].CreationTimestamp.After(live[j].CreationTimestamp.Time)
 	})
 	pod := live[0]
+	state := podState(pod, live)
 	container := ""
 	if len(pod.Spec.Containers) > 0 {
 		container = pod.Spec.Containers[0].Name
@@ -71,7 +104,7 @@ func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, ta
 	tailLines := int64(tail)
 	stream, err := r.client.CoreV1().Pods(namespace).GetLogs(pod.Name, &corev1.PodLogOptions{Container: container, TailLines: &tailLines}).Stream(ctx)
 	if err != nil {
-		return PodLog{Pod: pod.Name, Found: true, Err: err.Error()}
+		return withState(state, PodLog{Pod: pod.Name, Found: true, Err: err.Error()})
 	}
 	defer stream.Close()
 	var lines []string
@@ -83,5 +116,53 @@ func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, ta
 	if len(lines) > tail {
 		lines = lines[len(lines)-tail:]
 	}
-	return PodLog{Pod: pod.Name, Lines: lines, Found: true}
+	return withState(state, PodLog{Pod: pod.Name, Lines: lines, Found: true})
+}
+
+// withState keeps the query result but carries the lifecycle facts along.
+func withState(state, result PodLog) PodLog {
+	result.Ready, result.Phase, result.Restarts, result.PodRestarts = state.Ready, state.Phase, state.Restarts, state.PodRestarts
+	result.Reason, result.ExitCode, result.Message, result.LastTerminatedAt = state.Reason, state.ExitCode, state.Message, state.LastTerminatedAt
+	return result
+}
+
+// podState reads the lifecycle facts off the selected pod and totals the
+// restarts of its siblings, so one unhealthy replica is not mistaken for a
+// healthy deployment.
+func podState(selected corev1.Pod, all []corev1.Pod) PodLog {
+	state := PodLog{Found: true, Ready: len(selected.Status.ContainerStatuses) > 0, Phase: string(selected.Status.Phase)}
+	// The most recent termination across the pod's containers is the
+	// interesting one: a pod being replaced is explained by its last death.
+	keepTermination := func(term *corev1.ContainerStateTerminated) {
+		if term == nil || !term.FinishedAt.Time.After(state.LastTerminatedAt) {
+			return
+		}
+		state.LastTerminatedAt = term.FinishedAt.Time
+		state.Message = term.Message
+		code := term.ExitCode
+		state.ExitCode = &code
+		switch {
+		case term.Reason != "":
+			state.Reason = term.Reason
+		case term.Message != "":
+			state.Reason = "Error"
+		}
+	}
+	for i := range selected.Status.ContainerStatuses {
+		cs := &selected.Status.ContainerStatuses[i]
+		state.Restarts += cs.RestartCount
+		if !cs.Ready {
+			state.Ready = false
+		}
+		keepTermination(cs.LastTerminationState.Terminated)
+	}
+	if selected.Status.Reason != "" {
+		state.Reason = selected.Status.Reason
+	}
+	for _, p := range all {
+		for _, cs := range p.Status.ContainerStatuses {
+			state.PodRestarts += cs.RestartCount
+		}
+	}
+	return state
 }