# Stage 4a relay (STUN and TURN) — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** A platform-run STUN and TURN relay on every armed region box, enabled per app through `carlos relay enable`, with keys delivered as sealed config, so the Oficina Meet app connects calls across the internet with no third party in the path.

**Architecture:** One new package, `internal/relay`, holds everything pure: key derivation, the credential contract, the authority record, the pion server wrapper with its own socket wrapping for termination and metering, and the limits. The console gains a relay record per app, a relay-writer lease, an aggregate publisher and a reconcile, plus four CLI doors. The host gains a `carlos-relay.service` unit, a `carlos relay serve` verb, and a per-route config acknowledgement written by the config sync. Infra gains an `extra_ingress` block and a DNS record per region.

**Tech Stack:** Go 1.26, `github.com/pion/turn/v4` v4.1.4 (new dependency, pure Go, cross-compiles clean; checked 2026-09-07), `modernc.org/sqlite` (existing), OpenTofu (infra).

**Spec:** `docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md` (revision 6, codex accept). Parent: `docs/superpowers/specs/2026-09-07-media-platform-roadmap-design.md`.

## Global Constraints

- **The app-facing contract (spec §2) is frozen.** `CARLOS_RELAY` is standard padded base64 of a JSON array of `{region, urls, key, username_suffix}`; username is `<expiry>:<account>:<app>:<incarnation>:<kid>`; credential is `base64_std(HMAC-SHA256(key, username))`; max expiry four hours ahead. Test vector: master `SHA-256("carlos-relay-test-master")`, account `abc`, app `meet`, incarnation `0123456789abcdef`, kid `k1`, expiry `1900000000` → key `uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=`, credential `jRU479Y3GO45nrRHqxBf3BMeBHE6Fip1cwAccNv28MQ=`. Any change is a spec change.
- **Rulings (spec §9) are fixed:** four-hour credentials; member-level enable; console holds masters as sealed config; no TLS listener; two meters (`in`, `out`); limits as in spec §3.
- **Wire shapes are additive.** Every new bucket object is a new key; `relay.json`, `control/relay.json`, `control/relay-issuer.json`, `control/relay-writer.json`, `control/relay-ack/<box>.json`, `control/config-ack/<box>/<host>.json`, `metrics/relay/<box>/<hour>.json`. Nothing existing is retyped.
- **CAS is `Put(ctx, key, body, ifMatch)`** on `release.Store`; a mismatch returns `release.ErrCASConflict`. There is no other conditional primitive. (`internal/release/store.go:42`.)
- **Config values must pass `appconfig.ValidVarValue`**: no quotes, backslashes, newlines or NUL. That is why `CARLOS_RELAY` is base64.
- **`CGO_ENABLED=0` everywhere.** The CLI cross-compiles for darwin/arm64, darwin/amd64, linux/arm64, windows/amd64, windows/arm64 (`.amadan/ci.d/20-cross@5`). pion/turn is pure Go.
- **Every `carlos` command needs a `### carlos <verb>` heading in `docs/site/cli.md`** or `internal/docsite` fails the gate.
- **Build gate before any commit lands:** `go build ./...`, `go vet ./...`, `go test ./... -count=1`, `gofmt -l .` clean. Sandbox: `GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off`; adding the dependency needs the sandbox lifted once for `go get`.
- **The forge is `origin` (amadan.net).** Push there, `amadan branch describe`, wait for `amadan ci status -branch` to pass, `amadan branch merge … -expect <full sha>`. Never `gh`.
- **Delegation (AGENTS.md):** Tasks 1 to 5 are Sonnet from this brief; Task 6 (pion wrapper) and Task 10 (console operations with the lease) are Opus; final whole-branch review is Opus.

---

## File structure

| Path | Responsibility |
|---|---|
| `internal/relay/keys.go` | HKDF derivation, credential minting, username parse; the §2 contract |
| `internal/relay/masters.go` | masters file parse, `retire_at`, watched reload |
| `internal/relay/authority.go` | `control/relay.json` shape, read with lease clock, heartbeat write |
| `internal/relay/limits.go` | admission counters, rolling-hour budgets, token buckets |
| `internal/relay/conn.go` | wrapped `net.PacketConn` and listener conns: counting, buckets, EIP port gate, close |
| `internal/relay/server.go` | pion `ServerConfig` assembly: generator, auth handler, permission handler, bindings, termination timers |
| `internal/relay/meter.go` | per-app counters, minute samples, hourly object read-append-CAS |
| `internal/relay/serve.go` | `Serve(ctx, Config)`: wires the above, env parsing, PSI shedding loop |
| `internal/relay/record.go` | `apps/<acct>/<app>/relay.json` shape, Get/Put, transitions |
| `internal/relay/config.go` | build the `CARLOS_RELAY` value from a record and issuer masters |
| `internal/relay/vector_test.go` | the §2 test vector |
| `internal/host/units/carlos-relay.service` | the unit |
| `internal/host/bootstrap.go` | arm and start the unit |
| `internal/host/systemapply.go` | restart the relay on platform update |
| `internal/appconfig/sync.go`, `internal/appconfig/configack.go` | per-route config acknowledgement |
| `internal/console/apirelay.go` | doors: enable, status, disable, adopted, rotate |
| `internal/console/relayops.go` | writer lease, operations, aggregate publish, reconcile |
| `relay.go` (root) | `carlos relay` CLI: enable, status, disable, adopted, rotate, keygen, serve |
| `apimode.go` | `apiSession.relay*` client calls |
| `infra/modules/region-host/main.tf`, `variables.tf`, `outputs.tf` | `relay` variable, ingress, VPC CIDR output |
| `docs/site/cli.md` | `### carlos relay` |

---

### Task 1: Key derivation and the credential contract

**Files:**
- Create: `internal/relay/keys.go`
- Test: `internal/relay/vector_test.go`, `internal/relay/keys_test.go`

**Interfaces:**
- Produces:
  ```go
  const MaxCredentialLife = 4 * time.Hour
  func DeriveKey(master []byte, account, app, incarnation string) []byte      // 32 bytes
  func Suffix(account, app, incarnation, kid string) string                 // "acct:app:inc:kid"
  func Credential(key []byte, username string) string                      // base64 std
  type Username struct{ Expiry time.Time; Account, App, Incarnation, Kid string }
  func ParseUsername(s string) (Username, error)
  ```

- [ ] **Step 1: Write the vector test**

```go
package relay

import (
	"crypto/sha256"
	"encoding/base64"
	"testing"
)

func TestVector(t *testing.T) {
	m := sha256.Sum256([]byte("carlos-relay-test-master"))
	key := DeriveKey(m[:], "abc", "meet", "0123456789abcdef")
	if got := base64.StdEncoding.EncodeToString(key); got != "uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=" {
		t.Fatalf("key = %s", got)
	}
	u := "1900000000:" + Suffix("abc", "meet", "0123456789abcdef", "k1")
	if u != "1900000000:abc:meet:0123456789abcdef:k1" {
		t.Fatalf("username = %s", u)
	}
	if got := Credential(key, u); got != "jRU479Y3GO45nrRHqxBf3BMeBHE6Fip1cwAccNv28MQ=" {
		t.Fatalf("credential = %s", got)
	}
}
```

- [ ] **Step 2: Write the parse tests**

```go
func TestParseUsername(t *testing.T) {
	u, err := ParseUsername("1900000000:abc:meet:0123456789abcdef:k1")
	if err != nil || u.Account != "abc" || u.App != "meet" || u.Incarnation != "0123456789abcdef" || u.Kid != "k1" || u.Expiry.Unix() != 1900000000 {
		t.Fatalf("parse = %+v %v", u, err)
	}
	for _, bad := range []string{"", "1:2:3:4", "1:2:3:4:5:6", "x:abc:meet:0123456789abcdef:k1",
		"1900000000:ABC:meet:0123456789abcdef:k1", "1900000000:abc:9meet:0123456789abcdef:k1",
		"1900000000:abc:meet:short:k1", "1900000000:abc:meet:0123456789abcdef:K1"} {
		if _, err := ParseUsername(bad); err == nil {
			t.Errorf("ParseUsername(%q) accepted", bad)
		}
	}
}
```

- [ ] **Step 3: Run tests to verify they fail**

Run: `go test ./internal/relay/ -run 'TestVector|TestParseUsername' -v`
Expected: FAIL, package does not exist.

- [ ] **Step 4: Implement**

```go
// Package relay is the platform STUN/TURN relay: the app-facing credential
// contract (spec §2), key derivation (§4), and the server (§3).
package relay

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"time"

	"golang.org/x/crypto/hkdf"
	"io"
)

const (
	MaxCredentialLife = 4 * time.Hour
	hkdfSalt          = "carlos-relay-v1"
)

var (
	accountRe     = regexp.MustCompile(`^[a-z0-9]+$`)
	appRe         = regexp.MustCompile(`^[a-z][a-z0-9-]{0,31}$`)
	incarnationRe = regexp.MustCompile(`^[0-9a-f]{16}$`)
	kidRe         = regexp.MustCompile(`^[a-z0-9]{1,8}$`)
)

// DeriveKey is HKDF-SHA256(master, salt "carlos-relay-v1",
// info account||0x00||app||0x00||incarnation, 32).
func DeriveKey(master []byte, account, app, incarnation string) []byte {
	info := []byte(account + "\x00" + app + "\x00" + incarnation)
	r := hkdf.New(sha256.New, master, []byte(hkdfSalt), info)
	key := make([]byte, 32)
	if _, err := io.ReadFull(r, key); err != nil {
		panic("relay: hkdf: " + err.Error()) // cannot fail for 32 bytes
	}
	return key
}

func Suffix(account, app, incarnation, kid string) string {
	return account + ":" + app + ":" + incarnation + ":" + kid
}

// Credential is base64(HMAC-SHA256(key, username)), the TURN password.
func Credential(key []byte, username string) string {
	h := hmac.New(sha256.New, key)
	h.Write([]byte(username))
	return base64.StdEncoding.EncodeToString(h.Sum(nil))
}

type Username struct {
	Expiry                        time.Time
	Account, App, Incarnation, Kid string
}

// ParseUsername splits "<expiry>:<account>:<app>:<incarnation>:<kid>" and
// validates each field's alphabet. It does not check expiry against now;
// the auth handler does that with its own clock.
func ParseUsername(s string) (Username, error) {
	f := strings.Split(s, ":")
	if len(f) != 5 {
		return Username{}, fmt.Errorf("relay: username has %d fields, want 5", len(f))
	}
	exp, err := strconv.ParseInt(f[0], 10, 64)
	if err != nil || exp < 0 {
		return Username{}, fmt.Errorf("relay: bad expiry %q", f[0])
	}
	switch {
	case !accountRe.MatchString(f[1]):
		return Username{}, fmt.Errorf("relay: bad account %q", f[1])
	case !appRe.MatchString(f[2]):
		return Username{}, fmt.Errorf("relay: bad app %q", f[2])
	case !incarnationRe.MatchString(f[3]):
		return Username{}, fmt.Errorf("relay: bad incarnation %q", f[3])
	case !kidRe.MatchString(f[4]):
		return Username{}, fmt.Errorf("relay: bad kid %q", f[4])
	}
	return Username{Expiry: time.Unix(exp, 0).UTC(), Account: f[1], App: f[2], Incarnation: f[3], Kid: f[4]}, nil
}
```

`golang.org/x/crypto` is already in `go.mod`; `hkdf` is a subpackage of it.

- [ ] **Step 5: Run tests to verify they pass**

Run: `go test ./internal/relay/ -v`
Expected: PASS.

- [ ] **Step 6: Commit**

```bash
git add internal/relay/
git commit -m "relay: key derivation and the credential contract (spec §2, §4)"
```

---

### Task 2: The masters file

**Files:**
- Create: `internal/relay/masters.go`
- Test: `internal/relay/masters_test.go`

