File size: 5,419 Bytes
857a91b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
// Command matrix-runtime is the execution plane for Matrix Cloud. It serves an
// HTTP API for running MCP sandboxes, inspecting models and (in future)
// agents/tools, and can join a MatrixHub Cloud control plane.
package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"log"
	"net"
	"os"
	"os/signal"
	"strconv"
	"syscall"

	"github.com/agent-matrix/matrix-runtime/api"
	"github.com/agent-matrix/matrix-runtime/internal/config"
	"github.com/agent-matrix/matrix-runtime/internal/controlplane"
	"github.com/agent-matrix/matrix-runtime/internal/jobs"
	"github.com/agent-matrix/matrix-runtime/internal/store"
)

func main() {
	if len(os.Args) > 1 && os.Args[1] == "join" {
		if err := runJoin(os.Args[2:]); err != nil {
			log.Fatalf("join: %v", err)
		}
		return
	}
	if err := runServe(os.Args[1:]); err != nil {
		log.Fatalf("matrix-runtime: %v", err)
	}
}

func runServe(args []string) error {
	fs := flag.NewFlagSet("matrix-runtime", flag.ExitOnError)
	mode := fs.String("mode", "", "runtime mode: cloud-worker|customer-agent|hf-space|local-dev")
	port := fs.Int("port", 0, "HTTP port (overrides MATRIX_RUNTIME_PORT)")
	if err := fs.Parse(args); err != nil {
		return err
	}

	cfg := config.FromEnv(*mode)
	if *port != 0 {
		cfg.Port = *port
	}
	if err := cfg.Validate(); err != nil {
		return err
	}

	// Loud, actionable warnings for unsafe production configuration. These do not
	// abort startup (the readiness probe also reports them) but they should be
	// impossible to miss in logs.
	if cfg.IsProduction() {
		if cfg.APIToken == "" {
			log.Printf("WARNING: MATRIX_RUNTIME_API_TOKEN is not set in %s mode — the API is unauthenticated for non-session callers", cfg.Mode)
		}
		if cfg.DatabaseURL == "" {
			log.Printf("WARNING: using SQLite in %s mode — set MATRIX_RUNTIME_DATABASE_URL (Postgres) for multi-user/HA deployments", cfg.Mode)
		}
	}
	if cfg.MatrixShellEnabled {
		log.Printf("WARNING: MatrixShell is ENABLED — it executes commands in a local sandbox (disable with MATRIX_SHELL_ENABLED=false)")
	}

	mgr := jobs.NewManager(cfg)
	defer mgr.Shutdown()

	if err := mgr.Layout().EnsureDirs(); err != nil {
		log.Printf("warning: could not create data dirs under %s: %v", cfg.DataDir, err)
	}

	// Open the multitenant user store. When a PostgreSQL/Neon URL is configured
	// (hosted control plane, e.g. cloud.matrixhub.io) use it and isolate all
	// objects in cfg.DBSchema; otherwise fall back to a local SQLite file. A
	// failure here is non-fatal: the console loads but auth endpoints return 503.
	var (
		st     *store.Store
		err    error
		dbDesc string
	)
	if cfg.DatabaseURL != "" {
		st, err = store.OpenPostgres(cfg.DatabaseURL, cfg.DBSchema, cfg.DataDir)
		dbDesc = fmt.Sprintf("postgres (schema %q)", cfg.DBSchema)
	} else {
		st, err = store.Open(cfg.DBPath)
		dbDesc = cfg.DBPath
	}
	if err != nil {
		log.Printf("warning: could not open user store (%s): %v", dbDesc, err)
	} else {
		defer func() { _ = st.Close() }()
		mgr.SetInstallStore(st)
		log.Printf("user store ready: %s", dbDesc)
	}

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	// Background retention: purge old terminal jobs, scratch dirs and logs.
	mgr.StartJanitor(ctx)

	// Bind the configured port; if it's busy, walk forward to the next free one.
	ln, boundPort, err := listenWithFallback(cfg.Port, 50)
	if err != nil {
		return err
	}
	if boundPort != cfg.Port {
		log.Printf("port %d is in use — using %d instead", cfg.Port, boundPort)
	}
	cfg.Port = boundPort

	log.Printf("matrix-runtime %s (commit %s, built %s) starting: mode=%s runtime_id=%s addr=:%d data_dir=%s",
		config.Version, config.Commit, config.Date, cfg.Mode, cfg.EffectiveRuntimeID(), boundPort, cfg.DataDir)
	log.Printf("console + API ready: http://localhost:%d", boundPort)

	srv := api.NewServer(cfg, mgr, st)
	if err := srv.Serve(ctx, ln); err != nil && err.Error() != "http: Server closed" {
		return err
	}
	log.Printf("matrix-runtime stopped")
	return nil
}

// listenWithFallback binds the first free TCP port in [start, start+maxTries),
// returning the listener and the port it bound. A port already in use is
// skipped; other errors (e.g. permission) are returned.
func listenWithFallback(start, maxTries int) (net.Listener, int, error) {
	var lastErr error
	for p := start; p < start+maxTries && p <= 65535; p++ {
		ln, err := net.Listen("tcp", ":"+strconv.Itoa(p))
		if err == nil {
			return ln, p, nil
		}
		lastErr = err
		if errors.Is(err, syscall.EADDRINUSE) {
			continue
		}
		// Non "in use" error on the very first try is fatal; otherwise keep trying.
		if p == start {
			return nil, 0, err
		}
	}
	return nil, 0, fmt.Errorf("no free port in range %d-%d: %w", start, start+maxTries-1, lastErr)
}

func runJoin(args []string) error {
	fs := flag.NewFlagSet("join", flag.ExitOnError)
	cloudURL := fs.String("cloud-url", "", "MatrixHub Cloud URL")
	token := fs.String("token", "", "runtime join token")
	runtimeID := fs.String("runtime-id", "", "optional runtime id")
	workspace := fs.String("workspace", "", "optional workspace")
	if err := fs.Parse(args); err != nil {
		return err
	}
	path, err := controlplane.WriteJoinConfig(controlplane.JoinConfig{
		CloudURL:  *cloudURL,
		JoinToken: *token,
		RuntimeID: *runtimeID,
		Workspace: *workspace,
	})
	if err != nil {
		return err
	}
	fmt.Printf("Wrote join configuration to %s\n", path)
	return nil
}