# Production

Set up Temporal for durable workflows, stream events to UIs, apply adaptive rate limiting, and use system reminders.

Source: https://goa.design/docs/2-goa-ai/production/

Relative links resolve against the source URL above.


## Model Rate Limiting

Every model provider enforces rate limits. Exceed them and your requests fail with 429 errors. Worse: in a multi-replica deployment, each replica independently hammers the API, causing *aggregate* throttling that's invisible to individual processes.

### The Problem

**Scenario:** You deploy 10 replicas of your agent service. Each replica thinks it has 100K tokens/minute available. Combined, they send 1M tokens/minute—10x your actual quota. The provider throttles aggressively. Requests fail randomly across all replicas.

**Without rate limiting:**
- Requests fail unpredictably with 429s
- No visibility into remaining capacity
- Retries make congestion worse
- User experience degrades under load

**With adaptive rate limiting:**
- Each replica shares a coordinated budget
- Requests queue until capacity is available
- Backoff propagates across the cluster
- Graceful degradation instead of failures

### Overview

The `features/model/middleware` package provides an **AIMD-style adaptive rate
limiter** beneath the validated model client. It asks the provider for the exact
input-token count, blocks callers until that capacity is available, and adjusts
its input-tokens-per-minute budget in response to provider throttling. It never
estimates tokens and does not meter output quotas.

### AIMD Strategy

The limiter uses an **Additive Increase / Multiplicative Decrease (AIMD)** strategy:

| Event | Action | Formula |
|-------|--------|---------|
| Success | Probe (additive increase) | `TPM += recoveryRate` (5% of initial) |
| `ErrRateLimited` | Backoff (multiplicative decrease) | `TPM *= 0.5` |

The effective tokens-per-minute (TPM) is bounded by:
- **Minimum**: 10% of initial TPM (floor to prevent starvation)
- **Maximum**: The configured `maxTPM` ceiling

### Basic Usage

Create a single limiter per process and wrap your model client:

```go
import (
    "context"

    "goa.design/goa-ai/features/model/middleware"
    "goa.design/goa-ai/runtime/agent/runtime"
)

func main() {
    ctx := context.Background()
    rt := runtime.New(runtimeStore) // host-owned runtime storage

    // Vertex Gemini exposes the exact CountTokens operation required by the limiter.
    modelClient, err := rt.NewVertexGeminiModelClient(ctx, runtime.VertexConfig{
        ProjectID:    "my-gcp-project",
        Location:     "us-central1",
        DefaultModel: "gemini-2.5-flash",
    })
    if err != nil {
        panic(err)
    }

    limiter := middleware.NewAdaptiveRateLimiter(
        ctx,
        nil,     // process-local limiter
        "",      // unused for a process-local limiter
        60000,   // initial input tokens per minute
        120000,  // maximum input tokens per minute
    )

    rateLimitedClient, err := limiter.Middleware()(modelClient)
    if err != nil {
        panic(err)
    }

    if err := rt.RegisterModel("default", rateLimitedClient); err != nil {
        panic(err)
    }
}
```

### Cluster-Aware Rate Limiting

For multi-process deployments, coordinate rate limiting across instances using a Pulse replicated map:

```go
import (
    "context"

    "goa.design/goa-ai/features/model/middleware"
    "goa.design/pulse/rmap"
)

func main() {
    ctx := context.Background()

    // Create a Pulse replicated map backed by Redis
    rm, err := rmap.Join(ctx, "rate-limits", redisClient)
    if err != nil {
        panic(err)
    }
    defer rm.Close()

    limiter := middleware.NewAdaptiveRateLimiter(
        ctx,
        rm,
        "vertex:gemini",  // shared key for this model family
        60000,            // initial TPM
        120000,           // max TPM
    )

    rateLimitedClient, err := limiter.Middleware()(vertexClient)
    if err != nil {
        panic(err)
    }
}
```

While replicated-map reads and writes succeed:
- **Backoff propagates globally**: When any process receives `ErrRateLimited`, all processes reduce their budget
- **Probing is coordinated**: Successful requests increment the shared budget
- **Automatic reconciliation**: Processes watch for external changes and update their local limiters

Passing a nil map or empty key deliberately creates a process-local limiter.
When both are set, the shared key is absent, and the middleware cannot seed it,
the limiter also falls back to process-local operation. After startup, shared
backoff/probe update errors do not fail model calls; that process keeps its
local adaptive budget until a later replicated-map event reconciles it. Monitor
Redis availability if cluster-wide coordination is required.

### Exact Token Counting

The limiter calls the wrapped client's `CountTokens` operation before reserving
capacity. It requires `Exact=true` and never estimates.

Vertex Gemini exposes a native count operation. Bedrock uses Runtime
`CountTokens` where supported, but returns
`model.ErrTokenCountingUnsupported` for structured-output requests and for
models such as Claude Opus 4.7, Sonnet 5, and Mythos 5 that require the separate
AWS Mantle endpoint. A remote gateway preserves counting only when constructed
with `gateway.NewCountingRemoteClient`. OpenAI has no native token counter.

Wrapping a client succeeds even when counting is unsupported. The first
`Complete` or `Stream` call then returns
`model.ErrTokenCountingUnsupported` before inference.

