Compare commits

...

7 Commits

Author SHA1 Message Date
claude dc266056d1 docs: the shape and the rules for offloading model work (V-483)
483 is an umbrella and its children are the work, so what it owes them is
the shape they must all obey. docs/offload.md records it: the degradation
rule and where its line falls, admission control rather than a GPU
arbiter, the embedder staying on homesrv because it backs the classifier,
and the inventory of what runs a model on the box today.

CLAUDE.md gets a pointer, because an agent about to add a model caller or
touch a daemon seam needs to know this before it starts, not after.
2026-08-02 15:08:35 +02:00
kami 1c786b7156 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
2026-08-02 15:08:16 +02:00
claude a3af10a830 gitignore the root .env, it holds a live token (V-484)
It was untracked but not ignored, so one git add -A would have committed
MAVEN_AMBIENT_TOKEN. Same class as deploy/telegram.env, which is already
ignored.
2026-08-02 15:08:07 +02:00
claude c0de473382 ipc, worker: dial and bind through netaddr (V-484)
Five hardcoded transports, three in internal/ipc and two in
internal/worker, all now go through the seam address. The unix perms
logic moved into netaddr, so the two copies of parentDir and the umask
dance are gone.

peerCaller already returned ok=false for a non-unix conn, so the
SO_PEERCRED path degrades correctly on tcp with no change.
2026-08-02 15:08:07 +02:00
claude 3e534340bf ipc: pin that a scheme-less address still dials unix (V-484)
Six cases. The load-bearing one is the first: every deploy in the tree
writes a bare path, and it must keep meaning a unix socket with no
handshake in front of the payload.

The rest cover the tcp seam: a good token round-trips, a wrong one comes
back ErrUnauthorized, a stranger that speaks HTTP at the port is dropped
while the listener stays up for the next peer, and a tokenless tcp bind
fails rather than serving his turns to anyone who connects.
2026-08-02 15:08:07 +02:00
claude 1a704d704d ipc: a seam address that can name a transport (V-484)
internal/netaddr parses a daemon seam address and dials or binds it. A
scheme-less address is unix and behaves exactly as it does today: same
0700 parent dir, same 0600 socket, same bytes on the wire. tcp://host:port
is the new option, and it is what lets a module live on another host.