**Interfaces:**
- Produces:
  ```go
  type Master struct{ Kid string; Key []byte; RetireAt time.Time } // zero RetireAt = never
  type Masters struct{ /* unexported */ }
  func ParseMasters(r io.Reader) (*Masters, error)
  func (m *Masters) Lookup(kid string, now time.Time) (Master, bool) // false if absent or retired
  func (m *Masters) Kids() []string                                  // loaded kids, sorted
  func WatchMasters(ctx context.Context, path string, onChange func(*Masters, error))
  ```
- File format, one master per line: `<kid> <base64 key> <retire_at RFC3339 or "-">`.

- [ ] **Step 1: Write the tests**

```go
func TestParseMasters(t *testing.T) {
	in := "k1 " + b64(bytes.Repeat([]byte{1}, 32)) + " -\n" +
		"k0 " + b64(bytes.Repeat([]byte{2}, 32)) + " 2026-09-08T00:00:00Z\n" +
		"# comment\n\n"
	m, err := ParseMasters(strings.NewReader(in))
	if err != nil {
		t.Fatal(err)
	}
	if got := m.Kids(); !reflect.DeepEqual(got, []string{"k0", "k1"}) {
		t.Fatalf("kids = %v", got)
	}
	before := time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC)
	after := time.Date(2026, 9, 9, 0, 0, 0, 0, time.UTC)
	if _, ok := m.Lookup("k0", before); !ok {
		t.Fatal("k0 should be live before retire_at")
	}
	if _, ok := m.Lookup("k0", after); ok {
		t.Fatal("k0 should be retired after retire_at")
	}
	if _, ok := m.Lookup("k1", after); !ok {
		t.Fatal("k1 never retires")
	}
	if _, ok := m.Lookup("kx", before); ok {
		t.Fatal("unknown kid")
	}
}

func TestParseMastersRejectsBadLines(t *testing.T) {
	for _, bad := range []string{"k1 notbase64 -", "k1 " + b64([]byte{1}) + " -", "K1 " + b64(bytes.Repeat([]byte{1}, 32)) + " -", "k1 " + b64(bytes.Repeat([]byte{1}, 32)) + " yesterday"} {
		if _, err := ParseMasters(strings.NewReader(bad)); err == nil {
			t.Errorf("accepted %q", bad)
		}
	}
}

func b64(b []byte) string { return base64.StdEncoding.EncodeToString(b) }
```

- [ ] **Step 2: Run to verify failure**

Run: `go test ./internal/relay/ -run TestParseMasters -v`
Expected: FAIL, undefined.

- [ ] **Step 3: Implement**

```go
package relay

import (
	"bufio"
	"context"
	"encoding/base64"
	"fmt"
	"io"
	"os"
	"sort"
	"strings"
	"time"
)

type Master struct {
	Kid      string
	Key      []byte
	RetireAt time.Time // zero means never
}

type Masters struct{ byKid map[string]Master }

// ParseMasters reads "<kid> <base64 key> <retire_at|->" lines. Blank lines
// and "#" comments are skipped. Keys must decode to exactly 32 bytes.
func ParseMasters(r io.Reader) (*Masters, error) {
	m := &Masters{byKid: map[string]Master{}}
	sc := bufio.NewScanner(r)
	line := 0
	for sc.Scan() {
		line++
		s := strings.TrimSpace(sc.Text())
		if s == "" || strings.HasPrefix(s, "#") {
			continue
		}
		f := strings.Fields(s)
		if len(f) != 3 {
			return nil, fmt.Errorf("relay masters line %d: want 3 fields, got %d", line, len(f))
		}
		if !kidRe.MatchString(f[0]) {
			return nil, fmt.Errorf("relay masters line %d: bad kid %q", line, f[0])
		}
		key, err := base64.StdEncoding.DecodeString(f[1])
		if err != nil || len(key) != 32 {
			return nil, fmt.Errorf("relay masters line %d: key must be base64 of 32 bytes", line)
		}
		var retire time.Time
		if f[2] != "-" {
			retire, err = time.Parse(time.RFC3339, f[2])
			if err != nil {
				return nil, fmt.Errorf("relay masters line %d: bad retire_at %q", line, f[2])
			}
		}
		m.byKid[f[0]] = Master{Kid: f[0], Key: key, RetireAt: retire}
	}
	return m, sc.Err()
}

func (m *Masters) Lookup(kid string, now time.Time) (Master, bool) {
	mk, ok := m.byKid[kid]
	if !ok || (!mk.RetireAt.IsZero() && !now.Before(mk.RetireAt)) {
		return Master{}, false
	}
	return mk, true
}

func (m *Masters) Kids() []string {
	out := make([]string, 0, len(m.byKid))
	for k := range m.byKid {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

// WatchMasters re-reads path whenever its mtime changes, polling every
// five seconds (fsnotify would be a dependency for one file). onChange is
// called with the parse result or error each time the content changes,
// and once at start.
func WatchMasters(ctx context.Context, path string, onChange func(*Masters, error)) {
	var last time.Time
	tick := time.NewTicker(5 * time.Second)
	defer tick.Stop()
	load := func() {
		st, err := os.Stat(path)
		if err != nil {
			onChange(nil, err)
			return
		}
		if st.ModTime().Equal(last) {
			return
		}
		last = st.ModTime()
		f, err := os.Open(path)
		if err != nil {
			onChange(nil, err)
			return
		}
		defer f.Close()
		onChange(ParseMasters(f))
	}
	load()
	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
			load()
		}
	}
}
```

- [ ] **Step 4: Run tests, then commit**

Run: `go test ./internal/relay/ -v`
Expected: PASS.

```bash
git add internal/relay/masters.go internal/relay/masters_test.go
git commit -m "relay: masters file with retire_at and hot reload (spec §4)"
```

---

### Task 3: Authority record, lease clock, heartbeat

**Files:**
- Create: `internal/relay/authority.go`
- Test: `internal/relay/authority_test.go`

**Interfaces:**
- Produces:
  ```go
  const AuthorityKey = "control/relay.json"
  const LeaseMaxAge = 15 * time.Minute
  type AuthorityApp struct{ Account, App, Incarnation string; Regions []string }
  type Authority struct{ Apps []AuthorityApp; WrittenAt time.Time }
  func (a *Authority) Lookup(account, app string) (AuthorityApp, bool)
  func ReadAuthority(ctx, s release.Store) (*Authority, string /*etag*/, error)  // os.ErrNotExist → nil,"",nil
  type Lease struct{ /* unexported */ }
  func (l *Lease) Renew(now time.Time)         // on a successful read
  func (l *Lease) Live(now time.Time) bool     // now - last <= LeaseMaxAge
  func HeartbeatKey(box string) string         // "control/relay-ack/<box>.json"
  type Heartbeat struct{ Box, Region string; WrittenAt time.Time /* of the record held */; ReadAt time.Time; Listeners []string; Kids []string; MastersError string }
  func WriteHeartbeat(ctx, s release.Store, hb Heartbeat) error   // unconditional Put
  ```

- [ ] **Step 1: Tests**

```go
func TestAuthorityRoundTrip(t *testing.T) {
	s := release.NewMemStore() // see internal/release for the in-memory Store used by console tests
	ctx := context.Background()
	a, _, err := ReadAuthority(ctx, s)
	if err != nil || a != nil {
		t.Fatalf("absent = %v %v, want nil,nil", a, err)
	}
	body, _ := json.Marshal(Authority{Apps: []AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}}}, WrittenAt: time.Unix(1_800_000_000, 0).UTC()})
	if _, err := s.Put(ctx, AuthorityKey, body, ""); err != nil {
		t.Fatal(err)
	}
	a, etag, err := ReadAuthority(ctx, s)
	if err != nil || etag == "" {
		t.Fatal(err)
	}
	if got, ok := a.Lookup("abc", "meet"); !ok || got.Incarnation != "0123456789abcdef" {
		t.Fatalf("lookup = %+v %v", got, ok)
	}
	if _, ok := a.Lookup("abc", "other"); ok {
		t.Fatal("unknown app found")
	}
}

func TestLease(t *testing.T) {
	var l Lease
	t0 := time.Unix(1_800_000_000, 0)
	if l.Live(t0) {
		t.Fatal("never renewed should not be live")
	}
	l.Renew(t0)
	if !l.Live(t0.Add(14 * time.Minute)) {
		t.Fatal("14m should be live")
	}
	if l.Live(t0.Add(16 * time.Minute)) {
		t.Fatal("16m should not be live")
	}
}
```

If `release.NewMemStore` does not exist under that name, use whatever in-memory `release.Store` the console tests use (`grep -rn "func New.*Store" internal/release/*.go`); the console fixture's `rel` field is one.

- [ ] **Step 2: Run to verify failure, then implement**

```go
package relay

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

const (
	AuthorityKey = "control/relay.json"
	LeaseMaxAge  = 15 * time.Minute
)

type AuthorityApp struct {
	Account     string   `json:"account"`
	App         string   `json:"app"`
	Incarnation string   `json:"incarnation"`
	Regions     []string `json:"regions"`
}

type Authority struct {
	Apps      []AuthorityApp `json:"apps"`
	WrittenAt time.Time      `json:"written_at"`
}

func (a *Authority) Lookup(account, app string) (AuthorityApp, bool) {
	if a == nil {
		return AuthorityApp{}, false
	}
	for _, x := range a.Apps {
		if x.Account == account && x.App == app {
			return x, true
		}
	}
	return AuthorityApp{}, false
}

func ReadAuthority(ctx context.Context, s release.Store) (*Authority, string, error) {
	body, etag, err := s.Get(ctx, AuthorityKey)
	if errors.Is(err, os.ErrNotExist) {
		return nil, "", nil
	}
	if err != nil {
		return nil, "", err
	}
	var a Authority
	if err := json.Unmarshal(body, &a); err != nil {
		return nil, etag, err
	}
	return &a, etag, nil
}

// Lease is the relay's serving lease: live while the last SUCCESSFUL
// authority read is younger than LeaseMaxAge. A failed read renews nothing.
type Lease struct{ last time.Time }

func (l *Lease) Renew(now time.Time)     { l.last = now }
func (l *Lease) Live(now time.Time) bool { return !l.last.IsZero() && now.Sub(l.last) <= LeaseMaxAge }

func HeartbeatKey(box string) string { return "control/relay-ack/" + box + ".json" }

type Heartbeat struct {
	Box          string    `json:"box"`
	Region       string    `json:"region"`
	WrittenAt    time.Time `json:"written_at"` // of the authority record held; zero if none
	ReadAt       time.Time `json:"read_at"`
	Listeners    []string  `json:"listeners"`
	Kids         []string  `json:"kids"`
	MastersError string    `json:"masters_error,omitempty"`
}

func WriteHeartbeat(ctx context.Context, s release.Store, hb Heartbeat) error {
	body, err := json.Marshal(hb)
	if err != nil {
		return err
	}
	_, err = s.Put(ctx, HeartbeatKey(hb.Box), body, "")
	return err
}
```

- [ ] **Step 3: Run tests, commit**

```bash
git add internal/relay/authority.go internal/relay/authority_test.go
git commit -m "relay: authority record, serving lease, heartbeat (spec §3, §5)"
```

---

### Task 4: Limits

**Files:**
- Create: `internal/relay/limits.go`
- Test: `internal/relay/limits_test.go`

**Interfaces:**
- Produces:
  ```go
  type Limits struct {
      AppLive, AppPerMinute int; AppBytesPerHour int64
      AccountLive, AccountPerMinute int; AccountBytesPerHour int64
      BoxLive, BoxPerMinute int
      AllocBitsPerSec, BoxBitsPerSec int64
  }
  func DefaultLimits() Limits  // 64,30,20<<30; 256,60,60<<30; 2048,500; 8_000_000, 200_000_000
  func LimitsFromEnv(getenv func(string) string) Limits // CARLOS_RELAY_LIMIT_* overrides, names below
  type Identity struct{ Account, App string }
  type Admission struct{ /* unexported; one mutex */ }
  func NewAdmission(l Limits, now func() time.Time) *Admission
  func (a *Admission) Admit(id Identity) (ok bool, reason string)  // counts an attempt even on refusal
  func (a *Admission) Release(id Identity)
  func (a *Admission) AddBytes(id Identity, n int64) (over bool)   // rolling hour; over=true when a budget is crossed
  func (a *Admission) Seed(id Identity, at time.Time, n int64)      // rebuild rolling window on start
  func (a *Admission) OverBudget(id Identity, now time.Time) bool
  type Bucket struct{ /* token bucket */ }
  func NewBucket(bitsPerSec int64, now func() time.Time) *Bucket
  func (b *Bucket) Allow(nbytes int) bool
  ```
