amadan
oficina / meet Public
Meet slice 2: holding screen and room, direct connectivity — 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: Two people on the same network open a meeting link, set up camera and microphone on the holding screen, and see and hear each other in the room, with a guest waiting to be admitted by a member; every server-side rule of spec §2.4, §2.5, §4 and §5.1 enforced and tested, and a Playwright gate proving a real call.
Architecture: An in-process hub (internal/calls/hub) owns rooms, participations, admission, the cap, pair generations, replay rings and the tick, and knows nothing about HTTP. Two handlers wrap it: an SSE stream per peer and a JSON POST /signal. Two hand-written scripts do the rest in the browser: devices.js for the holding screen and room.js for signalling, perfect negotiation, resume and reconciliation. Media never touches the server. CARLOS_RELAY is absent in this slice, so calls connect on host candidates only, which is what the localhost browser gate exercises.
Tech Stack: Go 1.26, rastrillo v0.26.0, chi, gorm/SQLite (unchanged schema), html/template, vanilla JavaScript (ES2020, no bundler), Playwright 1.62 driven from a Go test through node.
Spec: docs/superpowers/specs/2026-09-07-meet-design.md revision 8: §2.4, §2.5, §3.3, §3.4, §3.5, §4.1 to §4.6, §5.1, §9, §12 item 2, §13 slice-2 lines. The programme: docs/superpowers/plans/2026-09-09-meet-slices-2-6-programme.md §6.
Global Constraints
- Module
amadan.net/oficina/meet; frameworkamadan.net/rastrillo/rastrillo v0.26.0;CGO_ENABLED=0;make cigreen at the end of every task;make browser(Playwright) green at the end of Tasks 8, 9 and 10. - Media never touches the server: no handler accepts a media byte; the hub reads a message's
type,from,to,seq,genand byte length and never its payload. - No payload, chat text, SDP, candidate, display name, title or link token is ever a log argument (spec §4.4); the captured-log test proves it.
- Authorization per endpoint is exactly spec §2.5's table: entry (
join, fresh stream) checks standing; everything after admission checks the participation;admitand tick promotion re-derive member kind and check the target's entry authority under the hub lock. - Recipients per message type are exactly spec §4.3's table; a waiting guest receives
hello,state, thenrosteron admission, orlost, orended, and nothing else. - Limits (spec §4.4): 64 KB body, 30 messages per second with burst 60 per peer (429 with
Retry-After), sixteen participations per room, eight waiting, one peer per session per room, four streams per person, 64-event subscriber buffer, unjoined and resume windows of thirty seconds, replay ring of thirty-two per participation, tombstones ten minutes. - The cap is four admitted peers, enforced at every transition into admitted under the hub lock, never on resume.
seqis monotonic per (sender, target) for the life of a participation;genis per unordered pair, held by the hub, starting at one.- Destructive actions (remove, takeover) are confirm pages with
ui/confirm-formand POSTs undercsrf.Protect; aGETnever ends a participation. - Every page still works with scripts off; the one exception, browser media participation, is written into
AGENTS.mdin Task 6. - Copy review is waived for this round (suite #5); new strings are still catalogued in
docs/superpowers/copy/2026-09-10-slice-2-strings.mdin Task 11. - Commits end with
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>andClaude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2. - Run from the slice worktree with
GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache; one git command per shell call; nevergit add .claude/.
---
File structure
| File | Responsibility | |---|---| | internal/calls/hub/hub.go | Hub, Room, Participation, peer ids, boot id, locks, limits, tombstones, Open/Resume/Close. | | internal/calls/hub/messages.go | Message envelope, Type constants, Roster and its builders, recipient rules. | | internal/calls/hub/admission.go | Join, Admit, Leave, Remove, Takeover, End, the cap, waiting queue, kind transitions, Tick. | | internal/calls/hub/relay.go | Relay (targeted, seq/gen checked), Reset, pair generations, replay rings, gap. | | internal/calls/hub/subscriber.go | Per-participation channel with the 64-event buffer and the lost {reason: "buffer"} outcome. | | internal/calls/hub/*_test.go | The §13 hub gates, no HTTP. | | internal/calls/stream.go | GET /m/{sqid}/stream: SSE, hello {peer, boot}, heartbeat, TTL, resume answers 404/409/410/503. | | internal/calls/signal.go | POST /m/{sqid}/signal: body limit, rate limit, type dispatch, refusal logging. | | internal/calls/participation.go | Standing-or-participation middleware for /m/{sqid} pages; standingKind adapter the hub calls on tick. | | internal/calls/handlers_room.go | Room page, removal and takeover confirm pages and POSTs. | | internal/calls/templates/room.html, confirm.html (reused), join.html (extended) | Server-rendered halves. | | internal/calls/static/devices.js | Holding screen: enumerate, preview, pickers, mute state, local storage, join button. | | internal/calls/static/room.js | SSE client, outbound queue, perfect negotiation, roster reconciliation, resume, terminal states, admission UI. | | internal/calls/static/calls.css | The grid, tiles and controls; vendored beside rastrillo's sheets. | | test/browser/call.mjs, package.json, Makefile browser target, internal/callstest/browser_test.go (//go:build browser) | The Playwright gate. | | AGENTS.md, docs/superpowers/copy/2026-09-10-slice-2-strings.md | The scriptless exception and the copy catalogue. |
---
Task 1: Hub core: rooms, participations, peer ids, limits
Files:
- Create:
internal/calls/hub/hub.go,internal/calls/hub/subscriber.go,internal/calls/hub/messages.go,internal/calls/hub/hub_test.go
Interfaces:
- Produces: ``
go package hub type PeerID string // 16 random bytes, hex, issued by Open type Kind int // KindGuest, KindMember type Reason string // "buffer","replaced","standing","expired","removed","ended" type Config struct{ Now func() time.Time; Logger *slog.Logger; StandingFor func(ctx context.Context, room string, personID int64) (Kind, bool, error) } func New(cfg Config) *Hub func (h *Hub) Boot() string type Participation struct{ Peer PeerID; Room string; Session string; PersonID int64; Name string; Kind Kind; Admitted bool; JoinedAt time.Time /* unexported: sub, resumeUntil, seqOut map[PeerID]uint64, ring, lastSeen map[PeerID]uint64 */ } type Message struct{ Type stringjson:"type"; From PeerIDjson:"from,omitempty"; Name stringjson:"name,omitempty"; To PeerIDjson:"to,omitempty"; Seq uint64json:"seq,omitempty"; Gen uint64json:"gen,omitempty"; Payload json.RawMessagejson:"payload,omitempty"; Reason Reasonjson:"reason,omitempty"; Roster *Rosterjson:"roster,omitempty"; Boot stringjson:"boot,omitempty"; Peer PeerIDjson:"peer,omitempty"; Waiting booljson:"waiting,omitempty"} type Roster struct{ Peers []RosterPeerjson:"peers"; Waiting []RosterPeerjson:"waiting,omitempty"} type RosterPeer struct{ Peer PeerIDjson:"peer"; Name stringjson:"name"; Kind stringjson:"kind"; Gen uint64json:"gen,omitempty"} type Sub struct{ /* C() <-chan Message; Lost() bool; Close() */ } var ErrRoomFull, ErrSessionHasPeer, ErrPersonStreams, ErrUnknownPeer, ErrNotYours error type OpenResult struct{ Peer PeerID; Sub *Sub; Existing PeerID /* set when ErrSessionHasPeer */ } func (h *Hub) Open(room, session string, personID int64, name string, kind Kind) (OpenResult, error) // fresh stream: creates an unjoined participation type ResumeResult struct{ Sub *Sub; Admitted bool; Replay []Message; Gap []PeerID; Roster *Roster } type ResumeError struct{ Status int; Reason Reason; Boot string } // 404 with Boot, 410 with Reason func (h *Hub) Resume(room, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) func (h *Hub) Detach(room string, peer PeerID) // stream closed: start the resume window func (h *Hub) Find(room, session string, peer PeerID) (*Participation, bool) // bound to session func (h *Hub) LiveForSession(room, session string) (PeerID, bool) // attached or in resume window`` - The 64-event buffer and
lost {reason: "buffer"}; unjoined participations dropped after thirty seconds; resume window thirty seconds; sixteen participations per room (503 before allocation); one peer per session per room (409 with the existing id); four streams per person; tombstones for ten minutes keyed by peer id with their reason.
- [ ] Step 1: Write the failing tests
package hub
import (
"context"
"io"
"log/slog"
"testing"
"time"
)
func testHub(now *time.Time) *Hub {
return New(Config{
Now: func() time.Time { return *now },
Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
StandingFor: func(ctx context.Context, room string, personID int64) (Kind, bool, error) {
if personID < 100 {
return KindMember, true, nil
}
return KindGuest, true, nil
},
})
}
func TestOpenIssuesADistinctPeerBoundToTheSession(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
a, err := h.Open("r1", "sessA", 1, "Ann", KindMember)
if err != nil {
t.Fatal(err)
}
b, _ := h.Open("r1", "sessB", 2, "Bob", KindMember)
if a.Peer == b.Peer || len(a.Peer) != 32 {
t.Fatalf("peer ids %q %q", a.Peer, b.Peer)
}
if _, ok := h.Find("r1", "sessB", a.Peer); ok {
t.Fatal("a peer was found through another session")
}
if _, ok := h.Find("r2", "sessA", a.Peer); ok {
t.Fatal("a peer was found through another room")
}
if p, ok := h.Find("r1", "sessA", a.Peer); !ok || p.Admitted {
t.Fatal("fresh participation should exist and be unjoined")
}
}
func TestOneLivePeerPerSessionPerRoom(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
a, _ := h.Open("r1", "sessA", 1, "Ann", KindMember)
_, err := h.Open("r1", "sessA", 1, "Ann", KindMember)
if err != ErrSessionHasPeer {
t.Fatalf("second stream: %v", err)
}
h.Detach("r1", a.Peer)
if _, err := h.Open("r1", "sessA", 1, "Ann", KindMember); err != ErrSessionHasPeer {
t.Fatal("a detached peer inside its resume window should still block a fresh stream")
}
now = now.Add(31 * time.Second)
h.Tick(context.Background())
if _, err := h.Open("r1", "sessA", 1, "Ann", KindMember); err != nil {
t.Fatalf("after the window: %v", err)
}
}
func TestSixteenParticipationsPerRoomThenRefuse(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
for i := 0; i < 16; i++ {
if _, err := h.Open("r1", "s"+string(rune('a'+i)), int64(100+i), "g", KindGuest); err != nil {
t.Fatalf("open %d: %v", i, err)
}
}
if _, err := h.Open("r1", "sZ", 200, "g", KindGuest); err != ErrRoomFull {
t.Fatalf("seventeenth: %v", err)
}
}
func TestUnjoinedParticipationDropsAfterThirtySeconds(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
a, _ := h.Open("r1", "sessA", 1, "Ann", KindMember)
now = now.Add(31 * time.Second)
h.Tick(context.Background())
if _, ok := h.Find("r1", "sessA", a.Peer); ok {
t.Fatal("unjoined participation survived the window")
}
if !a.Sub.Lost() {
t.Fatal("its subscriber was not closed")
}
}
func TestFourStreamsPerPerson(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
for i := 0; i < 4; i++ {
if _, err := h.Open("r"+string(rune('1'+i)), "s"+string(rune('1'+i)), 1, "Ann", KindMember); err != nil {
t.Fatal(err)
}
}
if _, err := h.Open("r9", "s9", 1, "Ann", KindMember); err != ErrPersonStreams {
t.Fatalf("fifth stream: %v", err)
}
}
func TestSlowSubscriberIsDeclaredLost(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
a, _ := h.Open("r1", "sessA", 1, "Ann", KindMember)
for i := 0; i < 70; i++ {
h.deliver("r1", a.Peer, Message{Type: "peer"})
}
if !a.Sub.Lost() {
t.Fatal("65th undelivered event did not declare the subscriber lost")
}
var last Message
for m := range a.Sub.C() {
last = m
}
if last.Type != "lost" || last.Reason != "buffer" {
t.Fatalf("last event %+v, want lost/buffer", last)
}
}- [ ] Step 2: Run to verify they fail
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/calls/hub/ 2>&1 | head -5 Expected: build failure, package does not exist.
- [ ] Step 3: Implement the core
subscriber.go:
package hub
import "sync/atomic"
const subBuffer = 64
// Sub is one stream's outbound channel. A subscriber that falls 64
// events behind is declared lost and told so (spec §4.4), as in Sheets.
type Sub struct {
ch chan Message
lost atomic.Bool
done chan struct{}
}
func newSub() *Sub { return &Sub{ch: make(chan Message, subBuffer), done: make(chan struct{})} }
func (s *Sub) C() <-chan Message { return s.ch }
func (s *Sub) Lost() bool { return s.lost.Load() }
// send is non-blocking; a full buffer declares the subscriber lost,
// drains room for the lost event, and closes.
func (s *Sub) send(m Message) {
select {
case <-s.done:
return
default:
}
select {
case s.ch <- m:
default:
s.terminate(Message{Type: "lost", Reason: ReasonBuffer})
}
}
func (s *Sub) terminate(final Message) {
if s.lost.Swap(true) {
return
}
close(s.done)
// Make room for the final event, then close so the range ends.
select {
case <-s.ch:
default:
}
s.ch <- final
close(s.ch)
}
func (s *Sub) Close() { s.terminate(Message{Type: "lost", Reason: ReasonBuffer}) }messages.go:
package hub
import "encoding/json"
type PeerID string
type Kind int
type Reason string
const (
KindGuest Kind = iota
KindMember
)
func (k Kind) String() string {
if k == KindMember {
return "member"
}
return "guest"
}
const (
ReasonBuffer Reason = "buffer"
ReasonReplaced Reason = "replaced"
ReasonStanding Reason = "standing"
ReasonExpired Reason = "expired"
ReasonRemoved Reason = "removed"
ReasonEnded Reason = "ended"
)
// Upstream types a browser may send.
const (
TypeJoin = "join"
TypeAdmit = "admit"
TypeLeave = "leave"
TypeOffer = "offer"
TypeAnswer = "answer"
TypeCandidate = "candidate"
TypeReset = "reset"
TypeChat = "chat" // slice 4
TypeShare = "share" // slice 4
)
// Downstream types the server emits.
const (
TypeHello = "hello"
TypeRoster = "roster"
TypePeer = "peer"
TypeGone = "gone"
TypeKnock = "knock"
TypeKnockGone = "knock-gone"
TypeFull = "full"
TypeEnded = "ended"
TypeLost = "lost"
TypePair = "pair"
TypeGap = "gap"
TypeState = "state"
)
// Message is the one envelope, built by the server. Payload is opaque.
type Message struct {
Type string `json:"type"`
From PeerID `json:"from,omitempty"`
Name string `json:"name,omitempty"`
To PeerID `json:"to,omitempty"`
Seq uint64 `json:"seq,omitempty"`
Gen uint64 `json:"gen,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
Reason Reason `json:"reason,omitempty"`
Roster *Roster `json:"roster,omitempty"`
Boot string `json:"boot,omitempty"`
Peer PeerID `json:"peer,omitempty"`
Waiting bool `json:"waiting,omitempty"`
}
type Roster struct {
Peers []RosterPeer `json:"peers"`
Waiting []RosterPeer `json:"waiting,omitempty"`
}
type RosterPeer struct {
Peer PeerID `json:"peer"`
Name string `json:"name"`
Kind string `json:"kind"`
Gen uint64 `json:"gen,omitempty"`
}hub.go (the data model and the four lifecycle entry points; admission and relay come in Tasks 2 and 3 but their fields are declared here):
package hub
import (
"context"
"crypto/rand"
"encoding/hex"
"errors"
"log/slog"
"sync"
"time"
)
const (
MaxParticipations = 16
MaxWaiting = 8
MaxAdmitted = 4
MaxStreamsPerPerson = 4
UnjoinedWindow = 30 * time.Second
ResumeWindow = 30 * time.Second
TombstoneLife = 10 * time.Minute
RingSize = 32
)
var (
ErrRoomFull = errors.New("hub: room holds sixteen participations")
ErrSessionHasPeer = errors.New("hub: this session already has a peer in the room")
ErrPersonStreams = errors.New("hub: this person has four streams open")
ErrUnknownPeer = errors.New("hub: no such peer")
ErrNotYours = errors.New("hub: peer is not bound to this session")
ErrNotAdmitted = errors.New("hub: peer is not admitted")
ErrNotMember = errors.New("hub: peer is not a member")
ErrNotWaiting = errors.New("hub: target is not waiting")
ErrStaleGen = errors.New("hub: stale generation")
ErrTooManyWaiting = errors.New("hub: eight already waiting")
)
type Config struct {
Now func() time.Time
Logger *slog.Logger
// StandingFor re-derives a person's kind and whether they still hold
// entry authority for the room. ok=false means standing none.
StandingFor func(ctx context.Context, room string, personID int64) (kind Kind, ok bool, err error)
}
type Participation struct {
Peer PeerID
Room string
Session string
PersonID int64
Name string
Kind Kind
Admitted bool
Joined bool // has posted join (waiting or admitted)
JoinedAt time.Time
OpenedAt time.Time
sub *Sub
resumeUntil time.Time // zero while a stream is attached
seqOut map[PeerID]uint64 // next seq per target (unused by the hub; kept by browsers)
lastSeen map[PeerID]uint64
ring []Message // targeted messages delivered to this peer, newest last, cap RingSize
}
type pairKey struct{ a, b PeerID }
func keyFor(x, y PeerID) pairKey {
if x < y {
return pairKey{x, y}
}
return pairKey{y, x}
}
type room struct {
sqid string
peers map[PeerID]*Participation
gens map[pairKey]uint64
ended bool
}
type tombstone struct {
reason Reason
at time.Time
}
type Hub struct {
cfg Config
boot string
mu sync.Mutex
rooms map[string]*room
bySession map[string]map[string]PeerID // room -> session -> peer
byPerson map[int64]int // open streams per person
tombstones map[PeerID]tombstone
}
func New(cfg Config) *Hub {
if cfg.Now == nil {
cfg.Now = time.Now
}
return &Hub{
cfg: cfg, boot: randomID(),
rooms: map[string]*room{}, bySession: map[string]map[string]PeerID{},
byPerson: map[int64]int{}, tombstones: map[PeerID]tombstone{},
}
}
func randomID() string {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
panic(err)
}
return hex.EncodeToString(b)
}
func (h *Hub) Boot() string { return h.boot }
func (h *Hub) roomLocked(sqid string) *room {
r, ok := h.rooms[sqid]
if !ok {
r = &room{sqid: sqid, peers: map[PeerID]*Participation{}, gens: map[pairKey]uint64{}}
h.rooms[sqid] = r
}
return r
}
type OpenResult struct {
Peer PeerID
Sub *Sub
Existing PeerID
}
// Open is a fresh stream: a new unjoined participation, subject to the
// room, session and person limits, all checked before allocation.
func (h *Hub) Open(sqid, session string, personID int64, name string, kind Kind) (OpenResult, error) {
h.mu.Lock()
defer h.mu.Unlock()
r := h.roomLocked(sqid)
if r.ended {
return OpenResult{}, ErrUnknownPeer
}
if existing, ok := h.bySession[sqid][session]; ok {
return OpenResult{Existing: existing}, ErrSessionHasPeer
}
if len(r.peers) >= MaxParticipations {
return OpenResult{}, ErrRoomFull
}
if h.byPerson[personID] >= MaxStreamsPerPerson {
return OpenResult{}, ErrPersonStreams
}
p := &Participation{
Peer: PeerID(randomID()), Room: sqid, Session: session, PersonID: personID, Name: name, Kind: kind,
OpenedAt: h.cfg.Now(), sub: newSub(), seqOut: map[PeerID]uint64{}, lastSeen: map[PeerID]uint64{},
}
r.peers[p.Peer] = p
if h.bySession[sqid] == nil {
h.bySession[sqid] = map[string]PeerID{}
}
h.bySession[sqid][session] = p.Peer
h.byPerson[personID]++
return OpenResult{Peer: p.Peer, Sub: p.sub}, nil
}
// Find answers a peer only through its own room and session.
func (h *Hub) Find(sqid, session string, peer PeerID) (*Participation, bool) {
h.mu.Lock()
defer h.mu.Unlock()
p, ok := h.findLocked(sqid, session, peer)
return p, ok
}
func (h *Hub) findLocked(sqid, session string, peer PeerID) (*Participation, bool) {
r, ok := h.rooms[sqid]
if !ok {
return nil, false
}
p, ok := r.peers[peer]
if !ok || p.Session != session {
return nil, false
}
return p, true
}
func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) {
h.mu.Lock()
defer h.mu.Unlock()
id, ok := h.bySession[sqid][session]
return id, ok
}
// Detach starts the resume window for a participation whose stream
// closed. Its subscriber is closed; a resume attaches a new one.
func (h *Hub) Detach(sqid string, peer PeerID) {
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
if !ok {
return
}
p, ok := r.peers[peer]
if !ok {
return
}
if p.sub != nil {
p.sub.terminate(Message{Type: TypeLost, Reason: ReasonBuffer})
p.sub = nil
}
p.resumeUntil = h.cfg.Now().Add(ResumeWindow)
}
// endLocked removes a participation for a terminal reason, closes its
// stream with lost{reason}, leaves a tombstone, and returns whether it
// was admitted or waiting so the caller can announce the departure.
func (h *Hub) endLocked(r *room, p *Participation, reason Reason) (wasAdmitted, wasWaiting bool) {
wasAdmitted, wasWaiting = p.Admitted, p.Joined && !p.Admitted
if p.sub != nil {
p.sub.terminate(Message{Type: TypeLost, Reason: reason})
}
delete(r.peers, p.Peer)
if h.bySession[r.sqid][p.Session] == p.Peer {
delete(h.bySession[r.sqid], p.Session)
}
if h.byPerson[p.PersonID] > 0 {
h.byPerson[p.PersonID]--
}
h.tombstones[p.Peer] = tombstone{reason: reason, at: h.cfg.Now()}
for k := range r.gens {
if k.a == p.Peer || k.b == p.Peer {
delete(r.gens, k)
}
}
return
}
// deliver is the one place an event reaches a subscriber; targeted
// messages are also kept in the recipient's replay ring (Task 3).
func (h *Hub) deliver(sqid string, to PeerID, m Message) {
h.mu.Lock()
defer h.mu.Unlock()
h.deliverLocked(sqid, to, m)
}
func (h *Hub) deliverLocked(sqid string, to PeerID, m Message) {
r, ok := h.rooms[sqid]
if !ok {
return
}
p, ok := r.peers[to]
if !ok {
return
}
if p.sub != nil {
p.sub.send(m)
}
}Tick (in admission.go, Task 2) is referenced by the tests here; for this task add a minimal Tick in hub.go that drops unjoined participations past UnjoinedWindow and resume-held ones past resumeUntil (as a leave, no tombstone), and expires tombstones. Task 2 extends it with kind re-derivation.
// Tick runs the fifteen-second housekeeping: unjoined and lapsed
// participations are dropped, tombstones expire. Task 2 adds kind
// re-derivation and promotion.
func (h *Hub) Tick(ctx context.Context) {
h.mu.Lock()
defer h.mu.Unlock()
now := h.cfg.Now()
for _, r := range h.rooms {
for _, p := range r.peers {
switch {
case !p.Joined && p.sub != nil && now.Sub(p.OpenedAt) > UnjoinedWindow:
h.endLocked(r, p, ReasonBuffer) // dropped quietly; the browser sees a closed stream
delete(h.tombstones, p.Peer)
case p.sub == nil && !p.resumeUntil.IsZero() && now.After(p.resumeUntil):
wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer)
delete(h.tombstones, p.Peer)
h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
}
}
}
for id, t := range h.tombstones {
if now.Sub(t.at) > TombstoneLife {
delete(h.tombstones, id)
}
}
}with announceDepartureLocked stubbed in this task (gone to admitted peers, knock-gone to admitted members; Task 2 fills the recipient helpers):
func (h *Hub) announceDepartureLocked(r *room, p *Participation, wasAdmitted, wasWaiting bool) {
switch {
case wasAdmitted:
h.broadcastLocked(r, Message{Type: TypeGone, From: p.Peer}, admittedPeers)
case wasWaiting:
h.broadcastLocked(r, Message{Type: TypeKnockGone, From: p.Peer}, admittedMembers)
}
}
type audience func(*Participation) bool
func admittedPeers(p *Participation) bool { return p.Admitted }
func admittedMembers(p *Participation) bool { return p.Admitted && p.Kind == KindMember }
func (h *Hub) broadcastLocked(r *room, m Message, who audience) {
for _, q := range r.peers {
if who(q) && q.sub != nil {
q.sub.send(m)
}
}
}- [ ] Step 4: Run the tests
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/calls/hub/ -v 2>&1 | tail -12 Expected: six tests PASS.
- [ ] Step 5: Gate and commit
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache make ci 2>&1 | tail -3
git add internal/calls/hub
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "hub: rooms, participations, peer ids, limits, tombstones
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 2: Admission: join, admit, leave, remove, takeover, end, the cap, the tick
Files:
- Create:
internal/calls/hub/admission.go,internal/calls/hub/admission_test.go - Modify:
internal/calls/hub/hub.go(Tickextended)
Interfaces:
- Produces: ``
go func (h *Hub) Join(ctx, sqid, session string, peer PeerID) error // standing re-checked via cfg.StandingFor; member: admit under cap or ErrRoomFull→fullto the joiner; guest: waiting or ErrTooManyWaiting func (h *Hub) Admit(ctx, sqid, session string, actor, target PeerID) error // actor admitted member (kind re-derived now); target waiting with entry authority now; cap under lock func (h *Hub) Leave(sqid, session string, peer PeerID) error func (h *Hub) Remove(ctx, sqid, session string, actor, target PeerID) error // §3.5, not a signalling message func (h *Hub) Takeover(sqid, session string) (old PeerID, err error) // §4.2: ends the session's live participation with the state-dependent departure func (h *Hub) End(sqid string) // meeting ended:endedto all, room dropped func (h *Hub) rosterFor(r *room, viewer *Participation) *Roster // waiting queue only for members; per-pair gen for the viewer`` - The cap check
admittedCountLocked(r) >= MaxAdmittedat both transitions;fullto the actor;rosterto the newly admitted;peerto the others;knockto admitted members on a guest join;knock-goneon any waiting departure;Tickre-derives kind for every joined participation throughcfg.StandingFor, ends members who lost standing withReasonStanding, ends waiting guests whose authority lapsed withReasonExpired, refreshesrosterto admitted peers on a kind change, and attempts promotion of a waiting peer that became a member.
- [ ] Step 1: Write the failing tests
package hub
import (
"context"
"testing"
"time"
)
// drain returns every event currently buffered on a sub, non-blocking.
func drain(s *Sub) []Message {
var out []Message
for {
select {
case m, ok := <-s.C():
if !ok {
return out
}
out = append(out, m)
default:
return out
}
}
}
func types(ms []Message) []string {
out := make([]string, len(ms))
for i, m := range ms {
out[i] = m.Type
}
return out
}
func TestMemberJoinsAndGuestWaits(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
if err := h.Join(ctx, "r1", "sm", m.Peer); err != nil {
t.Fatal(err)
}
ev := drain(m.Sub)
if len(ev) != 1 || ev[0].Type != TypeRoster || len(ev[0].Roster.Peers) != 1 {
t.Fatalf("member join events %v", types(ev))
}
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
if err := h.Join(ctx, "r1", "sg", g.Peer); err != nil {
t.Fatal(err)
}
if ev := drain(g.Sub); len(ev) != 0 {
t.Fatalf("waiting guest received %v", types(ev))
}
ev = drain(m.Sub)
if len(ev) != 1 || ev[0].Type != TypeKnock || ev[0].From != g.Peer || ev[0].Name != "Gus" {
t.Fatalf("member did not get the knock: %v", types(ev))
}
}
func TestAdmitSendsRosterToGuestPeerToOthersAndKnockGone(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
h.Join(ctx, "r1", "sm", m.Peer)
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
drain(m.Sub)
if err := h.Admit(ctx, "r1", "sm", m.Peer, g.Peer); err != nil {
t.Fatal(err)
}
gev := drain(g.Sub)
if len(gev) != 1 || gev[0].Type != TypeRoster || len(gev[0].Roster.Peers) != 2 || gev[0].Roster.Waiting != nil {
t.Fatalf("guest got %v (waiting queue must not reach a guest)", types(gev))
}
mev := types(drain(m.Sub))
if len(mev) != 2 || mev[0] != TypePeer || mev[1] != TypeKnockGone {
t.Fatalf("member got %v", mev)
}
}
func TestOnlyAnAdmittedMemberMayAdmit(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
h.Join(ctx, "r1", "sm", m.Peer)
g1, _ := h.Open("r1", "s1", 100, "G1", KindGuest)
h.Join(ctx, "r1", "s1", g1.Peer)
g2, _ := h.Open("r1", "s2", 101, "G2", KindGuest)
h.Join(ctx, "r1", "s2", g2.Peer)
h.Admit(ctx, "r1", "sm", m.Peer, g1.Peer)
if err := h.Admit(ctx, "r1", "s1", g1.Peer, g2.Peer); err != ErrNotMember {
t.Fatalf("admitted guest admitting: %v", err)
}
if err := h.Admit(ctx, "r1", "s2", g2.Peer, g2.Peer); err != ErrNotAdmitted {
t.Fatalf("waiting guest admitting: %v", err)
}
}
func TestCapAtJoinAndAtAdmit(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
var members []OpenResult
for i := 0; i < 5; i++ {
o, _ := h.Open("r1", "s"+string(rune('a'+i)), int64(1+i), "M", KindMember)
members = append(members, o)
h.Join(ctx, "r1", "s"+string(rune('a'+i)), o.Peer)
}
fifth := drain(members[4].Sub)
if len(fifth) != 1 || fifth[0].Type != TypeFull {
t.Fatalf("fifth member got %v, want full", types(fifth))
}
// One leaves; four queued guests and the freed slot yield one admission.
h.Leave("r1", "sd", members[3].Peer)
var guests []OpenResult
for i := 0; i < 4; i++ {
o, _ := h.Open("r1", "g"+string(rune('a'+i)), int64(100+i), "G", KindGuest)
guests = append(guests, o)
h.Join(ctx, "r1", "g"+string(rune('a'+i)), o.Peer)
}
drain(members[0].Sub)
if err := h.Admit(ctx, "r1", "sa", members[0].Peer, guests[0].Peer); err != nil {
t.Fatal(err)
}
if err := h.Admit(ctx, "r1", "sa", members[0].Peer, guests[1].Peer); err != nil {
t.Fatal(err) // Admit itself returns nil; the outcome is the `full` event
}
ev := types(drain(members[0].Sub))
if ev[len(ev)-1] != TypeFull {
t.Fatalf("second admit over the cap: %v, want full last", ev)
}
if p, _ := h.Find("r1", "gb", guests[1].Peer); p.Admitted {
t.Fatal("guest admitted over the cap")
}
}
func TestEightWaitingThenFull(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
for i := 0; i < 8; i++ {
o, _ := h.Open("r1", "g"+string(rune('a'+i)), int64(100+i), "G", KindGuest)
h.Join(ctx, "r1", "g"+string(rune('a'+i)), o.Peer)
}
o, _ := h.Open("r1", "gz", 200, "G", KindGuest)
if err := h.Join(ctx, "r1", "gz", o.Peer); err != ErrTooManyWaiting {
t.Fatalf("ninth waiting: %v", err)
}
if ev := drain(o.Sub); len(ev) != 1 || ev[0].Type != TypeFull {
t.Fatalf("ninth got %v", types(ev))
}
}
func TestGuestFirstArrivalIsInTheMembersRoster(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
h.Join(ctx, "r1", "sm", m.Peer)
ev := drain(m.Sub)
if len(ev) != 1 || ev[0].Type != TypeRoster || len(ev[0].Roster.Waiting) != 1 || ev[0].Roster.Waiting[0].Peer != g.Peer {
t.Fatalf("member roster %+v", ev)
}
}
func TestAdmitRefusesAnExpiredGrantBeforeTheTick(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
authority := map[int64]bool{100: true}
h := New(Config{Now: func() time.Time { return now }, StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
if id < 100 {
return KindMember, true, nil
}
return KindGuest, authority[id], nil
}})
ctx := context.Background()
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
h.Join(ctx, "r1", "sm", m.Peer)
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
authority[100] = false // grant expired, no tick yet
drain(m.Sub)
if err := h.Admit(ctx, "r1", "sm", m.Peer, g.Peer); err != nil {
t.Fatal(err)
}
if p, ok := h.Find("r1", "sg", g.Peer); ok && p.Admitted {
t.Fatal("expired guest was admitted")
}
gev := drain(g.Sub)
if len(gev) == 0 || gev[len(gev)-1].Type != TypeLost || gev[len(gev)-1].Reason != ReasonExpired {
t.Fatalf("expired guest got %v", types(gev))
}
if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeKnockGone {
t.Fatalf("member got %v", ev)
}
}
func TestTickDropsAMemberWhoLostStandingAndPromotesAWaitingGuest(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
kinds := map[int64]Kind{1: KindMember, 100: KindGuest}
ok := map[int64]bool{1: true, 100: true}
h := New(Config{Now: func() time.Time { return now }, StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
return kinds[id], ok[id], nil
}})
ctx := context.Background()
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
h.Join(ctx, "r1", "sm", m.Peer)
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
drain(m.Sub)
kinds[100] = KindMember // promoted while waiting
h.Tick(ctx)
if p, _ := h.Find("r1", "sg", g.Peer); !p.Admitted || p.Kind != KindMember {
t.Fatal("waiting guest promoted to member was not admitted on the tick")
}
ok[1] = false // member removed from the allowlist
drain(g.Sub)
h.Tick(ctx)
if _, found := h.Find("r1", "sm", m.Peer); found {
t.Fatal("removed member still in the room")
}
mev := drain(m.Sub)
if len(mev) == 0 || mev[len(mev)-1].Reason != ReasonStanding {
t.Fatalf("removed member got %v", types(mev))
}
if ev := types(drain(g.Sub)); len(ev) == 0 || ev[len(ev)-1] != TypeGone {
t.Fatalf("remaining peer got %v, want gone last", ev)
}
}
func TestTakeoverDepartureDependsOnState(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
h.Join(ctx, "r1", "sm", m.Peer)
ag, _ := h.Open("r1", "sag", 100, "Adm", KindGuest)
h.Join(ctx, "r1", "sag", ag.Peer)
h.Admit(ctx, "r1", "sm", m.Peer, ag.Peer)
w, _ := h.Open("r1", "sw", 101, "Wait", KindGuest)
h.Join(ctx, "r1", "sw", w.Peer)
u, _ := h.Open("r1", "su", 102, "Unj", KindGuest)
drain(m.Sub)
drain(ag.Sub)
// waiting takeover: knock-gone to members only
if _, err := h.Takeover("r1", "sw"); err != nil {
t.Fatal(err)
}
if ev := types(drain(ag.Sub)); len(ev) != 0 {
t.Fatalf("admitted guest saw a waiting departure: %v", ev)
}
if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeKnockGone {
t.Fatalf("member got %v", ev)
}
// unjoined takeover: nothing
h.Takeover("r1", "su")
_ = u
if ev := types(drain(m.Sub)); len(ev) != 0 {
t.Fatalf("unjoined takeover announced %v", ev)
}
// admitted takeover: gone to admitted peers, old stream lost/replaced
h.Takeover("r1", "sag")
agev := drain(ag.Sub)
if len(agev) == 0 || agev[len(agev)-1].Reason != ReasonReplaced {
t.Fatalf("replaced peer got %v", types(agev))
}
if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeGone {
t.Fatalf("member got %v", ev)
}
if _, err := h.Open("r1", "sag", 100, "Adm", KindGuest); err != nil {
t.Fatalf("fresh stream after takeover: %v", err)
}
}
func TestEndSendsEndedToEveryoneAndDropsTheRoom(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
h.Join(ctx, "r1", "sm", m.Peer)
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
h.End("r1")
for _, s := range []*Sub{m.Sub, g.Sub} {
ev := drain(s)
if len(ev) == 0 || ev[len(ev)-1].Type != TypeLost || ev[len(ev)-1].Reason != ReasonEnded {
t.Fatalf("got %v, want lost/ended last", types(ev))
}
}
if _, err := h.Open("r1", "sx", 5, "X", KindMember); err == nil {
t.Fatal("ended room accepted a stream")
}
}- [ ] Step 2: Run to verify they fail
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/calls/hub/ 2>&1 | grep -c 'undefined' Expected: non-zero (Join, Admit, Leave, Takeover, End undefined).
- [ ] Step 3: Implement admission
admission.go:
package hub
import "context"
func admittedCountLocked(r *room) int {
n := 0
for _, p := range r.peers {
if p.Admitted {
n++
}
}
return n
}
func waitingCountLocked(r *room) int {
n := 0
for _, p := range r.peers {
if p.Joined && !p.Admitted {
n++
}
}
return n
}
// rosterFor is the authoritative snapshot for one viewer: admitted peers
// with the viewer's pair generation, and the waiting queue only when the
// viewer is a member.
func (h *Hub) rosterFor(r *room, viewer *Participation) *Roster {
ro := &Roster{}
for _, p := range r.peers {
if !p.Admitted {
continue
}
rp := RosterPeer{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String()}
if p.Peer != viewer.Peer {
rp.Gen = h.genLocked(r, viewer.Peer, p.Peer)
}
ro.Peers = append(ro.Peers, rp)
}
if viewer.Kind == KindMember {
for _, p := range r.peers {
if p.Joined && !p.Admitted {
ro.Waiting = append(ro.Waiting, RosterPeer{Peer: p.Peer, Name: p.Name, Kind: "guest"})
}
}
}
return ro
}
// genLocked returns the pair's generation, starting it at one on first sight.
func (h *Hub) genLocked(r *room, a, b PeerID) uint64 {
k := keyFor(a, b)
if r.gens[k] == 0 {
r.gens[k] = 1
}
return r.gens[k]
}
// admitLocked is the one transition into the admitted state: cap under
// the lock, roster to the newcomer, peer to the others, knock-gone to
// members if they were waiting. Returns false when the room is full.
func (h *Hub) admitLocked(r *room, p *Participation) bool {
if admittedCountLocked(r) >= MaxAdmitted {
return false
}
wasWaiting := p.Joined && !p.Admitted
p.Admitted = true
p.Joined = true
p.JoinedAt = h.cfg.Now()
if p.sub != nil {
p.sub.send(Message{Type: TypeRoster, Roster: h.rosterFor(r, p)})
}
for _, q := range r.peers {
if q.Admitted && q.Peer != p.Peer && q.sub != nil {
q.sub.send(Message{Type: TypePeer, Peer: p.Peer, Name: p.Name, Payload: nil,
Roster: &Roster{Peers: []RosterPeer{{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String(), Gen: h.genLocked(r, q.Peer, p.Peer)}}}})
}
}
if wasWaiting {
h.broadcastLocked(r, Message{Type: TypeKnockGone, From: p.Peer}, admittedMembers)
}
return true
}
// Join is the entry. Standing is re-derived now (spec §2.5).
func (h *Hub) Join(ctx context.Context, sqid, session string, peer PeerID) error {
kind, ok, err := h.standing(ctx, sqid, session, peer)
if err != nil {
return err
}
h.mu.Lock()
defer h.mu.Unlock()
r := h.roomLocked(sqid)
p, found := h.findLocked(sqid, session, peer)
if !found {
return ErrUnknownPeer
}
if !ok {
h.endLocked(r, p, ReasonStanding)
return ErrNotYours
}
p.Kind = kind
if p.Joined {
return nil // idempotent
}
if kind == KindMember {
if !h.admitLocked(r, p) {
if p.sub != nil {
p.sub.send(Message{Type: TypeFull})
}
return ErrRoomFull
}
return nil
}
if waitingCountLocked(r) >= MaxWaiting {
if p.sub != nil {
p.sub.send(Message{Type: TypeFull})
}
return ErrTooManyWaiting
}
p.Joined = true
p.JoinedAt = h.cfg.Now()
h.broadcastLocked(r, Message{Type: TypeKnock, From: p.Peer, Name: p.Name}, admittedMembers)
return nil
}
// standing looks the peer up to find its person, then asks the app.
func (h *Hub) standing(ctx context.Context, sqid, session string, peer PeerID) (Kind, bool, error) {
h.mu.Lock()
p, found := h.findLocked(sqid, session, peer)
var personID int64
if found {
personID = p.PersonID
}
h.mu.Unlock()
if !found {
return 0, false, ErrUnknownPeer
}
return h.cfg.StandingFor(ctx, sqid, personID)
}
// Admit: the actor must be admitted with current member kind; the target
// waiting with entry authority now; the cap under the lock.
func (h *Hub) Admit(ctx context.Context, sqid, session string, actor, target PeerID) error {
actorKind, actorOK, err := h.standing(ctx, sqid, session, actor)
if err != nil {
return err
}
h.mu.Lock()
tp, found := h.rooms[sqid].peers[target]
var targetPerson int64
if found {
targetPerson = tp.PersonID
}
h.mu.Unlock()
if !found {
return ErrNotWaiting
}
targetKind, targetOK, err := h.cfg.StandingFor(ctx, sqid, targetPerson)
if err != nil {
return err
}
h.mu.Lock()
defer h.mu.Unlock()
r := h.roomLocked(sqid)
a, ok := h.findLocked(sqid, session, actor)
if !ok {
return ErrUnknownPeer
}
if !actorOK {
h.endLocked(r, a, ReasonStanding)
return ErrNotYours
}
a.Kind = actorKind
if !a.Admitted {
return ErrNotAdmitted
}
if a.Kind != KindMember {
return ErrNotMember
}
tp, ok = r.peers[target]
if !ok || !tp.Joined || tp.Admitted {
return ErrNotWaiting
}
if !targetOK {
wasAdmitted, wasWaiting := h.endLocked(r, tp, ReasonExpired)
h.announceDepartureLocked(r, tp, wasAdmitted, wasWaiting)
return nil
}
tp.Kind = targetKind
if !h.admitLocked(r, tp) {
if a.sub != nil {
a.sub.send(Message{Type: TypeFull})
}
}
return nil
}
func (h *Hub) Leave(sqid, session string, peer PeerID) error {
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
if !ok {
return ErrUnknownPeer
}
p, found := h.findLocked(sqid, session, peer)
if !found {
return ErrUnknownPeer
}
wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer)
delete(h.tombstones, p.Peer) // a leave is not a tombstoned ending
h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
return nil
}
// Remove is §3.5: the actor holds an admitted member participation in
// this room (kind re-derived), the target is any other participation.
func (h *Hub) Remove(ctx context.Context, sqid, session string, actor, target PeerID) error {
kind, ok, err := h.standing(ctx, sqid, session, actor)
if err != nil {
return err
}
h.mu.Lock()
defer h.mu.Unlock()
r := h.roomLocked(sqid)
a, found := h.findLocked(sqid, session, actor)
if !found || !ok {
return ErrUnknownPeer
}
a.Kind = kind
if !a.Admitted {
return ErrNotAdmitted
}
if a.Kind != KindMember {
return ErrNotMember
}
tp, found := r.peers[target]
if !found || tp.Peer == a.Peer {
return ErrUnknownPeer
}
wasAdmitted, wasWaiting := h.endLocked(r, tp, ReasonRemoved)
h.announceDepartureLocked(r, tp, wasAdmitted, wasWaiting)
return nil
}
// Takeover ends the session's live participation in the room with the
// state-dependent departure (spec §4.2), leaving a replaced tombstone.
func (h *Hub) Takeover(sqid, session string) (PeerID, error) {
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
if !ok {
return "", ErrUnknownPeer
}
id, ok := h.bySession[sqid][session]
if !ok {
return "", ErrUnknownPeer
}
p := r.peers[id]
wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonReplaced)
h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting)
return id, nil
}
// End is the meeting ending: ended to every stream, room dropped.
func (h *Hub) End(sqid string) {
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
if !ok {
r = h.roomLocked(sqid)
}
for _, p := range r.peers {
h.endLocked(r, p, ReasonEnded)
}
r.ended = true
delete(h.bySession, sqid)
}Extend Tick in hub.go: after the drops, for every joined participation call cfg.StandingFor (outside the lock: collect (room, peer, person) triples under the lock, call, then re-lock and apply) and apply the §2.4 table: member → none: end with ReasonStanding and announce; waiting guest → not ok: end with ReasonExpired and announce; kind changed on an admitted peer: set and send roster to admitted peers; waiting peer now member: admitLocked or full to it. Admitted guest with ok=false: nothing (ruling 14).
- [ ] Step 4: Run the tests
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/calls/hub/ -v 2>&1 | grep -E '^(=== RUN|--- (PASS|FAIL)|FAIL|ok)' | tail -20 Expected: all PASS.
- [ ] Step 5: Gate and commit
git add internal/calls/hub
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "hub: join, admit, leave, remove, takeover, end, the cap and the tick
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 3: Relay: targeted messages, seq and gen, replay ring, reset and pair, resume and gap
Files:
- Create:
internal/calls/hub/relay.go,internal/calls/hub/relay_test.go - Modify:
internal/calls/hub/hub.go(Resumeimplemented;deliverLockedrecords targeted messages into the ring)
Interfaces:
- Produces: ``
go func (h *Hub) Relay(sqid, session string, from PeerID, m Message) error // m.Type in offer/answer/candidate; from admitted; m.To admitted in this room; m.Gen == current gen else ErrStaleGen; delivers envelope {Type, From, Name, To, Seq, Gen, Payload} and rings it on the target func (h *Hub) Reset(sqid, session string, from, to PeerID) error // both admitted; gen++ under lock;pair {peer, gen}to both, ringed func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error)`` - Ring: newest last, capped at
RingSize; replay on resume = every ringed message withFrom == sender && Seq > last[sender], plus everypairwhoseGenexceeds the roster's for that peer; if the ring has dropped a message a sender sent abovelast(detected by the ring's oldestSeqfor that sender being greater thanlast[sender]+1while messages exist), emitgap {from}for that sender instead. Resumeanswers: unknown peer and no tombstone →ResumeError{Status: 404, Boot: h.boot}; tombstone →ResumeError{Status: 410, Reason}; peer bound to another session → 404 with boot (never leak); found → attach a newSub, clearresumeUntil; admitted:hellois the handler's, then the hub returnsRoster,Replay,Gap; waiting/unjoined:Admitted=false, nothing else.
- [ ] Step 1: Write the failing tests
package hub
import (
"context"
"encoding/json"
"testing"
"time"
)
func twoAdmitted(t *testing.T) (*Hub, *time.Time, OpenResult, OpenResult) {
t.Helper()
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
ctx := context.Background()
a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
h.Join(ctx, "r1", "sa", a.Peer)
b, _ := h.Open("r1", "sb", 2, "Bob", KindMember)
h.Join(ctx, "r1", "sb", b.Peer)
drain(a.Sub)
drain(b.Sub)
return h, &now, a, b
}
func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T) {
h, _, a, b := twoAdmitted(t)
ctx := context.Background()
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
drain(a.Sub)
drain(b.Sub)
err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1, From: "forged", Name: "Forged", Payload: json.RawMessage(`{"sdp":"x"}`)})
if err != nil {
t.Fatal(err)
}
bev := drain(b.Sub)
if len(bev) != 1 || bev[0].From != a.Peer || bev[0].Name != "Ann" || string(bev[0].Payload) != `{"sdp":"x"}` {
t.Fatalf("b got %+v", bev)
}
if ev := drain(a.Sub); len(ev) != 0 {
t.Fatalf("sender echoed %v", types(ev))
}
if ev := drain(g.Sub); len(ev) != 0 {
t.Fatalf("waiting guest received a relay: %v", types(ev))
}
if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: g.Peer, Seq: 2, Gen: 1}); err == nil {
t.Fatal("relay to a waiting peer accepted")
}
if err := h.Relay("r1", "sg", g.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrNotAdmitted {
t.Fatalf("waiting guest relaying: %v", err)
}
if err := h.Relay("r1", "sb", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer {
t.Fatalf("peer through another session: %v", err)
}
}
func TestStaleGenIsRefusedAndResetConverges(t *testing.T) {
h, _, a, b := twoAdmitted(t)
if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen {
t.Fatalf("gen 2 before any reset: %v", err)
}
if err := h.Reset("r1", "sa", a.Peer, b.Peer); err != nil {
t.Fatal(err)
}
if err := h.Reset("r1", "sb", b.Peer, a.Peer); err != nil { // simultaneous resets converge on the higher gen
t.Fatal(err)
}
aev, bev := drain(a.Sub), drain(b.Sub)
if len(aev) != 2 || aev[1].Type != TypePair || aev[1].Gen != 3 || aev[1].Peer != b.Peer {
t.Fatalf("a got %+v", aev)
}
if len(bev) != 2 || bev[1].Gen != 3 || bev[1].Peer != a.Peer {
t.Fatalf("b got %+v", bev)
}
if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1}); err != ErrStaleGen {
t.Fatal("old gen accepted after reset")
}
if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil {
t.Fatal(err)
}
}
func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) {
h, now, a, b := twoAdmitted(t)
ctx := context.Background()
for i := uint64(1); i <= 5; i++ {
h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)})
}
drain(a.Sub)
h.Detach("r1", a.Peer)
*now = now.Add(5 * time.Second)
res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 3})
if err != nil {
t.Fatal(err)
}
if !res.Admitted || res.Roster == nil || len(res.Replay) != 2 || res.Replay[0].Seq != 4 || res.Replay[1].Seq != 5 {
t.Fatalf("resume %+v", res)
}
// A waiting guest resumes and learns nothing.
g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
h.Join(ctx, "r1", "sg", g.Peer)
h.Detach("r1", g.Peer)
gres, err := h.Resume("r1", "sg", g.Peer, nil)
if err != nil {
t.Fatal(err)
}
if gres.Admitted || gres.Roster != nil || len(gres.Replay) != 0 {
t.Fatalf("waiting guest resume leaked %+v", gres)
}
}
func TestResumeGapWhenTheRingRolledOver(t *testing.T) {
h, now, a, b := twoAdmitted(t)
for i := uint64(1); i <= 40; i++ {
h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
}
drain(a.Sub)
h.Detach("r1", a.Peer)
*now = now.Add(time.Second)
res, _ := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2})
if len(res.Gap) != 1 || res.Gap[0] != b.Peer || len(res.Replay) != 0 {
t.Fatalf("expected a gap for b, got replay %d gap %v", len(res.Replay), res.Gap)
}
}
func TestResumeAnswers(t *testing.T) {
h, now, a, _ := twoAdmitted(t)
if _, err := h.Resume("r1", "sb", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 || err.(*ResumeError).Boot != h.Boot() {
t.Fatalf("resume through another session: %v", err)
}
h.Takeover("r1", "sa")
if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 410 || err.(*ResumeError).Reason != ReasonReplaced {
t.Fatalf("resume after takeover: %v", err)
}
*now = now.Add(TombstoneLife + time.Minute)
h.Tick(context.Background())
if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 {
t.Fatalf("resume after tombstone expiry: %v", err)
}
if _, err := h.Open("r1", "sa", 1, "Ann", KindMember); err != nil {
t.Fatalf("fresh stream after tombstone: %v", err)
}
}
func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) {
h, now, a, _ := twoAdmitted(t)
h.Detach("r1", a.Peer)
*now = now.Add(ResumeWindow + time.Second)
h.Tick(context.Background())
_, err := h.Resume("r1", "sa", a.Peer, nil)
re, ok := err.(*ResumeError)
if !ok || re.Status != 404 || re.Boot != h.Boot() {
t.Fatalf("lapsed resume: %v", err)
}
}- [ ] Step 2: Run to verify they fail
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/calls/hub/ 2>&1 | grep -c undefined Expected: non-zero.
- [ ] Step 3: Implement relay and resume
relay.go:
package hub
import "fmt"
func isTargeted(t string) bool { return t == TypeOffer || t == TypeAnswer || t == TypeCandidate }
// Relay delivers one targeted message. The envelope is rebuilt from the
// binding; the payload is never read.
func (h *Hub) Relay(sqid, session string, from PeerID, m Message) error {
if !isTargeted(m.Type) {
return fmt.Errorf("hub: %q is not a targeted type", m.Type)
}
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
if !ok {
return ErrUnknownPeer
}
p, found := h.findLocked(sqid, session, from)
if !found {
return ErrUnknownPeer
}
if !p.Admitted {
return ErrNotAdmitted
}
tp, ok := r.peers[m.To]
if !ok || !tp.Admitted {
return ErrUnknownPeer
}
if m.Gen != h.genLocked(r, p.Peer, tp.Peer) {
return ErrStaleGen
}
out := Message{Type: m.Type, From: p.Peer, Name: p.Name, To: tp.Peer, Seq: m.Seq, Gen: m.Gen, Payload: m.Payload}
h.ringLocked(tp, out)
if tp.sub != nil {
tp.sub.send(out)
}
return nil
}
func (h *Hub) ringLocked(p *Participation, m Message) {
p.ring = append(p.ring, m)
if len(p.ring) > RingSize {
p.ring = p.ring[len(p.ring)-RingSize:]
}
}
// Reset bumps the pair's generation and tells both sides.
func (h *Hub) Reset(sqid, session string, from, to PeerID) error {
h.mu.Lock()
defer h.mu.Unlock()
r, ok := h.rooms[sqid]
if !ok {
return ErrUnknownPeer
}
p, found := h.findLocked(sqid, session, from)
if !found {
return ErrUnknownPeer
}
if !p.Admitted {
return ErrNotAdmitted
}
tp, ok := r.peers[to]
if !ok || !tp.Admitted {
return ErrUnknownPeer
}
k := keyFor(p.Peer, tp.Peer)
h.genLocked(r, p.Peer, tp.Peer)
r.gens[k]++
gen := r.gens[k]
for _, pair := range [][2]*Participation{{p, tp}, {tp, p}} {
m := Message{Type: TypePair, Peer: pair[1].Peer, Gen: gen, To: pair[0].Peer}
h.ringLocked(pair[0], m)
if pair[0].sub != nil {
pair[0].sub.send(m)
}
}
return nil
}
type ResumeResult struct {
Sub *Sub
Admitted bool
Roster *Roster
Replay []Message
Gap []PeerID
}
type ResumeError struct {
Status int
Reason Reason
Boot string
}
func (e *ResumeError) Error() string { return fmt.Sprintf("hub: resume refused %d %s", e.Status, e.Reason) }
// Resume reattaches a stream to a participation in its window.
func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) {
h.mu.Lock()
defer h.mu.Unlock()
if t, ok := h.tombstones[peer]; ok {
return ResumeResult{}, &ResumeError{Status: 410, Reason: t.reason}
}
r, ok := h.rooms[sqid]
if !ok {
return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot}
}
p, found := h.findLocked(sqid, session, peer)
if !found {
return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot}
}
if p.sub != nil {
p.sub.terminate(Message{Type: TypeLost, Reason: ReasonReplaced}) // a second stream for the same peer replaces the first attachment only
}
p.sub = newSub()
p.resumeUntil = time.Time{}
res := ResumeResult{Sub: p.sub, Admitted: p.Admitted}
if !p.Admitted {
return res, nil
}
res.Roster = h.rosterFor(r, p)
// Replay: per sender, everything above last; a gap if the ring no
// longer holds the first message the browser has not seen.
oldest := map[PeerID]uint64{}
for _, m := range p.ring {
if isTargeted(m.Type) {
if o, ok := oldest[m.From]; !ok || m.Seq < o {
oldest[m.From] = m.Seq
}
}
}
gapped := map[PeerID]bool{}
for from, o := range oldest {
if o > last[from]+1 {
gapped[from] = true
res.Gap = append(res.Gap, from)
}
}
for _, m := range p.ring {
switch {
case isTargeted(m.Type):
if !gapped[m.From] && m.Seq > last[m.From] {
res.Replay = append(res.Replay, m)
}
case m.Type == TypePair:
if m.Gen > h.currentGenSeenLocked(res.Roster, m.Peer) {
res.Replay = append(res.Replay, m)
}
}
}
return res, nil
}
func (h *Hub) currentGenSeenLocked(ro *Roster, peer PeerID) uint64 {
for _, rp := range ro.Peers {
if rp.Peer == peer {
return rp.Gen
}
}
return 0
}Note: since the roster already carries the current generation for every admitted peer, a pair in the ring is only replayed when its Gen exceeds what the roster says, which by construction never happens for a current roster; keep the branch for correctness and let the browser also reconcile from the roster (spec §4.6). Add "time" to the imports.
- [ ] Step 4: Run the tests, gate, commit
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/calls/hub/ -race 2>&1 | tail -3 then make ci.
git add internal/calls/hub
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "hub: targeted relay with seq and gen, replay ring, reset and pair, resume and gap
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 4: The stream: SSE, hello with boot, heartbeat and TTL, resume answers
Files:
- Create:
internal/calls/stream.go,internal/calls/participation.go - Modify:
internal/calls/app.go(hub field, tick goroutine, routes),internal/callstest/harness_test.go(SSE client helper) - Test:
internal/callstest/stream_test.go
Interfaces:
- Produces:
app.hub *hub.Hubconstructed inbuildwithStandingForadaptingstandingFor(look up the meeting by sqid and the person by id;KindMemberwhenStandingMember,KindGuestwhenStandingGuest,ok=falsewhen none or the meeting ended); a fifteen-second tick goroutine started byApponly, plusTestHandle.TickForTest();GET /m/{sqid}/streaminside therequirePersongroup; SSE eventsevent: <type>\ndata: <json>\n\n;hello {peer, boot}first; heartbeat every 15 s as a comment line: hb\n\n; TTL five minutes then the handler returns (the browser resumes);Detachon any exit; resume via?peer=<id>&last=<json>; answers per spec §4.5 with JSON bodies; test hooksTestHandle.SetStreamTimings(heartbeat, ttl time.Duration). - Harness:
func (c *client) stream(target string) *sseReaderwithnext(timeout) (event string, data []byte, ok bool),close(), andhello() (peer, boot string).
- [ ] Step 1: Write the failing tests
package callstest
import (
"encoding/json"
"net/http"
"testing"
"time"
)
func TestStreamSaysHelloWithPeerAndBootAndRefusesTheSecondFreshStream(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
c := x.browser(t)
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
defer s.close()
peer, boot := s.hello()
if len(peer) != 32 || len(boot) != 32 {
t.Fatalf("hello peer=%q boot=%q", peer, boot)
}
rec := c.get("/m/" + sqid + "/stream")
if rec.Code != http.StatusConflict {
t.Fatalf("second fresh stream: %d, want 409", rec.Code)
}
var body struct{ Peer string }
json.Unmarshal(rec.Body.Bytes(), &body)
if body.Peer != peer {
t.Fatalf("409 names %q, want %q", body.Peer, peer)
}
}
func TestStreamNeedsStandingAndAResumeNeedsTheSession(t *testing.T) {
x, sqid, tok := memberWithMeeting(t)
stranger := x.browser(t)
// stranger has no standing: no grant, not a member
stranger2 := newHarness(t).browser(t)
_ = stranger2
if rec := stranger.get("/m/" + sqid + "/stream"); rec.Code != http.StatusSeeOther {
t.Fatalf("signed-out stream: %d", rec.Code)
}
c := x.browser(t)
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
peer, _ := s.hello()
s.close()
other := x.browser(t)
other.get("/j/" + tok)
other.signIn("guest@example.com")
rec := other.get("/m/" + sqid + "/stream?peer=" + peer)
if rec.Code != http.StatusNotFound {
t.Fatalf("resume through another session: %d, want 404", rec.Code)
}
}
func TestResumeAfterTheStreamClosesKeepsAdmission(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
c := x.browser(t)
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
peer, boot := s.hello()
c.signal(sqid, peer, `{"type":"join"}`)
if ev, _, ok := s.next(time.Second); !ok || ev != "roster" {
t.Fatalf("after join: %s ok=%v", ev, ok)
}
s.close()
s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer)
defer s2.close()
p2, b2 := s2.hello()
if p2 != peer || b2 != boot {
t.Fatalf("resume hello %s %s", p2, b2)
}
if ev, _, ok := s2.next(time.Second); !ok || ev != "roster" {
t.Fatalf("resume of an admitted peer should send roster, got %s", ev)
}
}
func TestTTLClosesTheStreamAndResumeWorksWithinTheWindow(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
x.ctl.SetStreamTimings(50*time.Millisecond, 300*time.Millisecond)
c := x.browser(t)
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
peer, _ := s.hello()
c.signal(sqid, peer, `{"type":"join"}`)
s.next(time.Second)
if !s.waitClosed(2 * time.Second) {
t.Fatal("stream did not close at the TTL")
}
s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer)
defer s2.close()
if p, _ := s2.hello(); p != peer {
t.Fatalf("resume after TTL: %s", p)
}
}
func TestLapsedResumeIs404WithTheSameBootAndAFreshJoinWorks(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
c := x.browser(t)
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
peer, boot := s.hello()
c.signal(sqid, peer, `{"type":"join"}`)
s.next(time.Second)
s.close()
x.advance(31 * time.Second)
x.ctl.TickForTest()
rec := c.get("/m/" + sqid + "/stream?peer=" + peer)
var body struct{ Boot string }
json.Unmarshal(rec.Body.Bytes(), &body)
if rec.Code != http.StatusNotFound || body.Boot != boot {
t.Fatalf("lapsed resume: %d %+v", rec.Code, body)
}
s3 := c.stream("/m/" + sqid + "/stream")
defer s3.close()
if p, _ := s3.hello(); p == "" || p == peer {
t.Fatalf("fresh join after lapse: %q", p)
}
}The harness additions (harness_test.go): an sseReader that performs the request with httptest.NewRecorder cannot stream; use a real httptest.Server wrapping x.h for stream tests. Add to harness: srv *httptest.Server started lazily by x.server(); client.stream issues a real HTTP GET with the client's cookies and Origin, reads lines in a goroutine, parses event:/data: pairs into a channel; next(timeout) receives; hello() reads the first event and unmarshals peer and boot; waitClosed waits for EOF; close() cancels the request. client.signal(sqid, peer, body string) *http.Response POSTs JSON {"peer": peer, ...body fields} with Content-Type: application/json and Origin through the same server. Because the CSRF middleware checks Origin against testOrigin (http://app.test), the httptest.Server's origin differs; set the harness Config.Origin to the server's URL when the server is started (start it in newHarness before building the app, so Origin is right from the first request) and make testOrigin a per-harness value x.origin. Update linkInPage to match x.origin. This is a harness refactor the first step of this task does before the tests.
- [ ] Step 2: Implement
participation.go:
package meet
import (
"context"
"amadan.net/oficina/meet/internal/calls/hub"
)
// hubStanding adapts standingFor for the hub: kind and whether the
// person still holds entry authority for the room.
func (a *app) hubStanding(ctx context.Context, sqid string, personID int64) (hub.Kind, bool, error) {
m, ok, err := a.store.meetingBySqid(ctx, sqid)
if err != nil || !ok || m.EndedAt != nil {
return hub.KindGuest, false, err
}
p, err := a.store.personByID(ctx, personID)
if err != nil {
return hub.KindGuest, false, err
}
st, err := a.standingFor(ctx, p, m)
if err != nil {
return hub.KindGuest, false, err
}
switch st {
case StandingMember:
return hub.KindMember, true, nil
case StandingGuest:
return hub.KindGuest, true, nil
}
return hub.KindGuest, false, nil
}Add store.personByID(ctx, id int64) (Person, error).
stream.go:
package meet
import (
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/go-chi/chi/v5"
"amadan.net/oficina/meet/internal/calls/hub"
"amadan.net/rastrillo/rastrillo/sessions"
)
// sessionKey is the session's identity for the hub: the cookie token
// hash is not exposed by rastrillo, so the subject plus a per-boot salt
// stands in. Two tabs of one browser share it; two people never do.
func (a *app) sessionKey(r *http.Request) string {
s, _ := sessions.Current(r)
return s.Subject + "|" + s.At.UTC().Format(time.RFC3339Nano)
}
func writeEvent(w http.ResponseWriter, flusher http.Flusher, m hub.Message) error {
b, err := json.Marshal(m)
if err != nil {
return err
}
if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", m.Type, b); err != nil {
return err
}
flusher.Flush()
return nil
}
// stream is GET /m/{sqid}/stream: fresh (standing) or resume (participation).
func (a *app) stream(w http.ResponseWriter, r *http.Request) {
sqid := chi.URLParam(r, "sqid")
p, _ := currentPerson(r)
session := a.sessionKey(r)
flusher, ok := w.(http.Flusher)
if !ok {
ErrorPage(w, r, http.StatusInternalServerError, "")
return
}
var sub *hub.Sub
var peer hub.PeerID
var first []hub.Message
if resume := r.URL.Query().Get("peer"); resume != "" {
var last map[hub.PeerID]uint64
if raw := r.URL.Query().Get("last"); raw != "" {
if err := json.Unmarshal([]byte(raw), &last); err != nil {
http.Error(w, `{"error":"bad last"}`, http.StatusBadRequest)
return
}
}
res, err := a.hub.Resume(sqid, session, hub.PeerID(resume), last)
if err != nil {
re, ok := err.(*hub.ResumeError)
if !ok {
ErrorPage(w, r, http.StatusInternalServerError, "")
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(re.Status)
json.NewEncoder(w).Encode(map[string]any{"reason": re.Reason, "boot": re.Boot})
return
}
sub, peer = res.Sub, hub.PeerID(resume)
first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot()})
if res.Admitted {
first = append(first, hub.Message{Type: hub.TypeRoster, Roster: res.Roster})
first = append(first, res.Replay...)
for _, from := range res.Gap {
first = append(first, hub.Message{Type: hub.TypeGap, From: from})
}
} else {
first = append(first, hub.Message{Type: hub.TypeState, Waiting: true})
}
} else {
m, ok, err := a.store.meetingBySqid(r.Context(), sqid)
if err != nil || !ok || m.EndedAt != nil {
ErrorPage(w, r, http.StatusNotFound, "")
return
}
st, err := a.standingFor(r.Context(), p, m)
if err != nil || st == StandingNone {
ErrorPage(w, r, http.StatusNotFound, "")
return
}
kind := hub.KindGuest
if st == StandingMember {
kind = hub.KindMember
}
res, err := a.hub.Open(sqid, session, p.ID, p.Name, kind)
switch err {
case nil:
case hub.ErrSessionHasPeer:
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusConflict)
json.NewEncoder(w).Encode(map[string]any{"peer": res.Existing})
return
case hub.ErrRoomFull, hub.ErrPersonStreams:
http.Error(w, `{"error":"full"}`, http.StatusServiceUnavailable)
return
default:
ErrorPage(w, r, http.StatusNotFound, "")
return
}
sub, peer = res.Sub, res.Peer
first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot()})
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-store")
w.Header().Set("X-Accel-Buffering", "no")
w.WriteHeader(http.StatusOK)
defer a.hub.Detach(sqid, peer)
rc := http.NewResponseController(w)
for _, m := range first {
if err := writeEvent(w, flusher, m); err != nil {
return
}
}
heartbeat := time.NewTicker(a.streamHeartbeat)
defer heartbeat.Stop()
ttl := time.NewTimer(a.streamTTL)
defer ttl.Stop()
for {
_ = rc.SetWriteDeadline(time.Now().Add(30 * time.Second))
select {
case <-r.Context().Done():
return
case <-ttl.C:
return
case <-heartbeat.C:
if _, err := fmt.Fprint(w, ": hb\n\n"); err != nil {
return
}
flusher.Flush()
case m, ok := <-sub.C():
if !ok {
return
}
if err := writeEvent(w, flusher, m); err != nil {
return
}
if m.Type == hub.TypeLost || m.Type == hub.TypeEnded {
return
}
}
}
}In app.go: fields hub *hub.Hub, streamHeartbeat, streamTTL (defaults 15 s and 5 min); construct the hub in build with Now: st.now, Logger, StandingFor: a.hubStanding; App starts go a.tickForever(ctx) (15 s ticker calling a.hub.Tick(ctx)); TestHandle.TickForTest() and SetStreamTimings. Route inside the requirePerson group: r.Get("/m/{sqid}/stream", a.stream). Also wire endMeeting to call a.hub.End(m.Sqid) after the store update.
- [ ] Step 3: Run tests, gate, commit
Run: GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/callstest/ -run 'Stream|Resume|TTL|Lapsed' -v 2>&1 | tail -15 then make ci.
git add internal/calls internal/callstest
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "stream: SSE per peer with hello and boot, heartbeat, TTL, resume answers
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 5: Signal: POST with limits, rate limit, dispatch, refusal logging, captured-log test
Files:
- Create:
internal/calls/signal.go,internal/calls/ratelimit.go - Test:
internal/callstest/signal_test.go
Interfaces:
- Produces:
POST /m/{sqid}/signalinside therequirePersongroup; body{"peer": "<id>", "type": "...", "to": "...", "seq": n, "gen": n, "payload": {...}}; refused above 64 KB with 413; 429 withRetry-After: 1above thirty per second (burst sixty) per peer, a token bucket keyed by peer id ina.signalLimiter; dispatch:join→hub.Join(standing re-derived by the hub);admit→hub.Admit(actor, payload.peer);leave→hub.Leave;offer|answer|candidate→hub.Relay;reset→hub.Reset(from, payload.to);chat|share→ 400 in this slice (slice 4); unknown → 400. Error mapping:ErrUnknownPeer/ErrNotYours→ 404;ErrNotAdmitted/ErrNotMember/ErrNotWaiting→ 403;ErrStaleGen→ 409{"error":"stale","gen":<current>};ErrRoomFull/ErrTooManyWaiting→ 200 (thefullevent carries the outcome). Every refusal logstype,room,peer,outcomeand nothing else; an accepted message logs nothing.
- [ ] Step 1: Write the failing tests
package callstest
import (
"bytes"
"log/slog"
"net/http"
"strings"
"testing"
"time"
)
func TestSignalNeedsAPeerBoundToThisSessionAndRoom(t *testing.T) {
x, sqid, tok := memberWithMeeting(t)
c := x.browser(t)
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
defer s.close()
peer, _ := s.hello()
if resp := c.signal(sqid, "0123456789abcdef0123456789abcdef", `{"type":"join"}`); resp.StatusCode != http.StatusNotFound {
t.Fatalf("foreign peer: %d", resp.StatusCode)
}
other := x.browser(t)
other.get("/j/" + tok)
other.signIn("guest@example.com")
if resp := other.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound {
t.Fatalf("peer through another session: %d", resp.StatusCode)
}
if resp := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
t.Fatalf("own join: %d", resp.StatusCode)
}
}
func TestSignalLimits(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
c := x.browser(t)
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
defer s.close()
peer, _ := s.hello()
c.signal(sqid, peer, `{"type":"join"}`)
big := `{"type":"candidate","to":"` + peer + `","seq":1,"gen":1,"payload":{"x":"` + strings.Repeat("a", 70*1024) + `"}}`
if resp := c.signal(sqid, peer, big); resp.StatusCode != http.StatusRequestEntityTooLarge {
t.Fatalf("64 KB: %d", resp.StatusCode)
}
var got429 bool
for i := 0; i < 100; i++ {
if resp := c.signal(sqid, peer, `{"type":"leave"}`); resp.StatusCode == http.StatusTooManyRequests {
if resp.Header.Get("Retry-After") == "" {
t.Fatal("429 without Retry-After")
}
got429 = true
break
}
}
if !got429 {
t.Fatal("100 messages in a burst never hit the rate limit")
}
}
func TestSignalStaleGenAnswers409WithTheCurrentGen(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
a := x.browser(t)
a.signIn("paul@example.com")
sa := a.stream("/m/" + sqid + "/stream")
defer sa.close()
pa, _ := sa.hello()
a.signal(sqid, pa, `{"type":"join"}`)
b := newMemberBrowser(t, x, "paul@rslw.com")
sb := b.stream("/m/" + sqid + "/stream")
defer sb.close()
pb, _ := sb.hello()
b.signal(sqid, pb, `{"type":"join"}`)
resp := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":1,"gen":7,"payload":{}}`)
if resp.StatusCode != http.StatusConflict {
t.Fatalf("stale gen: %d", resp.StatusCode)
}
}
// The §0.1 and §4.4 promise: no payload, name, title or token in any log
// line, on accepted and refused messages and on an error path.
func TestSignallingLogsCarryNoContent(t *testing.T) {
var buf bytes.Buffer
x := newHarnessWithLogger(t, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com", "paul@rslw.com")
c := x.browser(t)
c.signIn("paul@example.com")
sqid, tok := createMeeting(t, c, "Secret title", "")
s := c.stream("/m/" + sqid + "/stream")
defer s.close()
peer, _ := s.hello()
c.signal(sqid, peer, `{"type":"join"}`)
b := newMemberBrowser(t, x, "paul@rslw.com")
sb := b.stream("/m/" + sqid + "/stream")
defer sb.close()
pb, _ := sb.hello()
b.signal(sqid, pb, `{"type":"join"}`)
c.signal(sqid, peer, `{"type":"offer","to":"`+pb+`","seq":1,"gen":1,"payload":{"sdp":"v=0 SDPSECRET a=candidate 203.0.113.9"}}`)
c.signal(sqid, peer, `{"type":"candidate","to":"`+pb+`","seq":2,"gen":1,"payload":{"candidate":"candidate:1 1 udp 2 203.0.113.9 5000 typ host"}}`)
c.signal(sqid, peer, `{"type":"offer","to":"`+pb+`","seq":3,"gen":9,"payload":{"sdp":"STALESECRET"}}`)
c.signal(sqid, peer, `{"type":"bogus"}`)
x.advance(time.Minute)
x.ctl.TickForTest()
logs := buf.String()
for _, needle := range []string{"Secret title", "SDPSECRET", "203.0.113.9", "STALESECRET", tok, "paul@example.com"} {
if strings.Contains(logs, needle) {
t.Errorf("log carries %q", needle)
}
}
}Add newHarnessWithLogger(t, logger, members...) and newMemberBrowser(t, x, address) *client (a browser that signs in as an allowlisted member) to the harness.
- [ ] Step 2: Implement
ratelimit.go: a per-key token bucket, rate 30/s, burst 60, allow(key string, now time.Time) (ok bool, retryAfter time.Duration), entries evicted when idle for a minute (sweep on each call every thousand calls or on the tick).
signal.go:
package meet
import (
"encoding/json"
"errors"
"io"
"net/http"
"strconv"
"github.com/go-chi/chi/v5"
"amadan.net/oficina/meet/internal/calls/hub"
)
const maxSignalBody = 64 * 1024
type signalBody struct {
Peer hub.PeerID `json:"peer"`
Type string `json:"type"`
To hub.PeerID `json:"to,omitempty"`
Seq uint64 `json:"seq,omitempty"`
Gen uint64 `json:"gen,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
}
func (a *app) signal(w http.ResponseWriter, r *http.Request) {
sqid := chi.URLParam(r, "sqid")
session := a.sessionKey(r)
r.Body = http.MaxBytesReader(w, r.Body, maxSignalBody)
raw, err := io.ReadAll(r.Body)
if err != nil {
var tooBig *http.MaxBytesError
if errors.As(err, &tooBig) {
a.refuse(w, sqid, "", "", http.StatusRequestEntityTooLarge, "too-large")
return
}
a.refuse(w, sqid, "", "", http.StatusBadRequest, "read")
return
}
var body signalBody
if err := json.Unmarshal(raw, &body); err != nil || body.Peer == "" || body.Type == "" {
a.refuse(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "shape")
return
}
if ok, retry := a.signalLimiter.allow(string(body.Peer), a.store.now()); !ok {
w.Header().Set("Retry-After", strconv.Itoa(int(retry.Seconds())+1))
a.refuse(w, sqid, body.Peer, body.Type, http.StatusTooManyRequests, "rate")
return
}
if _, ok := a.hub.Find(sqid, session, body.Peer); !ok {
a.refuse(w, sqid, body.Peer, body.Type, http.StatusNotFound, "binding")
return
}
var herr error
switch body.Type {
case hub.TypeJoin:
herr = a.hub.Join(r.Context(), sqid, session, body.Peer)
case hub.TypeAdmit:
var p struct{ Peer hub.PeerID `json:"peer"` }
json.Unmarshal(body.Payload, &p)
herr = a.hub.Admit(r.Context(), sqid, session, body.Peer, p.Peer)
case hub.TypeLeave:
herr = a.hub.Leave(sqid, session, body.Peer)
case hub.TypeOffer, hub.TypeAnswer, hub.TypeCandidate:
herr = a.hub.Relay(sqid, session, body.Peer, hub.Message{Type: body.Type, To: body.To, Seq: body.Seq, Gen: body.Gen, Payload: body.Payload})
case hub.TypeReset:
herr = a.hub.Reset(sqid, session, body.Peer, body.To)
default:
a.refuse(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type")
return
}
switch {
case herr == nil, errors.Is(herr, hub.ErrRoomFull), errors.Is(herr, hub.ErrTooManyWaiting):
w.WriteHeader(http.StatusOK)
case errors.Is(herr, hub.ErrStaleGen):
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusConflict)
json.NewEncoder(w).Encode(map[string]any{"error": "stale"})
a.logRefusal(sqid, body.Peer, body.Type, "stale-gen")
case errors.Is(herr, hub.ErrNotAdmitted), errors.Is(herr, hub.ErrNotMember), errors.Is(herr, hub.ErrNotWaiting):
a.refuse(w, sqid, body.Peer, body.Type, http.StatusForbidden, "authority")
case errors.Is(herr, hub.ErrUnknownPeer), errors.Is(herr, hub.ErrNotYours):
a.refuse(w, sqid, body.Peer, body.Type, http.StatusNotFound, "binding")
default:
a.logger.Error("calls: signal", "room", sqid, "type", body.Type, "err", herr.Error())
w.WriteHeader(http.StatusInternalServerError)
}
}
func (a *app) refuse(w http.ResponseWriter, sqid string, peer hub.PeerID, typ string, status int, outcome string) {
a.logRefusal(sqid, peer, typ, outcome)
w.WriteHeader(status)
}
// logRefusal is the only signalling log line, and it carries no content.
func (a *app) logRefusal(sqid string, peer hub.PeerID, typ, outcome string) {
a.logger.Warn("calls: signal refused", "room", sqid, "peer", string(peer), "type", typ, "outcome", outcome)
}Route: r.Post("/m/{sqid}/signal", a.signal) in the requirePerson group. a.signalLimiter = newRateLimiter(30, 60) in build. Note the hub.Join error for a full room is a 200 because the event is the answer; ErrNotYours on join (lost standing) is 404.
- [ ] Step 3: Run tests, gate, commit
git add internal/calls internal/callstest
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "signal: POST with body and rate limits, dispatch, refusal logging without content
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 6: Room page, removal and takeover confirm pages, page authorization, AGENTS.md exception
Files:
- Create:
internal/calls/handlers_room.go,internal/calls/templates/room.html - Modify:
internal/calls/handlers_join.go(holding authorized by standing or participation),internal/calls/templates/join.html,internal/calls/app.go(routes),AGENTS.md - Test:
internal/callstest/room_test.go
Interfaces:
- Produces:
GET /m/{sqid}(room page; standing or live participation; scriptless output is the holding header, the paragraph, and a link to the holding screen; scripted output is the grid container, controls, admission queue, all rendered byroom.js);GET /m/{sqid}/remove/{peer}andPOST(actor must hold an admitted member participation, checked at execution throughhub.Remove);GET /m/{sqid}/takeoverandPOST(standing and a live participation for this session,hub.Takeover, then 303 to the room); page data for the room:Sqid, Title, Starts, Name, Standing, IsMember, Floor, Boot; adata-*block on the room page carryingsqid,boot,self-name,is-member,peer(empty) forroom.js. AGENTS.md"Screens work with JavaScript disabled" line becomes: "Screens work with JavaScript disabled, with one exception: browser media participation (the holding screen's camera preview and join button, and the call itself in the room) needs script; both pages render their title, standing, floor notice and navigation without it."
- [ ] Step 1: Write the failing tests
package callstest
import (
"net/http"
"net/url"
"strings"
"testing"
)
func TestRoomPageIsScriptlessHonestAndStandingGated(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
c := x.browser(t)
c.signIn("paul@example.com")
rec := c.get("/m/" + sqid)
body := rec.Body.String()
if rec.Code != http.StatusOK || !strings.Contains(body, "Board meeting") || !strings.Contains(body, `data-sqid="`+sqid+`"`) || !strings.Contains(body, "/m/"+sqid+"/join") {
t.Fatalf("room page: %d\n%s", rec.Code, body)
}
if !strings.Contains(body, "<noscript>") || !strings.Contains(body, "needs JavaScript") {
t.Fatal("room page lacks its scriptless paragraph")
}
stranger := newHarness(t).browser(t)
stranger.signIn("nobody@example.com")
if rec := stranger.get("/m/" + sqid); rec.Code != http.StatusNotFound {
t.Fatalf("no standing: %d", rec.Code)
}
}
func TestRemoveNeedsAConfirmPageAndAnAdmittedMemberActor(t *testing.T) {
x, sqid, tok := memberWithMeeting(t)
m := x.browser(t)
m.signIn("paul@example.com")
sm := m.stream("/m/" + sqid + "/stream")
defer sm.close()
pm, _ := sm.hello()
m.signal(sqid, pm, `{"type":"join"}`)
g := x.browser(t)
g.get("/j/" + tok)
g.signIn("guest@example.com")
sg := g.stream("/m/" + sqid + "/stream")
defer sg.close()
pg, _ := sg.hello()
g.signal(sqid, pg, `{"type":"join"}`)
if rec := m.get("/m/" + sqid + "/remove/" + pg); rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "Remove") {
t.Fatalf("confirm page: %d", rec.Code)
}
if rec := g.post("/m/"+sqid+"/remove/"+pm, nil); rec.Code != http.StatusNotFound {
t.Fatalf("waiting guest removing the member: %d", rec.Code)
}
if rec := m.post("/m/"+sqid+"/remove/"+pg, nil); rec.Code != http.StatusSeeOther {
t.Fatalf("member removing: %d", rec.Code)
}
ev, data, ok := sg.nextOf("lost", 2)
if !ok || !strings.Contains(string(data), `"removed"`) {
t.Fatalf("removed guest got %s %s", ev, data)
}
}
func TestTakeoverIsAConfirmedPostAndAGetEndsNothing(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
m := x.browser(t)
m.signIn("paul@example.com")
s1 := m.stream("/m/" + sqid + "/stream")
defer s1.close()
p1, _ := s1.hello()
m.signal(sqid, p1, `{"type":"join"}`)
if rec := m.get("/m/" + sqid + "/stream?takeover=1"); rec.Code != http.StatusConflict {
t.Fatalf("a GET with a query parameter must not take over: %d", rec.Code)
}
if rec := m.get("/m/" + sqid + "/takeover"); rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "another tab") {
t.Fatalf("takeover confirm: %d", rec.Code)
}
req := httptestPost("/m/"+sqid+"/takeover", url.Values{})
req.Header.Set("Origin", "https://evil.example")
if rec := m.do(req); rec.Code != http.StatusForbidden {
t.Fatalf("cross-origin takeover: %d", rec.Code)
}
if rec := m.post("/m/"+sqid+"/takeover", nil); rec.Code != http.StatusSeeOther {
t.Fatalf("takeover: %d", rec.Code)
}
if ev, data, ok := s1.nextOf("lost", 2); !ok || !strings.Contains(string(data), `"replaced"`) {
t.Fatalf("old stream got %s %s", ev, data)
}
if rec := m.get("/m/" + sqid + "/stream?peer=" + p1); rec.Code != http.StatusGone {
t.Fatalf("resume of the replaced peer: %d, want 410", rec.Code)
}
s2 := m.stream("/m/" + sqid + "/stream")
defer s2.close()
if p2, _ := s2.hello(); p2 == "" || p2 == p1 {
t.Fatalf("fresh stream after takeover: %q", p2)
}
}
func TestEndingTheMeetingEndsTheRoom(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
m := x.browser(t)
m.signIn("paul@example.com")
s := m.stream("/m/" + sqid + "/stream")
defer s.close()
p, _ := s.hello()
m.signal(sqid, p, `{"type":"join"}`)
m.post("/meetings/"+sqid+"/end", nil)
if ev, data, ok := s.nextOf("lost", 2); !ok || !strings.Contains(string(data), `"ended"`) {
t.Fatalf("got %s %s", ev, data)
}
}sseReader.nextOf(event string, timeoutSeconds int) skips events until the named one arrives.
- [ ] Step 2: Implement
handlers_room.go:
package meet
import (
"net/http"
"github.com/go-chi/chi/v5"
"amadan.net/oficina/meet/internal/calls/hub"
)
// roomAccess is §2.5's row for the two pages: standing, or a live
// participation for this session in this room.
func (a *app) roomAccess(w http.ResponseWriter, r *http.Request) (Meeting, Person, Standing, bool) {
sqid := chi.URLParam(r, "sqid")
m, ok, err := a.store.meetingBySqid(r.Context(), sqid)
if err != nil || !ok || m.EndedAt != nil {
ErrorPage(w, r, http.StatusNotFound, "")
return Meeting{}, Person{}, StandingNone, false
}
p, _ := currentPerson(r)
st, err := a.standingFor(r.Context(), p, m)
if err != nil {
ErrorPage(w, r, http.StatusInternalServerError, "")
return Meeting{}, Person{}, StandingNone, false
}
if st == StandingNone {
if _, live := a.hub.LiveForSession(sqid, a.sessionKey(r)); !live {
ErrorPage(w, r, http.StatusNotFound, "")
return Meeting{}, Person{}, StandingNone, false
}
st = StandingGuest
}
return m, p, st, true
}
func (a *app) room(w http.ResponseWriter, r *http.Request) {
m, p, st, ok := a.roomAccess(w, r)
if !ok {
return
}
render(w, "room", a.page(r, map[string]any{
"Sqid": m.Sqid, "Title": a.store.title(m), "Starts": a.fmtStart(m.StartsAt),
"Name": p.Name, "Standing": st.String(), "IsMember": st == StandingMember,
"Floor": !a.cfg.RelayConfigured, "Boot": a.hub.Boot(),
}))
}
func (a *app) confirmRemove(w http.ResponseWriter, r *http.Request) {
m, _, st, ok := a.roomAccess(w, r)
if !ok {
return
}
if st != StandingMember {
ErrorPage(w, r, http.StatusNotFound, "")
return
}
target := chi.URLParam(r, "peer")
render(w, "confirm", a.page(r, map[string]any{
"Title": "Remove this participant?", "Body": "They will be disconnected from \"" + a.store.title(m) + "\" and can only come back through the meeting link.",
"Action": "/m/" + m.Sqid + "/remove/" + target, "Label": "Remove", "CancelHref": "/m/" + m.Sqid, "Sqid": m.Sqid,
}))
}
func (a *app) remove(w http.ResponseWriter, r *http.Request) {
m, _, _, ok := a.roomAccess(w, r)
if !ok {
return
}
session := a.sessionKey(r)
actor, live := a.hub.LiveForSession(m.Sqid, session)
if !live {
ErrorPage(w, r, http.StatusNotFound, "")
return
}
if err := a.hub.Remove(r.Context(), m.Sqid, session, actor, hub.PeerID(chi.URLParam(r, "peer"))); err != nil {
ErrorPage(w, r, http.StatusNotFound, "")
return
}
http.Redirect(w, r, "/m/"+m.Sqid, http.StatusSeeOther)
}
func (a *app) confirmTakeover(w http.ResponseWriter, r *http.Request) {
m, _, _, ok := a.roomAccess(w, r)
if !ok {
return
}
render(w, "confirm", a.page(r, map[string]any{
"Title": "This meeting is open in another tab", "Body": "Use this tab instead? The other tab will leave the call.",
"Action": "/m/" + m.Sqid + "/takeover", "Label": "Use this tab", "CancelHref": "/m/" + m.Sqid + "/join", "Sqid": m.Sqid,
}))
}
func (a *app) takeover(w http.ResponseWriter, r *http.Request) {
m, _, _, ok := a.roomAccess(w, r)
if !ok {
return
}
if _, err := a.hub.Takeover(m.Sqid, a.sessionKey(r)); err != nil {
ErrorPage(w, r, http.StatusNotFound, "")
return
}
http.Redirect(w, r, "/m/"+m.Sqid, http.StatusSeeOther)
}Routes in the requirePerson group: r.Get("/m/{sqid}", a.room), r.Get("/m/{sqid}/remove/{peer}", a.confirmRemove), r.Post("/m/{sqid}/remove/{peer}", a.remove), r.Get("/m/{sqid}/takeover", a.confirmTakeover), r.Post("/m/{sqid}/takeover", a.takeover). Change holding to use roomAccess too, and give join.html the data-* block and the scripted containers (Task 7 fills them). room.html:
{{define "title"}}{{.Title}}{{end}}
{{define "content"}}
{{if .IsMember}}{{template "back-nav" dict "Href" "/" "Label" "Meetings"}}{{end}}
{{template "page-header" dict "Title" .Title}}
{{if .Starts}}<p>Starts {{.Starts}}.</p>{{end}}
<p>You are {{.Name}}, in the call as a {{.Standing}}.</p>
{{if .Floor}}<p rst-notice>This team's Meet has no relay yet. Calls usually connect only between people on the same network.</p>{{end}}
<noscript><p>The call needs JavaScript to reach your camera and the other participants; everything else on this page works without it. <a href="/m/{{.Sqid}}/join">Back to the holding screen</a>.</p></noscript>
<div id="room" data-sqid="{{.Sqid}}" data-boot="{{.Boot}}" data-self-name="{{.Name}}" data-is-member="{{if .IsMember}}1{{end}}" hidden>
<div id="tiles" class="meet-grid"></div>
<div id="controls" class="meet-controls">
<button type="button" id="mute" rst-btn>Mute</button>
<button type="button" id="camera" rst-btn>Camera off</button>
<button type="button" id="leave" rst-btn="danger">Leave</button>
</div>
<div id="queue" class="meet-queue" hidden><h2>Waiting to join</h2><ul id="knocks"></ul></div>
<p id="status" class="meet-status" role="status"></p>
</div>
{{if .IsMember}}<p><a href="/meetings/{{.Sqid}}/end" target="_blank" rel="noopener">End the meeting</a></p>{{end}}
<link rel="stylesheet" href="{{asset "static/calls.css"}}">
<script defer src="{{asset "static/room.js"}}"></script>
{{end}}Add "room" to pageNames, create static/calls.css with the grid, tile, controls and queue rules (small, rst-* tokens for colour), and a placeholder static/room.js that reveals #room (Task 8 replaces it). AGENTS.md line as specified above.
- [ ] Step 3: Run tests, gate, commit
git add internal AGENTS.md
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "room: page, removal and takeover confirm pages, participation-or-standing access, the scriptless exception
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 7: devices.js: the holding screen's scripted half
Files:
- Create:
internal/calls/static/devices.js - Modify:
internal/calls/templates/join.html
Interfaces:
- Produces: on the holding page with script, a
<video id="preview" muted autoplay playsinline>, three<select>s (camera, microphone, speaker; speaker only wheresetSinkIdexists), two toggles (mute, camera off), and a "Join" button that navigates to/m/{sqid}with the chosen device ids and mute state stored underlocalStorage["meet.devices"]as{camera, mic, speaker, muted, cameraOff}; without script, the paragraph and no button. Behaviour:navigator.mediaDevices.getUserMedia({audio: true, video: true})on load, thenenumerateDevices, populate selects (labels appear only after permission), restart the preview stream on selection change, stop all tracks on join. Errors (NotAllowedError,NotFoundError) show a sentence in#statusand disable Join.
- [ ] Step 1: Template
Replace the Task 7-relevant part of join.html's content:
<noscript><p>The call needs JavaScript to reach your camera and the other participants; everything else on this page works without it.</p></noscript>
<div id="devices" data-sqid="{{.Sqid}}" hidden>
<video id="preview" muted autoplay playsinline class="meet-preview"></video>
<div class="meet-pickers">
<label>Camera <select id="pick-camera"></select></label>
<label>Microphone <select id="pick-mic"></select></label>
<label id="speaker-row" hidden>Speaker <select id="pick-speaker"></select></label>
</div>
<div class="meet-controls">
<button type="button" id="toggle-mute" rst-btn aria-pressed="false">Mute</button>
<button type="button" id="toggle-camera" rst-btn aria-pressed="false">Camera off</button>
<button type="button" id="join" rst-btn="primary lg" disabled>Join the call</button>
</div>
<p id="status" class="meet-status" role="status"></p>
</div>
<link rel="stylesheet" href="{{asset "static/calls.css"}}">
<script defer src="{{asset "static/devices.js"}}"></script>Remove the slice-1 "Calls coming soon" paragraph.
- [ ] Step 2: devices.js
// devices.js: the holding screen. Enumerates devices, previews the
// camera, remembers choices in localStorage, and hands off to the room.
// Nothing here reaches the server.
(() => {
'use strict';
const root = document.getElementById('devices');
if (!root || !navigator.mediaDevices) return;
root.hidden = false;
const KEY = 'meet.devices';
const $ = (id) => document.getElementById(id);
const preview = $('preview'), pickCam = $('pick-camera'), pickMic = $('pick-mic'),
pickSpk = $('pick-speaker'), spkRow = $('speaker-row'), mute = $('toggle-mute'),
cam = $('toggle-camera'), join = $('join'), status = $('status');
let stream = null;
const prefs = load();
function load() {
try { return JSON.parse(localStorage.getItem(KEY) || '{}') || {}; } catch { return {}; }
}
function save() {
try { localStorage.setItem(KEY, JSON.stringify(prefs)); } catch {}
}
function say(text) { status.textContent = text; }
async function start() {
stop();
const constraints = {
audio: prefs.mic ? { deviceId: { exact: prefs.mic } } : true,
video: prefs.camera ? { deviceId: { exact: prefs.camera } } : true,
};
try {
stream = await navigator.mediaDevices.getUserMedia(constraints);
} catch (err) {
if (prefs.camera || prefs.mic) { delete prefs.camera; delete prefs.mic; save(); return start(); }
say(err.name === 'NotAllowedError'
? 'Meet needs permission to use your camera and microphone.'
: 'No camera or microphone was found.');
join.disabled = true;
return;
}
preview.srcObject = stream;
applyToggles();
await fill();
join.disabled = false;
say('');
}
function stop() {
if (stream) { for (const t of stream.getTracks()) t.stop(); stream = null; }
}
async function fill() {
const devices = await navigator.mediaDevices.enumerateDevices();
const fillSelect = (sel, kind, chosen) => {
sel.replaceChildren();
let n = 0;
for (const d of devices) {
if (d.kind !== kind) continue;
n++;
const o = document.createElement('option');
o.value = d.deviceId; o.textContent = d.label || `${kind.replace('input', '').replace('output', '')} ${n}`;
if (d.deviceId === chosen) o.selected = true;
sel.append(o);
}
return n;
};
fillSelect(pickCam, 'videoinput', prefs.camera || currentId('video'));
fillSelect(pickMic, 'audioinput', prefs.mic || currentId('audio'));
if (typeof HTMLMediaElement.prototype.setSinkId === 'function' && fillSelect(pickSpk, 'audiooutput', prefs.speaker) > 0) {
spkRow.hidden = false;
}
}
function currentId(kind) {
const t = stream && stream.getTracks().find((t) => t.kind === kind);
return t ? t.getSettings().deviceId : '';
}
function applyToggles() {
if (!stream) return;
for (const t of stream.getAudioTracks()) t.enabled = !prefs.muted;
for (const t of stream.getVideoTracks()) t.enabled = !prefs.cameraOff;
mute.setAttribute('aria-pressed', String(!!prefs.muted));
mute.textContent = prefs.muted ? 'Unmute' : 'Mute';
cam.setAttribute('aria-pressed', String(!!prefs.cameraOff));
cam.textContent = prefs.cameraOff ? 'Camera on' : 'Camera off';
}
pickCam.addEventListener('change', () => { prefs.camera = pickCam.value; save(); start(); });
pickMic.addEventListener('change', () => { prefs.mic = pickMic.value; save(); start(); });
pickSpk.addEventListener('change', () => { prefs.speaker = pickSpk.value; save(); });
mute.addEventListener('click', () => { prefs.muted = !prefs.muted; save(); applyToggles(); });
cam.addEventListener('click', () => { prefs.cameraOff = !prefs.cameraOff; save(); applyToggles(); });
join.addEventListener('click', () => { stop(); location.assign('/m/' + root.dataset.sqid); });
navigator.mediaDevices.addEventListener?.('devicechange', fill);
window.addEventListener('pagehide', stop);
start();
})();- [ ] Step 3: Test
Add to holding_test.go: the holding page as a member contains id="devices", devices.js, the <noscript> paragraph, and no longer contains "Calls coming soon". Run make ci, commit:
git add internal
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "holding screen: devices.js, preview, pickers, mute state, join
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 8: room.js: SSE client, outbound queue, perfect negotiation, reconciliation, resume, terminal states, admission UI
Files:
- Replace:
internal/calls/static/room.js - Modify:
internal/calls/static/calls.css
Interfaces (the client contract, spec §4.5, §4.6, §5.1):
- State:
peer,boot,gen[peerId],seqOut[peerId],lastIn[peerId],pcs[peerId] = {pc, polite, makingOffer, ignoreOffer, pendingCandidates[]},queue[peerId] = [{seq, message}]with one POST in flight per target,localstream from the devices prefs,admittedflag,roster. - Stream: open
EventSource('/m/{sqid}/stream'); onhellosetpeer, comparebootto the stored one, then POSTjoinonce per fresh stream; onerror(closed), resume with?peer=&last=after a 500 ms backoff, capped at 30 s; handle the resume refusals by reading the response whenEventSourcefails: sinceEventSourcehides status codes, do the resume withfetchfirst (HEAD-likeGETwithAccept: text/event-streamis not possible), so: on error,fetch(url + '&probe=1')where the server answers the same 404/409/410 JSON or204when the resume would succeed (addprobe=1handling tostream: run the refusal checks and return 204 without attaching), then open theEventSource. Document this instream.go. - Terminal reasons:
replaced,standing,expired,removed,endedand any 410 close everypc, stop the local stream, and show the sentence for that reason with the takeover link forreplaced.404with a differentboot: close every pc, fresh stream, join again.404same boot: same, but a409on the fresh stream shows the takeover link.409at first open: takeover link. - Roster reconciliation: on
roster, close pcs for ids not present, create pcs for ids present without one, and if a rostergenfor a peer is above the local one, rebuild that pc under the new gen. Onpeer, create. Ongone, close. Onpair, ifgenabove local, rebuild; else ignore. Ongap,resetto that peer. - Perfect negotiation (W3C reference):
polite = self < other;onnegotiationneeded→makingOffer = true; setLocalDescription(); send offer; onoffer:offerCollision = makingOffer || signalingState !== 'stable'; ignoreOffer = !polite && offerCollision; if ignoreOffer return; setRemoteDescription(); if offer: setLocalDescription(); send answer; oncandidate:addIceCandidatein try/catch, swallowed whenignoreOffer; candidates before a remote description are buffered. - Outbound:
send(to, type, payload)assignsseq = ++seqOut[to],gen = gen[to], enqueues; the pump POSTs/m/{sqid}/signalwith{peer, type, to, seq, gen, payload}, waits for 2xx; on 429 waitRetry-After; on other failures retry with backoff up to 30 s then close that pc and mark it for recreation on the next roster; on 409 stale gen drop the queue for that target (apairor roster will rebuild). - Inbound dedupe: drop a targeted message with
seq <= lastIn[from]; drop one withgen < gen[from]. - Admission UI: for members,
knockadds to#knockswith an "Admit" button that sendsadmit {peer};knock-goneremoves;roster.waitingfills it;fullshows "The room is full";state {waiting}shows "Waiting for someone to let you in". - Controls: mute and camera toggle track
enabled; leave sendsleave, closes everything, navigates to the holding screen. - Tiles: one
<video autoplay playsinline>per remote peer with the name; local tile muted; failed connection state shows "could not connect" with a Retry button that callspc.restartIce(). - Expose a test hook:
window.meet = { state() { return {peer, admitted, peers: Object.keys(pcs), frames: <Map peer→frames received via requestVideoFrameCallback or getStats>} } }used by the Playwright gate.
- [ ] Step 1: Write room.js
The implementer writes it to the contract above, keeping each concern in a named function (connectStream, resume, handle(event), reconcile(roster), createPC(id), closePC(id), negotiate, onOffer, onAnswer, onCandidate, send, pump, terminal(reason), renderTiles, renderQueue), with a header comment naming the spec sections. No library. Target size 600 to 800 lines. The Opus-tier implementer is expected to read spec §4.5, §4.6 and §5.1 before writing.
- [ ] Step 2: Add the probe branch to
stream.go
Before attaching a resume: if r.URL.Query().Get("probe") == "1", run the same Resume refusal checks without attaching (add hub.CanResume(sqid, session, peer) error returning the same ResumeError or nil), answer 204 on nil. For a fresh stream with probe=1, answer 409 if the session has a live peer, else 204. Test in stream_test.go: probe answers 204, 409, 410, 404 in the four situations.
- [ ] Step 3: Gate and commit
make ci (Go tests only; the browser gate is Task 9).
git add internal
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "room.js: SSE client, queue with seq and gen, perfect negotiation, reconciliation, resume, terminal states
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 9: The Playwright gate
Files:
- Create:
package.json,test/browser/call.mjs,internal/callstest/browser_test.go(//go:build browser) - Modify:
Makefile(browsertarget),.gitignore(node_modules/,package-lock.json)
Interfaces:
make browser:npm install --no-save playwright@1.62.1thengo test -tags browser -run TestCallInARealBrowser ./internal/callstest/.browser_test.go: starts the app on127.0.0.1:0withOriginset to that address andAllowedAddressespaul@example.com,paul@rslw.com, a recording mailer; signs two members in over HTTP and captures their session cookies (name fromrastrillo/sessionsCookieName,rastrillo_sessionon http); creates a meeting; runsnode test/browser/call.mjs <origin> <sqid> <cookieName> <cookieA> <cookieB> <scenario>for each scenario with a three-minute timeout, and fails on a non-zero exit. Scenarios in this slice:devices(holding screen enumerates and previews, join button enabled),call(two contexts join, each receives the other's video track with a non-zeroframescount within 20 s),renewal(withSetStreamTimingsshortened through a test-only headerX-Meet-Test-TTL: 2shonoured only whenConfig.Envis development; the call stays connected across a stream renewal),guest(a third context signs in through the link as a guest, knocks, is admitted from context A's queue, receives video; asserts throughwindow.meet.state()that it had nopeersbefore admission),removed(A removes the guest through the confirm page; the guest sees the terminal sentence),restart(the Go test restarts the app on the same port and asserts both contexts rejoin and reconnect),takeover(a second page in context A's browser gets the 409 link, confirms, the first page shows the terminal sentence and does not reconnect),crosssite(a page on another origin navigates to the takeover URL by GET and nothing ends),scriptless(a context with JavaScript disabled loads both pages and finds the paragraph).call.mjs:chromium.launch({ args: ['--use-fake-device-for-media-stream', '--use-fake-ui-for-media-stream'] }), onebrowser.newContext()per identity with the cookie set for the origin's host,page.waitForFunction(() => window.meet && window.meet.state().frames...); prints one line per assertion and exits non-zero on failure.
- [ ] Step 1: Makefile and package.json
# The browser gate. Not in ci: the one gate the team runs on every push
# must not depend on a toolchain the app itself does not use. Run it
# before every review that touches a page.
browser:
npm install --no-save playwright@1.62.1
go test -tags browser -run TestCallInARealBrowser -v ./internal/callstest/{
"name": "meet-browser-harness",
"private": true,
"description": "Playwright harness for the Meet browser gate. Not built; not published.",
"devDependencies": { "playwright": "1.62.1" },
"scripts": { "test:browser": "go test -tags browser -run TestCallInARealBrowser ./internal/callstest/" }
}- [ ] Step 2: The Go driver and call.mjs
Write both to the contract. Frame counting in the page: after ontrack, attach the stream to a hidden video and use video.requestVideoFrameCallback to count; fall back to getStats() framesDecoded when unavailable. The devices scenario asserts the camera select has at least one option and #join is enabled within 10 s.
- [ ] Step 3: Run the gate and commit
Run: make browser 2>&1 | tail -20 Expected: every scenario passes; record the run's output in the branch discussion.
git add Makefile package.json test internal/callstest/browser_test.go .gitignore
git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "browser gate: Playwright driver, two-context call, guest, removal, restart, takeover, scriptless
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2"---
Task 10: The remaining §13 hub gates and the captured-log test extension
Files:
- Modify:
internal/calls/hub/*_test.go,internal/callstest/signal_test.go
Cover, by name, what Tasks 1 to 5 did not: concurrent admissions (five goroutines admitting five waiting guests into one free slot, exactly one succeeds), a pair at or below the local generation ignored (hub side: Reset twice from one side yields gens 2 and 3 and a relay with gen 2 is stale), an admitted guest continuing past grant expiry on signal, stream?peer= and (stubbed until slice 3) /ice while being refused fresh entry (/m/{sqid}/join 404 after advance(2h+1m)), the 64-event buffer over the HTTP stream (lost {reason: "buffer"} then a successful resume), and a member dropped on the tick over HTTP (allowlist change through TestHandle.SetAllowlist(addrs...)). Then make ci and commit tests: remaining slice-2 hub and stream gates.
---
Task 11: Copy catalogue and branch record
Files:
- Create:
docs/superpowers/copy/2026-09-10-slice-2-strings.md
Every user-facing string in Tasks 6 to 8, verbatim, with its location, marked "copy review waived for this round per suite #5; catalogued for the later pass". Commit docs: slice 2 strings.
---
Self-review
Spec coverage (§12 item 2): devices.js (7); hub with peer binding (1), participation (1, 2), cap (2), limits (1, 5), resume with admission-branched output (3, 4), replay ring, gap, pair reset (3), terminal replaced (2, 3, 6), the tick (1, 2); stream and hello (4); every §4.3 message except chat and share (2, 3, 5); removal confirm page (6); room.js with perfect negotiation and the §4.6 queue (8); participation transitions (2, 10); AGENTS.md exception (6); CARLOS_RELAY absent (unchanged); the browser gate (9). §13 slice-2 Go gates are spread across Tasks 1, 2, 3, 4, 5, 6 and 10 and named there; the Playwright gates are Task 9's scenarios. /ice belongs to slice 3 and is only stubbed in Task 10's continuation test.
Placeholders: Tasks 8 and 9 give a contract and function list rather than full code because they are Opus-tier design tasks; every behaviour they must have is named, with the spec section, and the gate that proves it is in Task 9. Everything else has code.
Type consistency: hub.Message fields, PeerID, Kind, Reason, the error variables and ResumeError are declared in Task 1 and used unchanged after; Hub methods Open, Find, LiveForSession, Detach, Tick (1), Join, Admit, Leave, Remove, Takeover, End, rosterFor, genLocked, admitLocked (2), Relay, Reset, Resume, ringLocked (3), CanResume (8); app fields hub, streamHeartbeat, streamTTL, signalLimiter (4, 5); TestHandle.TickForTest, SetStreamTimings, SetAllowlist (4, 10); harness stream, signal, sseReader.next/nextOf/hello/waitClosed/close, newHarnessWithLogger, newMemberBrowser, per-harness origin (4, 5).
Rulings for the controller: the session key for the hub is the session subject plus the session's creation instant, since rastrillo does not expose the cookie hash; two tabs of one browser share it, which is the spec's intent. The resume probe (?probe=1) is an addition to the spec's §4.5 needed because EventSource hides status codes; record it as a spec revision note in the branch discussion.