1// Package app wires vev's runtime together via dependency injection. It is
2// the single place where usecases (client, daemon) are connected to concrete
3// adapters (ipc transport/listener, terminal); usecases themselves never
4// import adapters.
5package app
6
7import (
8 "bufio"
9 "context"
10 "crypto/rand"
11 "encoding/base64"
12 "errors"
13 "fmt"
14 "io"
15 "log/slog"
16 "net"
17 "net/http"
18 _ "net/http/pprof"
19 "os"
20 "os/exec"
21 "os/signal"
22 "path/filepath"
23 "strconv"
24 "strings"
25 "sync"
26 "syscall"
27 "text/tabwriter"
28 "time"
29
30 "github.com/bnema/vev/internal/adapters/clipboard"
31 "github.com/bnema/vev/internal/adapters/clock"
32 "github.com/bnema/vev/internal/adapters/config"
33 "github.com/bnema/vev/internal/adapters/dgram"
34 "github.com/bnema/vev/internal/adapters/ipc"
35 "github.com/bnema/vev/internal/adapters/lifecycle"
36 "github.com/bnema/vev/internal/adapters/noticefile"
37 "github.com/bnema/vev/internal/adapters/observability"
38 "github.com/bnema/vev/internal/adapters/pty"
39 remoteadapter "github.com/bnema/vev/internal/adapters/remote"
40 "github.com/bnema/vev/internal/adapters/shellcmd"
41 snapshotadapter "github.com/bnema/vev/internal/adapters/snapshot"
42 "github.com/bnema/vev/internal/adapters/sshstdio"
43 "github.com/bnema/vev/internal/adapters/term"
44 "github.com/bnema/vev/internal/domain"
45 "github.com/bnema/vev/internal/logging"
46 "github.com/bnema/vev/internal/persist"
47 "github.com/bnema/vev/internal/platform"
48 "github.com/bnema/vev/internal/ports"
49 "github.com/bnema/vev/internal/usecase/client"
50 "github.com/bnema/vev/internal/usecase/daemon"
51 "github.com/bnema/vev/internal/usecase/recovery"
52 pdgram "github.com/bnema/vev/pkg/dgram"
53 "github.com/bnema/vev/pkg/safedir"
54)
55
56// cmdKind identifies which sub-command the CLI parsed.
57type cmdKind int
58
59const (
60 kindAttach cmdKind = iota // ephemeral/new/attach — distinguished by intent
61 kindList
62 kindHost
63 kindKill
64 kindCmd
65 kindDaemon
66 kindDaemonLauncher
67 kindStdio
68 kindUDPBootstrap
69 kindUDPProxy
70 kindRemotePreview
71 kindHelp
72 kindVersion
73)
74
75// command is the parsed CLI invocation: what to do, plus the attach intent
76// and session name where relevant.
77type command struct {
78 kind cmdKind
79 intent uint8
80 name string
81 remoteTarget string
82 listHost string
83 listAll bool
84 hostAction string
85 hostTarget string
86 killAll bool
87 killDaemon bool
88 cmd cmdInvocation
89 remotePreviewPayload string
90}
91
92// usageError is a user-facing argument error; the app prints it (with usage)
93// rather than a stack of wrapped internals.
94type usageError struct{ msg string }
95
96func (e *usageError) Error() string { return "vev: " + e.msg }
97
98func usagef(format string, args ...any) error {
99 return &usageError{msg: fmt.Sprintf(format, args...)}
100}
101
102const usageText = `vev — a terminal multiplexer
103
104usage:
105 vev attach to (or create) an ephemeral session
106 vev new <name> create and attach to a named session
107 vev attach <name> attach to an existing session (alias: a)
108 vev attach user@host[:session]
109 attach through SSH to a remote vev daemon
110 vev ls list local sessions
111 vev ls <host> list sessions on a known remote host
112 vev ls --all list local and remote sessions
113 vev host add <host> add a pinned remote host
114 vev host rm <host> remove a pinned remote host
115 vev host list list known remote hosts
116 vev kill <name> kill a session
117 vev kill --all kill all sessions and stop the daemon
118 vev kill --daemon stop the active vev daemon
119 vev cmd <command> run a control command (vev cmd --help)
120 vev --help show this help
121 vev --version show version`
122
123// Build metadata. Defaults describe a plain `go build`; releases overwrite
124// them via -ldflags "-X github.com/bnema/vev/internal/app.version=..." (and
125// .commit / .date) — see .goreleaser.yaml.
126var (
127 version = "0.1.0-dev"
128 commit = "none"
129 date = "unknown"
130
131 openCatalogue = persist.OpenOrCreate
132 listenDaemon = func(dir string, observer ports.SerializedRuntimeObserver) (ports.Listener, error) {
133 return ipc.Listen(dir, ipc.WithRuntimeObserver(observer))
134 }
135)
136
137// versionLine renders the --version output.
138func versionLine() string {
139 return fmt.Sprintf("vev %s (commit %s, built %s)", version, commit, date)
140}
141
142// Run is the entry point invoked by main. It parses args into a command and
143// dispatches it.
144func Run(args []string) error {
145 cmd, err := parseArgs(args)
146 if err != nil {
147 return err
148 }
149 return dispatch(context.Background(), cmd)
150}
151
152// parseArgs turns the raw argv tail into a command. It is deliberately a
153// pure function so the full dispatch table can be unit-tested without any
154// I/O.
155func parseArgs(args []string) (command, error) {
156 if len(args) == 0 {
157 return command{kind: kindAttach, intent: ports.IntentEphemeral}, nil
158 }
159
160 switch args[0] {
161 case "--daemon":
162 return command{kind: kindDaemon}, nil
163 case "--daemon-launcher":
164 return command{kind: kindDaemonLauncher}, nil
165 case "_stdio":
166 if len(args) != 1 {
167 return command{}, usagef("`_stdio` does not accept a session name")
168 }
169 return command{kind: kindStdio}, nil
170 case "_udp-bootstrap":
171 if len(args) != 1 {
172 return command{}, usagef("`_udp-bootstrap` does not accept a session name")
173 }
174 return command{kind: kindUDPBootstrap}, nil
175 case "_udp-proxy":
176 if len(args) != 1 {
177 return command{}, usagef("`_udp-proxy` does not accept a session name")
178 }
179 return command{kind: kindUDPProxy}, nil
180 case "_remote-preview":
181 if len(args) != 2 || args[1] == "" {
182 return command{}, usagef("`_remote-preview` requires one encoded request")
183 }
184 return command{kind: kindRemotePreview, remotePreviewPayload: args[1]}, nil
185 case "new":
186 if len(args) < 2 || args[1] == "" {
187 return command{}, usagef("`new` requires a session name")
188 }
189 if len(args) > 2 {
190 return command{}, usagef("`new` does not support command overrides")
191 }
192 if err := domain.ValidateSessionName(args[1]); err != nil {
193 return command{}, err
194 }
195 return command{kind: kindAttach, intent: ports.IntentNew, name: args[1]}, nil
196 case "attach", "a":
197 if len(args) < 2 || args[1] == "" {
198 return command{}, usagef("`attach` requires a session name")
199 }
200 if len(args) > 2 {
201 return command{}, usagef("`attach` accepts exactly one session name or remote target")
202 }
203 cmd := command{kind: kindAttach, intent: ports.IntentAttach, name: args[1]}
204 if target, session, ok := parseRemoteAttachTarget(args[1]); ok {
205 if err := domain.ValidateRemoteHostTarget(target); err != nil {
206 return command{}, err
207 }
208 if session != "" {
209 if err := domain.ValidateSessionName(session); err != nil {
210 return command{}, err
211 }
212 }
213 cmd.remoteTarget = target
214 cmd.name = session
215 if session == "" {
216 cmd.intent = ports.IntentEphemeral
217 }
218 }
219 return cmd, nil
220 case "ls", "list":
221 return parseListArgs(args[1:])
222 case "host":
223 return parseHostArgs(args[1:])
224 case "cmd":
225 invocation, err := parseCmdArgs(args[1:])
226 if err != nil {
227 return command{}, err
228 }
229 return command{kind: kindCmd, cmd: invocation}, nil
230 case "kill":
231 if len(args) < 2 || args[1] == "" {
232 return command{}, usagef("`kill` requires a session name, --all, or --daemon")
233 }
234 if args[1] == "--" {
235 if len(args) != 3 || args[2] == "" {
236 return command{}, usagef("`kill --` requires a session name")
237 }
238 return command{kind: kindKill, name: args[2]}, nil
239 }
240 if len(args) > 2 {
241 return command{}, usagef("`kill` accepts exactly one session name, --all, or --daemon")
242 }
243 switch args[1] {
244 case "--all":
245 return command{kind: kindKill, killAll: true}, nil
246 case "--daemon":
247 return command{kind: kindKill, killDaemon: true}, nil
248 default:
249 return command{kind: kindKill, name: args[1]}, nil
250 }
251 case "-h", "--help", "help":
252 return command{kind: kindHelp}, nil
253 case "--version", "version":
254 return command{kind: kindVersion}, nil
255 default:
256 return command{}, usagef("unknown command %q", args[0])
257 }
258}
259
260// dispatch runs the parsed command.
261func dispatch(ctx context.Context, cmd command) error {
262 switch cmd.kind {
263 case kindHelp:
264 fmt.Println(usageText)
265 return nil
266 case kindVersion:
267 fmt.Println(versionLine())
268 return nil
269 case kindDaemon:
270 return runDaemon()
271 case kindDaemonLauncher:
272 return runDaemonLauncher()
273 case kindStdio:
274 return runStdio(ctx)
275 case kindUDPBootstrap:
276 return runUDPBootstrap(ctx)
277 case kindUDPProxy:
278 return runUDPProxy(ctx, os.Stdout)
279 case kindRemotePreview:
280 return runRemotePreview(ctx, cmd.remotePreviewPayload)
281 case kindList:
282 return runList(ctx, cmd)
283 case kindHost:
284 return runHostCommand(ctx, cmd, defaultRemoteHostDeps())
285 case kindKill:
286 return runKill(ctx, cmd.name, cmd.killAll, cmd.killDaemon)
287 case kindCmd:
288 return runCmd(ctx, cmd.cmd)
289 case kindAttach:
290 return runAttach(ctx, cmd.intent, cmd.name, cmd.remoteTarget)
291 default:
292 return usagef("unhandled command")
293 }
294}
295
296func parseListArgs(args []string) (command, error) {
297 if len(args) == 0 {
298 return command{kind: kindList}, nil
299 }
300 if len(args) > 1 {
301 return command{}, usagef("`ls` accepts at most one host or --all")
302 }
303 switch args[0] {
304 case "--all":
305 return command{kind: kindList, listAll: true}, nil
306 default:
307 if strings.HasPrefix(args[0], "-") {
308 return command{}, usagef("unknown flag %q for `ls`", args[0])
309 }
310 if err := domain.ValidateRemoteHostTarget(args[0]); err != nil {
311 return command{}, err
312 }
313 return command{kind: kindList, listHost: args[0]}, nil
314 }
315}
316
317func parseHostArgs(args []string) (command, error) {
318 if len(args) == 0 {
319 return command{}, usagef("`host` requires add, rm, or list")
320 }
321 switch args[0] {
322 case hostActionAdd, hostActionRm:
323 if len(args) < 2 || args[1] == "" {
324 return command{}, usagef("`host %s` requires a host target", args[0])
325 }
326 if len(args) > 2 {
327 return command{}, usagef("`host %s` accepts exactly one host target", args[0])
328 }
329 if err := domain.ValidateRemoteHostTarget(args[1]); err != nil {
330 return command{}, err
331 }
332 return command{kind: kindHost, hostAction: args[0], hostTarget: args[1]}, nil
333 case hostActionList:
334 if len(args) > 1 {
335 return command{}, usagef("`host list` accepts no arguments")
336 }
337 return command{kind: kindHost, hostAction: hostActionList}, nil
338 default:
339 return command{}, usagef("unknown host action %q", args[0])
340 }
341}
342
343// performanceTrace creates one serialized timestamp owner for this process.
344// An empty trace environment leaves all production behavior and wire bytes
345// unchanged.
346// Composition-root factory seams keep observer propagation testable without
347// opening real transports.
348var (
349 newPerformanceTrace = performanceTrace
350 newRemoteHostStore = remoteadapter.NewFileHostStore
351 newRemoteCatalogCache = remoteadapter.NewFileCatalogCache
352 newRemoteCatalogClient = remoteadapter.NewCatalogClient
353 newRemotePreviewClient = remoteadapter.NewPreviewClient
354 newRemoteDialerFactoryWithRuntimeObserver = func(observer ports.SerializedRuntimeObserver) ports.RemoteDialerFactory {
355 return remoteadapter.NewDialerFactoryWithRuntimeObserver(observer)
356 }
357 runClientWithDeps runClientFunc = func(
358 ctx context.Context,
359 deps client.Dependencies,
360 request client.AttachRequest,
361 ) error {
362 return client.NewRunner(deps).Run(ctx, request)
363 }
364)
365
366func performanceTrace(clk ports.Clock) (ports.SerializedRuntimeObserver, io.Closer, error) {
367 return performanceTraceWithFactories(clk, observability.NewJSONL, ports.NewRuntimeCorrelationObserver)
368}
369
370// performanceTraceWithFactories keeps setup rollback behavior directly
371// testable without requiring an operating-system file close failure.
372func performanceTraceWithFactories(
373 clk ports.Clock,
374 newSink func(string, ports.Clock, string) (ports.RuntimeObserver, io.Closer, error),
375 newCorrelation func(ports.RuntimeObserver, ports.RuntimeCorrelationInputs) (ports.RuntimeObserver, error),
376) (ports.SerializedRuntimeObserver, io.Closer, error) {
377 path, processID := os.Getenv("VEV_PERF_TRACE"), os.Getenv("VEV_PERF_PROCESS_ID")
378 if path == "" {
379 return nil, nil, nil
380 }
381 observer, closer, err := newSink(path, clk, processID)
382 if err != nil {
383 return nil, nil, err
384 }
385 // The harness supplies these manifest fields to every launched role. Keep a
386 // valid standalone trace for operators that set only the original trace
387 // variables, while ensuring harness traces match their process mapping.
388 inputs := ports.RuntimeCorrelationInputs{Scenario: os.Getenv("VEV_PERF_SCENARIO"), Run: 1}
389 if inputs.Scenario == "" {
390 inputs.Scenario = "runtime"
391 }
392 if rawRun := os.Getenv("VEV_PERF_RUN"); rawRun != "" {
393 inputs.Run, err = strconv.ParseUint(rawRun, 10, 64)
394 if err != nil {
395 setupErr := fmt.Errorf("invalid VEV_PERF_RUN %q: %w", rawRun, err)
396 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
397 }
398 if inputs.Run == 0 {
399 setupErr := fmt.Errorf("invalid VEV_PERF_RUN %q", rawRun)
400 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
401 }
402 }
403 observer, err = newCorrelation(observer, inputs)
404 if err != nil {
405 setupErr := fmt.Errorf("configure runtime trace correlation: %w", err)
406 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
407 }
408 reporter := ports.NewSerializedRuntimeObserver(observer, runtimeTraceQueueDepth)
409 return reporter, &runtimeTraceCloser{reporter: reporter, closer: closer}, nil
410}
411
412func closeTraceAfterSetupFailure(closer io.Closer, setupErr error) error {
413 if closer == nil {
414 return setupErr
415 }
416 if closeErr := closer.Close(); closeErr != nil {
417 return errors.Join(setupErr, fmt.Errorf("close performance trace after setup failure: %w", closeErr))
418 }
419 return setupErr
420}
421
422// runtimeTraceQueueDepth bounds all process-local trace producer handoffs.
423// A full queue emits the serialized diagnostic gap rather than delaying a
424// terminal, transport, or ACK progress path.
425const runtimeTraceQueueDepth = 256
426
427// runtimeTraceCloser owns the two-stage trace shutdown: drain the reporter
428// before closing the concrete timestamp/file owner. sync.Once makes every
429// deferred and explicit close path safe without closing shared workers twice.
430type runtimeTraceCloser struct {
431 reporter ports.SerializedRuntimeObserver
432 closer io.Closer
433 once sync.Once
434 err error
435}
436
437func (c *runtimeTraceCloser) Close() error {
438 if c == nil {
439 return nil
440 }
441 c.once.Do(func() {
442 if c.reporter != nil {
443 c.reporter.Flush()
444 c.reporter.Close()
445 }
446 if c.closer != nil {
447 c.err = c.closer.Close()
448 }
449 })
450 return c.err
451}
452
453func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {
454 return logging.Setup(logging.Config{
455 Dir: platform.StateDir(),
456 Component: component,
457 Level: logging.EnvLevel(),
458 MaxBytes: logging.DefaultMaxBytes,
459 RotateAtRuntime: rotateAtRuntime,
460 })
461}
462
463func logConfigWarnings(log *slog.Logger, warnings []domain.Warning) {
464 for _, warning := range warnings {
465 if warning.Line > 0 {
466 log.Warn("config warning", "line", warning.Line, "msg", warning.Msg)
467 continue
468 }
469 log.Warn("config warning", "msg", warning.Msg)
470 }
471}
472
473func snapshotDir() string {
474 return filepath.Join(platform.StateDir(), "snapshots")
475}
476
477func pprofAddrIsLoopback(addr string) bool {
478 host, _, err := net.SplitHostPort(addr)
479 if err != nil || host == "" {
480 return false
481 }
482 if strings.EqualFold(host, "localhost") {
483 return true
484 }
485 ip := net.ParseIP(host)
486 return ip != nil && ip.IsLoopback()
487}
488
489type ownedDaemonStart func(context.Context) error
490
491type lifecycleOwnership interface {
492 Release() error
493}
494
495func joinLifecycleReleaseError(retErr *error, owner lifecycleOwnership) {
496 *retErr = errors.Join(*retErr, owner.Release())
497}
498
499type lifecycleStartupDeps struct {
500 ensurePrivate func(string) error
501 acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error)
502 log *slog.Logger
503}
504
505const lifecycleAcquireRetry = 25 * time.Millisecond
506
507func lifecycleStartupDependencies(log *slog.Logger) lifecycleStartupDeps {
508 return lifecycleStartupDeps{
509 ensurePrivate: safedir.EnsurePrivate,
510 acquire: func(ctx context.Context, runtimeDir string, retry time.Duration) (lifecycleOwnership, error) {
511 return lifecycle.Acquire(ctx, runtimeDir, retry)
512 },
513 log: log,
514 }
515}
516
517func runWithLifecycleOwner(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart) error {
518 return runWithLifecycleOwnerDeps(ctx, runtimeDir, stateDir, start, lifecycleStartupDependencies(nil))
519}
520
521func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) {
522 log := deps.log
523 if log == nil {
524 log = slog.Default()
525 }
526 if err := deps.ensurePrivate(runtimeDir); err != nil {
527 return fmt.Errorf("vev: secure runtime directory: %w", err)
528 }
529 if err := deps.ensurePrivate(stateDir); err != nil {
530 return fmt.Errorf("vev: secure state directory: %w", err)
531 }
532 lockPath := lifecycle.Path(runtimeDir)
533 log.Info("lifecycle_owner_wait", "path", lockPath)
534 owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry)
535 if err != nil {
536 log.Warn("lifecycle_owner_wait_failed", "path", lockPath, "reason_code", "acquire-failed")
537 return fmt.Errorf("vev: acquire lifecycle ownership: %w", err)
538 }
539 log.Info("lifecycle_owner_acquired", "path", lockPath)
540 defer func() {
541 releaseErr := owner.Release()
542 if releaseErr != nil {
543 log.Error("lifecycle_owner_release_failed", "path", lockPath, "reason_code", "release-failed")
544 } else {
545 log.Info("lifecycle_owner_released", "path", lockPath)
546 }
547 retErr = errors.Join(retErr, releaseErr)
548 }()
549 if start == nil {
550 return errors.New("vev: nil owned daemon startup")
551 }
552 return start(ctx)
553}
554
555// runDaemon runs the daemon in the foreground (the hidden --daemon path,
556// entered by an auto-spawned child) while holding exclusive lifecycle
557// ownership through complete teardown.
558func runDaemon() (retErr error) {
559 ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
560 defer stop()
561 log, logCloser, err := configureLogging(logging.Daemon, true)
562 if err != nil {
563 return err
564 }
565 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
566 return runWithLifecycleOwnerDeps(ctx, ipc.SocketDir(), platform.StateDir(), func(ctx context.Context) error {
567 return runDaemonOwnedWithLogger(ctx, log)
568 }, lifecycleStartupDependencies(log))
569}
570
571func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
572 log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
573}
574
575func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
576 healthy, fresh, broken := 0, 0, 0
577 for _, record := range records {
578 switch {
579 case record.DegradedReason != "":
580 broken++
581 case record.Committed == nil:
582 fresh++
583 default:
584 healthy++
585 }
586 }
587 log.Info("daemon_startup_complete", "healthy", healthy, "fresh", fresh, "restoring", restoring, "broken", broken)
588}
589
590func constructDaemonBeforeSocketPublication(
591 construct func() *daemon.Daemon,
592 prepare func(*daemon.Daemon) error,
593 listen func() (ports.Listener, error),
594) (*daemon.Daemon, ports.Listener, error) {
595 d := construct()
596 if err := prepare(d); err != nil {
597 return d, nil, err
598 }
599 ln, err := listen()
600 return d, ln, err
601}
602
603func runDaemonOwned(ctx context.Context) (retErr error) {
604 log, logCloser, err := configureLogging(logging.Daemon, true)
605 if err != nil {
606 return err
607 }
608 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
609 return runDaemonOwnedWithLogger(ctx, log)
610}
611
612func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr error) {
613 clk := clock.New()
614 observer, observerCloser, err := newPerformanceTrace(clk)
615 if err != nil {
616 return fmt.Errorf("vev: performance trace: %w", err)
617 }
618 if observerCloser != nil {
619 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
620 }
621 remoteDiscoveryOpt, err := remoteDiscoveryDaemonOption(platform.StateDir(), observer, os.Getenv(envRemoteTransport))
622 if err != nil {
623 return err
624 }
625 if addr := os.Getenv("VEV_PPROF_ADDR"); addr != "" {
626 if !pprofAddrIsLoopback(addr) {
627 log.Warn("pprof bound to non-loopback address; /debug/pprof is unauthenticated", "addr", addr)
628 }
629 go func() {
630 if err := http.ListenAndServe(addr, nil); err != nil && !errors.Is(err, http.ErrServerClosed) {
631 log.Error("pprof server exited", "err", err)
632 }
633 }()
634 log.Info("pprof enabled", "addr", addr)
635 }
636
637 daemonOpts := []daemon.Option{remoteDiscoveryOpt}
638 if observer != nil {
639 daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer))
640 }
641 configPath := platform.ConfigPath()
642 cfg, warnings, err := config.Load(configPath)
643 if err != nil {
644 log.Warn("loading config failed; using defaults", "path", configPath, "err", err)
645 cfg = domain.Defaults()
646 }
647 logConfigWarnings(log, warnings)
648 daemonOpts = append(daemonOpts, daemon.WithConfig(cfg))
649 daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New()))
650 daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome))
651 snapshotRepository := snapshotadapter.NewRepositoryWithLogger(snapshotDir(), log)
652 daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository))
653 noticeStore := noticefile.New(platform.StateDir())
654 daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
655 stateDir := platform.StateDir()
656 storePath := persist.StorePath(stateDir)
657 opened, err := openCatalogue(stateDir)
658 if err != nil {
659 log.Error("catalogue_validation_failed", "path", storePath, "reason_code", "open-failed")
660 if errors.Is(err, persist.ErrCatalogueUnreadable) {
661 return unreadableCatalogueError(stateDir)
662 }
663 return fmt.Errorf("vev: open durable session state %s: %w", storePath, err)
664 }
665 recoveryMode := "current"
666 if opened.NewInstall {
667 recoveryMode = "new-install"
668 }
669 logCatalogueRecovery(log, opened.Records, recoveryMode)
670 coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
671 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
672 daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository))
673 log.Info("session persistence enabled", "path", storePath)
674 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
675
676 // Construct the catalogue-backed expected-session registry before socket
677 // publication. Phase 3 snapshot restoration remains asynchronous in Serve.
678 d, ln, err := constructDaemonBeforeSocketPublication(
679 func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
680 func(d *daemon.Daemon) error {
681 if err := d.CollectStartupGarbage(ctx); err != nil {
682 // GC is best-effort, but it is fully finished before socket
683 // publication. A failed pass leaves durable state untouched and
684 // restoration retains its per-session failure isolation.
685 log.Warn("snapshot_garbage_collection_failed", "err", err)
686 }
687 return nil
688 },
689 func() (ports.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
690 )
691 if err != nil {
692 closeErr := opened.Catalogue.Close()
693 log.Error("daemon startup preparation failed", "socket_dir", ipc.SocketDir(), "err", err)
694 return errors.Join(fmt.Errorf("vev: prepare daemon startup: %w", err), closeErr)
695 }
696 defer func() { _ = ln.Close() }()
697 log.Info("daemon starting", "socket", ln.Addr())
698
699 watchCtx, stopWatch := context.WithCancel(ctx)
700 defer stopWatch()
701 go func() {
702 if err := config.Watch(watchCtx, clk, configPath, func(cfg domain.Config, warnings []domain.Warning) {
703 logConfigWarnings(log, warnings)
704 d.ApplyConfig(cfg)
705 }); err != nil && !errors.Is(err, context.Canceled) {
706 log.Warn("config watcher stopped", "path", configPath, "err", err)
707 }
708 }()
709 if err := d.Serve(ctx, ln); err != nil {
710 log.Error("daemon exited", "err", err)
711 return err
712 }
713 log.Info("daemon exited cleanly")
714 return nil
715}
716
717// runAttach dials (auto-spawning the daemon if needed) and runs the client
718// attach loop. Logging goes to the shared file: the client must never write
719// to the console while the terminal is raw.
720func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) (retErr error) {
721 // Treat a controlling-terminal hangup or termination request as a graceful
722 // detach rather than an abrupt process death. Cancelling the context unwinds
723 // the client's pumps, closes its transport, and lets the deferred trace closer
724 // flush the in-flight receive's end mark — so teardown never truncates a span.
725 // Raw mode disables ISIG, so catching SIGINT here does not affect interactive
726 // Ctrl+C, which the daemon delivers to the remote shell as a normal keystroke.
727 ctx, stop := signal.NotifyContext(ctx, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT)
728 defer stop()
729
730 log, logCloser, err := configureLogging(logging.Client, false)
731 if err != nil {
732 return err
733 }
734 defer func() { _ = logCloser.Close() }()
735
736 clk := clock.New()
737 observer, observerCloser, err := newPerformanceTrace(clk)
738 if err != nil {
739 return fmt.Errorf("vev: performance trace: %w", err)
740 }
741 if observerCloser != nil {
742 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
743 }
744 return runAttachWithDeps(ctx, intent, name, remoteTarget, os.Getenv("VEV"), log, runAttachDeps{
745 localDialer: func() ports.Dialer {
746 return localDaemonDialer{dir: ipc.SocketDir(), observer: observer}
747 },
748 remoteDialerFactory: newRemoteDialerFactoryWithRuntimeObserver(observer),
749 selectedRemoteTransport: os.Getenv(envRemoteTransport),
750 runClient: runClientWithDeps,
751 createDetached: createDetachedLocalSession,
752 clipboard: clipboard.New(),
753 runtimeObserver: observer,
754 stateDir: platform.StateDir,
755 })
756}
757
758const envRemoteTransport = "VEV_REMOTE_TRANSPORT"
759
760func defaultLocalDialer() ports.Dialer { return localDaemonDialer{dir: ipc.SocketDir()} }
761
762func defaultRemoteDialerFactory() ports.RemoteDialerFactory {
763 return remoteadapter.NewDialerFactory()
764}
765
766type runClientFunc func(context.Context, client.Dependencies, client.AttachRequest) error
767
768type runAttachDeps struct {
769 localDialer func() ports.Dialer
770 remoteDialerFactory ports.RemoteDialerFactory
771 selectedRemoteTransport string
772 runClient runClientFunc
773 createDetached func(context.Context, string) error
774 runtimeObserver ports.SerializedRuntimeObserver
775 // clipboard reads a clipboard image on a remote attach's Ctrl+V (see
776 // docs/superpowers/specs/2026-07-04-clipboard-image-transfer-design.md).
777 // Only used for the remote-dialer branch below; local attaches never
778 // intercept Ctrl+V regardless of this field.
779 clipboard ports.ClipboardReader
780 // Optional remote-learning seams.
781 stateDir func() string
782 hostStore ports.RemoteHostStore
783}
784
785func remoteTransportModeFromEnv(value string) (ports.RemoteTransportMode, error) {
786 switch value {
787 case "", string(ports.RemoteTransportUDP):
788 return ports.RemoteTransportUDP, nil
789 case string(ports.RemoteTransportStdio):
790 return ports.RemoteTransportStdio, nil
791 default:
792 return "", fmt.Errorf("vev: invalid remote transport %q (want %q or %q)", value, ports.RemoteTransportUDP, ports.RemoteTransportStdio)
793 }
794}
795
796func validateRemoteAttachHandoff(target ports.AttachTarget) error {
797 if err := ports.ValidateAttachTarget(target); err != nil {
798 return err
799 }
800 if err := domain.ValidateRemoteHostTarget(target.Endpoint); err != nil {
801 return err
802 }
803 if err := domain.ValidateSessionName(target.Session); err != nil {
804 return err
805 }
806 if target.RemoteTarget != nil {
807 if target.EnvironmentPolicy != ports.EnvironmentPolicyDaemonOwned {
808 return errors.New("remote picker handoff must use daemon-owned environment")
809 }
810 if target.RemoteTarget.Endpoint != target.Endpoint || target.RemoteTarget.SessionName != target.Session {
811 return errors.New("remote picker handoff identity does not match route")
812 }
813 }
814 return nil
815}
816
817// remoteDiscoveryDaemonOption constructs the daemon-owned discovery ports from
818// the same validated transport selection used by direct remote attach.
819func remoteDiscoveryDaemonOption(stateDir string, observer ports.SerializedRuntimeObserver, transport string) (daemon.Option, error) {
820 mode, err := remoteTransportModeFromEnv(transport)
821 if err != nil {
822 return nil, err
823 }
824 discovery := daemon.WithRemoteDiscovery(
825 newRemoteHostStore(remoteadapter.HostStorePath(stateDir)),
826 newRemoteCatalogClient(),
827 newRemoteCatalogCache(remoteadapter.CatalogCachePath(stateDir)),
828 newRemoteDialerFactoryWithRuntimeObserver(observer),
829 mode,
830 )
831 preview := daemon.WithRemotePreview(newRemotePreviewClient())
832 return func(d *daemon.Daemon) {
833 discovery(d)
834 preview(d)
835 }, nil
836}
837
838const maxAttachTargetHandoffs = 32
839
840func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, activeSession string, log *slog.Logger, deps runAttachDeps) error {
841 if activeSession != "" {
842 if remoteTarget == "" && intent == ports.IntentNew {
843 return deps.createDetached(ctx, name)
844 }
845 return errors.New("vev: sessions should be nested with care; unset VEV to force")
846 }
847
848 runClient := deps.runClient
849 if runClient == nil {
850 runClient = runClientWithDeps
851 }
852 localDialer := deps.localDialer
853 if localDialer == nil {
854 localDialer = defaultLocalDialer
855 }
856 mode, modeErr := remoteTransportModeFromEnv(deps.selectedRemoteTransport)
857 factory := deps.remoteDialerFactory
858 var remoteSelection *domain.RemoteSessionTarget
859 var remoteEnvironmentPolicy ports.EnvironmentPolicy
860 remoteDisplayOrigin := remoteTarget
861 routeOrigin := ports.RouteOriginLocal
862 routeOriginKey := "local"
863 if remoteTarget != "" {
864 routeOrigin = ports.RouteOriginRemote
865 routeOriginKey = remoteTarget
866 if modeErr != nil {
867 return modeErr
868 }
869 if factory == nil {
870 factory = defaultRemoteDialerFactory()
871 }
872 }
873 pickerHandoff := remoteTarget == ""
874 pickerEnvironmentPolicy := func(target *domain.RemoteSessionTarget, policy ports.EnvironmentPolicy) ports.EnvironmentPolicy {
875 if pickerHandoff && target == nil {
876 return ports.EnvironmentPolicyDaemonOwned
877 }
878 return policy
879 }
880 handoff := func(target ports.AttachTarget) (ports.Dialer, client.AttachRequest, error) {
881 if err := validateRemoteAttachHandoff(target); err != nil {
882 return nil, client.AttachRequest{}, fmt.Errorf("vev: invalid remote attach handoff: %w", err)
883 }
884 if modeErr != nil {
885 return nil, client.AttachRequest{}, modeErr
886 }
887 if factory == nil {
888 factory = defaultRemoteDialerFactory()
889 }
890 var selection *domain.RemoteSessionTarget
891 if target.RemoteTarget != nil {
892 copyTarget := *target.RemoteTarget
893 selection = ©Target
894 }
895 dialer, err := factory.DialerForRemote(target.Endpoint, target.Session, mode, log)
896 if err != nil {
897 return nil, client.AttachRequest{}, err
898 }
899 policy := pickerEnvironmentPolicy(selection, target.EnvironmentPolicy)
900 displayOrigin := target.Endpoint
901 if selection != nil {
902 displayOrigin = selection.DisplayOrigin
903 }
904 return dialer, client.AttachRequest{
905 Intent: target.Intent,
906 SessionName: target.Session,
907 Remote: true,
908 Origin: ports.RouteOriginDiscovery,
909 OriginKey: target.Endpoint,
910 RemoteTarget: selection,
911 RemoteOrigin: displayOrigin,
912 EnvironmentPolicy: policy,
913 }, nil
914 }
915
916 handoffAttempts := 0
917 for {
918 var err error
919 if remoteTarget != "" {
920 if log != nil {
921 log.Info("attaching to remote session", "target", remoteTarget, "name", name, "transport", string(mode))
922 }
923 dialer, dialErr := factory.DialerForRemote(remoteTarget, name, mode, log)
924 if dialErr != nil {
925 return dialErr
926 }
927 err = runClient(ctx, client.Dependencies{
928 Dialer: dialer,
929 Terminal: term.New(),
930 Clock: clock.New(),
931 Clipboard: deps.clipboard,
932 Logger: log,
933 RuntimeObserver: deps.runtimeObserver,
934 RemoteHostLearner: attachRememberLearner(deps, remoteTarget, log),
935 AttachHandoff: handoff,
936 Remote: true,
937 Origin: ports.RouteOriginRemote,
938 OriginKey: remoteTarget,
939 }, client.AttachRequest{
940 Intent: intent,
941 SessionName: name,
942 Remote: true,
943 Origin: routeOrigin,
944 OriginKey: routeOriginKey,
945 RemoteTarget: remoteSelection,
946 RemoteOrigin: remoteDisplayOrigin,
947 EnvironmentPolicy: remoteEnvironmentPolicy,
948 })
949 } else {
950 if log != nil {
951 log.Info("attaching to local session", "intent", intent, "name", name)
952 }
953 err = runClient(ctx, client.Dependencies{
954 Dialer: localDialer(),
955 Terminal: term.New(),
956 Clock: clock.New(),
957 Logger: log,
958 RuntimeObserver: deps.runtimeObserver,
959 AttachHandoff: handoff,
960 Remote: false,
961 Origin: ports.RouteOriginLocal,
962 OriginKey: "local",
963 }, client.AttachRequest{Intent: intent, SessionName: name, Origin: ports.RouteOriginLocal, OriginKey: "local"})
964 }
965
966 var handoffErr *client.AttachTargetError
967 if !errors.As(err, &handoffErr) {
968 return err
969 }
970 if handoffErr == nil {
971 return fmt.Errorf("vev: invalid remote attach handoff")
972 }
973 if err := validateRemoteAttachHandoff(handoffErr.Target); err != nil {
974 return fmt.Errorf("vev: invalid remote attach handoff: %w", err)
975 }
976 if handoffAttempts >= maxAttachTargetHandoffs {
977 return fmt.Errorf("vev: attach handoff exceeded maximum of %d attempts", maxAttachTargetHandoffs)
978 }
979 handoffAttempts++
980 remoteTarget = handoffErr.Target.Endpoint
981 remoteDisplayOrigin = remoteTarget
982 routeOrigin = ports.RouteOriginDiscovery
983 routeOriginKey = remoteTarget
984 name = handoffErr.Target.Session
985 intent = handoffErr.Target.Intent
986 remoteSelection = nil
987 if handoffErr.Target.RemoteTarget != nil {
988 copyTarget := *handoffErr.Target.RemoteTarget
989 remoteSelection = ©Target
990 remoteDisplayOrigin = copyTarget.DisplayOrigin
991 }
992 remoteEnvironmentPolicy = pickerEnvironmentPolicy(remoteSelection, handoffErr.Target.EnvironmentPolicy)
993 if modeErr != nil {
994 return modeErr
995 }
996 if factory == nil {
997 factory = defaultRemoteDialerFactory()
998 }
999 }
1000}
1001
1002const daemonStopTimeout = 2 * time.Second
1003
1004type localDaemonDialer struct {
1005 dir string
1006 observer ports.SerializedRuntimeObserver
1007}
1008
1009func (d localDaemonDialer) Dial(ctx context.Context) (ports.Transport, error) {
1010 dial := realDial
1011 if d.observer != nil {
1012 dial = func(ctx context.Context, dir string) (ports.Transport, error) {
1013 return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(d.observer))
1014 }
1015 }
1016 return ensureDaemonWithLifecycle(ctx, d.dir, dial, realSpawn, defaultBackoff)
1017}
1018
1019func detachedLocalHello(name, cwd string) ports.Hello {
1020 termEnv := os.Getenv("TERM")
1021 return ports.Hello{
1022 Version: ports.ProtocolVersion,
1023 Intent: ports.IntentNew,
1024 Name: name,
1025 Size: domain.Size{Cols: 80, Rows: 24},
1026 TermEnv: termEnv,
1027 Cwd: cwd,
1028 TrueColor: client.DetectTrueColor(termEnv, os.Getenv("COLORTERM")),
1029 Env: os.Environ(),
1030 }
1031}
1032
1033func createDetachedLocalSession(ctx context.Context, name string) error {
1034 transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
1035 if err != nil {
1036 return err
1037 }
1038 defer func() { _ = transport.Close() }()
1039
1040 cwd, err := os.Getwd()
1041 if err != nil {
1042 cwd = ""
1043 }
1044 hello := detachedLocalHello(name, cwd)
1045 if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
1046 return fmt.Errorf("vev: creating detached session: %w", err)
1047 }
1048
1049 type detachedReply struct {
1050 frame ports.Frame
1051 err error
1052 }
1053 replyCh := make(chan detachedReply, 1)
1054 go func() {
1055 f, err := transport.Recv()
1056 replyCh <- detachedReply{frame: f, err: err}
1057 }()
1058
1059 select {
1060 case <-ctx.Done():
1061 _ = transport.Close()
1062 return ctx.Err()
1063 case reply := <-replyCh:
1064 if reply.err != nil {
1065 return fmt.Errorf("vev: awaiting detached session creation: %w", reply.err)
1066 }
1067 switch reply.frame.Type {
1068 case ports.MsgWelcome:
1069 return nil
1070 case ports.MsgError:
1071 em, derr := ports.UnmarshalErrorMsg(reply.frame.Payload)
1072 if derr != nil {
1073 return fmt.Errorf("vev: decoding error reply: %w", derr)
1074 }
1075 return &client.ProtocolError{Code: em.Code, Text: em.Text}
1076 default:
1077 return fmt.Errorf("vev: unexpected reply type %d to detached session creation", reply.frame.Type)
1078 }
1079 }
1080}
1081
1082func requestDaemonStop(ctx context.Context) error {
1083 ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)
1084 defer cancel()
1085
1086 transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1087 if err != nil {
1088 return fmt.Errorf("vev: stopping daemon: %w", err)
1089 }
1090 if owner != nil {
1091 return errors.Join(errors.New("vev: no daemon running"), owner.Release())
1092 }
1093 defer func() { _ = transport.Close() }()
1094 if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}); err != nil {
1095 return fmt.Errorf("vev: requesting daemon stop: %w", err)
1096 }
1097 if _, err := transport.Recv(); err != nil && !errors.Is(err, io.EOF) {
1098 return fmt.Errorf("vev: reading daemon stop reply: %w", err)
1099 }
1100
1101 owner, err = waitForLifecycleAvailability(ctx, ipc.SocketDir(), defaultBackoff)
1102 if err != nil {
1103 return fmt.Errorf("vev: waiting for daemon ownership transfer: %w", err)
1104 }
1105 return owner.Release()
1106}
1107
1108// runStdio is the hidden remote-side mode used by `ssh host vev _stdio`: it
1109// connects to the per-user daemon (auto-spawning it if needed) and proxies the
1110// framed protocol between process stdio and the daemon socket.
1111func runStdio(ctx context.Context) (retErr error) {
1112 log, logCloser, err := configureLogging(logging.Stdio, false)
1113 if err != nil {
1114 return err
1115 }
1116 defer func() { _ = logCloser.Close() }()
1117 log.Debug("stdio proxy starting")
1118
1119 clk := clock.New()
1120 observer, observerCloser, err := newPerformanceTrace(clk)
1121 if err != nil {
1122 return fmt.Errorf("vev: performance trace: %w", err)
1123 }
1124 if observerCloser != nil {
1125 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
1126 }
1127 transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
1128 return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
1129 }, realSpawn, defaultBackoff)
1130 if err != nil {
1131 return err
1132 }
1133 defer func() { _ = transport.Close() }()
1134
1135 // sshstdio.Close owns and closes writers that implement io.Closer so a
1136 // blocked Send can unwind. Keep process-global stdin/stdout outside that
1137 // ownership boundary: sshstdio recognizes process stdin as a shared
1138 // cancellable pump, while output is an owned pipe drained by a forwarding
1139 // goroutine.
1140 stdin, stdout := os.Stdin, os.Stdout
1141 stdoutReader, stdoutWriter := io.Pipe()
1142 forwardDone := make(chan struct{})
1143 go func() {
1144 defer close(forwardDone)
1145 if _, err := io.Copy(stdout, stdoutReader); err != nil {
1146 _ = stdoutReader.CloseWithError(err)
1147 }
1148 }()
1149 defer func() {
1150 _ = stdoutWriter.Close()
1151 <-forwardDone
1152 _ = stdoutReader.Close()
1153 }()
1154 stdio := sshstdio.NewTransport(stdin, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer))
1155 return runStdioProxy(ctx, stdio, transport, log)
1156}
1157
1158var (
1159 udpBootstrapTimeout = 5 * time.Second
1160 udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd {
1161 return exec.CommandContext(ctx, exe, args...)
1162 }
1163)
1164
1165// runUDPBootstrap starts a detached _udp-proxy, forwards its single readiness
1166// line, and exits so SSH stdio can close without owning the UDP proxy lifetime.
1167func runUDPBootstrap(ctx context.Context) error {
1168 // Always start a fresh proxy for a bootstrap request. A client attach begins
1169 // with MsgHello, which must reach the daemon handshake path; reusing an
1170 // already-running proxy would forward that Hello into the daemon connection's
1171 // post-handshake runConnLoop, where it is intentionally ignored. Every
1172 // bootstrap starts a fresh byte-only carriage.
1173 r, w, err := os.Pipe()
1174 if err != nil {
1175 return err
1176 }
1177 defer func() { _ = r.Close() }()
1178
1179 exe, err := os.Executable()
1180 if err != nil {
1181 _ = w.Close()
1182 return err
1183 }
1184 args := []string{"_udp-proxy"}
1185 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
1186 if err != nil {
1187 _ = w.Close()
1188 return err
1189 }
1190 defer func() { _ = devNull.Close() }()
1191
1192 cmd := udpProxyCommand(ctx, exe, args...)
1193 // _udp-proxy writes diagnostics through configureLogging(logging.Stdio, false);
1194 // stdio is detached here so the bootstrap SSH channel can close.
1195 cmd.Stdin = devNull
1196 cmd.Stdout = w
1197 cmd.Stderr = devNull
1198 cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
1199 if err := cmd.Start(); err != nil {
1200 _ = w.Close()
1201 return err
1202 }
1203 _ = w.Close()
1204
1205 readyCtx, cancel := context.WithTimeout(ctx, udpBootstrapTimeout)
1206 defer cancel()
1207 lineCh := make(chan string, 1)
1208 errCh := make(chan error, 1)
1209 go func() {
1210 line, err := bufio.NewReader(r).ReadString('\n')
1211 if err != nil {
1212 errCh <- err
1213 return
1214 }
1215 lineCh <- line
1216 }()
1217 select {
1218 case line := <-lineCh:
1219 if err := cmd.Process.Release(); err != nil {
1220 return err
1221 }
1222 _, err := fmt.Fprint(os.Stdout, line)
1223 return err
1224 case err := <-errCh:
1225 _ = cmd.Wait()
1226 return err
1227 case <-readyCtx.Done():
1228 _ = cmd.Process.Kill()
1229 _ = cmd.Wait()
1230 return fmt.Errorf("vev: udp bootstrap readiness: %w", readyCtx.Err())
1231 }
1232}
1233
1234// runUDPProxy is the detached long-lived remote-side datagram proxy.
1235func runUDPProxy(ctx context.Context, ready io.Writer) (retErr error) {
1236 log, logCloser, err := configureLogging(logging.Stdio, false)
1237 if err != nil {
1238 return err
1239 }
1240 defer func() { _ = logCloser.Close() }()
1241 log.Debug("udp carriage starting")
1242
1243 observer, observerCloser, err := newPerformanceTrace(clock.New())
1244 if err != nil {
1245 return fmt.Errorf("vev: performance trace: %w", err)
1246 }
1247 if observerCloser != nil {
1248 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
1249 }
1250 portRange, err := parseUDPPortRange(os.Getenv(envUDPPortRange))
1251 if err != nil {
1252 return err
1253 }
1254 conn, err := listenUDPInRange(ctx, portRange)
1255 if err != nil {
1256 return err
1257 }
1258 defer func() { _ = conn.Close() }()
1259 daemonTr, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
1260 return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
1261 }, realSpawn, defaultBackoff)
1262 if err != nil {
1263 return err
1264 }
1265 defer func() { _ = daemonTr.Close() }()
1266 udpOptions := udpProxyClientTransportOptions
1267 udpOptions.Observe = dgram.DiagnosticLogObserver(log)
1268 addr, ok := conn.LocalAddr().(*net.UDPAddr)
1269 if !ok {
1270 return errors.New("vev: udp proxy did not get a UDP address")
1271 }
1272 var (
1273 dg *dgram.Transport
1274 setupErr error
1275 )
1276 pdgram.SecretDo(func() {
1277 key := make([]byte, pdgram.KeySize)
1278 defer pdgram.Erase(key)
1279 if _, setupErr = rand.Read(key); setupErr != nil {
1280 return
1281 }
1282 dg, setupErr = dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions, dgram.WithRuntimeObserver(observer))
1283 if setupErr != nil {
1284 return
1285 }
1286 readiness := append([]byte("VEV-UDP "), strconv.Itoa(addr.Port)...)
1287 readiness = append(readiness, ' ')
1288 readiness = base64.StdEncoding.AppendEncode(readiness, key)
1289 readiness = append(readiness, '\n')
1290 defer pdgram.Erase(readiness)
1291 _, setupErr = ready.Write(readiness)
1292 pdgram.Erase(readiness)
1293 pdgram.Erase(key)
1294 })
1295 if setupErr != nil {
1296 if dg != nil {
1297 _ = dg.Close()
1298 }
1299 return setupErr
1300 }
1301 defer func() { _ = dg.Close() }()
1302 return runUDPProxyRuntime(ctx, dg, daemonTr, log)
1303}
1304
1305const udpProxyIdleTTL = 15 * time.Minute
1306
1307var udpProxyClientTransportOptions = dgram.Options{
1308 MaxPending: 32,
1309 DeadAfter: 2 * udpProxyIdleTTL,
1310}
1311
1312const (
1313 // envUDPPortRange configures the remote UDP proxy's listen port range so a
1314 // host firewall can allow a known range instead of a random ephemeral port.
1315 // Format: "START-END" (inclusive), a single "PORT", or "0" for a random
1316 // ephemeral port. Unset uses the default range below.
1317 envUDPPortRange = "VEV_UDP_PORT_RANGE"
1318 defaultUDPPortStart = 61000
1319 defaultUDPPortEnd = 61023
1320)
1321
1322// udpPortRange is an inclusive [start, end] UDP port range. start == 0 means a
1323// random ephemeral port (the old ":0" behavior).
1324type udpPortRange struct {
1325 start int
1326 end int
1327}
1328
1329// parseUDPPortRange parses VEV_UDP_PORT_RANGE. Empty -> default range. "0" ->
1330// ephemeral. "N" -> single port N. "A-B" -> inclusive range. Ports must be in
1331// 1..65535 (or 0 for ephemeral) and end must be >= start.
1332func parseUDPPortRange(value string) (udpPortRange, error) {
1333 value = strings.TrimSpace(value)
1334 if value == "" {
1335 return udpPortRange{start: defaultUDPPortStart, end: defaultUDPPortEnd}, nil
1336 }
1337 startStr, endStr, hasRange := strings.Cut(value, "-")
1338 start, err := parseUDPPort(startStr)
1339 if err != nil {
1340 return udpPortRange{}, fmt.Errorf("invalid %s %q: %w", envUDPPortRange, value, err)
1341 }
1342 if start == 0 {
1343 if hasRange {
1344 return udpPortRange{}, fmt.Errorf("invalid %s %q: port 0 (ephemeral) cannot be combined with a range", envUDPPortRange, value)
1345 }
1346 return udpPortRange{start: 0, end: 0}, nil
1347 }
1348 end := start
1349 if hasRange {
1350 end, err = parseUDPPort(endStr)
1351 if err != nil {
1352 return udpPortRange{}, fmt.Errorf("invalid %s %q: %w", envUDPPortRange, value, err)
1353 }
1354 }
1355 if end < start {
1356 return udpPortRange{}, fmt.Errorf("invalid %s %q: end %d before start %d", envUDPPortRange, value, end, start)
1357 }
1358 return udpPortRange{start: start, end: end}, nil
1359}
1360
1361func parseUDPPort(s string) (int, error) {
1362 p, err := strconv.Atoi(strings.TrimSpace(s))
1363 if err != nil {
1364 return 0, fmt.Errorf("port %q is not a number", strings.TrimSpace(s))
1365 }
1366 if p < 0 || p > 65535 {
1367 return 0, fmt.Errorf("port %d out of range 0-65535", p)
1368 }
1369 return p, nil
1370}
1371
1372// listenUDPInRange binds a UDP packet conn on the first free port in the range,
1373// skipping busy ports. A zero-start range binds a random ephemeral port.
1374func listenUDPInRange(ctx context.Context, r udpPortRange) (net.PacketConn, error) {
1375 var lc net.ListenConfig
1376 if r.start == 0 {
1377 return lc.ListenPacket(ctx, "udp", ":0")
1378 }
1379 if r.start < 0 || r.end < r.start {
1380 return nil, fmt.Errorf("invalid UDP port range %d-%d", r.start, r.end)
1381 }
1382 var lastErr error
1383 for port := r.start; port <= r.end; port++ {
1384 conn, err := lc.ListenPacket(ctx, "udp", fmt.Sprintf(":%d", port))
1385 if err == nil {
1386 return conn, nil
1387 }
1388 lastErr = err
1389 }
1390 return nil, fmt.Errorf("no free UDP port in range %d-%d: %w", r.start, r.end, lastErr)
1391}
1392
1393// runStdioProxy forwards the framed transport without interpreting protocol
1394// messages. Session selection remains in the Hello sent by the thin client.
1395func runStdioProxy(ctx context.Context, client, daemon ports.Transport, log *slog.Logger) error {
1396 return proxyTransports(ctx, client, daemon, log)
1397}
1398
1399func runUDPProxyRuntime(ctx context.Context, client, daemon ports.Transport, log *slog.Logger) error {
1400 return dgram.ProxyRuntime{Client: client, Daemon: daemon, Log: log, IdleTTL: udpProxyIdleTTL}.Run(ctx)
1401}
1402
1403func proxyTransports(ctx context.Context, a, b ports.Transport, log *slog.Logger) error {
1404 return dgram.ProxyRuntime{Client: a, Daemon: b, Log: log}.Run(ctx)
1405}
1406
1407// runList prints the daemon's session listing. With no daemon running, it
1408// falls back to the persisted stopped-session records. Remote forms list one
1409// known host or local+remote when --all is set.
1410func runList(ctx context.Context, cmd command) (retErr error) {
1411 if cmd.listAll || cmd.listHost != "" {
1412 return runRemoteList(ctx, cmd, defaultRemoteHostDeps())
1413 }
1414 sessions, err := listLocalSessions(ctx)
1415 if sessions != nil {
1416 printSessions(os.Stdout, sessions)
1417 }
1418 if err != nil {
1419 return err
1420 }
1421 return nil
1422}
1423
1424func listLocalSessions(ctx context.Context) (_ []ports.SessionInfo, retErr error) {
1425 transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1426 if err != nil {
1427 return nil, fmt.Errorf("vev: waiting for durable session state: %w", err)
1428 }
1429 if owner != nil {
1430 defer joinLifecycleReleaseError(&retErr, owner)
1431 records, loadErr := persist.LoadReadOnly(platform.StateDir())
1432 if loadErr != nil {
1433 if errors.Is(loadErr, persist.ErrCatalogueUnreadable) {
1434 return nil, unreadableCatalogueError(platform.StateDir())
1435 }
1436 return nil, fmt.Errorf("vev: reading stored sessions: %w", loadErr)
1437 }
1438 infos := make([]ports.SessionInfo, 0, len(records))
1439 for _, r := range records {
1440 state := ports.SessionDown
1441 if r.DegradedReason != "" {
1442 state = ports.SessionBroken
1443 }
1444 infos = append(infos, ports.SessionInfo{Name: r.Name, State: state})
1445 }
1446 return infos, nil
1447 }
1448 defer func() { _ = transport.Close() }()
1449
1450 if err := transport.Send(ports.Frame{Type: ports.MsgList, Payload: ports.MarshalList(ports.List{})}); err != nil {
1451 return nil, fmt.Errorf("vev: requesting session list: %w", err)
1452 }
1453 reply, err := transport.Recv()
1454 if err != nil {
1455 return nil, fmt.Errorf("vev: reading session list: %w", err)
1456 }
1457 return decodeSessionListReply(reply)
1458}
1459
1460func decodeSessionListReply(reply ports.Frame) ([]ports.SessionInfo, error) {
1461 if reply.Type == ports.MsgError {
1462 em, err := ports.UnmarshalErrorMsg(reply.Payload)
1463 if err != nil {
1464 return nil, fmt.Errorf("vev: decoding error reply: %w", err)
1465 }
1466 return nil, fmt.Errorf("vev: %s", em.Text)
1467 }
1468 if reply.Type != ports.MsgSessions {
1469 return nil, fmt.Errorf("vev: unexpected reply type %d to list", reply.Type)
1470 }
1471 sessions, err := ports.UnmarshalSessions(reply.Payload)
1472 if err != nil {
1473 return nil, fmt.Errorf("vev: decoding session list: %w", err)
1474 }
1475 return sessions.Sessions, nil
1476}
1477
1478// printSessions renders a session table (or a friendly note when empty).
1479func printSessions(w io.Writer, sessions []ports.SessionInfo) {
1480 if len(sessions) == 0 {
1481 _, _ = fmt.Fprintln(w, "no sessions")
1482 return
1483 }
1484 tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
1485 _, _ = fmt.Fprintln(tw, "NAME\tSTATE\tTABS\tATTACHED")
1486 for _, s := range sessions {
1487 state := "up"
1488 tabs := fmt.Sprintf("%d", s.Tabs)
1489 attached := "no"
1490 switch s.State {
1491 case ports.SessionDown:
1492 state = "down"
1493 tabs = "-"
1494 case ports.SessionBroken:
1495 state = "broken"
1496 tabs = "-"
1497 default:
1498 if s.Ephemeral {
1499 state = "temporary"
1500 }
1501 }
1502 if s.Attached {
1503 attached = "yes"
1504 }
1505 _, _ = fmt.Fprintf(tw, "%s\t%s\t%s\t%s\n", s.Name, state, tabs, attached)
1506 }
1507 _ = tw.Flush()
1508}
1509
1510func unreadableCatalogueError(stateDir string) error {
1511 return fmt.Errorf("%w: vev: durable session state at %s cannot be read and was left untouched.\n"+
1512 "vev does not erase it automatically. To start fresh, remove it:\n"+
1513 " rm -rf %s", persist.ErrCatalogueUnreadable, stateDir, stateDir)
1514}
1515
1516func runOfflineNamedKill(ctx context.Context, name string) (retErr error) {
1517 stateDir := platform.StateDir()
1518 if _, err := os.Stat(persist.StorePath(stateDir)); errors.Is(err, os.ErrNotExist) {
1519 return fmt.Errorf("vev: no such session: %s", name)
1520 } else if err != nil {
1521 return fmt.Errorf("vev: reading stored sessions: %w", err)
1522 }
1523 repository := snapshotadapter.NewRepository(snapshotDir())
1524 opened, err := openCatalogue(stateDir)
1525 if err != nil {
1526 if errors.Is(err, persist.ErrCatalogueUnreadable) {
1527 return unreadableCatalogueError(stateDir)
1528 }
1529 return fmt.Errorf("vev: opening stored sessions: %w", err)
1530 }
1531 defer func() { retErr = errors.Join(retErr, opened.Catalogue.Close()) }()
1532
1533 if _, ok, err := opened.Catalogue.Record(name); err != nil {
1534 return fmt.Errorf("vev: reading stored session: %w", err)
1535 } else if !ok {
1536 return fmt.Errorf("vev: no such session: %s", name)
1537 }
1538 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
1539 if err := coordinator.Delete(ctx, name); err != nil {
1540 return fmt.Errorf("vev: deleting stored session: %w", err)
1541 }
1542 return nil
1543}
1544
1545// runKill asks the daemon to terminate a named session, every session, or the daemon.
1546func runKill(ctx context.Context, name string, all, daemon bool) (retErr error) {
1547 transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1548 if err != nil {
1549 return fmt.Errorf("vev: waiting for durable session state: %w", err)
1550 }
1551 if owner != nil {
1552 defer joinLifecycleReleaseError(&retErr, owner)
1553 if name != "" && !all && !daemon {
1554 if err := runOfflineNamedKill(ctx, name); err != nil {
1555 return err
1556 }
1557 printKillSuccess(name, all, daemon)
1558 return nil
1559 }
1560 return fmt.Errorf("vev: no daemon running")
1561 }
1562 defer func() { _ = transport.Close() }()
1563
1564 if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{Name: name, All: all || daemon})}); err != nil {
1565 return fmt.Errorf("vev: requesting kill: %w", err)
1566 }
1567 reply, err := transport.Recv()
1568 if err != nil {
1569 // The daemon may close the connection after killing; treat a clean
1570 // EOF as success.
1571 if errors.Is(err, io.EOF) {
1572 printKillSuccess(name, all, daemon)
1573 return nil
1574 }
1575 return fmt.Errorf("vev: reading kill reply: %w", err)
1576 }
1577 if reply.Type == ports.MsgError {
1578 em, _ := ports.UnmarshalErrorMsg(reply.Payload)
1579 return fmt.Errorf("vev: %s", em.Text)
1580 }
1581 printKillSuccess(name, all, daemon)
1582 return nil
1583}
1584
1585func printKillSuccess(name string, all, daemon bool) {
1586 if daemon {
1587 fmt.Println("killed daemon")
1588 return
1589 }
1590 if all {
1591 fmt.Println("killed all sessions and stopped daemon")
1592 return
1593 }
1594 fmt.Printf("killed %s\n", name)
1595}