// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Ask the current master to yield, then assert ourselves as master.
	if m := d.master(); m != 0 && m != v.Device {
		if ip := d.get(m); ip != nil {
			c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
			fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)
		}
	}

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: link.Broadcast(nil), Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

-func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config) {

+func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP) {


Edit

				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: link.Broadcast(nil), Port: 50001})

				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})


The file /Users/noah/conductor/workspaces/deadca7/san-diego/link/cmd/master/main.go has been updated successfully. (file state is current in your context — no need to Read it back)

				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})

// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Ask the current master to yield, then assert ourselves as master.
	if m := d.master(); m != 0 && m != v.Device {
		if ip := d.get(m); ip != nil {
			c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
			fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)
		}
	}

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: link.Broadcast(nil), Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

-        c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: link.Broadcast(nil), Port: 50001})

+        c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})


yes we do want to drive tempo, and we also want to just close out packet dumps for now, so start it


Bash

go build ./... 2>&1 | tail && go vet ./... 2>&1 | grep -v rekordbox_pdb | tail && echo "BUILD+VET OK"

Build and vet the master tool


BUILD+VET OK


initialization failed: packages.Load error: err: exit status 1: stderr: go: go.work requires go >= 1.26.4 (running go 1.26.0)


Now I'm running a test session with the master device to validate it comes online, broadcasts its status, and captures the protocol handshake—I'll start the master process in the background with playback and tempo settings, then ask the user to press MASTER on a deck to trigger the 0x26 command so I can verify we respond with 0x27, and also test the become-master flow where our device sends 0x26 to the current master.


Builds and vets clean. Let me run a live master session — this drives tempo and should provoke the directed handoff/sync packets. Starting it now; I'll have you press MASTER/SYNC on a deck.


go run ./cmd/master -device 1 -tempo 124 -playing -out testdata/master.jsonl -secs 45 2>&1; echo "EXIT=$?"

Live master session 45s


Command running in background with ID: bx07v1hij. Output is being written to: /private/tmp/claude-501/-Users-noah-conductor-workspaces-deadca7-san-diego/5e143c63-1d6b-4233-91af-bb63e39e8beb/tasks/bx07v1hij.output. You will be notified when it completes. To check interim output, use Read on that file path.


🎬 **Master session running (45s)** — our virtual CDJ (device 1) is claiming tempo master and broadcasting status + beats at **124 BPM**. On the decks:

1. **Press MASTER on deck 3 or 4** (to grab master back from us) → that deck sends us a **handoff request `0x26`**, and we auto-reply `0x27` (watch for "answered master-handoff request" in my analysis).
2. **Press SYNC on a deck** → should produce **`0x2a`** sync-control.
3. If a deck is **synced to us**, it should follow our 124 BPM — that's us driving tempo.

I'll let the 45s finish, then report which directed packets we captured and confirm whether the decks accepted us as master / followed our tempo.


<task-notification>
<task-id>bx07v1hij</task-id>
<tool-use-id>toolu_01UMWjsgcVSY8HfjAEk5QsQ8</tool-use-id>
<output-file>/private/tmp/claude-501/-Users-noah-conductor-workspaces-deadca7-san-diego/5e143c63-1d6b-4233-91af-bb63e39e8beb/tasks/bx07v1hij.output</output-file>
<status>completed</status>
<summary>Background command "Live master session 45s" completed (exit code 0)</summary>
</task-notification>


cd /Users/noah/conductor/workspaces/deadca7/san-diego/link
echo "=== our events (stderr) ==="; grep -vE '^\{' /private/tmp/claude-501/-Users-noah-conductor-workspaces-deadca7-san-diego/5e143c63-1d6b-4233-91af-bb63e39e8beb/tasks/bx07v1hij.output | sort | uniq -c
echo "=== packet inventory + NEW types ==="
python3 -c '
import json,collections
c=collections.OrderedDict()
known={(50000,0x00),(50000,0x02),(50000,0x04),(50000,0x05),(50000,0x06),(50000,0x0a),(50000,0x08),(50001,0x03),(50001,0x04),(50001,0x0b),(50001,0x28),(50002,0x0a),(50002,0x40),(50002,0x05),(50002,0x06)}
for l in open("testdata/master.jsonl"):
    p=json.loads(l); d=bytes.fromhex(p["data"])
    if d[:10]!=bytes.fromhex("5173707431576d4a4f4c"): continue
    c[(p["port"],d[0x0a],p["src"].split(":")[0])]=c.get((p["port"],d[0x0a],p["src"].split(":")[0]),0)+1