The limiter probes upward after a successful unary call or clean stream end. It
backs off after a terminal `model.ErrRateLimited`. Opening or closing a stream
without reaching a terminal outcome does not alter capacity.

### Integration with Runtime

Wire rate-limited clients into the Goa-AI runtime:

```go
vertexLimiter := middleware.NewAdaptiveRateLimiter(ctx, nil, "", 60000, 120000)
limitedVertex, err := vertexLimiter.Middleware()(vertexClient)
if err != nil {
    panic(err)
}

rt := runtime.New(runtimeStore, runtime.WithEngine(temporalEng))
if err := rt.RegisterModel("gemini", limitedVertex); err != nil {
    panic(err)
}
```

### What Happens Under Load

| Traffic Level | Without Limiter | With Limiter |
|---------------|-----------------|--------------|
| Below quota | Requests succeed | Requests succeed |
| At quota | Random 429 failures | Requests queue, then succeed |
| Burst above quota | Cascade of failures, provider blocks | Backoff absorbs burst, gradual recovery |
| Sustained overload | All requests fail | Requests queue with bounded latency |

### Tuning Parameters

| Parameter | Default | Description |
|-----------|---------|-------------|
| `initialTPM` | (required) | Starting tokens-per-minute budget |
| `maxTPM` | (required) | Ceiling for probing |
| Floor | 10% of initial | Minimum budget (prevents starvation) |
| Recovery rate | 5% of initial | Additive increase per success |
| Backoff factor | 0.5 | Multiplicative decrease on 429 |

**Example:** With `initialTPM=60000, maxTPM=120000`:
- Floor: 6,000 TPM
- Recovery: +3,000 TPM per successful batch
- Backoff: halve current TPM on 429

### Monitoring

Measure model-call latency and terminal `model.ErrRateLimited` errors in the
telemetry around the wrapped client. Also monitor Redis when using a replicated
map, because the middleware intentionally keeps model calls running with a
process-local budget if shared-state initialization or updates fail.

### Best Practices

- **One limiter per model/provider**: Create separate limiters for different models to isolate their budgets
- **Set realistic initial TPM**: Start with your provider's documented rate limit or a conservative estimate
- **Use cluster-aware limiting in production**: Coordinate across replicas to avoid aggregate throttling
- **Monitor backoff events**: Log or emit metrics when backoffs occur to detect sustained throttling
- **Set maxTPM above initial**: Leave headroom for probing when traffic is below quota

---

## OpenTelemetry GenAI Observability

When a tracer is configured, Goa-AI emits vendor-neutral OpenTelemetry GenAI semantic-convention spans for planner-scoped agent operations:

- model calls use `gen_ai.operation.name="chat"` and span names like `chat {model}`
- tool completions use `gen_ai.operation.name="execute_tool"` and span names like `execute_tool {tool_name}`
- agent-as-tool delegation uses `gen_ai.operation.name="invoke_agent"` and span names like `invoke_agent {agent_name}`

These spans include `gen_ai.conversation.id`, `gen_ai.agent.id`, `gen_ai.agent.name`, `gen_ai.request.model`, `gen_ai.response.model`, token usage, finish reasons, tool identifiers, and streaming time-to-first-chunk when available. The runtime records identifiers, counts, timings, and errors by default; prompt text, chat history, tool arguments, and tool results remain application policy and are not attached automatically.

This keeps open-source Goa-AI telemetry portable across OpenTelemetry backends while still giving production systems enough structure to group a conversation, compare model latency and token usage, and inspect multi-agent tool chains.

---

## Prompt Overrides with Mongo Store

Production prompt management typically uses:

- baseline prompt specs registered in `runtime.PromptRegistry`, and
- scoped override records persisted in Mongo via `features/prompt/mongo`.

### Wiring

```go
import (
    promptmongo "goa.design/goa-ai/features/prompt/mongo"
    clientmongo "goa.design/goa-ai/features/prompt/mongo/clients/mongo"
    "goa.design/goa-ai/runtime/agent/runtime"
)

promptClient, err := clientmongo.New(clientmongo.Options{
    Client:     mongoClient,
    Database:   "assistant",
    Collection: "prompt_overrides", // optional (default is prompt_overrides)
})
if err != nil {
    panic(err)
}

promptStore, err := promptmongo.NewStore(promptClient)
if err != nil {
    panic(err)
}

rt := runtime.New(
    runtimeStore,
    runtime.WithEngine(temporalEng),
    runtime.WithPromptStore(promptStore),
)
```

### Override Resolution and Rollout

Override precedence is deterministic:

1. `session` scope
2. `facility` scope
3. `org` scope
4. global scope
5. baseline spec (when no override exists)

Recommended rollout strategy:

- Register new baseline specs first.
- Roll out overrides at broad scope (`org`), then narrow to `facility`/`session` for canaries.
- Track effective versions through `prompt_rendered` events and `model.Request.PromptRefs`.
- Roll back by writing a newer override at the same scope (or removing scope-specific overrides to fall back).

---

## Temporal Setup

This section covers setting up Temporal for durable agent workflows in production environments.

### Overview

