pull fetches turns appended since the last sync (by any channel except our own local writes) so the caller can splice them into history before the next turn. It advances the seen-sequence watermark. Safe to call on the turn's goroutine; returns nil when nothing is new or the hub is idle.
(ctx context.Context)
| 96 | // turn. It advances the seen-sequence watermark. Safe to call on the turn's |
| 97 | // goroutine; returns nil when nothing is new or the hub is idle. |
| 98 | func (hs *HubSync) pull(ctx context.Context) []models.Message { |
| 99 | hs.mu.Lock() |
| 100 | convID := hs.convID |
| 101 | since := hs.lastSeq |
| 102 | hs.mu.Unlock() |
| 103 | if convID == "" { |
| 104 | return nil |
| 105 | } |
| 106 | events, err := hs.client.ReadConversation(ctx, convID, since, 0) |
| 107 | if err != nil { |
| 108 | hs.logger.Warn("hub sync: pull failed", zap.Error(err)) |
| 109 | return nil |
| 110 | } |
| 111 | msgs := make([]models.Message, 0, len(events)) |
| 112 | hs.mu.Lock() |
| 113 | for _, ev := range events { |
| 114 | if ev.Seq > hs.lastSeq { |
| 115 | hs.lastSeq = ev.Seq |
| 116 | } |
| 117 | if ev.Channel == hubChannelLocal { |
| 118 | continue // our own turns are already in local history |
| 119 | } |
| 120 | msgs = append(msgs, ev.ToMessage()) |
| 121 | } |
| 122 | hs.mu.Unlock() |
| 123 | return msgs |
| 124 | } |
| 125 | |
| 126 | func (hs *HubSync) append(ctx context.Context, convID, role, content string) { |
| 127 | if content == "" { |