for (port,typ,src),n in sorted(c.items()):
    new=" <-- NEW" if (port,typ) not in known else ""
    print("%5d 0x%02x %-15s %5d%s"%(port,typ,src,n,new))
'

Analyze master session for directed packets


=== our events (stderr) ===
   1 EXIT=0
   1 online as device 1
=== packet inventory + NEW types ===
50000 0x00 192.168.0.160       3
50000 0x02 192.168.0.160       3
50000 0x04 192.168.0.160       1
50000 0x05 192.168.0.193       1
50000 0x06 192.168.0.160      28
50000 0x06 192.168.0.192      20
50000 0x06 192.168.0.193      22
50000 0x06 192.168.0.194      20
50000 0x0a 192.168.0.160       3
50001 0x03 192.168.0.193     147
50001 0x04 192.168.0.193       6
50001 0x0b 192.168.0.192    1383
50001 0x0b 192.168.0.194    1383
50001 0x28 192.168.0.160      85
50002 0x0a 192.168.0.160     207
50002 0x0a 192.168.0.192     250
50002 0x0a 192.168.0.194     207
50002 0x40 192.168.0.192      22
50002 0x40 192.168.0.194      19


 The master tool is online and broadcasting heartbeats and status messages on the expected ports, and the real decks are responding with their own status and position data. However, I'm not seeing any of the directed control packets that would indicate the user pressed MASTER or SYNC, or that the decks acknowledged the handoff request.

The key question is whether the real decks actually accepted us as master and started following our tempo. I need to check their status flags and BPM values during the session to see if their master flag dropped and their sync locked onto our 124 BPM. I should also verify whether our become-master logic even triggered — did we ever detect a deck advertising itself as master so we could send the handoff request? If the decks were paused or not synced, they wouldn't respond to tempo changes or initiate a handoff, and without a user pressing MASTER or SYNC on the decks, we wouldn't capture those directed packets either.


Our master emitter works — we're broadcasting **85 beats** and **207 status packets** as device 1. But no directed `0x26`/`0x27`/`0x2a` were captured, and we never logged "requested master" or "answered handoff" — so the decks didn't hand off to us. Let me check whether the real decks even recognized us as master (did their flags/tempo react?).


cd /Users/noah/conductor/workspaces/deadca7/san-diego/link
python3 -c '
import json
def u(b,o,n): return int.from_bytes(b[o:o+n],"big")
rows=[json.loads(l) for l in open("testdata/master.jsonl")]
for ip,dev in [("192.168.0.192","3"),("192.168.0.194","4")]:
    seen=[]
    for p in rows:
        d=bytes.fromhex(p["data"])
        if p["src"].startswith(ip) and p["port"]==50002 and len(d)>0x92 and d[0x0a]==0x0a:
            f=d[0x89]
            k=(bool(f&0x40),bool(f&0x20),bool(f&0x10),u(d,0x92,2)/100)  # playing,master,sync,bpm
            if k not in seen: seen.append(k)
    print("deck %s (playing,master,sync,bpm):"%dev)
    for k in seen: print("   ",k)
'

Check if decks recognized us as master


deck 3 (playing,master,sync,bpm):
    (False, True, True, 113.99)
    (False, False, True, 113.99)
deck 4 (playing,master,sync,bpm):
    (False, False, False, 102.0)


 Deck 3 relinquished master status when I asserted my master claim, which shows it recognized the assertion, but it didn't sync its tempo to my 124 BPM because it's paused and won't follow tempo changes while stopped. The sync flag stayed true even as it dropped master, suggesting the deck understood the protocol but is waiting for playback to resume before adjusting.

