| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "log/slog" |
| 6 | "strings" |
| 7 | "time" |
| 8 | |
| 9 | "reasonix/internal/provider" |
| 10 | ) |
| 11 | |
| 12 | // ErrCompactionRequired is returned when the prompt exceeds the provider limit |
| 13 | // and compaction could not produce a usable projection. Callers may retry. |
| 14 | var ErrCompactionRequired = errors.New("context exceeds provider limit and compaction failed") |
| 15 | |
| 16 | // modelVisibleMessages returns the provider-bound message list: a valid |
| 17 | // projection plus any post-projection appends, otherwise the full canonical |
| 18 | // transcript. LocalOnly stripping still happens in prepareSamplingRequest. |
| 19 | func (a *Agent) modelVisibleMessages() []provider.Message { |
| 20 | if a == nil || a.sess.conversation == nil { |
| 21 | return nil |
| 22 | } |
| 23 | msgs, _ := a.sess.conversation.snapshotMessagesVersion() |
| 24 | a.sess.compactionMu.Lock() |
| 25 | st := a.sess.compactionState |
| 26 | a.sess.compactionMu.Unlock() |
| 27 | if projectionValid(st, msgs, a.currentPromptCacheKey()) { |
| 28 | if visible := modelVisibleFromProjection(st.Projection, msgs); len(visible) > 0 { |
| 29 | return visible |
| 30 | } |
| 31 | } |
| 32 | return msgs |
| 33 | } |
| 34 | |
| 35 | func (a *Agent) currentProjectionVersion() uint64 { |
| 36 | if a == nil { |
| 37 | return 0 |
| 38 | } |
| 39 | a.sess.compactionMu.Lock() |
| 40 | defer a.sess.compactionMu.Unlock() |
| 41 | return a.sess.compactionState.Projection.ProjectionVersion |
| 42 | } |
| 43 | |
| 44 | // currentPromptCacheKey is the lineage key for the bound session + model. |
| 45 | func (a *Agent) currentPromptCacheKey() string { |
| 46 | if a == nil { |
| 47 | return "" |
| 48 | } |
| 49 | a.sess.compactionMu.Lock() |
| 50 | defer a.sess.compactionMu.Unlock() |
| 51 | return a.currentPromptCacheKeyLocked() |
| 52 | } |
| 53 | |
| 54 | func (a *Agent) currentPromptCacheKeyLocked() string { |
| 55 | return promptCacheKey(a.workspaceID, BranchID(a.sess.path), a.modelRef) |
| 56 | } |
| 57 | |
| 58 | // InvalidateProjection drops the in-memory and on-disk projection after |
| 59 | // lineage-changing operations (rewind, branch, fork, system/model change). |
| 60 | func (a *Agent) InvalidateProjection() { |
| 61 | if a == nil { |
| 62 | return |
| 63 | } |
| 64 | // A strong reasoning-replay overlay is indexed against the old canonical |
| 65 | // history. Clear it together with the compaction projection so rewind, |
| 66 | // branch, and model/system lineage changes cannot reuse a stale anchor. |
| 67 | a.sess.clearReasoningReplayStrongProjection() |
| 68 | a.sess.compactionMu.Lock() |
| 69 | path := a.sess.path |
| 70 | a.sess.compactionState = CompactionState{} |
| 71 | a.sess.compactionMu.Unlock() |
| 72 | a.sess.compaction.stuck = false |
| 73 | a.sess.compaction.stuckInputHash = "" |
| 74 | a.sess.compaction.consecutive = 0 |
| 75 | a.sess.compaction.failedTurn.Store(0) |
| 76 | a.sess.compaction.lastTurn.Store(0) |
| 77 | if path != "" { |
| 78 | if err := RemoveCompactionState(path); err != nil { |
| 79 | slog.Warn("agent: remove context projection", "err", err) |
| 80 | } |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | // InvalidateProjectionIfStale keeps the projection when it still matches the |
| 85 | // current transcript and performs the full invalidation otherwise. History |
| 86 | // rewrites that only touch messages past CoveredCount keep their fold. |
| 87 | func (a *Agent) InvalidateProjectionIfStale() { |
| 88 | if a == nil { |
| 89 | return |
| 90 | } |
| 91 | // This helper is called after a history rewrite even when the existing |
| 92 | // compaction fold remains valid (for example a tail-only rewind). The |
| 93 | // reasoning-replay overlay still belongs to the pre-rewrite shape. |
| 94 | a.sess.clearReasoningReplayStrongProjection() |
| 95 | a.sess.compactionMu.Lock() |
| 96 | st := a.sess.compactionState |
| 97 | if len(st.Projection.Messages) > 0 && a.sess.conversation != nil { |
| 98 | msgs, _ := a.sess.conversation.snapshotMessagesVersion() |
| 99 | if projectionValid(st, msgs, a.currentPromptCacheKeyLocked()) { |
| 100 | a.sess.compactionMu.Unlock() |
| 101 | return |
| 102 | } |
| 103 | } |
| 104 | a.sess.compactionMu.Unlock() |
| 105 | a.InvalidateProjection() |
| 106 | } |
| 107 | |
| 108 | // LoadProjectionSidecar loads the context sidecar into the agent. Corrupt or |
| 109 | // incompatible state is dropped so the next request rebuilds from canonical. |
| 110 | // Sidecars whose PromptCacheKey does not match the current agent lineage are |
| 111 | // discarded without deleting the file (another model may still own it). |
| 112 | func (a *Agent) LoadProjectionSidecar(sessionPath string) { |
| 113 | if a == nil { |
| 114 | return |
| 115 | } |
| 116 | a.sess.compactionMu.Lock() |
| 117 | a.sess.path = sessionPath |
| 118 | a.sess.compactionState = CompactionState{} |
| 119 | a.sess.checkpointState = "none" |
| 120 | a.sess.pendingModelContextCommit = nil |
| 121 | a.sess.compactionMu.Unlock() |
| 122 | if sessionPath == "" { |
| 123 | a.resetCompactionState() |
| 124 | return |
| 125 | } |
| 126 | st, ok, err := LoadCompactionState(sessionPath) |
| 127 | if err != nil { |
| 128 | slog.Warn("agent: load context projection", "err", err) |
| 129 | _ = RemoveCompactionState(sessionPath) |
| 130 | a.resetCompactionState() |
| 131 | return |
| 132 | } |
| 133 | if !ok { |
| 134 | a.resetCompactionState() |
| 135 | return |
| 136 | } |
| 137 | var msgs, preRepair []provider.Message |
| 138 | if a.sess.conversation != nil { |
| 139 | msgs, preRepair = a.sess.conversation.projectionValidationMessages() |
| 140 | } |
| 141 | needsNormalization := migratePromotedCoveredPrefixHash(&st, msgs) |
| 142 | a.sess.compactionMu.Lock() |
| 143 | key := a.currentPromptCacheKeyLocked() |
| 144 | normalized, keyOK := lineageKeyCompatible(st.PromptCacheKey, key) |
| 145 | // Keep receipt-only blocked/failed sidecars (no projection body) and legacy |
| 146 | // top-level BlockedInputHash so generation-scoped suppressions survive restart. |
| 147 | hasMaintenanceSignal := st.Projection.CoveredPrefixHash != "" || |
| 148 | st.BlockedInputHash != "" || |
| 149 | (st.LastReceipt != nil && (st.LastReceipt.Status == "blocked" || st.LastReceipt.Status == "failed" || |
| 150 | st.LastReceipt.Status == "applied")) |
| 151 | if key != "" && !keyOK { |
| 152 | // Lineage key changed (upgrade, model/workspace switch). Rebind when |
| 153 | // the projection body still matches the canonical covered prefix. |
| 154 | contentValid := projectionContentValid(st, msgs) |
| 155 | if !contentValid && migrateLegacyCoveredPrefixHash(&st, msgs, preRepair) { |
| 156 | contentValid = true |
| 157 | needsNormalization = true |
| 158 | } |
| 159 | if contentValid { |
| 160 | normalized, keyOK = key, true |
| 161 | } |
| 162 | } |
| 163 | if (key != "" && !keyOK) || !hasMaintenanceSignal { |
| 164 | a.sess.compactionState = CompactionState{} |
| 165 | a.sess.checkpointState = "none" |
| 166 | a.sess.compactionMu.Unlock() |
| 167 | return |
| 168 | } |
| 169 | // Only rewrite legacy native-editing lineage keys; exact matches stay pure-read. |
| 170 | if keyOK && key != "" && normalized != st.PromptCacheKey { |
| 171 | st.PromptCacheKey = normalized |
| 172 | needsNormalization = true |
| 173 | } |
| 174 | // Only mark restored when the projection still matches the transcript. |
| 175 | if !projectionContentValid(st, msgs) && migrateLegacyCoveredPrefixHash(&st, msgs, preRepair) { |
| 176 | needsNormalization = true |
| 177 | } |
| 178 | valid := len(st.Projection.Messages) > 0 && projectionValid(st, msgs, key) |
| 179 | if !valid && len(st.Projection.Messages) > 0 { |
| 180 | // Keep blocked receipts / telemetry; drop unusable projection body. |
| 181 | st.Projection = ContextProjection{} |
| 182 | } |
| 183 | a.sess.compactionState = st |
| 184 | if valid { |
| 185 | a.sess.checkpointState = "restored" |
| 186 | if needsNormalization { |
| 187 | if err := a.persistCompactionStateLocked(); err != nil { |
| 188 | slog.Warn("agent: persist normalized projection lineage", "err", err) |
| 189 | } |
| 190 | } |
| 191 | } else { |
| 192 | a.sess.checkpointState = "none" |
| 193 | } |
| 194 | a.sess.compactionMu.Unlock() |
| 195 | } |
| 196 | |
| 197 | // lineageKeyCompatible reports whether a stored PromptCacheKey still belongs to |
| 198 | // the current session/model lineage. Legacy native context-editing keys used a |
| 199 | // "|context-editing-native-..." suffix on an otherwise matching base key. |
| 200 | func lineageKeyCompatible(stored, current string) (normalized string, ok bool) { |
| 201 | stored, current = strings.TrimSpace(stored), strings.TrimSpace(current) |
| 202 | if current == "" { |
| 203 | // Unknown current lineage: accept any stored key as-is. |
| 204 | return stored, true |
| 205 | } |
| 206 | if stored == "" { |
| 207 | return "", false |
| 208 | } |
| 209 | if stored == current { |
| 210 | return current, true |
| 211 | } |
| 212 | const nativeSuffix = "|context-editing-native" |
| 213 | if strings.HasPrefix(stored, current+nativeSuffix) { |
| 214 | return current, true |
| 215 | } |
| 216 | if i := strings.Index(stored, nativeSuffix); i > 0 && stored[:i] == current { |
| 217 | return current, true |
| 218 | } |
| 219 | return "", false |
| 220 | } |
| 221 | |
| 222 | func (a *Agent) resetCompactionState() { |
| 223 | a.sess.compactionMu.Lock() |
| 224 | a.sess.compactionState = CompactionState{} |
| 225 | a.sess.checkpointState = "none" |
| 226 | a.sess.pendingModelContextCommit = nil |
| 227 | a.sess.compactionMu.Unlock() |
| 228 | } |
| 229 | |
| 230 | // BindSessionPath rebinds projection persistence to path. When loadSidecar is |
| 231 | // true the existing sidecar is loaded (resume/switch); otherwise in-memory |
| 232 | // projection is cleared without deleting another session's sidecar file. |
| 233 | func (a *Agent) BindSessionPath(path string, loadSidecar bool) { |
| 234 | if a == nil { |
| 235 | return |
| 236 | } |
| 237 | if loadSidecar { |
| 238 | a.LoadProjectionSidecar(path) |
| 239 | return |
| 240 | } |
| 241 | a.sess.compactionMu.Lock() |
| 242 | a.sess.path = path |
| 243 | a.sess.compactionState = CompactionState{} |
| 244 | a.sess.checkpointState = "none" |
| 245 | a.sess.pendingModelContextCommit = nil |
| 246 | a.sess.cacheState = CacheStateUnknown |
| 247 | a.sess.compactionMu.Unlock() |
| 248 | a.sess.compaction.stuck = false |
| 249 | a.sess.compaction.stuckInputHash = "" |
| 250 | a.sess.compaction.consecutive = 0 |
| 251 | a.sess.compaction.failedTurn.Store(0) |
| 252 | a.sess.compaction.lastTurn.Store(0) |
| 253 | } |
| 254 | |
| 255 | // SetSessionPath binds the transcript path used for projection persistence. |
| 256 | func (a *Agent) SetSessionPath(path string) { |
| 257 | if a == nil { |
| 258 | return |
| 259 | } |
| 260 | a.sess.compactionMu.Lock() |
| 261 | a.sess.path = path |
| 262 | a.sess.compactionMu.Unlock() |
| 263 | } |
| 264 | |
| 265 | // SessionPath returns the bound transcript path. |
| 266 | func (a *Agent) SessionPath() string { |
| 267 | if a == nil { |
| 268 | return "" |
| 269 | } |
| 270 | a.sess.compactionMu.Lock() |
| 271 | defer a.sess.compactionMu.Unlock() |
| 272 | return a.sess.path |
| 273 | } |
| 274 | |
| 275 | // SetCacheState records the resume-time cache estimate without rewriting history. |
| 276 | func (a *Agent) SetCacheState(state string) { |
| 277 | if a == nil { |
| 278 | return |
| 279 | } |
| 280 | switch state { |
| 281 | case CacheStateWarm, CacheStateCold, CacheStateUnknown: |
| 282 | default: |
| 283 | state = CacheStateUnknown |
| 284 | } |
| 285 | a.sess.compactionMu.Lock() |
| 286 | defer a.sess.compactionMu.Unlock() |
| 287 | a.sess.cacheState = state |
| 288 | if a.sess.compactionState.SchemaVersion == 0 && len(a.sess.compactionState.Projection.Messages) == 0 { |
| 289 | a.sess.compactionState.SchemaVersion = compactionStateSchemaCurrent |
| 290 | } |
| 291 | a.sess.compactionState.LastCacheState = state |
| 292 | a.sess.compactionState.UpdatedAt = time.Now().UTC() |
| 293 | } |
| 294 | |
| 295 | // CacheState returns the last estimated cache warm/cold/unknown label. |
| 296 | func (a *Agent) CacheState() string { |
| 297 | if a == nil { |
| 298 | return CacheStateUnknown |
| 299 | } |
| 300 | a.sess.compactionMu.Lock() |
| 301 | defer a.sess.compactionMu.Unlock() |
| 302 | if a.sess.cacheState == "" { |
| 303 | return CacheStateUnknown |
| 304 | } |
| 305 | return a.sess.cacheState |
| 306 | } |
| 307 | |
| 308 | func (a *Agent) persistCompactionStateLocked() error { |
| 309 | if a.sess.path == "" { |
| 310 | return nil |
| 311 | } |
| 312 | return SaveCompactionState(a.sess.path, a.sess.compactionState) |
| 313 | } |
| 314 | |
| 315 | // promptCacheKey builds a stable lineage key for session + model identity. |
| 316 | // It deliberately excludes message counts, timestamps, and projection hashes. |
| 317 | func promptCacheKey(workspaceID, sessionLineage, modelRef string) string { |
| 318 | parts := []string{ |
| 319 | strings.TrimSpace(workspaceID), |
| 320 | strings.TrimSpace(sessionLineage), |
| 321 | strings.TrimSpace(modelRef), |
| 322 | } |
| 323 | return strings.Join(parts, "|") |
| 324 | } |
| 325 |