- Env names: `CARLOS_RELAY_LIMIT_APP_LIVE`, `_APP_PER_MINUTE`, `_APP_BYTES_PER_HOUR`, `_ACCOUNT_LIVE`, `_ACCOUNT_PER_MINUTE`, `_ACCOUNT_BYTES_PER_HOUR`, `_BOX_LIVE`, `_BOX_PER_MINUTE`, `_ALLOC_BPS`, `_BOX_BPS`.

- [ ] **Step 1: Tests**

```go
func TestAdmissionCaps(t *testing.T) {
	now := time.Unix(1_800_000_000, 0)
	clock := func() time.Time { return now }
	l := DefaultLimits()
	l.AppLive, l.AppPerMinute, l.AccountLive, l.BoxLive = 2, 100, 3, 100
	a := NewAdmission(l, clock)
	meet := Identity{"abc", "meet"}
	docs := Identity{"abc", "docs"}
	for i := 0; i < 2; i++ {
		if ok, r := a.Admit(meet); !ok {
			t.Fatalf("admit %d refused: %s", i, r)
		}
	}
	if ok, r := a.Admit(meet); ok || r != "app live" {
		t.Fatalf("third meet = %v %q", ok, r)
	}
	if ok, _ := a.Admit(docs); !ok {
		t.Fatal("docs first should pass")
	}
	if ok, r := a.Admit(docs); ok || r != "account live" {
		t.Fatalf("account cap = %v %q", ok, r)
	}
	a.Release(meet)
	if ok, _ := a.Admit(docs); !ok {
		t.Fatal("after release, account has room")
	}
}

func TestAdmissionRate(t *testing.T) {
	now := time.Unix(1_800_000_000, 0)
	clock := func() time.Time { return now }
	l := DefaultLimits()
	l.AppPerMinute = 2
	a := NewAdmission(l, clock)
	id := Identity{"abc", "meet"}
	a.Admit(id); a.Release(id)
	a.Admit(id); a.Release(id)
	if ok, r := a.Admit(id); ok || r != "app rate" {
		t.Fatalf("rate = %v %q", ok, r)
	}
	now = now.Add(61 * time.Second)
	if ok, _ := a.Admit(id); !ok {
		t.Fatal("rate window should have rolled")
	}
}

func TestAdmissionBudgetRollsAndSeeds(t *testing.T) {
	now := time.Unix(1_800_000_000, 0)
	clock := func() time.Time { return now }
	l := DefaultLimits()
	l.AppBytesPerHour = 100
	a := NewAdmission(l, clock)
	id := Identity{"abc", "meet"}
	if a.AddBytes(id, 60) {
		t.Fatal("60 under 100")
	}
	if !a.AddBytes(id, 50) {
		t.Fatal("110 over 100")
	}
	now = now.Add(61 * time.Minute)
	if a.OverBudget(id, now) {
		t.Fatal("window rolled")
	}
	b := NewAdmission(l, clock)
	b.Seed(id, now.Add(-30*time.Minute), 90)
	if !b.AddBytes(id, 20) {
		t.Fatal("seeded 90 + 20 over 100")
	}
}

func TestBucket(t *testing.T) {
	now := time.Unix(1_800_000_000, 0)
	clock := func() time.Time { return now }
	b := NewBucket(8000, clock) // 1000 bytes/s
	if !b.Allow(1000) {
		t.Fatal("first second's worth")
	}
	if b.Allow(1) {
		t.Fatal("empty")
	}
	now = now.Add(500 * time.Millisecond)
	if !b.Allow(500) || b.Allow(1) {
		t.Fatal("refilled 500")
	}
}
```

- [ ] **Step 2: Implement**

```go
package relay

import (
	"strconv"
	"sync"
	"time"
)

type Limits struct {
	AppLive, AppPerMinute         int
	AppBytesPerHour               int64
	AccountLive, AccountPerMinute int
	AccountBytesPerHour           int64
	BoxLive, BoxPerMinute         int
	AllocBitsPerSec, BoxBitsPerSec int64
}

func DefaultLimits() Limits {
	return Limits{
		AppLive: 64, AppPerMinute: 30, AppBytesPerHour: 20 << 30,
		AccountLive: 256, AccountPerMinute: 60, AccountBytesPerHour: 60 << 30,
		BoxLive: 2048, BoxPerMinute: 500,
		AllocBitsPerSec: 8_000_000, BoxBitsPerSec: 200_000_000,
	}
}

func LimitsFromEnv(getenv func(string) string) Limits {
	l := DefaultLimits()
	i := func(name string, dst *int) {
		if v, err := strconv.Atoi(getenv("CARLOS_RELAY_LIMIT_" + name)); err == nil && v > 0 {
			*dst = v
		}
	}
	i64 := func(name string, dst *int64) {
		if v, err := strconv.ParseInt(getenv("CARLOS_RELAY_LIMIT_"+name), 10, 64); err == nil && v > 0 {
			*dst = v
		}
	}
	i("APP_LIVE", &l.AppLive); i("APP_PER_MINUTE", &l.AppPerMinute); i64("APP_BYTES_PER_HOUR", &l.AppBytesPerHour)
	i("ACCOUNT_LIVE", &l.AccountLive); i("ACCOUNT_PER_MINUTE", &l.AccountPerMinute); i64("ACCOUNT_BYTES_PER_HOUR", &l.AccountBytesPerHour)
	i("BOX_LIVE", &l.BoxLive); i("BOX_PER_MINUTE", &l.BoxPerMinute)
	i64("ALLOC_BPS", &l.AllocBitsPerSec); i64("BOX_BPS", &l.BoxBitsPerSec)
	return l
}

type Identity struct{ Account, App string }

type window struct { // sliding counters bucketed per minute, 60 buckets for an hour
	at [60]time.Time
	n  [60]int64
}

func (w *window) add(now time.Time, n int64) {
	m := now.Truncate(time.Minute)
	i := int(m.Unix()/60) % 60
	if !w.at[i].Equal(m) {
		w.at[i], w.n[i] = m, 0
	}
	w.n[i] += n
}

func (w *window) sum(now time.Time, span time.Duration) int64 {
	var s int64
	cut := now.Add(-span)
	for i := range w.at {
		if !w.at[i].IsZero() && w.at[i].After(cut) {
			s += w.n[i]
		}
	}
	return s
}

type counters struct {
	live    int
	starts  window // attempts per minute
	bytes   window // bytes per rolling hour
}

type Admission struct {
	mu   sync.Mutex
	l    Limits
	now  func() time.Time
	app  map[Identity]*counters
	acct map[string]*counters
	box  counters
}

func NewAdmission(l Limits, now func() time.Time) *Admission {
	return &Admission{l: l, now: now, app: map[Identity]*counters{}, acct: map[string]*counters{}}
}

func (a *Admission) get(id Identity) (*counters, *counters) {
	c, ok := a.app[id]
	if !ok {
		c = &counters{}
		a.app[id] = c
	}
	ac, ok := a.acct[id.Account]
	if !ok {
		ac = &counters{}
		a.acct[id.Account] = ac
	}
	return c, ac
}

// Admit counts an attempt for the app, the account and the box whether or
// not it is admitted, then checks live caps, rates and budgets in that order.
func (a *Admission) Admit(id Identity) (bool, string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	now := a.now()
	c, ac := a.get(id)
	c.starts.add(now, 1); ac.starts.add(now, 1); a.box.starts.add(now, 1)
	switch {
	case c.live >= a.l.AppLive:
		return false, "app live"
	case ac.live >= a.l.AccountLive:
		return false, "account live"
	case a.box.live >= a.l.BoxLive:
		return false, "box live"
	case c.starts.sum(now, time.Minute) > int64(a.l.AppPerMinute):
		return false, "app rate"
	case ac.starts.sum(now, time.Minute) > int64(a.l.AccountPerMinute):
		return false, "account rate"
	case a.box.starts.sum(now, time.Minute) > int64(a.l.BoxPerMinute):
		return false, "box rate"
	case c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour:
		return false, "app budget"
	case ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour:
		return false, "account budget"
	}
	c.live++; ac.live++; a.box.live++
	return true, ""
}

func (a *Admission) Release(id Identity) {
	a.mu.Lock()
	defer a.mu.Unlock()
	c, ac := a.get(id)
	if c.live > 0 { c.live-- }
	if ac.live > 0 { ac.live-- }
	if a.box.live > 0 { a.box.live-- }
}

// AddBytes records forwarded bytes and reports whether the app or account
// rolling-hour budget is now exceeded.
func (a *Admission) AddBytes(id Identity, n int64) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	now := a.now()
	c, ac := a.get(id)
	c.bytes.add(now, n); ac.bytes.add(now, n)
	return c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour || ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour
}

func (a *Admission) Seed(id Identity, at time.Time, n int64) {
	a.mu.Lock()
	defer a.mu.Unlock()
	c, ac := a.get(id)
	c.bytes.add(at, n); ac.bytes.add(at, n)
}

func (a *Admission) OverBudget(id Identity, now time.Time) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	c, ac := a.get(id)
	return c.bytes.sum(now, time.Hour) > a.l.AppBytesPerHour || ac.bytes.sum(now, time.Hour) > a.l.AccountBytesPerHour
}

// Bucket is a token bucket in bytes with a one-second burst.
type Bucket struct {
	mu     sync.Mutex
	rate   float64 // bytes per second
	tokens float64
	last   time.Time
	now    func() time.Time
}

func NewBucket(bitsPerSec int64, now func() time.Time) *Bucket {
	r := float64(bitsPerSec) / 8
	return &Bucket{rate: r, tokens: r, last: now(), now: now}
}

func (b *Bucket) Allow(nbytes int) bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	t := b.now()
	b.tokens += t.Sub(b.last).Seconds() * b.rate
	if b.tokens > b.rate {
		b.tokens = b.rate
	}
	b.last = t
	if float64(nbytes) > b.tokens {
		return false
	}
	b.tokens -= float64(nbytes)
	return true
}
```

- [ ] **Step 3: Run tests, commit**

```bash
git add internal/relay/limits.go internal/relay/limits_test.go
git commit -m "relay: admission, rolling budgets, token buckets (spec §3 limits)"
```

---

### Task 5: Wrapped sockets: counting, buckets, EIP port gate, close

**Files:**
- Create: `internal/relay/conn.go`
- Test: `internal/relay/conn_test.go`

**Interfaces:**
- Produces:
  ```go
  // Binding is what a relay socket belongs to.
  type Binding struct{ ID Identity; Expiry time.Time; Kid string; Src string /* typed source key */ }
  // relayConn wraps an allocation's relay PacketConn.
  type relayConn struct{ /* unexported */ }
  func newRelayConn(inner net.PacketConn, b *Binding, eip net.IP, ports *portSet, appBucket, boxBucket *Bucket, onIn func(Identity, int), onRejected func(Identity, int), onClose func(*relayConn)) *relayConn
  // WriteTo: client→peer forwarding. Drops (returns n,nil) when the bucket refuses or when dst is the EIP on a port not in ports.
  // ReadFrom: peer→relay; counted by the server only after pion forwards (see Task 6), so this wrapper counts nothing on read.
  // Close: idempotent; runs onClose once.
  type portSet struct{ /* unexported; mutex + map[int]bool */ }
  func (p *portSet) add(port int); func (p *portSet) remove(port int); func (p *portSet) has(port int) bool
  func typedSrc(a net.Addr) string   // "udp:" or "tcp:" + a.String(); the owner-binding key
  ```

- [ ] **Step 1: Tests**