I'm realizing the formal handoff exchange (0x26/0x27) never happened because I didn't initiate it properly — when I became master, the status hadn't arrived yet so the master flag wasn't set, and the decks didn't press their MASTER buttons to request from me. Deck 3 just saw two masters and yielded its flag without the directed packet exchange. To actually capture those handoff packets, I'd need either a deck to request master from me while I'm master, or my become-master logic to fire when the master flag is already populated. Refactoring the drive logic to defer the master claim: I'll have the reader detect incoming 0x27 responses and set the master flag true only after we receive that yield, while drive initially emits status with Master false, waits a few seconds to identify the current master, then sends 0x26 to it. If no master exists after the wait, we claim master immediately as a fallback. Refactoring to use shared state with atomics for master and handoff flags, passing them through to both the reader and drive goroutines. The reader will set master on receiving 0x27, while drive will probe for an existing master and only claim it after a timeout if none responds. I need to add the sync/atomic import and wire these changes through.


Revealing result: **deck 3 dropped its master flag (True→False) when we claimed master** — so the decks *do* recognize our status. But no `0x26`/`0x27` handshake happened because of a timing bug: our `drive()` asserts master immediately (so deck 3 yields silently) and our become-master request never fired (`d.master()` was still empty at startup). Let me fix it to do the proper sequence — observe, send `0x26` to the current master, flip to master on the `0x27` response.


Edit

	"os/signal"
	"sync"
	"sync/atomic"
	"time"


	"os/signal"
	"sync"
	"sync/atomic"
	"time"

// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Ask the current master to yield, then assert ourselves as master.
	if m := d.master(); m != 0 && m != v.Device {
		if ip := d.get(m); ip != nil {
			c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
			fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)
		}
	}

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

"os"
"os/signal"
"sync"
+  "sync/atomic"
"time"
"github.com/cockroachdb/errors"


	d := &devices{ip: map[byte]net.IP{}}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast) }

	d := &devices{ip: map[byte]net.IP{}}
	var master atomic.Bool

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }


	d := &devices{ip: map[byte]net.IP{}}
	var master atomic.Bool

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }

// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Ask the current master to yield, then assert ourselves as master.
	if m := d.master(); m != 0 && m != v.Device {
		if ip := d.get(m); ip != nil {
			c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
			fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)
		}
	}

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

}
d := &devices{ip: map[byte]net.IP{}}
+  var master atomic.Bool
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error {

-  g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, record) })

-  g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, record) })

+  g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })

+  g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

-  vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast) }

+  vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }


func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP) {

func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP, master *atomic.Bool) {


// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}
	var master atomic.Bool

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Ask the current master to yield, then assert ourselves as master.
	if m := d.master(); m != 0 && m != v.Device {
		if ip := d.get(m); ip != nil {
			c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
			fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)
		}
	}

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

-func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP) {

+func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP, master *atomic.Bool) {


Now replace the immediate-master logic with the proper observe-then-request sequence, and use `master.Load()` in status:


	// Ask the current master to yield, then assert ourselves as master.
	if m := d.master(); m != 0 && m != v.Device {
		if ip := d.get(m); ip != nil {
			c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
			fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)
		}
	}

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond

	// Become master via the proper handoff: observe who is master, request it to
	// yield (0x26), and flip to master when it answers (0x27). If nobody is
	// master after a short wait, just take the role.
	go func() {
		for i := 0; i < 25 && ctx.Err() == nil; i++ {
			if m := d.master(); m != 0 && m != v.Device {
				if ip := d.get(m); ip != nil {
					c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
					fmt.Fprintf(os.Stderr, "requested master handoff from player %d\n", m)
					return
				}
			}
			time.Sleep(200 * time.Millisecond)
		}
		master.Store(true)
		fmt.Fprintln(os.Stderr, "no existing master; taking the role")
	}()

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond


	// Become master via the proper handoff: observe who is master, request it to
	// yield (0x26), and flip to master when it answers (0x27). If nobody is
	// master after a short wait, just take the role.
	go func() {
		for i := 0; i < 25 && ctx.Err() == nil; i++ {
			if m := d.master(); m != 0 && m != v.Device {
				if ip := d.get(m); ip != nil {
					c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
					fmt.Fprintf(os.Stderr, "requested master handoff from player %d\n", m)
					return
				}
			}
			time.Sleep(200 * time.Millisecond)
		}
		master.Store(true)
		fmt.Fprintln(os.Stderr, "no existing master; taking the role")
	}()

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond

// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}
	var master atomic.Bool

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP, master *atomic.Bool) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Ask the current master to yield, then assert ourselves as master.
	if m := d.master(); m != 0 && m != v.Device {
		if ip := d.get(m); ip != nil {
			c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
			fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)
		}
	}

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

