fix(client): drop keys kept for a session when swapping to another
Only a resume of the same target replays kept input; a picker swap must not run keys typed for the previous session in the next one.
- commit
- 8bf6829b7e89ae4af58ff5a0f5be3a3983ca6b22
- author
- brice · · 2026-10-07 07:20 UTC
- parent
- 3980b13
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +55 -0
M internal/usecase/client/supervisor_attach.go +5 -0
| @@ -607,6 +607,11 @@ func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalI | |||
| 607 | 607 | // attachment detached cleanly, and the swap goes through Connecting | |
| 608 | 608 | // exactly like any committed selection. | |
| 609 | 609 | budget = attachmentResumeBudget{} | |
| 610 | + | // Keys kept for the previous session must not run in the next one; | |
| 611 | + | // only a resume of the same target replays them. | |
| 612 | + | if input != nil && input.pump != nil { | |
| 613 | + | input.pump.dropUnclaimed() | |
| 614 | + | } | |
| 610 | 615 | next, ok := s.withFreshStream(service, outcome.target) | |
| 611 | 616 | if !ok { | |
| 612 | 617 | return false, nil | |
| @@ -607,6 +607,11 @@ func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalI | |||||
| 607 | // attachment detached cleanly, and the swap goes through Connecting | 607 | // attachment detached cleanly, and the swap goes through Connecting | ||
| 608 | // exactly like any committed selection. | 608 | // exactly like any committed selection. | ||
| 609 | budget = attachmentResumeBudget{} | 609 | budget = attachmentResumeBudget{} | ||
| 610 | + | // Keys kept for the previous session must not run in the next one; | |||
| 611 | + | // only a resume of the same target replays them. | |||
| 612 | + | if input != nil && input.pump != nil { | |||
| 613 | + | input.pump.dropUnclaimed() | |||
| 614 | + | } | |||
| 610 | next, ok := s.withFreshStream(service, outcome.target) | 615 | next, ok := s.withFreshStream(service, outcome.target) | ||
| 611 | if !ok { | 616 | if !ok { | ||
| 612 | return false, nil | 617 | return false, nil | ||
M internal/usecase/client/supervisor_attach_test.go +50 -0
| @@ -604,6 +604,56 @@ func TestSupervisorPendingSwapContinuesRatherThanExiting(t *testing.T) { | |||
| 604 | 604 | } | |
| 605 | 605 | } | |
| 606 | 606 | ||
| 607 | + | // TestSupervisorSwapDropsKeysKeptForPreviousSession types into a session whose | |
| 608 | + | // link stalls, then swaps to another session: the stalled keys were meant for | |
| 609 | + | // the first session and must never run in the second. | |
| 610 | + | func TestSupervisorSwapDropsKeysKeptForPreviousSession(t *testing.T) { | |
| 611 | + | picker := newAttachTestPicker() | |
| 612 | + | harness := startAttachHarness(t, picker) | |
| 613 | + | base := newSessionTestStream() | |
| 614 | + | first := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})} | |
| 615 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 616 | + | return first, nil | |
| 617 | + | }) | |
| 618 | + | picker.commit(sessionTestRequest(true)) | |
| 619 | + | deliverReadyStream(t, base) | |
| 620 | + | awaitAttachedState(t, harness.sup) | |
| 621 | + | require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond) | |
| 622 | + | first.stall.Store(true) | |
| 623 | + | harness.reader.push([]byte("rm -rf build\r")) | |
| 624 | + | <-first.entered | |
| 625 | + | ||
| 626 | + | var mu sync.Mutex | |
| 627 | + | second := newSessionTestStream() | |
| 628 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |
| 629 | + | return second, nil | |
| 630 | + | }) | |
| 631 | + | base.deliver(navigationOffer(1)) | |
| 632 | + | awaitPresentation(t, harness.sup, PresentAttachedPicker) | |
| 633 | + | other := sessionTestRequest(true) | |
| 634 | + | other.Target = protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{2}, SessionName: "beta"} | |
| 635 | + | picker.recordOp(pickerOp{commit: true}, "row", other) | |
| 636 | + | // The Detach queues behind the stalled write; let the bounded drain expire | |
| 637 | + | // so the unsent keys are preserved, as on a dead link. | |
| 638 | + | for { | |
| 639 | + | timer := harness.clock.awaitTimer(t) | |
| 640 | + | if timer.delay == drainOutboxTimeout { | |
| 641 | + | timer.fire() | |
| 642 | + | break | |
| 643 | + | } | |
| 644 | + | } | |
| 645 | + | awaitStreamHello(t, &mu, &second) | |
| 646 | + | second.deliver(protocol.Welcome{SessionName: "beta"}) | |
| 647 | + | output := sessionTestOutput(1, "\x1b[Hbeta") | |
| 648 | + | output.Context.Route.Target = other.Target | |
| 649 | + | second.deliver(output) | |
| 650 | + | awaitAttachedState(t, harness.sup) | |
| 651 | + | ||
| 652 | + | harness.reader.push([]byte("z")) | |
| 653 | + | require.Eventually(t, func() bool { return sentInput(second) == "z" }, 5*time.Second, time.Millisecond, | |
| 654 | + | "second session input = %q", sentInput(second)) | |
| 655 | + | } | |
| 656 | + | ||
| 607 | 657 | // awaitStreamHello waits until the admitted stream has sent its first client | |
| 608 | 658 | // message, polling the guarded stream pointer rather than a stale snapshot. | |
| 609 | 659 | func awaitStreamHello(t *testing.T, mu *sync.Mutex, stream **sessionTestStream) { | |
| @@ -604,6 +604,56 @@ func TestSupervisorPendingSwapContinuesRatherThanExiting(t *testing.T) { | |||||
| 604 | } | 604 | } | ||
| 605 | } | 605 | } | ||
| 606 | 606 | ||||
| 607 | + | // TestSupervisorSwapDropsKeysKeptForPreviousSession types into a session whose | |||
| 608 | + | // link stalls, then swaps to another session: the stalled keys were meant for | |||
| 609 | + | // the first session and must never run in the second. | |||
| 610 | + | func TestSupervisorSwapDropsKeysKeptForPreviousSession(t *testing.T) { | |||
| 611 | + | picker := newAttachTestPicker() | |||
| 612 | + | harness := startAttachHarness(t, picker) | |||
| 613 | + | base := newSessionTestStream() | |||
| 614 | + | first := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})} | |||
| 615 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 616 | + | return first, nil | |||
| 617 | + | }) | |||
| 618 | + | picker.commit(sessionTestRequest(true)) | |||
| 619 | + | deliverReadyStream(t, base) | |||
| 620 | + | awaitAttachedState(t, harness.sup) | |||
| 621 | + | require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond) | |||
| 622 | + | first.stall.Store(true) | |||
| 623 | + | harness.reader.push([]byte("rm -rf build\r")) | |||
| 624 | + | <-first.entered | |||
| 625 | + | ||||
| 626 | + | var mu sync.Mutex | |||
| 627 | + | second := newSessionTestStream() | |||
| 628 | + | harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) { | |||
| 629 | + | return second, nil | |||
| 630 | + | }) | |||
| 631 | + | base.deliver(navigationOffer(1)) | |||
| 632 | + | awaitPresentation(t, harness.sup, PresentAttachedPicker) | |||
| 633 | + | other := sessionTestRequest(true) | |||
| 634 | + | other.Target = protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{2}, SessionName: "beta"} | |||
| 635 | + | picker.recordOp(pickerOp{commit: true}, "row", other) | |||
| 636 | + | // The Detach queues behind the stalled write; let the bounded drain expire | |||
| 637 | + | // so the unsent keys are preserved, as on a dead link. | |||
| 638 | + | for { | |||
| 639 | + | timer := harness.clock.awaitTimer(t) | |||
| 640 | + | if timer.delay == drainOutboxTimeout { | |||
| 641 | + | timer.fire() | |||
| 642 | + | break | |||
| 643 | + | } | |||
| 644 | + | } | |||
| 645 | + | awaitStreamHello(t, &mu, &second) | |||
| 646 | + | second.deliver(protocol.Welcome{SessionName: "beta"}) | |||
| 647 | + | output := sessionTestOutput(1, "\x1b[Hbeta") | |||
| 648 | + | output.Context.Route.Target = other.Target | |||
| 649 | + | second.deliver(output) | |||
| 650 | + | awaitAttachedState(t, harness.sup) | |||
| 651 | + | ||||
| 652 | + | harness.reader.push([]byte("z")) | |||
| 653 | + | require.Eventually(t, func() bool { return sentInput(second) == "z" }, 5*time.Second, time.Millisecond, | |||
| 654 | + | "second session input = %q", sentInput(second)) | |||
| 655 | + | } | |||
| 656 | + | ||||
| 607 | // awaitStreamHello waits until the admitted stream has sent its first client | 657 | // awaitStreamHello waits until the admitted stream has sent its first client | ||
| 608 | // message, polling the guarded stream pointer rather than a stale snapshot. | 658 | // message, polling the guarded stream pointer rather than a stale snapshot. | ||
| 609 | func awaitStreamHello(t *testing.T, mu *sync.Mutex, stream **sessionTestStream) { | 659 | func awaitStreamHello(t *testing.T, mu *sync.Mutex, stream **sessionTestStream) { | ||