Video calls (PairUX→ASCII), agent@ chat, and finger routes

ssh video-<code>@ joins a PairUX/LiveKit call rendered as truecolor
ASCII (▀ half-blocks, 2 pixels per cell); video@ prompts for a code.
Codes are minted by PairUX only. Pipeline: VP8 RTP → PLI keyframe
requests → ivfwriter remux → ffmpeg decode/scale → RGB24 → ANSI →
bubbletea over the SSH PTY. Subscriber-only, no audio in v1.

ssh agent@ opens a persisted chat with the operator's agent —
AGENTBBS_AGENT_CMD runs per message (stdin→stdout), e.g. `claude -p`.

ssh <member>@ with someone else's name prints a classic finger card
(.plan, member since, last seen) and disconnects; your own name still
enters the hub.

cmd/lkpublish: dev publisher for testing (explicit -fps pacing; lksdk
IVF replay mispaces from file timebase alone, measured 1fps from a
15fps file; dimensions required or dynacast pauses the track).

Verified end-to-end against livekit-server --dev: 128k truecolor cells
/ 20k distinct colors streamed over a real SSH session; chat round-trip
and finger card verified over SSH. Go toolchain pinned to 1.26 via
mise.toml (lksdk requirement).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

internal/calls/livekit.go

@@ -0,0 +1,223 @@
+// LiveKit subscriber: joins a PairUX room, takes the first remote VP8 video
+// track, remuxes RTP into IVF, and has ffmpeg decode + scale it into raw
+// RGB24 frames sized for the terminal grid.
+package calls
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"time"
+
+	lksdk "github.com/livekit/server-sdk-go/v2"
+	"github.com/pion/webrtc/v4"
+	"github.com/pion/webrtc/v4/pkg/media/ivfwriter"
+)
+
+// Config is the LiveKit deployment shared with PairUX.
+type Config struct {
+	URL, Key, Secret string
+}
+
+// ConfigFromEnv reads AGENTBBS_LIVEKIT_* with LIVEKIT_* fallbacks, matching
+// PairUX's env shape (LIVEKIT_API_KEY / LIVEKIT_API_SECRET).
+func ConfigFromEnv() Config {
+	pick := func(keys ...string) string {
+		for _, k := range keys {
+			if v := os.Getenv(k); v != "" {
+				return v
+			}
+		}
+		return ""
+	}
+	return Config{
+		URL:    pick("AGENTBBS_LIVEKIT_URL", "LIVEKIT_URL", "NEXT_PUBLIC_LIVEKIT_URL"),
+		Key:    pick("AGENTBBS_LIVEKIT_KEY", "LIVEKIT_API_KEY"),
+		Secret: pick("AGENTBBS_LIVEKIT_SECRET", "LIVEKIT_API_SECRET"),
+	}
+}
+
+func (c Config) ok() bool { return c.URL != "" && c.Key != "" && c.Secret != "" }
+
+// session is one live subscription: frames arrive on Frames sized pw*ph*3.
+type session struct {
+	Frames chan []byte
+	Status chan string
+	room   *lksdk.Room
+	ffmpeg *exec.Cmd
+	ivf    io.WriteCloser
+	done   chan struct{}
+}
+
+// join connects to room `code` as a hidden-ish subscriber and starts the
+// decode pipeline targeting pw x ph pixels.
+func join(cfg Config, code, identity string, pw, ph int) (*session, error) {
+	if !cfg.ok() {
+		return nil, fmt.Errorf("video calls are not configured on this host (LIVEKIT_URL/API_KEY/API_SECRET)")
+	}
+
+	s := &session{
+		Frames: make(chan []byte, 2),
+		Status: make(chan string, 8),
+		done:   make(chan struct{}),
+	}
+
+	// ffmpeg: IVF (VP8) on stdin → raw RGB24 frames on stdout.
+	ffin, ivfW := io.Pipe()
+	s.ivf = ivfW
+	if dump := os.Getenv("AGENTBBS_VIDEO_DEBUG"); dump != "" {
+		if f, err := os.Create(dump); err == nil {
+			s.ivf = teeWriteCloser{io.MultiWriter(ivfW, f), ivfW, f}
+		}
+	}
+	cmd := exec.Command("ffmpeg",
+		"-hide_banner", "-loglevel", "error",
+		"-probesize", "32", "-analyzeduration", "0",
+		"-fflags", "nobuffer", "-flags", "low_delay",
+		"-f", "ivf", "-i", "pipe:0",
+		"-vf", fmt.Sprintf("scale=%d:%d", pw, ph),
+		"-f", "rawvideo", "-pix_fmt", "rgb24", "pipe:1",
+	)
+	cmd.Stdin = ffin
+	cmd.Stderr = os.Stderr // surfaces decode errors in the server log
+	out, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	if err := cmd.Start(); err != nil {
+		return nil, fmt.Errorf("ffmpeg: %w", err)
+	}
+	s.ffmpeg = cmd
+
+	go func() { // frame pump: drop frames rather than lag
+		size := pw * ph * 3
+		for {
+			buf := make([]byte, size)
+			if _, err := io.ReadFull(out, buf); err != nil {
+				close(s.Frames)
+				return
+			}
+			select {
+			case s.Frames <- buf:
+			default:
+			}
+		}
+	}()
+
+	gotTrack := false
+	cb := &lksdk.RoomCallback{
+		ParticipantCallback: lksdk.ParticipantCallback{
+			OnTrackSubscribed: func(track *webrtc.TrackRemote, pub *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {
+				if gotTrack || track.Kind() != webrtc.RTPCodecTypeVideo {
+					return
+				}
+				if track.Codec().MimeType != webrtc.MimeTypeVP8 {
+					s.status("skipping non-VP8 track from " + rp.Identity())
+					return
+				}
+				gotTrack = true
+				s.status("video from " + rp.Identity())
+				// PLI is what makes the SFU start forwarding from a
+				// keyframe; without it a fresh subscriber starves.
+				rp.WritePLI(track.SSRC())
+				go s.keyframeTicker(rp, track.SSRC())
+				go s.consume(track)
+			},
+		},
+		OnDisconnected: func() { s.status("disconnected") },
+	}
+
+	room, err := lksdk.ConnectToRoom(cfg.URL, lksdk.ConnectInfo{
+		APIKey:              cfg.Key,
+		APISecret:           cfg.Secret,
+		RoomName:            code,
+		ParticipantIdentity: identity,
+	}, cb)
+	if err != nil {
+		s.Close()
+		return nil, fmt.Errorf("livekit: %w", err)
+	}
+	s.room = room
+	s.status("joined " + code + " — waiting for video…")
+	return s, nil
+}
+
+// keyframeTicker re-requests keyframes so late joins and packet loss recover
+// quickly; a periodic full refresh is cheap at terminal resolutions.
+func (s *session) keyframeTicker(rp *lksdk.RemoteParticipant, ssrc webrtc.SSRC) {
+	t := time.NewTicker(2 * time.Second)
+	defer t.Stop()
+	for range t.C {
+		select {
+		case <-s.done:
+			return
+		default:
+		}
+		rp.WritePLI(ssrc)
+	}
+}
+
+// consume remuxes the track's RTP into the IVF pipe until it ends.
+func (s *session) consume(track *webrtc.TrackRemote) {
+	w, err := ivfwriter.NewWith(s.ivf)
+	if err != nil {
+		s.status("ivf: " + err.Error())
+		return
+	}
+	n := 0
+	for {
+		pkt, _, err := track.ReadRTP()
+		if err != nil {
+			s.status(fmt.Sprintf("track ended after %d packets: %v", n, err))
+			_ = w.Close()
+			return
+		}
+		if err := w.WriteRTP(pkt); err != nil {
+			s.status("ivf write: " + err.Error())
+			return
+		}
+		n++
+		if n == 1 || n%500 == 0 {
+			s.status(fmt.Sprintf("receiving (%d rtp packets)", n))
+		}
+	}
+}
+
+func (s *session) status(msg string) {
+	select {
+	case s.Status <- msg:
+	default:
+	}
+}
+
+// teeWriteCloser mirrors the IVF stream to a debug file (AGENTBBS_VIDEO_DEBUG).
+type teeWriteCloser struct {
+	io.Writer
+	pipe io.WriteCloser
+	file io.Closer
+}
+
+func (t teeWriteCloser) Close() error {
+	_ = t.file.Close()
+	return t.pipe.Close()
+}
+
+// Close tears the whole pipeline down.
+func (s *session) Close() {
+	select {
+	case <-s.done:
+	default:
+		close(s.done)
+	}
+	if s.room != nil {
+		s.room.Disconnect()
+	}
+	if s.ivf != nil {
+		_ = s.ivf.Close()
+	}
+	if s.ffmpeg != nil && s.ffmpeg.Process != nil {
+		_ = s.ffmpeg.Process.Kill()
+		_ = s.ffmpeg.Wait()
+	}
+}