```go
func TestRelayConnGatesEIPPorts(t *testing.T) {
	inner := &fakePacketConn{}
	ports := &portSet{m: map[int]bool{50000: true}}
	eip := net.ParseIP("203.0.113.5")
	now := time.Unix(1_800_000_000, 0)
	clock := func() time.Time { return now }
	var in int
	c := newRelayConn(inner, &Binding{ID: Identity{"abc", "meet"}}, eip, ports, NewBucket(1<<40, clock), NewBucket(1<<40, clock),
		func(_ Identity, n int) { in += n }, nil, func(*relayConn) {})
	ok := &net.UDPAddr{IP: eip, Port: 50000}
	bad := &net.UDPAddr{IP: eip, Port: 3478}
	other := &net.UDPAddr{IP: net.ParseIP("198.51.100.9"), Port: 4444}
	c.WriteTo([]byte("abc"), ok)
	c.WriteTo([]byte("abc"), bad)
	c.WriteTo([]byte("abcd"), other)
	if len(inner.written) != 2 || in != 7 {
		t.Fatalf("written=%d in=%d, want 2 packets and 7 bytes", len(inner.written), in)
	}
}

func TestRelayConnBucketDrops(t *testing.T) {
	inner := &fakePacketConn{}
	now := time.Unix(1_800_000_000, 0)
	clock := func() time.Time { return now }
	c := newRelayConn(inner, &Binding{}, nil, &portSet{m: map[int]bool{}}, NewBucket(80, clock), NewBucket(1<<40, clock), func(Identity, int) {}, nil, func(*relayConn) {})
	dst := &net.UDPAddr{IP: net.ParseIP("198.51.100.9"), Port: 4444}
	c.WriteTo(make([]byte, 10), dst) // 10 bytes = the whole second
	c.WriteTo(make([]byte, 1), dst)
	if len(inner.written) != 1 {
		t.Fatalf("written = %d, want 1 (second dropped by bucket)", len(inner.written))
	}
}

func TestRelayConnCloseOnce(t *testing.T) {
	n := 0
	c := newRelayConn(&fakePacketConn{}, &Binding{}, nil, &portSet{m: map[int]bool{}}, nil, nil, nil, nil, func(*relayConn) { n++ })
	c.Close(); c.Close()
	if n != 1 {
		t.Fatalf("onClose ran %d times", n)
	}
}

func TestTypedSrc(t *testing.T) {
	u := &net.UDPAddr{IP: net.ParseIP("192.0.2.1"), Port: 5000}
	tc := &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 5000}
	if typedSrc(u) == typedSrc(tc) {
		t.Fatal("udp and tcp with the same address must not collide")
	}
}

// fakePacketConn records writes; ReadFrom blocks until Close.
type fakePacketConn struct {
	mu      sync.Mutex
	written [][]byte
	closed  chan struct{}
	once    sync.Once
}
func (f *fakePacketConn) init() { f.once.Do(func() { f.closed = make(chan struct{}) }) }
func (f *fakePacketConn) ReadFrom(b []byte) (int, net.Addr, error) { f.init(); <-f.closed; return 0, nil, net.ErrClosed }
func (f *fakePacketConn) WriteTo(b []byte, a net.Addr) (int, error) { f.mu.Lock(); defer f.mu.Unlock(); f.written = append(f.written, append([]byte(nil), b...)); return len(b), nil }
func (f *fakePacketConn) Close() error { f.init(); select { case <-f.closed: default: close(f.closed) }; return nil }
func (f *fakePacketConn) LocalAddr() net.Addr { return &net.UDPAddr{IP: net.IPv4zero, Port: 50000} }
func (f *fakePacketConn) SetDeadline(time.Time) error { return nil }
func (f *fakePacketConn) SetReadDeadline(time.Time) error { return nil }
func (f *fakePacketConn) SetWriteDeadline(time.Time) error { return nil }
```

- [ ] **Step 2: Implement**

```go
package relay

import (
	"net"
	"sync"
	"time"
)

// Binding is the identity an allocation was created under. Src is the
// typed source key (typedSrc) pion will present on later requests.
type Binding struct {
	ID     Identity
	Expiry time.Time
	Kid    string
	Src    string
}

// typedSrc keys the owner binding on transport AND address, because pion
// tags every allocation's tuple as UDP even for TCP clients and a UDP and a
// TCP client can share an address string.
func typedSrc(a net.Addr) string {
	switch a.(type) {
	case *net.TCPAddr:
		return "tcp:" + a.String()
	default:
		return "udp:" + a.String()
	}
}

type portSet struct {
	mu sync.Mutex
	m  map[int]bool
}

func (p *portSet) add(port int)         { p.mu.Lock(); p.m[port] = true; p.mu.Unlock() }
func (p *portSet) remove(port int)      { p.mu.Lock(); delete(p.m, port); p.mu.Unlock() }
func (p *portSet) has(port int) bool    { p.mu.Lock(); defer p.mu.Unlock(); return p.m[port] }

// relayConn is the PacketConn the relay hands pion for one allocation.
// Closing it is how the relay ends the allocation: pion's read loop deletes
// the allocation when ReadFrom errors (allocation.go:301 in pion/turn
// v4.1.4). WriteTo is the client→peer leg and is where forwarded bytes are
// counted, the per-allocation and box buckets applied, and the EIP port
// gate enforced.
type relayConn struct {
	net.PacketConn
	b          *Binding
	eip        net.IP
	ports      *portSet
	appBucket  *Bucket
	boxBucket  *Bucket
	onIn       func(Identity, int)
	onRejected func(Identity, int)
	onClose    func(*relayConn)
	once       sync.Once
}

func newRelayConn(inner net.PacketConn, b *Binding, eip net.IP, ports *portSet, appBucket, boxBucket *Bucket,
	onIn func(Identity, int), onRejected func(Identity, int), onClose func(*relayConn)) *relayConn {
	return &relayConn{PacketConn: inner, b: b, eip: eip, ports: ports, appBucket: appBucket, boxBucket: boxBucket,
		onIn: onIn, onRejected: onRejected, onClose: onClose}
}

func (c *relayConn) WriteTo(p []byte, addr net.Addr) (int, error) {
	if ua, ok := addr.(*net.UDPAddr); ok && c.eip != nil && ua.IP.Equal(c.eip) && !c.ports.has(ua.Port) {
		if c.onRejected != nil {
			c.onRejected(c.b.ID, len(p))
		}
		return len(p), nil // dropped: only live relay ports are reachable on our own EIP
	}
	if (c.appBucket != nil && !c.appBucket.Allow(len(p))) || (c.boxBucket != nil && !c.boxBucket.Allow(len(p))) {
		return len(p), nil // dropped by a token bucket
	}
	n, err := c.PacketConn.WriteTo(p, addr)
	if n > 0 && c.onIn != nil {
		c.onIn(c.b.ID, n)
	}
	return n, err
}

func (c *relayConn) Close() error {
	var err error
	c.once.Do(func() {
		err = c.PacketConn.Close()
		if c.onClose != nil {
			c.onClose(c)
		}
	})
	return err
}
```

- [ ] **Step 3: Run tests, commit**

```bash
git add internal/relay/conn.go internal/relay/conn_test.go
git commit -m "relay: wrapped relay socket with counting, buckets and the EIP port gate (spec §3)"
```

---

### Task 6: The pion server wrapper (Opus)

**Files:**
- Modify: `go.mod`, `go.sum` (add `github.com/pion/turn/v4 v4.1.4`)
- Create: `internal/relay/server.go`
- Test: `internal/relay/server_test.go` (end to end on loopback)

**Interfaces:**
- Consumes: Tasks 1 to 5.
- Produces:
  ```go
  type ServerConfig struct {
      Region      string
      RelayIP     net.IP        // the EIP advertised to clients
      UDP, TCP    string        // listen addresses, e.g. ":3478"; empty disables
      MinPort, MaxPort uint16   // 49152, 65535
      DenyCIDRs   []*net.IPNet  // from CARLOS_RELAY_DENY_CIDRS plus the built-in private ranges
      Masters     func() *Masters
      Authority   func() *Authority
      Lease       func(now time.Time) bool
      Limits      Limits
      Now         func() time.Time
      Log         func(string, ...any)
      Meter       *Meter        // Task 7; nil allowed in tests
  }
  type Server struct{ /* unexported */ }
  func NewServer(cfg ServerConfig) (*Server, error)
  func (s *Server) Close() error
  func (s *Server) EndAll(reason string)                       // lease lapse, shedding
  func (s *Server) EndWhere(func(*Binding) bool, reason string) // disable, incarnation change, region removed, budget, master retired
  func (s *Server) Live() int
  ```

- [ ] **Step 1: Add the dependency** (sandbox lifted for this one command)

Run: `GOFLAGS= GOPROXY=https://proxy.golang.org go get github.com/pion/turn/v4@v4.1.4 && go mod tidy`
Then: `GOFLAGS=-mod=mod GOPROXY=off go build ./...` must still pass; `git diff go.mod` shows only pion modules added.

- [ ] **Step 2: Write the end-to-end test**

```go
func TestServerEndToEnd(t *testing.T) {
	master := sha256.Sum256([]byte("carlos-relay-test-master"))
	masters, _ := ParseMasters(strings.NewReader("k1 " + b64(master[:]) + " -\n"))
	auth := &Authority{Apps: []AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"test"}}}, WrittenAt: time.Now()}
	now := time.Now()
	var mu sync.Mutex
	clock := func() time.Time { mu.Lock(); defer mu.Unlock(); return now }
	s, err := NewServer(ServerConfig{
		Region: "test", RelayIP: net.ParseIP("127.0.0.1"), UDP: "127.0.0.1:0",
		MinPort: 40000, MaxPort: 40100,
		Masters: func() *Masters { return masters }, Authority: func() *Authority { return auth },
		Lease: func(time.Time) bool { return true }, Limits: DefaultLimits(), Now: clock, Log: t.Logf,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()

	key := DeriveKey(master[:], "abc", "meet", "0123456789abcdef")
	exp := now.Add(time.Hour).Unix()
	username := fmt.Sprintf("%d:%s", exp, Suffix("abc", "meet", "0123456789abcdef", "k1"))
	client := turnClient(t, s.UDPAddr(), username, Credential(key, username))
	relayConn, err := client.Allocate()
	if err != nil {
		t.Fatalf("allocate: %v", err)
	}
	defer relayConn.Close()
	if s.Live() != 1 {
		t.Fatalf("live = %d", s.Live())
	}

	// Wrong app's credential from the same source is refused (owner binding).
	other := fmt.Sprintf("%d:%s", exp, Suffix("abc", "docs", "0123456789abcdef", "k1"))
	if _, err := turnClient(t, s.UDPAddr(), other, Credential(DeriveKey(master[:], "abc", "docs", "0123456789abcdef"), other)).Allocate(); err == nil {
		t.Fatal("unknown app allocated")
	}

	// Expiry ends the allocation with no client action.
	mu.Lock(); now = now.Add(2 * time.Hour); mu.Unlock()
	s.tick() // exported for tests: runs the expiry sweep once
	waitFor(t, func() bool { return s.Live() == 0 })
}

func TestServerRefusesPrivatePeer(t *testing.T) {
	// allocate as above, then client.CreatePermission(&net.UDPAddr{IP: net.ParseIP("10.0.0.1"), Port: 1}) must error.
}

func TestServerEndWhere(t *testing.T) {
	// allocate, then s.EndWhere(func(b *Binding) bool { return b.ID.App == "meet" }, "disable"); waitFor Live()==0.
}
```

`turnClient` builds a `turn.Client` with `turn.ClientConfig{STUNServerAddr, TURNServerAddr: addr, Conn: udp, Username, Password, Realm: "carlos"}` and calls `Listen()`; see `github.com/pion/turn/v4` examples `turn-client/udp`. `waitFor` polls for two seconds.

- [ ] **Step 3: Implement `server.go`**

Design, from spec §3:

- `NewServer` opens the UDP and TCP listeners, builds a `turn.ServerConfig{Realm: "carlos", AuthHandler: s.auth, QuotaHandler: s.quota, EventHandler: turn.EventHandler{OnAllocationCreated: s.created, OnAllocationDeleted: s.deleted}, PacketConnConfigs: [{PacketConn: udp, RelayAddressGenerator: s, PermissionHandler: s.permit}], ListenerConfigs: [{Listener: tcp, RelayAddressGenerator: s, PermissionHandler: s.permit}]}`.
- `Server` implements `turn.RelayAddressGenerator`: `Validate()`, `AllocatePacketConn(network string, requestedPort int) (net.PacketConn, net.Addr, error)` binds a UDP socket on `RelayIP`-side interface in `[MinPort, MaxPort]`, wraps it in `newRelayConn`, records it in `s.pending` keyed by relay port, adds the port to `s.ports`, and returns the wrapper and `&net.UDPAddr{IP: cfg.RelayIP, Port: port}`. `AllocateConn` returns `errUnsupported`.
- `auth(username, realm string, src net.Addr) ([]byte, bool)`: `ParseUsername`; expiry in `(now, now+MaxCredentialLife]`; `s.bound[typedSrc(src)]` if present must have the same Identity+Incarnation; `Authority().Lookup`; incarnation and region match; `Masters().Lookup(kid, now)`; `DeriveKey`; `Credential`; `turn.GenerateAuthKey(username, realm, cred)`; stash the parsed username in `s.lastAuth[typedSrc(src)]` so `created` can build the Binding. Count refusal reasons per app in `s.refusals`.
- `quota(username, realm string, src net.Addr) bool`: `Lease(now)` false → refuse; `Admit(Identity)` → refuse with reason logged.
- `created(src, dst net.Addr, proto, username, realm string, relayAddr net.Addr)`: take `s.pending[relayPort]`, set its Binding from `s.lastAuth[typedSrc(src)]`, move it to `s.allocs[typedSrc(src)]`, set `s.bound[typedSrc(src)] = binding`, start the expiry timer (`time.AfterFunc(expiry-now, conn.Close)`).
- `deleted(...)`: no-op; cleanup is in `relayConn.onClose`: remove the port from `s.ports`, `Admission.Release`, delete `s.bound[src]` only if it still points at this conn's binding, delete `s.allocs[src]` likewise.
- `permit(client net.Addr, peer net.IP) bool`: refuse if `peer` is in any `DenyCIDRs`, loopback, link-local, RFC 1918, ULA, or equals a relay listener's private address; allow `RelayIP` (the EIP; the port gate is in `relayConn.WriteTo`).
- `EndWhere`/`EndAll`: iterate `s.allocs`, `Close()` matching conns.
- `tick()`: close any conn whose `Binding.Expiry <= now` (belt for the timers), and any whose master is retired (`Masters().Lookup(kid, now)` false), and any whose app is `OverBudget`.
- Bytes: `onIn` → `Admission.AddBytes`; if over, `EndWhere(app)`. The `out` leg is counted by wrapping the listener conns' `WriteTo` (a `listenerConn` type in this file mirroring `relayConn` without the EIP gate, attributing by `s.bound[typedSrc(dst)]`). Both legs call `Meter.Add(id, in, out)` when `Meter != nil`.
- Built-in deny ranges: `127.0.0.0/8, ::1/128, 169.254.0.0/16, fe80::/10, 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, fc00::/7, 100.64.0.0/10`.

- [ ] **Step 4: Run the end-to-end tests, then the cross-compile check**

Run: `go test ./internal/relay/ -run TestServer -v`
Run: `for t in darwin/arm64 windows/amd64 linux/arm64; do GOOS=${t%/*} GOARCH=${t#*/} CGO_ENABLED=0 go build ./... || echo FAIL $t; done`
Expected: PASS, no FAIL lines.

- [ ] **Step 5: Commit**

```bash
git add go.mod go.sum internal/relay/server.go internal/relay/server_test.go
git commit -m "relay: pion server with our own address generator, owner binding, termination and peer policy (spec §3, §4)"
```

---

### Task 7: Meter

**Files:**
- Create: `internal/relay/meter.go`
- Test: `internal/relay/meter_test.go`

**Interfaces:**
- Produces:
  ```go
  func MeterKey(box, hour string) string   // "metrics/relay/<box>/<hour>.json"
  type MeterSample struct{ At time.Time; Apps map[string]MeterApp }  // key "<account>/<app>"
  type MeterApp struct{ In, Out, Rejected int64; Allocs int }
  type MeterChunk struct{ V int; Box, Hour string; Samples []MeterSample }
  type Meter struct{ /* unexported */ }
  func NewMeter(box string, now func() time.Time) *Meter
  func (m *Meter) Add(id Identity, in, out, rejected int64)
  func (m *Meter) SetAllocs(id Identity, n int)
  func (m *Meter) Flush(ctx, s release.Store) error   // one sample of deltas since last flush; read-append-CAS on the hour's object, retry 3x
  func LoadWindow(ctx, s release.Store, box string, now time.Time) ([]MeterSample, error) // previous and current hour's samples
  ```

- [ ] **Step 1: Tests**

```go
func TestMeterFlushAppendsDeltas(t *testing.T) {
	s := newMemStore()
	now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
	m := NewMeter("box1", func() time.Time { return now })
	id := Identity{"abc", "meet"}
	m.Add(id, 100, 200, 5)
	if err := m.Flush(context.Background(), s); err != nil { t.Fatal(err) }
	m.Add(id, 10, 20, 0)
	if err := m.Flush(context.Background(), s); err != nil { t.Fatal(err) }
	body, _, _ := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12"))
	var c MeterChunk
	json.Unmarshal(body, &c)
	if len(c.Samples) != 2 || c.Samples[0].Apps["abc/meet"].In != 100 || c.Samples[1].Apps["abc/meet"].Out != 20 {
		t.Fatalf("chunk = %+v", c)
	}
}

func TestLoadWindowSpansTwoHours(t *testing.T) {
	s := newMemStore()
	now := time.Date(2026, 9, 7, 12, 1, 0, 0, time.UTC)
	m := NewMeter("box1", func() time.Time { return now.Add(-30 * time.Minute) }) // 11:31
	m.Add(Identity{"abc", "meet"}, 7, 0, 0)
	m.Flush(context.Background(), s)
	samples, err := LoadWindow(context.Background(), s, "box1", now)
	if err != nil || len(samples) != 1 || samples[0].Apps["abc/meet"].In != 7 {
		t.Fatalf("window = %+v %v", samples, err)
	}
}
```

- [ ] **Step 2: Implement**

```go
package relay

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/metrics"
	"github.com/carlosframework/platform/internal/release"
)

func MeterKey(box, hour string) string { return "metrics/relay/" + box + "/" + hour + ".json" }

type MeterApp struct {
	In       int64 `json:"in"`
	Out      int64 `json:"out"`
	Rejected int64 `json:"rejected"`
	Allocs   int   `json:"allocs"`
}

type MeterSample struct {
	At   time.Time           `json:"at"`
	Apps map[string]MeterApp `json:"apps"`
}

type MeterChunk struct {
	V       int           `json:"v"`
	Box     string        `json:"box"`
	Hour    string        `json:"hour"`
	Samples []MeterSample `json:"samples"`
}

type Meter struct {
	mu   sync.Mutex
	box  string
	now  func() time.Time
	cur  map[Identity]*MeterApp
}

func NewMeter(box string, now func() time.Time) *Meter {
	return &Meter{box: box, now: now, cur: map[Identity]*MeterApp{}}
}

func (m *Meter) get(id Identity) *MeterApp {
	a, ok := m.cur[id]
	if !ok {
		a = &MeterApp{}
		m.cur[id] = a
	}
	return a
}

func (m *Meter) Add(id Identity, in, out, rejected int64) {
	m.mu.Lock()
	a := m.get(id)
	a.In += in; a.Out += out; a.Rejected += rejected
	m.mu.Unlock()
}

func (m *Meter) SetAllocs(id Identity, n int) {
	m.mu.Lock()
	m.get(id).Allocs = n
	m.mu.Unlock()
}

// Flush writes one sample of deltas since the previous flush, appended to
// this box's hourly object by read-append-CAS. The relay is the only
// writer of these objects, so the CAS only ever races with itself.
func (m *Meter) Flush(ctx context.Context, s release.Store) error {
	m.mu.Lock()
	now := m.now()
	sample := MeterSample{At: now, Apps: map[string]MeterApp{}}
	for id, a := range m.cur {
		sample.Apps[id.Account+"/"+id.App] = *a
		m.cur[id] = &MeterApp{Allocs: a.Allocs} // deltas reset, gauge kept
	}
	m.mu.Unlock()
	hour := metrics.HourOf(now)
	key := MeterKey(m.box, hour)
	var lastErr error
	for attempt := 0; attempt < 3; attempt++ {
		body, etag, err := s.Get(ctx, key)
		var c MeterChunk
		switch {
		case errors.Is(err, os.ErrNotExist):
			c = MeterChunk{V: 1, Box: m.box, Hour: hour}
			etag = ""
		case err != nil:
			return err
		default:
			if json.Unmarshal(body, &c) != nil {
				c = MeterChunk{V: 1, Box: m.box, Hour: hour}
			}
		}
		c.Samples = append(c.Samples, sample)
		out, _ := json.Marshal(c)
		if _, err = s.Put(ctx, key, out, etag); err == nil {
			return nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return err
		} else {
			lastErr = err
		}
	}
	return lastErr
}

// LoadWindow returns every sample from the previous and current hour's
// objects, for rebuilding rolling-hour budgets after a restart.
func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time) ([]MeterSample, error) {
	var out []MeterSample
	for _, h := range []string{metrics.HourOf(now.Add(-time.Hour)), metrics.HourOf(now)} {
		body, _, err := s.Get(ctx, MeterKey(box, h))
		if errors.Is(err, os.ErrNotExist) {
			continue
		}
		if err != nil {
			return nil, err
		}
		var c MeterChunk
		if json.Unmarshal(body, &c) == nil {
			out = append(out, c.Samples...)
		}
	}
	return out, nil
}
```

- [ ] **Step 3: Run tests, commit**

```bash
git add internal/relay/meter.go internal/relay/meter_test.go
git commit -m "relay: per-app meter with its own hourly objects (spec §3 metering)"
```

---

### Task 8: `Serve`, the unit, bootstrap arming, and the adopt restart

**Files:**
- Create: `internal/relay/serve.go`
- Create: `internal/host/units/carlos-relay.service`
- Modify: `internal/host/bootstrap.go:72` (`enabledUnits`), `:96-110` (`systemctlPlan`)
- Modify: `internal/host/systemapply.go:33` (unit const), `:382` (restart), `:568` (`restartEdge`)
- Modify: `internal/host/host_test.go:29` (`expectedUnits`)
- Modify: `relay.go` (root; created here with `serve` and `keygen`, other verbs in Task 11), `main.go:76` (dispatch)
- Test: `internal/host/host_test.go`, `internal/relay/serve_test.go`

**Interfaces:**
- Produces:
  ```go
  // internal/relay
  type Config struct {
      Box, Region string; MastersPath string; PublicIP net.IP; DenyCIDRs []*net.IPNet
      Store release.Store; Limits Limits; PressurePath string /* /proc/pressure/memory */
  }
  func ConfigFromEnv(ctx context.Context, getenv func(string) string, imdsIP, imdsRegion func() string, store release.Store) (Config, error)
  func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error
  ```
- Env read by `ConfigFromEnv`: `CARLOS_RELAY_PUBLIC_IP` (override), `CARLOS_RELAY_DENY_CIDRS` (comma list), `CARLOS_AWS_REGION` (override), `CARLOS_RELAY_MASTERS_PATH` (default `/etc/carlos/relay-masters`), `CARLOS_RELAY_LIMIT_*`.

- [ ] **Step 1: The unit file**