-  // Ask the current master to yield, then assert ourselves as master.

-  if m := d.master(); m != 0 && m != v.Device {

-      c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})

-      fmt.Fprintf(os.Stderr, "requested master from player %d\n", m)

+  // Become master via the proper handoff: observe who is master, request it to

+  // yield (0x26), and flip to master when it answers (0x27). If nobody is

+  // master after a short wait, just take the role.

+      if m := d.master(); m != 0 && m != v.Device {

+          c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})

+          fmt.Fprintf(os.Stderr, "requested master handoff from player %d\n", m)

+      time.Sleep(200 * time.Millisecond)

+    fmt.Fprintln(os.Stderr, "no existing master; taking the role")


s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}

s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: master.Load(), OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}


// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}
	var master atomic.Bool

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP, master *atomic.Bool) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Become master via the proper handoff: observe who is master, request it to
	// yield (0x26), and flip to master when it answers (0x27). If nobody is
	// master after a short wait, just take the role.
	go func() {
		for i := 0; i < 25 && ctx.Err() == nil; i++ {
			if m := d.master(); m != 0 && m != v.Device {
				if ip := d.get(m); ip != nil {
					c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
					fmt.Fprintf(os.Stderr, "requested master handoff from player %d\n", m)
					return
				}
			}
			time.Sleep(200 * time.Millisecond)
		}
		master.Store(true)
		fmt.Fprintln(os.Stderr, "no existing master; taking the role")
	}()

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

-      s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: true, OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}

+      s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: master.Load(), OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}


Now update the reader to set master on `0x27` (our handoff request was accepted):


func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {

func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, master *atomic.Bool, record func([]byte, int, *net.UDPAddr)) error {


// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}
	var master atomic.Bool

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP, master *atomic.Bool) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Become master via the proper handoff: observe who is master, request it to
	// yield (0x26), and flip to master when it answers (0x27). If nobody is
	// master after a short wait, just take the role.
	go func() {
		for i := 0; i < 25 && ctx.Err() == nil; i++ {
			if m := d.master(); m != 0 && m != v.Device {
				if ip := d.get(m); ip != nil {
					c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
					fmt.Fprintf(os.Stderr, "requested master handoff from player %d\n", m)
					return
				}
			}
			time.Sleep(200 * time.Millisecond)
		}
		master.Store(true)
		fmt.Fprintln(os.Stderr, "no existing master; taking the role")
	}()

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: master.Load(), OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

-func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, record func([]byte, int, *net.UDPAddr)) error {

+func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, master *atomic.Bool, record func([]byte, int, *net.UDPAddr)) error {


		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)

		if n > 0x0a {
			switch link.Type(b[0x0a]) {
			case link.TypeMasterRequest:
				c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
				master.Store(false)
				fmt.Fprintf(os.Stderr, "answered master-handoff request from %s (yielded)\n", src.IP)
			case link.TypeMasterResponse:
				master.Store(true)
				fmt.Fprintf(os.Stderr, "received master handoff (0x27) from %s; now master\n", src.IP)
			}
		}
		record(b, port, src)


		if n > 0x0a {
			switch link.Type(b[0x0a]) {
			case link.TypeMasterRequest:
				c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
				master.Store(false)
				fmt.Fprintf(os.Stderr, "answered master-handoff request from %s (yielded)\n", src.IP)
			case link.TypeMasterResponse:
				master.Store(true)
				fmt.Fprintf(os.Stderr, "received master handoff (0x27) from %s; now master\n", src.IP)
			}
		}
		record(b, port, src)

// Command master brings a virtual CDJ online as tempo master: it emits status
// and beat packets at a chosen tempo, takes the master role, answers handoff
// requests, and records all traffic so the directed master/sync packets can be
// captured. It can also push load-track and load-settings commands to a player.
//
//	go run ./cmd/master -device 1 -tempo 124 -playing -out master.jsonl -secs 60
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cockroachdb/errors"
	"github.com/nzoschke/deadca7/link"
	"golang.org/x/sync/errgroup"
)

type Packet struct {
	Data string `json:"data"`
	Len  int    `json:"len"`
	Port int    `json:"port"`
	Src  string `json:"src"`
	TS   int64  `json:"ts"`
}

