fix(agent): 超大 session 准备上下文时每条历史消息只转换一次

internal/agent/application/context_frag_test.go

@@ -251,7 +251,11 @@ func TestNormalizeContextMessagesRemapsCurrentAndMemory(t *testing.T) {
 
 	want := sanitizeMessages(messages)
 	want = repairToolCallClosures(want, syntheticToolClosureError)
-	got, gotCurrent, gotMemory := normalizeContextMessages(messages, &currentIndex, &memoryIndex)
+	normalized := normalizeContext(messages)
+	got, gotCurrent, gotMemory := normalized.models(), normalized.index(&currentIndex), normalized.index(&memoryIndex)
+	if sent := normalized.sent(); !reflect.DeepEqual(sent, modelMessagesToSDKMessages(got)) {
+		t.Fatalf("provider messages differ from converting the normalized ones: %#v", sent)
+	}
 
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("normalized messages changed: got %#v want %#v", got, want)

internal/agent/application/context_messages.go

@@ -0,0 +1,104 @@
+package application
+
+import (
+	"slices"
+	"strings"
+
+	sdk "github.com/felinics/twilight/sdk"
+)
+
+// contextMessage pairs a request message with its provider form. Converting
+// a stored message decodes and re-encodes its whole JSON, so a request
+// converts each message once: passes over it read the provider form, and a
+// pass that rewrites a message converts only the rewrite.
+type contextMessage struct {
+	model ModelMessage
+	sent  sdk.Message
+}
+
+func newContextMessage(model ModelMessage) contextMessage {
+	return contextMessage{model: model, sent: modelMessageToSDKMessage(model)}
+}
+
+func newContextMessages(models []ModelMessage) []contextMessage {
+	out := make([]contextMessage, 0, len(models))
+	for _, model := range models {
+		out = append(out, newContextMessage(model))
+	}
+	return out
+}
+
+func contextModelMessages(messages []contextMessage) []ModelMessage {
+	out := make([]ModelMessage, 0, len(messages))
+	for _, message := range messages {
+		out = append(out, message.model)
+	}
+	return out
+}
+
+func contextSentMessages(messages []contextMessage) []sdk.Message {
+	out := make([]sdk.Message, 0, len(messages))
+	for _, message := range messages {
+		out = append(out, message.sent)
+	}
+	return out
+}
+
+// normalizedContext is a request's messages after sanitizing and tool-call
+// closure repair. Sanitizing keeps or drops each message on its own, so the
+// first n inputs sanitize to a prefix of the sanitized messages; what
+// normalizing such a prefix alone yields is answered from those conversions.
+type normalizedContext struct {
+	messages  []contextMessage
+	sanitized []contextMessage
+	inputs    []int
+}
+
+func normalizeContext(messages []ModelMessage) normalizedContext {
+	var normalized normalizedContext
+	for i, message := range messages {
+		message, ok := sanitizeMessage(message)
+		if !ok {
+			continue
+		}
+		normalized.sanitized = append(normalized.sanitized, newContextMessage(message))
+		normalized.inputs = append(normalized.inputs, i)
+	}
+	normalized.messages = repairContextToolCalls(normalized.sanitized, syntheticToolClosureError)
+	return normalized
+}
+
+func (c normalizedContext) models() []ModelMessage {
+	return contextModelMessages(c.messages)
+}
+
+func (c normalizedContext) sent() []sdk.Message {
+	return contextSentMessages(c.messages)
+}
+
+// prefixLength is how many messages normalizing the first n inputs alone
+// yields.
+func (c normalizedContext) prefixLength(n int) int {
+	if n <= 0 {
+		return 0
+	}
+	kept, _ := slices.BinarySearch(c.inputs, n)
+	return len(repairContextToolCalls(c.sanitized[:kept:kept], syntheticToolClosureError))
+}
+
+// index is where input i lands: the last message normalizing inputs up to
+// and including i yields, when i is a user message normalizing keeps.
+func (c normalizedContext) index(i *int) *int {
+	if i == nil {
+		return nil
+	}
+	kept, found := slices.BinarySearch(c.inputs, *i)
+	if !found || !strings.EqualFold(strings.TrimSpace(c.sanitized[kept].model.Role), "user") {
+		return nil
+	}
+	prefix := repairContextToolCalls(c.sanitized[:kept+1:kept+1], syntheticToolClosureError)
+	if len(prefix) == 0 || !strings.EqualFold(strings.TrimSpace(prefix[len(prefix)-1].model.Role), "user") {
+		return nil
+	}
+	return intPointer(len(prefix) - 1)
+}

internal/agent/application/service.go

@@ -559,12 +559,14 @@ func (s *Service) resolveWithHTTPClient(ctx context.Context, req ChatRequest, mo
 	if !usePipeline && !req.ReusePersistedUserMessage {
 		messages = append(messages, reqMessages...)
 	}
-	trimmableMessages := normalizedContextPrefixLength(messages, historyMessageCount)
-	messages, currentMessageIndex, memoryMessageIndex = normalizeContextMessages(
-		messages,
-		currentMessageIndex,
-		memoryMessageIndex,
-	)
+	normalized := normalizeContext(messages)
+	if err := interruptedPreparation(ctx, "normalize context messages"); err != nil {
+		return resolvedContext{}, req, err
+	}
+	trimmableMessages := normalized.prefixLength(historyMessageCount)
+	currentMessageIndex = normalized.index(currentMessageIndex)
+	memoryMessageIndex = normalized.index(memoryMessageIndex)
+	messages = normalized.models()
 	runCfg.ContextHistoryTokenEstimates = make([]int, len(messages))
 	for index := range messages {
 		runCfg.ContextHistoryTokenEstimates[index] = estimateMessageTokens(messages[index])
@@ -599,10 +601,9 @@ func (s *Service) resolveWithHTTPClient(ctx context.Context, req ChatRequest, mo
 		headerifiedModelQuery = turnpkg.FormatUserHeader(headerInput, modelQuery)
 	}
 	runCfg.ContextFrags = historyContextFragsForMessages(messages, historyRecords)
-	forkMessages := nonNilModelMessages(messages)
-	runCfg.ForkContextSourceMessageIDs = historySourceMessageIDsForMessages(forkMessages, historyRecords)
-	runCfg.ContextProvenance = contextProvenance(forkMessages, historyRecords, compactionInputs(req))
-	runCfg.Messages = modelMessagesToSDKMessages(forkMessages)
+	runCfg.Messages = normalized.sent()
+	runCfg.ForkContextSourceMessageIDs = historySourceMessageIDsForMessages(messages, historyRecords)
+	runCfg.ContextProvenance = contextProvenance(messages, runCfg.Messages, historyRecords, compactionInputs(req))
 	runCfg.ContextMemoryMessageIndex = memoryMessageIndex
 	if usePipeline {
 		runCfg.ContextCurrentUserMessageIndex = currentMessageIndex
@@ -619,6 +620,9 @@ func (s *Service) resolveWithHTTPClient(ctx context.Context, req ChatRequest, mo
 	runCfg.InlineAttachments = extractNativeAttachmentParts(mergedAttachments)
 	runCfg.ContextScope = buildContextFragScope(req, displayName, runCfg.Identity)
 	runCfg = runCfg.RefreshContextFrag()
+	if err := interruptedPreparation(ctx, "compile context"); err != nil {
+		return resolvedContext{}, req, err
+	}
 
 	var injectedRecords *[]InjectedMessageRecord
 	if req.InjectCh != nil {
@@ -1447,14 +1451,6 @@ func (s *Service) prepareRunConfig(ctx context.Context, cfg native.RunConfig) na
 	return cfg.RefreshContextFrag()
 }
 
-func normalizeContextMessages(messages []ModelMessage, currentIndex, memoryIndex *int) ([]ModelMessage, *int, *int) {
-	cleaned := sanitizeMessages(messages)
-	cleaned = repairToolCallClosures(cleaned, syntheticToolClosureError)
-	return cleaned,
-		remapContextMessageIndex(messages, currentIndex),
-		remapContextMessageIndex(messages, memoryIndex)
-}
-
 func prependContextMessages(prefix, messages []ModelMessage, trackedIndex *int) ([]ModelMessage, *int) {
 	if len(prefix) == 0 {
 		return messages, trackedIndex
@@ -1469,28 +1465,6 @@ func prependContextMessages(prefix, messages []ModelMessage, trackedIndex *int)
 	return append(prefix, messages...), trackedIndex
 }
 
-func normalizedContextPrefixLength(messages []ModelMessage, rawPrefixLength int) int {
-	if rawPrefixLength <= 0 || len(messages) == 0 {
-		return 0
-	}
-	rawPrefixLength = min(rawPrefixLength, len(messages))
-	prefix := sanitizeMessages(messages[:rawPrefixLength])
-	prefix = repairToolCallClosures(prefix, syntheticToolClosureError)
-	return len(prefix)
-}
-
-func remapContextMessageIndex(messages []ModelMessage, index *int) *int {
-	if index == nil || *index < 0 || *index >= len(messages) || !strings.EqualFold(strings.TrimSpace(messages[*index].Role), "user") {
-		return nil
-	}
-	prefix := sanitizeMessages(messages[:*index+1])
-	prefix = repairToolCallClosures(prefix, syntheticToolClosureError)
-	if len(prefix) == 0 || !strings.EqualFold(strings.TrimSpace(prefix[len(prefix)-1].Role), "user") {
-		return nil
-	}
-	return intPointer(len(prefix) - 1)
-}
-
 func latestModelUserMessageIndex(messages []ModelMessage) *int {
 	known := false
 	for i, message := range messages {

internal/agent/application/service_context_switch_test.go

@@ -32,7 +32,7 @@ func TestContextProvenanceKeepsTheStoredRowThroughReasoningProjection(t *testing
 		t.Fatal(err)
 	}
 	records := projectInterruptedHistoryReasoning([]historyfrag.HistoryRecord{record})
-	got := contextProvenance([]ModelMessage{records[0].ModelMessage}, records, nil)
+	got := contextProvenance([]ModelMessage{records[0].ModelMessage}, modelMessagesToSDKMessages([]ModelMessage{records[0].ModelMessage}), records, nil)
 	if len(got) != 1 || got[0].Sources[0] != (turn.ContextMessageSource{Kind: "history", ID: "checkpoint"}) || records[0].ModelMessage.TextContent() == record.ModelMessage.TextContent() {
 		t.Fatalf("projected checkpoint lost stored identity: %+v", got)
 	}
@@ -45,12 +45,9 @@ func TestContextProvenanceSurvivesToolResultNormalization(t *testing.T) {
 		ToolCallID: "call-1", ToolName: "exec", Input: toolexec.ArgumentsFromValue(map[string]any{}),
 	}}}})[0]
 	result := ModelMessage{Role: "tool", ToolCallID: "call-1", Name: "exec", Content: json.RawMessage(`{"ok":true}`)}
-	records := []historyfrag.HistoryRecord{
-		{DBMessageID: "call-row", ModelMessage: call},
-		{DBMessageID: "result-row", ModelMessage: result},
-	}
-	messages, _, _ := normalizeContextMessages([]ModelMessage{call, result}, nil, nil)
-	got := contextProvenance(messages, records, nil)
+	records := []historyfrag.HistoryRecord{loadedRecord("call-row", call), loadedRecord("result-row", result)}
+	normalized := normalizeContext([]ModelMessage{records[0].ModelMessage, records[1].ModelMessage})
+	got := contextProvenance(normalized.models(), normalized.sent(), records, nil)
 	if len(got) != 2 || len(got[1].Sources) != 1 || got[1].Sources[0].ID != "result-row" {
 		t.Fatalf("normalized tool exchange lost its result source: %+v", got)
 	}
@@ -79,8 +76,9 @@ func TestContextProvenanceNamesTheCallRowForItsClosure(t *testing.T) {
 		}}}),
 		stored("later-row", "user", sdk.UserMessage("later question")),
 	}