Temporal provides durable execution for your Goa-AI agents. Agent runs become Temporal workflows with event-sourced history. Tool calls become activities with configurable retries. Every state transition is persisted. A restarted worker replays history and resumes exactly where it left off.

### How Durability Works

| Component | Role | Durability |
|-----------|------|------------|
| **Workflow** | Agent run orchestration | Event-sourced; survives restarts |
| **Plan Activity** | LLM inference call | Retries on transient failures |
| **Execute Tool Activity** | Tool invocation | Per-tool retry policies |
| **State** | Turn history, tool results | Persisted in workflow history |

**Concrete example:** Your agent calls an LLM, which returns 3 tool calls. Two tools complete. The third tool's service crashes.

- ❌ **Without Temporal:** The entire run fails. You re-run inference ($$$) and re-execute the two successful tools.
- ✅ **With Temporal:** Only the crashed tool retries. The workflow replays from history—no new LLM call, no re-running completed tools. Cost: one retry, not a full restart.

### What Survives Failures

| Failure Scenario | Without Temporal | With Temporal |
|------------------|------------------|---------------|
| Worker process crashes | Run lost, restart from zero | Replays from history, continues |
| Tool call times out | Run fails (or manual handling) | Automatic retry with backoff |
| Rate limit (429) | Run fails | Backs off, retries automatically |
| Network partition | Partial progress lost | Resumes after reconnect |
| Deploy during run | In-flight runs fail | Existing workflows stay on retained compatible workers; new workflows use the promoted version |

### Installation

**Option 1: Docker (Development)**

One-liner for local development:
```bash
docker run --rm -d --name temporal-dev -p 7233:7233 temporalio/auto-setup:latest
```

**Option 2: Temporalite (Development)**

```bash
go install go.temporal.io/server/cmd/temporalite@latest
temporalite start
```

**Option 3: Temporal Cloud (Production)**

