| 1 | package agent |
| 2 | |
| 3 | import "time" |
| 4 | |
| 5 | // SessionOpenTurn is a turn whose turn_begin marker has no matching turn_end |
| 6 | // on the selected head: the runtime that ran it stopped before it finished. |
| 7 | type SessionOpenTurn struct { |
| 8 | TurnID string |
| 9 | HeadID string |
| 10 | LeafID string |
| 11 | PreserveUser bool |
| 12 | StartedAt time.Time |
| 13 | } |
| 14 | |
| 15 | // QueueTurnBegin records the start of a foreground turn for a schema-2 |
| 16 | // session. The marker is appended with the next save, ahead of nothing it |
| 17 | // needs to precede: it names the leaf the turn started from, so recovery does |
| 18 | // not depend on entry order. It reports false for schema-1 sessions, which |
| 19 | // keep the in-flight meta marker. |
| 20 | func (s *Session) QueueTurnBegin(turnID string, preserveUser bool) bool { |
| 21 | if s == nil || turnID == "" { |
| 22 | return false |
| 23 | } |
| 24 | s.mu.Lock() |
| 25 | defer s.mu.Unlock() |
| 26 | if !s.head.dag { |
| 27 | return false |
| 28 | } |
| 29 | leaf := "" |
| 30 | if n := len(s.Messages); n > 0 { |
| 31 | leaf = s.Messages[n-1].ID |
| 32 | } |
| 33 | now := time.Now().UTC() |
| 34 | s.head.pending = append(s.head.pending, sessionDAGEntry{Type: sessionDAGTypeTurnBegin, Turn: turnID, Leaf: leaf, PreserveUser: preserveUser, At: now}) |
| 35 | s.head.openTurn = &sessionDAGTurn{turn: turnID, leaf: leaf, preserveUser: preserveUser, at: now} |
| 36 | s.version++ |
| 37 | return true |
| 38 | } |
| 39 | |
| 40 | // QueueTurnEnd closes a turn begun with QueueTurnBegin. It rides the next |
| 41 | // save, so the completed tail and its end marker land in one batch. |
| 42 | func (s *Session) QueueTurnEnd(turnID string) bool { |
| 43 | if s == nil || turnID == "" { |
| 44 | return false |
| 45 | } |
| 46 | s.mu.Lock() |
| 47 | defer s.mu.Unlock() |
| 48 | if !s.head.dag { |
| 49 | return false |
| 50 | } |
| 51 | s.head.pending = append(s.head.pending, sessionDAGEntry{Type: sessionDAGTypeTurnEnd, Turn: turnID, At: time.Now().UTC()}) |
| 52 | if s.head.openTurn != nil && s.head.openTurn.turn == turnID { |
| 53 | s.head.openTurn = nil |
| 54 | } |
| 55 | s.version++ |
| 56 | return true |
| 57 | } |
| 58 | |
| 59 | // OpenTurn reports the turn left open on this session's head, if any. |
| 60 | func (s *Session) OpenTurn() (SessionOpenTurn, bool) { |
| 61 | if s == nil { |
| 62 | return SessionOpenTurn{}, false |
| 63 | } |
| 64 | s.mu.RLock() |
| 65 | defer s.mu.RUnlock() |
| 66 | t := s.head.openTurn |
| 67 | if !s.head.dag || t == nil { |
| 68 | return SessionOpenTurn{}, false |
| 69 | } |
| 70 | return SessionOpenTurn{TurnID: t.turn, HeadID: s.head.ref.HeadID, LeafID: t.leaf, PreserveUser: t.preserveUser, StartedAt: t.at}, true |
| 71 | } |
| 72 | |
| 73 | // TurnContinuedOnOtherHead reports whether another head already carries a |
| 74 | // message that follows leafID: the "interrupted" turn moved there and kept |
| 75 | // running, so its tail on this head must not be treated as a crash remnant. |
| 76 | func (s *Session) TurnContinuedOnOtherHead(leafID string) bool { |
| 77 | if s == nil { |
| 78 | return false |
| 79 | } |
| 80 | s.mu.RLock() |
| 81 | defer s.mu.RUnlock() |
| 82 | st := s.head.state |
| 83 | if st == nil || !s.head.dag { |
| 84 | return false |
| 85 | } |
| 86 | mine := s.head.ref.HeadID |
| 87 | for _, n := range st.nodes { |
| 88 | if n.parent == leafID && n.head != mine && n.head != "" { |
| 89 | return true |
| 90 | } |
| 91 | } |
| 92 | return false |
| 93 | } |
| 94 | |
| 95 | // takePendingMarkers hands the queued turn markers to a save and clears the |
| 96 | // queue; the save stamps them with the head it writes to. |
| 97 | func (s *Session) takePendingMarkers() []sessionDAGEntry { |
| 98 | s.mu.Lock() |
| 99 | defer s.mu.Unlock() |
| 100 | pending := s.head.pending |
| 101 | s.head.pending = nil |
| 102 | return pending |
| 103 | } |
| 104 | |
| 105 | // requeuePendingMarkers puts markers back after a failed append so the next |
| 106 | // save carries them. |
| 107 | func (s *Session) requeuePendingMarkers(pending []sessionDAGEntry) { |
| 108 | if len(pending) == 0 { |
| 109 | return |
| 110 | } |
| 111 | s.mu.Lock() |
| 112 | defer s.mu.Unlock() |
| 113 | s.head.pending = append(pending, s.head.pending...) |
| 114 | } |
| 115 |