open-source/cursor-byok
1
0
Fork 0
mirror of https://wget.la/https://github.com/leookun/cursor-byok synced 2026-08-17 19:47:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #239 from DedSecer/fix/cursor-fork-context

Browse Source 
fix(forwarder): preserve Cursor fork context
This commit is contained in:
leokun
2026-08-03 00:21:14 +08:00
committed by GitHub
parent 5742073be5 14b286a466
commit 697fa99245
14 changed files with 1979 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/backend/forwarder/service.go
+77 -31
View File
@@ -9,6 +9,7 @@ import (
"log"
"sort"
"strings"
"sync"
"time"
"connectrpc.com/connect"
@@ -261,6 +262,8 @@ type Service struct {
execBridge execbridge.ExecBridge
interactionBridge interactionbridge.InteractionBridge
appendSeq *appendSequenceTracker
checkpointBlobMu sync.Mutex
checkpointBlobs map[string]*checkpointBlobCacheEntry
}
type agentModelMemory interface {
@@ -300,6 +303,7 @@ func NewService(historyRoot string, resolver modeladapter.ChannelResolver) *Serv
execBridge: execbridge.NewBridge(),
interactionBridge: interactionbridge.NewBridge(),
appendSeq: newAppendSequenceTracker(),
checkpointBlobs: make(map[string]*checkpointBlobCacheEntry),
}
service.startHistoryMaintenance()
store.SyncAllCursorTranscriptsBestEffort()
@@ -328,6 +332,7 @@ func newServiceWithDependencies(store *ConversationFileStore, projector *History
execBridge: execbridge.NewBridge(),
interactionBridge: interactionbridge.NewBridge(),
appendSeq: newAppendSequenceTracker(),
checkpointBlobs: make(map[string]*checkpointBlobCacheEntry),
}
}
@@ -557,6 +562,7 @@ func (service *Service) decodeInboundIntent(requestID string, message *agentv1.A
}
intent.ConversationID = conversationID
intent.ConversationState = runRequest.GetConversationState()
intent.PreFetchedBlobs = runRequest.GetPreFetchedBlobs()
intent.UserMessage = extractUserMessage(message)
intent.RequestContext = extractRequestContext(message)
if service.shouldIgnoreEmptyResumeRunRequest(requestID, runRequest, intent.UserMessage, intent.RequestContext) {
@@ -604,6 +610,7 @@ func (service *Service) decodeInboundIntent(requestID string, message *agentv1.A
intent.ConversationID = conversationID
intent.SubagentTypeName = strings.TrimSpace(prewarmRequest.GetSubagentTypeName())
intent.ConversationState = prewarmRequest.GetConversationState()
intent.PreFetchedBlobs = prewarmRequest.GetPreFetchedBlobs()
intent.Mode, intent.ModeSource, intent.HasExplicitMode, err = extractPrewarmMode(prewarmRequest)
if err != nil {
return InboundIntent{}, err
@@ -819,11 +826,14 @@ func (service *Service) snapshotVisibleTurns(conversation *ConversationFile) ([]
if service == nil || service.projector == nil || conversation == nil {
return nil, nil
}
state, err := service.projector.ProjectLegacyCheckpoint(conversation)
projection, err := service.projector.ProjectCheckpointProjection(conversation)
if err != nil {
return nil, err
}
return cloneByteSlices(state.GetTurns()), nil
if projection == nil || projection.State == nil {
return nil, fmt.Errorf("checkpoint projection is empty")
}
return cloneByteSlices(projection.State.GetTurns()), nil
}
// handleCancelIntent 处理取消请求，并向客户端发送执行桥 abort。
@@ -833,42 +843,60 @@ func (service *Service) handleCancelIntent(intent InboundIntent) error {
return fmt.Errorf("request is not active: %s", intent.RequestID)
}
hasCheckpoint := checkpointConversationInitialized(stream)
if hasCheckpoint {
cancelReason := firstNonEmpty(intent.CancelReason, "user aborted")
_, err := service.appendConversationEntries(stream, stream.ConversationID, []HistoryEntry{
newMetadataEntry(stream.TurnSeq, intent.RequestID, "control", map[string]any{
"status": "canceled",
"reason": cancelReason,
"replay_policy": cancelReplayPolicyForReason(cancelReason),
}),
})
if err != nil {
return err
}
}
stream.mu.Lock()
pendingExecs := make([]runtimecore.PendingExec, 0, len(stream.PendingExecs))
for _, pending := range stream.PendingExecs {
pendingExecs = append(pendingExecs, pending)
}
if stream.ProviderCancel != nil {
stream.ProviderCancel()
stream.ProviderCancel = nil
}
stream.ProviderActive = false
stream.CurrentProviderToken++
stream.CurrentCompactionToken++
stream.PendingProviderAction = providerActionNone
stream.PendingCompaction = nil
stream.UpdatedAt = time.Now().UTC()
stream.mu.Unlock()
if hasCheckpoint {
cancelReason := firstNonEmpty(intent.CancelReason, "user aborted")
cancelEntry := newMetadataEntry(stream.TurnSeq, intent.RequestID, "control", map[string]any{
"status": "canceled",
"reason": cancelReason,
"replay_policy": cancelReplayPolicyForReason(cancelReason),
})
if _, err := service.appendConversationEntries(stream, stream.ConversationID, []HistoryEntry{cancelEntry}); err != nil {
log.Printf("forwarder cancellation metadata persistence failed request_id=%s conversation_id=%s err=%v", stream.RequestID, stream.ConversationID, err)
if memoryErr := service.appendCheckpointEntries(stream, []HistoryEntry{cancelEntry}); memoryErr != nil {
return memoryErr
}
}
}
for _, pending := range pendingExecs {
_ = service.broker.Publish(intent.RequestID, StreamEvent{
Message: buildExecAbortMessage(pending),
})
}
if hasCheckpoint {
if err := service.publishCheckpoint(stream.RequestID, stream.ConversationID); err != nil {
return err
}
}
clearPendingProviderCompletion(stream)
terminalMessage := firstNonEmpty(intent.CancelReason, "[canceled] User aborted request")
stream.mu.Lock()
stream.PendingProviderAction = providerActionNone
stream.PendingExecs = make(map[string]runtimecore.PendingExec)
stream.PendingInteractions = make(map[string]runtimecore.PendingInteraction)
stream.UpdatedAt = time.Now().UTC()
stream.mu.Unlock()
service.setTurnPhase(stream, TurnPhaseCanceled)
return service.broker.Cancel(intent.RequestID, firstNonEmpty(intent.CancelReason, "[canceled] User aborted request"))
service.discardPendingCheckpoint(stream, fmt.Errorf("checkpoint superseded by cancellation"))
if hasCheckpoint {
if err := service.publishCheckpointWithTerminalAction(
stream.RequestID,
stream.ConversationID,
checkpointCancellationAction(terminalMessage),
); err != nil {
return service.failTerminalCheckpointSync(stream, err)
}
return nil
}
return service.finishCanceledTurnAfterCheckpoint(stream, terminalMessage)
}
// handleExecResult 处理客户端返回的执行桥结果，并在终态时把 tool_result 写回 history。
@@ -2122,9 +2150,18 @@ func (service *Service) completeSuccessfulTurn(stream *ActiveStream, completion
err,
)
}
if err := service.publishCheckpoint(requestID, conversationID); err != nil {
if err := service.publishCheckpointWithCompletion(requestID, conversationID, &completion); err != nil {
return err
}
return nil
}
func (service *Service) finishSuccessfulTurnAfterCheckpoint(stream *ActiveStream, completion pendingTurnCompletion) error {
if stream == nil {
return nil
}
requestID := firstNonEmpty(strings.TrimSpace(completion.RequestID), strings.TrimSpace(stream.RequestID))
usage := completion.Usage
if err := service.broker.Publish(requestID, StreamEvent{
Message: buildTurnEndedMessage(usage.InputTokens, usage.OutputTokens, usage.CacheReadTokens, usage.CacheWriteTokens),
}); err != nil {
@@ -2151,7 +2188,15 @@ func (service *Service) failStreamIfNonTerminal(stream *ActiveStream, terminalCo
}
// publishCheckpoint 按当前内存会话镜像投影出 checkpoint，并广播给所有 RunSSE 订阅者。
func (service *Service) publishCheckpoint(requestID string, _ string) error {
func (service *Service) publishCheckpoint(requestID string, conversationID string) error {
return service.publishCheckpointWithCompletion(requestID, conversationID, nil)
}
func (service *Service) publishCheckpointWithCompletion(requestID string, conversationID string, completion *pendingTurnCompletion) error {
return service.publishCheckpointWithTerminalAction(requestID, conversationID, checkpointCompletionAction(completion))
}
func (service *Service) publishCheckpointWithTerminalAction(requestID string, conversationID string, terminalAction checkpointTerminalAction) error {
stream, ok := service.broker.Get(requestID)
if !ok || stream == nil {
return fmt.Errorf("request is not active: %s", requestID)
@@ -2160,15 +2205,16 @@ func (service *Service) publishCheckpoint(requestID string, _ string) error {
if err != nil {
return err
}
state, err := service.projector.ProjectLegacyCheckpoint(conversation)
projection, err := service.projector.ProjectCheckpointProjection(conversation)
if err != nil {
return err
}
state.PendingToolCalls = buildPendingToolCalls(pendingExecs, pendingInteractions)
service.rewriteCheckpointTokenDetailsForClient(stream, conversation, state)
return service.broker.Publish(requestID, StreamEvent{
Message: buildCheckpointMessage(state),
})
if projection == nil || projection.State == nil {
return fmt.Errorf("checkpoint projection is empty")
}
projection.State.PendingToolCalls = buildPendingToolCalls(pendingExecs, pendingInteractions)
service.rewriteCheckpointTokenDetailsForClient(stream, conversation, projection.State)
return service.queueCheckpointProjection(stream, projection, terminalAction)
}
func (service *Service) rewriteCheckpointTokenDetailsForClient(stream *ActiveStream, conversation *ConversationFile, state *agentv1.ConversationStateStructure) {