```ini
# The platform STUN/TURN relay (media roadmap stage 4a). Armed only on a
# box that carries /etc/carlos/relay-masters; ConditionPathExists keeps a
# box without a master from ever listening. Shares the box with the edge and
# must lose first: bounded memory, PSI-driven shedding in the process, and
# the highest OOM adjustment there is.
[Unit]
Description=carlos relay (STUN and TURN)
After=network-online.target
Wants=network-online.target
ConditionPathExists=/etc/carlos/relay-masters

[Service]
EnvironmentFile=-/etc/carlos/host.env
ExecStart=/opt/carlos/carlos relay serve
Restart=always
RestartSec=1

User=carlos
Group=carlos
NoNewPrivileges=yes
ProtectSystem=strict
ProtectHome=yes
ReadWritePaths=/data/relay /run/carlos
PrivateTmp=yes
PrivateDevices=yes
ProtectClock=yes
ProtectHostname=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectKernelLogs=yes
ProtectControlGroups=yes
ProtectProc=invisible
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
RestrictNamespaces=yes
RestrictRealtime=yes
RestrictSUIDSGID=yes
LockPersonality=yes
MemoryDenyWriteExecute=yes
SystemCallFilter=@system-service
SystemCallErrorNumber=EPERM
SystemCallArchitectures=native
MemoryHigh=160M
MemoryMax=256M
OOMScoreAdjust=1000
TasksMax=512
Nice=5
CPUWeight=50

[Install]
WantedBy=multi-user.target
```

- [ ] **Step 2: Arm it in bootstrap** (`internal/host/bootstrap.go`)

Change line 72:
```go
var enabledUnits = []string{"carlos-edge.service", "carlos-relay.service", "carlos-adopt.timer", "carlos-adopt.path", "carlos-stage.path", "carlos-restart.path"}
```
Change `systemctlPlan` so the relay is started, not only enabled (timers and paths already are):
```go
for _, u := range units {
	if strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path") || u == "carlos-relay.service" {
		armed = append(armed, u)
	}
}
```
`ConditionPathExists` makes `start` a harmless no-op on a box with no masters file, so this is safe fleet-wide. Also create `/data/relay` in `bootstrap()` beside the other data dirs (find where `/data` subdirectories are created and add `relay` with the same owner and mode).

- [ ] **Step 3: Restart on platform update** (`internal/host/systemapply.go`)

Add beside line 33: `systemRelayUnit = "carlos-relay"`. Add after `restartEdge`:
```go
// restartRelay mirrors restartEdge for the relay; a unit that is inactive
// because its ConditionPathExists failed returns an error from restart,
// which is not a failure of the update, so it is logged and swallowed.
func (a *systemApplier) restartRelay() {
	a.ops.systemctl("reset-failed", systemRelayUnit)
	if _, err := a.ops.systemctl("try-restart", systemRelayUnit); err != nil {
		log.Printf("system: relay try-restart: %v", err)
	}
}
```
Call `a.restartRelay()` immediately after each `a.restartEdge()` succeeds (lines 382 and the rollback site near 692).

- [ ] **Step 4: Tests** in `internal/host/host_test.go`

Add `"carlos-relay.service"` to `expectedUnits` (keep the slice sorted). Add:
```go
func TestBootstrapStartsRelayUnit(t *testing.T) {
	plan := systemctlPlan(enabledUnits)
	var started []string
	for _, step := range plan {
		if step[0] == "start" {
			started = step[1:]
		}
	}
	if !slices.Contains(started, "carlos-relay.service") {
		t.Fatalf("relay not started by plan: %v", plan)
	}
	if slices.Contains(started, "carlos-edge.service") {
		t.Fatalf("edge must stay start-by-hand: %v", plan)
	}
}
```

- [ ] **Step 5: `Serve` and `ConfigFromEnv`**

```go
// Serve runs the relay until ctx is done: watches the masters file, reads
// authority every minute (and on control/serial change), heartbeats, flushes
// the meter, ticks expiry, and sheds under memory pressure.
func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
	var (
		mu       sync.Mutex
		masters  *Masters
		mastersE string
		auth     *Authority
		lease    Lease
	)
	now := time.Now
	meter := NewMeter(cfg.Box, now)
	if samples, err := LoadWindow(ctx, cfg.Store, cfg.Box, now()); err == nil {
		// seeded into Admission below after NewServer
		_ = samples
	}
	srv, err := NewServer(ServerConfig{
		Region: cfg.Region, RelayIP: cfg.PublicIP, UDP: ":3478", TCP: ":3478", MinPort: 49152, MaxPort: 65535,
		DenyCIDRs: cfg.DenyCIDRs, Limits: cfg.Limits, Now: now, Log: logf, Meter: meter,
		Masters:   func() *Masters { mu.Lock(); defer mu.Unlock(); return masters },
		Authority: func() *Authority { mu.Lock(); defer mu.Unlock(); return auth },
		Lease:     func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) },
	})
	if err != nil {
		return err
	}
	defer srv.Close()
	// (seed: for each sample in the window, srv.admission.Seed(id, sample.At, app.In+app.Out))

	go WatchMasters(ctx, cfg.MastersPath, func(m *Masters, err error) {
		mu.Lock()
		if err != nil { mastersE = err.Error() } else { masters, mastersE = m, "" }
		mu.Unlock()
		if err == nil { srv.tick() } // retire_at may have passed
	})

	readAuth := func() {
		a, _, err := ReadAuthority(ctx, cfg.Store)
		t := now()
		mu.Lock()
		if err == nil { auth = a; lease.Renew(t) } else { logf("relay: authority read: %v", err) }
		live := lease.Live(t)
		hb := Heartbeat{Box: cfg.Box, Region: cfg.Region, ReadAt: t, Listeners: srv.Listeners(), Kids: kidsOf(masters), MastersError: mastersE}
		if auth != nil { hb.WrittenAt = auth.WrittenAt }
		mu.Unlock()
		if !live { srv.EndAll("lease lapsed") }
		srv.EndWhere(func(b *Binding) bool { // authority changed under it
			x, ok := auth.Lookup(b.ID.Account, b.ID.App)
			return !ok || x.Incarnation != b.Incarnation || !slices.Contains(x.Regions, cfg.Region)
		}, "authority")
		if err := WriteHeartbeat(ctx, cfg.Store, hb); err != nil { logf("relay: heartbeat: %v", err) }
	}
	readAuth()
	authTick := time.NewTicker(time.Minute)
	serialTick := time.NewTicker(3 * time.Second)
	meterTick := time.NewTicker(time.Minute)
	expiryTick := time.NewTicker(10 * time.Second)
	psiTick := time.NewTicker(5 * time.Second)
	defer func() { for _, t := range []*time.Ticker{authTick, serialTick, meterTick, expiryTick, psiTick} { t.Stop() } }()
	var lastSerial int64 = -1
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-authTick.C:
			readAuth()
		case <-serialTick.C:
			if doc, _, err := release.ReadSerial(ctx, cfg.Store); err == nil && doc.Serial != lastSerial {
				lastSerial = doc.Serial
				readAuth()
			}
		case <-meterTick.C:
			if err := meter.Flush(ctx, cfg.Store); err != nil { logf("relay: meter flush: %v", err) }
		case <-expiryTick.C:
			srv.tick()
		case <-psiTick.C:
			if some, err := readPSI(cfg.PressurePath); err == nil {
				if some > 20 { srv.ShedOldest(1) } // one per tick until under 10; ShedOldest is a Server method closing the oldest binding
			}
		}
	}
}
```
`readPSI` parses the `some avg10=` field of `/proc/pressure/memory`. `Binding` gains an `Incarnation string` field set at `created` from the parsed username (add it in Task 6 if not already present). Write `ConfigFromEnv` reading the env names above, using `imdsIP()`/`imdsRegion()` when the overrides are empty, and refusing to start with no public IP.

- [ ] **Step 6: The verb**

Create `relay.go` at the root with `runRelay(args)` dispatching `serve` and `keygen` now (other sub-verbs in Task 11):
```go
func runRelayServe(args []string) {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
	defer stop()
	store, err := openDeployStoreForCLI(ctx) // see below
	if err != nil { log.Fatalf("relay serve: %v", err) }
	cfg, err := relay.ConfigFromEnv(ctx, os.Getenv, host.ReadPublicIP, host.ReadPlacementRegion, store)
	if err != nil { log.Fatalf("relay serve: %v", err) }
	if err := relay.Serve(ctx, cfg, log.Printf); err != nil { log.Fatalf("relay serve: %v", err) }
}

func runRelayKeygen(_ []string) {
	key := make([]byte, 32)
	rand.Read(key)
	kid := make([]byte, 4)
	rand.Read(kid)
	fmt.Printf("%s %s -\n", hex.EncodeToString(kid), base64.StdEncoding.EncodeToString(key))
}
```
Export `host.ReadPublicIP` and `host.ReadPlacementRegion` as thin wrappers over the existing unexported `readPublicIP`/`readPlacementRegion` (`internal/host/agent.go:1808`, `:1831`), and `host.OpenDeployStore(ctx)` over `openDeployStore` (`agent.go:2692`); the box label comes from `host.BoxLabel()` wrapping `metricsBoxLabel(os.Getenv, readMachineID())`. Add `case "relay": runRelay(args)` to `main.go` after `store`, and a line to `usage()`.

- [ ] **Step 7: Run all tests, cross-compile, commit**

Run: `go test ./internal/relay/ ./internal/host/ -count=1 && go build ./... && for t in darwin/arm64 windows/amd64 linux/arm64; do GOOS=${t%/*} GOARCH=${t#*/} CGO_ENABLED=0 go build ./... || echo FAIL $t; done`

```bash
git add internal/relay/serve.go internal/host/units/carlos-relay.service internal/host/bootstrap.go internal/host/systemapply.go internal/host/host_test.go internal/host/agent.go internal/host/metrics_loop.go relay.go main.go
git commit -m "relay: carlos relay serve, the unit, bootstrap arming and adopt restart (spec §3)"
```

---

### Task 9: The per-app record and the config value

**Files:**
- Create: `internal/relay/record.go`, `internal/relay/config.go`
- Test: `internal/relay/record_test.go`, `internal/relay/config_test.go`

**Interfaces:**
- Produces:
  ```go
  func RecordKey(acct, app string) string  // "apps/<acct>/<app>/relay.json"
  type RouteCutover struct{ Host string; Suffixes map[string]string /* region→suffix */; At time.Time; By string }
  type Record struct {
      Enabled bool; Incarnation string; Regions []string
      Op int64; Phase string /* "config"|"aggregate"|"done" */
      IssuedSuffixes map[string]string /* region→username_suffix last issued */
      Cutovers []RouteCutover /* member acks for TCP exec routes */
      EnabledAt, DisabledAt time.Time
  }
  func GetRecord(ctx, s release.Store, acct, app string) (*Record, string, error)
  func PutRecord(ctx, s release.Store, acct, app string, r *Record, etag string) (string, error)
  func NewIncarnation() string  // 16 lowercase hex from crypto/rand
  type RegionMaster struct{ Region, Kid string; Master []byte }
  type RegionURLs func(region string) []string   // urls for a region, from the platform domain
  func BuildConfig(acct, app string, r *Record, masters []RegionMaster, urls RegionURLs) (value string, suffixes map[string]string, err error)
  // value is base64(std, padded) of the JSON array in spec §2, elements in r.Regions order; masters must cover every region in r.Regions.
  func ParseConfig(value string) ([]ConfigEntry, error)   // for the config-ack (Task 12) and tests
  type ConfigEntry struct{ Region string; URLs []string; Key string; UsernameSuffix string }
  ```

- [ ] **Step 1: Tests**

