diff --git a/pkg/transport/lifecycle.go b/pkg/transport/lifecycle.go
index 3a6331336..58c0128cb 100644
--- a/pkg/transport/lifecycle.go
+++ b/pkg/transport/lifecycle.go
@@ -4,9 +4,13 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net"
+	"strings"
 	"sync"
+	"time"
 
 	"github.com/devsy-org/devsy/pkg/log"
+	"golang.org/x/crypto/ssh"
 )
 
 type CloseReason string
@@ -38,6 +42,7 @@ const (
 	TransportSideSSH      = SideSSH
 	TransportSideParent   = SideParent
 )
+const DefaultJoinTimeout = 5 * time.Second
 
 const (
 	TransportCloseUnknown           = CloseUnknown
@@ -153,6 +158,202 @@ type RunManagedOptions struct {
 	Handler       func(context.Context) error
 	Metadata      LogMetadata
 	TransportSide Side
+	JoinTimeout   time.Duration
+}
+
+type managedOutcome struct {
+	firstSide          Side
+	parentErr          error
+	handlerErr         error
+	transportErr       error
+	handlerCompleted   bool
+	transportCompleted bool
+}
+
+func isTeardownOrCancellationError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return isCancellationErr(err) || isClosedOrEOFErr(err)
+}
+
+func isCancellationErr(err error) bool {
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	return strings.Contains(err.Error(), "context canceled")
+}
+
+func isClosedOrEOFErr(err error) bool {
+	return isEOFErr(err) || isClosedNetErr(err)
+}
+
+func isEOFErr(err error) bool {
+	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) ||
+		errors.Is(err, io.ErrClosedPipe) {
+		return true
+	}
+	msg := err.Error()
+	return strings.Contains(msg, ": EOF") || strings.Contains(msg, "closed pipe")
+}
+
+func isClosedNetErr(err error) bool {
+	if errors.Is(err, net.ErrClosed) {
+		return true
+	}
+	return strings.Contains(err.Error(), "use of closed network connection")
+}
+
+func isExitMissingErr(err error) bool {
+	if err == nil {
+		return false
+	}
+	var exitMissing *ssh.ExitMissingError
+	if errors.As(err, &exitMissing) {
+		return true
+	}
+	return strings.Contains(err.Error(), "remote command exited without exit status or exit signal")
+}
+
+func isTransportTeardownError(err error) bool {
+	if err == nil {
+		return true
+	}
+	return isTeardownOrCancellationError(err) || isExitMissingErr(err)
+}
+
+func isMeaningfulHandlerErr(outcome managedOutcome) bool {
+	return outcome.handlerCompleted && outcome.handlerErr != nil &&
+		!isTeardownOrCancellationError(outcome.handlerErr)
+}
+
+func isGenuineProviderFailure(outcome managedOutcome) bool {
+	if !outcome.transportCompleted || outcome.transportErr == nil {
+		return false
+	}
+	return outcome.firstSide == SideProvider && !isTransportTeardownError(outcome.transportErr)
+}
+
+func resolveManagedErrors(outcome managedOutcome) error {
+	if outcome.firstSide == SideParent {
+		return outcome.parentErr
+	}
+	if isMeaningfulHandlerErr(outcome) {
+		return outcome.handlerErr
+	}
+	if isGenuineProviderFailure(outcome) {
+		return outcome.transportErr
+	}
+	if outcome.handlerCompleted && outcome.handlerErr == nil {
+		return nil
+	}
+	if err := resolveParentCancellation(outcome); err != nil {
+		return err
+	}
+	return resolveByFirstSide(outcome)
+}
+
+func resolveByFirstSide(outcome managedOutcome) error {
+	switch outcome.firstSide {
+	case SideSSH:
+		return resolveSSHFirst(outcome)
+	case SideProvider:
+		return resolveProviderFirst(outcome)
+	default:
+		return resolveFallback(outcome)
+	}
+}
+
+func resolveSSHFirst(outcome managedOutcome) error {
+	if !outcome.handlerCompleted || isTeardownOrCancellationError(outcome.handlerErr) {
+		if outcome.transportErr != nil {
+			return outcome.transportErr
+		}
+		return nil
+	}
+	return outcome.handlerErr
+}
+
+func resolveParentCancellation(outcome managedOutcome) error {
+	if outcome.parentErr != nil &&
+		(errors.Is(outcome.parentErr, context.Canceled) || errors.Is(outcome.parentErr, context.DeadlineExceeded)) {
+		if !outcome.handlerCompleted || isTeardownOrCancellationError(outcome.handlerErr) {
+			return outcome.parentErr
+		}
+	}
+	return nil
+}
+
+func resolveProviderFirst(outcome managedOutcome) error {
+	if !outcome.handlerCompleted || isTeardownOrCancellationError(outcome.handlerErr) {
+		return outcome.transportErr
+	}
+	return outcome.handlerErr
+}
+
+func resolveFallback(outcome managedOutcome) error {
+	if outcome.handlerCompleted && outcome.handlerErr != nil {
+		return outcome.handlerErr
+	}
+	if outcome.transportCompleted && outcome.transportErr != nil {
+		return outcome.transportErr
+	}
+	return outcome.parentErr
+}
+
+func waitForFirst(
+	parent context.Context,
+	transportSide Side,
+	handlerDone <-chan error,
+	connDone <-chan error,
+) (managedOutcome, error) {
+	var outcome managedOutcome
+	select {
+	case err := <-connDone:
+		outcome.firstSide = transportSide
+		outcome.transportErr = err
+		outcome.transportCompleted = true
+		return outcome, err
+	case err := <-handlerDone:
+		outcome.firstSide = SideSSH
+		outcome.handlerErr = err
+		outcome.handlerCompleted = true
+		return outcome, err
+	case <-parent.Done():
+		outcome.firstSide = SideParent
+		return outcome, parent.Err()
+	}
+}
+
+func initiateTeardown(conn ManagedConn, firstSide Side, handlerErr error) {
+	if firstSide == SideSSH && handlerErr == nil {
+		if cw, ok := conn.(CloseWriter); ok {
+			if err := cw.CloseWrite(); err == nil {
+				return
+			}
+		}
+	}
+	_ = conn.Close()
+}
+
+func joinRemaining(
+	joinCtx context.Context,
+	handlerDone <-chan error,
+	connDone <-chan error,
+	outcome *managedOutcome,
+) {
+	for !outcome.handlerCompleted || !outcome.transportCompleted {
+		select {
+		case err := <-handlerDone:
+			outcome.handlerErr = err
+			outcome.handlerCompleted = true
+		case err := <-connDone:
+			outcome.transportErr = err
+			outcome.transportCompleted = true
+		case <-joinCtx.Done():
+			return
+		}
+	}
 }
 
 func RunManaged(opts RunManagedOptions) error {
@@ -165,23 +366,28 @@ func RunManaged(opts RunManagedOptions) error {
 	if opts.Handler == nil {
 		return errors.New("handler is required")
 	}
+
+	joinTimeout := opts.JoinTimeout
+	if joinTimeout <= 0 {
+		joinTimeout = DefaultJoinTimeout
+	}
+
 	lifecycle, ctx := NewPersistentLifecycle(opts.Parent)
 	handlerDone := make(chan error, 1)
 	go func() { handlerDone <- opts.Handler(ctx) }()
 	connDone := make(chan error, 1)
 	go func() { connDone <- opts.Conn.Wait() }()
 
-	var info CloseInfo
-	select {
-	case err := <-connDone:
-		info = Classify(opts.TransportSide, err, opts.Parent)
-	case err := <-handlerDone:
-		info = Classify(SideSSH, err, opts.Parent)
-	case <-opts.Parent.Done():
-		info = Classify(SideParent, opts.Parent.Err(), opts.Parent)
-	}
-	lifecycle.Close(info)
+	outcome, firstErr := waitForFirst(opts.Parent, opts.TransportSide, handlerDone, connDone)
+	lifecycle.Close(Classify(outcome.firstSide, firstErr, opts.Parent))
+
+	initiateTeardown(opts.Conn, outcome.firstSide, outcome.handlerErr)
+	joinCtx, cancelJoin := context.WithTimeout(context.WithoutCancel(opts.Parent), joinTimeout)
+	defer cancelJoin()
+	joinRemaining(joinCtx, handlerDone, connDone, &outcome)
+
 	_ = opts.Conn.Close()
 	LogClose(lifecycle.CloseInfo(), opts.Metadata)
-	return lifecycle.CloseInfo().Err
+	outcome.parentErr = opts.Parent.Err()
+	return resolveManagedErrors(outcome)
 }
diff --git a/pkg/transport/lifecycle_test.go b/pkg/transport/lifecycle_test.go
index 7de355db2..3540819f9 100644
--- a/pkg/transport/lifecycle_test.go
+++ b/pkg/transport/lifecycle_test.go
@@ -3,7 +3,10 @@ package transport
 import (
 	"context"
 	"errors"
+	"fmt"
+	"io"
 	"net"
+	"sync"
 	"testing"
 	"time"
 )
@@ -111,6 +114,386 @@ func TestRunManagedReturnsParentCancellation(t *testing.T) {
 	}
 }
 