-	messages, _, _ := normalizeContextMessages([]ModelMessage{records[0].ModelMessage, records[1].ModelMessage}, nil, nil)
-	got := contextProvenance(messages, records, nil)
+	normalized := normalizeContext([]ModelMessage{records[0].ModelMessage, records[1].ModelMessage})
+	messages := normalized.models()
+	got := contextProvenance(messages, normalized.sent(), records, nil)
 	call := got[0].Sources
 	if len(messages) != 3 || len(got) != 3 || got[1].Message.Role != sdk.MessageRoleTool || !reflect.DeepEqual(got[1].Sources, call) ||
 		call[0].ID != "call-row" || got[1].Anchor || got[2].Sources[0].ID != "later-row" {
@@ -128,8 +126,9 @@ func TestContextProvenanceKeepsTheSourceOfAMovedDecisionResult(t *testing.T) {
 	for _, record := range records {
 		input = append(input, record.ModelMessage)
 	}
-	messages, _, _ := normalizeContextMessages(input, nil, nil)
-	got := contextProvenance(messages, records, nil)
+	normalized := normalizeContext(input)
+	messages := normalized.models()
+	got := contextProvenance(messages, normalized.sent(), records, nil)
 	if len(got) != len(messages) {
 		t.Fatalf("provenance = %+v, want every message named", got)
 	}
@@ -173,11 +172,12 @@ func TestContextProvenanceNamesWhatEachMessageRenders(t *testing.T) {
 	messages = append(messages, pipeline, live)
 	records := []historyfrag.HistoryRecord{{ModelMessage: messages[0], SourceKind: historyfrag.SourceCompactionLog, Ref: contextfrag.ContextRef{ID: "artifact-0"}}}
 	for i, id := range []string{"old-user", "old-assistant", "request", "call", "result"} {
-		records = append(records, historyfrag.HistoryRecord{ModelMessage: messages[i+1], DBMessageID: id})
+		records = append(records, loadedRecord(id, messages[i+1]))
+		messages[i+1] = records[i+1].ModelMessage
 	}
 	req := ChatRequest{PersistedUserMessageID: "request", inputInHistory: true}
 
-	got := contextProvenance(messages, records, compactionInputs(req))
+	got := contextProvenance(messages, modelMessagesToSDKMessages(messages), records, compactionInputs(req))
 	want := []struct {
 		kind, id string
 		anchor   bool
@@ -434,3 +434,10 @@ func TestSwitchContextChoosesWhatAFoldKeeps(t *testing.T) {
 		})
 	}
 }
+
+// loadedRecord is a history record as FromDBMessage loads it: its message
+// names the row it renders.
+func loadedRecord(id string, message ModelMessage) historyfrag.HistoryRecord {
+	message.ContextSource = &turn.ContextMessageSource{Kind: "history", ID: id}
+	return historyfrag.HistoryRecord{DBMessageID: id, ModelMessage: message}
+}

internal/agent/application/service_continuation.go

@@ -126,9 +126,9 @@ func (s *Service) prepareContinuationRunConfig(
 	// provider-valid. Applies to every continuation path that resumes after a
 	// deferred tool call. Provenance names the outgoing messages, closures
 	// included, so a fold removes each closure with its call.
-	outgoing := repairToolCallClosures(nonNilModelMessages(messages), syntheticToolClosureError)
-	base.ContextProvenance = contextProvenance(outgoing, retained, append(compactionInputs(req), req.continuationAnchors...))
-	base.Messages = modelMessagesToSDKMessages(outgoing)
+	outgoing := repairContextToolCalls(newContextMessages(messages), syntheticToolClosureError)
+	base.Messages = contextSentMessages(outgoing)
+	base.ContextProvenance = contextProvenance(contextModelMessages(outgoing), base.Messages, retained, append(compactionInputs(req), req.continuationAnchors...))
 	base.ContextCurrentUserMessageIndex = nil
 	base.ContextMemoryMessageIndex = nil
 	base.Query = ""

internal/agent/application/service_fork_context_test.go

@@ -3,25 +3,16 @@ package application
 import (
 	"testing"
 
+	sdk "github.com/felinics/twilight/sdk"
+
 	historyfrag "github.com/felinics/memoh/internal/agent/context/history"
+	"github.com/felinics/memoh/internal/agent/toolexec"
 )
 
 func TestHistorySourceMessageIDsFollowRetainedDatabaseMessages(t *testing.T) {
 	records := []historyfrag.HistoryRecord{
-		{
-			DBMessageID: "00000000-0000-0000-0000-000000000001",
-			ModelMessage: ModelMessage{
-				Role:    "user",
-				Content: newTextContent("first"),
-			},
-		},
-		{
-			DBMessageID: "00000000-0000-0000-0000-000000000002",
-			ModelMessage: ModelMessage{
-				Role:    "assistant",
-				Content: newTextContent("second"),
-			},
-		},
+		loadedRecord("00000000-0000-0000-0000-000000000001", ModelMessage{Role: "user", Content: newTextContent("first")}),
+		loadedRecord("00000000-0000-0000-0000-000000000002", ModelMessage{Role: "assistant", Content: newTextContent("second")}),
 	}
 	messages := []ModelMessage{
 		{Role: "system", Content: newTextContent("runtime workspace context")},
@@ -40,3 +31,22 @@ func TestHistorySourceMessageIDsFollowRetainedDatabaseMessages(t *testing.T) {
 		}
 	}
 }
+
+// A tool row answering a live call and an orphan one is rewritten to the
+// live result; the rewrite still renders that row.
+func TestHistorySourceMessageIDsSurviveToolClosureRepair(t *testing.T) {
+	call := sdkMessagesToModelMessages([]sdk.Message{{Role: sdk.MessageRoleAssistant, Content: []sdk.MessagePart{
+		sdk.ToolCallPart{ToolCallID: "live", ToolName: "exec", Input: toolexec.ArgumentsFromValue(map[string]any{})},
+	}}})[0]
+	results := sdkMessagesToModelMessages([]sdk.Message{sdk.ToolMessage(
+		sdk.ToolResultPart{ToolCallID: "live", ToolName: "exec", Result: toolexec.OutputFromValue("up")},
+		sdk.ToolResultPart{ToolCallID: "orphan", ToolName: "exec", Result: toolexec.OutputFromValue("up")},
+	)})[0]
+	records := []historyfrag.HistoryRecord{loadedRecord("call-row", call), loadedRecord("result-row", results)}
+
+	normalized := normalizeContext([]ModelMessage{records[0].ModelMessage, records[1].ModelMessage})
+	got := historySourceMessageIDsForMessages(normalized.models(), records)
+	if len(got) != 2 || got[1] != "result-row" || len(toolResultParts(normalized.sent()[1])) != 1 {
+		t.Fatalf("source IDs = %v for %+v, want the filtered result to keep result-row", got, normalized.sent())
+	}
+}

internal/agent/application/service_history.go

@@ -152,6 +152,9 @@ func (s *Service) loadHistoryRecords(ctx context.Context, fallback historyfrag.S
 			result = append(result, historyfrag.SummaryRecord(item.SummaryID, item.Summary, scope))
 			continue
 		}
+		if err := interruptedPreparation(ctx, "decode history rows"); err != nil {
+			return nil, false, err
+		}
 		msg, ok := kept[item.RowID]
 		if !ok {
 			return nil, false, fmt.Errorf("history checkpoint %s keeps row %s, which is not a context row", held.ID, item.RowID)
@@ -165,6 +168,9 @@ func (s *Service) loadHistoryRecords(ctx context.Context, fallback historyfrag.S
 	// The budget covers content only; metadata arrives undecoded. Release each
 	// row once converted so at most one row's metadata is decoded at a time.
 	for i := range msgs {
+		if err := interruptedPreparation(ctx, "decode history rows"); err != nil {
+			return nil, false, err
+		}
 		record, err := historyfrag.FromDBMessageWithLogger(s.logger, msgs[i], fallback)
 		if err != nil {
 			return nil, false, err