```go
func TestBuildConfigMatchesContract(t *testing.T) {
	m := sha256.Sum256([]byte("carlos-relay-test-master"))
	r := &Record{Enabled: true, Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}}
	urls := func(region string) []string {
		h := "relay." + region + ".oncarlos.com:3478"
		return []string{"stun:" + h, "turn:" + h + "?transport=udp", "turn:" + h + "?transport=tcp"}
	}
	val, suffixes, err := BuildConfig("abc", "meet", r, []RegionMaster{{Region: "eu-west-1", Kid: "k1", Master: m[:]}}, urls)
	if err != nil { t.Fatal(err) }
	if err := appconfig.ValidVarValue(val); err != nil { t.Fatalf("not env-safe: %v", err) }
	entries, err := ParseConfig(val)
	if err != nil || len(entries) != 1 { t.Fatalf("parse = %+v %v", entries, err) }
	e := entries[0]
	if e.Key != "uMhwW4P/9O5dQUIs91t2CCHGhwJbeiM0xQMQ2G0CJLU=" || e.UsernameSuffix != "abc:meet:0123456789abcdef:k1" || suffixes["eu-west-1"] != e.UsernameSuffix {
		t.Fatalf("entry = %+v", e)
	}
	if len(e.URLs) != 3 || e.URLs[0] != "stun:relay.eu-west-1.oncarlos.com:3478" { t.Fatalf("urls = %v", e.URLs) }
}

func TestBuildConfigRefusesMissingMaster(t *testing.T) {
	r := &Record{Enabled: true, Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1", "us-east-1"}}
	if _, _, err := BuildConfig("abc", "meet", r, []RegionMaster{{Region: "eu-west-1", Kid: "k1", Master: make([]byte, 32)}}, func(string) []string { return nil }); err == nil {
		t.Fatal("missing us-east-1 master accepted")
	}
}

func TestNewIncarnation(t *testing.T) {
	a, b := NewIncarnation(), NewIncarnation()
	if !incarnationRe.MatchString(a) || a == b { t.Fatalf("%q %q", a, b) }
}

func TestRecordRoundTrip(t *testing.T) {
	s := newMemStore()
	ctx := context.Background()
	r, etag, err := GetRecord(ctx, s, "abc", "meet")
	if err != nil || r != nil || etag != "" { t.Fatalf("absent = %v %q %v", r, etag, err) }
	etag, err = PutRecord(ctx, s, "abc", "meet", &Record{Enabled: true, Incarnation: NewIncarnation(), Op: 1, Phase: "config"}, "")
	if err != nil { t.Fatal(err) }
	if _, err := PutRecord(ctx, s, "abc", "meet", &Record{}, "stale-etag"); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("stale put = %v, want CAS conflict", err)
	}
	r, _, _ = GetRecord(ctx, s, "abc", "meet")
	if !r.Enabled || r.Op != 1 { t.Fatalf("record = %+v", r) }
}
```

- [ ] **Step 2: Implement** `record.go` (mirror `internal/objstore/objstore.go:132-170` Get/Put with `os.ErrNotExist` → nil) and `config.go`:

```go
type ConfigEntry struct {
	Region         string   `json:"region"`
	URLs           []string `json:"urls"`
	Key            string   `json:"key"`
	UsernameSuffix string   `json:"username_suffix"`
}

func BuildConfig(acct, app string, r *Record, masters []RegionMaster, urls RegionURLs) (string, map[string]string, error) {
	byRegion := map[string]RegionMaster{}
	for _, m := range masters { byRegion[m.Region] = m }
	entries := make([]ConfigEntry, 0, len(r.Regions))
	suffixes := map[string]string{}
	for _, region := range r.Regions {
		m, ok := byRegion[region]
		if !ok { return "", nil, fmt.Errorf("relay: no issuer master for region %s", region) }
		key := DeriveKey(m.Master, acct, app, r.Incarnation)
		suffix := Suffix(acct, app, r.Incarnation, m.Kid)
		suffixes[region] = suffix
		entries = append(entries, ConfigEntry{Region: region, URLs: urls(region), Key: base64.StdEncoding.EncodeToString(key), UsernameSuffix: suffix})
	}
	js, err := json.Marshal(entries)
	if err != nil { return "", nil, err }
	return base64.StdEncoding.EncodeToString(js), suffixes, nil
}

func ParseConfig(value string) ([]ConfigEntry, error) {
	js, err := base64.StdEncoding.DecodeString(value)
	if err != nil { return nil, fmt.Errorf("relay: CARLOS_RELAY is not base64: %w", err) }
	var out []ConfigEntry
	if err := json.Unmarshal(js, &out); err != nil { return nil, fmt.Errorf("relay: CARLOS_RELAY is not a JSON array: %w", err) }
	return out, nil
}
```

- [ ] **Step 3: Run tests, commit**

```bash
git add internal/relay/record.go internal/relay/config.go internal/relay/*_test.go
git commit -m "relay: per-app record and the CARLOS_RELAY value (spec §2, §5)"
```

---

### Task 10: Console operations: writer lease, aggregate publish, reconcile (Opus)

**Files:**
- Create: `internal/console/relayops.go`
- Test: `internal/console/relayops_test.go`

**Interfaces:**
- Consumes: Task 9; `appconfig.Apply` (`internal/appconfig/generation.go:119`), `appconfig.Seal`, `release.BumpSerial`, `s.cfg.Releases`, `s.cfg.SecretsPub`.
- Produces (methods on `*Server`, all requiring the writer lease):
  ```go
  const relayWriterKey = "control/relay-writer.json"
  const relayIssuerKey = "control/relay-issuer.json"   // {"<region>":"<kid>"}
  type relayWriter struct{ /* holder id, expiry; renew every 20s; 60s TTL */ }
  func (s *Server) relayWriterLoop(ctx)                 // acquires/renews; sets s.relayHeld atomically
  func (s *Server) relayHold() bool
  func (s *Server) relayMasters(region string) (kid string, master []byte, err error)  // console secret CARLOS_RELAY_MASTERS_<REGION> + issuer pointer
  func (s *Server) relayEnable(ctx, acct *Account, app string, regions []string, by string) (*relay.Record, error)
  func (s *Server) relayDisable(ctx, acct *Account, app string, by string) (*relay.Record, error)
  func (s *Server) relayAdopted(ctx, acct *Account, app, host string, by string) (*relay.Record, error)
  func (s *Server) relayRotate(ctx, region string) error   // gate on heartbeats, move issuer pointer, re-issue every enabled app in region
  func (s *Server) relayPublishAggregate(ctx) error        // etag-first rebuild from a listing, CAS, retry
  func (s *Server) relayReconcile(ctx) error               // every 5 min and at start; converges config and aggregate from every record
  ```
- Console config additions (`cmd/console/main.go` beside `storeOperators()`): `CARLOS_RELAY_REGIONS` (comma list, ordered) → `Config.RelayRegions []string`; `CARLOS_RELAY_MASTERS_<REGION>` read at use time through the console's own sealed-secret accessor (find how the console reads its own sealed secrets, `grep -rn "SecretsPriv\|consoleSecret" internal/console cmd/console`); `Config.PlatformDomain` for relay hostnames (`relay.<region>.<platform domain>`).

- [ ] **Step 1: Tests** (each with the console fixture; `fx.rel` is the `release.Store`)

```go
func TestRelayEnableWritesRecordConfigAndAggregate(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-enable") // like storeFixture plus RelayRegions=["eu-west-1"], a test master in the fixture's console secrets, SecretsPub set
	ctx := context.Background()
	rec, err := fx.srv.relayEnable(ctx, fx.acct(), "titogo", nil, "alice@keymail.dev")
	if err != nil { t.Fatal(err) }
	if !rec.Enabled || rec.Phase != "done" || rec.Op != 1 || len(rec.Incarnation) != 16 { t.Fatalf("record = %+v", rec) }
	m, _, _ := appconfig.Read(ctx, fx.rel, fx.sqid, "titogo") // whatever reads the default bundle manifest
	if v, ok := m.Vars["CARLOS_RELAY"]; !ok || v.Kind != "secret" { t.Fatalf("CARLOS_RELAY = %+v", v) }
	a, _, _ := relay.ReadAuthority(ctx, fx.rel)
	if x, ok := a.Lookup(fx.sqid, "titogo"); !ok || x.Incarnation != rec.Incarnation { t.Fatalf("aggregate = %+v", a) }
}

func TestRelayRepeatEnableKeepsIncarnationAndWritesNoConfig(t *testing.T) { /* enable twice; second returns same Incarnation, Op unchanged, config generation unchanged */ }

func TestRelayDisableThenEnableIsANewIncarnation(t *testing.T) { /* enable, disable (aggregate lacks app, CARLOS_RELAY unset), enable: Incarnation differs */ }

func TestRelayPublishAggregateIsEtagFirst(t *testing.T) {
	// Seed two enabled records A and B. Read the aggregate etag. Disable A directly via PutRecord + relayPublishAggregate.
	// Then call the internal publish helper with the STALE etag and a listing that still has A: it must return ErrCASConflict and, on retry, publish without A.
}

func TestRelayReconcileFinishesInterruptedEnable(t *testing.T) {
	// PutRecord{Enabled:true, Op:1, Phase:"config"} with no config and no aggregate; relayReconcile; expect CARLOS_RELAY set, aggregate has app, Phase "done".
}

func TestRelayReconcileUndoesStaleConfigAfterDisable(t *testing.T) {
	// Disabled record but CARLOS_RELAY present in config; relayReconcile; expect it unset.
}

func TestRelayReconcileRepairsDeletionWithoutRecord(t *testing.T) {
	// Aggregate lists an app with no record; relayReconcile; aggregate no longer lists it.
}

func TestRelayEnableRefusesRouteWithoutDelivery(t *testing.T) {
	// fx.addRoute(registry.Route{Host: "x.titogo...", App: "titogo", Addr: "10.0.0.5:8080"}) with no exec_delivery; relayEnable errors naming the host.
}

func TestRelayRotateRefusesUntilHeartbeatsCarryKid(t *testing.T) {
	// Two masters in the fixture secret for eu-west-1; issuer pointer k1; heartbeat for box1 lists only k1; relayRotate("eu-west-1") errors; write heartbeat listing k2; relayRotate succeeds; issuer pointer now k2; enabled app's IssuedSuffixes end in ":k2".
}

func TestRelayWriterLeaseExcludesSecondHolder(t *testing.T) {
	// Two Server values over one store; first acquires; second's acquire fails until the first's expiry passes on a fake clock.
}
```

