openvibers/openvibe.link Commits

v0.37.0: openvibe-node share, OpenVibe Connect from the Node (#97)

OP OpenVibers committed 11 days ago · 41db4140ab · 9 files, +1321 −1

* v0.37.0: openvibe-node share, OpenVibe Connect from the Node (CON-4, v0)

internal/connect speaks ovconnect-ws/1 like Connect's one-file client: one outgoing WebSocket per share to
wss://connect.openvibe.website/tunnel, HTTP streaming both ways with a per-stream inbox and pause/resume (a slow local
reader never stalls the tunnel), server-sent events, WebSocket upgrades as a byte pipe, reconnect with backoff, and a
stop on a wrong key, a closed share or a takeover. It only ever dials localhost on the share's port.

openvibe-node share --key … --port N serves one now; share add|list|remove keeps shares in connect-shares.json (0600)
and the service keeps every saved share up. Proved in production through Cloudflare (gate, GET, POST, SSE, upgrade,
close) and locally under load (40 SSE streams, 20 GETs, 10 x 3 MB at once).

* Windows keeps no Unix mode bits: the 0600 checks run elsewhere

Changes

ddiff --git a/README.md b/README.md
iindex 566d657..8bba743 100644
---- a/README.md
++++ b/README.md
@@@ -25,6 +25,10 @@ openvibe.bot shows this command with the robot's id and a fresh code. `--driver`
  OpenVibe.Network gets `--network <URL> --pairing pair_… --code ABCD-1234` instead of `--robot`. Details, manual
  install, Windows and configuration: [docs/install.md](docs/install.md).
  
