Merge pull request 'Docs: refresh the QA plan against the live task list' (#92) from task/483-design-offload-ml-to-the-workstation-kee into master

Reviewed-on: #92

.gitignore

@@ -66,3 +66,6 @@ coverage.out
 /HANDOFF.md
 /models/stt
 /models/tts
+
+# root .env — MAVEN_AMBIENT_TOKEN and friends, same class as deploy/telegram.env
+.env

internal/ipc/client.go

@@ -8,13 +8,15 @@ import (
 	"net"
 	"sync"
 	"time"
+
+	"github.com/kami/maven/internal/netaddr"
 )
 
-// Client — the module side of the boundary. Wraps a unix-socket connection
-// and satisfies CoreAPI, so a module imports ipc, holds a CoreAPI, and is
-// agnostic to whether it's been wired in-process (tests / daemon-embedded)
-// or over this socket (full topology). The swappability is the seam auth
-// will insert into without touching module code.
+// Client — the module side of the boundary. Wraps a connection to core and
+// satisfies CoreAPI, so a module imports ipc, holds a CoreAPI, and is
+// agnostic to whether it's been wired in-process (tests / daemon-embedded),
+// over a local unix socket, or over tcp to another host. The swappability is
+// the seam auth will insert into without touching module code.
 //
 // One Client ⇒ one conn ⇒ one concurrent request at a time. A module that
 // wants parallel requests opens one Client per goroutine; the store is the
@@ -22,7 +24,8 @@ import (
 // per-Client lock keeps frame interleaving impossible by construction.
 type Client struct {
 	conn net.Conn
-	path string // kept so a dropped conn can be re-dialed (core restart)
+	path string       // the address as configured, kept for errors and logs
+	addr netaddr.Addr // parsed, so a dropped conn can be re-dialed (core restart)
 	mu   sync.Mutex
 }
 
@@ -81,14 +84,22 @@ var readOnlyMethods = map[Method]bool{
 	MethodPing:                  true,
 }
 
-// Dial connects to a core socket at path and returns a Client. The module
-// owns its Client lifecycle; Close on shutdown.
+// Dial connects to core at path and returns a Client. The module owns its
+// Client lifecycle; Close on shutdown.
+//
+// path is a netaddr seam address: a bare path is the unix socket it has
+// always been, and "tcp://host:port?token=..." reaches a core on another
+// host. See internal/netaddr.
 func Dial(path string) (*Client, error) {
-	c, err := net.Dial("unix", path)
+	addr, err := netaddr.Parse(path)
 	if err != nil {
-		return nil, fmt.Errorf("ipc: dial %s: %w", path, err)
+		return nil, err
 	}
-	return &Client{conn: c, path: path}, nil
+	c, err := netaddr.Dial(addr)
+	if err != nil {
+		return nil, fmt.Errorf("ipc: dial %s: %w", addr, err)
+	}
+	return &Client{conn: c, path: path, addr: addr}, nil
 }
 
 func (c *Client) Close() error {
@@ -189,9 +200,9 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error {
 // re-dials clean. Caller holds c.mu.
 func (c *Client) roundtrip(m Method, raw json.RawMessage, resp *Response) error {
 	if c.conn == nil {
-		conn, err := net.Dial("unix", c.path)
+		conn, err := netaddr.Dial(c.addr)
 		if err != nil {
-			return fmt.Errorf("%w: dial %s: %v", errWriteLost, c.path, err)
+			return fmt.Errorf("%w: dial %s: %v", errWriteLost, c.addr, err)
 		}
 		c.conn = conn
 	}

internal/ipc/server.go

@@ -8,11 +8,11 @@ import (
 	"fmt"
 	"log"
 	"net"
-	"os"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/kami/maven/internal/netaddr"
 	"github.com/kami/maven/internal/store"
 	"golang.org/x/sys/unix"
 )
@@ -446,6 +446,7 @@ func mapErr(err error) error {
 type Server struct {
 	api  atomic.Value // stores CoreAPI
 	path string
+	addr netaddr.Addr
 
 	ln     net.Listener
 	wg     sync.WaitGroup
@@ -610,31 +611,29 @@ type CheckFunc func(ctx context.Context, m Method, params json.RawMessage) error
 // MethodAssertStepUp dispatch calls this instead of going through CoreAPI.
 type StepUpFunc func(ctx context.Context) error
 
-// Listen creates a Server bound to path. path's parent dir must exist and be
-// 0700 (we chmod it if we own it); the socket file itself is created 0600 so
-// only the same unix user can connect — the current "auth floor", same radius
-// as wg at the network boundary. Removing a stale socket at path first lets
-// the daemon restart cleanly.
+// Listen creates a Server bound to path.
+//
+// A bare path is a unix socket, unchanged: its parent dir is 0700 and the
+// socket file itself is 0600, so only the same unix user can connect — the
+// current "auth floor", same radius as wg at the network boundary. A stale
+// socket is removed first so the daemon restarts cleanly.
+//
+// A "tcp://host:port?token=..." address binds a network listener instead, for
+// a module that lives on another host. There is no filesystem there to be the
+// auth floor, so netaddr checks the shared token before this package sees the
+// connection and a token is mandatory. See internal/netaddr.
 func Listen(path string, api CoreAPI) (*Server, error) {
-	_ = os.Remove(path) // stale socket from a crashed daemon; ignore missing
-	if err := os.MkdirAll(parentDir(path), 0o700); err != nil {
-		return nil, fmt.Errorf("ipc: mkdir socket dir: %w", err)
-	}
-	// umask could widen the perms on socket creation; tighten then chmod to
-	// be explicit. 0600 ⇒ read+write by owner only.
-	oldMask := unix.Umask(0o077)
-	ln, err := net.Listen("unix", path)
-	unix.Umask(oldMask)
+	addr, err := netaddr.Parse(path)
 	if err != nil {
-		return nil, fmt.Errorf("ipc: listen %s: %w", path, err)
+		return nil, err
 	}
-	if err := os.Chmod(path, 0o600); err != nil {
-		_ = ln.Close()
-		_ = os.Remove(path)
-		return nil, fmt.Errorf("ipc: chmod socket: %w", err)
+	ln, err := netaddr.Listen(addr)
+	if err != nil {
+		return nil, err
 	}
 	s := &Server{
 		path: path,
+		addr: addr,
 		ln:   ln,
 		done: make(chan struct{}),
 	}
@@ -1268,7 +1267,7 @@ func (s *Server) Close() error {
 		// missing the seal costs every write since the last clean shutdown.
 		log.Printf("ipc: %d connection(s) still busy after %s, closing anyway", s.liveConns(), closeGrace)
 	}
-	_ = os.Remove(s.path)
+	netaddr.Cleanup(s.addr)
 	return err
 }
 
@@ -1338,25 +1337,6 @@ func (s *Server) Path() string { return s.path }
 // while the server is serving (dispatch loads api once per request via atomic).
 func (s *Server) SetAPI(api CoreAPI) { s.api.Store(api) }
 
-func parentDir(p string) string {
-	if i := lastIndexByte(p, '/'); i >= 0 {
-		if i == 0 {
-			return "/"
-		}
-		return p[:i]
-	}
-	return "."
-}
-
-func lastIndexByte(s string, b byte) int {
-	for i := len(s) - 1; i >= 0; i-- {
-		if s[i] == b {
-			return i
-		}
-	}
-	return -1
-}
-
 // peerCaller — read SO_PEERCRED off a unix conn to identify the connecting
 // process. Returns ok=false on a non-unix conn or a platform without
 // SO_PEERCRED; the caller then proceeds without a Caller (the socket perms

internal/netaddr/netaddr.go

@@ -0,0 +1,277 @@
+// Package netaddr parses a daemon seam address and dials or binds it.
+//
+// Every seam between Maven's daemons used to be a unix socket with the
+// network hardcoded at the call site — two dials in internal/ipc, one listen,
+// and the same pair again in internal/worker. That is correct for co-located
+// daemons and it is the reason a module cannot live on another host. This
+// package moves the choice into the address string so a deploy picks the
+// transport, not a recompile:
+//
+//	/run/maven/stt.sock                   unix   (the default, unchanged)
+//	unix:///run/maven/stt.sock            unix   (explicit, same thing)
+//	tcp://workstation:9310?token=hunter2  tcp
+//
+// A scheme-less address is unix and behaves exactly as it did before this
+// package existed: same 0700 parent dir, same 0600 socket, same bytes on the
+// wire with no handshake in front of them.
+//
+// Over TCP the filesystem permission that authenticated the unix socket is
+// gone, and what crosses this seam is audio of the owner speaking and the
+// text of his turns. So a TCP seam carries a shared token, checked before the
+// first protocol frame is read. Wireguard is supported underneath and is not
+// required.
+package netaddr
+
+import (
+	"crypto/subtle"
+	"errors"
+	"fmt"
+	"net"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// ErrUnauthorized — the peer presented a token the listener does not accept,
+// or presented none when one is required.
+var ErrUnauthorized = errors.New("netaddr: unauthorized")
+
+// Addr is a parsed seam endpoint.
+type Addr struct {
+	// Network is "unix" or "tcp".
+	Network string
+	// Address is the socket path (unix) or host:port (tcp).
+	Address string
+	// Token is the shared secret for a tcp seam. Empty for unix, where the
+	// filesystem does the same job.
+	Token string
+}
+
+// String renders the address for logs and errors. The token is never included.
+func (a Addr) String() string {
+	if a.Network == "unix" {
+		return a.Address
+	}
+	return a.Network + "://" + a.Address
+}
+
+// IsUnix reports whether this seam is a unix socket, and so is local, is
+// authenticated by file permissions, and needs no handshake.
+func (a Addr) IsUnix() bool { return a.Network == "unix" }
+
+// Parse reads a seam address. Anything without a "scheme://" prefix is a unix
+// socket path, which keeps every existing config and every default working
+// untouched.
+func Parse(s string) (Addr, error) {
+	if !strings.Contains(s, "://") {
+		return Addr{Network: "unix", Address: s}, nil
+	}
+	u, err := url.Parse(s)
+	if err != nil {
+		return Addr{}, fmt.Errorf("netaddr: parse %q: %w", s, err)
+	}
+	switch u.Scheme {
+	case "unix":
+		return Addr{Network: "unix", Address: u.Path}, nil
+	case "tcp":
+		if u.Host == "" {
+			return Addr{}, fmt.Errorf("netaddr: %q has no host:port", s)
+		}
+		return Addr{Network: "tcp", Address: u.Host, Token: u.Query().Get("token")}, nil
+	default:
+		return Addr{}, fmt.Errorf("netaddr: unsupported scheme %q", u.Scheme)
+	}
+}
+
+// MustParse is Parse for a literal known good at compile time. It panics on a
+// bad address, so use it in tests and constants, never on config input.
+func MustParse(s string) Addr {
+	a, err := Parse(s)
+	if err != nil {
+		panic(err)
+	}
+	return a
+}
+
+// handshakeTimeout bounds the token exchange. A peer that cannot write one
+// short line in this long is not going to serve a turn either.
+const handshakeTimeout = 5 * time.Second
+
+// greeting prefixes the token line. Versioned so a later mTLS seam can be
+// told apart from this one on the wire.
+const greeting = "MAVEN1 "
+
+// Dial connects to a. On a tcp seam it sends the token and waits for the
+// listener to accept it, so a returned conn is already authorized and the
+// caller can write its first protocol frame.
+func Dial(a Addr) (net.Conn, error) {
+	return DialTimeout(a, 0)
+}
+
+// DialTimeout is Dial with a bound on the connect. Zero means the operating
+// system default. The token exchange gets its own timeout either way.
+func DialTimeout(a Addr, timeout time.Duration) (net.Conn, error) {
+	var c net.Conn
+	var err error
+	if timeout > 0 {
+		c, err = net.DialTimeout(a.Network, a.Address, timeout)
+	} else {
+		c, err = net.Dial(a.Network, a.Address)
+	}
+	if err != nil {
+		return nil, err
+	}
+	if a.IsUnix() {
+		return c, nil
+	}
+	if err := clientHandshake(c, a.Token); err != nil {
+		_ = c.Close()
+		return nil, err
+	}
+	return c, nil
+}
+
+func clientHandshake(c net.Conn, token string) error {
+	_ = c.SetDeadline(time.Now().Add(handshakeTimeout))
+	defer c.SetDeadline(time.Time{})
+	if _, err := c.Write([]byte(greeting + token + "\n")); err != nil {
+		return fmt.Errorf("netaddr: send token: %w", err)
+	}
+	var reply [3]byte
+	if _, err := readFull(c, reply[:]); err != nil {
+		return fmt.Errorf("%w: %v", ErrUnauthorized, err)
+	}
+	if string(reply[:]) != "ok\n" {
+		return ErrUnauthorized
+	}
+	return nil
+}
+
+// Listener wraps a net.Listener so Accept performs the token check for a tcp
+// seam. A connection that fails the check is closed and never surfaces, so
+// the protocol above this layer only ever sees authorized peers.
+type Listener struct {
+	net.Listener
+	addr Addr
+}
+
+// Accept returns the next authorized connection. Unauthorized peers are
+// dropped and Accept keeps waiting: a bad token is a rejected stranger, not a
+// reason to stop serving.
+func (l *Listener) Accept() (net.Conn, error) {
+	for {
+		c, err := l.Listener.Accept()
+		if err != nil {
+			return nil, err
+		}
+		if l.addr.IsUnix() {
+			return c, nil
+		}
+		if err := serverHandshake(c, l.addr.Token); err != nil {
+			_ = c.Close()
+			continue
+		}
+		return c, nil
+	}
+}
+
+// Addr reports the parsed seam address this listener was built from.
+func (l *Listener) SeamAddr() Addr { return l.addr }
+
+func serverHandshake(c net.Conn, want string) error {
+	_ = c.SetDeadline(time.Now().Add(handshakeTimeout))
+	defer c.SetDeadline(time.Time{})
+	// The line is bounded: greeting, token, newline. Read a byte at a time so
+	// nothing of the first protocol frame is consumed when the token is short.
+	line := make([]byte, 0, 128)
+	var b [1]byte
+	for {
+		if _, err := readFull(c, b[:]); err != nil {
+			return err
+		}
+		if b[0] == '\n' {
+			break
+		}
+		line = append(line, b[0])
+		if len(line) > 512 {
+			return ErrUnauthorized
+		}
+	}
+	got, ok := strings.CutPrefix(string(line), greeting)
+	if !ok {
+		return ErrUnauthorized
+	}
+	if subtle.ConstantTimeCompare([]byte(got), []byte(want)) != 1 {
+		return ErrUnauthorized
+	}
+	if _, err := c.Write([]byte("ok\n")); err != nil {
+		return err
+	}
+	return nil
+}
+
+func readFull(c net.Conn, p []byte) (int, error) {
+	n := 0
+	for n < len(p) {
+		m, err := c.Read(p[n:])
+		n += m
+		if err != nil {
+			return n, err
+		}
+	}
+	return n, nil
+}
+
+// Listen binds a. A unix seam gets the perms it has always had: parent dir
+// 0700, socket 0600, and any stale socket from a crashed daemon removed
+// first. A tcp seam must carry a token, because there is no filesystem to
+// stand in for one.
+func Listen(a Addr) (*Listener, error) {
+	if a.IsUnix() {
+		ln, err := listenUnix(a.Address)
+		if err != nil {
+			return nil, err
+		}
+		return &Listener{Listener: ln, addr: a}, nil
+	}
+	if a.Token == "" {
+		return nil, fmt.Errorf("netaddr: listen %s: tcp seam requires a token", a)
+	}
+	ln, err := net.Listen("tcp", a.Address)
+	if err != nil {
+		return nil, fmt.Errorf("netaddr: listen %s: %w", a, err)
+	}
+	return &Listener{Listener: ln, addr: a}, nil
+}
+
+func listenUnix(path string) (net.Listener, error) {
+	_ = os.Remove(path) // stale socket from a crashed daemon; ignore missing
+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+		return nil, fmt.Errorf("netaddr: mkdir socket dir: %w", err)
+	}
+	// umask could widen the perms on socket creation; tighten then chmod to
+	// be explicit. 0600 ⇒ read+write by owner only.
+	oldMask := unix.Umask(0o077)
+	ln, err := net.Listen("unix", path)
+	unix.Umask(oldMask)
+	if err != nil {
+		return nil, fmt.Errorf("netaddr: listen %s: %w", path, err)
+	}
+	if err := os.Chmod(path, 0o600); err != nil {
+		_ = ln.Close()
+		_ = os.Remove(path)
+		return nil, fmt.Errorf("netaddr: chmod socket: %w", err)
+	}
+	return ln, nil
+}
+
+// Cleanup removes the socket file behind a unix seam. It is a no-op for tcp.
+func Cleanup(a Addr) {
+	if a.IsUnix() && a.Address != "" {
+		_ = os.Remove(a.Address)
+	}
+}

internal/netaddr/netaddr_test.go

@@ -0,0 +1,185 @@
+package netaddr
+
+import (
+	"errors"
+	"net"
+	"path/filepath"
+	"testing"
+)
+
+// A scheme-less address must stay unix. Every deploy in the tree writes a bare
+// path, so this is the test that says the transport change costs them nothing.
+func TestParseSchemelessIsUnix(t *testing.T) {
+	a, err := Parse("/run/maven/stt.sock")
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if !a.IsUnix() {
+		t.Fatalf("want unix, got %q", a.Network)
+	}
+	if a.Address != "/run/maven/stt.sock" {
+		t.Fatalf("address = %q", a.Address)
+	}
+	if a.Token != "" {
+		t.Fatalf("unix seam carries a token: %q", a.Token)
+	}
+}
+
+func TestParse(t *testing.T) {
+	cases := []struct {
+		in            string
+		net, addr, tk string
+		wantErr       bool
+	}{
+		{in: "", net: "unix", addr: ""},
+		{in: "unix:///run/maven/core.sock", net: "unix", addr: "/run/maven/core.sock"},
+		{in: "tcp://workstation:9310", net: "tcp", addr: "workstation:9310"},
+		{in: "tcp://workstation:9310?token=hunter2", net: "tcp", addr: "workstation:9310", tk: "hunter2"},
+		{in: "tcp://", wantErr: true},
+		{in: "udp://workstation:9310", wantErr: true},
+	}
+	for _, c := range cases {
+		a, err := Parse(c.in)
+		if c.wantErr {
+			if err == nil {
+				t.Errorf("Parse(%q) = %v, want error", c.in, a)
+			}
+			continue
+		}
+		if err != nil {
+			t.Errorf("Parse(%q): %v", c.in, err)
+			continue
+		}
+		if a.Network != c.net || a.Address != c.addr || a.Token != c.tk {
+			t.Errorf("Parse(%q) = %+v, want %s/%s/%s", c.in, a, c.net, c.addr, c.tk)
+		}
+	}
+}
+
+// The token must never reach a log line.
+func TestStringHidesToken(t *testing.T) {
+	a := MustParse("tcp://workstation:9310?token=hunter2")
+	if got := a.String(); got != "tcp://workstation:9310" {
+		t.Fatalf("String() = %q", got)
+	}
+}
+
+// A unix seam must round-trip with no handshake in front of the payload: the
+// first bytes the listener sees are the caller's, exactly as before.
+func TestUnixRoundTripHasNoHandshake(t *testing.T) {
+	a := MustParse(filepath.Join(t.TempDir(), "s.sock"))
+	ln, err := Listen(a)
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	defer ln.Close()
+	go echoOnce(ln)
+
+	c, err := Dial(a)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer c.Close()
+	if got := roundTrip(t, c, "hello"); got != "hello" {
+		t.Fatalf("got %q", got)
+	}
+}
+
+func TestTCPRoundTripWithToken(t *testing.T) {
+	ln, addr := listenLoopback(t, "s3cret")
+	defer ln.Close()
+	go echoOnce(ln)
+
+	c, err := Dial(addr)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer c.Close()
+	if got := roundTrip(t, c, "hello"); got != "hello" {
+		t.Fatalf("got %q", got)
+	}
+}
+
+func TestTCPWrongTokenIsRejected(t *testing.T) {
+	ln, addr := listenLoopback(t, "s3cret")
+	defer ln.Close()
+	// Accept keeps waiting past the bad peer, so nothing here should ever
+	// reach the echo. A conn that does means the token was not checked.
+	go echoOnce(ln)
+
+	bad := addr
+	bad.Token = "wrong"
+	if _, err := Dial(bad); !errors.Is(err, ErrUnauthorized) {
+		t.Fatalf("dial with wrong token: err = %v, want ErrUnauthorized", err)
+	}
+}
+
+// A stranger that speaks the protocol instead of the greeting is dropped, and
+// the listener stays up for the peer that follows it.
+func TestTCPUngreetedPeerDoesNotKillTheListener(t *testing.T) {
+	ln, addr := listenLoopback(t, "s3cret")
+	defer ln.Close()
+	go echoOnce(ln)
+
+	raw, err := net.Dial("tcp", addr.Address)
+	if err != nil {
+		t.Fatalf("raw dial: %v", err)
+	}
+	if _, err := raw.Write([]byte("GET / HTTP/1.1\n")); err != nil {
+		t.Fatalf("raw write: %v", err)
+	}
+	raw.Close()
+
+	c, err := Dial(addr)
+	if err != nil {
+		t.Fatalf("dial after stranger: %v", err)
+	}
+	defer c.Close()
+	if got := roundTrip(t, c, "still here"); got != "still here" {
+		t.Fatalf("got %q", got)
+	}
+}
+
+// A tcp seam with no token is a misconfiguration, and it must fail at bind
+// rather than serve the owner's turns to anyone who connects.
+func TestTCPListenRequiresToken(t *testing.T) {
+	if _, err := Listen(MustParse("tcp://127.0.0.1:0")); err == nil {
+		t.Fatal("listen on a tokenless tcp seam succeeded")
+	}
+}
+
+func listenLoopback(t *testing.T, token string) (*Listener, Addr) {
+	t.Helper()
+	ln, err := Listen(Addr{Network: "tcp", Address: "127.0.0.1:0", Token: token})
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	return ln, Addr{Network: "tcp", Address: ln.Addr().String(), Token: token}
+}
+
+func echoOnce(ln *Listener) {
+	c, err := ln.Accept()
+	if err != nil {
+		return
+	}
+	defer c.Close()
+	buf := make([]byte, 256)
+	n, err := c.Read(buf)
+	if err != nil {
+		return
+	}
+	_, _ = c.Write(buf[:n])
+}
+
+func roundTrip(t *testing.T, c net.Conn, msg string) string {
+	t.Helper()
+	if _, err := c.Write([]byte(msg)); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	buf := make([]byte, 256)
+	n, err := c.Read(buf)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	return string(buf[:n])
+}

internal/worker/client.go

@@ -24,6 +24,8 @@ import (
 	"net"
 	"sync"
 	"time"
+
+	"github.com/kami/maven/internal/netaddr"
 )
 
 // Client — one connection to one worker module. NOT goroutine-safe for
@@ -38,15 +40,25 @@ type Client struct {
 	dial func() (net.Conn, error)
 }
 
-// Dial opens a Client to the worker socket at path. The first call lazily
+// Dial opens a Client to the worker module at path. The first call lazily
 // dials; subsequent calls reuse the conn (a fresh dial happens on next call
 // after a teardown). Lazy dial keeps a worker that's restarting from
 // blocking core's startup; core attempts the dial on first use.
+//
+// path is a netaddr seam address. A bare path is the unix socket it has
+// always been; "tcp://workstation:9310?token=..." reaches a module on another
+// host, which is how stt and tts move to the machine with the GPU and the
+// microphone. A bad address surfaces on the first call, not here, because
+// Dial does not fail — see internal/netaddr.
 func Dial(path string) *Client {
+	addr, err := netaddr.Parse(path)
 	return &Client{
 		path: path,
 		dial: func() (net.Conn, error) {
-			return net.Dial("unix", path)
+			if err != nil {
+				return nil, err
+			}
+			return netaddr.Dial(addr)
 		},
 	}
 }

internal/worker/server.go

@@ -7,10 +7,11 @@
 // which module to dial; mixing the two is a config error caught cleanly by
 // the wire, not a runtime goroutine panic). One Server per module process.
 //
-// Socket perms mirror ipc.Server: dir 0700, socket 0600 ⇒ same unix user.
-// The module has no key, so the floor is "same user"; the wg/mTLS layers
-// are out of scope here (this socket never crosses the network radius —
-// it's local-only, point-to-point between two processes on the box).
+// The seam address decides the transport. On the default unix socket the
+// perms mirror ipc.Server — dir 0700, socket 0600 ⇒ same unix user — and that
+// is the whole auth floor, because the seam never leaves the box. A tcp
+// address moves the module to another host and takes that floor away, so
+// netaddr checks a shared token before the first frame. See internal/netaddr.
 package worker
 
 import (
@@ -18,11 +19,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"net"
-	"os"
 	"sync"
 	"sync/atomic"
 
-	"golang.org/x/sys/unix"
+	"github.com/kami/maven/internal/netaddr"
 )
 
 // Server — a worker module process's listener. Wires either a Transcriber,
@@ -34,6 +34,7 @@ type Server struct {
 	s Synthesizer
 
 	path string
+	addr netaddr.Addr
 	ln   net.Listener
 
 	wg   sync.WaitGroup
@@ -61,25 +62,24 @@ func NewSynthesizerServer(path string, s Synthesizer) *Server {
 // two separate processes per the restart-free / fail-independent invariant).
 func (srv *Server) SetSynthesizer(s Synthesizer) { srv.s = s }
 
-// Listen binds the unix socket with 0700 dir + 0600 socket perms (same floor
-// as internal/ipc). A stale socket at path is removed first so the worker
-// process restarts cleanly after a crash, no manual cleanup needed.
+// Listen binds the seam address the Server was built with.
+//
+// A bare path is a unix socket with 0700 dir + 0600 socket perms, the same
+// floor as internal/ipc, and a stale socket is removed first so the worker
+// process restarts cleanly after a crash. A "tcp://host:port?token=..."
+// address binds a network listener instead, so this module can run on the
+// workstation while core stays on homesrv; the token is mandatory there,
+// because there is no filesystem to be the auth floor. See internal/netaddr.
 func (srv *Server) Listen() error {
-	_ = os.Remove(srv.path)
-	if err := os.MkdirAll(parentDir(srv.path), 0o700); err != nil {
-		return fmt.Errorf("worker: mkdir socket dir: %w", err)
-	}
-	oldMask := unix.Umask(0o077)
-	ln, err := net.Listen("unix", srv.path)
-	unix.Umask(oldMask)
+	addr, err := netaddr.Parse(srv.path)
 	if err != nil {
-		return fmt.Errorf("worker: listen %s: %w", srv.path, err)
+		return err
 	}
-	if err := os.Chmod(srv.path, 0o600); err != nil {
-		_ = ln.Close()
-		_ = os.Remove(srv.path)
-		return fmt.Errorf("worker: chmod socket: %w", err)
+	ln, err := netaddr.Listen(addr)
+	if err != nil {
+		return err
 	}
+	srv.addr = addr
 	srv.ln = ln
 	return nil
 }
@@ -193,7 +193,7 @@ func (srv *Server) Close() error {
 	}
 	err := srv.ln.Close()
 	srv.wg.Wait()
-	_ = os.Remove(srv.path)
+	netaddr.Cleanup(srv.addr)
 	return err
 }
 
@@ -214,15 +214,3 @@ func marshalResult(v any) json.RawMessage {
 	b, _ := json.Marshal(v)
 	return b
 }
-
-func parentDir(p string) string {
-	for i := len(p) - 1; i >= 0; i-- {
-		if p[i] == '/' {
-			if i == 0 {
-				return "/"
-			}
-			return p[:i]
-		}
-	}
-	return "."
-}