| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "time" |
| 8 | |
| 9 | "reasonix/internal/provider" |
| 10 | ) |
| 11 | |
| 12 | type summaryProjectionCommit struct { |
| 13 | canonical, fold, projected []provider.Message |
| 14 | result foldSummary |
| 15 | transcriptVersion, projectionVersion, generation uint64 |
| 16 | activeTurn int64 |
| 17 | trigger, summary, inputHash, outputHash string |
| 18 | sourceTokens, projectionTokens int |
| 19 | // covered is the canonical length the frozen projection body represents; |
| 20 | // messages past it splice live from the transcript. |
| 21 | covered int |
| 22 | } |
| 23 | |
| 24 | // commitSummaryProjection CAS-installs a checkpoint under compactionMu: |
| 25 | // transcript version/hash, projection version, and generation must still match. |
| 26 | // The maintenance event is emitted only after the lock is released so a sink |
| 27 | // that re-enters ContextMaintenanceSnapshot cannot deadlock. |
| 28 | func (a *Agent) commitSummaryProjection(commit summaryProjectionCommit) (CompactionState, error) { |
| 29 | state := a.summaryProjectionState(commit) |
| 30 | a.sess.compactionMu.Lock() |
| 31 | current, currentVersion := a.sess.conversation.snapshotMessagesVersion() |
| 32 | if currentVersion != commit.transcriptVersion || |
| 33 | len(current) != len(commit.canonical) || |
| 34 | coveredPrefixHash(current, len(current)) != coveredPrefixHash(commit.canonical, len(commit.canonical)) || |
| 35 | a.sess.compactionState.Projection.ProjectionVersion != commit.projectionVersion || |
| 36 | a.sess.compactionState.Generation != commit.generation { |
| 37 | a.sess.compactionMu.Unlock() |
| 38 | return CompactionState{}, errCompressStaleContext |
| 39 | } |
| 40 | prev := a.sess.compactionState |
| 41 | a.sess.compactionState = state |
| 42 | accepted, err := a.persistInstalledProjectionLocked(context.Background(), state, current) |
| 43 | if err != nil { |
| 44 | if accepted { |
| 45 | a.sess.checkpointState = "pending" |
| 46 | a.sess.compactionMu.Unlock() |
| 47 | return CompactionState{}, fmt.Errorf("persist projection: %w", err) |
| 48 | } |
| 49 | a.sess.compactionState = prev |
| 50 | a.sess.compactionMu.Unlock() |
| 51 | if errors.Is(err, errCompressStaleContext) { |
| 52 | return CompactionState{}, err |
| 53 | } |
| 54 | return CompactionState{}, fmt.Errorf("persist projection: %w", err) |
| 55 | } |
| 56 | a.sess.checkpointState = "applied" |
| 57 | if commit.activeTurn != 0 && commit.trigger != CompactionTriggerManual { |
| 58 | a.sess.compaction.lastTurn.Store(commit.activeTurn) |
| 59 | } |
| 60 | receipt := state.LastReceipt |
| 61 | a.sess.compactionMu.Unlock() |
| 62 | a.emitContextMaintenance(receipt) |
| 63 | return state, nil |
| 64 | } |
| 65 | |
| 66 | func (a *Agent) persistInstalledProjectionLocked(ctx context.Context, state CompactionState, canonical []provider.Message) (bool, error) { |
| 67 | accepted := false |
| 68 | if recorder, ok := a.svc.sessionCheckpointer.(SessionModelContextRecorder); ok { |
| 69 | visible := modelVisibleFromProjection(state.Projection, canonical) |
| 70 | commit := cloneSessionModelContextCommit(SessionModelContextCommit{ |
| 71 | OperationID: state.LastReceipt.OperationID, |
| 72 | Reason: state.LastReceipt.Action, |
| 73 | Messages: visible, |
| 74 | }) |
| 75 | result, err := recorder.RecordSessionModelContext(ctx, commit) |
| 76 | accepted = result.Accepted |
| 77 | if err != nil { |
| 78 | if accepted { |
| 79 | a.sess.pendingModelContextCommit = &commit |
| 80 | } |
| 81 | return accepted, err |
| 82 | } |
| 83 | if result.Accepted && !result.Durable { |
| 84 | a.sess.pendingModelContextCommit = &commit |
| 85 | return true, errors.New("model context commit was accepted but is not durable") |
| 86 | } |
| 87 | } |
| 88 | if err := a.persistCompactionStateLocked(); err != nil { |
| 89 | if accepted { |
| 90 | visible := modelVisibleFromProjection(state.Projection, canonical) |
| 91 | commit := cloneSessionModelContextCommit(SessionModelContextCommit{ |
| 92 | OperationID: state.LastReceipt.OperationID, |
| 93 | Reason: state.LastReceipt.Action, |
| 94 | Messages: visible, |
| 95 | }) |
| 96 | a.sess.pendingModelContextCommit = &commit |
| 97 | } |
| 98 | return accepted, err |
| 99 | } |
| 100 | a.sess.pendingModelContextCommit = nil |
| 101 | return accepted, nil |
| 102 | } |
| 103 | |
| 104 | func (a *Agent) summaryProjectionState(commit summaryProjectionCommit) CompactionState { |
| 105 | projectionVersion := commit.projectionVersion + 1 |
| 106 | now := time.Now().UTC() |
| 107 | summaryHash := summaryContentHash(commit.summary) |
| 108 | coveredHash := coveredPrefixHash(commit.canonical, commit.covered) |
| 109 | receipt := &ContextMaintenanceReceipt{ |
| 110 | OperationID: fmt.Sprintf("summary-%d-%s", projectionVersion, commit.outputHash), Status: "applied", |
| 111 | Action: "summary", Trigger: commit.trigger, SourceProjection: commit.projectionVersion, |
| 112 | ProjectionVersion: projectionVersion, CoveredCount: commit.covered, CoveredPrefixHash: coveredHash, |
| 113 | InputHash: commit.inputHash, OutputHash: commit.outputHash, InputTokens: commit.sourceTokens, |
| 114 | ResultTokens: commit.projectionTokens, SavedTokens: max(0, commit.sourceTokens-commit.projectionTokens), |
| 115 | SummaryHash: summaryHash, CacheBreak: true, CreatedAt: now, |
| 116 | } |
| 117 | // LastReceipt is authoritative; do not mirror last_trigger/last_mode/token |
| 118 | // counters or top-level blocked_* fields (stripped again on save). |
| 119 | return CompactionState{ |
| 120 | SchemaVersion: compactionStateSchemaCurrent, TranscriptVersion: commit.transcriptVersion, |
| 121 | Generation: commit.generation + 1, PromptCacheKey: a.currentPromptCacheKey(), |
| 122 | Projection: ContextProjection{ |
| 123 | Messages: commit.projected, TranscriptVersion: commit.transcriptVersion, |
| 124 | ProjectionVersion: projectionVersion, CoveredCount: commit.covered, CoveredPrefixHash: coveredHash, |
| 125 | PinnedContextHash: pinnedContextCoverageHash(commit.canonical, commit.covered), |
| 126 | SummaryHash: summaryHash, SourceTokens: commit.sourceTokens, ProjectionTokens: commit.projectionTokens, |
| 127 | ViewInputHash: commit.inputHash, ViewOutputHash: commit.outputHash, CreatedAt: now, |
| 128 | }, |
| 129 | LastReceipt: receipt, UpdatedAt: now, |
| 130 | } |
| 131 | } |
| 132 |