Over tcp the filesystem permission that authenticated the unix socket is
gone, and what crosses this seam is audio of the owner speaking. So a tcp
listener requires a shared token, checked in constant time before the
first protocol frame is read, and a peer that fails is dropped without
taking the listener down with it.
2026-08-02 15:08:07 +02:00
kami e57adcb001 Merge pull request 'Docs: refresh the QA plan against the live task list' (#91) from task/459-docs-refresh-the-qa-plan-against-the-liv into master
Reviewed-on: #91
2026-08-02 15:07:21 +02:00
9 changed files with 667 additions and 89 deletions
+3
View File
@@ -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
+8
View File
@@ -28,6 +28,14 @@ model is a one-line change to `phraser.model_path` in `deploy/mavend.json`.
See `docs/rearchitecture.md` for the target architecture, `docs/design.md` for the folded design spec, and
`AGENTS.md` for local-preview + model-download recipes.
**Model work is moving to the workstation** (owner's call, 2026-08-02). homesrv cannot grow a
GPU and the workstation has 16GB of VRAM. So the resident model, STT and TTS become preferred
remotes with a floor on homesrv. The workstation is never assumed up. Fall back silently when
it would only do the job better. Name the gap when the 1.7B cannot do it at all. The embedder
stays on homesrv permanently, because it backs that floor. Read `docs/offload.md` before
touching a daemon seam or adding a model caller. Vikunja #483 is the umbrella, #484 to #487
are the work.
## Build & test
CGO daemons (`mavend`, `mavsttd`, `mavttsd`, `mavenclient`) need the vendored toolchain
+114
View File
@@ -0,0 +1,114 @@
# Offloading model work to the workstation
*Last verified: 2026-08-02 @ 5c05163. Living doc: correct it in place, do not append.*
Owner's call, 2026-08-02. Vikunja #483 is the umbrella. Tasks #484 to #487 are the
work, and this file holds the shape and the rules all four must obey.
## The goal
homesrv cannot grow a GPU. The workstation has 16GB of VRAM. Move the model work
to the workstation and leave homesrv running the logic that must be always-on,
deterministic and cheap.
## Why this is tractable
The split already exists structurally. `mavsttd` and `mavttsd` are separate
daemons that core reaches over a socket, not linked libraries. Moving them off-box
is a transport change, not a redesign.
The microphone is at the workstation, because that is where the owner sits and
homesrv is headless. So speech-to-text and the wake word are already on the
workstation side by construction. Audio never has to cross the LAN. Only the core
turn does.
## The constraint that shapes everything
The workstation's GPU is often busy: CPT runs, experiments, Correx, the manga-recap
pipeline. It also sleeps. homesrv does not.
So an offloaded model is never *the* model. It is the preferred one, with a floor
on homesrv. That is the shape the cascade already has, where a router error falls
through to the classifier.
## The degradation rule
Two cases, and the line between them is sharp.
**Fall back silently** when the workstation model would only do the job *better*:
routing, phrasing, a nudge. Falling back costs nothing that exists today, because
the resident Qwen3-1.7B is today's production quality. The owner should not be told
that his reply was phrased by the smaller model.
**Name the gap** when the resident model cannot do the job *at all*. A world
question that a 1.7B answers by inventing is the case. A wrong answer is worse
than "не могу сейчас". This is the rule CLAUDE.md already states for a sibling
service being down.
Nothing in between. A turn never breaks on the workstation being asleep.
## Admission control, not a scheduler
There is no GPU arbiter. That is a service with its own failure modes, and nothing
here needs work *distributed*. It needs admission control. The workstation
advertises free VRAM over a health endpoint, and Maven treats it as one more query
source that claims a turn or passes. llama-server also refuses to load when VRAM is
short, so the failure is detectable without cooperation from the owner's other
jobs.
The caller must be able to ask "is this peer usable right now" without a turn
hanging on a timeout. A dead remote is a normal state, not an error state.
## What stays on homesrv, permanently
The **embedder** (multilingual-e5-small, ONNX, CPU). It backs the classifier, which
must answer while the GPU is saturated. It is also cheap enough on CPU that moving
it buys nothing. Four callers:
| Caller | What for |
|---|---|
| `internal/router/classifier.go` | the routing floor |
| `cmd/mavend/actions_query.go` (`queryEmbed`) | memory recall |
| `cmd/mavend/feeds.go` | ingest embedding for every RSS item |
| `internal/crawl/watch.go` | ingest embedding for every crawled page |
`internal/speaker` becomes a fifth once it lands.
## Inventory: what runs a model on homesrv today
The **resident model** is one llama-server with seven callers:
| Caller | What for |
|---|---|
| `cmd/mavend/voicewire.go` | routing |
| `cmd/mavend/replier_llm.go` | replies |
| `cmd/mavend/tick.go` | digestion worker: `PhraseNudge`, `PhraseReminder` |
| `cmd/mavend/capture.go` | capture summarisation (unreachable, see #480) |
| `cmd/mavend/mail.go` | mail extraction (off, no IMAP) |
| `cmd/mavend/kiwixwire.go` | answering from a Kiwix, search or crawl passage |
| `memoryeval.go`, `modelswap.go` | admin and evals |
Then the embedder above, **whisper.cpp** in `mavsttd`, and **piper** in `mavttsd`.
`mavwaked` uses no model at all: an energy-threshold VAD over 30ms frames.
## Order
1. **Transport** (#484). Nothing else is possible until a seam can cross a host.
`internal/netaddr` landed in PR #92. A seam address now carries its own scheme,
and a scheme-less one is still unix. A tcp seam requires a shared token, because
the filesystem permission that authenticated the unix socket is gone.
2. **The resident model** (#485). Biggest quality delta. A 16GB card runs a 7-14B,
which fixes what the 1.7B gets wrong: world knowledge, and the persona the CPT
targets. The degradation path is already written and measured, since the
classifier scores 68.8% full accuracy at p50 16.6µs on its own.
3. **Speech-to-text and text-to-speech** (#486). They gain a real margin, but on
quality alone, and both already work.
4. **The wake word** (#487). Independent of all of the above.
## Assumptions
- The LAN is trusted enough that wireguard is supported but not required (owner's
call). What crosses the wire is still his utterances. That is why the tcp seam
carries its own token instead of assuming a network boundary.
- The workstation is not expected to be up. Every child task must still serve a
turn while it is down.
+24 -13
View File
@@ -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
}
+20 -40
View File
@@ -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
+277
View File
@@ -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)
}
}
+185
View File
@@ -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])
}
+14 -2
View File
@@ -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)
},
}
}
+22 -34
View File
@@ -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 "."
}