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

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

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

// 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

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

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

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.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

# 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:

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):

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:

// 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:

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

// 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):

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

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

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:

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

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)

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

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.

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 ./...

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

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

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)

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.