1package main
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "log/slog"
10 "os"
11 "os/exec"
12 "strings"
13 "sync"
14 "sync/atomic"
15 "time"
16
17 "github.com/bnema/vev/internal/adapters/dgram"
18 "github.com/bnema/vev/internal/adapters/ipc"
19 remoteadapter "github.com/bnema/vev/internal/adapters/remote"
20 "github.com/bnema/vev/internal/adapters/sshstdio"
21 "github.com/bnema/vev/internal/domain"
22 "github.com/bnema/vev/internal/ports"
23)
24
25const (
26 defaultTarget = "test@remote"
27 sessionName = "harness"
28)
29
30var requestID atomic.Uint64
31
32func main() {
33 if err := run(); err != nil {
34 fmt.Fprintf(os.Stderr, "remote picker harness: %v\n", err)
35 os.Exit(1)
36 }
37 fmt.Println("remote picker harness: PASS")
38}
39
40func run() (runErr error) {
41 target := os.Getenv("VEV_HARNESS_TARGET")
42 if target == "" {
43 target = defaultTarget
44 }
45 ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
46 defer cancel()
47
48 artifact := newHarnessArtifact(os.Getenv("VEV_HARNESS_ARTIFACT_DIR"))
49 if artifact != nil {
50 ctx = withHarnessArtifact(ctx, artifact)
51 defer func() {
52 if err := artifact.write(runErr == nil); err != nil {
53 runErr = errors.Join(runErr, err)
54 }
55 }()
56 }
57
58 catalog := remoteadapter.NewCatalogClient()
59 preview := remoteadapter.NewPreviewClient()
60 if err := runLocalPickerUnitedPhase(ctx, target, catalog, preview); err != nil {
61 return err
62 }
63 if err := runLiveStdioPhase(ctx, target, catalog, preview); err != nil {
64 return err
65 }
66 resumedTarget, err := runRestartResumePhase(ctx, target, catalog, preview)
67 if err != nil {
68 return err
69 }
70 if err := runUDPPhase(ctx, target); err != nil {
71 return err
72 }
73 if err := runCatalogFailurePhase(ctx, target, catalog); err != nil {
74 return err
75 }
76 if _, err := catalog.List(ctx, "test@unreachable.invalid"); err == nil {
77 return errors.New("unreachable catalog unexpectedly succeeded")
78 }
79 return runStaleTargetPhase(ctx, target, catalog, resumedTarget)
80}
81
82func runLiveStdioPhase(ctx context.Context, target string, catalog ports.RemoteCatalogClient, preview ports.RemotePreviewClient) error {
83 stdio, err := openTransport(ctx, ports.RemoteTransportStdio, target)
84 if err != nil {
85 return fmt.Errorf("open stdio transport: %w", err)
86 }
87 defer func() { _ = stdio.Close() }()
88 if err := attachAndCheck(ctx, stdio, ports.Hello{
89 Intent: ports.IntentNew, Name: sessionName, Env: localEnvironment(), Cwd: "/tmp/local-cwd",
90 TermEnv: "xterm-256color", TrueColor: true,
91 }); err != nil {
92 return fmt.Errorf("stdio live attach: %w", err)
93 }
94 if err := sendInputAndAwait(ctx, stdio, "printf 'VEV_REMOTE_HARNESS_ENV=%s:%s:%s:%s\\n' \"$VEV_TEST_ENV\" \"$XDG_RUNTIME_DIR\" \"$WAYLAND_DISPLAY\" \"$VEV\"", "local:/run/local:wayland-local:"); err != nil {
95 return fmt.Errorf("direct client environment: %w", err)
96 }
97 live, err := waitForSession(ctx, catalog, target, sessionName, "up")
98 if err != nil {
99 return fmt.Errorf("live catalog: %w", err)
100 }
101 liveTarget, err := targetForSession(target, live, false)
102 if err != nil {
103 return fmt.Errorf("live target: %w", err)
104 }
105 if _, err := preview.Preview(ctx, liveTarget, 40, 8); err != nil {
106 return fmt.Errorf("live preview: %w", err)
107 }
108 if err := sendCommand(ctx, stdio, "new-tab"); err != nil {
109 return fmt.Errorf("create second tab: %w", err)
110 }
111 liveWithTabs, err := waitForSessionWithTabCount(ctx, catalog, target, sessionName, "up", 2)
112 if err != nil {
113 return fmt.Errorf("multi-tab catalog: %w", err)
114 }
115 beforeTabs := ports.CatalogTabs(liveWithTabs)
116 if err := sendCommand(ctx, stdio, "close-tab"); err != nil {
117 return fmt.Errorf("remove tab: %w", err)
118 }
119 after, err := waitForSessionWithTabCount(ctx, catalog, target, sessionName, "up", 1)
120 if err != nil {
121 return fmt.Errorf("removed-tab catalog: %w", err)
122 }
123 removedID, ok := removedCatalogTabID(beforeTabs, ports.CatalogTabs(after))
124 if !ok {
125 return errors.New("removed-tab catalog did not identify exactly one removed tab")
126 }
127 removedSession := liveWithTabs
128 removedSession.ActiveTabID = removedID
129 removedTarget, err := targetForSession(target, removedSession, false)
130 if err != nil {
131 return fmt.Errorf("removed-tab target: %w", err)
132 }
133 if _, err := preview.Preview(ctx, removedTarget, 40, 8); err == nil {
134 return errors.New("removed tab unexpectedly produced a preview")
135 }
136 survivingTarget, err := targetForSession(target, after, false)
137 if err != nil {
138 return fmt.Errorf("surviving-tab target: %w", err)
139 }
140 if _, err := preview.Preview(ctx, survivingTarget, 40, 8); err != nil {
141 return fmt.Errorf("surviving tab preview: %w", err)
142 }
143 if err := detach(ctx, stdio); err != nil {
144 return fmt.Errorf("stop session: %w", err)
145 }
146 return nil
147}
148
149// runLocalPickerUnitedPhase drives the actual local picker against the SSH
150// daemon container. It validates the seam that individual catalog, preview,
151// and attach tests cannot: unified rows before and after selection, remote
152// content publication on the same local client transport, local chrome
153// composition, and continued input forwarding.
154func runLocalPickerUnitedPhase(ctx context.Context, target string, catalog ports.RemoteCatalogClient, preview ports.RemotePreviewClient) (err error) {
155 const (
156 localSession = "local-picker"
157 remoteSession = "picker"
158 previewMarker = "VEV_REMOTE_PICKER_ACTIVE"
159 inputMarker = "VEV_REMOTE_PICKER_INPUT_OK"
160 )
161
162 remote, err := openTransport(ctx, ports.RemoteTransportStdio, target)
163 if err != nil {
164 return fmt.Errorf("open remote picker source: %w", err)
165 }
166 remote.probe.configure("remote-picker-source")
167 ownsRemoteSession := false
168 defer func() {
169 _ = remote.Close()
170 if !ownsRemoteSession {
171 return
172 }
173 cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second)
174 defer cancel()
175 if cleanupErr := runRemoteCommand(cleanup, target, "vev", "kill", remoteSession); cleanupErr != nil {
176 err = errors.Join(err, fmt.Errorf("destroy remote picker session: %w", cleanupErr))
177 }
178 }()
179 if err := attachAndCheck(ctx, remote, ports.Hello{
180 Intent: ports.IntentNew, Name: remoteSession, Env: localEnvironment(), Cwd: "/tmp/remote-cwd",
181 TermEnv: "xterm-256color", TrueColor: true,
182 }); err != nil {
183 return fmt.Errorf("attach remote picker source: %w", err)
184 }
185 ownsRemoteSession = true
186 if err := sendCommand(ctx, remote, "new-tab"); err != nil {
187 return fmt.Errorf("create remote picker tab: %w", err)
188 }
189 if err := sendCommand(ctx, remote, "next-tab"); err != nil {
190 return fmt.Errorf("focus remote picker tab: %w", err)
191 }
192 if err := sendInputAndAwait(ctx, remote, "printf '"+previewMarker+"\\n'", previewMarker); err != nil {
193 return fmt.Errorf("seed remote picker preview: %w", err)
194 }
195 remoteCatalog, err := waitForSessionWithTabCount(ctx, catalog, target, remoteSession, "up", 2)
196 if err != nil {
197 return fmt.Errorf("remote picker catalog: %w", err)
198 }
199 if remoteCatalog.ActiveTabID == "" {
200 return errors.New("remote picker catalog omitted active tab")
201 }
202 tabs := ports.CatalogTabs(remoteCatalog)
203 if len(tabs) != 2 {
204 return errors.New("remote picker catalog did not retain both tabs")
205 }
206 selectedTarget, err := targetForSession(target, remoteCatalog, false)
207 if err != nil {
208 return fmt.Errorf("active remote picker target: %w", err)
209 }
210 if selectedTarget.LiveTabID != domain.TabStableID(remoteCatalog.ActiveTabID) {
211 return errors.New("remote picker catalog active tab did not resolve to the selected target")
212 }
213 activeTabIndex := -1
214 for i, tab := range tabs {
215 if domain.TabStableID(tab.ID) == selectedTarget.LiveTabID {
216 activeTabIndex = i
217 break
218 }
219 }
220 if activeTabIndex < 0 {
221 return errors.New("remote picker catalog active tab was absent from tab list")
222 }
223 selectedPreview, err := preview.Preview(ctx, selectedTarget, 55, 22)
224 if err != nil {
225 return fmt.Errorf("explicit remote tab preview: %w", err)
226 }
227 if !remotePreviewContains(selectedPreview, previewMarker) {
228 return errors.New("explicit remote tab preview omitted marker")
229 }
230
231 if err := runLocalCommand(ctx, "vev", "host", "add", target); err != nil {
232 return fmt.Errorf("register remote picker host: %w", err)
233 }
234 localDaemon, err := startLocalDaemon(ctx)
235 if err != nil {
236 return fmt.Errorf("start local picker daemon: %w", err)
237 }
238 defer stopLocalDaemon(localDaemon)
239
240 local, err := openLocalTransport(ctx)
241 if err != nil {
242 return fmt.Errorf("open local picker transport: %w", err)
243 }
244 // This is the one client-equivalent connection for the proof. Remote
245 // selection must replace its daemon-side owner without sending an endpoint
246 // or MsgAttachTarget to this transport.
247 local.probe.configure("local-picker")
248 defer func() { _ = local.Close() }()
249 if err := attachAndCheck(ctx, local, ports.Hello{
250 Intent: ports.IntentNew, Name: localSession, Env: localEnvironment(), Cwd: "/tmp/local-cwd",
251 TermEnv: "xterm-256color", TrueColor: true,
252 }); err != nil {
253 return fmt.Errorf("attach local picker session: %w", err)
254 }
255 if err := sendRawInput(local, "\x1b "); err != nil {
256 return fmt.Errorf("open command palette: %w", err)
257 }
258 if err := awaitOutputContains(ctx, local, "Commands"); err != nil {
259 return fmt.Errorf("await command palette: %w", err)
260 }
261 if err := sendRawInput(local, "SSP\r"); err != nil {
262 return fmt.Errorf("open session picker: %w", err)
263 }
264 if err := awaitOutputContains(ctx, local, remoteSession+"@remote"); err != nil {
265 return fmt.Errorf("await remote picker catalog row: %w", err)
266 }
267 beforeRows := capturePickerRows(local.probe)
268 if err := assertUnifiedPickerRows(beforeRows, localSession, remoteSession); err != nil {
269 return fmt.Errorf("capture unified picker rows before remote selection: %w", err)
270 }
271 // The local selected tab is first and the remote session header is not
272 // selectable, so one down reaches the first remote tab. Derive the rest
273 // from the catalog's stable active tab identity instead of assuming an
274 // ordinal target.
275 if err := sendRawInput(local, strings.Repeat("j", activeTabIndex+1)); err != nil {
276 return fmt.Errorf("select catalog active remote tab: %w", err)
277 }
278 if err := awaitRemotePickerPreview(ctx, local, previewMarker); err != nil {
279 return fmt.Errorf("remote picker preview: %w", err)
280 }
281 if err := sendRawInput(local, "\r"); err != nil {
282 return fmt.Errorf("commit remote picker target: %w", err)
283 }
284 if err := awaitOutputContains(ctx, local, remoteSession+"@remote"); err != nil {
285 return fmt.Errorf("publish remote content on local transport: %w", err)
286 }
287 if !local.probe.contains(previewMarker) {
288 return errors.New("published remote content omitted preview marker")
289 }
290 if err := assertNoDirectHandoff(local); err != nil {
291 return err
292 }
293 if err := assertNoRemoteChrome(local.probe, remoteSession); err != nil {
294 return fmt.Errorf("remote content chrome: %w", err)
295 }
296 // Split the marker so the terminal's echoed command line cannot satisfy the
297 // awaited-output assertion before the shell executes it.
298 if err := sendInputAndAwait(ctx, local, "printf '%s%s' 'VEV_REMOTE_PICKER' '_INPUT_OK'", inputMarker); err != nil {
299 return fmt.Errorf("continued input on local transport: %w", err)
300 }
301 if err := assertNoDirectHandoff(local); err != nil {
302 return err
303 }
304
305 // Reopen the picker through the same local connection while its owner is a
306 // remote view. The navigation shell must remain local and retain both sides
307 // of the unified session list.
308 if err := sendRawInput(local, "\x1b "); err != nil {
309 return fmt.Errorf("reopen command palette from remote content: %w", err)
310 }
311 if err := awaitOutputContains(ctx, local, "Commands"); err != nil {
312 return fmt.Errorf("await reopened command palette: %w", err)
313 }
314 if err := sendRawInput(local, "SSP\r"); err != nil {
315 return fmt.Errorf("reopen session picker from remote content: %w", err)
316 }
317 if err := awaitOutputContains(ctx, local, remoteSession+"@remote"); err != nil {
318 return fmt.Errorf("await reopened remote picker row: %w", err)
319 }
320 afterRows := capturePickerRows(local.probe)
321 if err := assertUnifiedPickerRows(afterRows, localSession, remoteSession); err != nil {
322 return fmt.Errorf("capture unified picker rows after remote selection: %w", err)
323 }
324 if err := assertNoDirectHandoff(local); err != nil {
325 return err
326 }
327 if err := sendRawInput(local, "\x1b"); err != nil {
328 return fmt.Errorf("close reopened session picker: %w", err)
329 }
330 return nil
331}
332
333func runRestartResumePhase(ctx context.Context, target string, catalog ports.RemoteCatalogClient, preview ports.RemotePreviewClient) (domain.RemoteSessionTarget, error) {
334 if err := runRemoteCommand(ctx, target, "vev", "kill", "--daemon"); err != nil {
335 return domain.RemoteSessionTarget{}, fmt.Errorf("stop remote daemon: %w", err)
336 }
337 if err := runRemoteCommand(ctx, target, "vev", "--daemon-launcher"); err != nil {
338 return domain.RemoteSessionTarget{}, fmt.Errorf("restart remote daemon: %w", err)
339 }
340 resumedCatalog, err := waitForSession(ctx, catalog, target, sessionName, "up")
341 if err != nil {
342 return domain.RemoteSessionTarget{}, fmt.Errorf("resumed catalog: %w", err)
343 }
344 resumedTarget, err := targetForSession(target, resumedCatalog, false)
345 if err != nil {
346 return domain.RemoteSessionTarget{}, fmt.Errorf("resumed target: %w", err)
347 }
348 resumed, err := openTransport(ctx, ports.RemoteTransportStdio, target)
349 if err != nil {
350 return domain.RemoteSessionTarget{}, fmt.Errorf("open restart stdio transport: %w", err)
351 }
352 defer func() { _ = resumed.Close() }()
353 if err := attachAndCheck(ctx, resumed, ports.Hello{
354 Intent: ports.IntentResume, Name: sessionName, Env: localEnvironment(), Cwd: "/tmp/local-cwd",
355 TermEnv: "xterm-256color", TrueColor: true, RemoteTarget: &resumedTarget,
356 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
357 }); err != nil {
358 return domain.RemoteSessionTarget{}, fmt.Errorf("exact restart resume: %w", err)
359 }
360 if _, err := preview.Preview(ctx, resumedTarget, 40, 8); err != nil {
361 return domain.RemoteSessionTarget{}, fmt.Errorf("restart preview: %w", err)
362 }
363 if err := touchRemoteFlag(ctx, target, "/tmp/vev-slow-preview"); err != nil {
364 return domain.RemoteSessionTarget{}, fmt.Errorf("enable slow preview: %w", err)
365 }
366 slowCtx, slowCancel := context.WithTimeout(ctx, 150*time.Millisecond)
367 _, slowErr := preview.Preview(slowCtx, resumedTarget, 40, 8)
368 slowCancel()
369 if err := touchRemoteFlag(ctx, target, "/tmp/vev-slow-preview-off"); err != nil {
370 return domain.RemoteSessionTarget{}, fmt.Errorf("disable slow preview: %w", err)
371 }
372 if slowErr == nil {
373 return domain.RemoteSessionTarget{}, errors.New("slow preview did not honor cancellation")
374 }
375 if err := detach(ctx, resumed); err != nil {
376 return domain.RemoteSessionTarget{}, fmt.Errorf("stop resumed session: %w", err)
377 }
378 return resumedTarget, nil
379}
380
381func runUDPPhase(ctx context.Context, target string) error {
382 udp, err := openTransport(ctx, ports.RemoteTransportUDP, target)
383 if err != nil {
384 return fmt.Errorf("open UDP transport: %w", err)
385 }
386 defer func() { _ = udp.Close() }()
387 if err := attachAndCheck(ctx, udp, ports.Hello{
388 Intent: ports.IntentNew, Name: "udp-harness", Env: localEnvironment(), Cwd: "/tmp/local-cwd",
389 TermEnv: "xterm-256color", TrueColor: true,
390 }); err != nil {
391 return fmt.Errorf("UDP attach: %w", err)
392 }
393 if err := detach(ctx, udp); err != nil {
394 return fmt.Errorf("UDP detach: %w", err)
395 }
396 return nil
397}
398
399func runCatalogFailurePhase(ctx context.Context, target string, catalog ports.RemoteCatalogClient) error {
400 if err := touchRemoteFlag(ctx, target, "/tmp/vev-bad-catalog"); err != nil {
401 return fmt.Errorf("enable mismatched catalog: %w", err)
402 }
403 if _, err := catalog.List(ctx, target); err == nil {
404 return errors.New("version-mismatched catalog unexpectedly decoded")
405 }
406 if err := touchRemoteFlag(ctx, target, "/tmp/vev-bad-catalog-off"); err != nil {
407 return fmt.Errorf("disable mismatched catalog: %w", err)
408 }
409 return nil
410}
411
412func runStaleTargetPhase(ctx context.Context, target string, catalog ports.RemoteCatalogClient, staleTarget domain.RemoteSessionTarget) error {
413 if err := runRemoteCommand(ctx, target, "vev", "kill", sessionName); err != nil {
414 return fmt.Errorf("destroy session for stale-target test: %w", err)
415 }
416 if err := waitForMissingSession(ctx, catalog, target, sessionName); err != nil {
417 return fmt.Errorf("wait for destroyed session: %w", err)
418 }
419 stale, err := openTransport(ctx, ports.RemoteTransportStdio, target)
420 if err != nil {
421 return fmt.Errorf("open stale-target transport: %w", err)
422 }
423 defer func() { _ = stale.Close() }()
424 if err := expectNoSuchTarget(ctx, stale, staleTarget); err != nil {
425 return fmt.Errorf("stale target fence: %w", err)
426 }
427 return nil
428}
429
430func removedCatalogTabID(before, after []ports.RemoteCatalogTab) (string, bool) {
431 remaining := make(map[string]struct{}, len(after))
432 for _, tab := range after {
433 remaining[tab.ID] = struct{}{}
434 }
435 var removed string
436 for _, tab := range before {
437 if _, exists := remaining[tab.ID]; exists {
438 continue
439 }
440 if removed != "" {
441 return "", false
442 }
443 removed = tab.ID
444 }
445 return removed, removed != ""
446}
447
448func localEnvironment() []string {
449 return []string{
450 "VEV_TEST_ENV=local",
451 "XDG_RUNTIME_DIR=/run/local",
452 "WAYLAND_DISPLAY=wayland-local",
453 "TERM=xterm-256color",
454 "COLORTERM=truecolor",
455 "TERM_PROGRAM=local-harness",
456 }
457}
458
459type harnessFrame struct {
460 frame ports.Frame
461 err error
462}
463
464type harnessTransport struct {
465 ports.Transport
466 probe *visualProbe
467 frames chan harnessFrame
468 queued []harnessFrame
469 stop chan struct{}
470 closeOnce sync.Once
471 closeErr error
472}
473
474func newHarnessTransport(transport ports.Transport) *harnessTransport {
475 h := &harnessTransport{
476 Transport: transport,
477 probe: newVisualProbe(domain.Size{Cols: defaultProbeCols, Rows: defaultProbeRows}),
478 frames: make(chan harnessFrame, 16),
479 stop: make(chan struct{}),
480 }
481 go h.receiveLoop()
482 return h
483}
484
485func (h *harnessTransport) receiveLoop() {
486 for {
487 frame, err := h.Transport.Recv()
488 select {
489 case h.frames <- harnessFrame{frame: frame, err: err}:
490 case <-h.stop:
491 return
492 }
493 if err != nil {
494 return
495 }
496 }
497}
498
499func (h *harnessTransport) Close() error {
500 h.closeOnce.Do(func() {
501 close(h.stop)
502 h.closeErr = h.Transport.Close()
503 })
504 return h.closeErr
505}
506
507func openTransport(ctx context.Context, mode ports.RemoteTransportMode, target string) (*harnessTransport, error) {
508 var transport ports.Transport
509 var err error
510 switch mode {
511 case ports.RemoteTransportStdio:
512 transport, err = sshstdio.DialContext(ctx, target, "", slog.New(slog.DiscardHandler))
513 case ports.RemoteTransportUDP:
514 dialer := dgram.NewRemoteDialerWithLogger(target, "", slog.New(slog.DiscardHandler))
515 dialer.BootstrapTimeout = 8 * time.Second
516 dialer.ProbeTimeout = 3 * time.Second
517 transport, err = dialer.Dial(ctx)
518 default:
519 return nil, fmt.Errorf("unsupported transport %q", mode)
520 }
521 if err != nil {
522 return nil, err
523 }
524 h := newHarnessTransport(transport)
525 h.probe.configure(string(mode))
526 if artifact := harnessArtifactFromContext(ctx); artifact != nil {
527 artifact.registerProbe(h.probe)
528 }
529 return h, nil
530}
531
532const commandStderrLimit = 4 << 10
533
534type boundedStderr struct {
535 bytes.Buffer
536 truncated bool
537}
538
539func (b *boundedStderr) Write(p []byte) (int, error) {
540 written := len(p)
541 remaining := commandStderrLimit - b.Len()
542 if remaining <= 0 {
543 b.truncated = b.truncated || written != 0
544 return written, nil
545 }
546 if len(p) > remaining {
547 p = p[:remaining]
548 b.truncated = true
549 }
550 _, _ = b.Buffer.Write(p)
551 return written, nil
552}
553
554func (b *boundedStderr) String() string {
555 stderr := b.Buffer.String()
556 if b.truncated {
557 stderr += "\n[stderr truncated]"
558 }
559 return stderr
560}
561
562func commandErrorWithStderr(err error, stderr string) error {
563 if stderr = strings.TrimSpace(stderr); stderr == "" {
564 return err
565 }
566 return fmt.Errorf("%w: stderr: %s", err, stderr)
567}
568
569func startLocalDaemon(ctx context.Context) (*exec.Cmd, error) {
570 cmd := exec.CommandContext(ctx, "vev", "--daemon")
571 stderr := &boundedStderr{}
572 cmd.Stdout = io.Discard
573 cmd.Stderr = stderr
574 if err := cmd.Start(); err != nil {
575 return nil, err
576 }
577
578 deadline, cancel := context.WithTimeout(ctx, 5*time.Second)
579 defer cancel()
580 for {
581 transport, err := ipc.DialContext(deadline, ipc.SocketDir())
582 if err == nil {
583 _ = transport.Close()
584 return cmd, nil
585 }
586 select {
587 case <-deadline.Done():
588 stopLocalProcess(cmd)
589 return nil, commandErrorWithStderr(deadline.Err(), stderr.String())
590 case <-time.After(25 * time.Millisecond):
591 }
592 }
593}
594
595func stopLocalDaemon(cmd *exec.Cmd) {
596 cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second)
597 defer cancel()
598 _ = runLocalCommand(cleanup, "vev", "kill", "--daemon")
599 stopLocalProcess(cmd)
600}
601
602func stopLocalProcess(cmd *exec.Cmd) {
603 if cmd == nil || cmd.Process == nil {
604 return
605 }
606 _ = cmd.Process.Kill()
607 _ = cmd.Wait()
608}
609
610func runLocalCommand(ctx context.Context, name string, args ...string) error {
611 return runCapturedCommand(ctx, name, args...)
612}
613
614func runCapturedCommand(ctx context.Context, name string, args ...string) error {
615 cmd := exec.CommandContext(ctx, name, args...)
616 stderr := &boundedStderr{}
617 cmd.Stdout = io.Discard
618 cmd.Stderr = stderr
619 if err := cmd.Run(); err != nil {
620 return commandErrorWithStderr(err, stderr.String())
621 }
622 return nil
623}
624
625func openLocalTransport(ctx context.Context) (*harnessTransport, error) {
626 transport, err := ipc.DialContext(ctx, ipc.SocketDir())
627 if err != nil {
628 return nil, err
629 }
630 h := newHarnessTransport(transport)
631 h.probe.configure("local")
632 if artifact := harnessArtifactFromContext(ctx); artifact != nil {
633 artifact.registerProbe(h.probe)
634 }
635 return h, nil
636}
637
638func attachAndCheck(ctx context.Context, tr *harnessTransport, hello ports.Hello) error {
639 hello.Version = ports.ProtocolVersion
640 hello.Size = domain.Size{Cols: 80, Rows: 24}
641 if hello.EnvironmentPolicy == 0 {
642 hello.EnvironmentPolicy = ports.EnvironmentPolicyClientOwned
643 }
644 payload := ports.MarshalHello(hello)
645 if err := tr.Send(ports.Frame{Type: ports.MsgHello, Payload: payload}); err != nil {
646 return err
647 }
648 frame, err := receiveWithTimeout(ctx, tr)
649 if err != nil {
650 return err
651 }
652 if frame.Type == ports.MsgError {
653 message, decodeErr := ports.UnmarshalErrorMsg(frame.Payload)
654 if decodeErr != nil {
655 return decodeErr
656 }
657 return fmt.Errorf("remote error code %d", message.Code)
658 }
659 if frame.Type != ports.MsgWelcome {
660 return fmt.Errorf("welcome frame type %d", frame.Type)
661 }
662 tr.probe.recordControl(probeEventWelcome)
663 if err := tr.Send(ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(ports.Theme{TrueColor: hello.TrueColor, SchemeKnown: true})}); err != nil {
664 return fmt.Errorf("send initial theme: %w", err)
665 }
666 return nil
667}
668
669func sendRawInput(tr *harnessTransport, data string) error {
670 return tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte(data)})})
671}
672
673func processOutputFrame(tr *harnessTransport, frame ports.Frame) error {
674 output, err := ports.UnmarshalOutput(frame.Payload)
675 if err != nil {
676 return err
677 }
678 result := tr.probe.apply(output)
679 if !result.Accepted {
680 if !tr.probe.resetPending {
681 if err := tr.Send(ports.Frame{Type: ports.MsgOutputResetRequest, Payload: ports.MarshalOutputResetRequest(ports.OutputResetRequest{})}); err != nil {
682 return err
683 }
684 tr.probe.resetPending = true
685 tr.probe.recordControl(probeEventOutputReset)
686 }
687 return nil
688 }
689 if !result.StateBearing {
690 return nil
691 }
692 payload, err := ports.MarshalAck(result.Ack)
693 if err != nil {
694 return err
695 }
696 if err := tr.Send(ports.Frame{Type: ports.MsgAck, Payload: payload}); err != nil {
697 return err
698 }
699 tr.probe.resetPending = false
700 tr.probe.markAcked(result)
701 return nil
702}
703
704func awaitOutputContains(ctx context.Context, tr *harnessTransport, want string) error {
705 deadline, cancel := context.WithTimeout(ctx, 5*time.Second)
706 defer cancel()
707 visible := tr.probe.contains(want)
708 if visible {
709 return nil
710 }
711 for {
712 frame, err := receiveWithTimeout(deadline, tr)
713 if err != nil {
714 return err
715 }
716 if frame.Type != ports.MsgOutput {
717 continue
718 }
719 if err := processOutputFrame(tr, frame); err != nil {
720 return err
721 }
722 nowVisible := tr.probe.contains(want)
723 if !visible && nowVisible {
724 return nil
725 }
726 visible = nowVisible
727 }
728}
729
730func awaitRemotePickerPreview(ctx context.Context, tr *harnessTransport, marker string) error {
731 deadline, cancel := context.WithTimeout(ctx, 5*time.Second)
732 defer cancel()
733 sawLoading := false
734 markerVisible := tr.probe.contains(marker)
735 for {
736 frame, err := receiveWithTimeout(deadline, tr)
737 if err != nil {
738 return fmt.Errorf("no rendered remote preview after %d output frames (loading=%t): %w", tr.probe.outputFrames, sawLoading, err)
739 }
740 if frame.Type != ports.MsgOutput {
741 continue
742 }
743 if err := processOutputFrame(tr, frame); err != nil {
744 return err
745 }
746 text := tr.probe.text()
747 sawLoading = sawLoading || strings.Contains(text, "loading remote preview")
748 if strings.Contains(text, "remote preview unavailable") {
749 return errors.New("remote preview unavailable")
750 }
751 nowMarkerVisible := strings.Contains(text, marker)
752 if !markerVisible && nowMarkerVisible {
753 return nil
754 }
755 markerVisible = nowMarkerVisible
756 }
757}
758
759func remotePreviewContains(preview ports.RemotePreview, want string) bool {
760 var text strings.Builder
761 for _, cell := range preview.Cells {
762 if !cell.Continuation {
763 text.WriteRune(cell.Rune)
764 }
765 }
766 return strings.Contains(text.String(), want)
767}
768
769func sendInputAndAwait(ctx context.Context, tr *harnessTransport, command, want string) error {
770 if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte(command + "\n")})}); err != nil {
771 return err
772 }
773 deadline, cancel := context.WithTimeout(ctx, 5*time.Second)
774 defer cancel()
775 visible := tr.probe.contains(want)
776 for {
777 frame, err := receiveWithTimeout(deadline, tr)
778 if err != nil {
779 return fmt.Errorf("waiting for marker after %d output frames/%d bytes: %w", tr.probe.outputFrames, tr.probe.outputBytes, err)
780 }
781 if frame.Type != ports.MsgOutput {
782 continue
783 }
784 if err := processOutputFrame(tr, frame); err != nil {
785 return fmt.Errorf("process output: %w", err)
786 }
787 nowVisible := tr.probe.contains(want)
788 if !visible && nowVisible {
789 return nil
790 }
791 visible = nowVisible
792 }
793}
794
795func sendCommand(ctx context.Context, tr *harnessTransport, slug string) error {
796 id := requestID.Add(1)
797 request := ports.CommandRequest{
798 Version: ports.ProtocolVersion,
799 RequestID: id,
800 Attached: true,
801 Slug: slug,
802 }
803 payload, err := ports.MarshalCommandRequest(request)
804 if err != nil {
805 return err
806 }
807 if err := tr.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
808 return err
809 }
810 deadline, cancel := context.WithTimeout(ctx, 5*time.Second)
811 defer cancel()
812 for {
813 frame, err := receiveWithTimeout(deadline, tr)
814 if err != nil {
815 return err
816 }
817 if frame.Type == ports.MsgOutput {
818 if err := processOutputFrame(tr, frame); err != nil {
819 return err
820 }
821 continue
822 }
823 if frame.Type != ports.MsgCommandResult {
824 continue
825 }
826 result, err := ports.UnmarshalCommandResult(frame.Payload)
827 if err != nil {
828 return err
829 }
830 if !result.OK {
831 return fmt.Errorf("remote command failed (code %d): %s", result.Code, result.Text)
832 }
833 return nil
834 }
835}
836
837func detach(ctx context.Context, tr *harnessTransport) error {
838 if err := tr.Send(ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})}); err != nil {
839 return err
840 }
841 deadline, cancel := context.WithTimeout(ctx, 5*time.Second)
842 defer cancel()
843 for {
844 frame, err := receiveWithTimeout(deadline, tr)
845 if err != nil {
846 return err
847 }
848 if frame.Type == ports.MsgDetached {
849 return nil
850 }
851 }
852}
853
854func expectNoSuchTarget(ctx context.Context, tr *harnessTransport, target domain.RemoteSessionTarget) error {
855 target.Endpoint = targetEndpoint(target.Endpoint)
856 target.DisplayOrigin = "remote"
857 target.SessionName = sessionName
858 if err := target.Validate(); err != nil {
859 return err
860 }
861 hello := ports.Hello{
862 Version: ports.ProtocolVersion,
863 Intent: ports.IntentResume,
864 Name: sessionName,
865 Size: domain.Size{Cols: 80, Rows: 24},
866 TermEnv: "xterm-256color",
867 TrueColor: true,
868 Env: localEnvironment(),
869 Cwd: "/tmp/local-cwd",
870 RemoteTarget: &target,
871 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
872 }
873 payload := ports.MarshalHello(hello)
874 if err := tr.Send(ports.Frame{Type: ports.MsgHello, Payload: payload}); err != nil {
875 return err
876 }
877 frame, err := receiveWithTimeout(ctx, tr)
878 if err != nil {
879 return err
880 }
881 if frame.Type != ports.MsgError {
882 return fmt.Errorf("stale target returned frame type %d", frame.Type)
883 }
884 message, err := ports.UnmarshalErrorMsg(frame.Payload)
885 if err != nil {
886 return err
887 }
888 if message.Code != ports.ErrNoSuchTarget {
889 return fmt.Errorf("stale target returned error code %d", message.Code)
890 }
891 return nil
892}
893
894func targetForSession(endpoint string, session ports.RemoteCatalogSession, stopped bool) (domain.RemoteSessionTarget, error) {
895 tabs := ports.CatalogTabs(session)
896 if len(tabs) == 0 {
897 return domain.RemoteSessionTarget{}, errors.New("catalog session has no tabs")
898 }
899 tab := tabs[0]
900 if !stopped && session.ActiveTabID != "" {
901 for _, candidate := range tabs {
902 if candidate.ID == session.ActiveTabID {
903 tab = candidate
904 break
905 }
906 }
907 }
908 target := domain.RemoteSessionTarget{
909 Endpoint: endpoint,
910 DisplayOrigin: "remote",
911 LifecycleID: session.LifecycleID,
912 SessionName: session.Name,
913 Stopped: stopped,
914 }
915 if stopped {
916 target.StoppedTab = domain.NewStableTabSelector(domain.TabStableID(tab.ID))
917 } else {
918 target.LiveTabID = domain.TabStableID(tab.ID)
919 }
920 if err := target.Validate(); err != nil {
921 return domain.RemoteSessionTarget{}, err
922 }
923 return target, nil
924}
925
926func targetEndpoint(endpoint string) string {
927 if endpoint == "" {
928 return defaultTarget
929 }
930 return endpoint
931}
932
933func waitForSession(ctx context.Context, catalog ports.RemoteCatalogClient, endpoint, name, state string) (ports.RemoteCatalogSession, error) {
934 return waitForSessionWithTabCount(ctx, catalog, endpoint, name, state, 0)
935}
936
937func waitForSessionWithTabCount(ctx context.Context, catalog ports.RemoteCatalogClient, endpoint, name, state string, tabs int) (ports.RemoteCatalogSession, error) {
938 deadline, cancel := context.WithTimeout(ctx, 10*time.Second)
939 defer cancel()
940 for {
941 result, err := catalog.List(deadline, endpoint)
942 if err == nil {
943 for _, session := range result.Sessions {
944 if session.Name == name && session.State == state && (tabs == 0 || len(ports.CatalogTabs(session)) == tabs) {
945 return session, nil
946 }
947 }
948 }
949 select {
950 case <-deadline.Done():
951 if err != nil {
952 return ports.RemoteCatalogSession{}, err
953 }
954 return ports.RemoteCatalogSession{}, fmt.Errorf("session %q did not reach state %q", name, state)
955 case <-time.After(100 * time.Millisecond):
956 }
957 }
958}
959
960func waitForMissingSession(ctx context.Context, catalog ports.RemoteCatalogClient, endpoint, name string) error {
961 deadline, cancel := context.WithTimeout(ctx, 10*time.Second)
962 defer cancel()
963 for {
964 result, err := catalog.List(deadline, endpoint)
965 if err == nil {
966 found := false
967 for _, session := range result.Sessions {
968 found = found || session.Name == name
969 }
970 if !found {
971 return nil
972 }
973 }
974 select {
975 case <-deadline.Done():
976 return errors.New("destroyed session remained in catalog")
977 case <-time.After(100 * time.Millisecond):
978 }
979 }
980}
981
982func touchRemoteFlag(ctx context.Context, target, path string) error {
983 if strings.HasSuffix(path, "-off") {
984 path = strings.TrimSuffix(path, "-off")
985 return runRemoteCommand(ctx, target, "rm", "-f", path)
986 }
987 return runRemoteCommand(ctx, target, "touch", path)
988}
989
990func runRemoteCommand(ctx context.Context, target string, command ...string) error {
991 spec := sshstdio.BuildCommandForRemoteCommand(target, command...)
992 return runCapturedCommand(ctx, spec.Path, spec.Args...)
993}
994
995func assertNoDirectHandoff(tr *harnessTransport) error {
996 if tr == nil || tr.probe == nil {
997 return errors.New("local picker transport has no visual probe")
998 }
999 // Inspect only the frames already queued at assertion start, retaining
1000 // non-handoff frames for the next receiveWithTimeout call.
1001 queued := len(tr.frames)
1002 for range queued {
1003 outcome := <-tr.frames
1004 if outcome.frame.Type == ports.MsgAttachTarget {
1005 tr.probe.recordIncoming(outcome.frame)
1006 } else {
1007 tr.queued = append(tr.queued, outcome)
1008 }
1009 }
1010 if tr.probe.unexpectedHandoffs != 0 {
1011 return fmt.Errorf("picker emitted %d direct MsgAttachTarget handoff frame(s)", tr.probe.unexpectedHandoffs)
1012 }
1013 return nil
1014}
1015
1016func receiveWithTimeout(ctx context.Context, tr *harnessTransport) (ports.Frame, error) {
1017 if len(tr.queued) != 0 {
1018 outcome := tr.queued[0]
1019 tr.queued = tr.queued[1:]
1020 if outcome.err == nil {
1021 tr.probe.recordIncoming(outcome.frame)
1022 }
1023 return outcome.frame, outcome.err
1024 }
1025 select {
1026 case <-ctx.Done():
1027 return ports.Frame{}, ctx.Err()
1028 case outcome := <-tr.frames:
1029 if outcome.err == nil {
1030 tr.probe.recordIncoming(outcome.frame)
1031 }
1032 return outcome.frame, outcome.err
1033 }
1034}