diff --git a/backend/internal/terminal/attachment.go b/backend/internal/terminal/attachment.go new file mode 100644 index 000000000..682f05ebc --- /dev/null +++ b/backend/internal/terminal/attachment.go @@ -0,0 +1,312 @@ +package terminal + +import ( + "context" + "errors" + "io" + "log/slog" + "sync" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +// PTYSource is what a terminal needs from the runtime: the argv that attaches a +// PTY to a session's pane, and a liveness check used to decide whether a dropped +// PTY should be re-attached or treated as a clean exit. The Zellij runtime adapter +// satisfies this via AttachCommand/IsAlive; the interface lives here, next to its +// only consumer, so terminal does not depend on a concrete adapter. +type PTYSource interface { + AttachCommand(handle ports.RuntimeHandle) ([]string, error) + IsAlive(ctx context.Context, handle ports.RuntimeHandle) (bool, error) +} + +// ptyProcess is a started PTY-backed attach process. It is the injection seam +// that keeps the attach loop testable without a real process: unit tests supply +// a scripted in-memory implementation; production uses a creack/pty-backed one +// (see pty_unix.go). +type ptyProcess interface { + io.ReadWriteCloser + Resize(rows, cols uint16) error +} + +// spawnFunc starts a PTY for argv, sized rows×cols from the start (zero means +// no size was recorded yet — kernel default). Spawning at the client's grid +// matters: the attach process reads the tty size once at startup, and sizing +// the PTY only after exec relies on SIGWINCH delivery that can race the +// process installing its handler — a missed signal leaves the zellij client +// laid out for a stale size. ctx cancellation must terminate the process. +type spawnFunc func(ctx context.Context, argv []string, rows, cols uint16) (ptyProcess, error) + +// reattach policy: a PTY that drops is re-attached while the underlying Zellij +// session is still alive, up to maxReattach consecutive failures. An attach that +// survived longer than reattachResetGrace before dropping resets the counter, so +// a long-lived pane that blips recovers but a tight crash-loop gives up. +const ( + defaultMaxReattach = 5 + defaultReattachResetTime = 5 * time.Second +) + +// attachment is ONE client's hold on a pane: a private `zellij attach` PTY +// spawned per mux open, streaming to a single sink. Zellij is the multiplexer — +// it owns the session's screen state and scrollback, and answers every fresh +// attach with its full init handshake (alt screen, SGR mouse tracking, +// bracketed paste) followed by a faithful repaint. That handshake is why the +// PTY is per-client and there is no server-side replay buffer: a byte ring +// can replay recent output, but the one-time mode negotiation at the head of +// the stream scrolls out of any bounded buffer, leaving late subscribers +// without mouse reporting (wheel scroll dead). A fresh attach per client makes +// Zellij re-send it, every time, by construction. +// +// onData must not block: the WS layer funnels frames onto its own buffered +// writer. onExit fires at most once, when the attach loop gives up (runtime +// dead, attach failure cap) — never on close(). +type attachment struct { + id string + handle ports.RuntimeHandle + src PTYSource + spawn spawnFunc + log *slog.Logger + onData func(data []byte) + onExit func() + + maxReattach int + resetGrace time.Duration + + mu sync.Mutex + pty ptyProcess + rows uint16 // last size the client asked for; re-applied on every attach + cols uint16 + closed bool + exited bool +} + +func newAttachment(id string, handle ports.RuntimeHandle, src PTYSource, spawn spawnFunc, onData func([]byte), onExit func(), log *slog.Logger) *attachment { + if log == nil { + log = slog.Default() + } + if onData == nil { + onData = func([]byte) {} + } + return &attachment{ + id: id, + handle: handle, + src: src, + spawn: spawn, + log: log, + onData: onData, + onExit: onExit, + maxReattach: defaultMaxReattach, + resetGrace: defaultReattachResetTime, + } +} + +// run drives attach → read-loop → re-attach until the pane exits cleanly, the +// attachment is closed, or ctx is cancelled. It is started once per attachment. +func (a *attachment) run(ctx context.Context) { + failures := 0 + for { + if a.isClosed() || ctx.Err() != nil { + return + } + + // Gate EVERY attach (including the first) on the runtime actually + // being alive. `zellij attach` resurrects EXITED sessions — re-running + // the serialized agent command — so attaching to a dead handle would + // re-create a runtime the daemon already destroyed, outside lifecycle + // control. A definitive "not alive" is a clean exit. A probe ERROR is + // not proof of death: it retries with backoff up to the same + // consecutive-failure cap as attach failures. + alive, err := a.src.IsAlive(ctx, a.handle) + if err != nil { + failures++ + if failures > a.maxReattach { + a.fail("liveness probe: " + err.Error()) + return + } + if !a.backoff(ctx, failures) { + return + } + continue + } + if !alive { + a.markExited() + return + } + + argv, err := a.src.AttachCommand(a.handle) + if err != nil { + a.fail("attach command: " + err.Error()) + return + } + rows, cols := a.size() + p, err := a.spawn(ctx, argv, rows, cols) + if err != nil { + failures++ + if failures > a.maxReattach { + a.fail("spawn pty: " + err.Error()) + return + } + if !a.backoff(ctx, failures) { + return + } + continue + } + + a.setPTY(p) + start := time.Now() + a.copyOut(p) + _ = p.Close() + + if time.Since(start) >= a.resetGrace { + failures = 0 + } + failures++ + + if failures > a.maxReattach { + a.markExited() + return + } + if !a.backoff(ctx, failures) { + return + } + a.log.Debug("terminal re-attaching", "id", a.id, "failures", failures) + } +} + +// copyOut pumps PTY output to the sink until the PTY closes or errors. +func (a *attachment) copyOut(p ptyProcess) { + buf := make([]byte, 32*1024) + for { + n, err := p.Read(buf) + if n > 0 { + chunk := make([]byte, n) + copy(chunk, buf[:n]) + a.onData(chunk) + } + if err != nil { + return + } + } +} + +// backoff sleeps between attach attempts; false means ctx was cancelled. +// Whether another attempt is warranted at all (liveness, failure cap) is +// decided at the top of the run loop, so a re-attach and a first attach share +// one gate. +func (a *attachment) backoff(ctx context.Context, failures int) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(reattachBackoff(failures)): + return true + } +} + +func reattachBackoff(failures int) time.Duration { + d := time.Duration(failures) * 200 * time.Millisecond + if d > time.Second { + d = time.Second + } + return d +} + +// write sends client keystrokes to the PTY. It is a no-op if no PTY is attached. +func (a *attachment) write(p []byte) error { + a.mu.Lock() + pty := a.pty + a.mu.Unlock() + if pty == nil { + return errors.New("terminal: no active pty") + } + _, err := pty.Write(p) + return err +} + +// resize records the client's grid and applies it to the live PTY. The size is +// remembered so an attach that is still in flight (or a later re-attach) starts +// at the client's grid instead of the kernel default — the open frame's +// cols/rows land here before the PTY exists. +func (a *attachment) resize(rows, cols uint16) error { + a.mu.Lock() + a.rows, a.cols = rows, cols + pty := a.pty + a.mu.Unlock() + if pty == nil { + return nil + } + return pty.Resize(rows, cols) +} + +// size returns the client's last requested grid (zero before the first +// open/resize recorded one). The spawn path reads it so the PTY starts at the +// client's grid instead of the kernel default. +func (a *attachment) size() (rows, cols uint16) { + a.mu.Lock() + defer a.mu.Unlock() + return a.rows, a.cols +} + +// setPTY publishes a freshly attached PTY and replays the client's last +// requested size onto it (see resize) — the spawn already started at the size +// read in run, but a resize frame can land between that read and registration +// here; the replay (Setsize + explicit WINCH) converges the late case. +func (a *attachment) setPTY(p ptyProcess) { + a.mu.Lock() + a.pty = p + rows, cols := a.rows, a.cols + a.mu.Unlock() + if rows > 0 && cols > 0 { + _ = p.Resize(rows, cols) + } +} + +// close detaches this client: stop re-attaching and kill the attach PTY. It +// never touches the Zellij session itself, which the zellij server keeps alive +// for other clients. +func (a *attachment) close() { + a.mu.Lock() + if a.closed { + a.mu.Unlock() + return + } + a.closed = true + pty := a.pty + a.pty = nil + a.mu.Unlock() + if pty != nil { + _ = pty.Close() + } +} + +func (a *attachment) isClosed() bool { + a.mu.Lock() + defer a.mu.Unlock() + return a.closed +} + +func (a *attachment) isExited() bool { + a.mu.Lock() + defer a.mu.Unlock() + return a.exited +} + +// markExited flips the attachment to exited and fires onExit once. +func (a *attachment) markExited() { + a.mu.Lock() + if a.exited { + a.mu.Unlock() + return + } + a.exited = true + a.mu.Unlock() + if a.onExit != nil { + a.onExit() + } +} + +// fail reports an unrecoverable attach error as an exit. +func (a *attachment) fail(reason string) { + a.log.Warn("terminal attachment failed", "id", a.id, "reason", reason) + a.markExited() +} diff --git a/backend/internal/terminal/attachment_integration_test.go b/backend/internal/terminal/attachment_integration_test.go new file mode 100644 index 000000000..10a82c0b4 --- /dev/null +++ b/backend/internal/terminal/attachment_integration_test.go @@ -0,0 +1,142 @@ +//go:build !windows + +package terminal + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/adapters/runtime/zellij" + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +// TestAttachmentStreamsRealZellijPane attaches a real PTY to a real Zellij +// session and asserts output streams back, then that killing the pane stops the +// attachment without a re-attach storm. Skipped when Zellij is unavailable. +func TestAttachmentStreamsRealZellijPane(t *testing.T) { + zellijBin, err := exec.LookPath("zellij") + if err != nil { + t.Skip("zellij unavailable") + } + + name := "ao-term-it-" + strconv.Itoa(os.Getpid()) + socketDir := filepath.Join("/tmp", name+"-socket") + if err := os.MkdirAll(socketDir, 0o755); err != nil { + t.Fatalf("mkdir socket dir: %v", err) + } + rt := zellij.New(zellij.Options{Binary: zellijBin, SocketDir: socketDir, ConfigDir: t.TempDir(), Timeout: 5 * time.Second}) + handle, err := rt.Create(context.Background(), ports.RuntimeConfig{ + SessionID: domain.SessionID(name), + WorkspacePath: t.TempDir(), + Argv: []string{"printf", "AO_READY\\n"}, + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + t.Cleanup(func() { _ = rt.Destroy(context.Background(), handle) }) + + var got safeBytes + a := newAttachment(name, handle, rt, defaultSpawn, got.add, nil, testLogger()) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + // Type a unique marker and expect it echoed back through the PTY. + eventually(t, 3*time.Second, func() bool { return a.write([]byte("echo AO_MARKER_42\n")) == nil }) + eventually(t, 5*time.Second, func() bool { return strings.Contains(got.string(), "AO_MARKER_42") }) + + // A fresh attach must carry zellij's init handshake: alt screen + SGR mouse + // tracking. This is the whole point of per-client attach — late clients see + // the mode negotiation, so wheel events are forwarded as mouse reports + // instead of going dead. + eventually(t, 5*time.Second, func() bool { + out := got.string() + return strings.Contains(out, "\x1b[?1049h") && strings.Contains(out, "\x1b[?1006h") + }) + + // Kill the session: the attachment must observe it as gone and not re-attach. + if err := rt.Destroy(context.Background(), handle); err != nil { + t.Fatalf("Destroy: %v", err) + } + eventually(t, 5*time.Second, func() bool { return a.isExited() }) +} + +// TestAttachmentReattachAdoptsNewSize is the end-to-end regression for the +// stale-size desync: client A holds the session at one grid, detaches, and +// client B immediately attaches at a different grid (the frontend's +// remount/reconnect flow). B's zellij client must adopt B's size — the inner +// pane's tty must report it — not stay laid out for A's. This is where the +// spawn-at-size + explicit-WINCH + SIGTERM-detach fixes meet a real zellij. +func TestAttachmentReattachAdoptsNewSize(t *testing.T) { + zellijBin, err := exec.LookPath("zellij") + if err != nil { + t.Skip("zellij unavailable") + } + + name := "ao-term-size-it-" + strconv.Itoa(os.Getpid()) + socketDir := filepath.Join("/tmp", name+"-socket") + if err := os.MkdirAll(socketDir, 0o755); err != nil { + t.Fatalf("mkdir socket dir: %v", err) + } + rt := zellij.New(zellij.Options{Binary: zellijBin, SocketDir: socketDir, ConfigDir: t.TempDir(), Timeout: 5 * time.Second}) + handle, err := rt.Create(context.Background(), ports.RuntimeConfig{ + SessionID: domain.SessionID(name), + WorkspacePath: t.TempDir(), + Argv: []string{"printf", "AO_READY\\n"}, + }) + if err != nil { + t.Fatalf("Create: %v", err) + } + t.Cleanup(func() { _ = rt.Destroy(context.Background(), handle) }) + + attachAt := func(rows, cols uint16) (*attachment, *safeBytes, context.CancelFunc) { + var got safeBytes + a := newAttachment(name, handle, rt, defaultSpawn, got.add, nil, testLogger()) + if err := a.resize(rows, cols); err != nil { + t.Fatalf("record size: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + go a.run(ctx) + return a, &got, cancel + } + + // Client A at 115x37: wait for the pane shell, then detach. + a, gotA, cancelA := attachAt(37, 115) + eventually(t, 5*time.Second, func() bool { return a.write([]byte("\n")) == nil }) + eventually(t, 5*time.Second, func() bool { return strings.Contains(gotA.string(), "AO_READY") }) + a.close() + cancelA() + + // Client B re-attaches immediately at 148x40 — no settle gap, same as the + // frontend reconnecting. The inner pane must see B's grid (zellij chrome + // shaves a couple rows/cols, so assert the reported cols land near 148 and + // far from 115). + b, gotB, cancelB := attachAt(40, 148) + defer cancelB() + defer b.close() + + eventually(t, 5*time.Second, func() bool { return b.write([]byte("echo SIZE:$(stty size)\n")) == nil }) + eventually(t, 10*time.Second, func() bool { + out := gotB.string() + i := strings.LastIndex(out, "SIZE:") + if i < 0 { + return false + } + fields := strings.Fields(strings.TrimPrefix(out[i:], "SIZE:")) + if len(fields) < 2 { + return false + } + cols, err := strconv.Atoi(strings.TrimFunc(fields[1], func(r rune) bool { return r < '0' || r > '9' })) + if err != nil { + return false + } + return cols > 130 // B's 148 minus zellij chrome; a stale A-layout reports ≤115 + }) +} diff --git a/backend/internal/terminal/attachment_test.go b/backend/internal/terminal/attachment_test.go new file mode 100644 index 000000000..476d03ac4 --- /dev/null +++ b/backend/internal/terminal/attachment_test.go @@ -0,0 +1,286 @@ +package terminal + +import ( + "context" + "io" + "log/slog" + "testing" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +func testLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } + +func newTestAttachment(src PTYSource, spawn spawnFunc, onData func([]byte), onExit func()) *attachment { + return newAttachment("t1", ports.RuntimeHandle{ID: "t1"}, src, spawn, onData, onExit, testLogger()) +} + +func TestAttachmentStreamsOutputToSink(t *testing.T) { + src := &fakeSource{alive: true} + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + + var sink safeBytes + a := newTestAttachment(src, sp.spawn, sink.add, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + pty.push([]byte("hello")) + eventually(t, time.Second, func() bool { return sink.string() == "hello" }) +} + +func TestAttachmentWriteAndResizeReachPTY(t *testing.T) { + src := &fakeSource{alive: true} + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + a := newTestAttachment(src, sp.spawn, nil, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + eventually(t, time.Second, func() bool { return a.write([]byte("ls\n")) == nil }) + eventually(t, time.Second, func() bool { return string(pty.writtenBytes()) == "ls\n" }) + + if err := a.resize(24, 80); err != nil { + t.Fatalf("resize: %v", err) + } + eventually(t, time.Second, func() bool { + rs := pty.resizeCalls() + return len(rs) == 1 && rs[0] == [2]uint16{24, 80} + }) +} + +// A size requested before the PTY exists (the open frame's cols/rows, or a +// resize racing the attach) must not be lost: the attach applies it the moment +// the PTY is up, instead of leaving the pane at the kernel default grid. +func TestAttachmentAppliesRequestedSizeOnAttach(t *testing.T) { + src := &fakeSource{alive: true} + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + a := newTestAttachment(src, sp.spawn, nil, nil) + + if err := a.resize(30, 100); err != nil { + t.Fatalf("resize before attach: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + eventually(t, time.Second, func() bool { + rs := pty.resizeCalls() + return len(rs) == 1 && rs[0] == [2]uint16{30, 100} + }) +} + +// The PTY must be SPAWNED at the recorded grid, not sized after exec: the +// attach process (zellij) reads the tty size once at startup, and a post-spawn +// TIOCSWINSZ depends on SIGWINCH delivery that can race the client installing +// its handler — a missed signal left the session laid out for the previous +// client's size (the "terminal doesn't repaint after a resize" desync). Also +// covers re-attach: a later resize must reach the NEXT spawn, not the first +// grid forever. +func TestAttachmentSpawnsPTYAtRecordedSize(t *testing.T) { + src := &fakeSource{alive: true} + first := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{first}} + a := newTestAttachment(src, sp.spawn, nil, nil) + a.resetGrace = time.Hour // keep the failure counter deterministic + + if err := a.resize(37, 115); err != nil { + t.Fatalf("resize before attach: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + eventually(t, time.Second, func() bool { + ss := sp.spawnSizes() + return len(ss) == 1 && ss[0] == [2]uint16{37, 115} + }) + + // Client resized, then the PTY dropped: the re-attach spawn must start at + // the latest grid. + if err := a.resize(40, 148); err != nil { + t.Fatalf("resize while attached: %v", err) + } + _ = first.Close() + + eventually(t, time.Second, func() bool { + ss := sp.spawnSizes() + return len(ss) == 2 && ss[1] == [2]uint16{40, 148} + }) +} + +func TestAttachmentSkipsReattachOnCleanExit(t *testing.T) { + src := &fakeSource{alive: true} // alive for the first attach + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + + exited := make(chan struct{}) + a := newTestAttachment(src, sp.spawn, nil, func() { close(exited) }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + eventually(t, time.Second, func() bool { return sp.calls() == 1 }) + src.setAlive(false) // Zellij session gone -> no re-attach + pty.Close() // pane ends + select { + case <-exited: + case <-time.After(time.Second): + t.Fatal("expected exit callback after clean pane exit") + } + if got := sp.calls(); got != 1 { + t.Fatalf("expected exactly one attach, got %d", got) + } +} + +// TestAttachmentNeverAttachesToDeadRuntime covers the resurrection bug: `zellij +// attach` on a killed-but-cached session resurrects it, re-running the agent +// command. An attachment whose runtime probes definitively dead must therefore +// report exited WITHOUT ever spawning an attach PTY — even on the very first +// open (the original code only checked liveness on re-attach). +func TestAttachmentNeverAttachesToDeadRuntime(t *testing.T) { + src := &fakeSource{alive: false} + sp := &fakeSpawner{} + + exited := make(chan struct{}) + a := newTestAttachment(src, sp.spawn, nil, func() { close(exited) }) + + go a.run(context.Background()) + select { + case <-exited: + case <-time.After(time.Second): + t.Fatal("expected exit when runtime is dead before first attach") + } + if got := sp.calls(); got != 0 { + t.Fatalf("attach must never run against a dead runtime, got %d spawns", got) + } +} + +// TestAttachmentRetriesProbeErrorsBeforeAttaching pins the hard rule that a +// failed liveness probe is NOT proof of death: a transient probe error must +// not flip the terminal to exited, and the attach proceeds once the probe +// recovers. +func TestAttachmentRetriesProbeErrorsBeforeAttaching(t *testing.T) { + src := &fakeSource{aliveErr: io.ErrUnexpectedEOF} + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + a := newTestAttachment(src, sp.spawn, nil, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + // While probes error the attachment must neither exit nor attach. + time.Sleep(50 * time.Millisecond) + if a.isExited() { + t.Fatal("probe error must not be treated as runtime death") + } + if got := sp.calls(); got != 0 { + t.Fatalf("attach must wait for a successful probe, got %d spawns", got) + } + + // Probe recovers -> the attach goes through. + src.setAliveResult(true, nil) + eventually(t, 2*time.Second, func() bool { return sp.calls() == 1 }) + if a.isExited() { + t.Fatal("attachment exited despite a live runtime") + } +} + +func TestAttachmentReattachesWhileSessionAlive(t *testing.T) { + src := &fakeSource{alive: true} // session still alive -> re-attach on drop + p1, p2 := newFakePTY(), newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{p1, p2}} + a := newTestAttachment(src, sp.spawn, nil, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go a.run(ctx) + + eventually(t, time.Second, func() bool { return sp.calls() >= 1 }) + if err := a.resize(24, 80); err != nil { + t.Fatalf("resize: %v", err) + } + p1.Close() // first attach drops + eventually(t, 2*time.Second, func() bool { return sp.calls() >= 2 }) + + // The client's grid survives the re-attach: the fresh PTY is sized to the + // last requested grid without waiting for the client to resize again. + eventually(t, time.Second, func() bool { + for _, rs := range p2.resizeCalls() { + if rs == [2]uint16{24, 80} { + return true + } + } + return false + }) + + // Now the session is gone: the next drop must not re-attach. + src.setAlive(false) + p2.Close() + eventually(t, 2*time.Second, func() bool { return a.isExited() }) +} + +func TestAttachmentFailsWhenAttachCommandErrors(t *testing.T) { + src := &fakeSource{alive: true, attachErr: io.ErrUnexpectedEOF} + sp := &fakeSpawner{} + + exited := make(chan struct{}) + a := newTestAttachment(src, sp.spawn, nil, func() { close(exited) }) + + go a.run(context.Background()) + select { + case <-exited: + case <-time.After(time.Second): + t.Fatal("expected exit when attach command fails") + } + if sp.calls() != 0 { + t.Fatalf("spawn should not run when attach command errors, got %d calls", sp.calls()) + } +} + +// close() is a detach, not a pane death: it must stop the attach loop and kill +// the client's PTY without firing onExit — the Zellij session is still alive +// and an exited frame would wrongly flip the client UI to its terminal state. +func TestAttachmentCloseDoesNotFireExit(t *testing.T) { + src := &fakeSource{alive: true} + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + + exited := make(chan struct{}) + a := newTestAttachment(src, sp.spawn, nil, func() { close(exited) }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + a.run(ctx) + close(done) + }() + + eventually(t, time.Second, func() bool { return sp.calls() == 1 }) + a.close() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("run must return after close") + } + select { + case <-exited: + t.Fatal("close must not fire onExit") + default: + } + if got := sp.calls(); got != 1 { + t.Fatalf("close must stop re-attaching, got %d spawns", got) + } +} diff --git a/backend/internal/terminal/doc.go b/backend/internal/terminal/doc.go index 44878ec34..9c262c7a6 100644 --- a/backend/internal/terminal/doc.go +++ b/backend/internal/terminal/doc.go @@ -1,20 +1,31 @@ -// Package terminal is the live-terminal streaming feature: it attaches to a -// session's Zellij pane over a PTY and multiplexes the byte stream to one or more -// WebSocket clients, alongside a session-state channel fed by the CDC +// Package terminal is the live-terminal streaming feature: each WebSocket +// client that opens a pane gets its OWN `zellij attach` PTY, piped over a +// ch-tagged wire protocol, alongside a session-state channel fed by the CDC // broadcaster. // +// Per-client attach (no shared PTY, no replay buffer): Zellij is the +// multiplexer — it owns the session's screen state and scrollback, and answers +// every fresh attach with its full init handshake (alt screen, SGR mouse +// tracking, bracketed paste) followed by a faithful repaint. Sharing one PTY +// and replaying a bounded byte ring to late subscribers loses exactly that +// handshake (it is emitted once, at the head of the stream), which left +// clients without mouse reporting — wheel scroll dead. Spawning a fresh attach +// per client makes Zellij re-send it, every time, by construction. The cost is +// one zellij client process per open pane per connection, which is what the +// zellij server is built for (yyork ships the same model). +// // Boundaries (see docs/architecture.md): // -// - This package owns the product workflow: PTY attach, output fan-out, a -// bounded replay buffer, re-attach resilience, and the ch-tagged wire -// protocol. It is transport-agnostic: it speaks to a small wsConn interface, -// not to any concrete WebSocket library. +// - This package owns the product workflow: per-client PTY attach, liveness +// gating, re-attach resilience, and the ch-tagged wire protocol. It is +// transport-agnostic: it speaks to a small wsConn interface, not to any +// concrete WebSocket library. // - internal/httpd owns the HTTP/WebSocket upgrade and adapts the accepted // socket to wsConn; it does not contain stream logic. // - The PTY itself is reached through PTYSource (satisfied by the Zellij runtime // adapter's AttachCommand/IsAlive) and spawned through an injectable -// spawnFunc, so the fan-out, buffering, and re-attach logic test without a -// real process, Zellij, or network. +// spawnFunc, so the attach and re-attach logic test without a real process, +// Zellij, or network. // // Raw PTY bytes never flow through the CDC change_log; only the session channel // is fed by cdc.Broadcaster. Terminal output is high-volume ephemeral data and diff --git a/backend/internal/terminal/fakes_test.go b/backend/internal/terminal/fakes_test.go index dad7a7203..2ff0688f0 100644 --- a/backend/internal/terminal/fakes_test.go +++ b/backend/internal/terminal/fakes_test.go @@ -116,14 +116,16 @@ type fakeSpawner struct { n int err error created []*fakePTY + sizes [][2]uint16 // rows×cols passed to each spawn call, in order } -func (f *fakeSpawner) spawn(context.Context, []string) (ptyProcess, error) { +func (f *fakeSpawner) spawn(_ context.Context, _ []string, rows, cols uint16) (ptyProcess, error) { f.mu.Lock() defer f.mu.Unlock() if f.err != nil { return nil, f.err } + f.sizes = append(f.sizes, [2]uint16{rows, cols}) var p *fakePTY if f.n < len(f.ptys) { p = f.ptys[f.n] @@ -141,6 +143,12 @@ func (f *fakeSpawner) calls() int { return f.n } +func (f *fakeSpawner) spawnSizes() [][2]uint16 { + f.mu.Lock() + defer f.mu.Unlock() + return append([][2]uint16(nil), f.sizes...) +} + // eventually polls cond until true or the deadline, failing the test otherwise. func eventually(t *testing.T, d time.Duration, cond func() bool) { t.Helper() diff --git a/backend/internal/terminal/logger_test.go b/backend/internal/terminal/logger_test.go index a13233415..0384a65a7 100644 --- a/backend/internal/terminal/logger_test.go +++ b/backend/internal/terminal/logger_test.go @@ -12,8 +12,8 @@ func TestNilLoggerFallsBackToDefault(t *testing.T) { if mgr.log == nil { t.Fatal("manager logger is nil") } - s := newSession("t1", ports.RuntimeHandle{ID: "t1"}, &fakeSource{}, (&fakeSpawner{}).spawn, nil) - if s.log == nil { - t.Fatal("session logger is nil") + a := newAttachment("t1", ports.RuntimeHandle{ID: "t1"}, &fakeSource{}, (&fakeSpawner{}).spawn, nil, nil, nil) + if a.log == nil { + t.Fatal("attachment logger is nil") } } diff --git a/backend/internal/terminal/manager.go b/backend/internal/terminal/manager.go index 79ba6134c..7b22608b6 100644 --- a/backend/internal/terminal/manager.go +++ b/backend/internal/terminal/manager.go @@ -34,9 +34,11 @@ const ( defaultWriteBuffer = 1024 ) -// Manager owns the set of live terminal sessions and serves WebSocket clients. -// Sessions outlive any single connection: multiple clients can attach to the -// same pane, and a client reconnect re-subscribes to the existing session. +// Manager serves WebSocket clients, spawning one attach PTY per opened pane +// per connection. There is no shared per-pane state to outlive a connection: +// the Zellij server owns the session (screen, scrollback, modes), and every +// fresh attach gets its full handshake + repaint. A client reconnect simply +// attaches again. type Manager struct { src PTYSource events EventSource @@ -44,13 +46,13 @@ type Manager struct { log *slog.Logger heartbeat time.Duration - // ctx scopes every session's PTY lifetime; cancelled by Close. + // ctx scopes every attachment's PTY lifetime; cancelled by Close. ctx context.Context cancel context.CancelFunc - mu sync.Mutex - sessions map[string]*session - closed bool + mu sync.Mutex + attachments map[*attachment]struct{} + closed bool } // Option configures a Manager. @@ -70,14 +72,14 @@ func NewManager(src PTYSource, events EventSource, log *slog.Logger, opts ...Opt } ctx, cancel := context.WithCancel(context.Background()) m := &Manager{ - src: src, - events: events, - spawn: defaultSpawn, - log: log, - heartbeat: defaultHeartbeat, - ctx: ctx, - cancel: cancel, - sessions: map[string]*session{}, + src: src, + events: events, + spawn: defaultSpawn, + log: log, + heartbeat: defaultHeartbeat, + ctx: ctx, + cancel: cancel, + attachments: map[*attachment]struct{}{}, } for _, opt := range opts { opt(m) @@ -85,8 +87,8 @@ func NewManager(src PTYSource, events EventSource, log *slog.Logger, opts ...Opt return m } -// Close tears down every session and stops re-attach loops. Safe to call once on -// daemon shutdown. +// Close tears down every live attachment and stops re-attach loops. Safe to +// call once on daemon shutdown. func (m *Manager) Close() { m.mu.Lock() if m.closed { @@ -94,42 +96,35 @@ func (m *Manager) Close() { return } m.closed = true - sessions := make([]*session, 0, len(m.sessions)) - for _, s := range m.sessions { - sessions = append(sessions, s) + attachments := make([]*attachment, 0, len(m.attachments)) + for a := range m.attachments { + attachments = append(attachments, a) } - m.sessions = map[string]*session{} + m.attachments = map[*attachment]struct{}{} m.mu.Unlock() m.cancel() - for _, s := range sessions { - s.close() + for _, a := range attachments { + a.close() } } -// openSession returns the live session for id, starting it on first open. The id -// is the runtime handle id (Zellij handle). -func (m *Manager) openSession(id string) (*session, error) { +// track registers a live attachment so Close can tear it down; it refuses new +// attachments once the manager is closed. +func (m *Manager) track(a *attachment) error { m.mu.Lock() defer m.mu.Unlock() if m.closed { - return nil, context.Canceled + return context.Canceled } - if s, ok := m.sessions[id]; ok { - return s, nil - } - handle := ports.RuntimeHandle{ID: id} - s := newSession(id, handle, m.src, m.spawn, m.log) - m.sessions[id] = s - go func() { - s.run(m.ctx) - m.mu.Lock() - if cur, ok := m.sessions[id]; ok && cur == s { - delete(m.sessions, id) - } - m.mu.Unlock() - }() - return s, nil + m.attachments[a] = struct{}{} + return nil +} + +func (m *Manager) forget(a *attachment) { + m.mu.Lock() + delete(m.attachments, a) + m.mu.Unlock() } // Serve runs the protocol loop for one client connection until it errors, the @@ -144,7 +139,7 @@ func (m *Manager) Serve(ctx context.Context, conn wsConn) { conn: conn, cancel: cancel, out: make(chan serverMsg, defaultWriteBuffer), - terms: map[string]func(){}, + terms: map[string]*attachment{}, } defer c.cleanup() @@ -171,7 +166,7 @@ type connState struct { out chan serverMsg mu sync.Mutex - terms map[string]func() // terminal id -> unsubscribe + terms map[string]*attachment // terminal id -> this conn's own attach PTY unsubEvts func() closed bool } @@ -192,25 +187,28 @@ func (c *connState) handle(msg clientMsg) { func (c *connState) handleTerminal(msg clientMsg) { switch msg.Type { case msgOpen: - c.openTerminal(msg.ID) + c.openTerminal(msg.ID, msg.Rows, msg.Cols) case msgData: raw, err := base64.StdEncoding.DecodeString(msg.Data) if err != nil { return } - if s := c.lookup(msg.ID); s != nil { - _ = s.write(raw) + if a := c.lookup(msg.ID); a != nil { + _ = a.write(raw) } case msgResize: - if s := c.lookup(msg.ID); s != nil { - _ = s.resize(msg.Rows, msg.Cols) + if a := c.lookup(msg.ID); a != nil { + _ = a.resize(msg.Rows, msg.Cols) } case msgClose: c.closeTerminal(msg.ID) } } -func (c *connState) openTerminal(id string) { +// openTerminal spawns this connection's own attach PTY for the pane. rows/cols +// are the client's grid from the open frame, applied as the PTY's initial size +// (a resize that raced ahead of the attach would otherwise be lost). +func (c *connState) openTerminal(id string, rows, cols uint16) { if id == "" { c.enqueue(serverMsg{Ch: chTerminal, Type: msgError, Error: "missing terminal id"}) return @@ -218,29 +216,14 @@ func (c *connState) openTerminal(id string) { c.mu.Lock() if _, ok := c.terms[id]; ok { c.mu.Unlock() - return // already open on this conn; avoid duplicate replay + return // already open on this conn; avoid a duplicate attach } c.mu.Unlock() - s, err := c.mgr.openSession(id) - if err != nil { - c.enqueue(serverMsg{Ch: chTerminal, ID: id, Type: msgError, Error: err.Error()}) - return - } - - // Ack before subscribing so opened always precedes the replay and any - // data/exited frames subscribe delivers (the single out channel preserves - // this order). Reversing it would let a reconnecting client with buffered - // content, or one opening an already-dead pane, see data/exited before the - // open acknowledgement. - c.enqueue(serverMsg{Ch: chTerminal, ID: id, Type: msgOpened}) - - // exitFired guards the subscribe-to-assign window: the session can exit (and - // run onExit) at any point after subscribe returns exited=false, including - // before c.terms[id] is assigned below. onExit and the assign both read/write - // this flag and the map only under c.mu, so no atomic is needed. - var exitFired bool - unsub, exited := s.subscribe( + // a is captured by onExit before assignment; safe because the attach loop — + // the only thing that fires onExit — starts after the registration below. + var a *attachment + a = newAttachment(id, ports.RuntimeHandle{ID: id}, c.mgr.src, c.mgr.spawn, func(data []byte) { c.enqueue(serverMsg{ Ch: chTerminal, @@ -252,55 +235,53 @@ func (c *connState) openTerminal(id string) { func() { // Clear the connection's entry for this id before sending exited so // a client that reopens the moment it sees exited finds no stale - // entry and is served instead of dropped by the open guard. Ordering - // the delete after the frame would race that reopen. markExited fires - // this without s.mu held, so taking c.mu is safe. + // entry and is served instead of dropped by the open guard. Guard on + // identity: that reopen may already have installed a fresh + // attachment under the same id, which must not be evicted. c.mu.Lock() - exitFired = true - delete(c.terms, id) + if c.terms[id] == a { + delete(c.terms, id) + } c.mu.Unlock() c.enqueue(serverMsg{Ch: chTerminal, ID: id, Type: msgExited}) }, - ) - // An already-exited session sent its exited frame from subscribe and has - // nothing to unsubscribe. Don't register it: leaving c.terms[id] set would - // trip the open guard above and silently drop every later open for this id - // on this connection (e.g. after the pane respawns) until close/reconnect. - if exited { + c.mgr.log) + if rows > 0 && cols > 0 { + _ = a.resize(rows, cols) // recorded now, applied when the PTY attaches + } + if err := c.mgr.track(a); err != nil { + c.enqueue(serverMsg{Ch: chTerminal, ID: id, Type: msgError, Error: err.Error()}) return } c.mu.Lock() - c.terms[id] = unsub - // If onExit already ran in the subscribe-to-assign window its delete was a - // no-op (the key did not exist yet), so the assign above just resurrected a - // stale entry for a dead pane. Re-apply the delete while still holding c.mu. - if exitFired { - delete(c.terms, id) - } + c.terms[id] = a c.mu.Unlock() + + // Ack before starting the attach loop so opened always precedes any + // data/exited frames (the single out channel preserves this order). A + // dead pane is reported as opened followed by exited. + c.enqueue(serverMsg{Ch: chTerminal, ID: id, Type: msgOpened}) + + go func() { + a.run(c.mgr.ctx) + c.mgr.forget(a) + }() } func (c *connState) closeTerminal(id string) { c.mu.Lock() - unsub := c.terms[id] + a := c.terms[id] delete(c.terms, id) c.mu.Unlock() - if unsub != nil { - unsub() + if a != nil { + a.close() } } -func (c *connState) lookup(id string) *session { +func (c *connState) lookup(id string) *attachment { c.mu.Lock() - _, open := c.terms[id] - c.mu.Unlock() - if !open { - return nil - } - c.mgr.mu.Lock() - s := c.mgr.sessions[id] - c.mgr.mu.Unlock() - return s + defer c.mu.Unlock() + return c.terms[id] } func (c *connState) handleSubscribe(msg clientMsg) { @@ -332,8 +313,8 @@ func (c *connState) handleSubscribe(msg clientMsg) { } // enqueue pushes a frame to the writer. If the buffer is full the client is too -// slow to keep up; tear the connection down rather than stall fan-out for other -// subscribers of the same pane. +// slow to keep up; tear the connection down rather than block the attachment's +// PTY read loop behind it. func (c *connState) enqueue(msg serverMsg) { select { case c.out <- msg: @@ -385,19 +366,20 @@ func (c *connState) cleanup() { return } c.closed = true - unsubs := make([]func(), 0, len(c.terms)+1) - for _, u := range c.terms { - unsubs = append(unsubs, u) - } - c.terms = map[string]func(){} - if c.unsubEvts != nil { - unsubs = append(unsubs, c.unsubEvts) - c.unsubEvts = nil + attachments := make([]*attachment, 0, len(c.terms)) + for _, a := range c.terms { + attachments = append(attachments, a) } + c.terms = map[string]*attachment{} + unsubEvts := c.unsubEvts + c.unsubEvts = nil c.mu.Unlock() - for _, u := range unsubs { - u() + for _, a := range attachments { + a.close() + } + if unsubEvts != nil { + unsubEvts() } _ = c.conn.Close("server: connection closed") } diff --git a/backend/internal/terminal/manager_test.go b/backend/internal/terminal/manager_test.go index 1584487d8..dc0730b88 100644 --- a/backend/internal/terminal/manager_test.go +++ b/backend/internal/terminal/manager_test.go @@ -9,7 +9,6 @@ import ( "time" "github.com/aoagents/agent-orchestrator/backend/internal/cdc" - "github.com/aoagents/agent-orchestrator/backend/internal/ports" ) // fakeConn is an in-memory wsConn driven by channels. @@ -113,22 +112,16 @@ func nextTerminal(t *testing.T, c *fakeConn) serverMsg { } } -// Opening a terminal whose session has already exited but is not yet reaped from -// m.sessions must (1) send opened before exited and (2) not register the noop -// unsubscribe, so a later open for the same id on this connection is still -// served instead of being silently dropped by the already-open guard. -func TestServeOpenAlreadyExitedSessionDoesNotBlockReopen(t *testing.T) { - src := &fakeSource{} +// Opening a pane whose runtime is already dead must (1) send opened before +// exited (the dead pane is reported, not errored) and (2) clear the conn's +// entry, so a later open for the same id on this connection is still served +// instead of being silently dropped by the already-open guard. +func TestServeOpenDeadRuntimeReportsExitedAndAllowsReopen(t *testing.T) { + src := &fakeSource{alive: false} sp := &fakeSpawner{} mgr := NewManager(src, nil, testLogger(), WithSpawn(sp.spawn), WithHeartbeat(0)) defer mgr.Close() - exited := newSession("t1", ports.RuntimeHandle{ID: "t1"}, src, sp.spawn, testLogger()) - exited.markExited() - mgr.mu.Lock() - mgr.sessions["t1"] = exited - mgr.mu.Unlock() - conn := newFakeConn() ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -141,6 +134,9 @@ func TestServeOpenAlreadyExitedSessionDoesNotBlockReopen(t *testing.T) { if m := nextTerminal(t, conn); m.Type != msgExited { t.Fatalf("second frame = %q, want exited", m.Type) } + if got := sp.calls(); got != 0 { + t.Fatalf("attach must never run against a dead runtime, got %d spawns", got) + } conn.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgOpen} if m := nextTerminal(t, conn); m.Type != msgOpened { @@ -174,38 +170,30 @@ func TestServeExitAfterOpenClearsEntryAllowingReopen(t *testing.T) { recv(t, conn, chTerminal, msgOpened, 2*time.Second) } -// The subscribe-to-assign window: a session can exit (running onExit, which -// deletes c.terms[id]) between subscribe returning exited=false and openTerminal -// assigning c.terms[id] = unsub. If the assign resurrects that entry without -// re-checking, the stale entry traps every later open for the id on this -// connection. Close the pty concurrently with the open (IsAlive false => no -// re-attach) so the exit races the assign across many iterations; every reopen -// must be served (opened), never silently dropped by the open guard. +// An attachment that exits the moment it is opened (dead runtime) fires onExit +// from its run goroutine, racing the reopen that follows the exited frame. The +// identity-guarded delete in onExit must never evict a successor attachment +// registered under the same id: every reopen must be served (opened), never +// silently dropped by the already-open guard. Stressed across many iterations +// to shake the exit/reopen interleavings out. func TestServeReopenAfterImmediateExitNeverStuck(t *testing.T) { for i := 0; i < 400; i++ { src := &fakeSource{} - src.setAlive(false) // dropped pty must not re-attach -> session exits - p := newFakePTY() // alive at subscribe; closed below to race the assign - sp := &fakeSpawner{ptys: []*fakePTY{p}} + src.setAlive(false) // dead runtime -> the open exits without attaching + sp := &fakeSpawner{} mgr := NewManager(src, nil, testLogger(), WithSpawn(sp.spawn), WithHeartbeat(0)) conn := newFakeConn() ctx, cancel := context.WithCancel(context.Background()) go mgr.Serve(ctx, conn) - // Send the open and close the pty concurrently: the session's exit - // (onExit -> delete c.terms[id]) then races openTerminal's assign of - // c.terms[id] = unsub. On the iterations where exit lands in the - // subscribe-to-assign window, an unguarded assign resurrects a stale - // entry for the dead pane, trapping every later open for this id. conn.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgOpen} - go p.Close() recv(t, conn, chTerminal, msgOpened, time.Second) recv(t, conn, chTerminal, msgExited, time.Second) - // The reopen must be served even when the first open's session exited in - // the subscribe-to-assign window. + // The reopen must be served even while the first open's exit teardown is + // still in flight. conn.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgOpen} recv(t, conn, chTerminal, msgOpened, time.Second) @@ -290,12 +278,116 @@ func TestServeClosesConnOnReadEnd(t *testing.T) { } } +// Each connection opening the same pane gets its OWN attach PTY — that is the +// per-client model: zellij replays its init handshake + full repaint to every +// fresh attach, so no client depends on bytes another client consumed. Output +// pushed to one client's PTY must reach only that client, and closing one +// client's terminal must not touch the other's PTY. +func TestServePerClientAttachIsolation(t *testing.T) { + src := &fakeSource{alive: true} + p1, p2 := newFakePTY(), newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{p1, p2}} + mgr := NewManager(src, nil, testLogger(), WithSpawn(sp.spawn), WithHeartbeat(0)) + defer mgr.Close() + + connA, connB := newFakeConn(), newFakeConn() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go mgr.Serve(ctx, connA) + go mgr.Serve(ctx, connB) + + connA.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgOpen} + recv(t, connA, chTerminal, msgOpened, time.Second) + eventually(t, time.Second, func() bool { return sp.calls() == 1 }) + + connB.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgOpen} + recv(t, connB, chTerminal, msgOpened, time.Second) + eventually(t, time.Second, func() bool { return sp.calls() == 2 }) + + // Zellij fans output out per attach; here each fake PTY stands in for one + // attach process, so its bytes must reach exactly its own connection. + p1.push([]byte("for-A")) + data := recv(t, connA, chTerminal, msgData, time.Second) + got, _ := base64.StdEncoding.DecodeString(data.Data) + if string(got) != "for-A" { + t.Fatalf("conn A data = %q", got) + } + select { + case m := <-connB.out: + t.Fatalf("conn B received %s/%s for conn A's PTY output", m.Ch, m.Type) + default: + } + + // Closing A's terminal detaches A only: B's PTY stays live. + connA.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgClose} + eventually(t, time.Second, func() bool { + select { + case <-p1.closed: + return true + default: + return false + } + }) + select { + case <-p2.closed: + t.Fatal("closing conn A's terminal must not close conn B's PTY") + default: + } +} + +// The open frame carries the client's grid; the PTY must start at that size +// rather than the kernel default, even though the attach is asynchronous. +func TestServeOpenAppliesInitialSize(t *testing.T) { + src := &fakeSource{alive: true} + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + mgr := NewManager(src, nil, testLogger(), WithSpawn(sp.spawn), WithHeartbeat(0)) + defer mgr.Close() + + conn := newFakeConn() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go mgr.Serve(ctx, conn) + + conn.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgOpen, Rows: 40, Cols: 120} + recv(t, conn, chTerminal, msgOpened, time.Second) + eventually(t, time.Second, func() bool { + rs := pty.resizeCalls() + return len(rs) == 1 && rs[0] == [2]uint16{40, 120} + }) +} + +// Manager.Close must kill every live attach PTY: a PTY left open keeps its +// attach process running and deadlocks daemon shutdown. +func TestManagerCloseKillsLiveAttachments(t *testing.T) { + src := &fakeSource{alive: true} + pty := newFakePTY() + sp := &fakeSpawner{ptys: []*fakePTY{pty}} + mgr := NewManager(src, nil, testLogger(), WithSpawn(sp.spawn), WithHeartbeat(0)) + + conn := newFakeConn() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go mgr.Serve(ctx, conn) + + conn.in <- clientMsg{Ch: chTerminal, ID: "t1", Type: msgOpen} + recv(t, conn, chTerminal, msgOpened, time.Second) + eventually(t, time.Second, func() bool { return sp.calls() == 1 }) + + mgr.Close() + select { + case <-pty.closed: + case <-time.After(time.Second): + t.Fatal("Manager.Close must close live attach PTYs") + } +} + func TestEnqueueOverflowCancelsConn(t *testing.T) { cancelled := make(chan struct{}) c := &connState{ out: make(chan serverMsg, 1), cancel: func() { close(cancelled) }, - terms: map[string]func(){}, + terms: map[string]*attachment{}, } c.enqueue(serverMsg{Ch: chTerminal, Type: msgData}) // fills buffer c.enqueue(serverMsg{Ch: chTerminal, Type: msgData}) // overflow -> cancel diff --git a/backend/internal/terminal/pty_unix.go b/backend/internal/terminal/pty_unix.go index a250a0379..6f1730943 100644 --- a/backend/internal/terminal/pty_unix.go +++ b/backend/internal/terminal/pty_unix.go @@ -8,19 +8,30 @@ import ( "os" "os/exec" "sync" + "syscall" + "time" "github.com/creack/pty" ) -// defaultSpawn starts argv on a real PTY via creack/pty. ctx cancellation kills -// the process. Windows uses a stub (see pty_windows.go) until a ConPTY path is -// added. -func defaultSpawn(ctx context.Context, argv []string) (ptyProcess, error) { +// defaultSpawn starts argv on a real PTY via creack/pty, sized rows×cols from +// birth when a size is known: `zellij attach` reads the tty size once at +// startup, and a post-spawn TIOCSWINSZ depends on SIGWINCH delivery that can +// race the client installing its handler — StartWithSize makes the first read +// correct by construction. ctx cancellation kills the process. Windows uses a +// stub (see pty_windows.go) until a ConPTY path is added. +func defaultSpawn(ctx context.Context, argv []string, rows, cols uint16) (ptyProcess, error) { if len(argv) == 0 { return nil, errors.New("terminal: empty attach command") } cmd := exec.CommandContext(ctx, argv[0], argv[1:]...) - f, err := pty.Start(cmd) + var f *os.File + var err error + if rows > 0 && cols > 0 { + f, err = pty.StartWithSize(cmd, &pty.Winsize{Rows: rows, Cols: cols}) + } else { + f, err = pty.Start(cmd) + } if err != nil { return nil, err } @@ -38,23 +49,60 @@ func (p *creackPTY) Read(b []byte) (int, error) { return p.f.Read(b) } func (p *creackPTY) Write(b []byte) (int, error) { return p.f.Write(b) } func (p *creackPTY) Resize(rows, cols uint16) error { - return pty.Setsize(p.f, &pty.Winsize{Rows: rows, Cols: cols}) + err := pty.Setsize(p.f, &pty.Winsize{Rows: rows, Cols: cols}) + // Always follow with an explicit SIGWINCH: the kernel only raises one when + // the size actually changed, so a re-asserted (identical) grid would never + // reach a zellij client that missed or lost the original signal — the + // session would stay laid out for a stale size, with no repaint until the + // next real change (the frontend re-sends its grid after each resize burst + // for exactly this self-heal; see useTerminalSession). The client re-reads + // the tty and re-reports to its server; when already in sync it's a no-op. + if p.cmd.Process != nil { + _ = p.cmd.Process.Signal(syscall.SIGWINCH) + } + return err } -// Close stops the attach process and releases the PTY. Zellij attach exits cleanly -// when the master closes, but kill the process to be sure it does not linger. +// detachGrace is how long Close waits for a SIGTERM'd attach process to exit +// on its own before falling back to SIGKILL. A zellij client that is being +// drained detaches in ~50ms; the grace only runs out for a wedged process. +const detachGrace = 250 * time.Millisecond + +// Close stops the attach process and releases the PTY. // -// It is idempotent: both the session run loop (after copyOut returns) and -// session.close (via Manager.Close) call Close on the same PTY, and cmd.Wait -// must run exactly once. A second concurrent Wait on the same process blocks -// forever, deadlocking daemon shutdown when a terminal is still attached. +// SIGTERM first, SIGKILL as fallback: a SIGTERM'd `zellij attach` deregisters +// itself from the zellij server before exiting, while a SIGKILL'd one leaves +// deregistration to the server noticing the dead socket. A dead-but-registered +// client pins the session's size (zellij sizes a session to its smallest +// client), so the next attach renders for the ghost's grid — the "terminal +// doesn't repaint to the new size" desync. The master stays open through the +// grace so the run loop's copyOut keeps draining the client's shutdown output +// (a blocked tty write would stall the graceful exit past the grace). +// +// It is idempotent: both the attachment run loop (after copyOut returns) and +// attachment.close (via closeTerminal, conn cleanup, or Manager.Close) call +// Close on the same PTY, and cmd.Wait must run exactly once. A second +// concurrent Wait on the same process blocks forever, deadlocking daemon +// shutdown when a terminal is still attached. func (p *creackPTY) Close() error { p.closeOnce.Do(func() { - p.closeErr = p.f.Close() + done := make(chan struct{}) + go func() { + _ = p.cmd.Wait() + close(done) + }() if p.cmd.Process != nil { - _ = p.cmd.Process.Kill() + _ = p.cmd.Process.Signal(syscall.SIGTERM) } - _ = p.cmd.Wait() + select { + case <-done: + case <-time.After(detachGrace): + if p.cmd.Process != nil { + _ = p.cmd.Process.Kill() + } + <-done + } + p.closeErr = p.f.Close() }) return p.closeErr } diff --git a/backend/internal/terminal/pty_unix_test.go b/backend/internal/terminal/pty_unix_test.go index 7d8c04ff5..76e26337a 100644 --- a/backend/internal/terminal/pty_unix_test.go +++ b/backend/internal/terminal/pty_unix_test.go @@ -4,6 +4,7 @@ package terminal import ( "context" + "strings" "testing" "time" ) @@ -13,7 +14,7 @@ import ( // exactly once. Without the sync.Once a second Wait blocks forever, so this test // would hang (caught by the watchdog) rather than fail. func TestCreackPTYCloseIsIdempotent(t *testing.T) { - p, err := defaultSpawn(context.Background(), []string{"/bin/sh", "-c", "sleep 30"}) + p, err := defaultSpawn(context.Background(), []string{"/bin/sh", "-c", "sleep 30"}, 0, 0) if err != nil { t.Fatalf("spawn: %v", err) } @@ -31,3 +32,115 @@ func TestCreackPTYCloseIsIdempotent(t *testing.T) { t.Fatal("creackPTY.Close did not return: double Close deadlocked on cmd.Wait") } } + +// TestCreackPTYResizeSignalsOnIdenticalSize guards the resize self-heal: the +// kernel only raises SIGWINCH when TIOCSWINSZ actually changes the size, so a +// re-asserted (identical) grid relies on Resize's explicit signal. A zellij +// client that lost the original update would otherwise keep its server laid +// out for a stale size forever — the "terminal doesn't repaint after resizing +// the pane" desync. +func TestCreackPTYResizeSignalsOnIdenticalSize(t *testing.T) { + p, err := defaultSpawn(context.Background(), + []string{"/bin/sh", "-c", `trap 'echo WINCHED' WINCH; while :; do sleep 0.05; done`}, 0, 0) + if err != nil { + t.Fatalf("spawn: %v", err) + } + defer p.Close() + + // Give the shell a beat to install the trap, then resize twice to the SAME + // size. The first call changes the size (fresh PTYs start at 0x0) and the + // second is identical — only the explicit signal can deliver it. + time.Sleep(200 * time.Millisecond) + if err := p.Resize(24, 80); err != nil { + t.Fatalf("resize 1: %v", err) + } + time.Sleep(200 * time.Millisecond) + if err := p.Resize(24, 80); err != nil { + t.Fatalf("resize 2: %v", err) + } + + deadline := time.Now().Add(5 * time.Second) + var out strings.Builder + buf := make([]byte, 4096) + for time.Now().Before(deadline) { + n, err := p.Read(buf) + if n > 0 { + out.WriteString(string(buf[:n])) + if strings.Count(out.String(), "WINCHED") >= 2 { + return + } + } + if err != nil { + break + } + } + t.Fatalf("expected 2 WINCHED traps (one per Resize, including the identical one), got output: %q", out.String()) +} + +// TestCreackPTYSpawnsAtRequestedSize: the child must see the requested grid on +// its very first TIOCGWINSZ, with no SIGWINCH involved — sizing after exec +// races the client installing its WINCH handler (a missed signal strands the +// zellij session at the previous client's size). +func TestCreackPTYSpawnsAtRequestedSize(t *testing.T) { + p, err := defaultSpawn(context.Background(), []string{"/bin/sh", "-c", "stty size"}, 40, 140) + if err != nil { + t.Fatalf("spawn: %v", err) + } + defer p.Close() + + deadline := time.Now().Add(5 * time.Second) + var out strings.Builder + buf := make([]byte, 4096) + for time.Now().Before(deadline) { + n, readErr := p.Read(buf) + if n > 0 { + out.WriteString(string(buf[:n])) + if strings.Contains(out.String(), "40 140") { + return + } + } + if readErr != nil { + break + } + } + t.Fatalf("child did not see the spawn size 40x140, got output: %q", out.String()) +} + +// TestCreackPTYCloseTermsBeforeKill: Close must give the attach process a +// chance to exit on SIGTERM (a zellij client deregisters from its server on +// SIGTERM; a straight SIGKILL leaves a ghost client that pins the session's +// size), and must still return promptly for a process that ignores SIGTERM. +func TestCreackPTYCloseTermsBeforeKill(t *testing.T) { + t.Run("cooperative process exits within the grace", func(t *testing.T) { + p, err := defaultSpawn(context.Background(), + []string{"/bin/sh", "-c", `trap 'exit 0' TERM; while :; do sleep 0.05; done`}, 0, 0) + if err != nil { + t.Fatalf("spawn: %v", err) + } + time.Sleep(200 * time.Millisecond) // let the trap install + start := time.Now() + _ = p.Close() + if elapsed := time.Since(start); elapsed >= detachGrace { + t.Fatalf("Close took %v: SIGTERM path did not let a cooperative process exit before the kill grace", elapsed) + } + }) + + t.Run("TERM-ignoring process is killed after the grace", func(t *testing.T) { + p, err := defaultSpawn(context.Background(), + []string{"/bin/sh", "-c", `trap '' TERM; while :; do sleep 0.05; done`}, 0, 0) + if err != nil { + t.Fatalf("spawn: %v", err) + } + time.Sleep(200 * time.Millisecond) + done := make(chan struct{}) + go func() { + _ = p.Close() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Close did not return for a TERM-ignoring process: SIGKILL fallback missing") + } + }) +} diff --git a/backend/internal/terminal/pty_windows.go b/backend/internal/terminal/pty_windows.go index f88ef557b..6d8107f3c 100644 --- a/backend/internal/terminal/pty_windows.go +++ b/backend/internal/terminal/pty_windows.go @@ -10,6 +10,6 @@ import ( // defaultSpawn is not implemented on Windows: the POSIX PTY path uses // creack/pty. The rest of the package compiles and tests on Windows with an // injected spawner. -func defaultSpawn(_ context.Context, _ []string) (ptyProcess, error) { +func defaultSpawn(_ context.Context, _ []string, _, _ uint16) (ptyProcess, error) { return nil, errors.New("terminal: PTY streaming is not supported on Windows yet") } diff --git a/backend/internal/terminal/ring.go b/backend/internal/terminal/ring.go deleted file mode 100644 index 8ed303cdf..000000000 --- a/backend/internal/terminal/ring.go +++ /dev/null @@ -1,40 +0,0 @@ -package terminal - -// defaultRingMax caps per-terminal replay history. A late subscriber gets at -// most this many bytes of recent output so it can paint a usable screen without -// the whole session backlog. Matches the legacy 50KB ring. -const defaultRingMax = 50 * 1024 - -// ringBuffer is a byte ring holding the most recent output of one terminal. It -// is owned by session and accessed under session.mu. -type ringBuffer struct { - buf []byte - max int -} - -func newRingBuffer(maxBytes int) *ringBuffer { - if maxBytes <= 0 { - maxBytes = defaultRingMax - } - return &ringBuffer{max: maxBytes} -} - -// append adds p and drops the oldest bytes beyond max. A single write larger -// than max is truncated to its last max bytes. -func (r *ringBuffer) append(p []byte) { - if len(p) >= r.max { - r.buf = append(r.buf[:0], p[len(p)-r.max:]...) - return - } - r.buf = append(r.buf, p...) - if len(r.buf) > r.max { - r.buf = append(r.buf[:0], r.buf[len(r.buf)-r.max:]...) - } -} - -// snapshot returns a copy of the current contents (oldest first). -func (r *ringBuffer) snapshot() []byte { - out := make([]byte, len(r.buf)) - copy(out, r.buf) - return out -} diff --git a/backend/internal/terminal/ring_test.go b/backend/internal/terminal/ring_test.go deleted file mode 100644 index 26b91358c..000000000 --- a/backend/internal/terminal/ring_test.go +++ /dev/null @@ -1,38 +0,0 @@ -package terminal - -import ( - "bytes" - "strings" - "testing" -) - -func TestRingBufferKeepsTailWithinCap(t *testing.T) { - r := newRingBuffer(8) - r.append([]byte("abcd")) - r.append([]byte("efgh")) - r.append([]byte("ij")) // total 10 > 8, drop oldest 2 - - if got := string(r.snapshot()); got != "cdefghij" { - t.Fatalf("snapshot = %q, want %q", got, "cdefghij") - } -} - -func TestRingBufferTruncatesOversizeWrite(t *testing.T) { - r := newRingBuffer(4) - r.append([]byte(strings.Repeat("x", 3))) - r.append([]byte("abcdefgh")) // single write larger than cap - - if got := string(r.snapshot()); got != "efgh" { - t.Fatalf("snapshot = %q, want %q", got, "efgh") - } -} - -func TestRingBufferSnapshotIsCopy(t *testing.T) { - r := newRingBuffer(16) - r.append([]byte("data")) - snap := r.snapshot() - snap[0] = 'X' - if !bytes.Equal(r.snapshot(), []byte("data")) { - t.Fatal("snapshot must not alias internal buffer") - } -} diff --git a/backend/internal/terminal/session.go b/backend/internal/terminal/session.go deleted file mode 100644 index f0ab25fbf..000000000 --- a/backend/internal/terminal/session.go +++ /dev/null @@ -1,344 +0,0 @@ -package terminal - -import ( - "context" - "errors" - "io" - "log/slog" - "sync" - "time" - - "github.com/aoagents/agent-orchestrator/backend/internal/ports" -) - -// PTYSource is what a terminal needs from the runtime: the argv that attaches a -// PTY to a session's pane, and a liveness check used to decide whether a dropped -// PTY should be re-attached or treated as a clean exit. The Zellij runtime adapter -// satisfies this via AttachCommand/IsAlive; the interface lives here, next to its -// only consumer, so terminal does not depend on a concrete adapter. -type PTYSource interface { - AttachCommand(handle ports.RuntimeHandle) ([]string, error) - IsAlive(ctx context.Context, handle ports.RuntimeHandle) (bool, error) -} - -// ptyProcess is a started PTY-backed attach process. It is the injection seam -// that keeps fan-out, buffering, and re-attach testable without a real process: -// unit tests supply a scripted in-memory implementation; production uses a -// creack/pty-backed one (see pty_unix.go). -type ptyProcess interface { - io.ReadWriteCloser - Resize(rows, cols uint16) error -} - -// spawnFunc starts a PTY for argv. ctx cancellation must terminate the process. -type spawnFunc func(ctx context.Context, argv []string) (ptyProcess, error) - -// reattach policy: a PTY that drops is re-attached while the underlying Zellij -// session is still alive, up to maxReattach consecutive failures. An attach that -// survived longer than reattachResetGrace before dropping resets the counter, so -// a long-lived pane that blips recovers but a tight crash-loop gives up. -const ( - defaultMaxReattach = 5 - defaultReattachResetTime = 5 * time.Second -) - -// subscriber receives one terminal's output frames. It must not block: session -// fan-out calls subscribers while serializing replay/delivery under its mutex, -// so the WS layer funnels frames onto its own buffered writer. -type subscriber func(data []byte) - -// session is one attached terminal pane, fanned out to N subscribers. It owns a -// single PTY (re-attached on drop) and a replay ring buffer. -type session struct { - id string - handle ports.RuntimeHandle - src PTYSource - spawn spawnFunc - log *slog.Logger - ring *ringBuffer - - maxReattach int - resetGrace time.Duration - - mu sync.Mutex - pty ptyProcess - subs map[int]subscriber - exitSubs map[int]func() - nextSub int - closed bool - exited bool - - doneOnce sync.Once - done chan struct{} -} - -func newSession(id string, handle ports.RuntimeHandle, src PTYSource, spawn spawnFunc, log *slog.Logger) *session { - if log == nil { - log = slog.Default() - } - return &session{ - id: id, - handle: handle, - src: src, - spawn: spawn, - log: log, - ring: newRingBuffer(defaultRingMax), - maxReattach: defaultMaxReattach, - resetGrace: defaultReattachResetTime, - subs: map[int]subscriber{}, - exitSubs: map[int]func(){}, - done: make(chan struct{}), - } -} - -// run drives attach → read-loop → re-attach until the pane exits cleanly, the -// session is closed, or ctx is cancelled. It is started once per session. -func (s *session) run(ctx context.Context) { - defer s.markDone() - - failures := 0 - for { - if s.isClosed() || ctx.Err() != nil { - return - } - - // Gate EVERY attach (including the first) on the runtime actually - // being alive. `zellij attach` resurrects EXITED sessions — re-running - // the serialized agent command — so attaching to a dead handle would - // re-create a runtime the daemon already destroyed, outside lifecycle - // control. A definitive "not alive" is a clean exit. A probe ERROR is - // not proof of death: it retries with backoff up to the same - // consecutive-failure cap as attach failures. - alive, err := s.src.IsAlive(ctx, s.handle) - if err != nil { - failures++ - if failures > s.maxReattach { - s.fail("liveness probe: " + err.Error()) - return - } - if !s.backoff(ctx, failures) { - return - } - continue - } - if !alive { - s.markExited() - return - } - - argv, err := s.src.AttachCommand(s.handle) - if err != nil { - s.fail("attach command: " + err.Error()) - return - } - p, err := s.spawn(ctx, argv) - if err != nil { - failures++ - if failures > s.maxReattach { - s.fail("spawn pty: " + err.Error()) - return - } - if !s.backoff(ctx, failures) { - return - } - continue - } - - s.setPTY(p) - start := time.Now() - s.copyOut(p) - _ = p.Close() - - if time.Since(start) >= s.resetGrace { - failures = 0 - } - failures++ - - if failures > s.maxReattach { - s.markExited() - return - } - if !s.backoff(ctx, failures) { - return - } - s.log.Debug("terminal re-attaching", "id", s.id, "failures", failures) - } -} - -// copyOut pumps PTY output into the ring buffer and out to subscribers until the -// PTY closes or errors. -func (s *session) copyOut(p ptyProcess) { - buf := make([]byte, 32*1024) - for { - n, err := p.Read(buf) - if n > 0 { - chunk := make([]byte, n) - copy(chunk, buf[:n]) - s.deliver(chunk) - } - if err != nil { - return - } - } -} - -// backoff sleeps between attach attempts; false means ctx was cancelled. -// Whether another attempt is warranted at all (liveness, failure cap) is -// decided at the top of the run loop, so a re-attach and a first attach share -// one gate. -func (s *session) backoff(ctx context.Context, failures int) bool { - select { - case <-ctx.Done(): - return false - case <-time.After(reattachBackoff(failures)): - return true - } -} - -func reattachBackoff(failures int) time.Duration { - d := time.Duration(failures) * 200 * time.Millisecond - if d > time.Second { - d = time.Second - } - return d -} - -// subscribe registers an output callback and an exit callback, replays the ring -// buffer to the new subscriber, and returns an unsubscribe func. If the pane has -// already exited, onExit fires immediately and exited is true; the caller must -// not treat the returned no-op unsubscribe as a live registration (there is -// nothing to track and re-opening must stay possible). -func (s *session) subscribe(onData subscriber, onExit func()) (unsubscribe func(), exited bool) { - s.mu.Lock() - if s.exited { - s.mu.Unlock() - if onExit != nil { - onExit() - } - return func() {}, true - } - id := s.nextSub - s.nextSub++ - s.subs[id] = onData - if onExit != nil { - s.exitSubs[id] = onExit - } - // Deliver the replay while still holding s.mu. deliver (the copyOut path) - // also takes s.mu around append+fanout, so the two are fully serialized: a - // chunk is either in this snapshot (and was fanned out before this - // subscriber registered) or it is fanned out after this returns, never both. - // Releasing the lock before replaying would let a chunk land in both the - // snapshot and a concurrent fanout, delivering it twice (or let an exit - // frame overtake the replay). onData is a non-blocking enqueue, so holding - // the lock across it cannot deadlock. - replay := s.ring.snapshot() - if len(replay) > 0 { - onData(replay) - } - s.mu.Unlock() - - return func() { - s.mu.Lock() - delete(s.subs, id) - delete(s.exitSubs, id) - s.mu.Unlock() - }, false -} - -// deliver appends a chunk to the ring and fans it out to current subscribers as -// one atomic step under s.mu. Holding the lock across both is what lets -// subscribe (which snapshots + replays under the same lock) guarantee -// exactly-once delivery: append+fanout and register+snapshot+replay can never -// interleave. Each fn is a non-blocking enqueue, so the lock is held only -// briefly and cannot deadlock. -func (s *session) deliver(chunk []byte) { - s.mu.Lock() - defer s.mu.Unlock() - s.ring.append(chunk) - for _, fn := range s.subs { - fn(chunk) - } -} - -// write sends client keystrokes to the PTY. It is a no-op if no PTY is attached. -func (s *session) write(p []byte) error { - s.mu.Lock() - pty := s.pty - s.mu.Unlock() - if pty == nil { - return errors.New("terminal: no active pty") - } - _, err := pty.Write(p) - return err -} - -func (s *session) resize(rows, cols uint16) error { - s.mu.Lock() - pty := s.pty - s.mu.Unlock() - if pty == nil { - return nil - } - return pty.Resize(rows, cols) -} - -func (s *session) setPTY(p ptyProcess) { - s.mu.Lock() - s.pty = p - s.mu.Unlock() -} - -// close tears the session down: stop re-attaching and kill the PTY. -func (s *session) close() { - s.mu.Lock() - if s.closed { - s.mu.Unlock() - return - } - s.closed = true - pty := s.pty - s.pty = nil - s.mu.Unlock() - if pty != nil { - _ = pty.Close() - } -} - -func (s *session) isClosed() bool { - s.mu.Lock() - defer s.mu.Unlock() - return s.closed -} - -func (s *session) isExited() bool { - s.mu.Lock() - defer s.mu.Unlock() - return s.exited -} - -// markExited flips the pane to exited and notifies/clears subscribers. -func (s *session) markExited() { - s.mu.Lock() - if s.exited { - s.mu.Unlock() - return - } - s.exited = true - exits := make([]func(), 0, len(s.exitSubs)) - for _, fn := range s.exitSubs { - exits = append(exits, fn) - } - s.exitSubs = map[int]func(){} - s.mu.Unlock() - for _, fn := range exits { - fn() - } -} - -// fail reports an unrecoverable attach error to subscribers as an exit. -func (s *session) fail(reason string) { - s.log.Warn("terminal session failed", "id", s.id, "reason", reason) - s.markExited() -} - -func (s *session) markDone() { s.doneOnce.Do(func() { close(s.done) }) } diff --git a/backend/internal/terminal/session_integration_test.go b/backend/internal/terminal/session_integration_test.go deleted file mode 100644 index 09c952d76..000000000 --- a/backend/internal/terminal/session_integration_test.go +++ /dev/null @@ -1,62 +0,0 @@ -//go:build !windows - -package terminal - -import ( - "context" - "os" - "os/exec" - "path/filepath" - "strconv" - "strings" - "testing" - "time" - - "github.com/aoagents/agent-orchestrator/backend/internal/adapters/runtime/zellij" - "github.com/aoagents/agent-orchestrator/backend/internal/domain" - "github.com/aoagents/agent-orchestrator/backend/internal/ports" -) - -// TestSessionStreamsRealZellijPane attaches a real PTY to a real Zellij session and -// asserts output streams back, then that killing the pane stops the session -// without a re-attach storm. Skipped when Zellij is unavailable. -func TestSessionStreamsRealZellijPane(t *testing.T) { - zellijBin, err := exec.LookPath("zellij") - if err != nil { - t.Skip("zellij unavailable") - } - - name := "ao-term-it-" + strconv.Itoa(os.Getpid()) - socketDir := filepath.Join("/tmp", name+"-socket") - if err := os.MkdirAll(socketDir, 0o755); err != nil { - t.Fatalf("mkdir socket dir: %v", err) - } - rt := zellij.New(zellij.Options{Binary: zellijBin, SocketDir: socketDir, ConfigDir: t.TempDir(), Timeout: 5 * time.Second}) - handle, err := rt.Create(context.Background(), ports.RuntimeConfig{ - SessionID: domain.SessionID(name), - WorkspacePath: t.TempDir(), - Argv: []string{"printf", "AO_READY\\n"}, - }) - if err != nil { - t.Fatalf("Create: %v", err) - } - t.Cleanup(func() { _ = rt.Destroy(context.Background(), handle) }) - - s := newSession(name, handle, rt, defaultSpawn, testLogger()) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go s.run(ctx) - - var got safeBytes - s.subscribe(got.add, nil) - - // Type a unique marker and expect it echoed back through the PTY. - eventually(t, 3*time.Second, func() bool { return s.write([]byte("echo AO_MARKER_42\n")) == nil }) - eventually(t, 5*time.Second, func() bool { return strings.Contains(got.string(), "AO_MARKER_42") }) - - // Kill the session: the terminal session must observe it as gone and not re-attach. - if err := rt.Destroy(context.Background(), handle); err != nil { - t.Fatalf("Destroy: %v", err) - } - eventually(t, 5*time.Second, func() bool { return s.isExited() }) -} diff --git a/backend/internal/terminal/session_test.go b/backend/internal/terminal/session_test.go deleted file mode 100644 index 4de33d305..000000000 --- a/backend/internal/terminal/session_test.go +++ /dev/null @@ -1,201 +0,0 @@ -package terminal - -import ( - "context" - "io" - "log/slog" - "testing" - "time" - - "github.com/aoagents/agent-orchestrator/backend/internal/ports" -) - -func testLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } - -func newTestSession(src PTYSource, spawn spawnFunc) *session { - return newSession("t1", ports.RuntimeHandle{ID: "t1"}, src, spawn, testLogger()) -} - -func TestSessionFansOutLiveOutputToSubscribers(t *testing.T) { - src := &fakeSource{alive: true} - pty := newFakePTY() - sp := &fakeSpawner{ptys: []*fakePTY{pty}} - s := newTestSession(src, sp.spawn) - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go s.run(ctx) - - var a, b safeBytes - s.subscribe(a.add, nil) - s.subscribe(b.add, nil) - - pty.push([]byte("hello")) - eventually(t, time.Second, func() bool { return a.string() == "hello" && b.string() == "hello" }) -} - -func TestSessionReplaysRingBufferOnSubscribe(t *testing.T) { - src := &fakeSource{alive: true} - pty := newFakePTY() - sp := &fakeSpawner{ptys: []*fakePTY{pty}} - s := newTestSession(src, sp.spawn) - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go s.run(ctx) - - pty.push([]byte("scrollback")) - eventually(t, time.Second, func() bool { return ringLen(s) == len("scrollback") }) - - var late safeBytes - s.subscribe(late.add, nil) - eventually(t, time.Second, func() bool { return late.string() == "scrollback" }) -} - -func ringLen(s *session) int { - s.mu.Lock() - defer s.mu.Unlock() - return len(s.ring.snapshot()) -} - -func TestSessionWriteAndResizeReachPTY(t *testing.T) { - src := &fakeSource{alive: true} - pty := newFakePTY() - sp := &fakeSpawner{ptys: []*fakePTY{pty}} - s := newTestSession(src, sp.spawn) - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go s.run(ctx) - - eventually(t, time.Second, func() bool { return s.write([]byte("ls\n")) == nil }) - eventually(t, time.Second, func() bool { return string(pty.writtenBytes()) == "ls\n" }) - - if err := s.resize(24, 80); err != nil { - t.Fatalf("resize: %v", err) - } - eventually(t, time.Second, func() bool { - rs := pty.resizeCalls() - return len(rs) == 1 && rs[0] == [2]uint16{24, 80} - }) -} - -func TestSessionSkipsReattachOnCleanExit(t *testing.T) { - src := &fakeSource{alive: true} // alive for the first attach - pty := newFakePTY() - sp := &fakeSpawner{ptys: []*fakePTY{pty}} - s := newTestSession(src, sp.spawn) - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - exited := make(chan struct{}) - go s.run(ctx) - s.subscribe(func([]byte) {}, func() { close(exited) }) - - eventually(t, time.Second, func() bool { return sp.calls() == 1 }) - src.setAlive(false) // Zellij session gone -> no re-attach - pty.Close() // pane ends - select { - case <-exited: - case <-time.After(time.Second): - t.Fatal("expected exit callback after clean pane exit") - } - if got := sp.calls(); got != 1 { - t.Fatalf("expected exactly one attach, got %d", got) - } -} - -// TestSessionNeverAttachesToDeadRuntime covers the resurrection bug: `zellij -// attach` on a killed-but-cached session resurrects it, re-running the agent -// command. A session whose runtime probes definitively dead must therefore -// report exited WITHOUT ever spawning an attach PTY — even on the very first -// open (the original code only checked liveness on re-attach). -func TestSessionNeverAttachesToDeadRuntime(t *testing.T) { - src := &fakeSource{alive: false} - sp := &fakeSpawner{} - s := newTestSession(src, sp.spawn) - - exited := make(chan struct{}) - s.subscribe(func([]byte) {}, func() { close(exited) }) - - go s.run(context.Background()) - select { - case <-exited: - case <-time.After(time.Second): - t.Fatal("expected exit when runtime is dead before first attach") - } - if got := sp.calls(); got != 0 { - t.Fatalf("attach must never run against a dead runtime, got %d spawns", got) - } -} - -// TestSessionRetriesProbeErrorsBeforeAttaching pins the hard rule that a -// failed liveness probe is NOT proof of death: a transient probe error must -// not flip the terminal to exited, and the attach proceeds once the probe -// recovers. -func TestSessionRetriesProbeErrorsBeforeAttaching(t *testing.T) { - src := &fakeSource{aliveErr: io.ErrUnexpectedEOF} - pty := newFakePTY() - sp := &fakeSpawner{ptys: []*fakePTY{pty}} - s := newTestSession(src, sp.spawn) - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go s.run(ctx) - - // While probes error the session must neither exit nor attach. - time.Sleep(50 * time.Millisecond) - if s.isExited() { - t.Fatal("probe error must not be treated as runtime death") - } - if got := sp.calls(); got != 0 { - t.Fatalf("attach must wait for a successful probe, got %d spawns", got) - } - - // Probe recovers -> the attach goes through. - src.setAliveResult(true, nil) - eventually(t, 2*time.Second, func() bool { return sp.calls() == 1 }) - if s.isExited() { - t.Fatal("session exited despite a live runtime") - } -} - -func TestSessionReattachesWhileSessionAlive(t *testing.T) { - src := &fakeSource{alive: true} // session still alive -> re-attach on drop - p1, p2 := newFakePTY(), newFakePTY() - sp := &fakeSpawner{ptys: []*fakePTY{p1, p2}} - s := newTestSession(src, sp.spawn) - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go s.run(ctx) - - eventually(t, time.Second, func() bool { return sp.calls() >= 1 }) - p1.Close() // first attach drops - eventually(t, 2*time.Second, func() bool { return sp.calls() >= 2 }) - - // Now the session is gone: the next drop must not re-attach. - src.setAlive(false) - p2.Close() - eventually(t, 2*time.Second, func() bool { return s.isExited() }) -} - -func TestSessionFailsWhenAttachCommandErrors(t *testing.T) { - src := &fakeSource{alive: true, attachErr: io.ErrUnexpectedEOF} - sp := &fakeSpawner{} - s := newTestSession(src, sp.spawn) - - exited := make(chan struct{}) - s.subscribe(func([]byte) {}, func() { close(exited) }) - - go s.run(context.Background()) - select { - case <-exited: - case <-time.After(time.Second): - t.Fatal("expected exit when attach command fails") - } - if sp.calls() != 0 { - t.Fatalf("spawn should not run when attach command errors, got %d calls", sp.calls()) - } -} diff --git a/docs/backend-code-structure.md b/docs/backend-code-structure.md index 68d9e008e..efee5d58f 100644 --- a/docs/backend-code-structure.md +++ b/docs/backend-code-structure.md @@ -237,15 +237,16 @@ Does not belong here: ### `internal/terminal` -`terminal` owns the terminal session protocol and PTY/session management used by -the HTTP mux. +`terminal` owns the terminal session protocol and PTY attach management used by +the HTTP mux. Every client that opens a pane gets its own `zellij attach` PTY — +zellij owns screen state and scrollback and replays its init handshake + full +repaint per attach, so there is no shared per-pane buffer. Belongs here: -- terminal session lifecycle; +- per-client attachment lifecycle (liveness gating, re-attach backoff); - input/output framing independent of HTTP; -- PTY-backed session handling; -- ring buffers and terminal protocol tests. +- PTY-backed attach handling and terminal protocol tests. `httpd` adapts WebSocket connections to terminal interfaces; `terminal` should not import `httpd`. diff --git a/frontend/src/renderer/components/TerminalPane.tsx b/frontend/src/renderer/components/TerminalPane.tsx index 0adc146ac..4ec8330ed 100644 --- a/frontend/src/renderer/components/TerminalPane.tsx +++ b/frontend/src/renderer/components/TerminalPane.tsx @@ -56,12 +56,12 @@ function AttachedTerminal({ session, theme, daemonReady }: TerminalPaneProps) { if (!terminal) return; // Reuse means the previous session's screen would linger; clear before // re-pointing. Screen-clear only, never reset(): every pane PTY is - // `zellij attach` with identical modes, and a full RIS would wipe the - // mouse-tracking mode zellij enabled at attach — the 50KB ring replay - // can't re-enable it, leaving wheel scroll dead after the first session - // switch (yyork's frozen-scroll regression, solved there the same way). - // Skipped on the very first attachment: the buffer is empty and the first - // fit may not have run yet. + // `zellij attach` with identical modes, so the previous session's mouse + // tracking stays valid while the new attach's handshake + repaint stream + // in — a full RIS would leave wheel scroll dead for that window (yyork's + // frozen-scroll regression, solved there the same way). Skipped on the + // very first attachment: the buffer is empty and the first fit may not + // have run yet. if (hadAttachmentRef.current) { terminal.clear(); } diff --git a/frontend/src/renderer/components/XtermTerminal.tsx b/frontend/src/renderer/components/XtermTerminal.tsx index a27c5e8e1..fe1e3cf03 100644 --- a/frontend/src/renderer/components/XtermTerminal.tsx +++ b/frontend/src/renderer/components/XtermTerminal.tsx @@ -64,12 +64,11 @@ function loadRenderer(term: Terminal): void { const terminalThemes = buildTerminalThemes(); // Erase scrollback (3J) + display (2J) and home the cursor — yyork's -// terminalResetSequence. Deliberately NOT term.reset(): a full RIS also wipes -// the DEC private modes zellij enabled when its attach process started (SGR -// mouse tracking, alt screen), and the mux's 50KB ring replay no longer -// contains those init sequences — after a RIS, xterm never re-enters -// mouse-tracking mode, wheel events stop being forwarded to zellij, and -// scrolling goes dead. +// terminalResetSequence. Deliberately NOT term.reset(): every pane PTY is a +// fresh per-client `zellij attach` whose handshake re-asserts the DEC private +// modes (SGR mouse tracking, alt screen) anyway, but a full RIS would drop +// them for the window until that handshake arrives — a flash where wheel +// events stop reaching zellij. The clear only wipes pixels; modes stay up. const CLEAR_SEQUENCE = "\x1b[3J\x1b[2J\x1b[H"; export function XtermTerminal(props: XtermTerminalProps) { diff --git a/frontend/src/renderer/hooks/useTerminalSession.test.tsx b/frontend/src/renderer/hooks/useTerminalSession.test.tsx index 882ad7e52..06f76aaf5 100644 --- a/frontend/src/renderer/hooks/useTerminalSession.test.tsx +++ b/frontend/src/renderer/hooks/useTerminalSession.test.tsx @@ -171,7 +171,7 @@ describe("useTerminalSession", () => { expect(muxes[0].resizes).toContainEqual(["handle-1", 120, 40]); }); - it("collapses a drag's burst of grid changes into one trailing PTY resize", () => { + it("collapses a drag's burst of grid changes into one trailing PTY resize, then re-asserts it", () => { const { terminal, muxes } = setup(); const initialResizes = muxes[0].resizes.length; // connect() sends the opening size terminal.emitResize(100, 30); @@ -179,6 +179,29 @@ describe("useTerminalSession", () => { terminal.emitResize(120, 40); act(() => void vi.advanceTimersByTime(100)); expect(muxes[0].resizes.slice(initialResizes)).toEqual([["handle-1", 120, 40]]); + // The settled grid goes out once more: paired with the backend's explicit + // SIGWINCH (pty_unix.go) it re-syncs a zellij client that lost the + // original update, which otherwise kept the session laid out for the old + // size until the next real grid change. + act(() => void vi.advanceTimersByTime(250)); + expect(muxes[0].resizes.slice(initialResizes)).toEqual([ + ["handle-1", 120, 40], + ["handle-1", 120, 40], + ]); + }); + + it("a new resize burst supersedes a pending re-assert", () => { + const { terminal, muxes } = setup(); + const initialResizes = muxes[0].resizes.length; + terminal.emitResize(100, 30); + act(() => void vi.advanceTimersByTime(100)); // settles -> sent, re-assert pending + terminal.emitResize(120, 40); // user keeps dragging before the re-assert fires + act(() => void vi.advanceTimersByTime(100 + 250)); + expect(muxes[0].resizes.slice(initialResizes)).toEqual([ + ["handle-1", 100, 30], + ["handle-1", 120, 40], + ["handle-1", 120, 40], + ]); }); it("marks exit in the terminal and refetches workspace state instead of writing status", () => { @@ -208,7 +231,7 @@ describe("useTerminalSession", () => { act(() => void vi.advanceTimersByTime(500)); expect(muxes).toHaveLength(2); expect(muxes[0].disposed).toBe(true); - expect(terminal.clears).toBe(1); // the server replays the output ring on open + expect(terminal.clears).toBe(1); // the fresh zellij attach repaints over a blank grid expect(muxes[1].opens).toEqual([["handle-1", 80, 24]]); act(() => muxes[1].emitOpened("handle-1")); expect(view.result.current.state).toBe("attached"); diff --git a/frontend/src/renderer/hooks/useTerminalSession.ts b/frontend/src/renderer/hooks/useTerminalSession.ts index 4d2a240c7..8829c70b0 100644 --- a/frontend/src/renderer/hooks/useTerminalSession.ts +++ b/frontend/src/renderer/hooks/useTerminalSession.ts @@ -25,9 +25,9 @@ export type AttachableTerminal = { writeln: (line: string) => void; /** * Erase screen + scrollback and home the cursor, preserving terminal modes. - * Never a full reset (RIS): that would drop the mouse-tracking mode zellij - * enabled at attach, which the ring replay cannot re-establish — killing - * wheel scroll (see XtermTerminal's CLEAR_SEQUENCE). + * Never a full reset (RIS): that would drop zellij's mouse-tracking mode + * for the gap until the fresh attach's handshake re-asserts it — a window + * with wheel scroll dead (see XtermTerminal's CLEAR_SEQUENCE). */ clear: () => void; onData: (listener: (data: string) => void) => { dispose: () => void }; @@ -55,6 +55,15 @@ const RETRY_MAX_MS = 8_000; // sizes; the attached program should get one SIGWINCH when the drag settles, // not dozens (yyork's terminal-panel does the same at its socket layer). const RESIZE_DEBOUNCE_MS = 100; +// One follow-up frame with the same grid after each settled resize. xterm only +// fires onResize on actual grid changes and the kernel only raises SIGWINCH on +// actual size changes, so a resize update the zellij client loses (raced +// mid-attach, coalesced during a drag) would otherwise desync the session's +// layout from the pane until the NEXT real change — the terminal keeps +// painting at the old size. The backend answers every resize frame with an +// explicit SIGWINCH (pty_unix.go), so this re-assert makes the client re-read +// and re-report its grid; when everything is already in sync it's a no-op. +const RESIZE_REASSERT_MS = 250; function defaultCreateMux(): TerminalMux { // Resolved per connect, not per hook: a daemon restart can change the port. @@ -161,11 +170,16 @@ export function useTerminalSession(session: WorkspaceSession | undefined, option const input = terminal.onData((data) => mux.sendInput(handle, data)); // xterm only fires onResize when the grid actually changed; the debounce // additionally collapses a drag's burst of changes into one PTY resize. + // Each settled resize is re-asserted once (see RESIZE_REASSERT_MS); both + // stages share resizeTimer so a new burst or teardown cancels either. const resize = terminal.onResize(({ cols, rows }) => { if (r.resizeTimer) clearTimeout(r.resizeTimer); r.resizeTimer = setTimeout(() => { - r.resizeTimer = null; mux.resize(handle, cols, rows); + r.resizeTimer = setTimeout(() => { + r.resizeTimer = null; + mux.resize(handle, cols, rows); + }, RESIZE_REASSERT_MS); }, RESIZE_DEBOUNCE_MS); }); r.disposers.push( @@ -174,10 +188,11 @@ export function useTerminalSession(session: WorkspaceSession | undefined, option ); // Connection status is chrome (the pane's banner), never buffer content — - // the PTY owns the buffer. On reattach the server replays the recent-output - // ring (backend internal/terminal/ring.go); clear the stale screen so it - // isn't doubled. Screen-clear only, never reset(): RIS would wipe zellij's - // mouse-tracking mode and freeze wheel scrolling after every reconnect. + // the PTY owns the buffer. Each open spawns a fresh server-side `zellij + // attach` (backend internal/terminal/attachment.go) that answers with its + // init handshake + a full repaint; clear the stale screen so the repaint + // lands on a blank grid. Screen-clear only, never reset(): RIS would drop + // zellij's mouse-tracking mode until the handshake lands. if (!r.firstAttach) { terminal.clear(); }