fix(runs): close the action watch stream only when the action itself is done

A timed-out attempt emits a terminal TIMED_OUT event for attempt N
before the action restarts as attempt N+1, and closing on attempt
terminality alone ended the WatchActionDetails stream mid-retry. Close
only when the action's phase is terminal and action_events has caught
up with it: the highest-numbered attempt is terminal and is the
action's current attempt.

Checked the other half of the concern too: the actions-table updater
already lists TIMED_OUT as a retryable phase, so the Queued row that
follows a mid-retry timeout is accepted — no stickiness to fix there.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: shaon-chowdhury-euc <shaon.chowdhury@eucalyptus.vc>

Author: shaon-chowdhury-euc

runs/service/action_stream_complete_test.go

@@ -0,0 +1,76 @@
+package service
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/common"
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow"
+)
+
+func details(statusPhase common.ActionPhase, statusAttempts uint32, attempts ...*workflow.ActionAttempt) *workflow.ActionDetails {
+	return &workflow.ActionDetails{
+		Status: &workflow.ActionStatus{
+			Phase:    statusPhase,
+			Attempts: statusAttempts,
+		},
+		Attempts: attempts,
+	}
+}
+
+func attempt(n uint32, phase common.ActionPhase) *workflow.ActionAttempt {
+	return &workflow.ActionAttempt{Attempt: n, Phase: phase}
+}
+
+func TestActionStreamComplete_MidRetryTimeoutKeepsStreamOpen(t *testing.T) {
+	// The #7910 review case: attempt 1 timed out (terminal event recorded), but the
+	// action is restarting as attempt 2. Neither ordering of the two writes may
+	// close the stream.
+
+	// Event landed first: terminal TIMED_OUT for attempt 1, actions row still RUNNING.
+	assert.False(t, actionStreamComplete(details(
+		common.ActionPhase_ACTION_PHASE_RUNNING, 1,
+		attempt(1, common.ActionPhase_ACTION_PHASE_TIMED_OUT),
+	)))
+
+	// Row moved first: action already QUEUED at attempt 2, last event still attempt 1.
+	assert.False(t, actionStreamComplete(details(
+		common.ActionPhase_ACTION_PHASE_QUEUED, 2,
+		attempt(1, common.ActionPhase_ACTION_PHASE_TIMED_OUT),
+	)))
+}
+
+func TestActionStreamComplete_TerminalTimeoutCloses(t *testing.T) {
+	assert.True(t, actionStreamComplete(details(
+		common.ActionPhase_ACTION_PHASE_TIMED_OUT, 2,
+		attempt(1, common.ActionPhase_ACTION_PHASE_TIMED_OUT),
+		attempt(2, common.ActionPhase_ACTION_PHASE_TIMED_OUT),
+	)))
+}
+
+func TestActionStreamComplete_SuccessCloses(t *testing.T) {
+	assert.True(t, actionStreamComplete(details(
+		common.ActionPhase_ACTION_PHASE_SUCCEEDED, 1,
+		attempt(1, common.ActionPhase_ACTION_PHASE_SUCCEEDED),
+	)))
+}
+
+func TestActionStreamComplete_ActionTerminalButEventsLagStaysOpen(t *testing.T) {
+	// The pre-existing eventual-consistency case the old predicate also guarded:
+	// the actions table is terminal but action_events has not caught up yet.
+	assert.False(t, actionStreamComplete(details(
+		common.ActionPhase_ACTION_PHASE_FAILED, 2,
+		attempt(1, common.ActionPhase_ACTION_PHASE_FAILED),
+	)))
+	assert.False(t, actionStreamComplete(details(
+		common.ActionPhase_ACTION_PHASE_FAILED, 1,
+		attempt(1, common.ActionPhase_ACTION_PHASE_RUNNING),
+	)))
+}
+
+func TestActionStreamComplete_NoAttemptsStaysOpen(t *testing.T) {
+	assert.False(t, actionStreamComplete(details(
+		common.ActionPhase_ACTION_PHASE_SUCCEEDED, 1,
+	)))
+}

runs/service/run_service.go

@@ -866,20 +866,29 @@ func IsTerminalPhase(phase common.ActionPhase) bool {
 		phase == common.ActionPhase_ACTION_PHASE_RECOVERED
 }
 
