openvibers/openvibe.link Commits

exchange serve: sell a local model through the Exchange endpoint flow (#120)

OP OpenVibers committed 6 days ago · a395363427 · 5 files, +757 −2

Changes

ddiff --git a/README.md b/README.md
iindex a45691a..7077493 100644
---- a/README.md
++++ b/README.md
@@@ -33,6 +33,14 @@ serves the share this computer was picked for at `https://<name>--<you>.openvibe
`openvibe-node share --key ovc_… --port 8790` with a key), and `openvibe-node share add` keeps it up with the service;
`openvibe-node status` lists each share ([docs/connect.md](docs/connect.md)).
++**Sell a local model** on [OpenVibe.Exchange](https://openvibe.exchange):
++`openvibe-node exchange serve --model qwen3:32b --public-url https://gpu.example.com/v1`. It serves the model from the
++local OpenAI-compatible server (Ollama by default at `127.0.0.1:11434`), registers the public https address through the
++Exchange's endpoint-seller flow with your `OPENVIBE_DEVELOPER_KEY`, and removes the offer when you stop it with Ctrl-C.
++The public URL must be reachable by the Exchange and is judged by its egress guard; keep it behind your own tunnel or
++reverse proxy. The local listener has no buyer authentication (the Exchange's endpoint flow sends no buyer identity), so
++bind it to loopback, which is the default.
++
```sh
openvibe-node # guided first run: pair, check the link, report this machine's first job
openvibe-node status # link, plugins, video, battery, shares, stop latch
@@@ -44,6 +52,7 @@ openvibe-node tray # the same in your notification area, with Approve/Deny
openvibe-node memory scan # what your coding tools remember: read-only, local, secrets removed
openvibe-node memory upload # send what they remember to your own memory store, after you say yes
openvibe-node memory sync # write your profile and rules back into the tools, one marked block per file, opt-in per tool
++openvibe-node exchange serve --model qwen3:32b --public-url https://gpu.example.com/v1 # sell a local model
```
## What is in here
ddiff --git a/STATUS.json b/STATUS.json
iindex 52a61c5..eee4e4f 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -23,7 +23,8 @@
"OpenVibe Connect shares (CON-4): `openvibe-node share <port>` serves the share this computer was picked for, with no key; every saved share is in `openvibe-node status` (address, connected or why not, requests) and stops when the Node stops",
"one-command install on a Raspberry Pi or other Linux robot (I2C, SPI, camera tools, kit detection)",
"system service on systemd, launchd and Windows (or a Windows sign-in start)",
-- "ESP32 library (OpenVibeNode): pairing, control link, drive and actuator commands, telemetry, heartbeat, deadman and e-stop"
++ "ESP32 library (OpenVibeNode): pairing, control link, drive and actuator commands, telemetry, heartbeat, deadman and e-stop",
++ "`openvibe-node exchange serve --model <id>`: serve a local Ollama/llama.cpp/vLLM model through an OpenAI-compatible listener and register it on OpenVibe.Exchange's endpoint-seller API with the person's OPENVIBE_DEVELOPER_KEY, then remove the offer on Ctrl-C"
],
"notYet": [
"code signing and notarization for Windows and macOS (the release has SHA256SUMS only)",
@@@ -31,5 +32,5 @@
"a tray on macOS (approvals there are a dialog with Deny as its default)",
"the computer plugin's browser on Windows"
],
-- "updated": "2026-11-06"
++ "updated": "2026-11-07"
}
ddiff --git a/cmd/openvibe-node/exchange.go b/cmd/openvibe-node/exchange.go
nnew file mode 100644
iindex 0000000..d91042f
---- /dev/null
++++ b/cmd/openvibe-node/exchange.go
@@@ -0,0 +1,535 @@
++package main
++
++// `openvibe-node exchange serve --model <id>`: serve a model that runs on this machine (Ollama, llama.cpp, vLLM …)
++// as a seller offer on OpenVibe.Exchange. It listens on a local OpenAI-compatible port, forwards ONE model to the
++// local server that actually runs it, and registers this machine's public https address through the Exchange's
++// existing endpoint-seller flow (POST /api/v1/exchange/seller/offers) with the person's Network developer key. The
++// Exchange reaches the public URL only through its egress guard and verifies the endpoint with canaries exactly as
++// it verifies an OpenVibe Node machine; on Ctrl-C the offer is removed again.
++//
++// The local listener does not authenticate the Exchange: the Exchange's endpoint flow builds a fresh request and
++// sends no buyer identity by design. Keep --public-url behind the person's own tunnel/reverse proxy, and bind
++// --listen to loopback (the default) so only that tunnel reaches it.
++
++import (
++ "bytes"
++ "context"
++ "encoding/json"
++ "errors"
++ "flag"
++ "fmt"
++ "io"
++ "net"
++ "net/http"
++ "net/url"
++ "os"
++ "os/signal"
++ "strings"
++ "syscall"
++ "time"
++)
++
++const (
++ exchangeDefaultURL = "https://api.openvibe.exchange"
++ exchangeDefaultUpstream = "http://127.0.0.1:11434/v1"
++ exchangeDefaultListen = "127.0.0.1:11435"
++ exchangeMaxBody = 2 * 1024 * 1024
++ exchangeMaxAnswer = 8 * 1024 * 1024
++ exchangeTokenEnv = "OPENVIBE_DEVELOPER_KEY"
++)
++
++const exchangeUsage = `usage:
++ openvibe-node exchange serve --model <id> [flags]
++
++Serve a local model as an OpenVibe.Exchange seller offer through the endpoint-seller flow.
++
++ --model <id> the model id buyers ask for (required), e.g. qwen3:32b
++ --upstream URL the local OpenAI-compatible server (default http://127.0.0.1:11434/v1, Ollama)
++ --upstream-model <id> the model name the upstream knows, when it differs from --model
++ --listen ADDR the local address to serve (default 127.0.0.1:11435)
++ --public-url URL the https base URL the Exchange calls (required unless --no-register), e.g.
++ https://gpu.example.com/v1; /chat/completions is appended
++ --price-in N credits per million input tokens (default 0: volunteer)
++ --price-out N credits per million output tokens (default 0)
++ --price N one price for both sides, unless --price-in/--price-out is given
++ --concurrency N calls in flight at once, 1-8 (default 2)
++ --hours-start HH:MM when the offer opens, UTC by default (default 00:00)
++ --hours-end HH:MM when it closes (default 23:59)
++ --tz ZONE the time zone of the hours (default UTC)
++ --audience WHO everyone, followers or group (default everyone)
++ --group ID the group, when --audience group
++ --name NAME a name for the offer
++ --token-env NAME where the person's Network developer key lives (default OPENVIBE_DEVELOPER_KEY)
++ --exchange URL the Exchange API origin (default https://api.openvibe.exchange)
++ --no-register serve locally only, without registering an offer
++
++The command prints the local address and the registered offer id, then serves until Ctrl-C. It removes the offer on
++exit. The developer key is read from the environment and is never printed.`
++
++type exchangeServeOptions struct {
++ Model string
++ Upstream string
++ UpstreamModel string
++ Listen string
++ PublicURL string
++ Price float64
++ PriceIn float64
++ PriceOut float64
++ Concurrency int
++ HoursStart string
++ HoursEnd string
++ TZ string
++ Audience string
++ Group string
++ Name string
++ TokenEnv string
++ ExchangeURL string
++ NoRegister bool
++ UpstreamToken string // a local upstream's own key, from --upstream-token-env (never printed)
++
++ // Ready is the tests' seam: called once the listener is up and the offer registered (empty when not registered).
++ Ready func(localAddr, offerID string)
++ // Client is the tests' seam for the Exchange and the upstream; nil uses a bounded default.
++ Client *http.Client
++}
++
++type exchangeOfferAnswer struct {
++ Offer struct {
++ ID string `json:"id"`
++ } `json:"offer"`
++}
++
++func cmdExchange(args []string, stdout, stderr io.Writer) int {
++ if len(args) == 0 || args[0] != "serve" {
++ fmt.Fprintln(stderr, exchangeUsage)
++ return 2
++ }
++ fs := flag.NewFlagSet("exchange serve", flag.ContinueOnError)
++ fs.SetOutput(stderr)
++ var o exchangeServeOptions
++ var upstreamTokenEnv string
++ fs.StringVar(&o.Model, "model", "", "the model id buyers ask for")
++ fs.StringVar(&o.Upstream, "upstream", exchangeDefaultUpstream, "the local OpenAI-compatible server")
++ fs.StringVar(&o.UpstreamModel, "upstream-model", "", "the model name the upstream knows")
++ fs.StringVar(&o.Listen, "listen", exchangeDefaultListen, "the local address to serve")
++ fs.StringVar(&o.PublicURL, "public-url", "", "the public https base URL the Exchange calls")
++ fs.Float64Var(&o.Price, "price", 0, "one price for both sides, credits per million tokens")
++ fs.Float64Var(&o.PriceIn, "price-in", 0, "credits per million input tokens")
++ fs.Float64Var(&o.PriceOut, "price-out", 0, "credits per million output tokens")
++ fs.IntVar(&o.Concurrency, "concurrency", 2, "calls in flight at once (1-8)")
++ fs.StringVar(&o.HoursStart, "hours-start", "00:00", "when the offer opens, HH:MM")
++ fs.StringVar(&o.HoursEnd, "hours-end", "23:59", "when the offer closes, HH:MM")
++ fs.StringVar(&o.TZ, "tz", "UTC", "the time zone of the hours")
++ fs.StringVar(&o.Audience, "audience", "everyone", "everyone, followers or group")
++ fs.StringVar(&o.Group, "group", "", "the group, when --audience group")
++ fs.StringVar(&o.Name, "name", "", "a name for the offer")
++ fs.StringVar(&o.TokenEnv, "token-env", exchangeTokenEnv, "where the Network developer key lives")
++ fs.StringVar(&o.ExchangeURL, "exchange", exchangeDefaultURL, "the Exchange API origin")
++ fs.BoolVar(&o.NoRegister, "no-register", false, "serve locally only")
++ fs.StringVar(&upstreamTokenEnv, "upstream-token-env", "", "where the local upstream's key lives, when it needs one")
++ if err := fs.Parse(args[1:]); err != nil {
++ return 2
++ }
++ if fs.NArg() != 0 {
++ fmt.Fprintln(stderr, exchangeUsage)
++ return 2
++ }
++ if upstreamTokenEnv != "" {
++ o.UpstreamToken = os.Getenv(upstreamTokenEnv)
++ if o.UpstreamToken == "" {
++ fmt.Fprintf(stderr, "error: %s is empty (the local upstream needs a key)\n", upstreamTokenEnv)
++ return 1
++ }
++ }
++ ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
++ defer stop()
++ if err := runExchangeServe(ctx, o, stdout); err != nil {
++ if errors.Is(err, context.Canceled) {
++ return 0
++ }
++ fmt.Fprintln(stderr, "error:", err)
++ return 1
++ }
++ return 0
++}
++
++func normalizeExchangeBase(raw, field string) (*url.URL, error) {
++ u, err := url.Parse(strings.TrimSpace(raw))
++ if err != nil || u.Scheme == "" || u.Host == "" {
++ return nil, fmt.Errorf("%s must be an http(s) URL", field)
++ }
++ if u.Scheme != "http" && u.Scheme != "https" {
++ return nil, fmt.Errorf("%s must be an http(s) URL", field)
++ }
++ if u.User != nil {
++ return nil, fmt.Errorf("%s must not carry credentials", field)
++ }
++ return u, nil
++}
++
++// judgePublicURL applies the client side of the Exchange's own rule: https, a host name (never an IP literal or a
++// private name), port 443 or 8443, no query or fragment. The Exchange judges it again; this fails fast and honestly.
++func judgePublicURL(raw string) (*url.URL, error) {
++ u, err := url.Parse(strings.TrimSpace(raw))
++ if err != nil || u.Scheme == "" || u.Host == "" {
++ return nil, errors.New("--public-url must be an https URL")
++ }
++ if u.Scheme != "https" {
++ return nil, errors.New("--public-url must be https")
++ }
++ if u.User != nil {
++ return nil, errors.New("--public-url must not carry credentials")
++ }
++ if u.RawQuery != "" || u.Fragment != "" {
++ return nil, errors.New("--public-url must not carry a query or fragment")
++ }
++ if p := u.Port(); p != "" && p != "443" && p != "8443" {
++ return nil, errors.New("--public-url must use port 443 or 8443")
++ }
++ host := strings.ToLower(strings.Trim(u.Hostname(), "[]"))
++ if host == "" {
++ return nil, errors.New("--public-url must name a host")
++ }
++ if ip := net.ParseIP(host); ip != nil {
++ return nil, errors.New("--public-url must name a host, not an IP address")
++ }
++ if host == "localhost" || strings.HasSuffix(host, ".localhost") || strings.HasSuffix(host, ".local") ||
++ strings.HasSuffix(host, ".internal") || strings.HasSuffix(host, ".home.arpa") {
++ return nil, errors.New("--public-url must be a public host name")
++ }
++ u.Path = strings.TrimRight(u.Path, "/")
++ return u, nil
++}
++
++func hhmm(s string) bool {
++ if len(s) != 5 || s[2] != ':' {
++ return false
++ }
++ for i, c := range s {
++ if i == 2 {
++ continue
++ }
++ if c < '0' || c > '9' {
++ return false
++ }
++ }
++ h := int(s[0]-'0')*10 + int(s[1]-'0')
++ m := int(s[3]-'0')*10 + int(s[4]-'0')
++ return h < 24 && m < 60
++}
++
++func (o *exchangeServeOptions) validate() error {
++ if strings.TrimSpace(o.Model) == "" {
++ return errors.New("--model is the model id buyers ask for (for example qwen3:32b)")
++ }
++ if o.UpstreamModel == "" {
++ o.UpstreamModel = o.Model
++ }
++ if _, err := normalizeExchangeBase(o.Upstream, "--upstream"); err != nil {
++ return err
++ }
++ o.Upstream = strings.TrimRight(o.Upstream, "/")
++ if _, err := normalizeExchangeBase(o.ExchangeURL, "--exchange"); err != nil {
++ return err
++ }
++ o.ExchangeURL = strings.TrimRight(o.ExchangeURL, "/")
++ if _, err := net.ResolveTCPAddr("tcp", o.Listen); err != nil {
++ return fmt.Errorf("--listen %q is not an address: %v", o.Listen, err)
++ }
++ if !o.NoRegister {
++ if _, err := judgePublicURL(o.PublicURL); err != nil {
++ return err
++ }
++ o.PublicURL = strings.TrimRight(o.PublicURL, "/")
++ if strings.TrimSpace(os.Getenv(o.TokenEnv)) == "" {
++ return fmt.Errorf("%s is not set: make a Network developer key and export it there (or use --no-register to serve locally only)", o.TokenEnv)
++ }
++ }
++ if o.Price > 0 && o.PriceIn == 0 && o.PriceOut == 0 {
++ o.PriceIn, o.PriceOut = o.Price, o.Price
++ }
++ if o.PriceIn < 0 || o.PriceOut < 0 {
++ return errors.New("--price-in and --price-out cannot be negative")
++ }
++ if o.Concurrency < 1 || o.Concurrency > 8 {
++ return errors.New("--concurrency is an integer from 1 to 8")
++ }
++ if !hhmm(o.HoursStart) || !hhmm(o.HoursEnd) {
++ return errors.New("--hours-start and --hours-end are HH:MM")
++ }
++ if strings.TrimSpace(o.TZ) == "" || len(o.TZ) > 64 {
++ return errors.New("--tz is a time zone name, at most 64 characters")
++ }
++ if o.Audience != "everyone" && o.Audience != "followers" && o.Audience != "group" {
++ return errors.New("--audience is everyone, followers or group")
++ }
++ if o.Audience == "group" && strings.TrimSpace(o.Group) == "" {
++ return errors.New("--audience group needs --group")
++ }
++ if len(o.Name) > 100 {
++ return errors.New("--name is at most 100 characters")
++ }
++ return nil
++}
++
++func exchangeClient(o *exchangeServeOptions) *http.Client {
++ if o.Client != nil {
++ return o.Client
++ }
++ return &http.Client{Timeout: 60 * time.Second}
++}
++
++func exchangeToken(o *exchangeServeOptions) string { return strings.TrimSpace(os.Getenv(o.TokenEnv)) }
++
++func registerExchangeOffer(ctx context.Context, o *exchangeServeOptions) (string, error) {
++ price := map[string]float64{"in_per_million": o.PriceIn, "out_per_million": o.PriceOut}
++ body := map[string]any{
++ "base_url": o.PublicURL,
++ "models": []string{o.Model},
++ "price": price,
++ "hours": map[string]string{"start": o.HoursStart, "end": o.HoursEnd, "tz": o.TZ},
++ "concurrency": o.Concurrency,
++ "audience": o.Audience,
++ }
++ if o.Group != "" {
++ body["group"] = o.Group
++ }
++ if o.Name != "" {
++ body["name"] = o.Name
++ }
++ raw, err := json.Marshal(body)
++ if err != nil {
++ return "", err
++ }
++ req, err := http.NewRequestWithContext(ctx, http.MethodPost, o.ExchangeURL+"/api/v1/exchange/seller/offers", bytes.NewReader(raw))
++ if err != nil {
++ return "", err
++ }
++ req.Header.Set("Content-Type", "application/json")
++ req.Header.Set("Accept", "application/json")
++ req.Header.Set("Authorization", "Bearer "+exchangeToken(o))
++ req.Header.Set("User-Agent", "openvibe-node/"+version)
++ res, err := exchangeClient(o).Do(req)
++ if err != nil {
++ return "", fmt.Errorf("the Exchange could not be reached: %w", err)
++ }
++ defer res.Body.Close()
++ answer, _ := io.ReadAll(io.LimitReader(res.Body, 64*1024))
++ if res.StatusCode < 200 || res.StatusCode >= 300 {
++ return "", fmt.Errorf("the Exchange answered HTTP %d: %s", res.StatusCode, exchangeDetail(answer))
++ }
++ var out exchangeOfferAnswer
++ if err := json.Unmarshal(answer, &out); err != nil || out.Offer.ID == "" {
++ return "", errors.New("the Exchange answered without an offer id")
++ }
++ return out.Offer.ID, nil
++}
++
++func removeExchangeOffer(ctx context.Context, o *exchangeServeOptions, id string) error {
++ req, err := http.NewRequestWithContext(ctx, http.MethodDelete, o.ExchangeURL+"/api/v1/exchange/seller/offers/"+url.PathEscape(id), nil)
++ if err != nil {
++ return err
++ }
++ req.Header.Set("Accept", "application/json")
++ req.Header.Set("Authorization", "Bearer "+exchangeToken(o))
++ req.Header.Set("User-Agent", "openvibe-node/"+version)
++ res, err := exchangeClient(o).Do(req)
++ if err != nil {
++ return err
++ }
++ defer res.Body.Close()
++ io.Copy(io.Discard, io.LimitReader(res.Body, 64*1024))
++ if res.StatusCode < 200 || res.StatusCode >= 300 {
++ return fmt.Errorf("the Exchange answered HTTP %d to the removal", res.StatusCode)
++ }
++ return nil
++}
++
++// exchangeDetail is the problem document's sentence, never the whole body and never a token (which the Exchange never
++// sends anyway).
++func exchangeDetail(body []byte) string {
++ var p struct {
++ Detail string `json:"detail"`
++ Title string `json:"title"`
++ Code string `json:"code"`
++ }
++ if json.Unmarshal(body, &p) == nil {
++ for _, s := range []string{p.Detail, p.Title, p.Code} {
++ if strings.TrimSpace(s) != "" {
++ return strings.TrimSpace(s)
++ }
++ }
++ }
++ return strings.TrimSpace(string(body))[:min(len(strings.TrimSpace(string(body))), 300)]
++}
++
++func exchangeModels(o *exchangeServeOptions) []byte {
++ out, _ := json.Marshal(map[string]any{
++ "object": "list",
++ "data": []map[string]any{{
++ "id": o.Model, "object": "model", "created": time.Now().Unix(), "owned_by": "openvibe-node",
++ }},
++ })
++ return out
++}
++
++func exchangeError(w http.ResponseWriter, status int, message string) {
++ w.Header().Set("Content-Type", "application/json")
++ w.WriteHeader(status)
++ json.NewEncoder(w).Encode(map[string]any{"error": map[string]any{"message": message, "type": "invalid_request_error"}})
++}
++
++func exchangeUpstreamPath(base string) string {
++ u, err := url.Parse(base)
++ if err != nil {
++ return base + "/chat/completions"
++ }
++ if strings.HasSuffix(strings.TrimRight(u.Path, "/"), "/v1") {
++ u.Path = strings.TrimRight(u.Path, "/") + "/chat/completions"
++ } else {
++ u.Path = strings.TrimRight(u.Path, "/") + "/v1/chat/completions"
++ }
++ return u.String()
++}
++
++// exchangeChat forwards one completion to the local upstream. Only the registered model is served; the request is
++// rebuilt around the model, the conversation, the output limit and the temperature, never the buyer's identity.
++func exchangeChat(o *exchangeServeOptions, w http.ResponseWriter, r *http.Request) {
++ if r.Method != http.MethodPost {
++ exchangeError(w, http.StatusMethodNotAllowed, "POST only")
++ return
++ }
++ raw, err := io.ReadAll(io.LimitReader(r.Body, exchangeMaxBody+1))
++ if err != nil {
++ exchangeError(w, http.StatusBadRequest, "the request body could not be read")
++ return
++ }
++ if len(raw) > exchangeMaxBody {
++ exchangeError(w, http.StatusRequestEntityTooLarge, "the request body is too large")
++ return
++ }
++ var asked struct {
++ Model string `json:"model"`
++ Stream bool `json:"stream"`
++ }
++ if err := json.Unmarshal(raw, &asked); err != nil {
++ exchangeError(w, http.StatusBadRequest, "the request is not JSON")
++ return
++ }
++ if asked.Model != "" && asked.Model != o.Model && asked.Model != o.UpstreamModel {
++ exchangeError(w, http.StatusNotFound, fmt.Sprintf("this node serves %s, not %s", o.Model, asked.Model))
++ return
++ }
++ var body map[string]any
++ if err := json.Unmarshal(raw, &body); err != nil || body == nil {
++ exchangeError(w, http.StatusBadRequest, "the request is not a JSON object")
++ return
++ }
++ body["model"] = o.UpstreamModel
++ body["stream"] = false // the endpoint-seller flow reads one answer, never a stream
++ forwarded, _ := json.Marshal(body)
++
++ ctx, cancel := context.WithTimeout(r.Context(), 10*time.Minute)
++ defer cancel()
++ req, err := http.NewRequestWithContext(ctx, http.MethodPost, exchangeUpstreamPath(o.Upstream), bytes.NewReader(forwarded))
++ if err != nil {
++ exchangeError(w, http.StatusBadGateway, "the local model could not be called")
++ return
++ }
++ req.Header.Set("Content-Type", "application/json")
++ req.Header.Set("Accept", "application/json")
++ if o.UpstreamToken != "" {
++ req.Header.Set("Authorization", "Bearer "+o.UpstreamToken)
++ }
++ res, err := exchangeClient(o).Do(req)
++ if err != nil {
++ exchangeError(w, http.StatusBadGateway, "the local model could not be reached")
++ return
++ }
++ defer res.Body.Close()
++ answer, err := io.ReadAll(io.LimitReader(res.Body, exchangeMaxAnswer+1))
++ if err != nil || len(answer) > exchangeMaxAnswer {
++ exchangeError(w, http.StatusBadGateway, "the local model's answer could not be read")
++ return
++ }
++ contentType := res.Header.Get("Content-Type")
++ if contentType == "" {
++ contentType = "application/json"
++ }
++ w.Header().Set("Content-Type", contentType)
++ w.WriteHeader(res.StatusCode)
++ w.Write(answer)
++}
++
++func exchangeMux(o *exchangeServeOptions) *http.ServeMux {
++ mux := http.NewServeMux()
++ models := func(w http.ResponseWriter, r *http.Request) {
++ if r.Method != http.MethodGet {
++ exchangeError(w, http.StatusMethodNotAllowed, "GET only")
++ return
++ }
++ w.Header().Set("Content-Type", "application/json")
++ w.Write(exchangeModels(o))
++ }
++ chat := func(w http.ResponseWriter, r *http.Request) { exchangeChat(o, w, r) }
++ mux.HandleFunc("/v1/models", models)
++ mux.HandleFunc("/v1/chat/completions", chat)
++ // A --public-url without /v1 gets /chat/completions appended by the Exchange; accept that too.
++ mux.HandleFunc("/models", models)
++ mux.HandleFunc("/chat/completions", chat)
++ return mux
++}
++
++// runExchangeServe is the command without the signal handling, so a test can cancel its context.
++func runExchangeServe(ctx context.Context, o exchangeServeOptions, stdout io.Writer) error {
++ if err := o.validate(); err != nil {
++ return err
++ }
++ ln, err := net.Listen("tcp", o.Listen)
++ if err != nil {
++ return fmt.Errorf("--listen %s: %w", o.Listen, err)
++ }
++ server := &http.Server{Handler: exchangeMux(&o), ReadHeaderTimeout: 10 * time.Second}
++ served := make(chan error, 1)
++ go func() { served <- server.Serve(ln) }()
++
++ offerID := ""
++ if !o.NoRegister {
++ registerCtx, cancel := context.WithTimeout(ctx, 45*time.Second)
++ offerID, err = registerExchangeOffer(registerCtx, &o)
++ cancel()
++ if err != nil {
++ server.Close()
++ return err
++ }
++ }
++ fmt.Fprintf(stdout, "Serving %s on http://%s (upstream %s)\n", o.Model, ln.Addr().String(), o.Upstream)
++ if offerID != "" {
++ fmt.Fprintf(stdout, "Registered %s on the Exchange (%s) at %s\n", o.Model, o.ExchangeURL, o.PublicURL)
++ fmt.Fprintf(stdout, "Offer %s; it is removed when this command stops.\n", offerID)
++ }
++ if o.Ready != nil {
++ o.Ready(ln.Addr().String(), offerID)
++ }
++
++ select {
++ case <-ctx.Done():
++ case err := <-served:
++ if err != nil && !errors.Is(err, http.ErrServerClosed) {
++ return err
++ }
++ }
++ shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
++ defer cancel()
++ server.Shutdown(shutdownCtx)
++ if offerID != "" {
++ removeCtx, cancelRemove := context.WithTimeout(context.Background(), 20*time.Second)
++ defer cancelRemove()
++ if err := removeExchangeOffer(removeCtx, &o, offerID); err != nil {
++ fmt.Fprintf(stdout, "The offer %s could not be removed: %v\n", offerID, err)
++ } else {
++ fmt.Fprintf(stdout, "Removed offer %s\n", offerID)
++ }
++ }
++ return nil
++}
ddiff --git a/cmd/openvibe-node/exchange_test.go b/cmd/openvibe-node/exchange_test.go
nnew file mode 100644
iindex 0000000..1257a85
---- /dev/null
++++ b/cmd/openvibe-node/exchange_test.go
@@@ -0,0 +1,205 @@
++package main
++
++import (
++ "bytes"
++ "context"
++ "encoding/json"
++ "io"
++ "net/http"
++ "net/http/httptest"
++ "strings"
++ "sync"
++ "testing"
++ "time"
++)
++
++// `openvibe-node exchange serve`: the seller offer is registered through the endpoint-seller flow with the person's
++// developer key, the local model is served without its own auth, the model the buyer names is rewritten to the one the
++// local upstream knows, and Ctrl-C removes the offer again.
++func TestExchangeServeRegistersForwardsAndRemoves(t *testing.T) {
++ var mu sync.Mutex
++ var registered map[string]any
++ var registerAuth, removedPath, removedAuth string
++ exchange := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++ mu.Lock()
++ defer mu.Unlock()
++ w.Header().Set("Content-Type", "application/json")
++ switch {
++ case r.Method == http.MethodPost && r.URL.Path == "/api/v1/exchange/seller/offers":
++ registerAuth = r.Header.Get("Authorization")
++ json.NewDecoder(r.Body).Decode(&registered)
++ w.WriteHeader(http.StatusCreated)
++ io.WriteString(w, `{"offer":{"id":"ep_test"}}`)
++ case r.Method == http.MethodDelete && strings.HasPrefix(r.URL.Path, "/api/v1/exchange/seller/offers/"):
++ removedPath = r.URL.Path
++ removedAuth = r.Header.Get("Authorization")
++ io.WriteString(w, `{"id":"ep_test","removed":true}`)
++ default:
++ w.WriteHeader(http.StatusNotFound)
++ io.WriteString(w, `{"detail":"not found"}`)
++ }
++ }))
++ defer exchange.Close()
++
++ var upstreamModel string
++ var upstreamStream bool
++ upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++ if r.URL.Path != "/v1/chat/completions" {
++ t.Errorf("the upstream path: %s", r.URL.Path)
++ }
++ var body map[string]any
++ json.NewDecoder(r.Body).Decode(&body)
++ upstreamModel, _ = body["model"].(string)
++ upstreamStream, _ = body["stream"].(bool)
++ w.Header().Set("Content-Type", "application/json")
++ io.WriteString(w, `{"id":"chatcmpl_1","choices":[{"index":0,"message":{"role":"assistant","content":"local answer"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":2}}`)
++ }))
++ defer upstream.Close()
++
++ t.Setenv("OV_TEST_EXCHANGE_TOKEN", "dev-token")
++ ctx, cancel := context.WithCancel(context.Background())
++ ready := make(chan string, 1)
++ done := make(chan error, 1)
++ go func() {
++ done <- runExchangeServe(ctx, exchangeServeOptions{
++ Model: "qwen3:32b", Upstream: upstream.URL, UpstreamModel: "qwen3-upstream",
++ Listen: "127.0.0.1:0", PublicURL: "https://seller.example/v1",
++ PriceIn: 10, PriceOut: 30, Concurrency: 2, HoursStart: "00:00", HoursEnd: "23:59", TZ: "UTC",
++ Audience: "everyone", Name: "test box", TokenEnv: "OV_TEST_EXCHANGE_TOKEN", ExchangeURL: exchange.URL,
++ Ready: func(addr, _ string) { ready <- addr },
++ }, io.Discard)
++ }()
++
++ var addr string
++ select {
++ case addr = <-ready:
++ case err := <-done:
++ t.Fatalf("serve stopped before it was ready: %v", err)
++ case <-time.After(5 * time.Second):
++ t.Fatal("serve did not become ready")
++ }
++
++ // The model list is the one model this node sells.
++ res, err := http.Get("http://" + addr + "/v1/models")
++ if err != nil {
++ t.Fatal(err)
++ }
++ var models struct {
++ Data []struct {
++ ID string `json:"id"`
++ } `json:"data"`
++ }
++ json.NewDecoder(res.Body).Decode(&models)
++ res.Body.Close()
++ if len(models.Data) != 1 || models.Data[0].ID != "qwen3:32b" {
++ t.Fatalf("models: %+v", models)
++ }
++
++ // One completion: the buyer's model is accepted, rewritten for the upstream, and stream is forced off.
++ res, err = http.Post("http://"+addr+"/v1/chat/completions", "application/json", strings.NewReader(`{"model":"qwen3:32b","messages":[{"role":"user","content":"hi"}],"stream":true}`))
++ if err != nil {
++ t.Fatal(err)
++ }
++ answer, _ := io.ReadAll(res.Body)
++ res.Body.Close()
++ if res.StatusCode != http.StatusOK || !strings.Contains(string(answer), "local answer") {
++ t.Fatalf("answer %d %s", res.StatusCode, answer)
++ }
++ if upstreamModel != "qwen3-upstream" {
++ t.Fatalf("upstream model %q", upstreamModel)
++ }
++ if upstreamStream {
++ t.Fatal("stream was not forced off")
++ }
++
++ // A model this node does not sell is refused before the upstream sees it.
++ res, err = http.Post("http://"+addr+"/v1/chat/completions", "application/json", strings.NewReader(`{"model":"other/model","messages":[{"role":"user","content":"hi"}]}`))
++ if err != nil {
++ t.Fatal(err)
++ }
++ res.Body.Close()
++ if res.StatusCode != http.StatusNotFound {
++ t.Fatalf("another model: %d", res.StatusCode)
++ }
++
++ cancel()
++ select {
++ case err := <-done:
++ if err != nil {
++ t.Fatalf("serve: %v", err)
++ }
++ case <-time.After(5 * time.Second):
++ t.Fatal("serve did not stop")
++ }
++
++ mu.Lock()
++ defer mu.Unlock()
++ if registerAuth != "Bearer dev-token" {
++ t.Fatalf("registration auth: %q", registerAuth)
++ }
++ if removedPath != "/api/v1/exchange/seller/offers/ep_test" || removedAuth != "Bearer dev-token" {
++ t.Fatalf("removal %s %q", removedPath, removedAuth)
++ }
++ if registered["base_url"] != "https://seller.example/v1" {
++ t.Fatalf("base_url: %v", registered["base_url"])
++ }
++ if m, _ := registered["models"].([]any); len(m) != 1 || m[0] != "qwen3:32b" {
++ t.Fatalf("models: %v", registered["models"])
++ }
++ price, _ := registered["price"].(map[string]any)
++ if price["in_per_million"] != float64(10) || price["out_per_million"] != float64(30) {
++ t.Fatalf("price: %v", registered["price"])
++ }
++ if registered["concurrency"] != float64(2) || registered["audience"] != "everyone" {
++ t.Fatalf("offer: %v", registered)
++ }
++ hours, _ := registered["hours"].(map[string]any)
++ if hours["start"] != "00:00" || hours["end"] != "23:59" || hours["tz"] != "UTC" {
++ t.Fatalf("hours: %v", registered["hours"])
++ }
++}
++
++func TestExchangeServeRefusesAPrivateOrIPPublicURL(t *testing.T) {
++ t.Setenv("OV_TEST_EXCHANGE_TOKEN", "dev-token")
++ for _, pub := range []string{"http://seller.example/v1", "https://127.0.0.1/v1", "https://[::1]/v1", "https://localhost/v1", "https://seller.example:8080/v1"} {
++ err := runExchangeServe(context.Background(), exchangeServeOptions{
++ Model: "qwen3:32b", Upstream: "http://127.0.0.1:11434/v1", Listen: "127.0.0.1:0",
++ PublicURL: pub, Concurrency: 2, HoursStart: "00:00", HoursEnd: "23:59", TZ: "UTC",
++ Audience: "everyone", TokenEnv: "OV_TEST_EXCHANGE_TOKEN", ExchangeURL: "http://127.0.0.1:1",
++ }, io.Discard)
++ if err == nil || !strings.Contains(err.Error(), "public-url") {
++ t.Fatalf("%s: %v", pub, err)
++ }
++ }
++}
++
++func TestExchangeServeRefusesARefusedDeveloperKey(t *testing.T) {
++ exchange := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++ w.Header().Set("Content-Type", "application/json")
++ w.WriteHeader(http.StatusUnauthorized)
++ io.WriteString(w, `{"detail":"the developer key was refused"}`)
++ }))
++ defer exchange.Close()
++ t.Setenv("OV_TEST_EXCHANGE_TOKEN", "wrong-token")
++ err := runExchangeServe(context.Background(), exchangeServeOptions{
++ Model: "qwen3:32b", Upstream: "http://127.0.0.1:11434/v1", Listen: "127.0.0.1:0",
++ PublicURL: "https://seller.example/v1", Concurrency: 2, HoursStart: "00:00", HoursEnd: "23:59", TZ: "UTC",
++ Audience: "everyone", TokenEnv: "OV_TEST_EXCHANGE_TOKEN", ExchangeURL: exchange.URL,
++ }, io.Discard)
++ if err == nil || !strings.Contains(err.Error(), "HTTP 401") || !strings.Contains(err.Error(), "developer key was refused") {
++ t.Fatalf("a refused key: %v", err)
++ }
++ if strings.Contains(err.Error(), "wrong-token") {
++ t.Fatal("the token is in the error")
++ }
++}
++
++func TestExchangeUsage(t *testing.T) {
++ var out, errb bytes.Buffer
++ if code := run([]string{"exchange"}, &out, &errb); code != 2 || !strings.Contains(errb.String(), "exchange serve --model") {
++ t.Fatalf("usage: %d %q", code, errb.String())
++ }
++ if code := run([]string{"exchange", "serve"}, &out, &errb); code != 1 || !strings.Contains(errb.String(), "--model") {
++ t.Fatalf("no model: %d %q", code, errb.String())
++ }
++}
ddiff --git a/cmd/openvibe-node/main.go b/cmd/openvibe-node/main.go
iindex f8cdcc1..5b68d88 100644
---- a/cmd/openvibe-node/main.go
++++ b/cmd/openvibe-node/main.go
@@@ -84,6 +84,8 @@ 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 exchange serve --model <id> [--public-url URL]
++ serve a local model as an OpenVibe.Exchange seller offer
openvibe-node stream --probe --whip-url URL [--seconds N] [--size WxH] [--json]
publish a QR of the millisecond clock for the headless latency viewer (the ticket comes from
OPENVIBE_STREAM_TICKET, never argv)
@@@ -148,6 +150,9 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
// The memory scan has its own flags, so it parses its own arguments.
return cmdMemory(rest, stdin, stdout, stderr)
}
++ if cmd == "exchange" { // its own flags: serve, and the seller offer's
++ return cmdExchange(rest, stdout, stderr)
++ }
if cmd == "desk-sandbox" { // its own flags, up to "--": a browser's flags after it are never the Node's
return cmdDeskSandbox(rest, stderr)
}