func main() {
	device := flag.Int("device", 1, "virtual CDJ device number (must be 1-4 to be master)")
	loadTrack := flag.String("load-track", "", "load a track: target,source,id (e.g. 3,3,1)")
	loadSettings := flag.Int("load-settings", 0, "send load-settings to this target player")
	name := flag.String("name", "deadca7", "virtual CDJ device name")
	out := flag.String("out", "master.jsonl", "output JSONL fixture path")
	playing := flag.Bool("playing", false, "emit beats and a playing status")
	secs := flag.Int("secs", 60, "run duration in seconds")
	tempo := flag.Float64("tempo", 120, "tempo in BPM to broadcast")
	flag.Parse()

	if *device < 1 || *device > 4 {
		fmt.Fprintln(os.Stderr, "device must be 1-4 to act as master")
		os.Exit(1)
	}

	cfg := config{
		device:       byte(*device),
		loadSettings: byte(*loadSettings),
		loadTrack:    *loadTrack,
		name:         *name,
		out:          *out,
		playing:      *playing,
		secs:         time.Duration(*secs) * time.Second,
		tempo:        *tempo,
	}
	if err := run(cfg); err != nil {
		fmt.Fprintf(os.Stderr, "%+v\n", err)
		os.Exit(1)
	}
}

type config struct {
	device       byte
	loadSettings byte
	loadTrack    string
	name         string
	out          string
	playing      bool
	secs         time.Duration
	tempo        float64
}

func run(cfg config) error {
	f, err := os.Create(cfg.out)
	if err != nil {
		return errors.WithStack(err)
	}
	defer f.Close()

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()
	ctx, cancel = context.WithTimeout(ctx, cfg.secs)
	defer cancel()

	c0, err := listenUDP(50000)
	if err != nil {
		return err
	}
	defer c0.Close()
	if err := link.SetBroadcast(c0); err != nil {
		return err
	}
	c1, err := listenUDP(50001)
	if err != nil {
		return err
	}
	defer c1.Close()
	c2, err := listenUDP(50002)
	if err != nil {
		return err
	}
	defer c2.Close()

	vcdj, ipNet, err := online(c0, cfg.device, cfg.name)
	if err != nil {
		return err
	}

	enc := json.NewEncoder(f)
	var encMu sync.Mutex
	record := func(b []byte, port int, src *net.UDPAddr) {
		encMu.Lock()
		defer encMu.Unlock()
		enc.Encode(Packet{Data: fmt.Sprintf("%x", b), Len: len(b), Port: port, Src: src.String(), TS: time.Now().UnixMilli()})
	}

	d := &devices{ip: map[byte]net.IP{}}
	var master atomic.Bool

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error {
		return vcdj.Online(ctx, c0, ipNet, func(b []byte, src *net.UDPAddr) {
			if k, err := link.DecodeKeepAlive(b); err == nil {
				d.set(k.Device, k.IP)
			}
			record(b, 50000, src)
		})
	})
	g.Go(func() error { return reader(ctx, c1, 50001, d, vcdj, c2, &master, record) })
	g.Go(func() error { return reader(ctx, c2, 50002, d, vcdj, c2, &master, record) })

	bcast := link.Broadcast(ipNet)
	vcdj.OnOnline = func() { drive(ctx, vcdj, c1, c2, d, cfg, bcast, &master) }

	if err := g.Wait(); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	return nil
}

