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 resetProtocolIncompatibleRecords(ctx context.Context, log *slog.Logger, coordinator *recovery.Coordinator, records, incompatible []domain.CatalogueRecord) ([]domain.CatalogueRecord, error) {
572 if len(incompatible) == 0 {
573 return records, nil
574 }
575 byName := make(map[string]int, len(records))
576 for i := range records {
577 byName[records[i].Name] = i
578 }
579 for _, old := range incompatible {
580 fresh, committed, err := coordinator.ResetProtocolIncompatible(ctx, old.Name, old.IncarnationID)
581 if !committed {
582 if err != nil {
583 return nil, fmt.Errorf("reset protocol-incompatible session %q: %w", old.Name, err)
584 }
585 return nil, fmt.Errorf("reset protocol-incompatible session %q: catalogue authority changed", old.Name)
586 }
587 if index, ok := byName[old.Name]; ok {
588 records[index] = fresh
589 }
590 log.Info("protocol_incompatible_session_reset", "session", old.Name, "stored_incarnation", old.IncarnationID.String())
591 if err != nil {
592 log.Warn("protocol_incompatible_snapshot_cleanup_failed", "session", old.Name, "err", err)
593 }
594 }
595 return records, nil
596}
597
598func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
599 log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
600}
601
602func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
603 healthy, fresh, broken := 0, 0, 0
604 for _, record := range records {
605 switch {
606 case record.DegradedReason != "":
607 broken++
608 case record.Committed == nil:
609 fresh++
610 default:
611 healthy++
612 }
613 }
614 log.Info("daemon_startup_complete", "healthy", healthy, "fresh", fresh, "restoring", restoring, "broken", broken)
615}
616
617func constructDaemonBeforeSocketPublication(
618 construct func() *daemon.Daemon,
619 prepare func(*daemon.Daemon) error,
620 listen func() (ports.Listener, error),
621) (*daemon.Daemon, ports.Listener, error) {
622 d := construct()
623 if err := prepare(d); err != nil {
624 return d, nil, err
625 }
626 ln, err := listen()
627 return d, ln, err
628}
629
630func runDaemonOwned(ctx context.Context) (retErr error) {
631 log, logCloser, err := configureLogging(logging.Daemon, true)
632 if err != nil {
633 return err
634 }
635 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
636 return runDaemonOwnedWithLogger(ctx, log)
637}
638
639func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr error) {
640 clk := clock.New()
641 observer, observerCloser, err := newPerformanceTrace(clk)
642 if err != nil {
643 return fmt.Errorf("vev: performance trace: %w", err)
644 }
645 if observerCloser != nil {
646 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
647 }
648 remoteDiscoveryOpt, err := remoteDiscoveryDaemonOption(platform.StateDir(), observer, os.Getenv(envRemoteTransport))
649 if err != nil {
650 return err
651 }
652 if addr := os.Getenv("VEV_PPROF_ADDR"); addr != "" {
653 if !pprofAddrIsLoopback(addr) {
654 log.Warn("pprof bound to non-loopback address; /debug/pprof is unauthenticated", "addr", addr)
655 }
656 go func() {
657 if err := http.ListenAndServe(addr, nil); err != nil && !errors.Is(err, http.ErrServerClosed) {
658 log.Error("pprof server exited", "err", err)
659 }
660 }()
661 log.Info("pprof enabled", "addr", addr)
662 }
663
664 daemonOpts := []daemon.Option{remoteDiscoveryOpt}
665 if observer != nil {
666 daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer))
667 }
668 configPath := platform.ConfigPath()
669 cfg, warnings, err := config.Load(configPath)
670 if err != nil {
671 log.Warn("loading config failed; using defaults", "path", configPath, "err", err)
672 cfg = domain.Defaults()
673 }
674 logConfigWarnings(log, warnings)
675 daemonOpts = append(daemonOpts, daemon.WithConfig(cfg))
676 daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New()))
677 daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome))
678 snapshotRepository := snapshotadapter.NewRepositoryWithLogger(snapshotDir(), log)
679 daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository))
680 noticeStore := noticefile.New(platform.StateDir())
681 daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
682 stateDir := platform.StateDir()
683 storePath := persist.StorePath(stateDir)
684 opened, err := openCatalogue(stateDir)
685 if err != nil {
686 log.Error("catalogue_validation_failed", "path", storePath, "reason_code", "open-failed")
687 if errors.Is(err, persist.ErrCatalogueUnreadable) {
688 return unreadableCatalogueError(stateDir)
689 }
690 return fmt.Errorf("vev: open durable session state %s: %w", storePath, err)
691 }
692 recoveryMode := "current"
693 if opened.NewInstall {
694 recoveryMode = "new-install"
695 }
696 coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
697 opened.Records, err = resetProtocolIncompatibleRecords(ctx, log, coordinator, opened.Records, opened.IncompatibleRecords)
698 if err != nil {
699 return errors.Join(fmt.Errorf("vev: reset protocol-incompatible durable sessions: %w", err), opened.Catalogue.Close())
700 }
701 if len(opened.IncompatibleRecords) != 0 {
702 recoveryMode = "protocol-reset"
703 }
704 logCatalogueRecovery(log, opened.Records, recoveryMode)
705 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
706 daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository))
707 log.Info("session persistence enabled", "path", storePath)
708 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
709
710 // Construct the catalogue-backed expected-session registry before socket
711 // publication. Phase 3 snapshot restoration remains asynchronous in Serve.
712 d, ln, err := constructDaemonBeforeSocketPublication(
713 func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
714 func(d *daemon.Daemon) error {
715 if err := d.CollectStartupGarbage(ctx); err != nil {
716 // GC is best-effort, but it is fully finished before socket
717 // publication. A failed pass leaves durable state untouched and
718 // restoration retains its per-session failure isolation.
719 log.Warn("snapshot_garbage_collection_failed", "err", err)
720 }
721 return nil
722 },
723 func() (ports.Listener, error) { return listenDaemon(ipc.SocketDir(), observer) },
724 )
725 if err != nil {
726 closeErr := opened.Catalogue.Close()
727 log.Error("daemon startup preparation failed", "socket_dir", ipc.SocketDir(), "err", err)
728 return errors.Join(fmt.Errorf("vev: prepare daemon startup: %w", err), closeErr)
729 }
730 defer func() { _ = ln.Close() }()
731 log.Info("daemon starting", "socket", ln.Addr())
732
733 watchCtx, stopWatch := context.WithCancel(ctx)
734 defer stopWatch()
735 go func() {
736 if err := config.Watch(watchCtx, clk, configPath, func(cfg domain.Config, warnings []domain.Warning) {
737 logConfigWarnings(log, warnings)
738 d.ApplyConfig(cfg)
739 }); err != nil && !errors.Is(err, context.Canceled) {
740 log.Warn("config watcher stopped", "path", configPath, "err", err)
741 }
742 }()
743 if err := d.Serve(ctx, ln); err != nil {
744 log.Error("daemon exited", "err", err)
745 return err
746 }
747 log.Info("daemon exited cleanly")
748 return nil
749}
750
751// runAttach dials (auto-spawning the daemon if needed) and runs the client
752// attach loop. Logging goes to the shared file: the client must never write
753// to the console while the terminal is raw.
754func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) (retErr error) {
755 // Treat a controlling-terminal hangup or termination request as a graceful
756 // detach rather than an abrupt process death. Cancelling the context unwinds
757 // the client's pumps, closes its transport, and lets the deferred trace closer
758 // flush the in-flight receive's end mark — so teardown never truncates a span.
759 // Raw mode disables ISIG, so catching SIGINT here does not affect interactive
760 // Ctrl+C, which the daemon delivers to the remote shell as a normal keystroke.
761 ctx, stop := signal.NotifyContext(ctx, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT)
762 defer stop()
763
764 log, logCloser, err := configureLogging(logging.Client, false)
765 if err != nil {
766 return err
767 }
768 defer func() { _ = logCloser.Close() }()
769
770 clk := clock.New()
771 observer, observerCloser, err := newPerformanceTrace(clk)
772 if err != nil {
773 return fmt.Errorf("vev: performance trace: %w", err)
774 }
775 if observerCloser != nil {
776 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
777 }
778 return runAttachWithDeps(ctx, intent, name, remoteTarget, os.Getenv("VEV"), log, runAttachDeps{
779 localDialer: func() ports.Dialer {
780 return localDaemonDialer{dir: ipc.SocketDir(), observer: observer}
781 },
782 remoteDialerFactory: newRemoteDialerFactoryWithRuntimeObserver(observer),
783 selectedRemoteTransport: os.Getenv(envRemoteTransport),
784 runClient: runClientWithDeps,
785 createDetached: createDetachedLocalSession,
786 clipboard: clipboard.New(),
787 runtimeObserver: observer,
788 stateDir: platform.StateDir,
789 })
790}
791
792const envRemoteTransport = "VEV_REMOTE_TRANSPORT"
793
794func defaultLocalDialer() ports.Dialer { return localDaemonDialer{dir: ipc.SocketDir()} }
795
796func defaultRemoteDialerFactory() ports.RemoteDialerFactory {
797 return remoteadapter.NewDialerFactory()
798}
799
800type runClientFunc func(context.Context, client.Dependencies, client.AttachRequest) error
801
802type runAttachDeps struct {
803 localDialer func() ports.Dialer
804 remoteDialerFactory ports.RemoteDialerFactory
805 selectedRemoteTransport string
806 runClient runClientFunc
807 createDetached func(context.Context, string) error
808 runtimeObserver ports.SerializedRuntimeObserver
809 // clipboard reads a clipboard image on a remote attach's Ctrl+V (see
810 // docs/superpowers/specs/2026-07-04-clipboard-image-transfer-design.md).
811 // Only used for the remote-dialer branch below; local attaches never
812 // intercept Ctrl+V regardless of this field.
813 clipboard ports.ClipboardReader
814 // Optional remote-learning seams.
815 stateDir func() string
816 hostStore ports.RemoteHostStore
817}
818
819func remoteTransportModeFromEnv(value string) (ports.RemoteTransportMode, error) {
820 switch value {
821 case "", string(ports.RemoteTransportUDP):
822 return ports.RemoteTransportUDP, nil
823 case string(ports.RemoteTransportStdio):
824 return ports.RemoteTransportStdio, nil
825 default:
826 return "", fmt.Errorf("vev: invalid remote transport %q (want %q or %q)", value, ports.RemoteTransportUDP, ports.RemoteTransportStdio)
827 }
828}
829
830func validateRemoteAttachHandoff(target ports.AttachTarget) error {
831 if err := ports.ValidateAttachTarget(target); err != nil {
832 return err
833 }
834 if err := domain.ValidateRemoteHostTarget(target.Endpoint); err != nil {
835 return err
836 }
837 if err := domain.ValidateSessionName(target.Session); err != nil {
838 return err
839 }
840 if target.RemoteTarget != nil {
841 if target.EnvironmentPolicy != ports.EnvironmentPolicyDaemonOwned {
842 return errors.New("remote picker handoff must use daemon-owned environment")
843 }
844 if target.RemoteTarget.Endpoint != target.Endpoint || target.RemoteTarget.SessionName != target.Session {
845 return errors.New("remote picker handoff identity does not match route")
846 }
847 }
848 return nil
849}
850
851// remoteDiscoveryDaemonOption constructs the daemon-owned discovery ports from
852// the same validated transport selection used by direct remote attach.
853func remoteDiscoveryDaemonOption(stateDir string, observer ports.SerializedRuntimeObserver, transport string) (daemon.Option, error) {
854 mode, err := remoteTransportModeFromEnv(transport)
855 if err != nil {
856 return nil, err
857 }
858 discovery := daemon.WithRemoteDiscovery(
859 newRemoteHostStore(remoteadapter.HostStorePath(stateDir)),
860 newRemoteCatalogClient(),
861 newRemoteCatalogCache(remoteadapter.CatalogCachePath(stateDir)),
862 newRemoteDialerFactoryWithRuntimeObserver(observer),
863 mode,
864 )
865 preview := daemon.WithRemotePreview(newRemotePreviewClient())
866 return func(d *daemon.Daemon) {
867 discovery(d)
868 preview(d)
869 }, nil
870}
871
872const maxAttachTargetHandoffs = 32
873
874func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, activeSession string, log *slog.Logger, deps runAttachDeps) error {
875 if activeSession != "" {
876 if remoteTarget == "" && intent == ports.IntentNew {
877 return deps.createDetached(ctx, name)
878 }
879 return errors.New("vev: sessions should be nested with care; unset VEV to force")
880 }
881
882 runClient := deps.runClient
883 if runClient == nil {
884 runClient = runClientWithDeps
885 }
886 localDialer := deps.localDialer
887 if localDialer == nil {
888 localDialer = defaultLocalDialer
889 }
890 mode, modeErr := remoteTransportModeFromEnv(deps.selectedRemoteTransport)
891 factory := deps.remoteDialerFactory
892 var remoteSelection *domain.RemoteSessionTarget
893 var remoteEnvironmentPolicy ports.EnvironmentPolicy
894 remoteDisplayOrigin := domain.RemoteDisplayOrigin(remoteTarget)
895 routeOrigin := ports.RouteOriginLocal
896 routeOriginKey := "local"
897 if remoteTarget != "" {
898 routeOrigin = ports.RouteOriginRemote
899 routeOriginKey = remoteTarget
900 if modeErr != nil {
901 return modeErr
902 }
903 if factory == nil {
904 factory = defaultRemoteDialerFactory()
905 }
906 }
907 pickerHandoff := remoteTarget == ""
908 pickerEnvironmentPolicy := func(target *domain.RemoteSessionTarget, policy ports.EnvironmentPolicy) ports.EnvironmentPolicy {
909 if pickerHandoff && target == nil {
910 return ports.EnvironmentPolicyDaemonOwned
911 }
912 return policy
913 }
914 handoff := func(target ports.AttachTarget) (ports.Dialer, client.AttachRequest, error) {
915 if err := validateRemoteAttachHandoff(target); err != nil {
916 return nil, client.AttachRequest{}, fmt.Errorf("vev: invalid remote attach handoff: %w", err)
917 }
918 if modeErr != nil {
919 return nil, client.AttachRequest{}, modeErr
920 }
921 if factory == nil {
922 factory = defaultRemoteDialerFactory()
923 }
924 var selection *domain.RemoteSessionTarget
925 if target.RemoteTarget != nil {
926 copyTarget := *target.RemoteTarget
927 selection = ©Target
928 }
929 dialer, err := factory.DialerForRemote(target.Endpoint, target.Session, mode, log)
930 if err != nil {
931 return nil, client.AttachRequest{}, err
932 }
933 policy := pickerEnvironmentPolicy(selection, target.EnvironmentPolicy)
934 displayOrigin := domain.RemoteDisplayOrigin(target.Endpoint)
935 if selection != nil {
936 displayOrigin = selection.DisplayOrigin
937 }
938 return dialer, client.AttachRequest{
939 Intent: target.Intent,
940 SessionName: target.Session,
941 Remote: true,
942 Origin: ports.RouteOriginDiscovery,
943 OriginKey: target.Endpoint,
944 RemoteTarget: selection,
945 RemoteOrigin: displayOrigin,
946 EnvironmentPolicy: policy,
947 }, nil
948 }
949
950 handoffAttempts := 0
951 for {
952 var err error
953 if remoteTarget != "" {
954 if log != nil {
955 log.Info("attaching to remote session", "target", remoteTarget, "name", name, "transport", string(mode))
956 }
957 dialer, dialErr := factory.DialerForRemote(remoteTarget, name, mode, log)
958 if dialErr != nil {
959 return dialErr
960 }
961 err = runClient(ctx, client.Dependencies{
962 Dialer: dialer,
963 Terminal: term.New(),
964 Clock: clock.New(),
965 Clipboard: deps.clipboard,
966 Logger: log,
967 RuntimeObserver: deps.runtimeObserver,
968 RemoteHostLearner: attachRememberLearner(deps, remoteTarget, log),
969 AttachHandoff: handoff,
970 Remote: true,
971 Origin: ports.RouteOriginRemote,
972 OriginKey: remoteTarget,
973 }, client.AttachRequest{
974 Intent: intent,
975 SessionName: name,
976 Remote: true,
977 Origin: routeOrigin,
978 OriginKey: routeOriginKey,
979 RemoteTarget: remoteSelection,
980 RemoteOrigin: remoteDisplayOrigin,
981 EnvironmentPolicy: remoteEnvironmentPolicy,
982 })
983 } else {
984 if log != nil {
985 log.Info("attaching to local session", "intent", intent, "name", name)
986 }
987 err = runClient(ctx, client.Dependencies{
988 Dialer: localDialer(),
989 Terminal: term.New(),
990 Clock: clock.New(),
991 Logger: log,
992 RuntimeObserver: deps.runtimeObserver,
993 AttachHandoff: handoff,
994 Remote: false,
995 Origin: ports.RouteOriginLocal,
996 OriginKey: "local",
997 }, client.AttachRequest{Intent: intent, SessionName: name, Origin: ports.RouteOriginLocal, OriginKey: "local"})
998 }
999
1000 var handoffErr *client.AttachTargetError
1001 if !errors.As(err, &handoffErr) {
1002 return err
1003 }
1004 if handoffErr == nil {
1005 return fmt.Errorf("vev: invalid remote attach handoff")
1006 }
1007 if err := validateRemoteAttachHandoff(handoffErr.Target); err != nil {
1008 return fmt.Errorf("vev: invalid remote attach handoff: %w", err)
1009 }
1010 if handoffAttempts >= maxAttachTargetHandoffs {
1011 return fmt.Errorf("vev: attach handoff exceeded maximum of %d attempts", maxAttachTargetHandoffs)
1012 }
1013 handoffAttempts++
1014 remoteTarget = handoffErr.Target.Endpoint
1015 remoteDisplayOrigin = domain.RemoteDisplayOrigin(remoteTarget)
1016 routeOrigin = ports.RouteOriginDiscovery
1017 routeOriginKey = remoteTarget
1018 name = handoffErr.Target.Session
1019 intent = handoffErr.Target.Intent
1020 remoteSelection = nil
1021 if handoffErr.Target.RemoteTarget != nil {
1022 copyTarget := *handoffErr.Target.RemoteTarget
1023 remoteSelection = ©Target
1024 remoteDisplayOrigin = copyTarget.DisplayOrigin
1025 }
1026 remoteEnvironmentPolicy = pickerEnvironmentPolicy(remoteSelection, handoffErr.Target.EnvironmentPolicy)
1027 if modeErr != nil {
1028 return modeErr
1029 }
1030 if factory == nil {
1031 factory = defaultRemoteDialerFactory()
1032 }
1033 }
1034}
1035
1036const daemonStopTimeout = 2 * time.Second
1037
1038type localDaemonDialer struct {
1039 dir string
1040 observer ports.SerializedRuntimeObserver
1041}
1042
1043func (d localDaemonDialer) Dial(ctx context.Context) (ports.Transport, error) {
1044 dial := realDial
1045 if d.observer != nil {
1046 dial = func(ctx context.Context, dir string) (ports.Transport, error) {
1047 return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(d.observer))
1048 }
1049 }
1050 return ensureDaemonWithLifecycle(ctx, d.dir, dial, realSpawn, defaultBackoff)
1051}
1052
1053func detachedLocalHello(name, cwd string) ports.Hello {
1054 termEnv := os.Getenv("TERM")
1055 return ports.Hello{
1056 Version: ports.ProtocolVersion,
1057 Intent: ports.IntentNew,
1058 Name: name,
1059 Size: domain.Size{Cols: 80, Rows: 24},
1060 TermEnv: termEnv,
1061 Cwd: cwd,
1062 TrueColor: client.DetectTrueColor(termEnv, os.Getenv("COLORTERM")),
1063 Env: os.Environ(),
1064 }
1065}
1066
1067func createDetachedLocalSession(ctx context.Context, name string) error {
1068 transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
1069 if err != nil {
1070 return err
1071 }
1072 defer func() { _ = transport.Close() }()
1073
1074 cwd, err := os.Getwd()
1075 if err != nil {
1076 cwd = ""
1077 }
1078 hello := detachedLocalHello(name, cwd)
1079 if err := transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
1080 return fmt.Errorf("vev: creating detached session: %w", err)
1081 }
1082
1083 type detachedReply struct {
1084 frame ports.Frame
1085 err error
1086 }
1087 replyCh := make(chan detachedReply, 1)
1088 go func() {
1089 f, err := transport.Recv()
1090 replyCh <- detachedReply{frame: f, err: err}
1091 }()
1092
1093 select {
1094 case <-ctx.Done():
1095 _ = transport.Close()
1096 return ctx.Err()
1097 case reply := <-replyCh:
1098 if reply.err != nil {
1099 return fmt.Errorf("vev: awaiting detached session creation: %w", reply.err)
1100 }
1101 switch reply.frame.Type {
1102 case ports.MsgWelcome:
1103 return nil
1104 case ports.MsgError:
1105 em, derr := ports.UnmarshalErrorMsg(reply.frame.Payload)
1106 if derr != nil {
1107 return fmt.Errorf("vev: decoding error reply: %w", derr)
1108 }
1109 return &client.ProtocolError{Code: em.Code, Text: em.Text}
1110 default:
1111 return fmt.Errorf("vev: unexpected reply type %d to detached session creation", reply.frame.Type)
1112 }
1113 }
1114}
1115
1116func requestDaemonStop(ctx context.Context) error {
1117 ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)
1118 defer cancel()
1119
1120 transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1121 if err != nil {
1122 return fmt.Errorf("vev: stopping daemon: %w", err)
1123 }
1124 if owner != nil {
1125 return errors.Join(errors.New("vev: no daemon running"), owner.Release())
1126 }
1127 defer func() { _ = transport.Close() }()
1128 if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}); err != nil {
1129 return fmt.Errorf("vev: requesting daemon stop: %w", err)
1130 }
1131 if _, err := transport.Recv(); err != nil && !errors.Is(err, io.EOF) {
1132 return fmt.Errorf("vev: reading daemon stop reply: %w", err)
1133 }
1134
1135 owner, err = waitForLifecycleAvailability(ctx, ipc.SocketDir(), defaultBackoff)
1136 if err != nil {
1137 return fmt.Errorf("vev: waiting for daemon ownership transfer: %w", err)
1138 }
1139 return owner.Release()
1140}
1141
1142// runStdio is the hidden remote-side mode used by `ssh host vev _stdio`: it
1143// connects to the per-user daemon (auto-spawning it if needed) and proxies the
1144// framed protocol between process stdio and the daemon socket.
1145func runStdio(ctx context.Context) (retErr error) {
1146 log, logCloser, err := configureLogging(logging.Stdio, false)
1147 if err != nil {
1148 return err
1149 }
1150 defer func() { _ = logCloser.Close() }()
1151 log.Debug("stdio proxy starting")
1152
1153 clk := clock.New()
1154 observer, observerCloser, err := newPerformanceTrace(clk)
1155 if err != nil {
1156 return fmt.Errorf("vev: performance trace: %w", err)
1157 }
1158 if observerCloser != nil {
1159 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
1160 }
1161 transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
1162 return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
1163 }, realSpawn, defaultBackoff)
1164 if err != nil {
1165 return err
1166 }
1167 defer func() { _ = transport.Close() }()
1168
1169 // sshstdio.Close owns and closes writers that implement io.Closer so a
1170 // blocked Send can unwind. Keep process-global stdin/stdout outside that
1171 // ownership boundary: sshstdio recognizes process stdin as a shared
1172 // cancellable pump, while output is an owned pipe drained by a forwarding
1173 // goroutine.
1174 stdin, stdout := os.Stdin, os.Stdout
1175 stdoutReader, stdoutWriter := io.Pipe()
1176 forwardDone := make(chan struct{})
1177 go func() {
1178 defer close(forwardDone)
1179 if _, err := io.Copy(stdout, stdoutReader); err != nil {
1180 _ = stdoutReader.CloseWithError(err)
1181 }
1182 }()
1183 defer func() {
1184 _ = stdoutWriter.Close()
1185 <-forwardDone
1186 _ = stdoutReader.Close()
1187 }()
1188 stdio := sshstdio.NewTransport(stdin, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer))
1189 return runStdioProxy(ctx, stdio, transport, log)
1190}
1191
1192var (
1193 udpBootstrapTimeout = 5 * time.Second
1194 udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd {
1195 return exec.CommandContext(ctx, exe, args...)
1196 }
1197)
1198
1199// runUDPBootstrap starts a detached _udp-proxy, forwards its single readiness
1200// line, and exits so SSH stdio can close without owning the UDP proxy lifetime.
1201func runUDPBootstrap(ctx context.Context) error {
1202 // Always start a fresh proxy for a bootstrap request. A client attach begins
1203 // with MsgHello, which must reach the daemon handshake path; reusing an
1204 // already-running proxy would forward that Hello into the daemon connection's
1205 // post-handshake runConnLoop, where it is intentionally ignored. Every
1206 // bootstrap starts a fresh byte-only carriage.
1207 r, w, err := os.Pipe()
1208 if err != nil {
1209 return err
1210 }
1211 defer func() { _ = r.Close() }()
1212
1213 exe, err := os.Executable()
1214 if err != nil {
1215 _ = w.Close()
1216 return err
1217 }
1218 args := []string{"_udp-proxy"}
1219 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
1220 if err != nil {
1221 _ = w.Close()
1222 return err
1223 }
1224 defer func() { _ = devNull.Close() }()
1225
1226 cmd := udpProxyCommand(ctx, exe, args...)
1227 // _udp-proxy writes diagnostics through configureLogging(logging.Stdio, false);
1228 // stdio is detached here so the bootstrap SSH channel can close.
1229 cmd.Stdin = devNull
1230 cmd.Stdout = w
1231 cmd.Stderr = devNull
1232 cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
1233 if err := cmd.Start(); err != nil {
1234 _ = w.Close()
1235 return err
1236 }
1237 _ = w.Close()
1238
1239 readyCtx, cancel := context.WithTimeout(ctx, udpBootstrapTimeout)
1240 defer cancel()
1241 lineCh := make(chan string, 1)
1242 errCh := make(chan error, 1)
1243 go func() {
1244 line, err := bufio.NewReader(r).ReadString('\n')
1245 if err != nil {
1246 errCh <- err
1247 return
1248 }
1249 lineCh <- line
1250 }()
1251 select {
1252 case line := <-lineCh:
1253 if err := cmd.Process.Release(); err != nil {
1254 return err
1255 }
1256 _, err := fmt.Fprint(os.Stdout, line)
1257 return err
1258 case err := <-errCh:
1259 _ = cmd.Wait()
1260 return err
1261 case <-readyCtx.Done():
1262 _ = cmd.Process.Kill()
1263 _ = cmd.Wait()
1264 return fmt.Errorf("vev: udp bootstrap readiness: %w", readyCtx.Err())
1265 }
1266}
1267
1268// runUDPProxy is the detached long-lived remote-side datagram proxy.
1269func runUDPProxy(ctx context.Context, ready io.Writer) (retErr error) {
1270 log, logCloser, err := configureLogging(logging.Stdio, false)
1271 if err != nil {
1272 return err
1273 }
1274 defer func() { _ = logCloser.Close() }()
1275 log.Debug("udp carriage starting")
1276
1277 observer, observerCloser, err := newPerformanceTrace(clock.New())
1278 if err != nil {
1279 return fmt.Errorf("vev: performance trace: %w", err)
1280 }
1281 if observerCloser != nil {
1282 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
1283 }
1284 portRange, err := parseUDPPortRange(os.Getenv(envUDPPortRange))
1285 if err != nil {
1286 return err
1287 }
1288 conn, err := listenUDPInRange(ctx, portRange)
1289 if err != nil {
1290 return err
1291 }
1292 defer func() { _ = conn.Close() }()
1293 daemonTr, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
1294 return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
1295 }, realSpawn, defaultBackoff)
1296 if err != nil {
1297 return err
1298 }
1299 defer func() { _ = daemonTr.Close() }()
1300 udpOptions := udpProxyClientTransportOptions
1301 udpOptions.Observe = dgram.DiagnosticLogObserver(log)
1302 addr, ok := conn.LocalAddr().(*net.UDPAddr)
1303 if !ok {
1304 return errors.New("vev: udp proxy did not get a UDP address")
1305 }
1306 var (
1307 dg *dgram.Transport
1308 setupErr error
1309 )
1310 pdgram.SecretDo(func() {
1311 key := make([]byte, pdgram.KeySize)
1312 defer pdgram.Erase(key)
1313 if _, setupErr = rand.Read(key); setupErr != nil {
1314 return
1315 }
1316 dg, setupErr = dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions, dgram.WithRuntimeObserver(observer))
1317 if setupErr != nil {
1318 return
1319 }
1320 readiness := append([]byte("VEV-UDP "), strconv.Itoa(addr.Port)...)
1321 readiness = append(readiness, ' ')
1322 readiness = base64.StdEncoding.AppendEncode(readiness, key)
1323 readiness = append(readiness, '\n')
1324 defer pdgram.Erase(readiness)
1325 _, setupErr = ready.Write(readiness)
1326 pdgram.Erase(readiness)
1327 pdgram.Erase(key)
1328 })
1329 if setupErr != nil {
1330 if dg != nil {
1331 _ = dg.Close()
1332 }
1333 return setupErr
1334 }
1335 defer func() { _ = dg.Close() }()
1336 return runUDPProxyRuntime(ctx, dg, daemonTr, log)
1337}
1338
1339const udpProxyIdleTTL = 15 * time.Minute
1340
1341var udpProxyClientTransportOptions = dgram.Options{
1342 MaxPending: 32,
1343 DeadAfter: 2 * udpProxyIdleTTL,
1344}
1345
1346const (
1347 // envUDPPortRange configures the remote UDP proxy's listen port range so a
1348 // host firewall can allow a known range instead of a random ephemeral port.
1349 // Format: "START-END" (inclusive), a single "PORT", or "0" for a random
1350 // ephemeral port. Unset uses the default range below.
1351 envUDPPortRange = "VEV_UDP_PORT_RANGE"
1352 defaultUDPPortStart = 61000
1353 defaultUDPPortEnd = 61023
1354)
1355
1356// udpPortRange is an inclusive [start, end] UDP port range. start == 0 means a
1357// random ephemeral port (the old ":0" behavior).
1358type udpPortRange struct {
1359 start int
1360 end int
1361}
1362
1363// parseUDPPortRange parses VEV_UDP_PORT_RANGE. Empty -> default range. "0" ->
1364// ephemeral. "N" -> single port N. "A-B" -> inclusive range. Ports must be in
1365// 1..65535 (or 0 for ephemeral) and end must be >= start.
1366func parseUDPPortRange(value string) (udpPortRange, error) {
1367 value = strings.TrimSpace(value)
1368 if value == "" {
1369 return udpPortRange{start: defaultUDPPortStart, end: defaultUDPPortEnd}, nil
1370 }
1371 startStr, endStr, hasRange := strings.Cut(value, "-")
1372 start, err := parseUDPPort(startStr)
1373 if err != nil {
1374 return udpPortRange{}, fmt.Errorf("invalid %s %q: %w", envUDPPortRange, value, err)
1375 }
1376 if start == 0 {
1377 if hasRange {
1378 return udpPortRange{}, fmt.Errorf("invalid %s %q: port 0 (ephemeral) cannot be combined with a range", envUDPPortRange, value)
1379 }
1380 return udpPortRange{start: 0, end: 0}, nil
1381 }
1382 end := start
1383 if hasRange {
1384 end, err = parseUDPPort(endStr)
1385 if err != nil {
1386 return udpPortRange{}, fmt.Errorf("invalid %s %q: %w", envUDPPortRange, value, err)
1387 }
1388 }
1389 if end < start {
1390 return udpPortRange{}, fmt.Errorf("invalid %s %q: end %d before start %d", envUDPPortRange, value, end, start)
1391 }
1392 return udpPortRange{start: start, end: end}, nil
1393}
1394
1395func parseUDPPort(s string) (int, error) {
1396 p, err := strconv.Atoi(strings.TrimSpace(s))
1397 if err != nil {
1398 return 0, fmt.Errorf("port %q is not a number", strings.TrimSpace(s))
1399 }
1400 if p < 0 || p > 65535 {
1401 return 0, fmt.Errorf("port %d out of range 0-65535", p)
1402 }
1403 return p, nil
1404}
1405
1406// listenUDPInRange binds a UDP packet conn on the first free port in the range,
1407// skipping busy ports. A zero-start range binds a random ephemeral port.
1408func listenUDPInRange(ctx context.Context, r udpPortRange) (net.PacketConn, error) {
1409 var lc net.ListenConfig
1410 if r.start == 0 {
1411 return lc.ListenPacket(ctx, "udp", ":0")
1412 }
1413 if r.start < 0 || r.end < r.start {
1414 return nil, fmt.Errorf("invalid UDP port range %d-%d", r.start, r.end)
1415 }
1416 var lastErr error
1417 for port := r.start; port <= r.end; port++ {
1418 conn, err := lc.ListenPacket(ctx, "udp", fmt.Sprintf(":%d", port))
1419 if err == nil {
1420 return conn, nil
1421 }
1422 lastErr = err
1423 }
1424 return nil, fmt.Errorf("no free UDP port in range %d-%d: %w", r.start, r.end, lastErr)
1425}
1426
1427// runStdioProxy forwards the framed transport without interpreting protocol
1428// messages. Session selection remains in the Hello sent by the thin client.
1429func runStdioProxy(ctx context.Context, client, daemon ports.Transport, log *slog.Logger) error {
1430 return proxyTransports(ctx, client, daemon, log)
1431}
1432
1433func runUDPProxyRuntime(ctx context.Context, client, daemon ports.Transport, log *slog.Logger) error {
1434 return dgram.ProxyRuntime{Client: client, Daemon: daemon, Log: log, IdleTTL: udpProxyIdleTTL}.Run(ctx)
1435}
1436
1437func proxyTransports(ctx context.Context, a, b ports.Transport, log *slog.Logger) error {
1438 return dgram.ProxyRuntime{Client: a, Daemon: b, Log: log}.Run(ctx)
1439}
1440
1441// runList prints the daemon's session listing. With no daemon running, it
1442// falls back to the persisted stopped-session records. Remote forms list one
1443// known host or local+remote when --all is set.
1444func runList(ctx context.Context, cmd command) (retErr error) {
1445 if cmd.listAll || cmd.listHost != "" {
1446 return runRemoteList(ctx, cmd, defaultRemoteHostDeps())
1447 }
1448 sessions, err := listLocalSessions(ctx)
1449 if sessions != nil {
1450 printSessions(os.Stdout, sessions)
1451 }
1452 if err != nil {
1453 return err
1454 }
1455 return nil
1456}
1457
1458func listLocalSessions(ctx context.Context) (_ []ports.SessionInfo, retErr error) {
1459 transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1460 if err != nil {
1461 return nil, fmt.Errorf("vev: waiting for durable session state: %w", err)
1462 }
1463 if owner != nil {
1464 defer joinLifecycleReleaseError(&retErr, owner)
1465 records, loadErr := persist.LoadReadOnly(platform.StateDir())
1466 if loadErr != nil {
1467 if errors.Is(loadErr, persist.ErrCatalogueUnreadable) {
1468 return nil, unreadableCatalogueError(platform.StateDir())
1469 }
1470 return nil, fmt.Errorf("vev: reading stored sessions: %w", loadErr)
1471 }
1472 infos := make([]ports.SessionInfo, 0, len(records))
1473 for _, r := range records {
1474 state := ports.SessionDown
1475 if r.DegradedReason != "" {
1476 state = ports.SessionBroken
1477 }
1478 infos = append(infos, ports.SessionInfo{Name: r.Name, State: state})
1479 }
1480 return infos, nil
1481 }
1482 defer func() { _ = transport.Close() }()
1483
1484 if err := transport.Send(ports.Frame{Type: ports.MsgList, Payload: ports.MarshalList(ports.List{})}); err != nil {
1485 return nil, fmt.Errorf("vev: requesting session list: %w", err)
1486 }
1487 reply, err := transport.Recv()
1488 if err != nil {
1489 return nil, fmt.Errorf("vev: reading session list: %w", err)
1490 }
1491 return decodeSessionListReply(reply)
1492}
1493
1494func decodeSessionListReply(reply ports.Frame) ([]ports.SessionInfo, error) {
1495 if reply.Type == ports.MsgError {
1496 em, err := ports.UnmarshalErrorMsg(reply.Payload)
1497 if err != nil {
1498 return nil, fmt.Errorf("vev: decoding error reply: %w", err)
1499 }
1500 return nil, fmt.Errorf("vev: %s", em.Text)
1501 }
1502 if reply.Type != ports.MsgSessions {
1503 return nil, fmt.Errorf("vev: unexpected reply type %d to list", reply.Type)
1504 }
1505 sessions, err := ports.UnmarshalSessions(reply.Payload)
1506 if err != nil {
1507 return nil, fmt.Errorf("vev: decoding session list: %w", err)
1508 }
1509 return sessions.Sessions, nil
1510}
1511
1512// printSessions renders a session table (or a friendly note when empty).
1513func printSessions(w io.Writer, sessions []ports.SessionInfo) {
1514 if len(sessions) == 0 {
1515 _, _ = fmt.Fprintln(w, "no sessions")
1516 return
1517 }
1518 tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
1519 _, _ = fmt.Fprintln(tw, "NAME\tSTATE\tTABS\tATTACHED")
1520 for _, s := range sessions {
1521 state := "up"
1522 tabs := fmt.Sprintf("%d", s.Tabs)
1523 attached := "no"
1524 switch s.State {
1525 case ports.SessionDown:
1526 state = "down"
1527 tabs = "-"
1528 case ports.SessionBroken:
1529 state = "broken"
1530 tabs = "-"
1531 default:
1532 if s.Ephemeral {
1533 state = "temporary"
1534 }
1535 }
1536 if s.Attached {
1537 attached = "yes"
1538 }
1539 _, _ = fmt.Fprintf(tw, "%s\t%s\t%s\t%s\n", s.Name, state, tabs, attached)
1540 }
1541 _ = tw.Flush()
1542}
1543
1544func unreadableCatalogueError(stateDir string) error {
1545 return fmt.Errorf("%w: vev: durable session state at %s cannot be read and was left untouched.\n"+
1546 "vev does not erase it automatically. To start fresh, remove it:\n"+
1547 " rm -rf %s", persist.ErrCatalogueUnreadable, stateDir, stateDir)
1548}
1549
1550func runOfflineNamedKill(ctx context.Context, name string) (retErr error) {
1551 stateDir := platform.StateDir()
1552 if _, err := os.Stat(persist.StorePath(stateDir)); errors.Is(err, os.ErrNotExist) {
1553 return fmt.Errorf("vev: no such session: %s", name)
1554 } else if err != nil {
1555 return fmt.Errorf("vev: reading stored sessions: %w", err)
1556 }
1557 repository := snapshotadapter.NewRepository(snapshotDir())
1558 opened, err := openCatalogue(stateDir)
1559 if err != nil {
1560 if errors.Is(err, persist.ErrCatalogueUnreadable) {
1561 return unreadableCatalogueError(stateDir)
1562 }
1563 return fmt.Errorf("vev: opening stored sessions: %w", err)
1564 }
1565 defer func() { retErr = errors.Join(retErr, opened.Catalogue.Close()) }()
1566
1567 if _, ok, err := opened.Catalogue.Record(name); err != nil {
1568 return fmt.Errorf("vev: reading stored session: %w", err)
1569 } else if !ok {
1570 return fmt.Errorf("vev: no such session: %s", name)
1571 }
1572 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
1573 if err := coordinator.Delete(ctx, name); err != nil {
1574 return fmt.Errorf("vev: deleting stored session: %w", err)
1575 }
1576 return nil
1577}
1578
1579// runKill asks the daemon to terminate a named session, every session, or the daemon.
1580func runKill(ctx context.Context, name string, all, daemon bool) (retErr error) {
1581 transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1582 if err != nil {
1583 return fmt.Errorf("vev: waiting for durable session state: %w", err)
1584 }
1585 if owner != nil {
1586 defer joinLifecycleReleaseError(&retErr, owner)
1587 if name != "" && !all && !daemon {
1588 if err := runOfflineNamedKill(ctx, name); err != nil {
1589 return err
1590 }
1591 printKillSuccess(name, all, daemon)
1592 return nil
1593 }
1594 return fmt.Errorf("vev: no daemon running")
1595 }
1596 defer func() { _ = transport.Close() }()
1597
1598 if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{Name: name, All: all || daemon})}); err != nil {
1599 return fmt.Errorf("vev: requesting kill: %w", err)
1600 }
1601 reply, err := transport.Recv()
1602 if err != nil {
1603 // The daemon may close the connection after killing; treat a clean
1604 // EOF as success.
1605 if errors.Is(err, io.EOF) {
1606 printKillSuccess(name, all, daemon)
1607 return nil
1608 }
1609 return fmt.Errorf("vev: reading kill reply: %w", err)
1610 }
1611 if reply.Type == ports.MsgError {
1612 em, _ := ports.UnmarshalErrorMsg(reply.Payload)
1613 return fmt.Errorf("vev: %s", em.Text)
1614 }
1615 printKillSuccess(name, all, daemon)
1616 return nil
1617}
1618
1619func printKillSuccess(name string, all, daemon bool) {
1620 if daemon {
1621 fmt.Println("killed daemon")
1622 return
1623 }
1624 if all {
1625 fmt.Println("killed all sessions and stopped daemon")
1626 return
1627 }
1628 fmt.Printf("killed %s\n", name)
1629}