Sign up at [temporal.io](https://temporal.io) and configure your client with cloud credentials.

**Option 4: Self-Hosted (Production)**

Deploy Temporal using Docker Compose or Kubernetes. See the [Temporal documentation](https://docs.temporal.io) for deployment guides.

### Runtime Configuration

Goa-AI abstracts the execution backend behind the `Engine` interface. Swap engines without changing agent code:

**In-Memory Engine** (development):
```go
// Default: no external dependencies
rt := runtime.New(storageinmem.New())
```

**Temporal Engine** (production):
```go
import (
    runtimeTemporal "goa.design/goa-ai/runtime/agent/engine/temporal"
    temporalclient "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
    "go.temporal.io/sdk/workflow"
)

const releaseBuildID = "git-sha-or-image-digest"

temporalEng, err := runtimeTemporal.NewWorker(runtimeTemporal.Options{
    ClientOptions: &temporalclient.Options{
        HostPort:  "127.0.0.1:7233",
        Namespace: "default",
    },
    WorkerOptions: runtimeTemporal.WorkerOptions{
        TaskQueue: "orchestrator.chat",
        Options: worker.Options{
            DeploymentOptions: worker.DeploymentOptions{
                UseVersioning: true,
                Version: worker.WorkerDeploymentVersion{
                    DeploymentName: "assistant",
                    BuildID:        releaseBuildID,
                },
                DefaultVersioningBehavior: workflow.VersioningBehaviorPinned,
            },
        },
    },
})
if err != nil {
    panic(err)
}
defer temporalEng.Close()

rt := runtime.New(runtimeStore, runtime.WithEngine(temporalEng))
```

### Runtime Storage Ownership

`runtime.New` requires one `storage.Store`. In production, one host service must
own the database that contains session state, run metadata, continuation
checkpoints, and run records that cannot change after insertion. Agent workers
call that owner through a typed API; they do not open separate connections to
the owner's collections.
Product data remains with the product service. For example, a chat service keeps
its transcript, ratings, and search fields even when another service owns the
Goa-AI runtime store.

[Memory & Sessions](../memory-sessions/#store-lifecycle-changes-and-records-together)
defines the complete storage contract: lifecycle changes and records are stored
together, exact retries return the accepted result, new children require a
running parent, stored event JSON is strict, and cancellation provenance is
preserved. [Runtime](../runtime/#ensuring-a-final-record-and-its-delivery)
defines how hosts validate and deliver final run events after engine history
closes. Keep these rules in the owning runtime store rather than reproducing
them in each worker.

The move from `session.Store` plus `runlog.Store` is a coordinated storage
change. Before the new runtime writes, existing run metadata, checkpoints, and
records must satisfy the integrated `storage.Store` contract. Deploy the
storage owner and every worker that calls it together. Old split-store writers
and new integrated-store writers must not overlap.

Perform this conversion offline with a disposable migration program run from a
temporary job or database pod. The host owns that program because it owns the
database schema and deployment environment; it is not part of the normal
runtime release and should be deleted after the cutover is verified.

1. Back up the runtime database and verify that no old or new runtime writer is
   active.
2. Run the migration in verification mode and correct every rejected record.
3. Apply the conversion, then verify the schema, indexes, session state, run
   metadata, current-format checkpoints, and immutable run records.
4. Deploy the storage owner and all workers together, then remove the temporary
   migration program.

This physical storage conversion does not translate suspension formats. Saved
work must separately satisfy the [current continuation contract](../runtime/#external-input-and-workflow-continuations).
Keep completed history and stored tool results intact.

Once the conversion starts, rollback is a database restore, not a mixed-version
deployment. If conversion or verification fails, keep runtime traffic closed
and restore the complete pre-conversion backup before running old writers.

Do not set `ClientOptions.DataConverter`. The Temporal engine rejects a custom
converter and installs Goa-AI's bounded converter itself so every worker and
client uses the same persisted contract.

### Temporal Payload Contract

Every workflow or activity argument list has an aggregate encoded limit of
`engine.MaxPayloadBytes` (1 MiB). Before encoding, the converter also rejects a
value graph deeper than 64 levels or larger than 100,000 visited values. It
does not truncate oversized data.

`planner.ToolResult` is an in-process value and cannot cross a Temporal
boundary. Workflows carry `api.ToolEvent` with canonical JSON bytes instead.
When a valid tool result can exceed 1 MiB, the tool executor must save it in
application-owned storage and return a typed reference; the runtime does not
silently replace the result.

### Timing and Activity Retries

Use the DSL for semantic run budgets: how long the whole run may take, how long a
planner attempt may run, and how long a tool attempt may run.

Goa-AI starts each agent workflow once. It does not restart the whole workflow
after failure because an earlier attempt may already have called tools or saved
a final lifecycle record. Durability comes from workflow-history replay and
retries of individual planner, tool, hook, and storage activities.

```go
Agent("operator", "Production operations agent", func() {
    RunPolicy(func() {
        DefaultCaps(MaxToolCalls(20), MaxRecoveryTurns(3))
        Timing(func() {
            Budget("5m")
            Plan("45s")
            Tools("90s")
        })
    })
})
```

The Temporal adapter owns workflow-engine mechanics such as queue-wait and
liveness timeouts. Configure those on the engine, not in the DSL:

```go
temporalEng, err := runtimeTemporal.NewWorker(runtimeTemporal.Options{
    ClientOptions: &client.Options{
        HostPort:  "127.0.0.1:7233",
        Namespace: "default",
    },
    WorkerOptions: runtimeTemporal.WorkerOptions{
        TaskQueue: "orchestrator.chat",
    },
    ActivityDefaults: runtimeTemporal.ActivityDefaults{
        Planner: runtimeTemporal.ActivityTimeoutDefaults{
            QueueWaitTimeout: 30 * time.Second,
            LivenessTimeout:  20 * time.Second,
        },
        Tool: runtimeTemporal.ActivityTimeoutDefaults{
            QueueWaitTimeout: 2 * time.Minute,
            LivenessTimeout:  20 * time.Second,
        },
    },
})
```

Generated plan/resume, execute-tool, and hook-publishing activities use retry
policies that are safe only when retries are logically idempotent. Hook events
carry stable event keys, and tool executions should persist or replay canonical
results by `ToolCallID` rather than repeating irreversible side effects.

### Worker Setup

Workers poll task queues and execute workflows/activities. Workers are automatically started for each registered agent—no manual worker configuration needed in most cases.

### Transparent Rollouts

Temporal durability and transparent releases are separate guarantees. Temporal
stores workflow history. Your deployment must keep compatible worker code and
every required downstream service available while that history is still in
use.

The configuration above opts the worker into [Temporal Worker Deployment
Versioning](https://docs.temporal.io/production-deployment/worker-deployments/worker-versioning).
`releaseBuildID` must identify one immutable binary or container image. Never
reuse a build ID for different workflow code or use a mutable tag such as
`latest`.

Release a worker version in this order:

1. Start the new workers beside all retained worker versions.
2. Wait for the new process to pass readiness and register successfully with
   Temporal.
3. Make the new Worker Deployment Version current. Temporal assigns new
   workflows to it while existing workflows stay pinned to the version that
   started them.
4. If the worker process also serves an API, route normal API traffic only to
   the current ready build. Keep old pods alive for Temporal without sending
   new API requests to them. A separate API deployment is another valid design,
   but it is not required.
5. Remove an old version only after Temporal reports it drained. A stopped pod
   is not proof that no workflow still needs that code.

Each accepted user input starts one top-level Goa-AI workflow. Goa-AI ends that
workflow when it requests human or external input and stores a private
checkpoint under the completed run ID. The accepted answer starts a new
workflow on the current worker version. The new version must therefore accept
the saved checkpoint format, generated result codecs, and required tool names.
The current runtime accepts only `goa-ai.run-suspension.v8`; earlier checkpoint
versions are rejected. Follow [Generated contract changes](#generated-contract-changes)
for an incompatible upgrade rather than this overlapping-worker procedure.
Worker versioning cannot translate incompatible stored values.

The rest of the application must preserve availability during the same
overlap:

- A downstream Service must always have at least one ready endpoint. Use a
  readiness-gated rolling replacement; a `Recreate` rollout introduces a gap.
- Downstream APIs must accept calls from retained and current workers.
- Database migrations normally must support both releases until the old version
  is drained. Use an expand-then-contract sequence rather than replacing a
  schema before old code stops using it. The unified runtime-store migration
  described above is deliberately different: it requires one coordinated
  cutover and does not allow old and new writers to overlap.
- If a process serves both API traffic and Temporal work, the traffic selector
  must identify the current build independently from Temporal's access to
  retained workers.

#### Registry-backed tool providers

Goa-AI tool providers also support readiness-gated rolling replacement. Provider
replicas with the same generated schema and admission revision join the same
registry admission and may overlap. When either value changes, the replacement
provider stays alive and retries registration while the old admission remains
authoritative. The old provider stops claiming calls, settles accepted work,
and releases its lease before the new admission can execute, so two different
tool contracts never serve the same toolset at once.

A valid `CallTool` request that finds an active toolset with no healthy provider
waits within its existing execution deadline. Request publication then verifies
the selected provider in the same Redis operation that appends the call. If the
old provider started draining after the health check, the unpublished call
selects the replacement and tries again without extending its deadline. The
provider assignment becomes permanent only when publication succeeds. Caller
cancellation ends only that transport attempt; an exact retry can continue the
unpublished call. Deadline expiry records the normal durable
`call_not_admitted` decision.

The registry owns provider health, not deployment intent. It cannot distinguish
a rollout handoff from another provider outage, so the same bounded wait applies
to both. It does not inspect pod names or version strings and it does not ask the
model to retry. This contract lets consumers use one rolling release policy for
provider changes without allowing incompatible provider generations to overlap.

Registry clients, servers, and providers must still use a compatible wire
protocol during the release. If that envelope must change incompatibly, first
release code that accepts both forms; the registry does not negotiate protocol
versions during a rolling overlap.

Worker Deployment Versioning protects workflow replay. It does not protect a
workflow from an unavailable dependency, an incompatible API, or an
incompatible checkpoint.

#### Generated contract changes

When generated agents, completion packages, or persisted runtime payloads
change incompatibly, do not apply the mixed-version procedure above. Regenerate
all agents and completions, drain or stop affected work, and deploy the runtime,
workers, and callers as one coordinated release. Goa-AI does not provide a
dual-read mode for generated runtime contracts.

The runtime accepts only the exact `goa-ai.run-suspension.v8` schema. Planners
that wait for questions, clarification, or external tools preserve the
provider's `ModelToolCallID`; the workflow assigns the separate runtime
`ToolCallID` before it saves the suspension. Other suspension schemas do not
resume. Version eight also retains the advertised choices when an accepted
recovery plan waits for input. Follow the host-owned preservation and
resumability decision in [External Input and Workflow Continuations](../runtime/#external-input-and-workflow-continuations)
before replacing workers that own older saved work; do not add a dual reader or
infer fields.

Update cooperating workers for every workflow and activity queue, generated
packages, and callers before accepting new work. Once new workers save v8
suspensions, older workers cannot resume those suspensions: reverting an image
alone does not restore resumability. Preserve completed history and stored tool
results; any physical storage conversion is a separate host-owned operation.

#### Release verification

Do not call the release transparent until all of these checks pass:

- A workflow started before promotion completes on its original build.
- A new workflow starts and completes on the current build.
- An external-input request created before promotion continues successfully as
  a new workflow after promotion.
- API traffic reaches only the current ready build.
- Old workers stay ready until Temporal reports them drained.
- Every downstream Service retains a ready endpoint throughout replacement.
- The observation window contains no new workflow failures, container
  restarts, or readiness gaps.

The one-workflow-per-turn and cross-workflow event identity contract is covered
in [External Input and Workflow Continuations](../runtime/#external-input-and-workflow-continuations).

### Best Practices

- **Use separate namespaces** for different environments (dev, staging, prod)
- **Configure retry policies** per toolset based on reliability characteristics
- **Monitor workflow execution** using Temporal's UI and observability tools
- **Set appropriate timeouts** for activities—balance reliability vs. hung detection
- **Use Temporal Cloud** for production to avoid operational burden

---

## Streaming UI

This section shows how to stream agent events to UIs in real-time using Goa-AI's streaming infrastructure.

### Overview

Goa-AI publishes **session-owned streams** of typed events that can be delivered to UIs via:
- Server-Sent Events (SSE)
- WebSockets
- Message buses (Pulse, Redis Streams, etc.)

All stream-visible events for a session are appended to a single stream: `session/<session_id>`. Each event carries both `run_id` and `session_id` so UIs can group events into per-run lanes/cards. Nested agent runs are linked via `child_run_linked` events. UIs close SSE/WebSocket deterministically when they observe `run_stream_end` for the active run.

### Stream Sink Interface

Implement the `stream.Sink` interface:

```go
type Sink interface {
    Send(ctx context.Context, event stream.Event) error
    Close(ctx context.Context) error
}
```

### Event Types

The `stream` package defines concrete event types that implement `stream.Event`. Common ones for UIs are:

| Event Type | Description |
|------------|-------------|
| `AssistantReply` | Assistant message chunks (streaming text) |
| `PlannerThought` | Planner thinking blocks (notes and structured reasoning) |
| `ToolStart` | Tool execution started |
| `ToolUpdate` | Tool execution progress (expected child count updates) |
| `ToolEnd` | Tool execution completed (result, error, telemetry) |
| `AwaitClarification` | Planner is waiting for human clarification |
| `AwaitExternalTools` | Planner is waiting for external tool results |
| `Usage` | Token usage per model invocation |
| `Workflow` | Run lifecycle and phase updates |
| `ChildRunLinked` | Link from a parent tool call to a child agent run |
| `RunStreamEnd` | Explicit stream boundary marker for a run (no more stream-visible events will appear for that run) |

Transports typically type-switch on `stream.Event` for compile-time safety:

```go
switch e := evt.(type) {
case stream.AssistantReply:
    // e.Data.Text
case stream.PlannerThought:
    // e.Data.Note or structured thinking fields
case stream.ToolStart:
    // e.Data.ToolCallID, e.Data.ToolName, e.Data.Payload
case stream.ToolEnd:
    // e.Data.Result, e.Data.Error, e.Data.ResultPreview
case stream.ChildRunLinked:
    // e.Data.ToolName, e.Data.ToolCallID, e.Data.ChildRunID, e.Data.ChildAgentID
case stream.RunStreamEnd:
    // run has no more stream-visible events
}
```

### Example: SSE Sink

```go
type SSESink struct {
    w http.ResponseWriter
}

func (s *SSESink) Send(ctx context.Context, event stream.Event) error {
    switch e := event.(type) {
    case stream.AssistantReply:
        fmt.Fprintf(s.w, "data: assistant: %s\n\n", e.Data.Text)
    case stream.PlannerThought:
        if e.Data.Note != "" {
            fmt.Fprintf(s.w, "data: thinking: %s\n\n", e.Data.Note)
        }
    case stream.ToolStart:
        fmt.Fprintf(s.w, "data: tool_start: %s\n\n", e.Data.ToolName)
    case stream.ToolEnd:
        fmt.Fprintf(s.w, "data: tool_end: %s status=%v\n\n",
            e.Data.ToolName, e.Data.Error == nil)
    case stream.ChildRunLinked:
        fmt.Fprintf(s.w, "data: child_run_linked: %s child=%s\n\n",
            e.Data.ToolName, e.Data.ChildRunID)
    case stream.RunStreamEnd:
        fmt.Fprintf(s.w, "data: run_stream_end: %s\n\n", e.RunID())
    }
    s.w.(http.Flusher).Flush()
    return nil
}

func (s *SSESink) Close(ctx context.Context) error {
    return nil
}
```

### Session Stream Subscription (Pulse)

In production, UIs consume the session stream (`session/<session_id>`) from a shared bus (Pulse / Redis Streams) and filter by `run_id`. Close SSE/WebSocket when you observe `run_stream_end` for the active run.

### Global Stream Sink

To stream all runs through a global sink (for example, Pulse), configure the runtime with a stream sink:

```go
rt := runtime.New(
    runtimeStore,
    runtime.WithStream(pulseSink), // or your custom sink
)
```

The runtime installs a default `stream.Subscriber` that:
- maps hook events to `stream.Event` values
- uses the **default `StreamProfile`**, which emits assistant replies, planner thoughts, tool start/update/end, awaits, usage, workflow, `child_run_linked` links, and the terminal `run_stream_end` marker

### Stream Profiles

Not every consumer needs every event. **Stream profiles** filter events for different audiences, reducing noise and bandwidth for specific use cases.

| Profile | Use Case | Included Events |
|---------|----------|-----------------|
| `UserChatProfile()` | End-user chat UI | Assistant replies, tool start/end, workflow completion |
| `AgentDebugProfile()` | Developer debugging | Everything including planner thoughts |
| `MetricsProfile()` | Observability pipelines | Usage and workflow events only |

**Using built-in profiles:**

```go
// User-facing chat: replies, tool status, completion
profile := stream.UserChatProfile()

// Debug view: everything including planner thoughts
profile := stream.AgentDebugProfile()

// Metrics pipeline: just usage and workflow events
profile := stream.MetricsProfile()

sub, _ := stream.NewSubscriberWithProfile(sink, profile)
```

**Custom profiles:**

```go
// Fine-grained control over which events to emit
profile := stream.StreamProfile{
    Assistant:  true,
    Thoughts:   false,  // Skip planner thinking
    ToolStart:  true,
    ToolUpdate: true,
    ToolEnd:    true,
    Usage:      false,  // Skip usage events
    Workflow:   true,
    ChildRuns:  true,   // Include parent tool → child run links
}

sub, _ := stream.NewSubscriberWithProfile(sink, profile)
```

Custom profiles are useful when:
- You need specific events for a specialized consumer (e.g., progress tracking)
- You want to reduce payload size for mobile clients
- You're building analytics pipelines that only need certain events

### Advanced: Pulse & Stream Bridges

For production setups, you often want to:
- publish events to a shared bus (e.g., Pulse)
- use a **session-owned stream** on that bus (`session/<session_id>`)

Goa-AI provides:
- `features/stream/pulse` – a `stream.Sink` implementation backed by Pulse
- `runtime/agent/stream/bridge` – helpers to wire the hook bus to any sink

Typical wiring:

```go
pulseClient := pulse.NewClient(redisClient)
s, err := pulseSink.NewSink(pulseSink.Options{
    Client: pulseClient,
    // Optional: override stream naming (defaults to `session/<SessionID>`).
    StreamID: func(ev stream.Event) (string, error) {
        if ev.SessionID() == "" {
            return "", errors.New("missing session id")
        }
        return fmt.Sprintf("session/%s", ev.SessionID()), nil
    },
})
if err != nil { log.Fatal(err) }

rt := runtime.New(
    runtimeStore,
    runtime.WithEngine(eng),
    runtime.WithStream(s),
)
```

---

## System Reminders

Models drift. They forget instructions. They ignore context that was clear 10 turns ago. When your agent executes long-running tasks, you need a way to inject *dynamic, contextual guidance* without polluting the user conversation.

### The Problem

**Scenario:** Your agent manages a todo list. After 20 turns, the user asks "what's next?" but the model has drifted—it doesn't remember there's a pending todo in progress. You need to nudge it *without* the user seeing an awkward "REMINDER: you have a todo in progress" message.

**Without system reminders:**
- You bloat the system prompt with every possible scenario
- Guidance gets lost in long conversations
- No way to inject context based on tool results
- Users see internal agent scaffolding

**With system reminders:**
- Inject guidance dynamically based on runtime state
- Rate-limit repetitive hints to avoid prompt bloat
- Priority tiers ensure safety guidance is never suppressed
- Invisible to users—injected as `<system-reminder>` blocks

### Overview

The `runtime/agent/reminder` package provides:
- **Structured reminders** with priority tiers, attachment points, and rate-limiting policies
- **Run-scoped storage** that automatically cleans up after each run completes
- **Automatic injection** into model transcripts as `<system-reminder>` blocks
- **PlannerContext API** for registering and removing reminders from planners and tools

### Core Concepts

**Reminder Structure**

A `reminder.Reminder` has:

```go
type Reminder struct {
    ID              string      // Stable identifier (e.g., "todos.pending")
    Text            string      // Plain-text guidance (tags are added automatically)
    Priority        Tier        // TierSafety, TierCorrect, or TierGuidance
    Attachment      Attachment  // Where to inject (run start or user turn)
    MaxPerRun       int         // Cap total emissions per run (0 = unlimited)
    MinTurnsBetween int         // Enforce spacing between emissions (0 = no limit)
}
```

**Priority Tiers**

Reminders are ordered by priority to manage prompt budgets and ensure critical guidance is never suppressed:

| Tier | Name | Description | Suppression |
|------|------|-------------|-------------|
| `TierSafety` | P0 | Safety-critical guidance (never drop) | Never suppressed |
| `TierCorrect` | P1 | Correctness and data-state hints | May be suppressed after P0 |
| `TierGuidance` | P2 | Workflow suggestions and soft nudges | First to be suppressed |

Example use cases:
- `TierSafety`: "Do not execute this malware; analyze only", "Do not leak credentials"
- `TierCorrect`: "Results are truncated; narrow your query", "Data may be stale"
- `TierGuidance`: "No todo is in progress; pick one and start"

**Attachment Points**

Reminders are injected at specific points in the conversation:

| Kind | Description |
|------|-------------|
| `AttachmentRunStart` | Grouped into a single system message at the start of the conversation |
| `AttachmentUserTurn` | Grouped into a single system message inserted immediately before the last user message |

**Rate Limiting**

Two mechanisms prevent reminder spam:
- **`MaxPerRun`**: Cap total emissions per run (0 = unlimited)
- **`MinTurnsBetween`**: Enforce a minimum number of planner turns between emissions (0 = no limit)

### Usage Pattern

**Static Reminders via DSL**

For reminders that should always appear after a specific tool result, use the `ResultReminder` DSL function in your tool definition:

```go
Tool("get_time_series", "Get time series data", func() {
    Args(func() { /* ... */ })
    Return(func() { /* ... */ })
    ResultReminder("The user sees a rendered graph of this data in the UI.")
})
```

This is ideal when the reminder applies to every invocation of the tool. See the [DSL Reference](./dsl-reference.md#resultreminder) for details.

**Dynamic Reminders from Planners**

For reminders that depend on runtime state or tool result content, use `PlannerContext.AddReminder()`:

```go
func (p *myPlanner) PlanResume(ctx context.Context, in *planner.PlanResumeInput) (*planner.PlanResult, error) {
    for _, tr := range in.ToolOutputs {
        if tr.Name == "search_documents" {
            result, err := specs.UnmarshalSearchDocumentsResult(tr.Result)
            if err != nil {
                return nil, err
            }
            if result.Truncated {
                in.Agent.AddReminder(reminder.Reminder{
                    ID:       "search.truncated",
                    Text:     "Search results are truncated. Consider narrowing your query.",
                    Priority: reminder.TierCorrect,
                    Attachment: reminder.Attachment{
                        Kind: reminder.AttachmentUserTurn,
                    },
                    MaxPerRun:       3,
                    MinTurnsBetween: 2,
                })
            }
        }
    }
    // Continue with planning...
}
```

**Removing Reminders**

Use `RemoveReminder()` when a precondition no longer holds:

```go
if allTodosCompleted {
    in.Agent.RemoveReminder("todos.no_active")
}
```

**Preserving Rate-Limit Counters**

`AddReminder()` preserves emission counters when updating an existing reminder by ID. If you need to change reminder content but maintain rate limits:

```go
in.Agent.AddReminder(reminder.Reminder{
    ID:              "todos.pending",
    Text:            buildUpdatedText(snap),
    Priority:        reminder.TierGuidance,
    Attachment:      reminder.Attachment{Kind: reminder.AttachmentUserTurn},
    MinTurnsBetween: 3,
})
```

**Anti-pattern**: Don't call `RemoveReminder()` followed by `AddReminder()` for the same ID—this resets counters and bypasses `MinTurnsBetween`.

### Injection and Formatting

**Automatic Tagging**

The runtime automatically wraps reminder text in `<system-reminder>` tags when injecting into transcripts:

```go
// You provide plain text:
Text: "Results are truncated. Narrow your query."

// Runtime injects:
<system-reminder>Results are truncated. Narrow your query.</system-reminder>
```

**Explaining Reminders to Models**

Include `reminder.DefaultExplanation` in your system prompt so models know how to interpret `<system-reminder>` blocks:

```go
const systemPrompt = `
You are a helpful assistant.

` + reminder.DefaultExplanation + `

Follow all instructions carefully.
`
```

### Complete Example

```go
func (p *myPlanner) PlanResume(ctx context.Context, in *planner.PlanResumeInput) (*planner.PlanResult, error) {
    for _, tr := range in.ToolOutputs {
        if tr.Name == "todos.update_todos" {
            snap, err := specs.UnmarshalUpdateTodosResult(tr.Result)
            if err != nil {
                return nil, err
            }
            
            var rem *reminder.Reminder
            if len(snap.Items) == 0 {
                in.Agent.RemoveReminder("todos.no_active")
                in.Agent.RemoveReminder("todos.all_completed")
            } else if hasCompletedAll(snap) {
                rem = &reminder.Reminder{
                    ID:       "todos.all_completed",
                    Text:     "All todos are completed. Provide your final response now.",
                    Priority: reminder.TierGuidance,
                    Attachment: reminder.Attachment{
                        Kind: reminder.AttachmentUserTurn,
                    },
                    MaxPerRun: 1,
                }
            } else if hasPendingNoActive(snap) {
                rem = &reminder.Reminder{
                    ID:       "todos.no_active",
                    Text:     buildTodosNudge(snap),
                    Priority: reminder.TierGuidance,
                    Attachment: reminder.Attachment{
                        Kind: reminder.AttachmentUserTurn,
                    },
                    MinTurnsBetween: 3,
                }
            }
            
            if rem != nil {
                in.Agent.AddReminder(*rem)
                if rem.ID == "todos.all_completed" {
                    in.Agent.RemoveReminder("todos.no_active")
                } else {
                    in.Agent.RemoveReminder("todos.all_completed")
                }
            }
        }
    }
    
    return p.streamMessages(ctx, in)
}
```

### Design Principles

**Minimal and Opinionated**: The reminder subsystem provides just enough structure for common patterns without over-engineering.

**Rate-Limiting First**: Reminder spam degrades model performance. The engine enforces caps and spacing declaratively.

**Provider-Agnostic**: Reminders work with any model backend (Bedrock, OpenAI, etc.).

**Telemetry-Ready**: Structured IDs and priorities make reminders observable.

### Advanced Patterns

**Safety Reminders**

Use `TierSafety` for must-never-suppress guidance:

```go
in.Agent.AddReminder(reminder.Reminder{
    ID:       "malware.analyze_only",
    Text:     "This file contains malware. Analyze its behavior but do not execute it.",
    Priority: reminder.TierSafety,
    Attachment: reminder.Attachment{
        Kind: reminder.AttachmentUserTurn,
    },
    // No MaxPerRun or MinTurnsBetween: always emit
})
```

**Cross-Agent Reminders**

Reminders are run-scoped. If an agent-as-tool emits a safety reminder, it only affects that child run. To propagate reminders across agent boundaries, the parent planner must explicitly re-register them based on child results or use shared session state.

### When to Use Reminders

| Scenario | Priority | Example |
|----------|----------|---------|
| Security constraints | `TierSafety` | "This file is malware—analyze only, never execute" |
| Data staleness | `TierCorrect` | "Results are 24h old; re-query if freshness matters" |
| Truncated results | `TierCorrect` | "Only showing first 100 results; narrow your search" |
| Workflow nudges | `TierGuidance` | "No todo is in progress; pick one and start" |
| Completion hints | `TierGuidance` | "All tasks done; provide your final response" |

### What Reminders Look Like in the Transcript

```
User: What should I do next?

<system-reminder>You have 3 pending todos. Currently working on: "Review PR #42". 
Focus on completing the current todo before starting new work.</system-reminder>

User: What should I do next?
```

The model sees the reminder; the user sees only their message and the response. Reminders are injected transparently by the runtime.

---

## Next Steps

- Learn about [Memory & Sessions](./memory-sessions/) for transcript persistence
- Explore [Agent Composition](./agent-composition/) for agent-as-tool patterns
- Read about [Toolsets](./toolsets/) for tool execution models