Bring wire protocol into full YAW/2 spec compliance

Five interop issues fixed against PROTOCOL.md:

- Join signature now covers nonce_raw || net_ascii (64-char UTF-8 hex
  string) as specified in §5.1, not net_raw_bytes. Both anchor server
  and client updated to match.

- Chat wire fields renamed to spec names: text/ts (Unix ms int64)
  replacing body/sent_at (ISO timestamp). Flat layout on PeerMessage
  matches §8 exactly; store and TUI updated accordingly.

- Direct messages now use the spec "pm" type (flat {type,mid,text,ts})
  instead of chat+to. Receiver reconstructs a ChatMessage with
  dm:<short-id> room for IPC/storage. §8 compliant.

- File transfer message types changed to spec hyphenated names:
  file-offer, file-accept, file-cancel, file-done with spec field
  names (name/size not filename/size_bytes). §9 compliant.

- DataChannel open-race (§14 gotcha #3) fixed with sync.Once: doOpen
  fires on OnOpen callback or immediately if the channel is already
  open when WireDataChannel is called (answerer race).

Also fixes two bugs found during testing:

- mid was missing from outgoing wire messages, causing all received
  messages to arrive with mid="" and collide on the UNIQUE DB
  constraint. mid is now included on all sent chat/pm messages; a
  random mid is generated for any received message that omits it.

- Test scripts hardened: kill -9 + active lsof polling replaces blind
  sleep for port cleanup; join_network sent before peer_field queries
  (local_peer is now network-scoped and nil until joined).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

internal/anchor/client.go

@@ -81,8 +81,8 @@ func runOnce(ctx context.Context, anchorURL, netHash string, id *crypto.Identity
 			if err != nil {
 				return fmt.Errorf("bad challenge nonce: %w", err)
 			}
-			netBytes, _ := hex.DecodeString(netHash)
-			sig := id.Sign(append(nonceBytes, netBytes...))
+			// §5.1: sig covers nonce_raw || net_ascii (64-char hex string as UTF-8)
+			sig := id.Sign(append(nonceBytes, []byte(netHash)...))
 			sendCh <- proto.AnchorMessage{
 				Type: proto.AnchorJoin,
 				ID:   string(id.PeerID()),

internal/ipc/ipc.go

@@ -16,7 +16,6 @@ import (
 	"net"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/waste-go/internal/invite"
 	"github.com/waste-go/internal/netmgr"
 	"github.com/waste-go/internal/proto"
@@ -49,9 +48,11 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) {
 
 	writeCh := make(chan []byte, 128)
 	done := make(chan struct{})
+	writerDone := make(chan struct{})
 
 	// Writer goroutine.
 	go func() {
+		defer close(writerDone)
 		w := bufio.NewWriter(conn)
 		for line := range writeCh {
 			line = append(line, '\n')
@@ -140,30 +141,63 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) {
 				send(errMsg("send_message: not joined to any network"))
 				continue
 			}
-			msg := &proto.ChatMessage{
-				Mid:    randomHex(16),
-				ID:     uuid.NewString(),
-				From:   n.Identity.PeerID(),
-				To:     cmd.To,
-				Room:   cmd.Room,
-				Body:   cmd.Body,
-				SentAt: time.Now(),
-			}
-			payload, err := json.Marshal(proto.PeerMessage{Type: proto.MsgChat, Chat: msg})
-			if err != nil {
-				continue
-			}
+			ts := time.Now().UnixMilli()
 			if cmd.To != nil {
-				n.Mesh.SendTo(*cmd.To, payload)
+				// DM → spec "pm" type: flat {type, mid, text, ts} on the wire
+				mid := randomHex(16)
+				wire, err := json.Marshal(proto.PeerMessage{
+					Type: proto.MsgPm,
+					Mid:  mid,
+					Text: cmd.Body,
+					Ts:   ts,
+				})
+				if err != nil {
+					continue
+				}
+				n.Mesh.SendTo(*cmd.To, wire)
+				// Store locally with dm:<short-id> room convention
+				local := &proto.ChatMessage{
+					Mid:  mid,
+					From: n.Identity.PeerID(),
+					To:   cmd.To,
+					Room: "dm:" + (*cmd.To).Short(),
+					Text: cmd.Body,
+					Ts:   ts,
+				}
+				n.Mesh.SaveMessage(local)
+				n.Mesh.Emit(proto.IpcMessage{
+					Type:      proto.EvtMessageReceived,
+					NetworkID: n.ID,
+					Message:   local,
+				})
 			} else {
-				n.Mesh.Broadcast(payload)
+				// Group chat → spec "chat" type: flat {type, mid, room, text, ts}
+				mid := randomHex(16)
+				wire, err := json.Marshal(proto.PeerMessage{
+					Type: proto.MsgChat,
+					Mid:  mid,
+					Room: cmd.Room,
+					Text: cmd.Body,
+					Ts:   ts,
+				})
+				if err != nil {
+					continue
+				}
+				n.Mesh.Broadcast(wire)
+				local := &proto.ChatMessage{
+					Mid:  mid,
+					From: n.Identity.PeerID(),
+					Room: cmd.Room,
+					Text: cmd.Body,
+					Ts:   ts,
+				}
+				n.Mesh.SaveMessage(local)
+				n.Mesh.Emit(proto.IpcMessage{
+					Type:      proto.EvtMessageReceived,
+					NetworkID: n.ID,
+					Message:   local,
+				})
 			}
-			n.Mesh.SaveMessage(msg)
-			n.Mesh.Emit(proto.IpcMessage{
-				Type:      proto.EvtMessageReceived,
-				NetworkID: n.ID,
-				Message:   msg,
-			})
 
 		case proto.CmdGetState:
 			send(stateSnapshot(mgr))
@@ -222,6 +256,7 @@ func handleClient(conn net.Conn, mgr *netmgr.Manager) {
 
 	close(done)
 	close(writeCh)
+	<-writerDone // wait for writer to flush before conn.Close() fires
 	log.Printf("ipc: UI client disconnected")
 }
 

internal/mesh/peer.go

@@ -2,10 +2,13 @@
 package mesh
 
 import (
+	"crypto/rand"
 	"encoding/hex"
 	"encoding/json"
 	"log"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/pion/webrtc/v3"
 
@@ -21,7 +24,7 @@ type Anchor interface {
 }
 
 // WireDataChannel sets up open/message/close handlers on a "yaw" DataChannel.
-// Must be called before the DataChannel opens.
+// Safe to call whether the channel is already open or not (§6 open-race).
 func WireDataChannel(
 	dc *webrtc.DataChannel,
 	pc *webrtc.PeerConnection,
@@ -31,7 +34,8 @@ func WireDataChannel(
 ) {
 	sendCh := make(chan []byte, 64)
 
-	dc.OnOpen(func() {
+	var once sync.Once
+	doOpen := func() {
 		log.Printf("peer: DataChannel open with %s", peerID.Short())
 
 		// Send hello — bind our identity to this DTLS session.
@@ -69,7 +73,13 @@ func WireDataChannel(
 				}
 			}
 		}()
-	})
+	}
+
+	dc.OnOpen(func() { once.Do(doOpen) })
+	// §6 gotcha: answerer's DC may already be open when OnDataChannel fires.
+	if dc.ReadyState() == webrtc.DataChannelStateOpen {
+		once.Do(doOpen)
+	}
 
 	dc.OnMessage(func(msg webrtc.DataChannelMessage) {
 		if msg.IsString {
@@ -152,8 +162,30 @@ func handleDCMessage(data []byte, from proto.PeerID, localID *crypto.Identity, m
 
 func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
 	switch msg.Type {
+	case proto.MsgChat:
+		chat := &proto.ChatMessage{
+			Mid:  midOrRandom(msg.Mid),
+			From: from,
+			Room: msg.Room,
+			Text: msg.Text,
+			Ts:   msg.Ts,
+		}
+		m.SaveMessage(chat)
+		m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: chat})
+
+	case proto.MsgPm:
+		// Private message — reconstruct as ChatMessage for IPC/storage using dm:<short-id> room.
+		chat := &proto.ChatMessage{
+			Mid:  midOrRandom(msg.Mid),
+			From: from,
+			Room: "dm:" + from.Short(),
+			Text: msg.Text,
+			Ts:   msg.Ts,
+		}
+		m.SaveMessage(chat)
+		m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: chat})
+
 	case proto.MsgFileListReq:
-		// Peer wants our file list — reply directly on their send channel.
 		files := m.ScanShareDir()
 		resp, err := json.Marshal(proto.PeerMessage{
 			Type:         proto.MsgFileListResp,
@@ -165,7 +197,6 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
 		m.SendTo(from, resp)
 
 	case proto.MsgFileListResp:
-		// Received a remote peer's file list — forward to IPC subscribers.
 		if msg.FileListResp != nil {
 			m.Emit(proto.IpcMessage{
 				Type:   proto.EvtFileList,
@@ -174,11 +205,13 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
 			})
 		}
 
-	case proto.MsgChat:
-		if msg.Chat != nil {
-			m.SaveMessage(msg.Chat)
-			m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: msg.Chat})
-		}
+	case proto.MsgFileOffer:
+		m.Emit(proto.IpcMessage{
+			Type:   proto.EvtIncomingFile,
+			PeerID: peerIDPtr(from),
+			Offer:  &proto.FileOffer{Xid: msg.Xid, Name: msg.Name, Size: msg.Size, SHA256: msg.SHA256},
+		})
+
 	case proto.MsgPeerGossip:
 		if msg.Gossip != nil {
 			log.Printf("mesh: gossip from %s: %d peer hints", from.Short(), len(msg.Gossip.Peers))
@@ -187,19 +220,24 @@ func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
 		log.Printf("mesh: ping from %s", from.Short())
 	case proto.MsgPong:
 		log.Printf("mesh: pong from %s", from.Short())
-	case proto.MsgFileOffer:
-		if msg.FileOffer != nil {
-			m.Emit(proto.IpcMessage{
-				Type:   proto.EvtIncomingFile,
-				PeerID: peerIDPtr(from),
-				Offer:  msg.FileOffer,
-			})
-		}
 	default:
 		log.Printf("mesh: unknown message type %q from %s", msg.Type, from.Short())
 	}
 }
 
+// midOrRandom returns mid if non-empty, otherwise generates a random 16-byte hex string.
+// Ensures every stored message has a unique mid even from peers that don't send one.
+func midOrRandom(mid string) string {
+	if mid != "" {
+		return mid
+	}
+	b := make([]byte, 16)
+	if _, err := rand.Read(b); err != nil {
+		return hex.EncodeToString([]byte(time.Now().String()))
+	}
+	return hex.EncodeToString(b)
+}
+
 func dtlsFingerprints(pc *webrtc.PeerConnection) (local, remote []byte) {
 	if ld := pc.LocalDescription(); ld != nil {
 		local = fingerprintFromSDP(ld.SDP)

internal/proto/proto.go

@@ -35,40 +35,64 @@ type MsgType string
 
 const (
 	MsgChat         MsgType = "chat"
+	MsgPm           MsgType = "pm"           // private message, §8
 	MsgPeerGossip   MsgType = "peer_gossip"
 	MsgFileListReq  MsgType = "file_list_req"
 	MsgFileListResp MsgType = "file_list_resp"
-	MsgFileOffer    MsgType = "file_offer"
-	MsgFileResp     MsgType = "file_response"
-	MsgFileDone     MsgType = "file_done"
+	MsgFileOffer    MsgType = "file-offer"   // §9, hyphenated per spec
+	MsgFileAccept   MsgType = "file-accept"
+	MsgFileCancel   MsgType = "file-cancel"
+	MsgFileDone     MsgType = "file-done"
 	MsgPing         MsgType = "ping"
 	MsgPong         MsgType = "pong"
 )
 
+// PmMessage is a private message sent directly over a single peer link (§8 "pm").
+// The sender/receiver are implicit from the DataChannel; no room or from fields on the wire.
+type PmMessage struct {
+	Text string `json:"text"`
+	Ts   int64  `json:"ts"` // Unix milliseconds
+}
+
 // PeerMessage is the top-level container sent over the "yaw" DataChannel.
+// The spec types (hello, chat, pm, file-offer …) are flat JSON objects; we
+// embed the fields directly using inline structs where needed, but for structured
+// types we include the payload pointer. Unknown fields are ignored (forward compat).
 // File chunks go over a separate binary DataChannel labeled "f:<xid>".
 type PeerMessage struct {
 	Type MsgType `json:"type"`
 
-	// Only one of these will be set, depending on Type.
-	Chat         *ChatMessage  `json:"chat,omitempty"`
+	// chat / pm fields (flat on the wire per spec §8)
+	Mid  string `json:"mid,omitempty"`  // optional dedup id; required when relay hops > 0
+	Room string `json:"room,omitempty"` // chat only
+	Text string `json:"text,omitempty"` // chat and pm
+	Ts   int64  `json:"ts,omitempty"`   // chat and pm (Unix ms)
+
+	// Non-spec extensions (unknown types are silently ignored by other impls)
 	Gossip       *PeerGossip   `json:"gossip,omitempty"`
 	FileListResp *FileListResp `json:"file_list_resp,omitempty"`
-	FileOffer    *FileOffer    `json:"file_offer,omitempty"`
-	FileResp     *FileResponse `json:"file_response,omitempty"`
-	FileDone     *FileDone     `json:"file_done,omitempty"`
-	Seq          *uint64       `json:"seq,omitempty"` // for ping/pong
+
+	// file transfer (§9) — fields are flat on the wire
+	Xid    string `json:"xid,omitempty"`
+	Name   string `json:"name,omitempty"`
+	Size   int64  `json:"size,omitempty"`
+	SHA256 string `json:"sha256,omitempty"`
+
+	// file-done / file-cancel / file-accept just need xid (already above)
+	Reason string `json:"reason,omitempty"` // file-cancel
+
+	Seq *uint64 `json:"seq,omitempty"` // ping/pong
 }
 
-// ChatMessage is a message to a room or a DM.
+// ChatMessage is a group chat message (wire type "chat", §8).
+// Also used internally for persisting PMs after they are received.
 type ChatMessage struct {
-	Mid    string    `json:"mid"`          // random 16-byte hex, for deduplication (YAW/2 §8)
-	ID     string    `json:"id"`           // internal uuid, kept for local use
-	From   PeerID    `json:"from"`
-	To     *PeerID   `json:"to,omitempty"` // nil = broadcast to room
-	Room   string    `json:"room"`
-	Body   string    `json:"body"`
-	SentAt time.Time `json:"sent_at"`
+	Mid  string `json:"mid,omitempty"` // optional dedup id (required when relay hops > 0)
+	From PeerID `json:"from,omitempty"` // set by receiver from DC context; not on wire for pm
+	To   *PeerID `json:"to,omitempty"`  // internal only — not transmitted; set for DMs
+	Room string `json:"room"`
+	Text string `json:"text"`
+	Ts   int64  `json:"ts"` // Unix milliseconds
 }
 
 // PeerGossip shares known peer addresses.
@@ -95,27 +119,13 @@ type FileListResp struct {
 	Files []FileEntry `json:"files"`
 }
 
-// FileOffer initiates a file transfer.
+// FileOffer is used internally when emitting EvtIncomingFile to the IPC layer.
+// On the wire, file-offer fields are flat inside PeerMessage (xid/name/size/sha256).
 type FileOffer struct {
-	Mid       string `json:"mid"`        // dedup id
-	Xid       string `json:"xid"`        // transfer id, used as DataChannel label "f:<xid>"
-	Filename  string `json:"filename"`
-	SizeBytes int64  `json:"size_bytes"`
-	SHA256    string `json:"sha256"`     // hex
-}
-
-// FileResponse accepts or declines a FileOffer.
-type FileResponse struct {
-	Mid      string `json:"mid"`
-	Xid      string `json:"xid"`
-	Accepted bool   `json:"accepted"`
-}
-
-// FileDone signals that all chunks have been sent. Receiver verifies SHA256.
-type FileDone struct {
-	Mid    string `json:"mid"`
 	Xid    string `json:"xid"`
-	SHA256 string `json:"sha256"` // hex
+	Name   string `json:"name"`
+	Size   int64  `json:"size"`
+	SHA256 string `json:"sha256"`
 }
 
 // ── DataChannel hello (YAW/2 §6) ─────────────────────────────────────────────

internal/store/store.go

@@ -57,10 +57,11 @@ func (s *Store) Close() error {
 // SaveMessage persists a chat message. Duplicate mids are silently ignored
 // (INSERT OR IGNORE), so calling this more than once is safe.
 func (s *Store) SaveMessage(msg *proto.ChatMessage) error {
+	sentAt := time.UnixMilli(msg.Ts).UTC()
 	_, err := s.db.Exec(
 		`INSERT OR IGNORE INTO messages (mid, room, from_peer, body, sent_at)
 		 VALUES (?, ?, ?, ?, ?)`,
-		msg.Mid, msg.Room, string(msg.From), msg.Body, msg.SentAt.UTC(),
+		msg.Mid, msg.Room, string(msg.From), msg.Text, sentAt,
 	)
 	return err
 }
@@ -96,12 +97,12 @@ func (s *Store) RecentMessages(room string, limit int) ([]proto.ChatMessage, err
 		var m proto.ChatMessage
 		var from string
 		var sentAt time.Time
-		if err := rows.Scan(&m.Mid, &from, &m.Body, &sentAt); err != nil {
+		if err := rows.Scan(&m.Mid, &from, &m.Text, &sentAt); err != nil {
 			return nil, err
 		}
 		m.From = proto.PeerID(from)
 		m.Room = room
-		m.SentAt = sentAt
+		m.Ts = sentAt.UnixMilli()
 		msgs = append(msgs, m)
 	}
 	// Reverse so oldest-first.

internal/store/store_test.go

@@ -16,11 +16,11 @@ func TestRoundTrip(t *testing.T) {
 	defer st.Close()
 
 	msg := &proto.ChatMessage{
-		Mid:    "aabbccdd00112233",
-		From:   proto.PeerID("deadbeef"),
-		Room:   "general",
-		Body:   "hello world",
-		SentAt: time.Now().UTC().Truncate(time.Second),
+		Mid:  "aabbccdd00112233",
+		From: proto.PeerID("deadbeef"),
+		Room: "general",
+		Text: "hello world",
+		Ts:   time.Now().UTC().Truncate(time.Second).UnixMilli(),
 	}
 
 	if err := st.SaveMessage(msg); err != nil {
@@ -40,7 +40,7 @@ func TestRoundTrip(t *testing.T) {
 		t.Fatalf("got %d messages, want 1", len(msgs))
 	}
 	got := msgs[0]
-	if got.Mid != msg.Mid || got.Body != msg.Body || got.Room != msg.Room {
+	if got.Mid != msg.Mid || got.Text != msg.Text || got.Room != msg.Room {
 		t.Fatalf("message mismatch: %+v", got)
 	}
 }
@@ -79,11 +79,11 @@ func TestRecentMessagesOrdering(t *testing.T) {
 	base := time.Now().UTC().Truncate(time.Second)
 	for i := range 5 {
 		st.SaveMessage(&proto.ChatMessage{
-			Mid:    string(rune('a'+i)) + "000000000000000",
-			From:   "deadbeef",
-			Room:   "general",
-			Body:   string(rune('a' + i)),
-			SentAt: base.Add(time.Duration(i) * time.Second),
+			Mid:  string(rune('a'+i)) + "000000000000000",
+			From: "deadbeef",
+			Room: "general",
+			Text: string(rune('a' + i)),
+			Ts:   base.Add(time.Duration(i) * time.Second).UnixMilli(),
 		})
 	}
 
@@ -96,7 +96,7 @@ func TestRecentMessagesOrdering(t *testing.T) {
 	}
 	// Must be oldest-first.
 	for i := 1; i < len(msgs); i++ {
-		if msgs[i].SentAt.Before(msgs[i-1].SentAt) {
+		if msgs[i].Ts < msgs[i-1].Ts {
 			t.Fatalf("messages not in ascending order at index %d", i)
 		}
 	}