diff --git a/backend/internal/app/cloud_agent_context_budget_test.go b/backend/internal/app/cloud_agent_context_budget_test.go index e49f90f90..a0e64f5c5 100644 --- a/backend/internal/app/cloud_agent_context_budget_test.go +++ b/backend/internal/app/cloud_agent_context_budget_test.go @@ -76,6 +76,13 @@ func TestCloudAgentEstimatedTokensIsConservativeForChinese(t *testing.T) { } } +func TestCloudAgentCompactionReadingDoesNotTrustInvalidConfiguredWindow(t *testing.T) { + budget := cloudAgentContextBudget{InputBudgetTokens: 1000, CompactAtTokens: 850, Source: "channel-model", Configured: false} + if _, configured := cloudAgentCompactionReadingFor(budget, 900); configured { + t.Fatal("invalid capability window was treated as a real token budget") + } +} + func TestFitCloudAgentModelContextUsesTokenBudget(t *testing.T) { request := canonicalAgentRequest{Messages: []map[string]any{{"role": "user", "content": strings.Repeat("中文", 5000)}}} if err := fitCloudAgentModelContext(&request, 1_000); err == nil { diff --git a/backend/internal/app/cloud_agent_context_compaction.go b/backend/internal/app/cloud_agent_context_compaction.go index 6866a9a81..805660ef1 100644 --- a/backend/internal/app/cloud_agent_context_compaction.go +++ b/backend/internal/app/cloud_agent_context_compaction.go @@ -68,7 +68,7 @@ func cloudAgentCompactionReadingFor(budget cloudAgentContextBudget, projectedTok if budget.InputBudgetTokens > 0 { reading.PressureRatio = math.Round(float64(projectedTokens)/float64(budget.InputBudgetTokens)*10000) / 10000 } - if budget.Source == "" || budget.Source == "default" { + if !budget.Configured || budget.Source == "" || budget.Source == "default" { return reading, false } return reading, true @@ -470,7 +470,7 @@ func (s *Service) writeCloudAgentContextCheckpoint(run *model.CloudAgentExecutio recent := cloudAgentCompleteTurnTail(state.Canonical.Messages, cloudAgentContextKeepPairs) history, err := cloudAgentCheckpointHistory(checkpoint, recent) if err != nil { - return fmt.Errorf("%w: %v", errCloudAgentCheckpoint, err) + return cloudAgentCheckpointFailure("compaction checkpoint history", err) } return s.repo.MutateCloudAgent(run.UserID, run.ID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error { state.ContextCheckpoint = &checkpoint diff --git a/backend/internal/app/cloud_agent_context_compaction_test.go b/backend/internal/app/cloud_agent_context_compaction_test.go index 3efaae64e..48101c4b3 100644 --- a/backend/internal/app/cloud_agent_context_compaction_test.go +++ b/backend/internal/app/cloud_agent_context_compaction_test.go @@ -4,6 +4,7 @@ package app import ( "context" "encoding/json" + "fmt" "strings" "testing" @@ -120,6 +121,43 @@ func TestCloudAgentCompactionRequestsWhenTokenLineReached(t *testing.T) { } } +// Read results are replay caches, not the source of truth. A few large canvas +// snapshots must not make the requested compaction checkpoint impossible to +// persist, otherwise the outer error is misreported as a context overflow and +// the compression task is never even enqueued. +func TestCloudAgentCompactionCanCheckpointAfterPruningOversizedReadCache(t *testing.T) { + s, _, run, state := func() (*Service, *gorm.DB, *model.CloudAgentExecution, cloudAgentRuntime) { + s, db, root := reliableAgentRoot(t) + current, decoded := decodeCloudAgentCompactionState(t, s, root.ID) + return s, db, current, decoded + }() + + state.ToolReadResults = make(map[string]cloudAgentCachedToolResult) + state.ToolReadReplays = make(map[string]int) + for index := 0; index < 8; index++ { + key := fmt.Sprintf("canvas_get_state:{\"offset\":%d}", index) + state.ToolReadResults[key] = cloudAgentCachedToolResult{Result: json.RawMessage(`{"content":"` + strings.Repeat("x", 100000) + `"}`)} + state.ToolReadReplays[key] = index + 1 + } + state.ActiveTaskID = "" + budget := cloudAgentContextBudgetFor(8_000, 2_000, "test") + requested, err := s.cloudAgentRequestCompaction(run, &state, budget, budget.CompactAtTokens) + if err != nil || !requested { + t.Fatalf("大只读缓存不应阻塞压缩请求: requested=%v err=%v", requested, err) + } + + persisted, decoded := decodeCloudAgentCompactionState(t, s, run.ID) + if decoded.ContextCompaction == nil || decoded.ContextCompaction.Status != "requested" { + t.Fatalf("压缩请求没有落库: %+v", decoded.ContextCompaction) + } + if len(decoded.ToolReadResults) >= 8 || len(decoded.ToolReadResults) != len(decoded.ToolReadReplays) { + t.Fatalf("超大只读缓存没有按需清理: results=%d replays=%d", len(decoded.ToolReadResults), len(decoded.ToolReadReplays)) + } + if len(persisted.StateJSON) >= cloudAgentStateJSONLimit { + t.Fatalf("裁剪缓存后 StateJSON 仍超过上限: %d", len(persisted.StateJSON)) + } +} + // 没有配模型上下文窗口时退回字节/条数兜底判据，且兜底判据同样必须"暂停而不是判死"。 func TestCloudAgentCompactionFallsBackToBytesWithoutModelWindow(t *testing.T) { s, db, root := reliableAgentRoot(t) diff --git a/backend/internal/app/cloud_agent_runtime.go b/backend/internal/app/cloud_agent_runtime.go index fb9a69178..1a3ae30c6 100644 --- a/backend/internal/app/cloud_agent_runtime.go +++ b/backend/internal/app/cloud_agent_runtime.go @@ -7,6 +7,7 @@ import ( "fmt" "log" "math" + "sort" "strings" "time" "unicode/utf8" @@ -22,6 +23,44 @@ import ( // A deterministic checkpoint failure must not be retried forever like a transient DB error. var errCloudAgentCheckpoint = errors.New("invalid Agent checkpoint") +// cloudAgentCheckpointError keeps the save stage attached to a deterministic +// checkpoint failure. Previously every failure was flattened to the same +// sentinel and advanceCloudAgent consequently told users that context had +// overflowed even when the actual problem was a corrupt event or an encoding +// failure. +type cloudAgentCheckpointError struct { + Stage string + Err error +} + +func (e *cloudAgentCheckpointError) Error() string { + if e == nil { + return errCloudAgentCheckpoint.Error() + } + if e.Stage == "" { + return fmt.Sprintf("%v: %v", errCloudAgentCheckpoint, e.Err) + } + return fmt.Sprintf("%v (%s): %v", errCloudAgentCheckpoint, e.Stage, e.Err) +} + +func (e *cloudAgentCheckpointError) Unwrap() error { + if e == nil { + return errCloudAgentCheckpoint + } + return e.Err +} + +func (e *cloudAgentCheckpointError) Is(target error) bool { + return target == errCloudAgentCheckpoint || (e != nil && errors.Is(e.Err, target)) +} + +func cloudAgentCheckpointFailure(stage string, err error) error { + if err == nil { + err = errors.New("unknown checkpoint failure") + } + return &cloudAgentCheckpointError{Stage: stage, Err: err} +} + type CloudAgentEvent struct { EventID string `json:"eventId"` RunID string `json:"runId"` @@ -625,6 +664,79 @@ func cloudAgentContainsString(values []string, target string) bool { } return false } + +const cloudAgentStateJSONLimit = 512 << 10 + +func cloudAgentCheckpointJSON(state *cloudAgentRuntime) ([]byte, error) { + checkpoint := *state + checkpoint.Canonical.Messages = nil + checkpoint.TextHistory = nil + checkpoint.Events = nil + return json.Marshal(checkpoint) +} + +// cloudAgentPruneReadCacheForCheckpoint is a last-resort durability guard. +// ToolReadResults is a replay cache: losing one entry means a later read may +// execute again, but allowing that cache to make the whole runtime impossible +// to checkpoint loses the run and prevents context compaction from starting. +// Remove the largest entries first so a single oversized canvas/skill result +// cannot strand the run above the 512 KiB state limit. +func cloudAgentPruneReadCacheForCheckpoint(state *cloudAgentRuntime, limit int) (removed int) { + if state == nil || len(state.ToolReadResults) == 0 || limit <= 0 { + return 0 + } + raw, err := cloudAgentCheckpointJSON(state) + if err != nil || len(raw) <= limit { + return 0 + } + type candidate struct { + key string + bytes int + inContext bool + error bool + } + candidates := make([]candidate, 0, len(state.ToolReadResults)) + for key, cached := range state.ToolReadResults { + entryBytes, _ := json.Marshal(cached) + candidates = append(candidates, candidate{ + key: key, + bytes: len(entryBytes), + inContext: cloudAgentReadResultInContext(state, cached.Result), + error: cached.Error != "", + }) + } + // Prefer dropping errors and entries already absent from the current + // transcript. Keep results currently visible to the model for as long as + // possible because compaction may need them for a replay after eviction. + sort.Slice(candidates, func(i, j int) bool { + if candidates[i].inContext != candidates[j].inContext { + return !candidates[i].inContext + } + if candidates[i].error != candidates[j].error { + return candidates[i].error + } + if candidates[i].bytes != candidates[j].bytes { + return candidates[i].bytes > candidates[j].bytes + } + return candidates[i].key < candidates[j].key + }) + for _, item := range candidates { + if raw, err = cloudAgentCheckpointJSON(state); err != nil || len(raw) <= limit { + break + } + delete(state.ToolReadResults, item.key) + delete(state.ToolReadReplays, item.key) + removed++ + } + if len(state.ToolReadResults) == 0 { + state.ToolReadResults = nil + } + if len(state.ToolReadReplays) == 0 { + state.ToolReadReplays = nil + } + return removed +} + func cloudAgentSave(run *model.CloudAgentExecution, state *cloudAgentRuntime) error { if run == nil || state == nil { return errors.New("Agent runtime state is missing") @@ -639,25 +751,31 @@ func cloudAgentSave(run *model.CloudAgentExecution, state *cloudAgentRuntime) er } if run.ID != "" { if err := validateCloudAgentRuntime(run, state); err != nil { - return fmt.Errorf("%w: %v", errCloudAgentCheckpoint, err) + return cloudAgentCheckpointFailure("runtime validation", err) } for index, event := range state.Events { sequence := state.EventSeqBase + index + 1 if event.Seq != sequence || event.RunID != run.ID || event.EventID != fmt.Sprintf("%s:%d", run.ID, sequence) { - return fmt.Errorf("%w: Agent event sequence or identity is invalid", errCloudAgentCheckpoint) + return cloudAgentCheckpointFailure("event identity", errors.New("Agent event sequence or identity is invalid")) } } } - checkpoint := *state - checkpoint.Canonical.Messages = nil - checkpoint.TextHistory = nil - checkpoint.Events = nil - raw, err := json.Marshal(checkpoint) + // The canonical transcript and event journal are persisted separately. The + // bounded StateJSON therefore contains only runtime metadata and replay + // caches, and the latter must never be allowed to block a checkpoint. + raw, err := cloudAgentCheckpointJSON(state) if err != nil { - return fmt.Errorf("%w: %v", errCloudAgentCheckpoint, err) + return cloudAgentCheckpointFailure("state encode", err) } - if len(raw) > 512<<10 { - return fmt.Errorf("%w: Agent 状态超过 512KB 上限", errCloudAgentCheckpoint) + if len(raw) > cloudAgentStateJSONLimit { + cloudAgentPruneReadCacheForCheckpoint(state, cloudAgentStateJSONLimit) + raw, err = cloudAgentCheckpointJSON(state) + if err != nil { + return cloudAgentCheckpointFailure("state encode after cache pruning", err) + } + } + if len(raw) > cloudAgentStateJSONLimit { + return cloudAgentCheckpointFailure("state size", fmt.Errorf("Agent 状态超过 512KB 上限（%d bytes）", len(raw))) } run.CanvasID, run.ActiveTaskID, run.MediaTaskID = state.Request.CanvasID, state.ActiveTaskID, state.MediaTaskID run.ParentID = state.ParentID @@ -671,7 +789,7 @@ func cloudAgentSave(run *model.CloudAgentExecution, state *cloudAgentRuntime) er for _, event := range state.Events { body, err := json.Marshal(event) if err != nil { - return fmt.Errorf("%w: encode Agent event: %v", errCloudAgentCheckpoint, err) + return cloudAgentCheckpointFailure("event encode", err) } run.Journal = append(run.Journal, model.CloudAgentEventRecord{RunID: run.ID, UserID: run.UserID, Sequence: event.Seq, EventJSON: string(body), CreatedAt: event.CreatedAt}) } @@ -679,14 +797,14 @@ func cloudAgentSave(run *model.CloudAgentExecution, state *cloudAgentRuntime) er for index, message := range state.Canonical.Messages { body, err := json.Marshal(message) if err != nil { - return fmt.Errorf("%w: encode Agent message: %v", errCloudAgentCheckpoint, err) + return cloudAgentCheckpointFailure("canonical message encode", err) } run.Transcript = append(run.Transcript, model.CloudAgentMessageRecord{RunID: run.ID, UserID: run.UserID, Kind: "canonical", Sequence: index + 1, MessageJSON: string(body)}) } for index, message := range state.TextHistory { body, err := json.Marshal(message) if err != nil { - return fmt.Errorf("%w: encode Agent history: %v", errCloudAgentCheckpoint, err) + return cloudAgentCheckpointFailure("history encode", err) } run.Transcript = append(run.Transcript, model.CloudAgentMessageRecord{RunID: run.ID, UserID: run.UserID, Kind: "history", Sequence: index + 1, MessageJSON: string(body)}) } @@ -858,7 +976,17 @@ func (s *Service) wakeCloudAgentScheduler() { func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error) { defer func() { if errors.Is(err, errCloudAgentCheckpoint) { - err = s.terminateCloudAgent(run, "Agent 上下文或执行记录超过安全限制，本轮已停止；已有任务结果保留在任务中心") + message := "Agent 运行状态保存失败，本轮已停止；已有任务结果保留在任务中心" + var checkpointErr *cloudAgentCheckpointError + if errors.As(err, &checkpointErr) { + log.Printf("agent checkpoint rejected run=%s stage=%s error=%v", run.ID, checkpointErr.Stage, checkpointErr.Err) + if checkpointErr.Stage == "state size" { + message = "Agent 上下文或执行记录超过安全限制，本轮已停止；已有任务结果保留在任务中心" + } + } else { + message = "Agent 运行状态保存失败，本轮已停止；已有任务结果保留在任务中心" + } + err = s.terminateCloudAgent(run, message) } }() if run.CleanupPending {