feat: Surfacing heartbeat counts for activities

chasm/lib/activity/activity.go

@@ -698,9 +698,11 @@ func (a *Activity) RecordHeartbeat(
 	if err != nil {
 		return nil, err
 	}
+	prevHeartbeat, _ := a.LastHeartbeat.TryGet(ctx)
 	a.LastHeartbeat = chasm.NewDataField(ctx, &activitypb.ActivityHeartbeatState{
-		RecordedTime: timestamppb.New(ctx.Now(a)),
-		Details:      input.Request.GetHeartbeatRequest().GetDetails(),
+		RecordedTime:        timestamppb.New(ctx.Now(a)),
+		Details:             input.Request.GetHeartbeatRequest().GetDetails(),
+		TotalHeartbeatCount: prevHeartbeat.GetTotalHeartbeatCount() + 1,
 	})
 	if heartbeatTimeout := a.GetHeartbeatTimeout().AsDuration(); heartbeatTimeout > 0 {
 		ctx.AddTask(
@@ -802,6 +804,7 @@ func (a *Activity) buildActivityExecutionInfo(ctx chasm.Context) *apiactivitypb.
 		Header:                  requestData.GetHeader(),
 		HeartbeatDetails:        heartbeat.GetDetails(),
 		HeartbeatTimeout:        a.GetHeartbeatTimeout(),
+		TotalHeartbeatCount:     heartbeat.GetTotalHeartbeatCount(),
 		LastAttemptCompleteTime: attempt.GetCompleteTime(),
 		LastFailure:             attempt.GetLastFailureDetails().GetFailure(),
 		LastHeartbeatTime:       heartbeat.GetRecordedTime(),

chasm/lib/activity/activity_test.go

@@ -9,8 +9,10 @@ import (
 	commonpb "go.temporal.io/api/common/v1"
 	taskqueuepb "go.temporal.io/api/taskqueue/v1"
 	"go.temporal.io/server/api/historyservice/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	tokenspb "go.temporal.io/server/api/token/v1"
 	"go.temporal.io/server/chasm"
-	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
+	activitypb "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/namespace"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
@@ -250,6 +252,57 @@ func TestActivityTerminate(t *testing.T) {
 	}
 }
 
+func TestRecordHeartbeat_IncrementsHeartbeatCount(t *testing.T) {
+	testTime := time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
+	testNamespaceID := "test-namespace-id"
+
+	ctx := &chasm.MockMutableContext{
+		MockContext: chasm.MockContext{
+			HandleNow: func(chasm.Component) time.Time { return testTime },
+		},
+	}
+
+	ref := &persistencespb.ChasmComponentRef{
+		NamespaceId: testNamespaceID,
+		BusinessId:  "test-business-id",
+		RunId:       "test-run-id",
+	}
+	refBytes, err := ref.Marshal()
+	require.NoError(t, err)
+
+	token := &tokenspb.Task{
+		Attempt:      1,
+		ComponentRef: refBytes,
+	}
+
+	attemptState := &activitypb.ActivityAttemptState{Count: 1}
+	a := &Activity{
+		ActivityState: &activitypb.ActivityState{
+			Status: activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
+		},
+		LastAttempt: chasm.NewDataField(ctx, attemptState),
+	}
+
+	input := WithToken[*historyservice.RecordActivityTaskHeartbeatRequest]{
+		Token: token,
+		Request: &historyservice.RecordActivityTaskHeartbeatRequest{
+			NamespaceId: testNamespaceID,
+		},
+	}
+
+	_, err = a.RecordHeartbeat(ctx, input)
+	require.NoError(t, err)
+	heartbeat, ok := a.LastHeartbeat.TryGet(ctx)
+	require.True(t, ok)
+	require.Equal(t, int64(1), heartbeat.GetTotalHeartbeatCount())
+
+	_, err = a.RecordHeartbeat(ctx, input)
+	require.NoError(t, err)
+	heartbeat, ok = a.LastHeartbeat.TryGet(ctx)
+	require.True(t, ok)
+	require.Equal(t, int64(2), heartbeat.GetTotalHeartbeatCount())
+}
+
 func TestContextMetadata(t *testing.T) {
 	t.Run("returns activity type and task queue", func(t *testing.T) {
 		ctx := &chasm.MockMutableContext{}

chasm/lib/activity/proto/v1/activity_state.proto

@@ -158,6 +158,8 @@ message ActivityHeartbeatState {
   temporal.api.common.v1.Payloads details = 1;
   // Time the last heartbeat was recorded.
   google.protobuf.Timestamp recorded_time = 2;
+  // Total number of heartbeats recorded across all attempts of this activity, including retries.
+  int64 total_heartbeat_count = 3;
 }
 
 message ActivityRequestData {

go.mod

@@ -63,7 +63,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.43.0
 	go.opentelemetry.io/otel/sdk/metric v1.43.0
 	go.opentelemetry.io/otel/trace v1.43.0
-	go.temporal.io/api v1.62.10-0.20260421204157-0617d4e3bba2
+	go.temporal.io/api v1.62.10-0.20260422203216-0d99afb772ad
 	go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2
 	go.temporal.io/sdk v1.41.1
 	go.uber.org/fx v1.24.0

go.sum

@@ -469,8 +469,8 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R
 go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U=
 go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0=
 go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4=
-go.temporal.io/api v1.62.10-0.20260421204157-0617d4e3bba2 h1:9s2PjMyiiRg49fmjgWDo5RIx2MuWu5K4S8a7pThNpzU=
-go.temporal.io/api v1.62.10-0.20260421204157-0617d4e3bba2/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
+go.temporal.io/api v1.62.10-0.20260422203216-0d99afb772ad h1:QaJLjzQXfc/oRdUyUdA3mihv2yZadCgelcaDwTYB7mM=
+go.temporal.io/api v1.62.10-0.20260422203216-0d99afb772ad/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
 go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2 h1:1hKeH3GyR6YD6LKMHGCZ76t6h1Sgha0hXVQBxWi3dlQ=
 go.temporal.io/auto-scaled-workers v0.0.0-20260407181057-edd947d743d2/go.mod h1:T8dnzVPeO+gaUTj9eDgm/lT2lZH4+JXNvrGaQGyVi50=
 go.temporal.io/sdk v1.41.1 h1:yOpvsHyDD1lNuwlGBv/SUodCPhjv9nDeC9lLHW/fJUA=

proto/internal/temporal/server/api/persistence/v1/executions.proto

@@ -698,6 +698,9 @@ message ActivityInfo {
   // Replication: This field is part of ActivityInfo and is automatically replicated
   // via state-based replication. No special handling is needed.
   temporal.server.api.clock.v1.VectorClock started_clock = 52;
+
+  // Total number of heartbeats recorded across all attempts of this activity, including retries.
+  int64 total_heartbeat_count = 53;
 }
 
 // timer_map column

service/history/workflow/mutable_state_impl.go

@@ -1974,6 +1974,7 @@ func (ms *MutableStateImpl) UpdateActivityProgress(
 	}
 	ai.Version = ms.GetCurrentVersion()
 	ai.LastHeartbeatDetails = request.Details
+	ai.TotalHeartbeatCount++
 	now := ms.timeSource.Now()
 	ai.LastHeartbeatUpdateTime = timestamppb.New(now)
 	ms.updateActivityInfos[ai.ScheduledEventId] = ai

service/history/workflow/mutable_state_impl_test.go

@@ -7389,3 +7389,22 @@ func (s *mutableStateSuite) TestApplyWorkflowExecutionOptionsUpdatedEvent_TimeSk
 		})
 	}
 }
+
+func (s *mutableStateSuite) TestUpdateActivityProgress_IncrementsHeartbeatCount() {
+	dbState := s.buildWorkflowMutableState()
+	scheduleEventID := int64(100)
+	dbState.ActivityInfos[scheduleEventID] = &persistencespb.ActivityInfo{
+		ScheduledEventId: scheduleEventID,
+	}
+	ms, err := NewMutableStateFromDB(s.mockShard, s.mockEventsCache, s.logger, s.namespaceEntry, dbState, 123)
+	s.NoError(err)
+
+	ai := ms.pendingActivityInfoIDs[scheduleEventID]
+	s.Equal(int64(0), ai.TotalHeartbeatCount)
+
+	ms.UpdateActivityProgress(ai, &workflowservice.RecordActivityTaskHeartbeatRequest{})
+	s.Equal(int64(1), ai.TotalHeartbeatCount)
+
+	ms.UpdateActivityProgress(ai, &workflowservice.RecordActivityTaskHeartbeatRequest{})
+	s.Equal(int64(2), ai.TotalHeartbeatCount)
+}