- [ ] **Step 2: Implement** in this order, each behind the lease: `relayWriterLoop` (CAS on `relayWriterKey` with `{holder, expires_at}`; acquire if absent or expired; renew every 20s; on failed renew set held=false and let in-flight ops fail their next `requireRelayLease()` check); `relayMasters` (parse `CARLOS_RELAY_MASTERS_<REGION>` as `kid:key[,kid:key]`, read `relayIssuerKey`, return the named one); `relayPublishAggregate` exactly per spec §5 (read aggregate etag first, then `ListDirs("apps/")` and each account's apps to find every `relay.json`, build from records with `Enabled`, `Put` with the first etag, on `ErrCASConflict` restart, four attempts); `relayApplyConfig(ctx, acct, app string, rec *relay.Record)` (build via `relay.BuildConfig` with `urls(region) = []string{"stun:relay."+region+"."+domain+":3478", "turn:...?transport=udp", "turn:...?transport=tcp"}`, `appconfig.Seal` the value with `s.cfg.SecretsPub`, `appconfig.Apply(ctx, s.cfg.Releases, acct.Sqid, app, map[string]appconfig.Var{"CARLOS_RELAY": {Kind: "secret", Envelope: env}}, nil)`; for disable `Apply(..., nil, []string{"CARLOS_RELAY"})`); `relayEnable`/`relayDisable` as the spec's transition table with a per-app `sync.Mutex` from `s.relayAppLocks sync.Map`; the route-delivery check using `s.registryRoutesForApp` (whatever the instances API uses to list an app's routes and their `Backing`/`Addr`) and `appconfig`'s exec-delivery resolution; `relayReconcile` per spec §5 (for every record: if enabled, expected config value equals current → skip, else re-apply; if disabled and present → unset; then unconditional `relayPublishAggregate`); `relayRotate` per spec §5 step 4. Start `relayWriterLoop` and a five-minute `relayReconcile` ticker from the console's existing background-loop startup, gated on `len(s.cfg.RelayRegions) > 0`.

- [ ] **Step 3: Run tests, commit**

```bash
git add internal/console/relayops.go internal/console/relayops_test.go cmd/console/main.go internal/console/server.go
git commit -m "console: relay operations under a writer lease, etag-first aggregate, converging reconcile (spec §5)"
```

---

### Task 11: Console doors and the CLI

**Files:**
- Create: `internal/console/apirelay.go`
- Modify: `internal/console/authcli.go:700` (routes), `apimode.go` (client calls), `relay.go` (root; add enable/status/disable/adopted/rotate), `docs/site/cli.md` (`### carlos relay`)
- Test: `internal/console/apirelay_test.go`

**Interfaces:**
- Routes (`s.cliRoute`, `svcrole.Allow(svcrole.Admin)` for all but status which allows read roles too, matching store):
  ```
  POST /api/cli/apps/{account}/{app}/relay            body {"regions":[...]}        → relayEnable
  GET  /api/cli/apps/{account}/{app}/relay                                          → status
  POST /api/cli/apps/{account}/{app}/relay/disable                                 → relayDisable
  POST /api/cli/apps/{account}/{app}/relay/adopted    body {"host":"..."}          → relayAdopted
  POST /api/cli/relay/rotate                          body {"region":"..."}        → relayRotate (operator: CARLOS_RELAY_OPERATORS, checked after auth, byte-identical 404 on failure)
  ```
- Status reply shape:
  ```go
  type apiRelayReply struct {
      Enabled bool; Incarnation string; Regions []string; Op int64; Phase string
      Issued map[string]string                    // region → username_suffix
      Routes []apiRelayRoute                      // per route: Host, Shape, Delivery, AckedAt, Cutover
      Regions []apiRelayRegion                    // Region, HeartbeatAge string, Current bool, LeaseLive bool, Listeners, Kids []string, Stale bool
      DisablePending []string                     // regions not yet past the disable
      RetireNoEarlierThan map[string]string       // region → RFC3339, when rotate is in progress
  }
  ```

- [ ] **Step 1: Door tests** through `rawReq`/`getJSON`/`postJSON` with `relayURL(fx, app) = fx.ts.URL + "/api/cli/apps/" + acct + "/" + app + "/relay"`: 404 byte-identical for a non-member and an unknown app (compare bodies with the store door's 404); refusal message when `RelayRegions` empty; enable → 200 with `enabled:true` and 16-hex incarnation; status after enable shows the region row from a seeded heartbeat with `stale:true` when the heartbeat is six minutes old; disable → `enabled:false`; adopted for an unknown host → 404 with message; rotate without operator → 404.

- [ ] **Step 2: Implement doors** on the `handleAPIStoreGet` pattern (`internal/console/apistore.go:299`): `apiAccountApp` gate, `storeNoBucket` equivalent when `s.cfg.Releases == nil`, `jsonError` on failures, `writeJSON` on success. Status assembles the reply from `relay.GetRecord`, the heartbeats under `control/relay-ack/`, the config-acks under `control/config-ack/` (Task 12's shape), and the route list.

- [ ] **Step 3: CLI** in `relay.go`: `runRelay` dispatch adds `enable|status|disable|adopted|rotate`; each mirrors `runStoreStatus` (`store.go:157`): flags via `appFlag`, `accountFlags`, `--console`; `requireRelayAPISession` copied from `requireStoreAPISession` with its text adapted; `apiSession.relayEnable/relayGet/relayDisable/relayAdopted/relayRotate` in `apimode.go` on the `storeGet` shape (`apimode.go:1592`); `printRelayStatus` prints the reply as: desired state line, issued kids per region, one line per route with its delivery state, one line per region with heartbeat age and lease, and the retire-no-earlier-than line when present.

- [ ] **Step 4: Docs**: add to `docs/site/cli.md` under "Your apps", after `### carlos store`:

````markdown
### carlos relay

A STUN and TURN relay the platform runs for your app's WebRTC calls, so a
browser can find its public address and, when two networks cannot reach
each other directly, relay through the deployment instead of a third party.

```sh
carlos relay enable --app meet
carlos relay status --app meet
```

`enable` turns it on for one app and delivers `CARLOS_RELAY` to the app's
config: a base64-wrapped JSON list of regions, each with its ICE server URLs,
a key and a username suffix. The app mints a TURN credential per call from
the key (see the relay design spec, §2). `status` shows what was issued,
which routes have picked it up, and the health of each region's relay,
none of which needs box access. `disable` removes it and ends every
allocation the app holds. `adopted --host <route>` is how a member tells the
platform that a TCP-address route it restarted by hand now runs the new
key. `rotate --region <r>` is the operator's verb after staging a new
master; `keygen` prints one. `serve` is the unit's entry point and not a
thing to run by hand.
````

- [ ] **Step 5: Run the docsite gate and all tests, commit**

Run: `go test ./internal/docsite/ ./internal/console/ -count=1 && go build ./...`

```bash
git add internal/console/apirelay.go internal/console/apirelay_test.go internal/console/authcli.go apimode.go relay.go docs/site/cli.md
git commit -m "relay: console doors and carlos relay enable|status|disable|adopted|rotate (spec §5)"
```

---

### Task 12: The per-route config acknowledgement

**Files:**
- Create: `internal/appconfig/configack.go`
- Modify: `internal/appconfig/sync.go:331-434` (`syncHost` gains an ack writer callback), `internal/host/agent.go:2937` (pass the store and box)
- Test: `internal/appconfig/configack_test.go`, `internal/appconfig/sync_test.go`

**Interfaces:**
- Produces:
  ```go
  func ConfigAckKey(box, host string) string   // "control/config-ack/<box>/<host>.json"
  type ConfigAck struct {
      Box, Host, Shape string          // shape: "unit" | "socket-exec" | "tcp-exec-file"
      At time.Time
      Restarted bool; RestartError string
      RelaySuffixes map[string]string   // region → username_suffix parsed from CARLOS_RELAY, nil if absent
  }
  func WriteConfigAck(ctx, s release.Store, ack ConfigAck) error
  // Sync gains a parameter: ack func(ConfigAck) (nil = no acks)
  func Sync(ctx, reg, s, priv, envDir string, restart func(host string) error, ack func(ConfigAck)) error
  ```

- [ ] **Step 1: Tests**: `TestSyncWritesConfigAckNamingRelaySuffixes`: a route with a manifest carrying `CARLOS_RELAY` (built with `relay.BuildConfig` in the test) is synced with a recording `ack` func; the ack has `Shape: "unit"`, `Restarted: true`, `RelaySuffixes["eu-west-1"] == "abc:meet:...:k1"`. `TestSyncAckReportsRestartFailure`: restart returns an error; ack has `Restarted: false` and the error text. `TestSyncAckWithoutRelayHasNilSuffixes`.

- [ ] **Step 2: Implement**: in `syncHost`, after the rename succeeds, build `ConfigAck{Host: rt.Host, Shape: shapeOf(rt), At: time.Now()}`; parse `CARLOS_RELAY` out of the rendered vars (the decrypted manifest is in scope; use `relay.ParseConfig` on its value if present and fill `RelaySuffixes`); call `restart`, set `Restarted`/`RestartError`; call `ack(a)` if non-nil. Import cycle check: `internal/appconfig` importing `internal/relay` is fine only if `internal/relay` does not import `appconfig`; it does not (Task 9's test imports `appconfig` but the package does not). `shapeOf`: `rt.Backing == "unit"` → `"unit"`; `rt.Socket != ""` → `"socket-exec"`; else `"tcp-exec-file"`. The exec-file path (`syncexec.go`) writes the same ack with `Restarted: false`. In `agent.go:2937`, pass `func(a appconfig.ConfigAck) { a.Box = box; if err := appconfig.WriteConfigAck(ctx, store, a); err != nil { log.Printf("adopt: config ack %s: %v", a.Host, err) } }`.

- [ ] **Step 3: Run tests, commit**

```bash
git add internal/appconfig/configack.go internal/appconfig/configack_test.go internal/appconfig/sync.go internal/appconfig/syncexec.go internal/appconfig/sync_test.go internal/host/agent.go
git commit -m "appconfig: per-route config acknowledgement naming rendered relay identities (spec §5)"
```

---

### Task 13: Infrastructure

**Files:**
- Modify: `infra/modules/region-host/variables.tf`, `main.tf`, `outputs.tf`
- Modify (pinfra repo, separate PR there): `flagship/region-ie.tf` and siblings; `host-init.sh` if it stages `host.env`

- [ ] **Step 1: Module**: add
```hcl
variable "relay" {
  description = "Open the platform STUN/TURN relay's ports (udp+tcp 3478, udp 49152-65535). Off by default; a box also needs /etc/carlos/relay-masters before the unit starts."
  type        = bool
  default     = false
}
```
and in `main.tf`, three more `dynamic "ingress"` entries driven by `var.relay ? [1] : []` with descriptions `"relay stun/turn udp"` (3478/3478/udp), `"relay stun/turn tcp"` (3478/3478/tcp), `"relay allocations"` (49152/65535/udp), all `0.0.0.0/0`. Add `output "vpc_cidr" { value = data.aws_vpc.selected.cidr_block }` (find the VPC data source the module already uses). Run `tofu validate` in the module directory.

- [ ] **Step 2: Flagship** (pinfra repo): `relay = true` on the Ireland module first; an `aws_route53_record` `relay.eu-west-1.<platform domain>` A → `module.host_ie.eip_public_ip`; document in `flagship/README.md` that `/etc/carlos/relay-masters` and `CARLOS_RELAY_DENY_CIDRS=<vpc_cidr output>` are staged by hand like `host.env`, and that console config gets `CARLOS_RELAY_REGIONS=eu-west-1` and the sealed `CARLOS_RELAY_MASTERS_EU_WEST_1`.

- [ ] **Step 3: Commit** (platform repo part)

```bash
git add infra/modules/region-host/
git commit -m "infra: relay variable opens stun/turn ingress and outputs the VPC CIDR (spec §6)"
```

---

### Task 14: Whole-branch gate and landing

- [ ] **Step 1: Full gate**: `go build ./... && go vet ./... && go test ./... -count=1 && gofmt -l .` (empty output) and the three cross-compiles.
- [ ] **Step 2: Spec coverage walk**: for each of spec §3's bullets, §4's steps, §5's transitions and §7's unit-test list, name the test that covers it; add any missing test.
- [ ] **Step 3: Opus review** of the whole branch against the spec, with `superpowers:requesting-code-review`.
- [ ] **Step 4: Land**: push to `origin`, `amadan branch describe carlos/platform relay-stun-turn-impl -body -`, wait for `amadan ci status -branch relay-stun-turn-impl` to pass, `amadan branch merge carlos/platform relay-stun-turn-impl -expect <full sha>`.
- [ ] **Step 5: Flagship rollout** (operator, from the spec's §7 gate): stage masters and deny CIDRs on the Ireland box; apply pinfra; promote; confirm `carlos-relay.service` active; `carlos relay enable --app <meet app>`; run the §7 flagship checks in order; message the Meet session.

---

## Self-review

**Spec coverage.** §2 contract → Tasks 1, 9 (vector test in 1, env-safety in 9). §3 process/unit → 8; hardening → 8; listeners and generator → 6; owner binding → 5, 6; termination → 6, 8; peer policy and EIP gate → 5, 6; limits → 4, 6; metering → 5, 6, 7; authority and lease → 3, 8. §4 masters → 2; console custody → 10; verification steps → 6. §5 records → 9; writer lease, transitions, publish, reconcile, rotate → 10; doors, status, adopted → 11; config-ack → 12. §6 → 13. §7 flagship gate → 14. §8 exclusions: none built. Gaps: none found.

**Placeholders.** Task 6 step 3 and Task 10 step 2 are design-level for the Opus tasks by intent, with every function and behaviour named; Task 6's test file names the three tests. No "TBD".

**Type consistency.** `Identity{Account, App}` used in Tasks 4 to 8; `Binding` gains `Incarnation` in Task 6 (noted in Task 8); `relay.Record` fields in Task 9 match Task 10 and 11 uses; `ConfigAck.RelaySuffixes` in Task 12 matches Task 11's status reading; `Sync` signature change in Task 12 must update the one caller at `agent.go:2937` and any test calling `Sync`.