+type controlledManagedConn struct {
+	net.Conn
+	mu                  sync.Mutex
+	triggerOnce         sync.Once
+	closeOnce           sync.Once
+	waitErr             error
+	triggerWait         chan struct{}
+	waitResultPublished chan struct{}
+	closed              chan struct{}
+}
+
+func newControlledManagedConn() *controlledManagedConn {
+	return &controlledManagedConn{
+		Conn:                &stubConn{},
+		triggerWait:         make(chan struct{}),
+		waitResultPublished: make(chan struct{}),
+		closed:              make(chan struct{}),
+	}
+}
+
+func (c *controlledManagedConn) Wait() error {
+	<-c.triggerWait
+	close(c.waitResultPublished)
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.waitErr
+}
+
+func (c *controlledManagedConn) Close() error {
+	c.closeOnce.Do(func() {
+		close(c.closed)
+		c.triggerOnce.Do(func() { close(c.triggerWait) })
+	})
+	return nil
+}
+
+func (c *controlledManagedConn) TriggerCleanupError(err error) {
+	c.mu.Lock()
+	c.waitErr = err
+	c.mu.Unlock()
+	c.triggerOnce.Do(func() { close(c.triggerWait) })
+}
+
+func TestRunManagedHandlerSuccessBeatsCleanupError(t *testing.T) {
+	conn := newControlledManagedConn()
+	errTeardown := errors.New("wait: remote command exited without exit status or exit signal")
+
+	err := RunManaged(RunManagedOptions{
+		Parent:        context.Background(),
+		Conn:          conn,
+		TransportSide: SideProvider,
+		Handler: func(ctx context.Context) error {
+			conn.TriggerCleanupError(errTeardown)
+			<-ctx.Done()
+			return nil
+		},
+	})
+	if err != nil {
+		t.Fatalf("RunManaged() = %v, want nil", err)
+	}
+}
+
+func TestRunManagedHandlerErrorWinsOverCleanupError(t *testing.T) {
+	conn := newControlledManagedConn()
+	errTeardown := errors.New("wait: remote command exited without exit status or exit signal")
+	wantErr := errors.New("command exited with status 127")
+
+	err := RunManaged(RunManagedOptions{
+		Parent:        context.Background(),
+		Conn:          conn,
+		TransportSide: SideProvider,
+		Handler: func(ctx context.Context) error {
+			conn.TriggerCleanupError(errTeardown)
+			<-ctx.Done()
+			return wantErr
+		},
+	})
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("RunManaged() = %v, want %v", err, wantErr)
+	}
+}
+
+func TestRunManagedProviderFailureWinsOverLaterHandlerSuccess(t *testing.T) {
+	conn := newControlledManagedConn()
+	providerErr := errors.New("connection reset by peer")
+
+	err := RunManaged(RunManagedOptions{
+		Parent:        context.Background(),
+		Conn:          conn,
+		TransportSide: SideProvider,
+		Handler: func(ctx context.Context) error {
+			conn.TriggerCleanupError(providerErr)
+			<-ctx.Done()
+			return nil
+		},
+	})
+	if !errors.Is(err, providerErr) {
+		t.Fatalf("RunManaged() = %v, want %v", err, providerErr)
+	}
+}
+
+func TestRunManagedGenuineTransportFailure(t *testing.T) {
+	conn := newControlledManagedConn()
+	providerErr := errors.New("connection reset by peer")
+	conn.TriggerCleanupError(providerErr)
+
+	err := RunManaged(RunManagedOptions{
+		Parent:        context.Background(),
+		Conn:          conn,
+		TransportSide: SideProvider,
+		Handler: func(ctx context.Context) error {
+			<-ctx.Done()
+			return ctx.Err()
+		},
+	})
+	if !errors.Is(err, providerErr) {
+		t.Fatalf("RunManaged() = %v, want %v", err, providerErr)
+	}
+}
+
+func TestRunManagedStressHandlerSuccessBeatsCleanup(t *testing.T) {
+	errTeardown := errors.New("wait: remote command exited without exit status or exit signal")
+	for i := range 100 {
+		conn := newControlledManagedConn()
+		err := RunManaged(RunManagedOptions{
+			Parent:        context.Background(),
+			Conn:          conn,
+			TransportSide: SideProvider,
+			Handler: func(ctx context.Context) error {
+				conn.TriggerCleanupError(errTeardown)
+				<-conn.waitResultPublished
+				return nil
+			},
+		})
+		if err != nil {
+			t.Fatalf("iteration %d: RunManaged() = %v, want nil", i, err)
+		}
+	}
+}
+
+func TestRunManagedBoundedSecondSideShutdown(t *testing.T) {
+	conn := newTestManagedConn(nil)
+	handlerStuck := make(chan struct{})
+	defer close(handlerStuck)
+
+	done := make(chan error, 1)
+	start := time.Now()
+	go func() {
+		done <- RunManaged(RunManagedOptions{
+			Parent:        context.Background(),
+			Conn:          conn,
+			TransportSide: SideProvider,
+			JoinTimeout:   50 * time.Millisecond,
+			Handler: func(ctx context.Context) error {
+				<-handlerStuck
+				return nil
+			},
+		})
+	}()
+
+	_ = conn.Close()
+
+	select {
+	case <-done:
+		if time.Since(start) > 2*time.Second {
+			t.Fatal("RunManaged took too long to return after bounded join timeout")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("RunManaged hung waiting for second side")
+	}
+}
+
+type closeWriteFailingConn struct {
+	*testManagedConn
+	closeWriteCalled bool
+}
+
+func (c *closeWriteFailingConn) CloseWrite() error {
+	c.closeWriteCalled = true
+	return errors.ErrUnsupported
+}
+
+func TestRunManagedCloseWriteFailureFallsBackToClose(t *testing.T) {
+	conn := &closeWriteFailingConn{
+		testManagedConn: newTestManagedConn(nil),
+	}
+
+	err := RunManaged(RunManagedOptions{
+		Parent:        context.Background(),
+		Conn:          conn,
+		TransportSide: SideProvider,
+		JoinTimeout:   2 * time.Second,
+		Handler: func(ctx context.Context) error {
+			return nil
+		},
+	})
+	if err != nil {
+		t.Fatalf("RunManaged() = %v, want nil", err)
+	}
+	if !conn.closeWriteCalled {
+		t.Fatal("CloseWrite was not called")
+	}
+	select {
+	case <-conn.closed:
+	default:
+		t.Fatal("Close was not called after CloseWrite failed")
+	}
+}
+
+func TestResolveManagedErrors_SuccessAndCancellation(t *testing.T) {
+	errTeardown := errors.New("wait: remote command exited without exit status or exit signal")
+	errCanceled := context.Canceled
+
+	tests := []struct {
+		name    string
+		outcome managedOutcome
+		wantErr error
+	}{
+		{
+			name: "parent cancelled first wins",
+			outcome: managedOutcome{
+				firstSide:          SideParent,
+				parentErr:          errCanceled,
+				handlerErr:         errCanceled,
+				transportErr:       errTeardown,
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: errCanceled,
+		},
+		{
+			name: "handler success beats transport teardown when transport was first",
+			outcome: managedOutcome{
+				firstSide:          SideProvider,
+				handlerErr:         nil,
+				transportErr:       errTeardown,
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: nil,
+		},
+		{
+			name: "handler success beats transport teardown when handler was first",
+			outcome: managedOutcome{
+				firstSide:          SideSSH,
+				handlerErr:         nil,
+				transportErr:       errTeardown,
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := resolveManagedErrors(tt.outcome)
+			if !errors.Is(got, tt.wantErr) {
+				t.Fatalf("resolveManagedErrors() = %v, want %v", got, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestResolveManagedErrors_HandlerError(t *testing.T) {
+	errTeardown := errors.New("wait: remote command exited without exit status or exit signal")
+	errUser := errors.New("command exited with status 127")
+
+	tests := []struct {
+		name    string
+		outcome managedOutcome
+		wantErr error
+	}{
+		{
+			name: "handler error beats transport teardown when handler was first",
+			outcome: managedOutcome{
+				firstSide:          SideSSH,
+				handlerErr:         errUser,
+				transportErr:       errTeardown,
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: errUser,
+		},
+		{
+			name: "handler error beats transport teardown when transport was first",
+			outcome: managedOutcome{
+				firstSide:          SideProvider,
+				handlerErr:         errUser,
+				transportErr:       errTeardown,
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: errUser,
+		},
+		{
+			name: "transport error returned when handler did not complete in SideSSH",
+			outcome: managedOutcome{
+				firstSide:          SideSSH,
+				transportErr:       errTeardown,
+				handlerCompleted:   false,
+				transportCompleted: true,
+			},
+			wantErr: errTeardown,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := resolveManagedErrors(tt.outcome)
+			if !errors.Is(got, tt.wantErr) {
+				t.Fatalf("resolveManagedErrors() = %v, want %v", got, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestResolveManagedErrors_ProviderFailure(t *testing.T) {
+	errProvider := errors.New("provider reset")
+	errCanceled := context.Canceled
+
+	tests := []struct {
+		name    string
+		outcome managedOutcome
+		wantErr error
+	}{
+		{
+			name: "genuine provider failure wins over handler cancellation consequence",
+			outcome: managedOutcome{
+				firstSide:          SideProvider,
+				handlerErr:         errCanceled,
+				transportErr:       errProvider,
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: errProvider,
+		},
+		{
+			name: "genuine provider failure wins when handler timed out",
+			outcome: managedOutcome{
+				firstSide:          SideProvider,
+				transportErr:       errProvider,
+				handlerCompleted:   false,
+				transportCompleted: true,
+			},
+			wantErr: errProvider,
+		},
+		{
+			name: "genuine provider failure wins over later handler success",
+			outcome: managedOutcome{
+				firstSide:          SideProvider,
+				handlerErr:         nil,
+				transportErr:       errProvider,
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: errProvider,
+		},
+		{
+			name: "clean provider exit with handler EOF returns nil",
+			outcome: managedOutcome{
+				firstSide:          SideProvider,
+				handlerErr:         fmt.Errorf("read: %w", io.EOF),
+				handlerCompleted:   true,
+				transportCompleted: true,
+			},
+			wantErr: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := resolveManagedErrors(tt.outcome)
+			if !errors.Is(got, tt.wantErr) {
+				t.Fatalf("resolveManagedErrors() = %v, want %v", got, tt.wantErr)
+			}
+		})
+	}
+}
+
 type stubConn struct{}
 
 func (*stubConn) Read([]byte) (int, error)         { return 0, net.ErrClosed }