++**Put a port on the web** with [OpenVibe Connect](https://connect.openvibe.website): `openvibe-node share --key ovc_… --port 8790`
++serves it at `https://<name>--<you>.openvibe.website` behind OpenVibe sign-in, and `openvibe-node share add` keeps it
++up with the service ([docs/connect.md](docs/connect.md)).
++
  ```sh
  openvibe-node status     # link, plugins, video, battery, stop latch
  openvibe-node stop       # kill switch: stop every motor now and keep it stopped
ddiff --git a/STATUS.json b/STATUS.json
iindex 870b89e..0082c85 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -5,7 +5,7 @@
    "wave": "T15",
    "plan": "Plan track T15 (Bot complete) and ADR-043 (Bot: devices, pairing, control and safety); the computer plugin and the job worker serve OpenVibe.Actor and the Codes job runner (tracks T14/T17)",
    "kind": "client",
--  "version": "0.36.0",
++  "version": "0.37.0",
    "release": "v0.36.0: the ai plugin (0.2.0) passes tools to the machine's own model server and hands back the calls the model asked for, never running one, so an agent on OpenVibe (Actor) can work on a model this machine serves; its capabilities say tools: true (contracts 0.277.0). v0.35.0: the Node applies the roles the network's machines hub gives it at each check-in (codes.worker starts the job worker it was installed with, desks.host and ai.host are checked against what the machine was installed for, an unknown role is ignored, and the local configuration always wins), and its status reports them back; a desk's first browser start in a cold sandbox gets 90 s. v0.34.0: on Windows a killed plugin takes every process it started with it (a Job Object with kill-on-close), as the process group does on Linux and macOS, so a bridge helper cannot keep a robot's link open after the Node stops or restarts its plugin; the Node reads what the coding tools remember (read-only, local, redacted) and `openvibe-node memory` uploads it to your own store; the heartbeat carries the machine's fleet of coding agents; the Codes job runner pin is v0.5.2",
    "runtime": "go 1.24 (one static binary per platform); plugins in Python 3; an ESP32 Arduino library",
    "database": "none",
ddiff --git a/cmd/openvibe-node/main.go b/cmd/openvibe-node/main.go
iindex 2d4b909..a946191 100644
---- a/cmd/openvibe-node/main.go
++++ b/cmd/openvibe-node/main.go
@@@ -33,6 +33,7 @@ import (
  	"time"
  
  	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/connect"
  	"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
  	"github.com/OpenVibers/OpenVibe.Node/internal/link"
  	"github.com/OpenVibers/OpenVibe.Node/internal/localctl"
@@@ -77,6 +78,9 @@ Usage:
    openvibe-node memory scan [--tools a,b] [--project DIR …] [--json]
                      read the memory files of the coding tools on this machine
                      (read-only, local, secrets removed)
++  openvibe-node share --key ovc_… --port N           put this machine's port N on the web (OpenVibe Connect),
++                                                   behind OpenVibe sign-in, until Ctrl-C
++  openvibe-node share add|list|remove …             the shares the Node service keeps up (openvibe-node share)
    openvibe-node version
  
  Global flags (before or after the command):
@@@ -133,6 +137,9 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
  		session = fs.Bool("user-session", false, "Windows: start at sign-in in your own session instead of as a service (install, uninstall)")
  		keyID   = fs.String("key-id", "", "accept only this Bot signing key id (profile verify)")
  		autoArg = fs.String("autostart", "", "on or off: start the tray when you sign in (tray)")
++		shKey   = fs.String("key", "", "the share's ovc_… connect key from connect.openvibe.website (share)")
++		shPort  = fs.Int("port", 0, "the local port to share (share)")
++		shSrv   = fs.String("connect-server", "", "OpenVibe Connect's tunnel URL; default wss://connect.openvibe.website/tunnel (share)")
  	)
  	// Accept flags anywhere: split positionals out.
  	var positional []string
@@@ -222,6 +229,11 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
  			return 2
  		}
  		err = cmdProfileVerify(g, positional[1], *server, *keyID, stdout)
++	case "share":
++		err = cmdShare(g, positional, shareArgs{key: *shKey, port: *shPort, server: *shSrv}, stdout, stderr)
++		if errors.Is(err, errUsage) {
++			return 2
++		}
  	case "version", "--version":
  		fmt.Fprintf(stdout, "openvibe-node %s %s/%s\n", version, runtime.GOOS, runtime.GOARCH)
  	default:
@@@ -569,6 +581,10 @@ func cmdRun(g *globals, dryRun bool) error {
  	}
  	g.log.Info("openvibe-node starting", "version", version, "config", g.paths.ConfigDir, "plugins", len(cfg.Plugins))
  	return service.Run(serviceOptions(g, ""), func(ctx context.Context) error {
++		if !dryRun {
++			// OpenVibe Connect: every saved share stays up while the Node runs (openvibe-node share add).
++			go connect.RunAll(ctx, g.paths.ConfigDir, connect.Options{Log: g.log, Version: version})
++		}
  		if err := n.Run(ctx); !errors.Is(err, node.ErrQuit) {
  			return err
  		}
ddiff --git a/cmd/openvibe-node/share.go b/cmd/openvibe-node/share.go
nnew file mode 100644
iindex 0000000..f27f247
---- /dev/null
++++ b/cmd/openvibe-node/share.go
@@@ -0,0 +1,129 @@
++package main
++
++import (
++	"context"
++	"errors"
++	"fmt"
++	"io"
++	"os"
++	"os/signal"
++	"strings"
++	"syscall"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/connect"
++)
++
++// shareArgs is `openvibe-node share …` (OpenVibe Connect, Website Connect v0).
++type shareArgs struct {
++	key    string
++	port   int
++	server string
++}
++
++const shareUsage = `usage:
++  openvibe-node share --key ovc_… --port N      serve the share now, until Ctrl-C
++  openvibe-node share add --key ovc_… --port N  keep it up from the Node service (restart the service to start it)
++  openvibe-node share list                      the shares the service keeps up
++  openvibe-node share remove <PORT|KEY|ADDRESS> stop keeping one up
++
++Make a share and its key at https://connect.openvibe.website. The Node only ever connects to localhost on the
++share's port; every visitor signs in with OpenVibe first.`
++
++func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Writer) error {
++	sub := ""
++	if len(positional) > 0 {
++		sub = positional[0]
++	}
++	if a.key == "" {
++		a.key = os.Getenv("OV_CONNECT_KEY")
++	}
++	switch sub {
++	case "", "run":
++		if a.key == "" || a.port == 0 {
++			fmt.Fprintln(stderr, shareUsage)
++			return errUsage
++		}
++		return shareNow(g, a, stdout)
++	case "add":
++		if a.key == "" || a.port == 0 {
++			fmt.Fprintln(stderr, shareUsage)
++			return errUsage
++		}
++		if !strings.HasPrefix(a.key, "ovc_") {
++			return errors.New("the key is the share's ovc_… connect key, from its page on connect.openvibe.website")
++		}
++		if err := g.paths.Ensure(); err != nil {
++			return err
++		}
++		if err := connect.Add(g.paths.ConfigDir, connect.Share{Key: a.key, Port: a.port, Server: a.server}); err != nil {
++			return err
++		}
++		fmt.Fprintf(stdout, "Saved: localhost:%d (key %s). Restart the Node service to start it: the service keeps every saved share up.\n", a.port, connect.Redact(a.key))
++		return nil
++	case "list":
++		shares, err := connect.Load(g.paths.ConfigDir)
++		if err != nil {
++			return err
++		}
++		if len(shares) == 0 {
++			fmt.Fprintln(stdout, "No saved shares. Add one with: openvibe-node share add --key ovc_… --port N")
++			return nil
++		}
++		for _, s := range shares {
++			name := s.Name
++			if name == "" {
++				name = "(address shown once it connects)"
++			}
++			fmt.Fprintf(stdout, "localhost:%d  %s  key %s\n", s.Port, name, connect.Redact(s.Key))
++		}
++		return nil
++	case "remove":
++		if len(positional) != 2 {
++			fmt.Fprintln(stderr, shareUsage)
++			return errUsage
++		}
++		n, err := connect.Remove(g.paths.ConfigDir, positional[1])
++		if err != nil {
++			return err
++		}
++		if n == 0 {
++			return fmt.Errorf("no saved share matches %q", positional[1])
++		}
++		fmt.Fprintf(stdout, "Removed %d. Restart the Node service to stop serving it.\n", n)
++		return nil
++	default:
++		fmt.Fprintln(stderr, shareUsage)
++		return errUsage
++	}
++}
++
++// shareNow serves one share in the foreground, with a line per request, until Ctrl-C or the server stops it.
++func shareNow(g *globals, a shareArgs, stdout io.Writer) error {
++	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
++	defer stop()
++	err := connect.Run(ctx, connect.Share{Key: a.key, Port: a.port, Server: a.server}, connect.Options{
++		Log:     g.log,
++		Version: version,
++		OnReady: func(r connect.Ready) {
++			fmt.Fprintf(stdout, "Connected. %s → localhost:%d (only the people you let in can open it). Ctrl-C to stop.\n", r.URL, a.port)
++		},
++		OnRequest: func(method, path string, status int, d time.Duration) {
++			if len(path) > 120 {
++				path = path[:120]
++			}
++			if status == 0 {
++				fmt.Fprintf(stdout, "%s %s → nothing answered on port %d\n", method, path, a.port)
++				return
++			}
++			fmt.Fprintf(stdout, "%s %s → %d (%d ms)\n", method, path, status, d.Milliseconds())
++		},
++	})
++	if errors.Is(err, context.Canceled) {
++		return nil
++	}
++	return err
++}
++
++// errUsage: the usage was printed; exit 2 without another line.
++var errUsage = errors.New("usage")
ddiff --git a/cmd/openvibe-node/share_test.go b/cmd/openvibe-node/share_test.go
nnew file mode 100644
iindex 0000000..6d78785
---- /dev/null
++++ b/cmd/openvibe-node/share_test.go
@@@ -0,0 +1,54 @@
++package main
++
++import (
++	"bytes"
++	"os"
++	"path/filepath"
++	"runtime"
++	"strings"
++	"testing"
++)
++
++// `openvibe-node share`: usage, add, list (the key only redacted), remove, and the file only the Node's account reads.
++func TestShareCommands(t *testing.T) {
++	home := t.TempDir()
++	key := "ovc_" + strings.Repeat("K", 43)
++	var out, errb bytes.Buffer
++
++	if code := run([]string{"share", "--home", home}, &out, &errb); code != 2 || !strings.Contains(errb.String(), "connect.openvibe.website") {
++		t.Fatalf("usage: %d %q", code, errb.String())
++	}
++	out.Reset()
++	errb.Reset()
++	if code := run([]string{"share", "add", "--key", key, "--port", "8790", "--home", home}, &out, &errb); code != 0 {
++		t.Fatalf("add: %d %s", code, errb.String())
++	}
++	if strings.Contains(out.String(), key) {
++		t.Fatal("add printed the whole key")
++	}
++	out.Reset()
++	if code := run([]string{"share", "list", "--home", home}, &out, &errb); code != 0 || !strings.Contains(out.String(), "localhost:8790") || strings.Contains(out.String(), key) {
++		t.Fatalf("list: %d %q", code, out.String())
++	}
++	matches, _ := filepath.Glob(filepath.Join(home, "**", "connect-shares.json"))
++	if len(matches) == 0 {
++		matches, _ = filepath.Glob(filepath.Join(home, "*", "connect-shares.json"))
++	}
++	if len(matches) == 0 {
++		matches, _ = filepath.Glob(filepath.Join(home, "connect-shares.json"))
++	}
++	if len(matches) != 1 {
++		t.Fatalf("the saved file: %v", matches)
++	}
++	if fi, _ := os.Stat(matches[0]); runtime.GOOS != "windows" && fi.Mode().Perm() != 0o600 { // Windows keeps no Unix mode bits
++		t.Fatalf("mode %v", fi.Mode().Perm())
++	}
++	out.Reset()
++	if code := run([]string{"share", "remove", "8790", "--home", home}, &out, &errb); code != 0 || !strings.Contains(out.String(), "Removed 1") {
++		t.Fatalf("remove: %d %q", code, out.String())
++	}
++	errb.Reset()
++	if code := run([]string{"share", "add", "--key", "nope", "--port", "1", "--home", home}, &out, &errb); code != 1 || !strings.Contains(errb.String(), "ovc_") {
++		t.Fatalf("a bad key: %d %q", code, errb.String())
++	}
++}
ddiff --git a/docs/connect.md b/docs/connect.md
nnew file mode 100644
iindex 0000000..2366490
---- /dev/null
++++ b/docs/connect.md
@@@ -0,0 +1,50 @@
++# OpenVibe Connect from the Node
++
++[OpenVibe Connect](https://connect.openvibe.website) puts one port on your computer on the web at its own address,
++`https://<name>--<you>.openvibe.website`, and only the people you pick can open it, after they sign in with OpenVibe.
++The Node speaks Connect's protocol itself (`internal/connect`, `ovconnect-ws/1`), so a share can stay up as part of
++the Node service, across reboots, without the one-file client.
++
++## Make a share
++
++Sign in at connect.openvibe.website, give the share a name and the port, and pick who can open it. The page shows its
++connect key once (`ovc_…`). Keep it like a password: it lets a computer serve that share.
++
++## Serve it now
++
++```sh
++openvibe-node share --key ovc_… --port 8790
++```
++
++It prints the address, then one line per request, until Ctrl-C. The key can also come from `OV_CONNECT_KEY`, which
++keeps it off the command line and out of your shell history:
++
++```sh
++OV_CONNECT_KEY=ovc_… openvibe-node share --port 8790
++```
++
++## Keep it up with the Node service
++
++```sh
++openvibe-node share add --key ovc_… --port 8790
++openvibe-node share list
++openvibe-node share remove 8790        # or the key, or the address
++```
++
++The saved shares are in the Node's config directory, `connect-shares.json`, readable only by the Node's account
++(0600). The service starts every saved share when it starts, so restart it after `add` or `remove`. A share the server
++ends for good (its key was replaced, it was closed, or another computer took it over) stays stopped and says so in the
++log; `share list` shows each share's address once it has connected.
++
++## What the Node does and does not do
++
++- It opens one outgoing WebSocket per share to `wss://connect.openvibe.website/tunnel` (port 443, through Cloudflare),
++  so it works behind any router, firewall or phone hotspot, with no open ports.
++- It only ever connects to `localhost` on the share's own port. The server cannot make it reach anything else.
++- Requests stream both ways with backpressure; server-sent events stream; WebSocket upgrades become a byte pipe.
++- It reconnects after a drop with backoff from 1 to 30 seconds, and the server pings every 25 seconds.
++- Your app sees `Host: localhost:<port>`, the share's address in `X-Forwarded-Host`, and who the visitor is in
++  `X-OpenVibe-User` and `X-OpenVibe-Subject`. It never sees the gate's own cookie.
++
++Who may open a share, public shares and the limits are Connect's: see connect.openvibe.website and
++[OpenVibe.Connect's README](https://github.com/OpenVibers/OpenVibe.Connect).
ddiff --git a/internal/connect/connect.go b/internal/connect/connect.go
nnew file mode 100644
iindex 0000000..7613cd2
---- /dev/null
++++ b/internal/connect/connect.go
@@@ -0,0 +1,554 @@
++// Package connect keeps OpenVibe Connect shares up from the Node: each share is one port on this machine served at
++// https://<label>--<user>.openvibe.website behind OpenVibe sign-in (OpenVibe.Connect, ADR-057, Website Connect v0).
++//
++// The Node opens one outgoing WebSocket per share to wss://connect.openvibe.website/tunnel and speaks ovconnect-ws/1,
++// the same protocol as the one-file client (OpenVibe.Connect public/ov-connect.js):
++//
++//	→ {"t":"hello","v":1,"key":"ovc_…","client":"openvibe-node/…","port":8790}
++//	← {"t":"ready","v":1,"share":{"id","host","url","mode"}}            or {"t":"error",…} and a close
++//	← {"t":"req","id":N,"method","path","headers":[[k,v]…]}             an HTTP request for the local port
++//	← {"t":"raw","id":N,"head":"GET /ws HTTP/1.1\r\n…"}                  a WebSocket upgrade: a byte pipe
++//	→ {"t":"res","id":N,"status":200,"headers":[[k,v]…]}
++//	↔ binary [kind 1 byte][id 4 bytes BE][payload]: kind 1 data, 2 end
++//	↔ {"t":"pause"|"resume"|"abort","id":N}
++//
++// It only ever connects to localhost on the share's own port: the server cannot make the Node reach anything else.
++// A wrong key (close 4002), a closed share (4003) or another client taking over (4001) stops that share; anything
++// else reconnects with backoff from 1 s to 30 s.
++package connect
++
++import (
++	"bufio"
++	"bytes"
++	"context"
++	"encoding/binary"
++	"encoding/json"
++	"errors"
++	"fmt"
++	"io"
++	"log/slog"
++	"net"
++	"net/http"
++	"strconv"
++	"strings"
++	"sync"
++	"time"
++
++	"github.com/gorilla/websocket"
++)
++
++// DefaultServer is OpenVibe Connect's tunnel endpoint.
++const DefaultServer = "wss://connect.openvibe.website/tunnel"
++
++const (
++	protocolVersion = 1
++	kindData        = 1
++	kindEnd         = 2
++	chunkSize       = 32 * 1024
++)
++
++// Share is one share the Node keeps up: its connect key and the local port it serves.
++type Share struct {
++	Key    string `json:"key"`
++	Port   int    `json:"port"`
++	Server string `json:"server,omitempty"` // default DefaultServer
++	Name   string `json:"name,omitempty"`   // the share's address, learnt from "ready" (shown by `share list`)
++}
++
++// Ready is what the server says once a share is connected.
++type Ready struct {
++	ID   string `json:"id"`
++	Host string `json:"host"`
++	URL  string `json:"url"`
++	Mode string `json:"mode"`
++}
++
++// ErrStopped is returned when the server ends a share for good: a wrong key, a closed share, or another client.
++var ErrStopped = errors.New("connect: share stopped by the server")
++
++// Options tune a client; the zero value is right for production.
++type Options struct {
++	Log       *slog.Logger
++	Version   string
++	LocalHost string      // default "localhost"
++	OnReady   func(Ready) // called after each successful hello
++	OnRequest func(method, path string, status int, d time.Duration)
++	Dialer    *websocket.Dialer
++	MinWait   time.Duration // reconnect backoff floor (default 1 s)
++	MaxWait   time.Duration // reconnect backoff ceiling (default 30 s)
++}
++
++// Run keeps one share connected until ctx ends or the server stops it (ErrStopped).
++func Run(ctx context.Context, s Share, o Options) error {
++	if s.Port < 1 || s.Port > 65535 {
++		return fmt.Errorf("connect: port %d is not a port", s.Port)
++	}
++	if !strings.HasPrefix(s.Key, "ovc_") {
++		return errors.New("connect: the key is the share's ovc_… connect key")
++	}
++	if o.Log == nil {
++		o.Log = slog.New(slog.NewTextHandler(io.Discard, nil))
++	}
++	if o.MinWait == 0 {
++		o.MinWait = time.Second
++	}
++	if o.MaxWait == 0 {
++		o.MaxWait = 30 * time.Second
++	}
++	wait := o.MinWait
++	for {
++		err := runOnce(ctx, s, o, func() { wait = o.MinWait })
++		if ctx.Err() != nil {
++			return ctx.Err()
++		}
++		if errors.Is(err, ErrStopped) {
++			return err
++		}
++		o.Log.Info("connect: disconnected; trying again", "port", s.Port, "in", wait.String(), "err", errString(err))
++		select {
++		case <-ctx.Done():
++			return ctx.Err()
++		case <-time.After(wait):
++		}
++		wait *= 2
++		if wait > o.MaxWait {
++			wait = o.MaxWait
++		}
++	}
++}
++
++func errString(err error) string {
++	if err == nil {
++		return ""
++	}
++	return err.Error()
++}
++
++type tunnel struct {
++	ws      *websocket.Conn
++	wmu     sync.Mutex
++	share   Share
++	o       Options
++	mu      sync.Mutex
++	streams map[uint32]*stream
++}
++
++type stream struct {
++	body   *io.PipeWriter // http: the request body's writer
++	conn   net.Conn       // raw: the local socket
++	paused chan struct{}  // non-nil while the server asked us to pause sending
++	pmu    sync.Mutex
++
++	// The inbox: data frames from the server wait here so a slow local reader never stalls the tunnel's read loop.
++	// Over inboxHigh bytes the server is asked to pause this stream; under inboxLow it resumes.
++	qmu      sync.Mutex
++	q        [][]byte
++	qbytes   int
++	ended    bool
++	held     bool
++	dead     bool
++	wake     chan struct{} // never closed: drop sets dead and wakes the drainer
++	writerOn bool
++}
++
++const (
++	inboxHigh = 1 << 20
++	inboxLow  = 256 << 10
++)
++
++func (st *stream) waitResume(ctx context.Context) {
++	st.pmu.Lock()
++	ch := st.paused
++	st.pmu.Unlock()
++	if ch == nil {
++		return
++	}
++	select {
++	case <-ch:
++	case <-ctx.Done():
++	}
++}
++
++func (st *stream) setPaused(p bool) {
++	st.pmu.Lock()
++	defer st.pmu.Unlock()
++	if p && st.paused == nil {
++		st.paused = make(chan struct{})
++	} else if !p && st.paused != nil {
++		close(st.paused)
++		st.paused = nil
++	}
++}
++
++func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
++	server := s.Server
++	if server == "" {
++		server = DefaultServer
++	}
++	dialer := o.Dialer
++	if dialer == nil {
++		dialer = &websocket.Dialer{HandshakeTimeout: 15 * time.Second, Proxy: http.ProxyFromEnvironment}
++	}
++	ws, _, err := dialer.DialContext(ctx, server, http.Header{"User-Agent": {"openvibe-node/" + o.Version}})
++	if err != nil {
++		return err
++	}
++	ctx, cancel := context.WithCancel(ctx)
++	defer cancel()
++	t := &tunnel{ws: ws, share: s, o: o, streams: map[uint32]*stream{}}
++	defer t.closeAll()
++	go func() { <-ctx.Done(); _ = ws.Close() }()
++	// The server pings every 25 s; answering is gorilla's default ping handler. A dead link shows as a read error.
++	if err := t.sendJSON(map[string]any{"t": "hello", "v": protocolVersion, "key": s.Key, "client": "openvibe-node/" + o.Version, "port": s.Port}); err != nil {
++		return err
++	}
++	for {
++		mt, data, err := ws.ReadMessage()
++		if err != nil {
++			var ce *websocket.CloseError
++			if errors.As(err, &ce) && (ce.Code == 4001 || ce.Code == 4002 || ce.Code == 4003) {
++				o.Log.Warn("connect: the server stopped this share", "port", s.Port, "code", ce.Code, "reason", ce.Text)
++				return fmt.Errorf("%w: %s", ErrStopped, ce.Text)
++			}
++			return err
++		}
++		if mt == websocket.BinaryMessage {
++			t.onBinary(ctx, data)
++			continue
++		}
++		var m struct {
++			T       string     `json:"t"`
++			ID      uint32     `json:"id"`
++			Method  string     `json:"method"`
++			Path    string     `json:"path"`
++			Headers [][]string `json:"headers"`
++			Head    string     `json:"head"`
++			Share   Ready      `json:"share"`
++			Message string     `json:"message"`
++		}
++		if json.Unmarshal(data, &m) != nil {
++			continue
++		}
++		switch m.T {
++		case "ready":
++			onReady()
++			o.Log.Info("connect: connected", "url", m.Share.URL, "port", s.Port)
++			if o.OnReady != nil {
++				o.OnReady(m.Share)
++			}
++		case "error":
++			o.Log.Warn("connect: the server refused", "port", s.Port, "message", m.Message)
++		case "req":
++			t.onRequest(ctx, m.ID, m.Method, m.Path, m.Headers)
++		case "raw":
++			t.onRaw(ctx, m.ID, m.Head)
++		case "pause", "resume":
++			if st := t.get(m.ID); st != nil {
++				st.setPaused(m.T == "pause")
++			}
++		case "abort":
++			t.drop(m.ID)
++		}
++	}
++}
++
++func (t *tunnel) get(id uint32) *stream {
++	t.mu.Lock()
++	defer t.mu.Unlock()
++	return t.streams[id]
++}
++
++func (t *tunnel) put(id uint32, st *stream) {
++	t.mu.Lock()
++	t.streams[id] = st
++	t.mu.Unlock()
++}
++
++func (t *tunnel) drop(id uint32) {
++	t.mu.Lock()
++	st := t.streams[id]
++	delete(t.streams, id)
++	t.mu.Unlock()
++	if st == nil {
++		return
++	}
++	if st.body != nil {
++		_ = st.body.CloseWithError(errors.New("aborted"))
++	}
++	if st.conn != nil {
++		_ = st.conn.Close()
++	}
++	st.setPaused(false)
++	st.qmu.Lock()
++	st.dead = true
++	st.q, st.qbytes = nil, 0
++	wake := st.wake
++	st.qmu.Unlock()
++	if wake != nil {
++		select {
++		case wake <- struct{}{}:
++		default:
++		}
++	}
++}
++
++func (t *tunnel) closeAll() {
++	t.mu.Lock()
++	ids := make([]uint32, 0, len(t.streams))
++	for id := range t.streams {
++		ids = append(ids, id)
++	}
++	t.mu.Unlock()
++	for _, id := range ids {
++		t.drop(id)
++	}
++}
++
++func (t *tunnel) sendJSON(v any) error {
++	b, err := json.Marshal(v)
++	if err != nil {
++		return err
++	}
++	t.wmu.Lock()
++	defer t.wmu.Unlock()
++	_ = t.ws.SetWriteDeadline(time.Now().Add(30 * time.Second))
++	return t.ws.WriteMessage(websocket.TextMessage, b)
++}
++
++func (t *tunnel) sendFrame(kind byte, id uint32, payload []byte) error {
++	buf := make([]byte, 5+len(payload))
++	buf[0] = kind
++	binary.BigEndian.PutUint32(buf[1:5], id)
++	copy(buf[5:], payload)
++	t.wmu.Lock()
++	defer t.wmu.Unlock()
++	_ = t.ws.SetWriteDeadline(time.Now().Add(30 * time.Second))
++	return t.ws.WriteMessage(websocket.BinaryMessage, buf)
++}
++
++func (t *tunnel) abort(id uint32, code, message string) {
++	_ = t.sendJSON(map[string]any{"t": "abort", "id": id, "code": code, "message": message})
++	t.drop(id)
++}
++
++func (t *tunnel) onBinary(ctx context.Context, data []byte) {
++	if len(data) < 5 {
++		return
++	}
++	kind, id, payload := data[0], binary.BigEndian.Uint32(data[1:5]), data[5:]
++	st := t.get(id)
++	if st == nil {
++		return
++	}
++	st.qmu.Lock()
++	if kind == kindData {
++		st.q = append(st.q, append([]byte(nil), payload...))
++		st.qbytes += len(payload)
++	} else if kind == kindEnd {
++		st.ended = true
++	}
++	pause := st.qbytes > inboxHigh && !st.held
++	if pause {
++		st.held = true
++	}
++	if !st.writerOn {
++		st.writerOn = true
++		st.wake = make(chan struct{}, 1)
++		go t.drain(id, st)
++	}
++	wake := st.wake
++	st.qmu.Unlock()
++	if pause {
++		_ = t.sendJSON(map[string]any{"t": "pause", "id": id})
++	}
++	select {
++	case wake <- struct{}{}:
++	default:
++	}
++}
++
++// drain writes a stream's inbox to the local side, in order, off the read loop.
++func (t *tunnel) drain(id uint32, st *stream) {
++	for {
++		<-st.wake
++		for {
++			st.qmu.Lock()
++			if st.dead {
++				st.qmu.Unlock()
++				return
++			}
++			if len(st.q) == 0 {
++				ended := st.ended
++				st.qmu.Unlock()
++				if ended {
++					if st.body != nil {
++						_ = st.body.Close()
++					} else if c, ok := st.conn.(interface{ CloseWrite() error }); ok {
++						_ = c.CloseWrite()
++					}
++					return
++				}
++				break
++			}
++			chunk := st.q[0]
++			st.q = st.q[1:]
++			st.qbytes -= len(chunk)
++			resume := st.held && st.qbytes < inboxLow
++			if resume {
++				st.held = false
++			}
++			st.qmu.Unlock()
++			var err error
++			if st.body != nil {
++				_, err = st.body.Write(chunk)
++			} else if st.conn != nil {
++				_, err = st.conn.Write(chunk)
++			}
++			if err != nil {
++				t.drop(id)
++				return
++			}
++			if resume {
++				_ = t.sendJSON(map[string]any{"t": "resume", "id": id})
++			}
++		}
++	}
++}
++
++func (t *tunnel) localAddr() string {
++	h := t.o.LocalHost
++	if h == "" {
++		h = "localhost"
++	}
++	return net.JoinHostPort(h, strconv.Itoa(t.share.Port))
++}
++
++// onRequest carries one HTTP request to the local port. The request body arrives as data frames on a pipe, so a large
++// upload streams; the response streams back in chunks with the server's pause/resume.
++func (t *tunnel) onRequest(ctx context.Context, id uint32, method, path string, headers [][]string) {
++	pr, pw := io.Pipe()
++	st := &stream{body: pw}
++	t.put(id, st)
++	go func() {
++		started := time.Now()
++		req, err := http.NewRequestWithContext(ctx, method, "http://"+t.localAddr()+path, pr)
++		if err != nil {
++			t.abort(id, "local_error", "bad request")
++			return
++		}
++		for _, h := range headers {
++			if len(h) != 2 {
++				continue
++			}
++			if strings.EqualFold(h[0], "host") {
++				req.Host = h[1]
++				continue
++			}
++			if strings.EqualFold(h[0], "content-length") {
++				if n, err := strconv.ParseInt(h[1], 10, 64); err == nil {
++					req.ContentLength = n
++				}
++				continue
++			}
++			req.Header.Add(h[0], h[1])
++		}
++		if method == http.MethodGet || method == http.MethodHead {
++			// A body-less request still gets its "end" frame; nothing waits on the pipe.
++			if req.ContentLength <= 0 {
++				req.Body = http.NoBody
++				req.ContentLength = 0
++				go func() { _, _ = io.Copy(io.Discard, pr) }()
++			}
++		}
++		resp, err := localClient.Do(req)
++		if err != nil {
++			code := "local_error"
++			var oe *net.OpError
++			if errors.As(err, &oe) && oe.Op == "dial" {
++				code = "local_refused"
++			}
++			if t.o.OnRequest != nil {
++				t.o.OnRequest(method, path, 0, time.Since(started))
++			}
++			t.abort(id, code, err.Error())
++			return
++		}
++		defer resp.Body.Close()
++		out := make([][]string, 0, len(resp.Header)+1)
++		for k, vs := range resp.Header {
++			for _, v := range vs {
++				out = append(out, []string{strings.ToLower(k), v})
++			}
++		}
++		if resp.ContentLength >= 0 && resp.Header.Get("Content-Length") == "" && method != http.MethodHead {
++			out = append(out, []string{"content-length", strconv.FormatInt(resp.ContentLength, 10)})
++		}
++		if err := t.sendJSON(map[string]any{"t": "res", "id": id, "status": resp.StatusCode, "headers": out}); err != nil {
++			return
++		}
++		if t.o.OnRequest != nil {
++			t.o.OnRequest(method, path, resp.StatusCode, time.Since(started))
++		}
++		buf := make([]byte, chunkSize)
++		for {
++			st.waitResume(ctx)
++			n, rerr := resp.Body.Read(buf)
++			if n > 0 {
++				if err := t.sendFrame(kindData, id, buf[:n]); err != nil {
++					return
++				}
++			}
++			if rerr == io.EOF {
++				_ = t.sendFrame(kindEnd, id, nil)
++				t.mu.Lock()
++				delete(t.streams, id)
++				t.mu.Unlock()
++				return
++			}
++			if rerr != nil {
++				t.abort(id, "local_error", "the response broke off")
++				return
++			}
++		}
++	}()
++}
++
++// onRaw opens a plain TCP connection to the local port for an upgrade, writes the request head the server built, and
++// then only moves bytes. The local app answers the 101 itself.
++func (t *tunnel) onRaw(ctx context.Context, id uint32, head string) {
++	conn, err := (&net.Dialer{Timeout: 10 * time.Second}).DialContext(ctx, "tcp", t.localAddr())
++	if err != nil {
++		t.abort(id, "local_refused", err.Error())
++		return
++	}
++	st := &stream{conn: conn}
++	t.put(id, st)
++	if _, err := io.Copy(conn, bufio.NewReader(bytes.NewReader([]byte(head)))); err != nil {
++		t.abort(id, "local_error", err.Error())
++		return
++	}
++	go func() {
++		buf := make([]byte, chunkSize)
++		for {
++			st.waitResume(ctx)
++			n, rerr := conn.Read(buf)
++			if n > 0 {
++				if err := t.sendFrame(kindData, id, buf[:n]); err != nil {
++					return
++				}
++			}
++			if rerr != nil {
++				_ = t.sendFrame(kindEnd, id, nil)
++				t.drop(id)
++				return
++			}
++		}
++	}()
++}
++
++// localClient talks to the local app: no proxy (it is localhost), no redirects followed (the visitor's browser
++// follows them), no automatic decompression (the bytes pass as the app sent them).
++var localClient = &http.Client{
++	Transport:     &http.Transport{Proxy: nil, DisableCompression: true, MaxIdleConnsPerHost: 16, IdleConnTimeout: 60 * time.Second, ResponseHeaderTimeout: 2 * time.Minute},
++	CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
++}
ddiff --git a/internal/connect/connect_test.go b/internal/connect/connect_test.go
nnew file mode 100644
iindex 0000000..6a4b322
---- /dev/null
++++ b/internal/connect/connect_test.go
@@@ -0,0 +1,376 @@
++package connect
++
++import (
++	"bufio"
++	"context"
++	"encoding/binary"
++	"encoding/json"
++	"errors"
++	"io"
++	"net"
++	"net/http"
++	"net/http/httptest"
++	"os"
++	"path/filepath"
++	"runtime"
++	"strconv"
++	"strings"
++	"sync"
++	"testing"
++	"time"
++
++	"github.com/gorilla/websocket"
++)
++
++// fakeServer plays OpenVibe Connect's side of ovconnect-ws/1 for one test: it reads the hello, answers ready (or
++// refuses), then lets the test send requests and read what the Node answers.
++type fakeServer struct {
++	t      *testing.T
++	srv    *httptest.Server
++	conns  chan *serverConn
++	refuse int // a close code to answer the hello with (0: ready)
++	hellos chan map[string]any
++}
++
++type serverConn struct {
++	ws     *websocket.Conn
++	wmu    sync.Mutex
++	mu     sync.Mutex
++	frames map[uint32]chan frame
++}
++
++type frame struct {
++	json map[string]any
++	data []byte
++	end  bool
++}
++
++func newFakeServer(t *testing.T, refuse int) *fakeServer {
++	f := &fakeServer{t: t, conns: make(chan *serverConn, 4), refuse: refuse, hellos: make(chan map[string]any, 4)}
++	up := websocket.Upgrader{}
++	f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++		ws, err := up.Upgrade(w, r, nil)
++		if err != nil {
++			return
++		}
++		_, data, err := ws.ReadMessage()
++		if err != nil {
++			return
++		}
++		var hello map[string]any
++		_ = json.Unmarshal(data, &hello)
++		f.hellos <- hello
++		if f.refuse != 0 {
++			_ = ws.WriteMessage(websocket.TextMessage, []byte(`{"t":"error","code":"key_invalid","message":"That connect key is not one we know."}`))
++			_ = ws.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(f.refuse, "That connect key is not one we know."))
++			time.Sleep(50 * time.Millisecond)
++			_ = ws.Close()
++			return
++		}
++		_ = ws.WriteMessage(websocket.TextMessage, []byte(`{"t":"ready","v":1,"share":{"id":"shr_1","host":"agents--alex.openvibe.website","url":"https://agents--alex.openvibe.website","mode":"owner"}}`))
++		c := &serverConn{ws: ws, frames: map[uint32]chan frame{}}
++		f.conns <- c
++		for {
++			mt, data, err := ws.ReadMessage()
++			if err != nil {
++				return
++			}
++			if mt == websocket.BinaryMessage {
++				id := binary.BigEndian.Uint32(data[1:5])
++				c.ch(id) <- frame{data: append([]byte(nil), data[5:]...), end: data[0] == kindEnd}
++				continue
++			}
++			var m map[string]any
++			_ = json.Unmarshal(data, &m)
++			id, _ := m["id"].(float64)
++			c.ch(uint32(id)) <- frame{json: m}
++		}
++	}))
++	t.Cleanup(f.srv.Close)
++	return f
++}
++
++func (f *fakeServer) url() string { return "ws" + strings.TrimPrefix(f.srv.URL, "http") + "/tunnel" }
++
++func (c *serverConn) ch(id uint32) chan frame {
++	c.mu.Lock()
++	defer c.mu.Unlock()
++	if c.frames[id] == nil {
++		c.frames[id] = make(chan frame, 256)
++	}
++	return c.frames[id]
++}
++
++func (c *serverConn) sendJSON(v any) {
++	b, _ := json.Marshal(v)
++	c.wmu.Lock()
++	defer c.wmu.Unlock()
++	_ = c.ws.WriteMessage(websocket.TextMessage, b)
++}
++
++func (c *serverConn) sendFrame(kind byte, id uint32, p []byte) {
++	buf := make([]byte, 5+len(p))
++	buf[0] = kind
++	binary.BigEndian.PutUint32(buf[1:5], id)
++	copy(buf[5:], p)
++	c.wmu.Lock()
++	defer c.wmu.Unlock()
++	_ = c.ws.WriteMessage(websocket.BinaryMessage, buf)
++}
++
++// collect reads one stream's answer: the head, then data until end (or an abort).
++func (c *serverConn) collect(t *testing.T, id uint32) (status int, headers map[string]string, body []byte, abort string) {
++	t.Helper()
++	headers = map[string]string{}
++	deadline := time.After(5 * time.Second)
++	for {
++		select {
++		case f := <-c.ch(id):
++			if f.json != nil {
++				switch f.json["t"] {
++				case "res":
++					status = int(f.json["status"].(float64))
++					for _, h := range f.json["headers"].([]any) {
++						p := h.([]any)
++						headers[p[0].(string)] = p[1].(string)
++					}
++				case "abort":
++					return status, headers, body, f.json["code"].(string)
++				}
++				continue
++			}
++			if f.end {
++				return status, headers, body, ""
++			}
++			body = append(body, f.data...)
++		case <-deadline:
++			t.Fatalf("stream %d: no answer", id)
++		}
++	}
++}
++
++func localApp(t *testing.T) (port int, seen chan *http.Request) {
++	seen = make(chan *http.Request, 16)
++	mux := http.NewServeMux()
++	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
++		b, _ := io.ReadAll(r.Body)
++		r.Body = io.NopCloser(strings.NewReader(string(b)))
++		seen <- r
++		w.Header().Set("X-App", "yes")
++		w.Header().Set("Content-Type", "application/json")
++		_ = json.NewEncoder(w).Encode(map[string]string{"method": r.Method, "path": r.URL.RequestURI(), "host": r.Host, "user": r.Header.Get("X-OpenVibe-User"), "body": string(b)})
++	})
++	mux.HandleFunc("/big", func(w http.ResponseWriter, r *http.Request) {
++		_, _ = w.Write([]byte(strings.Repeat("z", 3<<20)))
++	})
++	ln, err := net.Listen("tcp", "127.0.0.1:0")
++	if err != nil {
++		t.Fatal(err)
++	}
++	srv := &http.Server{Handler: mux}
++	go func() { _ = srv.Serve(ln) }()
++	t.Cleanup(func() { _ = srv.Close() })
++	// A raw upgrade: answer 101, then echo with a prefix.
++	return ln.Addr().(*net.TCPAddr).Port, seen
++}
++
++func echoUpgradeApp(t *testing.T) int {
++	ln, err := net.Listen("tcp", "127.0.0.1:0")
++	if err != nil {
++		t.Fatal(err)
++	}
++	t.Cleanup(func() { _ = ln.Close() })
++	go func() {
++		for {
++			c, err := ln.Accept()
++			if err != nil {
++				return
++			}
++			go func(c net.Conn) {
++				defer c.Close()
++				br := bufio.NewReader(c)
++				req, err := http.ReadRequest(br)
++				if err != nil || req.Header.Get("Upgrade") == "" {
++					return
++				}
++				_, _ = c.Write([]byte("HTTP/1.1 101 Switching Protocols\r\nUpgrade: echo\r\nConnection: Upgrade\r\n\r\n"))
++				buf := make([]byte, 1024)
++				for {
++					n, err := br.Read(buf)
++					if n > 0 {
++						_, _ = c.Write(append([]byte("echo:"), buf[:n]...))
++					}
++					if err != nil {
++						return
++					}
++				}
++			}(c)
++		}
++	}()
++	return ln.Addr().(*net.TCPAddr).Port
++}
++
++func start(t *testing.T, f *fakeServer, port int) (*serverConn, chan Ready, context.CancelFunc, chan error) {
++	ctx, cancel := context.WithCancel(context.Background())
++	ready := make(chan Ready, 4)
++	done := make(chan error, 1)
++	go func() {
++		done <- Run(ctx, Share{Key: "ovc_" + strings.Repeat("A", 43), Port: port, Server: f.url()}, Options{Version: "test", LocalHost: "127.0.0.1", OnReady: func(r Ready) { ready <- r }, MinWait: 20 * time.Millisecond})
++	}()
++	t.Cleanup(cancel)
++	select {
++	case c := <-f.conns:
++		return c, ready, cancel, done
++	case err := <-done:
++		t.Fatalf("Run ended: %v", err)
++	case <-time.After(5 * time.Second):
++		t.Fatal("no connection")
++	}
++	return nil, nil, nil, nil
++}
++
++func TestHelloAndHTTP(t *testing.T) {
++	f := newFakeServer(t, 0)
++	port, seen := localApp(t)
++	c, ready, _, _ := start(t, f, port)
++	hello := <-f.hellos
++	if hello["t"] != "hello" || hello["v"].(float64) != 1 || hello["port"].(float64) != float64(port) || !strings.HasPrefix(hello["key"].(string), "ovc_") {
++		t.Fatalf("hello %v", hello)
++	}
++	if r := <-ready; r.Host != "agents--alex.openvibe.website" {
++		t.Fatalf("ready %v", r)
++	}
++
++	// A GET with the headers the server sends: Host is the server's choice, the visitor's identity passes.
++	c.sendJSON(map[string]any{"t": "req", "id": 1, "method": "GET", "path": "/hello?x=1", "headers": [][]string{{"host", "localhost:" + strconv.Itoa(port)}, {"x-openvibe-user", "alex"}}})
++	c.sendFrame(kindEnd, 1, nil)
++	status, headers, body, abort := c.collect(t, 1)
++	if abort != "" || status != 200 || headers["x-app"] != "yes" {
++		t.Fatalf("GET: %d %v %q", status, headers, abort)
++	}
++	var got map[string]string
++	_ = json.Unmarshal(body, &got)
++	if got["path"] != "/hello?x=1" || got["user"] != "alex" || got["host"] != "localhost:"+strconv.Itoa(port) {
++		t.Fatalf("the app saw %v", got)
++	}
++	<-seen
++
++	// A POST whose body arrives in pieces.
++	c.sendJSON(map[string]any{"t": "req", "id": 2, "method": "POST", "path": "/post", "headers": [][]string{{"content-type", "text/plain"}}})
++	c.sendFrame(kindData, 2, []byte("hello "))
++	c.sendFrame(kindData, 2, []byte("from the phone"))
++	c.sendFrame(kindEnd, 2, nil)
++	_, _, body, _ = c.collect(t, 2)
++	_ = json.Unmarshal(body, &got)
++	if got["body"] != "hello from the phone" || got["method"] != "POST" {
++		t.Fatalf("POST: %v", got)
++	}
++
++	// A 3 MB answer arrives whole, in chunks.
++	c.sendJSON(map[string]any{"t": "req", "id": 3, "method": "GET", "path": "/big", "headers": [][]string{}})
++	c.sendFrame(kindEnd, 3, nil)
++	_, _, body, _ = c.collect(t, 3)
++	if len(body) != 3<<20 {
++		t.Fatalf("big: %d bytes", len(body))
++	}
++}
++
++func TestNothingListeningIsLocalRefused(t *testing.T) {
++	f := newFakeServer(t, 0)
++	ln, _ := net.Listen("tcp", "127.0.0.1:0")
++	port := ln.Addr().(*net.TCPAddr).Port
++	_ = ln.Close()
++	c, _, _, _ := start(t, f, port)
++	c.sendJSON(map[string]any{"t": "req", "id": 1, "method": "GET", "path": "/", "headers": [][]string{}})
++	c.sendFrame(kindEnd, 1, nil)
++	if _, _, _, abort := c.collect(t, 1); abort != "local_refused" {
++		t.Fatalf("abort %q", abort)
++	}
++}
++
++func TestUpgradeIsABytePipe(t *testing.T) {
++	f := newFakeServer(t, 0)
++	port := echoUpgradeApp(t)
++	c, _, _, _ := start(t, f, port)
++	c.sendJSON(map[string]any{"t": "raw", "id": 7, "head": "GET /ws HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: echo\r\n\r\n"})
++	got := ""
++	deadline := time.After(5 * time.Second)
++	sent := false
++	for !strings.Contains(got, "echo:ping") {
++		select {
++		case fr := <-c.ch(7):
++			got += string(fr.data)
++			if strings.Contains(got, "\r\n\r\n") && !sent {
++				sent = true
++				c.sendFrame(kindData, 7, []byte("ping"))
++			}
++		case <-deadline:
++			t.Fatalf("no echo: %q", got)
++		}
++	}
++	if !strings.HasPrefix(got, "HTTP/1.1 101") {
++		t.Fatalf("got %q", got)
++	}
++}
++
++func TestAWrongKeyStops(t *testing.T) {
++	f := newFakeServer(t, 4002)
++	done := make(chan error, 1)
++	go func() {
++		done <- Run(context.Background(), Share{Key: "ovc_" + strings.Repeat("B", 43), Port: 80, Server: f.url()}, Options{MinWait: 10 * time.Millisecond})
++	}()
++	select {
++	case err := <-done:
++		if !errors.Is(err, ErrStopped) {
++			t.Fatalf("err %v", err)
++		}
++	case <-time.After(5 * time.Second):
++		t.Fatal("did not stop")
++	}
++}
++
++func TestReconnectsAfterADrop(t *testing.T) {
++	f := newFakeServer(t, 0)
++	port, _ := localApp(t)
++	c, ready, _, _ := start(t, f, port)
++	<-ready
++	_ = c.ws.Close() // the link drops (Wi-Fi, sleep, a restart on the server)
++	select {
++	case <-f.conns:
++	case <-time.After(5 * time.Second):
++		t.Fatal("no reconnect")
++	}
++	select {
++	case <-ready:
++	case <-time.After(5 * time.Second):
++		t.Fatal("not ready again")
++	}
++}
++
++func TestSavedSharesAreTheOwnersOnly(t *testing.T) {
++	dir := t.TempDir()
++	key := "ovc_" + strings.Repeat("C", 43)
++	if err := Add(dir, Share{Key: key, Port: 8790}); err != nil {
++		t.Fatal(err)
++	}
++	if err := Add(dir, Share{Key: key, Port: 8791}); err != nil {
++		t.Fatal(err)
++	}
++	list, _ := Load(dir)
++	if len(list) != 1 || list[0].Port != 8791 {
++		t.Fatalf("the same key replaces: %v", list)
++	}
++	fi, _ := os.Stat(filepath.Join(dir, File))
++	if runtime.GOOS != "windows" && fi.Mode().Perm() != 0o600 { // Windows keeps no Unix mode bits
++		t.Fatalf("mode %v", fi.Mode().Perm())
++	}
++	if n, _ := Remove(dir, "8791"); n != 1 {
++		t.Fatal("remove by port")
++	}
++	if list, _ := Load(dir); len(list) != 0 {
++		t.Fatal("removed")
++	}
++	if Redact(key) != "ovc_CCCCCC…" {
++		t.Fatalf("redact %q", Redact(key))
++	}
++}
ddiff --git a/internal/connect/store.go b/internal/connect/store.go
nnew file mode 100644
iindex 0000000..226a285
---- /dev/null
++++ b/internal/connect/store.go
@@@ -0,0 +1,137 @@
++package connect
++
++import (
++	"context"
++	"encoding/json"
++	"errors"
++	"fmt"
++	"os"
++	"path/filepath"
++	"sort"
++	"sync"
++)
++
++// The shares the Node keeps up live in <config>/connect-shares.json, readable only by the Node's own account (0600):
++// a connect key lets a computer serve that share, so it is kept like a credential and never logged or printed whole.
++
++// File is the saved shares' file name inside the config directory.
++const File = "connect-shares.json"
++
++type saved struct {
++	Shares []Share `json:"shares"`
++}
++
++// Load reads the saved shares; a missing file is no shares.
++func Load(dir string) ([]Share, error) {
++	b, err := os.ReadFile(filepath.Join(dir, File))
++	if errors.Is(err, os.ErrNotExist) {
++		return nil, nil
++	}
++	if err != nil {
++		return nil, err
++	}
++	var s saved
++	if err := json.Unmarshal(b, &s); err != nil {
++		return nil, fmt.Errorf("%s: %w", File, err)
++	}
++	return s.Shares, nil
++}
++
++func save(dir string, shares []Share) error {
++	sort.Slice(shares, func(i, j int) bool { return shares[i].Port < shares[j].Port })
++	b, err := json.MarshalIndent(saved{Shares: shares}, "", "  ")
++	if err != nil {
++		return err
++	}
++	if err := os.MkdirAll(dir, 0o700); err != nil {
++		return err
++	}
++	tmp := filepath.Join(dir, File+".tmp")
++	if err := os.WriteFile(tmp, append(b, '\n'), 0o600); err != nil {
++		return err
++	}
++	return os.Rename(tmp, filepath.Join(dir, File))
++}
++
++// Add saves a share (the same key again replaces its port and name).
++func Add(dir string, s Share) error {
++	shares, err := Load(dir)
++	if err != nil {
++		return err
++	}
++	out := shares[:0]
++	for _, x := range shares {
++		if x.Key != s.Key {
++			out = append(out, x)
++		}
++	}
++	return save(dir, append(out, s))
++}
++
++// Remove forgets the share with this key, or every share on this port; it reports how many it removed.
++func Remove(dir string, keyOrPort string) (int, error) {
++	shares, err := Load(dir)
++	if err != nil {
++		return 0, err
++	}
++	out := shares[:0]
++	n := 0
++	for _, x := range shares {
++		if x.Key == keyOrPort || fmt.Sprint(x.Port) == keyOrPort || (x.Name != "" && x.Name == keyOrPort) {
++			n++
++			continue
++		}
++		out = append(out, x)
++	}
++	if n == 0 {
++		return 0, nil
++	}
++	return n, save(dir, out)
++}
++
++// Redact shows a key's first characters only.
++func Redact(key string) string {
++	if len(key) <= 10 {
++		return "ovc_…"
++	}
++	return key[:10] + "…"
++}
++
++// RunAll keeps every saved share up until ctx ends; a share the server stops is left stopped (its key is wrong, it
++// was closed, or another computer took it over). It remembers each share's address as the server names it.
++func RunAll(ctx context.Context, dir string, o Options) {
++	shares, err := Load(dir)
++	if err != nil {
++		if o.Log != nil {
++			o.Log.Warn("connect: saved shares unreadable", "err", err)
++		}
++		return
++	}
++	var wg sync.WaitGroup
++	var mu sync.Mutex
++	for _, s := range shares {
++		s := s
++		so := o
++		so.OnReady = func(r Ready) {
++			if s.Name == r.Host {
++				return
++			}
++			mu.Lock()
++			defer mu.Unlock()
++			s.Name = r.Host
++			_ = Add(dir, s)
++		}
++		wg.Add(1)
++		go func() {
++			defer wg.Done()
++			err := Run(ctx, s, so)
++			if errors.Is(err, ErrStopped) && o.Log != nil {
++				o.Log.Warn("connect: share stopped; remove it with `openvibe-node share remove`", "port", s.Port, "key", Redact(s.Key))
++			}
++		}()
++	}
++	if len(shares) > 0 && o.Log != nil {
++		o.Log.Info("connect: keeping shares up", "count", len(shares))
++	}
++	wg.Wait()
++}