// drive emits status and beats as master and runs any one-shot commands.
func drive(ctx context.Context, v *link.VCDJ, c1, c2 *net.UDPConn, d *devices, cfg config, bcast net.IP, master *atomic.Bool) {
	if cfg.loadTrack != "" {
		var target, source, id int
		if _, err := fmt.Sscanf(cfg.loadTrack, "%d,%d,%d", &target, &source, &id); err == nil {
			if ip := d.get(byte(target)); ip != nil {
				c2.WriteToUDP(v.LoadTrack(byte(target), byte(source), link.SlotSD, link.TrackTypeRekordbox, uint32(id)), &net.UDPAddr{IP: ip, Port: 50002})
				fmt.Fprintf(os.Stderr, "sent load-track to player %d: track %d\n", target, id)
			}
		}
	}
	if cfg.loadSettings != 0 {
		if ip := d.get(cfg.loadSettings); ip != nil {
			c2.WriteToUDP(v.LoadSettings(cfg.loadSettings), &net.UDPAddr{IP: ip, Port: 50002})
			fmt.Fprintf(os.Stderr, "sent load-settings to player %d\n", cfg.loadSettings)
		}
	}

	// Become master via the proper handoff: observe who is master, request it to
	// yield (0x26), and flip to master when it answers (0x27). If nobody is
	// master after a short wait, just take the role.
	go func() {
		for i := 0; i < 25 && ctx.Err() == nil; i++ {
			if m := d.master(); m != 0 && m != v.Device {
				if ip := d.get(m); ip != nil {
					c1.WriteToUDP(v.MasterRequest(), &net.UDPAddr{IP: ip, Port: 50001})
					fmt.Fprintf(os.Stderr, "requested master handoff from player %d\n", m)
					return
				}
			}
			time.Sleep(200 * time.Millisecond)
		}
		master.Store(true)
		fmt.Fprintln(os.Stderr, "no existing master; taking the role")
	}()

	interval := time.Duration(60000/cfg.tempo) * time.Millisecond
	start := time.Now()
	status := time.NewTicker(200 * time.Millisecond)
	defer status.Stop()
	beat := time.NewTicker(interval)
	defer beat.Stop()

	var counter uint32
	for {
		select {
		case <-ctx.Done():
			return
		case <-status.C:
			n := int(time.Since(start) / interval)
			counter++
			s := link.StatusOut{Beat: n + 1, BeatInBar: byte(n%4) + 1, Counter: counter, Master: master.Load(), OnAir: true, Playing: cfg.playing, Synced: true, Tempo: cfg.tempo}
			for _, ip := range d.all() {
				c2.WriteToUDP(v.StatusPacket(s), &net.UDPAddr{IP: ip, Port: 50002})
			}
		case <-beat.C:
			if cfg.playing {
				n := int(time.Since(start) / interval)
				c1.WriteToUDP(v.BeatPacket(byte(n%4)+1, cfg.tempo), &net.UDPAddr{IP: bcast, Port: 50001})
			}
		}
	}
}

// reader records every datagram and answers master-handoff requests sent to us.
func reader(ctx context.Context, c *net.UDPConn, port int, d *devices, v *link.VCDJ, c2 *net.UDPConn, master *atomic.Bool, record func([]byte, int, *net.UDPAddr)) error {
	go func() { <-ctx.Done(); c.SetReadDeadline(time.Now()) }()
	buf := make([]byte, 1500)
	for {
		n, src, err := c.ReadFromUDP(buf)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return errors.WithStack(err)
		}
		b := append([]byte(nil), buf[:n]...)
		if port == 50002 {
			if s, err := link.DecodeStatus(b); err == nil {
				d.set(s.Device, src.IP)
				if s.Master {
					d.setMaster(s.Device)
				}
			}
		}
		if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {
			c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})
			fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)
		}
		record(b, port, src)
	}
}

type devices struct {
	mu  sync.Mutex
	ip  map[byte]net.IP
	mas byte
}

func (d *devices) set(dev byte, ip net.IP) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.ip[dev] = append(net.IP(nil), ip...)
}

func (d *devices) setMaster(dev byte) { d.mu.Lock(); d.mas = dev; d.mu.Unlock() }

func (d *devices) master() byte { d.mu.Lock(); defer d.mu.Unlock(); return d.mas }

func (d *devices) get(dev byte) net.IP { d.mu.Lock(); defer d.mu.Unlock(); return d.ip[dev] }

func (d *devices) all() []net.IP {
	d.mu.Lock()
	defer d.mu.Unlock()
	var out []net.IP
	for _, ip := range d.ip {
		out = append(out, ip)
	}
	return out
}

func listenUDP(port int) (*net.UDPConn, error) {
	c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
	return c, errors.WithStack(err)
}

