From a7f949482829e983cd00383ce92af21f98ab445f Mon Sep 17 00:00:00 2001
From: leookun <leokun.cn@gmail.com>
Date: Fri, 7 Aug 2026 02:19:29 +0800
Subject: [PATCH] fix: checkpoint

---
 frontend/src/components/ui/Combobox.vue       |  4 +-
 .../backend/forwarder/checkpoint_blobs.go     | 45 ++++++++--
 .../forwarder/checkpoint_blobs_test.go        | 51 ++++++++---
 internal/backend/forwarder/projector.go       | 27 +++++-
 .../backend/forwarder/projector_fork_test.go  | 90 +++++++++++++++++++
 internal/backend/forwarder/types.go           |  1 +
 6 files changed, 196 insertions(+), 22 deletions(-)

diff --git a/frontend/src/components/ui/Combobox.vue b/frontend/src/components/ui/Combobox.vue
index f8a71c3..291ffe3 100644
--- a/frontend/src/components/ui/Combobox.vue
+++ b/frontend/src/components/ui/Combobox.vue
@@ -295,10 +295,10 @@ onBeforeUnmount(() => {
         ref="menuRef"
         :id="listboxID"
         role="listbox"
-        class="fixed z-[999] overflow-hidden rounded-[8px] border border-[#3f3f3f] bg-[#232323] p-1 shadow-[0_16px_30px_-12px_rgba(0,0,0,0.7)]"
+        class="fixed z-[999] overflow-y-auto rounded-[8px] border border-[#3f3f3f] bg-[#232323] p-1 shadow-[0_16px_30px_-12px_rgba(0,0,0,0.7)]"
         :style="menuStyle"
       >
-        <ul v-show="filteredOptions.length" role="presentation" class="overflow-y-auto py-1">
+        <ul v-show="filteredOptions.length" role="presentation" class="py-1">
           <li
             v-for="option in normalizedOptions"
             v-show="filteredOptions.includes(option)"
diff --git a/internal/backend/forwarder/checkpoint_blobs.go b/internal/backend/forwarder/checkpoint_blobs.go
index c07c07f..edb65ed 100644
--- a/internal/backend/forwarder/checkpoint_blobs.go
+++ b/internal/backend/forwarder/checkpoint_blobs.go
@@ -94,6 +94,13 @@ func (service *Service) queueCheckpointProjection(stream *ActiveStream, projecti
 	if service.checkpointProjectionReady(stream) {
 		return service.publishReadyCheckpoint(stream)
 	}
+	// Keep the latest live UI state ahead of an immediate client abort. Blob writes are
+	// ordered before this snapshot; acknowledgements still gate terminal completion.
+	if completion == nil {
+		if err := service.publishPendingCheckpoint(stream); err != nil {
+			return service.finishAfterCheckpointSyncFailure(stream, fmt.Errorf("publish pending checkpoint: %w", err))
+		}
+	}
 	service.scheduleStreamTimer(
 		stream,
 		providerTimerKey(streamTimerCheckpointBlobs, ""),
@@ -106,6 +113,31 @@ func (service *Service) queueCheckpointProjection(stream *ActiveStream, projecti
 	return nil
 }
 
+func (service *Service) publishPendingCheckpoint(stream *ActiveStream) error {
+	if service == nil || stream == nil {
+		return nil
+	}
+	stream.mu.Lock()
+	pending := stream.PendingCheckpoint
+	if pending == nil || pending.Published {
+		stream.mu.Unlock()
+		return nil
+	}
+	pending.Published = true
+	state := pending.State
+	stream.UpdatedAt = time.Now().UTC()
+	stream.mu.Unlock()
+	if err := service.broker.Publish(stream.RequestID, StreamEvent{Message: buildCheckpointMessage(state)}); err != nil {
+		stream.mu.Lock()
+		if stream.PendingCheckpoint == pending {
+			pending.Published = false
+		}
+		stream.mu.Unlock()
+		return err
+	}
+	return nil
+}
+
 func (service *Service) checkpointProjectionReady(stream *ActiveStream) bool {
 	if stream == nil {
 		return false
@@ -176,15 +208,18 @@ func (service *Service) publishReadyCheckpoint(stream *ActiveStream) error {
 	stream.PendingCheckpoint = nil
 	state := pending.State
 	completion := clonePendingTurnCompletion(pending.Completion)
+	published := pending.Published
 	stream.UpdatedAt = time.Now().UTC()
 	stream.mu.Unlock()
 	clearStreamTimer(stream, providerTimerKey(streamTimerCheckpointBlobs, ""))
-	if err := service.broker.Publish(stream.RequestID, StreamEvent{Message: buildCheckpointMessage(state)}); err != nil {
-		if completion != nil {
-			log.Printf("forwarder checkpoint publish skipped before successful terminal request_id=%s err=%v", stream.RequestID, err)
-			return service.finishSuccessfulTurnAfterCheckpoint(stream, *completion)
+	if !published {
+		if err := service.broker.Publish(stream.RequestID, StreamEvent{Message: buildCheckpointMessage(state)}); err != nil {
+			if completion != nil {
+				log.Printf("forwarder checkpoint publish skipped before successful terminal request_id=%s err=%v", stream.RequestID, err)
+				return service.finishSuccessfulTurnAfterCheckpoint(stream, *completion)
+			}
+			return err
 		}
-		return err
 	}
 	if completion != nil {
 		return service.finishSuccessfulTurnAfterCheckpoint(stream, *completion)
diff --git a/internal/backend/forwarder/checkpoint_blobs_test.go b/internal/backend/forwarder/checkpoint_blobs_test.go
index 9ada5d3..7cf0f64 100644
--- a/internal/backend/forwarder/checkpoint_blobs_test.go
+++ b/internal/backend/forwarder/checkpoint_blobs_test.go
@@ -8,26 +8,34 @@ import (
 	"cursor/gen/agentv1"
 )
 
-func TestCheckpointBlobSyncPublishesCheckpointAfterAcknowledgements(t *testing.T) {
+func TestCheckpointBlobSyncPublishesNonTerminalCheckpointBeforeAcknowledgements(t *testing.T) {
 	service, stream, projection := testCheckpointBlobProjection(t)
 	if err := service.queueCheckpointProjection(stream, projection, nil); err != nil {
 		t.Fatalf("queueCheckpointProjection() error = %v", err)
 	}
 	events := readCheckpointTestEvents(t, service, stream)
-	if len(events) != len(projection.Blobs) {
-		t.Fatalf("events before ACK = %d, want %d Blob writes", len(events), len(projection.Blobs))
+	if len(events) != len(projection.Blobs)+1 {
+		t.Fatalf("events before ACK = %d, want %d Blob writes and one checkpoint", len(events), len(projection.Blobs))
 	}
-	for _, event := range events {
+	for _, event := range events[:len(projection.Blobs)] {
 		if event.Message.GetKvServerMessage().GetSetBlobArgs() == nil {
 			t.Fatalf("event before ACK = %#v, want set_blob_args", event.Message)
 		}
 	}
+	if checkpoint := events[len(events)-1].Message.GetConversationCheckpointUpdate(); checkpoint == nil || len(checkpoint.GetTurns()) != 1 {
+		t.Fatalf("last event before ACK = %#v, want one Blob-backed turn", events[len(events)-1].Message)
+	}
 
 	acknowledgeCheckpointBlobs(t, service, stream)
 	events = readCheckpointTestEvents(t, service, stream)
-	checkpoint := events[len(events)-1].Message.GetConversationCheckpointUpdate()
-	if checkpoint == nil || len(checkpoint.GetTurns()) != 1 {
-		t.Fatalf("last event checkpoint = %#v, want one Blob-backed turn", checkpoint)
+	checkpointCount := 0
+	for _, event := range events {
+		if event.Message.GetConversationCheckpointUpdate() != nil {
+			checkpointCount++
+		}
+	}
+	if checkpointCount != 1 {
+		t.Fatalf("checkpoints after ACK = %d, want 1", checkpointCount)
 	}
 }
 
@@ -40,6 +48,12 @@ func TestCheckpointBlobSyncPublishesCheckpointBeforeSuccessfulTerminal(t *testin
 	if err := service.queueCheckpointProjection(stream, projection, completion); err != nil {
 		t.Fatalf("queueCheckpointProjection() error = %v", err)
 	}
+	eventsBeforeACK := readCheckpointTestEvents(t, service, stream)
+	for _, event := range eventsBeforeACK {
+		if event.Message.GetConversationCheckpointUpdate() != nil || event.Message.GetInteractionUpdate().GetTurnEnded() != nil || event.End {
+			t.Fatalf("event before ACK = %#v, want only Blob writes", event)
+		}
+	}
 	acknowledgeCheckpointBlobs(t, service, stream)
 
 	events := readCheckpointTestEvents(t, service, stream)
@@ -84,11 +98,21 @@ func TestCheckpointBlobTimeoutDoesNotFailSuccessfulTurn(t *testing.T) {
 	}
 }
 
-func TestCancellationDiscardsPendingCheckpointAndIgnoresLateAcknowledgements(t *testing.T) {
+func TestCancellationKeepsPublishedCheckpointAndIgnoresLateAcknowledgements(t *testing.T) {
 	service, stream, projection := testCheckpointBlobProjection(t)
 	if err := service.queueCheckpointProjection(stream, projection, nil); err != nil {
 		t.Fatalf("queueCheckpointProjection() error = %v", err)
 	}
+	eventsBeforeCancel := readCheckpointTestEvents(t, service, stream)
+	checkpointBeforeCancel := 0
+	for _, event := range eventsBeforeCancel {
+		if event.Message.GetConversationCheckpointUpdate() != nil {
+			checkpointBeforeCancel++
+		}
+	}
+	if checkpointBeforeCancel != 1 {
+		t.Fatalf("checkpoints before cancel = %d, want 1", checkpointBeforeCancel)
+	}
 	stream.mu.Lock()
 	requestIDs := make([]uint32, 0, len(stream.PendingCheckpointBlobWrites))
 	for requestID := range stream.PendingCheckpointBlobWrites {
@@ -114,16 +138,19 @@ func TestCancellationDiscardsPendingCheckpointAndIgnoresLateAcknowledgements(t *
 	}
 
 	events := readCheckpointTestEvents(t, service, stream)
-	var checkpoint, canceledEnd bool
+	checkpointCount := 0
+	var canceledEnd bool
 	for _, event := range events {
-		checkpoint = checkpoint || event.Message.GetConversationCheckpointUpdate() != nil
+		if event.Message.GetConversationCheckpointUpdate() != nil {
+			checkpointCount++
+		}
 		canceledEnd = canceledEnd || event.End && event.TerminalErrorCode == "canceled"
 	}
 	stream.mu.Lock()
 	pending := stream.PendingCheckpoint
 	stream.mu.Unlock()
-	if checkpoint || !canceledEnd || pending != nil {
-		t.Fatalf("cancel events checkpoint=%v canceled_end=%v pending=%v", checkpoint, canceledEnd, pending != nil)
+	if checkpointCount != 1 || !canceledEnd || pending != nil {
+		t.Fatalf("cancel events checkpoints=%d canceled_end=%v pending=%v", checkpointCount, canceledEnd, pending != nil)
 	}
 }
 
diff --git a/internal/backend/forwarder/projector.go b/internal/backend/forwarder/projector.go
index f8f58d7..0b5980f 100644
--- a/internal/backend/forwarder/projector.go
+++ b/internal/backend/forwarder/projector.go
@@ -621,10 +621,22 @@ func projectCheckpointTurnBlobs(conversation *ConversationFile, blobs *checkpoin
 		}
 		grouped[entry.TurnSeq] = append(grouped[entry.TurnSeq], entry)
 	}
-
-	turnIDs := make([][]byte, 0, len(order))
+	logicalTurns := make([][]HistoryEntry, 0, len(order))
 	for _, turnSeq := range order {
 		entries := grouped[turnSeq]
+		if checkpointTurnHasUserMessage(entries) {
+			logicalTurns = append(logicalTurns, append([]HistoryEntry(nil), entries...))
+			continue
+		}
+		if len(logicalTurns) == 0 {
+			continue
+		}
+		last := len(logicalTurns) - 1
+		logicalTurns[last] = append(logicalTurns[last], entries...)
+	}
+
+	turnIDs := make([][]byte, 0, len(logicalTurns))
+	for _, entries := range logicalTurns {
 		completedToolCalls, err := collectCheckpointCompletedToolCalls(entries)
 		if err != nil {
 			return nil, err
@@ -734,7 +746,7 @@ func projectCheckpointTurnBlobs(conversation *ConversationFile, blobs *checkpoin
 				})
 			}
 		}
-		if len(userMessageID) == 0 && len(steps) == 0 {
+		if len(userMessageID) == 0 {
 			continue
 		}
 		stepIDs := make([][]byte, 0, len(steps))
@@ -765,6 +777,15 @@ func projectCheckpointTurnBlobs(conversation *ConversationFile, blobs *checkpoin
 	return turnIDs, nil
 }
 
+func checkpointTurnHasUserMessage(entries []HistoryEntry) bool {
+	for _, entry := range entries {
+		if strings.TrimSpace(entry.Kind) == "user_message" {
+			return true
+		}
+	}
+	return false
+}
+
 func collectCheckpointCompletedToolCalls(entries []HistoryEntry) (map[string]json.RawMessage, error) {
 	completed := make(map[string]json.RawMessage)
 	for _, entry := range entries {
diff --git a/internal/backend/forwarder/projector_fork_test.go b/internal/backend/forwarder/projector_fork_test.go
index 2a8a7be..1a88a61 100644
--- a/internal/backend/forwarder/projector_fork_test.go
+++ b/internal/backend/forwarder/projector_fork_test.go
@@ -87,6 +87,94 @@ func TestProjectCheckpointProjectionBuildsResolvableForkState(t *testing.T) {
 	}
 }
 
+func TestProjectCheckpointProjectionMergesResumeActivityIntoPreviousUserTurn(t *testing.T) {
+	userPayload, err := protojson.Marshal(&agentv1.UserMessage{
+		Text:      "original question",
+		MessageId: "message-1",
+	})
+	if err != nil {
+		t.Fatalf("marshal user message: %v", err)
+	}
+	firstAnswer := newAssistantTextEntry(1, "request-1", "before resume", "", "")
+	firstAnswer.Seq = 2
+	resumedAnswer := newAssistantTextEntry(2, "request-resume", "after resume", "", "")
+	resumedAnswer.Seq = 3
+	conversation := &ConversationFile{
+		ConversationID:     "conversation-1",
+		RootConversationID: "conversation-1",
+		Mode:               "agent",
+		NextTurnSeq:        3,
+		NextEntrySeq:       4,
+		Entries: []HistoryEntry{
+			{Seq: 1, TurnSeq: 1, RequestID: "request-1", Role: "user", Kind: "user_message", Payload: userPayload},
+			firstAnswer,
+			resumedAnswer,
+		},
+	}
+
+	projection, err := NewHistoryProjector().ProjectCheckpointProjection(conversation)
+	if err != nil {
+		t.Fatalf("ProjectCheckpointProjection() error = %v", err)
+	}
+	if len(projection.State.GetTurns()) != 1 {
+		t.Fatalf("checkpoint turns = %d, want one logical user turn", len(projection.State.GetTurns()))
+	}
+
+	blobs := make(map[string][]byte, len(projection.Blobs))
+	for _, blob := range projection.Blobs {
+		blobs[string(blob.ID)] = blob.Data
+	}
+	turn := &agentv1.ConversationTurnStructure{}
+	if err := proto.Unmarshal(blobs[string(projection.State.GetTurns()[0])], turn); err != nil {
+		t.Fatalf("decode checkpoint turn: %v", err)
+	}
+	agentTurn := turn.GetAgentConversationTurn()
+	if agentTurn == nil {
+		t.Fatal("checkpoint turn does not contain an agent turn")
+	}
+	if len(agentTurn.GetUserMessage()) == 0 {
+		t.Fatal("checkpoint turn lost the original user message Blob id")
+	}
+	if _, ok := blobs[string(agentTurn.GetUserMessage())]; !ok {
+		t.Fatal("checkpoint turn references a missing user message Blob")
+	}
+	if agentTurn.GetRequestId() != "request-1" {
+		t.Fatalf("checkpoint request id = %q, want original request", agentTurn.GetRequestId())
+	}
+
+	steps := checkpointProjectionSteps(t, projection)
+	if len(steps) != 2 || steps[0].GetAssistantMessage().GetText() != "before resume" || steps[1].GetAssistantMessage().GetText() != "after resume" {
+		t.Fatalf("checkpoint steps did not preserve resumed activity: %#v", steps)
+	}
+
+	replay, err := promptengine.DecodeReplayMessages(projection.State.GetRootPromptMessagesJson())
+	if err != nil {
+		t.Fatalf("decode root prompt replay: %v", err)
+	}
+	if len(replay) != 3 || replay[0].Role != "user" || replay[1].Content != "before resume" || replay[2].Content != "after resume" {
+		t.Fatalf("checkpoint turn merge changed model replay: %#v", replay)
+	}
+}
+
+func TestProjectCheckpointProjectionOmitsActivityWithoutAnyUserMessage(t *testing.T) {
+	conversation := &ConversationFile{
+		ConversationID: "conversation-1",
+		Mode:           "agent",
+		NextTurnSeq:    2,
+		Entries: []HistoryEntry{
+			newAssistantTextEntry(1, "request-resume", "orphaned resume output", "", ""),
+		},
+	}
+
+	projection, err := NewHistoryProjector().ProjectCheckpointProjection(conversation)
+	if err != nil {
+		t.Fatalf("ProjectCheckpointProjection() error = %v", err)
+	}
+	if len(projection.State.GetTurns()) != 0 || len(projection.Blobs) != 0 {
+		t.Fatalf("checkpoint emitted a turn without a user message: %#v", projection)
+	}
+}
+
 func TestProjectCheckpointProjectionKeepsForkPointIsolatedFromLaterHistory(t *testing.T) {
 	firstUser, err := protojson.Marshal(&agentv1.UserMessage{Text: "first question", MessageId: "message-1"})
 	if err != nil {
@@ -288,6 +376,7 @@ func TestProjectCheckpointProjectionKeepsStartedToolCallWhenResultPayloadIsMissi
 		Mode:           "agent",
 		NextTurnSeq:    2,
 		Entries: []HistoryEntry{
+			testCheckpointUserEntry(t),
 			newToolCallEntry(1, "request-1", "call-1", "Read", "", "", startedToolCall),
 			newToolResultEntry(1, "request-1", "call-1", "Read", `{"path":"/tmp/example.txt"}`, "read failed", "", nil),
 		},
@@ -316,6 +405,7 @@ func TestProjectCheckpointProjectionAppendsLegacyResultWithoutToolCallEntry(t *t
 		Mode:           "agent",
 		NextTurnSeq:    2,
 		Entries: []HistoryEntry{
+			testCheckpointUserEntry(t),
 			newToolResultEntry(1, "request-1", "call-1", "Read", `{"path":"/tmp/example.txt"}`, "not readable", "", completedToolCall),
 		},
 	}
diff --git a/internal/backend/forwarder/types.go b/internal/backend/forwarder/types.go
index 01b69f2..7eb7788 100644
--- a/internal/backend/forwarder/types.go
+++ b/internal/backend/forwarder/types.go
@@ -228,6 +228,7 @@ type pendingCheckpointPublish struct {
 	State      *agentv1.ConversationStateStructure
 	Required   map[string]struct{}
 	Completion *pendingTurnCompletion
+	Published  bool
 }
 
 type PendingCompaction struct {