-// lastAttemptIsTerminal returns true when the highest-numbered attempt has reached a
-// terminal phase. Used by WatchActionDetails to close the stream only after action_events
-// reflects the terminal transition, not just the actions table.
-func lastAttemptIsTerminal(attempts []*workflow.ActionAttempt) bool {
+// actionStreamComplete reports whether WatchActionDetails has nothing further to
+// deliver. Attempt-level terminality alone is not enough to close on: a timed-out
+// attempt emits a terminal TIMED_OUT event for attempt N while the action restarts
+// as attempt N+1 in the same reconcile, so closing on the event alone ends the
+// stream mid-retry. Nor is the actions table alone: it can be terminal before
+// action_events reflects it. The stream is complete only when both agree — the
+// action's own phase is terminal, and the highest-numbered attempt is terminal
+// and is the action's current attempt.
+func actionStreamComplete(details *workflow.ActionDetails) bool {
+	attempts := details.GetAttempts()
 	if len(attempts) == 0 {
 		return false
 	}
-	var last *workflow.ActionAttempt
+	last := attempts[0]
 	for _, a := range attempts {
-		if last == nil || a.GetAttempt() > last.GetAttempt() {
+		if a.GetAttempt() > last.GetAttempt() {
 			last = a
 		}
 	}
-	return IsTerminalPhase(last.GetPhase())
+	status := details.GetStatus()
+	return IsTerminalPhase(status.GetPhase()) &&
+		IsTerminalPhase(last.GetPhase()) &&
+		last.GetAttempt() == status.GetAttempts()
 }
 
 // GetActionData is deprecated and no longer implemented. Clients should use
@@ -1241,8 +1250,9 @@ func (s *RunService) WatchActionDetails(
 		return err
 	}
 
-	// Close only once action_events reflects the terminal phase, not just actions table.
-	if lastAttemptIsTerminal(details.GetAttempts()) {
+	// Close only once the action is terminal AND action_events reflects it — a
+	// terminal event for a retrying attempt must not end the stream.
+	if actionStreamComplete(details) {
 		return nil
 	}
 
@@ -1270,8 +1280,8 @@ func (s *RunService) WatchActionDetails(
 			}); err != nil {
 				return err
 			}
-			// Close once action_events reflects the terminal phase.
-			if lastAttemptIsTerminal(details.GetAttempts()) {
+			// Close once the action is terminal and action_events reflects it.
+			if actionStreamComplete(details) {
 				return nil
 			}
 		}

runs/test/api/watch_action_details_test.go

@@ -202,3 +202,97 @@ func TestWatchActionDetails_GetActionNotFound(t *testing.T) {
 	assert.False(t, stream.Receive())
 	assert.Error(t, stream.Err())
 }
+
+// TestWatchActionDetails_StaysOpenAcrossTimeoutRetry covers the #7910 review
+// finding: a timed-out attempt emits a terminal TIMED_OUT event for attempt N
+// before the action restarts as attempt N+1, and the stream must survive that
+// window. Writes below follow the executor's ordering exactly — terminal event
+// first, actions row second — and the stream must only close on the terminal
+// attempt that is also the action's current attempt.
+func TestWatchActionDetails_StaysOpenAcrossTimeoutRetry(t *testing.T) {
+	t.Cleanup(func() { cleanupTestDB(t) })
+
+	ctx := context.Background()
+	httpClient := newClient()
+	runClient := workflowconnect.NewRunServiceClient(httpClient, endpoint)
+	internalClient := workflowconnect.NewInternalRunServiceClient(httpClient, endpoint)
+
+	runID := &common.RunIdentifier{
+		Org:     testOrg,
+		Project: testProject,
+		Domain:  testDomain,
+		Name:    "r" + uniqueString(),
+	}
+	actionName := "action-1"
+	actionID := &common.ActionIdentifier{Run: runID, Name: actionName}
+
+	createTestAction(t, ctx, internalClient, runID, actionName, nil)
+
+	updateStatus := func(phase common.ActionPhase, attempts uint32) {
+		t.Helper()
+		_, err := internalClient.UpdateActionStatus(ctx, connect.NewRequest(&workflow.UpdateActionStatusRequest{
+			ActionId: actionID,
+			Status: &workflow.ActionStatus{
+				Phase:    phase,
+				Attempts: attempts,
+			},
+		}))
+		require.NoError(t, err)
+	}
+
+	event := func(attempt uint32, phase common.ActionPhase) {
+		t.Helper()
+		recordActionEvent(t, ctx, internalClient, &workflow.ActionEvent{
+			Id:          actionID,
+			Attempt:     attempt,
+			Phase:       phase,
+			Version:     0,
+			UpdatedTime: timestamppb.New(time.Now()),
+		})
+	}
+
+	// Attempt 1 is running.
+	updateStatus(common.ActionPhase_ACTION_PHASE_RUNNING, 1)
+	event(1, common.ActionPhase_ACTION_PHASE_RUNNING)
+
+	watchCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
+	stream, err := runClient.WatchActionDetails(watchCtx, connect.NewRequest(&workflow.WatchActionDetailsRequest{
+		ActionId: actionID,
+	}))
+	require.NoError(t, err)
+	defer stream.Close()
+
+	require.True(t, stream.Receive(), "initial state: %v", stream.Err())
+
+	// Attempt 1 times out and the action retries — event first, row second, the
+	// executor's ordering. The terminal TIMED_OUT event for attempt 1 lands while
+	// the row still says RUNNING/1; the old attempt-only predicate closed here.
+	event(1, common.ActionPhase_ACTION_PHASE_TIMED_OUT)
+	updateStatus(common.ActionPhase_ACTION_PHASE_QUEUED, 2)
+	event(2, common.ActionPhase_ACTION_PHASE_QUEUED)
+
+	// Attempt 2 exhausts retries: terminal for the attempt AND the action.
+	event(2, common.ActionPhase_ACTION_PHASE_TIMED_OUT)
+	updateStatus(common.ActionPhase_ACTION_PHASE_TIMED_OUT, 2)
+
+	// Drain until the server closes the stream. Notifications may coalesce, so
+	// assert on what must be true of the whole: the stream survived past the
+	// mid-retry terminal event (we observe attempt 2 at all), and the final
+	// message is the terminal attempt 2.
+	var last *workflow.ActionDetails
+	sawAttempt2 := false
+	for stream.Receive() {
+		last = stream.Msg().Details
+		for _, a := range last.GetAttempts() {
+			if a.GetAttempt() == 2 {
+				sawAttempt2 = true
+			}
+		}
+	}
+	require.NoError(t, stream.Err(), "stream must close cleanly, not on ctx timeout")
+	require.NotNil(t, last, "expected at least one update after the mid-retry timeout")
+	assert.True(t, sawAttempt2, "stream closed before the retry attempt was ever delivered")
+	assert.Equal(t, common.ActionPhase_ACTION_PHASE_TIMED_OUT, last.GetStatus().GetPhase())
+	assert.Equal(t, uint32(2), last.GetStatus().GetAttempts())
+}