| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | |
| 11 | "reasonix/internal/event" |
| 12 | ) |
| 13 | |
| 14 | // ForkAvailability names why one source turn can or cannot start a child |
| 15 | // session. A surface shows the reason instead of collapsing every refusal into |
| 16 | // one "unavailable" message. |
| 17 | type ForkAvailability string |
| 18 | |
| 19 | const ( |
| 20 | // ForkAvailable means the turn closed at an atomic commit boundary. |
| 21 | ForkAvailable ForkAvailability = "available" |
| 22 | // ForkTurnOpen means the turn has no terminal turn/end record yet. |
| 23 | ForkTurnOpen ForkAvailability = "turn_open" |
| 24 | // ForkActiveAuthority means the cut would inherit in-flight execution state. |
| 25 | ForkActiveAuthority ForkAvailability = "active_authority" |
| 26 | // ForkHistoryUnverifiable means the source keeps no persisted turn records, |
| 27 | // so no boundary can be proven. Text matching, elapsed time, or a turn that |
| 28 | // merely looks finished must never substitute for one. |
| 29 | ForkHistoryUnverifiable ForkAvailability = "history_unverifiable" |
| 30 | // ForkStaleSource means the request no longer addresses the session boundary |
| 31 | // the caller displayed. |
| 32 | ForkStaleSource ForkAvailability = "stale_source" |
| 33 | // ForkUnsupported means this surface has no create-only session fork. |
| 34 | ForkUnsupported ForkAvailability = "unsupported" |
| 35 | ) |
| 36 | |
| 37 | // ForkTarget is one source turn a client may fork from. It is derived only from |
| 38 | // committed events, so the same source yields the same targets whether it is |
| 39 | // live in this process, owned by another process, or read cold from disk. |
| 40 | type ForkTarget struct { |
| 41 | TurnID string `json:"turnId"` |
| 42 | TurnNumber int `json:"turnNumber"` |
| 43 | StartSequence uint64 `json:"startSequence"` |
| 44 | EndSequence uint64 `json:"endSequence"` |
| 45 | // BoundarySequence is the complete atomic commit boundary observed by the |
| 46 | // caller. CreateFork requires the same value so a delayed request cannot be |
| 47 | // reinterpreted against another projection. |
| 48 | BoundarySequence uint64 `json:"boundarySequence"` |
| 49 | Status event.TurnStatus `json:"status"` |
| 50 | // MessageID is the stable transcript identity of this turn's final assistant |
| 51 | // reply, empty when the turn committed none. |
| 52 | MessageID string `json:"messageId,omitempty"` |
| 53 | Available bool `json:"available"` |
| 54 | Reason ForkAvailability `json:"reason,omitempty"` |
| 55 | } |
| 56 | |
| 57 | // ForkTargetSet is the fork state of one source session. Targets stay empty and |
| 58 | // Verifiable stays false for legacy history that keeps messages without turn |
| 59 | // records, which is what lets a surface say the boundary is unverifiable |
| 60 | // instead of offering a cut it cannot prove. |
| 61 | type ForkTargetSet struct { |
| 62 | Source SessionRef `json:"source"` |
| 63 | Targets []ForkTarget `json:"targets"` |
| 64 | Verifiable bool `json:"verifiable"` |
| 65 | } |
| 66 | |
| 67 | func forkProjectionAvailability(projection Projection, boundary uint64) ForkAvailability { |
| 68 | if boundary == 0 || boundary != projection.CommittedSequence { |
| 69 | return ForkHistoryUnverifiable |
| 70 | } |
| 71 | if projection.TurnID != "" || len(projection.Interactions) != 0 || len(projection.ActiveTools) != 0 { |
| 72 | return ForkActiveAuthority |
| 73 | } |
| 74 | return ForkAvailable |
| 75 | } |
| 76 | |
| 77 | // ForkTargets lists the source's turns in display order. The open turn is |
| 78 | // included as ForkTurnOpen so a surface can explain why its own turn is not |
| 79 | // forkable yet without disabling the turns that already finished. |
| 80 | func ForkTargets(projection Projection) ForkTargetSet { |
| 81 | targets := make([]ForkTarget, 0, len(projection.Turns)+1) |
| 82 | for _, turn := range projection.Turns { |
| 83 | if projection.HiddenTurns[turn.TurnID] { |
| 84 | continue |
| 85 | } |
| 86 | target := ForkTarget{ |
| 87 | TurnID: turn.TurnID, TurnNumber: len(targets) + 1, |
| 88 | StartSequence: turn.StartSequence, EndSequence: turn.EndSequence, |
| 89 | BoundarySequence: turn.BoundarySequence, |
| 90 | Status: turn.Status, MessageID: turn.MessageID, |
| 91 | Available: turn.Availability == ForkAvailable, |
| 92 | } |
| 93 | if !target.Available && turn.Availability != "" { |
| 94 | target.Reason = turn.Availability |
| 95 | } else if !target.Available { |
| 96 | target.Reason = ForkHistoryUnverifiable |
| 97 | } |
| 98 | targets = append(targets, target) |
| 99 | } |
| 100 | if projection.TurnID != "" { |
| 101 | targets = append(targets, ForkTarget{ |
| 102 | TurnID: projection.TurnID, TurnNumber: len(targets) + 1, |
| 103 | StartSequence: projection.CurrentTurnStart, Status: projection.TurnStatus, |
| 104 | MessageID: projection.CurrentTurnMessageID, Reason: ForkTurnOpen, |
| 105 | }) |
| 106 | } |
| 107 | return ForkTargetSet{Targets: targets, Verifiable: len(projection.Turns) > 0 || projection.TurnID != ""} |
| 108 | } |
| 109 | |
| 110 | // ForkSequence resolves the cut for one turn identity. Only a turn that closed |
| 111 | // at an atomic commit boundary resolves; an open turn and an unknown identity |
| 112 | // are refused rather than silently redirected to the newest turn. |
| 113 | func ForkSequence(projection Projection, turnID string) (uint64, ForkAvailability, error) { |
| 114 | turnID = strings.TrimSpace(turnID) |
| 115 | if turnID == "" { |
| 116 | return 0, "", fmt.Errorf("session: fork needs a turn id") |
| 117 | } |
| 118 | if projection.HiddenTurns[turnID] { |
| 119 | return 0, ForkHistoryUnverifiable, nil |
| 120 | } |
| 121 | if projection.TurnID == turnID { |
| 122 | return 0, ForkTurnOpen, nil |
| 123 | } |
| 124 | for _, turn := range projection.Turns { |
| 125 | if turn.TurnID != turnID { |
| 126 | continue |
| 127 | } |
| 128 | if turn.Availability != ForkAvailable { |
| 129 | reason := turn.Availability |
| 130 | if reason == "" { |
| 131 | reason = ForkHistoryUnverifiable |
| 132 | } |
| 133 | return 0, reason, nil |
| 134 | } |
| 135 | return turn.BoundarySequence, ForkAvailable, nil |
| 136 | } |
| 137 | return 0, ForkHistoryUnverifiable, nil |
| 138 | } |
| 139 | |
| 140 | // ForkSequenceForNumber resolves the cut for a display turn number. It exists |
| 141 | // only for clients that still address turns by their 1-based position; new |
| 142 | // clients carry the stable turn identity instead. |
| 143 | func ForkSequenceForNumber(projection Projection, turn int) (uint64, ForkAvailability, error) { |
| 144 | set := ForkTargets(projection) |
| 145 | if turn < 1 || turn > len(set.Targets) { |
| 146 | return 0, ForkHistoryUnverifiable, fmt.Errorf("session: turn %d is not a forkable turn", turn) |
| 147 | } |
| 148 | return ForkSequence(projection, set.Targets[turn-1].TurnID) |
| 149 | } |
| 150 | |
| 151 | // ForkTargetSetFor reads the fork state of one session without requiring a live |
| 152 | // runtime. A session owned by another process is read from its durable commits |
| 153 | // and keeps its lease untouched. |
| 154 | func (s *Service) ForkTargetSetFor(ctx context.Context, ref SessionRef) (ForkTargetSet, error) { |
| 155 | if s == nil { |
| 156 | return ForkTargetSet{}, fmt.Errorf("session: nil service") |
| 157 | } |
| 158 | if err := ref.validate(s.hostID); err != nil { |
| 159 | return ForkTargetSet{}, err |
| 160 | } |
| 161 | projection, err := s.forkTurnProjection(ctx, ref) |
| 162 | if err != nil { |
| 163 | return ForkTargetSet{}, err |
| 164 | } |
| 165 | set := ForkTargets(projection) |
| 166 | set.Source = ref |
| 167 | return set, nil |
| 168 | } |
| 169 | |
| 170 | // forkTurnProjection reads only the projection a fork resolves its cut from. A |
| 171 | // live runtime already holds that projection in memory, so it is read without |
| 172 | // reconstructing the durable transcript: surfaces refresh their fork state |
| 173 | // after every turn, next to the running turn. A session with no runtime in this |
| 174 | // process is read cold from its durable commits, which keeps the lease of the |
| 175 | // process that owns it untouched. |
| 176 | func (s *Service) forkTurnProjection(ctx context.Context, ref SessionRef) (Projection, error) { |
| 177 | if runtime, ok := s.Runtime(ref); ok { |
| 178 | return runtime.Session().ExecutionSnapshot().Projection, nil |
| 179 | } |
| 180 | snapshot, err := s.query.Snapshot(ctx, ref) |
| 181 | if err != nil { |
| 182 | return Projection{}, err |
| 183 | } |
| 184 | return snapshot.Projection, nil |
| 185 | } |
| 186 | |
| 187 | // ForkUnavailableError reports a refused cut together with the reason a surface |
| 188 | // shows. Every refusal keeps its own reason so "still running", "read-only" and |
| 189 | // "no boundary" never collapse into one message. |
| 190 | type ForkUnavailableError struct { |
| 191 | TurnID string |
| 192 | Reason ForkAvailability |
| 193 | } |
| 194 | |
| 195 | func (e *ForkUnavailableError) Error() string { |
| 196 | return fmt.Sprintf("session: turn %q cannot start a fork (%s)", e.TurnID, e.Reason) |
| 197 | } |
| 198 | |
| 199 | // ForkRequest identifies one create-a-child-session request. The host resolves |
| 200 | // the cut from persisted turn records and requires BoundarySequence to match |
| 201 | // the atomic boundary the client observed. Array positions and checkpoint |
| 202 | // numbers are never accepted as authority. |
| 203 | type ForkRequest struct { |
| 204 | Source SessionRef |
| 205 | // TurnID is the stable identity of the completed turn to cut after. |
| 206 | TurnID string |
| 207 | // BoundarySequence is the exact atomic boundary the client observed for the |
| 208 | // turn. The host re-resolves it and refuses a stale or reinterpreted anchor. |
| 209 | BoundarySequence uint64 |
| 210 | // ChildID is optional; the host mints one when empty. |
| 211 | ChildID string |
| 212 | // OperationID identifies this creation request. A retried submission with |
| 213 | // the same operation id addresses the same child instead of minting a |
| 214 | // second fork of one turn. |
| 215 | OperationID string |
| 216 | } |
| 217 | |
| 218 | // ForkResult reports the created child and the turn it was cut at. |
| 219 | type ForkResult struct { |
| 220 | Child SessionRef |
| 221 | Turn ForkTarget |
| 222 | } |
| 223 | |
| 224 | // CreateFork publishes an independent child session from one completed turn of |
| 225 | // the source. It never switches, closes, or writes to the source: a running |
| 226 | // parent keeps running, and a parent owned by another process keeps its lease. |
| 227 | func (s *Service) CreateFork(ctx context.Context, request ForkRequest) (ForkResult, error) { |
| 228 | if s == nil { |
| 229 | return ForkResult{}, fmt.Errorf("session: nil service") |
| 230 | } |
| 231 | if err := request.Source.validate(s.hostID); err != nil { |
| 232 | return ForkResult{}, err |
| 233 | } |
| 234 | filesystem, ok := s.persistence.(*FilesystemPersistence) |
| 235 | if !ok { |
| 236 | return ForkResult{}, errors.New("session: persistence does not support filesystem fork") |
| 237 | } |
| 238 | projection, err := s.forkTurnProjection(ctx, request.Source) |
| 239 | if err != nil { |
| 240 | return ForkResult{}, err |
| 241 | } |
| 242 | sequence, availability, err := ForkSequence(projection, request.TurnID) |
| 243 | if err != nil { |
| 244 | return ForkResult{}, err |
| 245 | } |
| 246 | if availability != ForkAvailable { |
| 247 | return ForkResult{}, &ForkUnavailableError{TurnID: request.TurnID, Reason: availability} |
| 248 | } |
| 249 | if request.BoundarySequence == 0 || request.BoundarySequence != sequence { |
| 250 | return ForkResult{}, &ForkUnavailableError{TurnID: request.TurnID, Reason: ForkStaleSource} |
| 251 | } |
| 252 | target, ok := forkTargetByID(projection, request.TurnID) |
| 253 | if !ok { |
| 254 | return ForkResult{}, &ForkUnavailableError{TurnID: request.TurnID, Reason: ForkHistoryUnverifiable} |
| 255 | } |
| 256 | childID := strings.TrimSpace(request.ChildID) |
| 257 | if childID == "" { |
| 258 | if operation := strings.TrimSpace(request.OperationID); operation != "" { |
| 259 | childID = deterministicID("fork\x00" + request.Source.SessionID + "\x00" + request.TurnID + "\x00" + operation) |
| 260 | } else { |
| 261 | childID = randomID() |
| 262 | } |
| 263 | } |
| 264 | if err := validateSessionID(childID); err != nil { |
| 265 | return ForkResult{}, err |
| 266 | } |
| 267 | parent, closeParent, err := s.forkSource(ctx, request.Source) |
| 268 | if err != nil { |
| 269 | return ForkResult{}, err |
| 270 | } |
| 271 | defer closeParent() |
| 272 | childRef := SessionRef{HostID: s.hostID, SessionID: childID} |
| 273 | childDir := filepath.Join(filesystem.Root, childID) |
| 274 | parentDir := parent.dir() |
| 275 | if matched, readErr := forkChildMatches(childDir, parentDir, sequence); readErr == nil { |
| 276 | // A retried request reuses the child it already published. A different |
| 277 | // session that merely holds this identity is a real conflict. |
| 278 | if matched { |
| 279 | return ForkResult{Child: childRef, Turn: target}, nil |
| 280 | } |
| 281 | return ForkResult{}, fmt.Errorf("session: child session %q already exists", childID) |
| 282 | } else if !os.IsNotExist(readErr) { |
| 283 | return ForkResult{}, readErr |
| 284 | } |
| 285 | if _, err := parent.Fork(ctx, childDir, childID, sequence); err != nil { |
| 286 | // The check above is no reservation: two callers with one operation id |
| 287 | // both reach the publish, and the child the winner published is this |
| 288 | // request's own result, so the race resolves as the idempotent success. |
| 289 | if matched, readErr := forkChildMatches(childDir, parentDir, sequence); readErr == nil && matched { |
| 290 | return ForkResult{Child: childRef, Turn: target}, nil |
| 291 | } |
| 292 | // Both refusals mean the boundary this target advertised cannot carry a |
| 293 | // safe child. Each keeps its own reason so the surface says which one. |
| 294 | switch { |
| 295 | case errors.Is(err, ErrForkActiveAuthority): |
| 296 | return ForkResult{}, &ForkUnavailableError{TurnID: request.TurnID, Reason: ForkActiveAuthority} |
| 297 | case errors.Is(err, ErrForkBoundaryNotAtomic): |
| 298 | return ForkResult{}, &ForkUnavailableError{TurnID: request.TurnID, Reason: ForkHistoryUnverifiable} |
| 299 | default: |
| 300 | return ForkResult{}, err |
| 301 | } |
| 302 | } |
| 303 | return ForkResult{Child: childRef, Turn: target}, nil |
| 304 | } |
| 305 | |
| 306 | // forkChildMatches reports whether childDir already holds the child this request |
| 307 | // asked for: the same inherited cut, from the same parent directory. The read |
| 308 | // error is passed through so a caller can tell "no child yet" from a store it |
| 309 | // cannot read. |
| 310 | func forkChildMatches(childDir, parentDir string, sequence uint64) (bool, error) { |
| 311 | manifest, err := readStoredManifest(filepath.Join(childDir, "manifest.json")) |
| 312 | if err != nil { |
| 313 | return false, err |
| 314 | } |
| 315 | return manifest.InheritedEvents == sequence && manifest.Source != nil && |
| 316 | filepath.Clean(manifest.Source.Path) == filepath.Clean(parentDir), nil |
| 317 | } |
| 318 | |
| 319 | // forkSource returns the session to read the durable prefix from, plus its |
| 320 | // release. A live source is used as-is so its accepted tail is flushed first; a |
| 321 | // source with no runtime in this process is opened read-only, which neither |
| 322 | // restores the parent agent nor disturbs another process's lease. |
| 323 | func (s *Service) forkSource(ctx context.Context, ref SessionRef) (*Session, func(), error) { |
| 324 | if runtime, ok := s.Runtime(ref); ok { |
| 325 | return runtime.session, func() {}, nil |
| 326 | } |
| 327 | parent, err := s.persistence.Open(ref.SessionID, ReadOnly) |
| 328 | if err != nil { |
| 329 | return nil, nil, err |
| 330 | } |
| 331 | return parent, func() { _ = parent.Close(context.WithoutCancel(ctx)) }, nil |
| 332 | } |
| 333 | |
| 334 | func forkTargetByID(projection Projection, turnID string) (ForkTarget, bool) { |
| 335 | turnID = strings.TrimSpace(turnID) |
| 336 | for _, target := range ForkTargets(projection).Targets { |
| 337 | if target.TurnID == turnID { |
| 338 | return target, true |
| 339 | } |
| 340 | } |
| 341 | return ForkTarget{}, false |
| 342 | } |
| 343 |