Genkit Flows on Kubernetes, Routed by a Decision Model: senro + jev in Go

Introduction

Genkit flows as senro steps showed how to put a Genkit flow inside a senro step and get retries, caching and an event stream for every model call. Every step in that article ran on your own machine.

This one moves the steps into Kubernetes and adds a second kind of model. Gemini writes. jev decides. jev is a decision model: it does not generate text, it evaluates a state against typed questions and answers each one with a probability, in a few hundred milliseconds. Genkit Go 1.14 reaches it through the System One plugin, and a decision is one genkit.GenerateData call whose output type is the question set.

That makes jev the right model to sit at the switch points of a pipeline. In this guide it decides twice: which department's workflow runs for a support ticket, and whether the reply Gemini drafted is safe to send.

Everything below runs against the real Gemini API (gemini-3.8-flash) and the real TypeSafe API (jev-latest), on Genkit Go v1.14.0, senro v1.4.5 and a local kind cluster on Kubernetes 1.37. The code is in genkit-k8s-flows-jev.

Why a decision model decides and a chat model writes

Asking a chat model "which team should own this ticket, answer with one word" works until it does not: you get prose around the word, a department that is not on the list, no idea how sure the model was, and a second or two of latency for a decision that should take milliseconds. A decision model is built for exactly that call.

	Chat model as router	Decision model as router
Answer	Text you parse	A typed field: `Choice[Dept]`, `Noul`, `Score[L]`
Options	In the prompt, hoped for in the output	In the Go type, enforced by the schema
Confidence	None, or whatever the model says about itself	A probability per option, calibrated per model version
Tools	Can be called by accident	Refused by the protocol
Latency	Seconds	A few hundred milliseconds
Cost per decision	Tokens for the state and the prose	Tokens for the state

And the reverse is just as true. jev cannot write the reply to the customer, explain an outage or summarise a brief. That is what gemini-3.8-flash is for. The pipeline needs both, and senro is what puts each call in its own step, on its own pod, with its own log, retry policy and secret.

What we are building

A support-ticket pipeline. A raw email comes in; a reply goes out, or a brief goes to a human, or both.

Two runs, because of one rule in senro: a When condition is evaluated once, at run start, against facts the run already has, never against something a step produces. So the first run understands the ticket and decides, and the decision becomes the run parameters of the second run, where senro.When(senro.ParamIs(...)) prunes every branch the verdict did not pick before anything starts. Pruned steps settle as skipped_condition and the run stays green.

Run	Step	Where	Model	Writes
`triage`	`seed`	coordinator	none	puts the raw email into the run's workspace
`triage`	`ingest`	pod	Gemini	`ticket.json`, a structured ticket
`triage`	`triage`	pod	jev	`decision.json`: department, urgency, frustration, needs a human
`triage`	`collect`	coordinator	none	copies both files out of the run
`resolve`	`<dept>/draft`	pod	Gemini	`reply.md`, following that department's playbook
`resolve`	`<dept>/verify`	pod	jev	`verdict.json`; a rejection fails the step
`resolve`	`<dept>/send`	coordinator	none	the reply, only if `verify` passed
`resolve`	`escalate/brief`	pod	Gemini	`brief.md` for a human, when jev said so

Inside a branch, jev is a guardrail: verify reads the draft against the playbook, and if it rejects the draft, the step fails and senro skips send. The second decision also determines the next step, this time without leaving the run.

What you need

Go 1.27 or newer, Docker, kind and kubectl
GEMINI_API_KEY and TYPESAFE_API_KEY in your environment
the senro CLI if you want the terminal UI (optional)

go get github.com/firebase/genkit/go@v1.14.0 go get github.com/xavidop/senro@v1.4.5 go get github.com/xavidop/mamori

mamori is the library senro uses to resolve secrets. It is how the two API keys reach the pods without ever being in a pod spec.

Step 1: a cluster senro can reach