func online(c0 *net.UDPConn, device byte, name string) (*link.VCDJ, *net.IPNet, error) {
	c0.SetReadDeadline(time.Now().Add(5 * time.Second))
	defer c0.SetReadDeadline(time.Time{})
	buf := make([]byte, 1500)
	for {
		n, _, err := c0.ReadFromUDP(buf)
		if err != nil {
			return nil, nil, errors.WithStack(err)
		}
		k, err := link.DecodeKeepAlive(buf[:n])
		if err != nil || k.Device == device {
			continue
		}
		iface, ipNet, err := link.Iface(k.IP)
		if err != nil {
			return nil, nil, err
		}
		return &link.VCDJ{Device: device, Log: func(s string) { fmt.Fprintln(os.Stderr, s) }, MAC: iface.HardwareAddr, Name: name}, ipNet, nil
	}
}

-    if port == 50001 && n > 0x0a && link.Type(b[0x0a]) == link.TypeMasterRequest {

-      c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})

-      fmt.Fprintf(os.Stderr, "answered master-handoff request from %s\n", src.IP)

+        c2.WriteToUDP(v.MasterResponse(), &net.UDPAddr{IP: src.IP, Port: 50002})

+        fmt.Fprintf(os.Stderr, "answered master-handoff request from %s (yielded)\n", src.IP)

+        fmt.Fprintf(os.Stderr, "received master handoff (0x27) from %s; now master\n", src.IP)


 I'm realizing there's a potential conflict with the message type bytes — 0x26 for the handoff request comes through the beat port (50001) while 0x27 for the response goes through 50002, but my switch statement handles both ports and relies on these distinct type bytes to differentiate the messages, which should work fine since each type only appears on its designated port.

Now I'm checking whether the deck's status packets could collide with anything — on 50001 we have beat (0x28), position (0x0b), and on-air (0x03), none of which overlap with 0x26 or 0x27, and on 50002 the status (0x0a) and media responses (0x05/0x06) don't collide with 0x27 either, so there's no ambiguity in the type checking.

The real concern is whether we'll receive our own status packets back — we won't, since we send status to other devices' unicast addresses and we're bound to 0.0.0.0:50002, so we won't capture our own outgoing unicast traffic, but the decks will send their status to us since we're announced as device 1, and when we become master and emit that status to the deck, the deck yields and drops its master flag, so our master state comes only from the decks' status packets, not from our own emissions.

There's a potential issue though: if the become-master goroutine sends a MasterRequest and the deck never responds with 0x27, we'll never set master to true, but that's acceptable for capture purposes since we'll at least record our own 0x26 attempt, and user input can help trigger the handoff anyway — time to build and test this.


Bash

go build ./... 2>&1 | tail && echo "BUILD OK"

Build master tool with handoff fix


BUILD OK


Builds. Let me re-run the master session — now it'll properly request the handoff from the current master and we should capture the `0x26`/`0x27` exchange.


go run ./cmd/master -device 1 -tempo 124 -playing -out testdata/master.jsonl -secs 40 2>&1; echo "EXIT=$?"

Re-run master session with handoff sequence


Command running in background with ID: bkkwaj6ss. Output is being written to: /private/tmp/claude-501/-Users-noah-conductor-workspaces-deadca7-san-diego/5e143c63-1d6b-4233-91af-bb63e39e8beb/tasks/bkkwaj6ss.output. You will be notified when it completes. To check interim output, use Read on that file path.


The master tool needs to send a 0x26 command to whichever deck is currently set as master—likely deck 3 from before. I should let the process run and see if that deck is still master, then capture the handshake exchange. If needed, I can prompt the user to ensure a deck is marked as master so the handoff request has a valid target.


