| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "fmt" |
| 7 | "log/slog" |
| 8 | "os" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/provider" |
| 12 | "reasonix/internal/store" |
| 13 | ) |
| 14 | |
| 15 | type dagRoute int |
| 16 | |
| 17 | const ( |
| 18 | dagRouteSchemaOne dagRoute = iota |
| 19 | dagRouteNative |
| 20 | dagRouteUpgrade |
| 21 | ) |
| 22 | |
| 23 | // SessionLogSchemaEnv is the emergency switch back to the schema-1 writer for |
| 24 | // sessions that have not been upgraded yet; upgraded logs stay schema 2. |
| 25 | const SessionLogSchemaEnv = "REASONIX_SESSION_LOG" |
| 26 | |
| 27 | func dagWriterEnabled() bool { |
| 28 | return os.Getenv(SessionLogSchemaEnv) != "v1" |
| 29 | } |
| 30 | |
| 31 | // probeLogForSave classifies the event log for a save, refusing a schema this |
| 32 | // build cannot own and healing a torn schema-1 tail before anything appends |
| 33 | // behind it. |
| 34 | func probeLogForSave(path string) (sessionEventLogProbe, error) { |
| 35 | probe, err := probeSessionEventLog(path) |
| 36 | if err != nil { |
| 37 | return probe, err |
| 38 | } |
| 39 | if probe.futureSchema { |
| 40 | return probe, fmt.Errorf("session event log for %s uses schema %d; this build supports up to %d", path, probe.schemaVersion, sessionDAGSchemaVersion) |
| 41 | } |
| 42 | if probe.native && probe.size > 0 { |
| 43 | if err := repairSessionEventLogTail(path); err != nil { |
| 44 | return probe, fmt.Errorf("repair session event log: %w", err) |
| 45 | } |
| 46 | } |
| 47 | return probe, nil |
| 48 | } |
| 49 | |
| 50 | // dagSaveRoute decides whether a save runs on the schema-2 path. An existing |
| 51 | // schema-1 log is only upgraded by the lease holder; a session with no log |
| 52 | // yet (new, or a bare checkpoint) starts schema 2 unless another runtime |
| 53 | // holds it. Foreign files and overlong names stay on the schema-1 path. |
| 54 | func (s *Session) dagSaveRoute(path string, probe sessionEventLogProbe) dagRoute { |
| 55 | if probe.dag { |
| 56 | return dagRouteNative |
| 57 | } |
| 58 | if !dagWriterEnabled() || !probe.native { |
| 59 | return dagRouteSchemaOne |
| 60 | } |
| 61 | if probe.size > 0 { |
| 62 | if !SessionLeaseHeldByCurrentRuntime(path) { |
| 63 | return dagRouteSchemaOne |
| 64 | } |
| 65 | return dagRouteUpgrade |
| 66 | } |
| 67 | if SessionLeaseHeldByOtherRuntime(path) { |
| 68 | return dagRouteSchemaOne |
| 69 | } |
| 70 | return dagRouteUpgrade |
| 71 | } |
| 72 | |
| 73 | // saveDAGLocked persists msgs to a schema-2 log: it replays (or extends) the |
| 74 | // cached graph, diffs the in-memory transcript against this session's head, |
| 75 | // appends the resulting entries in one batch, and refreshes the derived |
| 76 | // files. Concurrent writers never conflict; a writer that only fell behind |
| 77 | // its own head reports a stale-prefix conflict so the caller adopts disk. |
| 78 | func (s *Session) saveDAGLocked(path string, mode sessionSaveMode, route dagRoute, msgs []provider.Message, version uint64, rewriteVersion int, digest [sha256.Size]byte) error { |
| 79 | ctx := context.Background() |
| 80 | now := time.Now().UTC() |
| 81 | baseRevision, _, err := sessionContentRevision(path) |
| 82 | ledgerUnreadable := err != nil |
| 83 | if ledgerUnreadable { |
| 84 | // A persisted session with an unreadable ledger fails closed, as in |
| 85 | // schema 1; a brand-new transcript still lands first and the record |
| 86 | // step below reports the sidecar. |
| 87 | if sessionArtifactsHaveContent(path) { |
| 88 | return err |
| 89 | } |
| 90 | baseRevision = 0 |
| 91 | } |
| 92 | var st *sessionDAGState |
| 93 | if route == dagRouteUpgrade { |
| 94 | st, err = s.upgradeLogForSave(ctx, path, now) |
| 95 | } else { |
| 96 | st, err = s.dagStateForSave(ctx, path, now) |
| 97 | } |
| 98 | if err != nil { |
| 99 | return err |
| 100 | } |
| 101 | s.ensureMessageIDsForSave(msgs) |
| 102 | plan, err := s.planDAGWrite(path, st, msgs, mode, now) |
| 103 | if err != nil { |
| 104 | return err |
| 105 | } |
| 106 | deferProjection := mode.defersProjection() |
| 107 | pending := s.takePendingMarkers() |
| 108 | for i := range pending { |
| 109 | pending[i].Head = plan.head |
| 110 | } |
| 111 | plan.entries = append(plan.entries, pending...) |
| 112 | if len(plan.entries) == 0 { |
| 113 | s.adoptDAGPosition(st, plan) |
| 114 | s.republishDAGDerivedIfPending(ctx, path, st, plan, msgs, digest, baseRevision) |
| 115 | s.maintainDAGLog(ctx, path, st, plan, mode, now) |
| 116 | s.markCheckpointPersisted(path, digest, version, baseRevision, rewriteVersion, msgs, deferProjection) |
| 117 | return nil |
| 118 | } |
| 119 | reserved, err := invalidateSessionListingProjection(path) |
| 120 | if err != nil { |
| 121 | if !ledgerUnreadable { |
| 122 | return fmt.Errorf("invalidate session listing projection: %w", err) |
| 123 | } |
| 124 | reserved = 0 |
| 125 | } |
| 126 | tail := st.lastGoodEnd |
| 127 | if _, err := appendSessionDAGEntries(path, plan.entries, true); err != nil { |
| 128 | s.requeuePendingMarkers(pending) |
| 129 | return err |
| 130 | } |
| 131 | if err := st.replayFrom(ctx, tail, defaultSessionReplayLimits); err != nil { |
| 132 | return err |
| 133 | } |
| 134 | if st.damaged { |
| 135 | return fmt.Errorf("session log %s: appended entries did not replay", path) |
| 136 | } |
| 137 | plan.applyIDRenames(s) |
| 138 | s.adoptDAGPosition(st, plan) |
| 139 | // The compatibility checkpoint lands before the ledger, as in schema 1, so |
| 140 | // a metadata failure never leaves the anchor behind the log. |
| 141 | selected := st.selectedHead() |
| 142 | displayCurrent := selected == plan.head && writeDAGCheckpointCache(path, plan, msgs, baseRevision) |
| 143 | revision, err := recordSessionContentRevision(path, digest, baseRevision, reserved) |
| 144 | if err != nil { |
| 145 | return err |
| 146 | } |
| 147 | s.publishDAGDerived(ctx, path, st, plan, msgs, digest, revision, selected, displayCurrent, deferProjection) |
| 148 | s.maintainDAGLog(ctx, path, st, plan, mode, now) |
| 149 | s.markCheckpointPersisted(path, digest, version, revision, rewriteVersion, msgs, deferProjection) |
| 150 | return nil |
| 151 | } |
| 152 | |
| 153 | // dagStateForSave returns the replayed graph, extending the cached state by |
| 154 | // the bytes appended since its last observed tail when the generation and |
| 155 | // size still allow it, and settles a torn tail before any append. |
| 156 | func (s *Session) dagStateForSave(ctx context.Context, path string, now time.Time) (*sessionDAGState, error) { |
| 157 | logPath := store.SessionEventLog(path) |
| 158 | s.mu.RLock() |
| 159 | cached := s.head.state |
| 160 | s.mu.RUnlock() |
| 161 | header, ok, err := readSessionDAGHeader(path) |
| 162 | if err != nil { |
| 163 | return nil, err |
| 164 | } |
| 165 | var st *sessionDAGState |
| 166 | if cached != nil && ok && cached.path == logPath && header.generation == cached.generation && !cached.damaged { |
| 167 | if info, err := os.Stat(logPath); err == nil && info.Size() >= cached.lastGoodEnd { |
| 168 | st = cached |
| 169 | if err := st.replayFrom(ctx, st.lastGoodEnd, defaultSessionReplayLimits); err != nil { |
| 170 | return nil, err |
| 171 | } |
| 172 | } |
| 173 | } |
| 174 | if st == nil { |
| 175 | st, err = replaySessionDAG(ctx, logPath, defaultSessionReplayLimits) |
| 176 | if err != nil { |
| 177 | return nil, err |
| 178 | } |
| 179 | } |
| 180 | if st.damaged { |
| 181 | if err := settleDAGTail(ctx, path, st, now); err != nil { |
| 182 | return nil, err |
| 183 | } |
| 184 | } |
| 185 | return st, nil |
| 186 | } |
| 187 | |
| 188 | // settleDAGTail waits out the quiet window a torn tail might still be |
| 189 | // finishing, then repairs it; appending after an unrepaired partial line |
| 190 | // would bury this writer's own entry inside it. |
| 191 | func settleDAGTail(ctx context.Context, path string, st *sessionDAGState, now time.Time) error { |
| 192 | logPath := store.SessionEventLog(path) |
| 193 | info, err := os.Stat(logPath) |
| 194 | if err != nil { |
| 195 | return err |
| 196 | } |
| 197 | if age := now.Sub(info.ModTime()); age < sessionDAGTailRepairMinAge { |
| 198 | time.Sleep(sessionDAGTailRepairMinAge - age) |
| 199 | st.damaged = false |
| 200 | if err := st.replayFrom(ctx, st.lastGoodEnd, defaultSessionReplayLimits); err != nil { |
| 201 | return err |
| 202 | } |
| 203 | if !st.damaged { |
| 204 | return nil |
| 205 | } |
| 206 | } |
| 207 | repaired, err := repairSessionDAGTail(path, st, time.Now().UTC()) |
| 208 | if err != nil { |
| 209 | return err |
| 210 | } |
| 211 | if !repaired { |
| 212 | return fmt.Errorf("session log %s has a torn tail that is still being written", path) |
| 213 | } |
| 214 | return nil |
| 215 | } |
| 216 | |
| 217 | // upgradeLogForSave replaces the schema-1 log (or bare checkpoint) with |
| 218 | // generation 1 of a schema-2 log built from the transcript on disk. The |
| 219 | // in-memory delta is then written by the ordinary diff, exactly like any |
| 220 | // other save. Callers hold the file lock and satisfied dagSaveRoute. |
| 221 | func (s *Session) upgradeLogForSave(ctx context.Context, path string, now time.Time) (*sessionDAGState, error) { |
| 222 | disk, err := loadSessionTranscript(ctx, path, defaultSessionReplayLimits, nil) |
| 223 | if err != nil && !os.IsNotExist(err) { |
| 224 | return nil, err |
| 225 | } |
| 226 | msgs := migrateLegacyProviderContent(NormalizeSession(disk.msgs)) |
| 227 | assignLegacyMessageIDs(path, msgs) |
| 228 | var inFlight *InFlightTurnMeta |
| 229 | if meta, ok, err := LoadBranchMeta(path); err == nil && ok { |
| 230 | inFlight = meta.InFlightTurn |
| 231 | } |
| 232 | if err := upgradeSessionLogToDAG(path, msgs, disk.times, inFlight, now); err != nil { |
| 233 | return nil, err |
| 234 | } |
| 235 | st, err := replaySessionDAG(ctx, store.SessionEventLog(path), defaultSessionReplayLimits) |
| 236 | if err != nil { |
| 237 | return nil, err |
| 238 | } |
| 239 | if len(msgs) > 0 { |
| 240 | slog.Info("session: upgraded event log to schema 2", "path", path, "messages", len(msgs)) |
| 241 | } |
| 242 | return st, nil |
| 243 | } |
| 244 | |
| 245 | // ensureMessageIDsForSave mints ids for messages that reached the session |
| 246 | // without one and writes them back by position so the next save sees the |
| 247 | // same ids the log now holds. |
| 248 | func (s *Session) ensureMessageIDsForSave(msgs []provider.Message) { |
| 249 | minted := false |
| 250 | for i := range msgs { |
| 251 | if msgs[i].ID == "" { |
| 252 | msgs[i].ID = NewMessageID() |
| 253 | minted = true |
| 254 | } |
| 255 | } |
| 256 | if !minted { |
| 257 | return |
| 258 | } |
| 259 | s.mu.Lock() |
| 260 | defer s.mu.Unlock() |
| 261 | for i := range msgs { |
| 262 | if i < len(s.Messages) && s.Messages[i].ID == "" && messagesEqualForStorage(s.Messages[i], msgs[i]) { |
| 263 | s.Messages[i].ID = msgs[i].ID |
| 264 | } |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | func (s *Session) adoptDAGPosition(st *sessionDAGState, plan *dagWritePlan) { |
| 269 | h := st.heads[plan.head] |
| 270 | leaf := "" |
| 271 | if h != nil { |
| 272 | leaf = h.leaf |
| 273 | } |
| 274 | s.mu.Lock() |
| 275 | defer s.mu.Unlock() |
| 276 | s.head.ref = HeadRef{HeadID: plan.head, LeafID: leaf, LogGeneration: st.generation, LogOffset: st.lastGoodEnd} |
| 277 | s.head.dag = true |
| 278 | s.head.state = st |
| 279 | s.head.headCount = len(st.heads) |
| 280 | if plan.forked { |
| 281 | s.head.events = append(s.head.events, HeadEvent{Kind: HeadEventForkedConcurrent, HeadID: plan.head, OtherWriter: plan.otherWriter}) |
| 282 | } |
| 283 | } |
| 284 | |
| 285 | // writeDAGCheckpointCache refreshes the .jsonl random-read model for the |
| 286 | // selected head, extending it in place for a pure append. It reports whether |
| 287 | // the cache now matches msgs; failures are logged, never fatal. |
| 288 | func writeDAGCheckpointCache(path string, plan *dagWritePlan, msgs []provider.Message, baseRevision int64) bool { |
| 289 | if plan.pureAppend { |
| 290 | current, err := appendSessionDisplayReadModel(path, msgs, plan.appendFrom, baseRevision) |
| 291 | if err != nil { |
| 292 | slog.Warn("session: keeping save after display read-model append failure", "path", path, "err", err) |
| 293 | } |
| 294 | if current { |
| 295 | return true |
| 296 | } |
| 297 | } |
| 298 | if err := writeSessionMessages(path, msgs); err != nil { |
| 299 | slog.Warn("session: keeping save after display read-model write failure", "path", path, "err", err) |
| 300 | return false |
| 301 | } |
| 302 | return true |
| 303 | } |
| 304 | |
| 305 | // publishDAGDerived refreshes the display index when the .jsonl cache is |
| 306 | // current for this head, and the head index plus meta mirror on every save. |
| 307 | func (s *Session) publishDAGDerived(ctx context.Context, path string, st *sessionDAGState, plan *dagWritePlan, msgs []provider.Message, digest [sha256.Size]byte, revision int64, selected string, displayCurrent, deferProjection bool) { |
| 308 | if displayCurrent { |
| 309 | appendFrom := -1 |
| 310 | if plan.pureAppend { |
| 311 | appendFrom = plan.appendFrom |
| 312 | } |
| 313 | if err := refreshCheckpointDisplayIndex(path, msgs, digest, revision, appendFrom, deferProjection); err != nil { |
| 314 | slog.Warn("session: keeping save after display index write failure", "path", path, "err", err) |
| 315 | } |
| 316 | } |
| 317 | if err := writeSessionDAGIndex(ctx, path, st); err != nil { |
| 318 | slog.Warn("session: keeping save after head index write failure", "path", path, "err", err) |
| 319 | } |
| 320 | if err := UpdateBranchMeta(path, false, func(meta *BranchMeta) error { |
| 321 | meta.HeadID = selected |
| 322 | meta.HeadCount = len(st.heads) |
| 323 | meta.LogSchema = sessionDAGSchemaVersion |
| 324 | meta.LogGeneration = st.generation |
| 325 | return nil |
| 326 | }); err != nil { |
| 327 | slog.Warn("session: head metadata update deferred", "path", path, "err", err) |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | // maintainDAGLog rotates the log when it has outgrown its live chains or a |
| 332 | // redaction needs its bytes physically erased, but only under the |
| 333 | // single-writer proof; otherwise the log simply keeps growing for now. |
| 334 | func (s *Session) maintainDAGLog(ctx context.Context, path string, st *sessionDAGState, plan *dagWritePlan, mode sessionSaveMode, now time.Time) { |
| 335 | if mode != sessionSaveRewriteCompact && st.holes == 0 && !sessionDAGLogOversized(st) { |
| 336 | return |
| 337 | } |
| 338 | if err := sessionDAGSingleWriterProof(path, st, now); err != nil { |
| 339 | slog.Info("session: log rotation deferred", "path", path, "reason", err) |
| 340 | return |
| 341 | } |
| 342 | if err := rotateSessionDAG(path, st, now); err != nil { |
| 343 | slog.Warn("session: log rotation failed", "path", path, "err", err) |
| 344 | return |
| 345 | } |
| 346 | fresh, err := replaySessionDAG(ctx, store.SessionEventLog(path), defaultSessionReplayLimits) |
| 347 | if err != nil { |
| 348 | slog.Warn("session: replay after rotation failed", "path", path, "err", err) |
| 349 | return |
| 350 | } |
| 351 | s.adoptDAGPosition(fresh, &dagWritePlan{head: plan.head}) |
| 352 | if err := writeSessionDAGIndex(ctx, path, fresh); err != nil { |
| 353 | slog.Warn("session: head index write after rotation failed", "path", path, "err", err) |
| 354 | } |
| 355 | } |
| 356 |