senro's Kubernetes executor never reads KUBECONFIG, ~/.kube/config or your current context. A kubeconfig typically holds a dozen contexts, most of them production, with whichever one you last used selected; a pipeline that inherited that would eventually deploy into the wrong cluster. Instead, the cluster comes from three environment variables, read at run start:

Variable	What it is
`SENRO_K8S_SERVER`	The apiserver URL
`SENRO_K8S_CA_FILE`	The cluster CA, as a PEM file. senro will not skip verification
`SENRO_K8S_TOKEN_FILE`	A bearer token in a file (or a client certificate and key)

So a local cluster needs four things: kind, a namespace, an identity with the permissions the executor documents, and those three files. The repo's scripts/cluster-up.sh does all of it; this is what it runs.

kind create cluster --name senro-ai --kubeconfig .kind/kubeconfig export KUBECONFIG=$PWD/.kind/kubeconfig kubectl apply -f k8s/rbac.yaml

The RBAC is the exact set senro needs and nothing more: in the namespace, create, get and delete on pods, get on pods/log, create and get on pods/exec, and create, patch, delete on secrets. At cluster scope, list and get on nodes, which is how senro reads the platform it has to compile for.

The get on pods/exec is the one to remember: senro uses the WebSocket exec protocol, which is a GET on that subresource, and a Role with only create fails every step that carries a workspace.

apiVersion: rbac.authorization.k8s.io/v1 kind: Role metadata: name: senro-steps namespace: support rules: - apiGroups: [""] resources: ["pods"] verbs: ["create", "get", "delete"] - apiGroups: [""] resources: ["pods/log"] verbs: ["get"] - apiGroups: [""] resources: ["pods/exec"] verbs: ["create", "get"] - apiGroups: [""] resources: ["secrets"] verbs: ["create", "patch", "delete"]

Then the three files, pulled out of the kubeconfig kind wrote and a short-lived ServiceAccount token:

kubectl config view --minify --raw -o jsonpath='{.clusters[0].cluster.server}' > .kind/server kubectl config view --minify --raw \ -o jsonpath='{.clusters[0].cluster.certificate-authority-data}' | base64 -d > .kind/ca.crt kubectl -n support create token senro --duration=4h > .kind/token export SENRO_K8S_SERVER=$(cat .kind/server) export SENRO_K8S_CA_FILE=$PWD/.kind/ca.crt export SENRO_K8S_TOKEN_FILE=$PWD/.kind/token

A file rather than SENRO_K8S_TOKEN inline, because an environment variable is visible in /proc/<pid>/environ and inherited by every child process the coordinator spawns.

Step 2: the flows

Five flows, in one file. Nothing here knows about senro or Kubernetes.

Gemini turns an email into a ticket

type Ticket struct { Customer string `json:"customer"` Company string `json:"company,omitempty"` Summary string `json:"summary"` Ask string `json:"ask"` Facts []string `json:"facts"` } f.Ingest = genkit.DefineFlow(g, "ingestTicket", func(ctx context.Context, raw string) (Ticket, error) { t, _, err := genkit.GenerateData[Ticket](ctx, g, ai.WithModelName("googleai/gemini-3.8-flash"), ai.WithSystem("You turn raw support emails into a structured ticket. "+ "Facts are identifiers, amounts, dates and error codes quoted from the email, nothing inferred."), ai.WithPrompt("Email:\n\n%s", raw)) if err != nil { return Ticket{}, err } if t == nil { return Ticket{}, fmt.Errorf("ingest: the model returned no ticket") } return *t, nil })

jev decides what the ticket is

With the System One plugin, the output type is the question set. Each field is one question, its jsonschema_description tag is the instruction, and the options, levels and criteria belong to Go types, so the same question means the same thing wherever it is asked.