🎬 **Re-running (40s)** with the proper handoff sequence. To capture the exchange, **please make sure one deck is set as MASTER** (e.g. deck 3 — tap its MASTER button so it's lit). Our tool will observe it's master, send it a `0x26` handoff request, and it should reply `0x27`. If you also **press MASTER on the other deck**, that produces another handoff request directed at us. I'll analyze when the 40s completes.


<task-notification>
<task-id>bkkwaj6ss</task-id>
<tool-use-id>toolu_017JLYpQ8koRSBWB2HDsVXgM</tool-use-id>
<output-file>/private/tmp/claude-501/-Users-noah-conductor-workspaces-deadca7-san-diego/5e143c63-1d6b-4233-91af-bb63e39e8beb/tasks/bkkwaj6ss.output</output-file>
<status>completed</status>
<summary>Background command "Re-run master session with handoff sequence" completed (exit code 0)</summary>
</task-notification>


cd /Users/noah/conductor/workspaces/deadca7/san-diego/link
echo "=== our events ==="; grep -vE '^\{' /private/tmp/claude-501/-Users-noah-conductor-workspaces-deadca7-san-diego/5e143c63-1d6b-4233-91af-bb63e39e8beb/tasks/bkkwaj6ss.output | grep -v EXIT | sort | uniq -c
echo "=== directed packets captured? (0x26/0x27/0x2a) ==="
python3 -c '
import json,collections
c=collections.OrderedDict()
for l in open("testdata/master.jsonl"):
    p=json.loads(l); d=bytes.fromhex(p["data"])
    if d[:10]!=bytes.fromhex("5173707431576d4a4f4c"): continue
    t=d[0x0a]
    if (p["port"],t) in [(50001,0x26),(50002,0x27),(50001,0x2a),(50002,0x26),(50001,0x27)]:
        c[(p["port"],t,p["src"].split(":")[0])]=c.get((p["port"],t,p["src"].split(":")[0]),0)+1
if c:
    for (port,typ,src),n in sorted(c.items()): print("  port=%d type=0x%02x src=%s count=%d <-- DIRECTED"%(port,typ,src,n))
else:
    print("  none captured")
'

Check for directed handoff packets


=== our events ===
   1 no existing master; taking the role
   1 online as device 1
=== directed packets captured? (0x26/0x27/0x2a) ===
  none captured


 No deck was advertising the master flag during this session, so we couldn't initiate a handoff request—we just assumed the master role by default. The user never pressed MASTER on any deck either, which would have sent us a 0x26 handoff. Interestingly, deck 3 had been master in an earlier capture, but not in this particular session.

The directed-packet capture really hinges on having a deck in master state or the user triggering a MASTER button press while we're master. Our implementation is solid—we've verified the emitter works and the decks recognize us (deck 3 proved that earlier with the status and beat round-trip). Rather than keep chasing this without the right deck state, I should either give the user precise instructions for one focused attempt with a deck in master, or acknowledge that the core feature—driving tempo—is built and tested, and the directed-packet capture is just waiting on the right deck configuration.

Actually, I can make this cleaner: have our tool record its own emitted packets (0x26 and 0x2a) directly to the jsonl so the dump contains them regardless of whether a deck responds. That way we capture samples of every packet type we're supposed to emit, and if a deck does respond with 0x27, we'll have that too. For 0x2a, there's no real deck sample to capture since it's a command we emit, not something decks send — our tested builder closes that out. For 0x26 and 0x27, I need the deck in master state to capture the real handoff exchange. Let me run one more focused attempt with explicit instructions to keep deck 3 as master, and if that works we get the real 0x27; if not, we've already verified the builders work and decks recognize our packets. I'll skip the threading complexity and just run the final test with clear deck master instructions to capture that real 0x27 packet.


Still "no existing master" — no deck was advertising the master flag this run, so our handoff request had no target (and no one pressed MASTER toward us). Our emitter is provably correct (deck 3 *yielded* to us earlier; our status/beat round-trip through our own decoders), but capturing the real `0x26`/`0x27` exchange needs a deck actually holding master.

One clean attempt should nail it. The key: **a deck must be MASTER before I start.**


go run ./cmd/master -device 1 -tempo 124 -playing -out testdata/master.jsonl -secs 40 2>&1; echo EXIT=$?

Final master run to capture real handoff


Command running in background with ID: be2bpdy9m. Output is being written to: /private/tmp/claude-501/-Users-noah-conductor-workspaces-deadca7-san-diego/5e143c63-1d6b-4233-91af-bb63e39e8beb/tasks/be2bpdy9m.output. You will be notified when it completes. To check interim output, use Read on that file path.


🎬 **Final run (40s).** Two things make this work — please do **both**:

1. **Before/right now: set deck 3 to MASTER** — tap its MASTER button so the master indicator is **lit** and leave it on. (A deck must currently hold master for our `0x26` request to have a target.)
2. Once it's lit, our tool will request the handoff and deck 3 should reply `0x27`. If you *also* tap **MASTER on deck 4** mid-run, that sends *us* a `0x26` we'll answer.

I'll analyze the moment it finishes — this should capture the real master-handoff exchange and close out the last packets.