import systemonex "github.com/firebase/genkit/go/plugins/systemone/exp" type Dept string func (Dept) Criteria() map[Dept]string { return map[Dept]string{ "billing": "Charges, invoices, refunds, subscriptions, plan changes", "technical": "Bugs, errors, outages, latency, integrations, webhooks, API", "account": "Login, password, permissions, data export or deletion, privacy", "other": "None of the above", } } type Frustration int func (Frustration) Levels() []string { return []string{"Calm", "Annoyed", "Angry", "Furious"} } type Urgent struct{} func (Urgent) Criteria() (yes, no string) { return "Names a deadline, says now or today, or describes an outage in progress", "No time pressure is expressed, or the customer says there is no rush" } type Triage struct { Department systemonex.Choice[Dept] `json:"department" jsonschema_description:"Which team should own this ticket?"` IsUrgent systemonex.NoulOf[Urgent] `json:"is_urgent" jsonschema_description:"Does the ticket communicate time pressure?"` Frustration systemonex.Score[Frustration] `json:"frustration" jsonschema_description:"How frustrated is the customer?"` NeedsHuman systemonex.Noul `json:"needs_human" jsonschema_description:"Does resolving this need a human: a legal threat, a contract or SLA dispute, data loss, or a refund outside the playbook?"` }

Four question kinds cover the protocol. Choice[T] picks one of T.Criteria() and gives a probability per option plus a Confidence. Score[L] rates on L.Levels() and answers Level() and Label(). Noul is "is this true?" as a probability, where 0.5 means the model could not tell. NoulOf[C] is the same with C.Criteria() saying what yes and no mean.

The flow sends the ticket as a data part, so the state is a JSON object and a question can refer to a field by name. It then turns the four answers into a flat verdict, with thresholds that belong to the caller:

type Decision struct { Department string `json:"department"` Confidence float64 `json:"confidence"` Urgent bool `json:"urgent"` Frustration string `json:"frustration"` Escalate bool `json:"escalate"` Model string `json:"model"` } f.Triage = genkit.DefineFlow(g, "triageTicket", func(ctx context.Context, t Ticket) (Decision, error) { out, resp, err := genkit.GenerateData[Triage](ctx, g, ai.WithModelName("typesafe/jev-latest"), ai.WithPromptParts(ai.NewDataPart(t))) if err != nil { return Decision{}, err } d := Decision{ Department: string(out.Department.Choice), Confidence: out.Department.Confidence, Urgent: out.IsUrgent.Probability >= 0.5, Frustration: out.Frustration.Label(), Model: systemonex.ResponseInfo(resp).Model, } // A low-confidence routing decision is not a routing decision. if d.Confidence < 0.5 { d.Department = "other" } d.Escalate = d.Department == "other" || out.NeedsHuman.Probability >= 0.6 || out.Frustration.Level() >= 3 return d, nil })

ResponseInfo(resp).Model records which jev version answered. Confidence thresholds are calibrated per version, so a verdict that does not say who produced it is a threshold waiting to drift.

Gemini drafts, jev reviews

The draft flow is an ordinary GenerateData[Reply] call with the playbook and the ticket in the prompt. One difference from the triage flow: the ticket goes to Gemini as JSON text. A data part is how a decision model takes an object state; a chat model takes a prompt.

f.Draft = genkit.DefineFlow(g, "draftReply", func(ctx context.Context, in DraftInput) (Reply, error) { r, _, err := genkit.GenerateData[Reply](ctx, g, ai.WithModelName("googleai/gemini-3.8-flash"), ai.WithSystem("You are a support agent for Acme Cloud. Write the reply the customer will receive. "+ "Follow the playbook strictly. Never state amounts, dates or identifiers that are not in the ticket. "+ "Plain text, under 180 words, sign as 'Acme Cloud Support'."), ai.WithPrompt("Playbook:\n%s\n\n%s\n\nTicket:\n%s", in.Playbook, urgency, asJSON(in.Ticket))) // ... })

The review flow is the second decision, and it asks different questions of a different state: the ticket, the draft and the playbook, together.

type Overpromise struct{} func (Overpromise) Criteria() (yes, no string) { return "Commits to a refund amount, a credit, a release date, a deadline, or a deletion the playbook does not allow", "Every commitment in the reply is one the playbook permits" } type Tone int func (Tone) Levels() []string { return []string{"Cold or dismissive", "Neutral", "Warm and specific"} } type Review struct { Addresses systemonex.Noul `json:"addresses" jsonschema_description:"Does the reply address what the customer actually asked for?"` Overpromises systemonex.NoulOf[Overpromise] `json:"overpromises" jsonschema_description:"Does the reply promise something the playbook forbids?"` Invents systemonex.Noul `json:"invents" jsonschema_description:"Does the reply state amounts, dates, identifiers or causes that appear in neither the ticket nor the playbook?"` Tone systemonex.Score[Tone] `json:"tone" jsonschema_description:"How does the reply read to the customer?"` } f.Verify = genkit.DefineFlow(g, "verifyReply", func(ctx context.Context, in VerifyInput) (Verdict, error) { out, _, err := genkit.GenerateData[Review](ctx, g, ai.WithModelName("typesafe/jev-latest"), ai.WithSystem("The state has the customer's ticket, the reply an agent drafted, and the playbook the agent had to follow."), ai.WithPromptParts(ai.NewDataPart(in))) if err != nil { return Verdict{}, err } v := Verdict{Addresses: out.Addresses.Probability, Overpromises: out.Overpromises.Probability, Invents: out.Invents.Probability, Tone: out.Tone.Label()} if v.Addresses < 0.6 { v.Reasons = append(v.Reasons, "the reply does not address the customer's ask") } if v.Overpromises >= 0.5 { v.Reasons = append(v.Reasons, "the reply promises something the playbook forbids") } if v.Invents >= 0.5 { v.Reasons = append(v.Reasons, "the reply invents facts") } if out.Tone.Level() == 0 { v.Reasons = append(v.Reasons, "the tone is cold") } v.Approved = len(v.Reasons) == 0 return v, nil })

A system message is never part of the state with this plugin. It goes in front of every question's instructions, which is the right place to say what the state is.

Step 3: flows as steps that can run in a pod

A senro func step is a registered Go function. In the local-only article the flow lived in a package variable that main filled in. In a pod that does not work, because senro runs a func step by sending your pipeline binary into the pod and re-entering it as a fresh process; main never gets to the line that initialised Genkit. So Genkit is initialised lazily, inside the step, from the secrets senro delivered to that pod:

var ( engineOnce sync.Once engine *Flows engineErr error ) func flows(ctx senro.Ctx) (*Flows, error) { engineOnce.Do(func() { var plugins []api.Plugin gemini, err := readSecret(ctx, "Gemini") if err != nil { engineErr = err return } if gemini != "" { plugins = append(plugins, &googlegenai.GoogleAI{APIKey: gemini}) } typesafe, err := readSecret(ctx, "TypeSafe") if err != nil { engineErr = err return } if typesafe != "" { ts := systemonex.TypeSafe() ts.APIKey = typesafe plugins = append(plugins, ts) } g := genkit.Init(ctx, genkit.WithPlugins(plugins...)) engine = defineFlows(g) }) return engine, engineErr }

readSecret reads the file that ctx.Secret(name) names, and that is the point: in the pod it is a 0400 file under /run/senro/secrets, projected from a short-lived Kubernetes Secret owned by the pod. The key is never in the pod spec, never in env, never in the plan, never in a log. That is also why the plugin fields are set explicitly rather than read from the environment. And senro.Ctx embeds context.Context, so it goes straight into genkit.Init and flow.Run.

The steps themselves are short. Each one reads from the case workspace, runs a flow, and writes back:

const workspace = "case" func init() { senro.RegisterFunc("ai/ingest", Ingest) senro.RegisterFunc("ai/triage", TriageStep) senro.RegisterFunc("ai/draft", Draft) senro.RegisterFunc("ai/verify", Verify) senro.RegisterFunc("ai/brief", Brief) senro.RegisterFunc("ai/collect", Collect) senro.RegisterFunc("ai/seed", Seed) senro.RegisterFunc("ai/send", Send) } func...