Merge branch 'main' into fix/editproperly
1 file · +61 -6
M packages/agent/modes/telegram/bridge.go +61 -6
| @@ -44,12 +44,13 @@ type Bridge struct { | |||
| 44 | 44 | Save func(Config) error | |
| 45 | 45 | Host Host | |
| 46 | 46 | ||
| 47 | - | mu sync.Mutex | |
| 48 | - | running bool | |
| 49 | - | cancel context.CancelFunc | |
| 50 | - | me *User | |
| 51 | - | chatID int64 // populated after first DM from the paired user | |
| 52 | - | replyBuf strings.Builder | |
| 47 | + | mu sync.Mutex | |
| 48 | + | running bool | |
| 49 | + | cancel context.CancelFunc | |
| 50 | + | me *User | |
| 51 | + | chatID int64 // populated after first DM from the paired user | |
| 52 | + | replyBuf strings.Builder | |
| 53 | + | workingCancel context.CancelFunc | |
| 53 | 54 | ||
| 54 | 55 | // nextReplyFromTelegram is set when the next assistant reply | |
| 55 | 56 | // should be sent bare (no "zot: " prefix) because the turn was | |
| @@ -140,9 +141,14 @@ func (b *Bridge) Start(parent context.Context) error { | |||
| 140 | 141 | func (b *Bridge) Stop() { | |
| 141 | 142 | b.mu.Lock() | |
| 142 | 143 | cancel := b.cancel | |
| 144 | + | workingCancel := b.workingCancel | |
| 143 | 145 | b.running = false | |
| 144 | 146 | b.cancel = nil | |
| 147 | + | b.workingCancel = nil | |
| 145 | 148 | b.mu.Unlock() | |
| 149 | + | if workingCancel != nil { | |
| 150 | + | workingCancel() | |
| 151 | + | } | |
| 146 | 152 | if cancel != nil { | |
| 147 | 153 | cancel() | |
| 148 | 154 | } | |
| @@ -160,11 +166,16 @@ func (b *Bridge) OnAssistantText(text string) { | |||
| 160 | 166 | b.mu.Lock() | |
| 161 | 167 | // prefix := "zot: " | |
| 162 | 168 | prefix := "" | |
| 169 | + | workingCancel := b.workingCancel | |
| 170 | + | b.workingCancel = nil | |
| 163 | 171 | if b.nextReplyFromTelegram { | |
| 164 | 172 | prefix = "" | |
| 165 | 173 | b.nextReplyFromTelegram = false | |
| 166 | 174 | } | |
| 167 | 175 | b.mu.Unlock() | |
| 176 | + | if workingCancel != nil { | |
| 177 | + | workingCancel() | |
| 178 | + | } | |
| 168 | 179 | b.sendToPaired(text, prefix) | |
| 169 | 180 | } | |
| 170 | 181 | ||
| @@ -178,6 +189,47 @@ func (b *Bridge) OnUserTyped(text string) { | |||
| 178 | 189 | b.sendToPaired(text, "you: ") | |
| 179 | 190 | } | |
| 180 | 191 | ||
| 192 | + | // indicateWorking keeps Telegram's typing indicator visible while a turn | |
| 193 | + | // submitted from Telegram is waiting or running. Telegram clears chat actions | |
| 194 | + | // after a few seconds, so refresh it until the reply is ready or the bridge | |
| 195 | + | // stops. | |
| 196 | + | func (b *Bridge) indicateWorking(ctx context.Context, chatID int64) { | |
| 197 | + | workingCtx, cancel := context.WithCancel(ctx) | |
| 198 | + | ||
| 199 | + | b.mu.Lock() | |
| 200 | + | previous := b.workingCancel | |
| 201 | + | b.workingCancel = cancel | |
| 202 | + | b.mu.Unlock() | |
| 203 | + | if previous != nil { | |
| 204 | + | previous() | |
| 205 | + | } | |
| 206 | + | ||
| 207 | + | go func() { | |
| 208 | + | ticker := time.NewTicker(4 * time.Second) | |
| 209 | + | defer ticker.Stop() | |
| 210 | + | for { | |
| 211 | + | _ = b.Client.SendChatAction(workingCtx, chatID, "typing") | |
| 212 | + | select { | |
| 213 | + | case <-workingCtx.Done(): | |
| 214 | + | return | |
| 215 | + | case <-ticker.C: | |
| 216 | + | } | |
| 217 | + | } | |
| 218 | + | }() | |
| 219 | + | } | |
| 220 | + | ||
| 221 | + | // stopWorking clears the local typing-indicator refresh loop. Telegram removes | |
| 222 | + | // the visible indicator itself when the next message arrives or it expires. | |
| 223 | + | func (b *Bridge) stopWorking() { | |
| 224 | + | b.mu.Lock() | |
| 225 | + | cancel := b.workingCancel | |
| 226 | + | b.workingCancel = nil | |
| 227 | + | b.mu.Unlock() | |
| 228 | + | if cancel != nil { | |
| 229 | + | cancel() | |
| 230 | + | } | |
| 231 | + | } | |
| 232 | + | ||
| 181 | 233 | // sendToPaired writes text (with an optional prefix, chunked to | |
| 182 | 234 | // Telegram's 4096-char cap) to the paired chat. No-op when the | |
| 183 | 235 | // bridge is stopped or before the paired chat id is known. | |
| @@ -342,12 +394,14 @@ func (b *Bridge) handleUpdate(ctx context.Context, u Update) { | |||
| 342 | 394 | return | |
| 343 | 395 | case "/stop": | |
| 344 | 396 | b.Host.CancelTurn() | |
| 397 | + | b.stopWorking() | |
| 345 | 398 | _ = b.Client.SendMessage(ctx, msg.Chat.ID, | |
| 346 | 399 | "cancelled the current turn.", msg.MessageID) | |
| 347 | 400 | return | |
| 348 | 401 | } | |
| 349 | 402 | if isStopCommand(text) { | |
| 350 | 403 | b.Host.CancelTurn() | |
| 404 | + | b.stopWorking() | |
| 351 | 405 | _ = b.Client.SendMessage(ctx, msg.Chat.ID, | |
| 352 | 406 | "cancelled the current turn.", msg.MessageID) | |
| 353 | 407 | return | |
| @@ -382,6 +436,7 @@ func (b *Bridge) handleUpdate(ctx context.Context, u Update) { | |||
| 382 | 436 | b.mu.Lock() | |
| 383 | 437 | b.nextReplyFromTelegram = true | |
| 384 | 438 | b.mu.Unlock() | |
| 439 | + | b.indicateWorking(ctx, msg.Chat.ID) | |
| 385 | 440 | b.Host.SubmitOrQueue(prompt, images) | |
| 386 | 441 | } | |
| 387 | 442 | ||
| @@ -44,12 +44,13 @@ type Bridge struct { | |||||
| 44 | Save func(Config) error | 44 | Save func(Config) error | ||
| 45 | Host Host | 45 | Host Host | ||
| 46 | 46 | ||||
| 47 | - | mu sync.Mutex | 47 | + | mu sync.Mutex |
| 48 | - | running bool | 48 | + | running bool |
| 49 | - | cancel context.CancelFunc | 49 | + | cancel context.CancelFunc |
| 50 | - | me *User | 50 | + | me *User |
| 51 | - | chatID int64 // populated after first DM from the paired user | 51 | + | chatID int64 // populated after first DM from the paired user |
| 52 | - | replyBuf strings.Builder | 52 | + | replyBuf strings.Builder |
| 53 | + | workingCancel context.CancelFunc | |||
| 53 | 54 | ||||
| 54 | // nextReplyFromTelegram is set when the next assistant reply | 55 | // nextReplyFromTelegram is set when the next assistant reply | ||
| 55 | // should be sent bare (no "zot: " prefix) because the turn was | 56 | // should be sent bare (no "zot: " prefix) because the turn was | ||
| @@ -140,9 +141,14 @@ func (b *Bridge) Start(parent context.Context) error { | |||||
| 140 | func (b *Bridge) Stop() { | 141 | func (b *Bridge) Stop() { | ||
| 141 | b.mu.Lock() | 142 | b.mu.Lock() | ||
| 142 | cancel := b.cancel | 143 | cancel := b.cancel | ||
| 144 | + | workingCancel := b.workingCancel | |||
| 143 | b.running = false | 145 | b.running = false | ||
| 144 | b.cancel = nil | 146 | b.cancel = nil | ||
| 147 | + | b.workingCancel = nil | |||
| 145 | b.mu.Unlock() | 148 | b.mu.Unlock() | ||
| 149 | + | if workingCancel != nil { | |||
| 150 | + | workingCancel() | |||
| 151 | + | } | |||
| 146 | if cancel != nil { | 152 | if cancel != nil { | ||
| 147 | cancel() | 153 | cancel() | ||
| 148 | } | 154 | } | ||
| @@ -160,11 +166,16 @@ func (b *Bridge) OnAssistantText(text string) { | |||||
| 160 | b.mu.Lock() | 166 | b.mu.Lock() | ||
| 161 | // prefix := "zot: " | 167 | // prefix := "zot: " | ||
| 162 | prefix := "" | 168 | prefix := "" | ||
| 169 | + | workingCancel := b.workingCancel | |||
| 170 | + | b.workingCancel = nil | |||
| 163 | if b.nextReplyFromTelegram { | 171 | if b.nextReplyFromTelegram { | ||
| 164 | prefix = "" | 172 | prefix = "" | ||
| 165 | b.nextReplyFromTelegram = false | 173 | b.nextReplyFromTelegram = false | ||
| 166 | } | 174 | } | ||
| 167 | b.mu.Unlock() | 175 | b.mu.Unlock() | ||
| 176 | + | if workingCancel != nil { | |||
| 177 | + | workingCancel() | |||
| 178 | + | } | |||
| 168 | b.sendToPaired(text, prefix) | 179 | b.sendToPaired(text, prefix) | ||
| 169 | } | 180 | } | ||
| 170 | 181 | ||||
| @@ -178,6 +189,47 @@ func (b *Bridge) OnUserTyped(text string) { | |||||
| 178 | b.sendToPaired(text, "you: ") | 189 | b.sendToPaired(text, "you: ") | ||
| 179 | } | 190 | } | ||
| 180 | 191 | ||||
| 192 | + | // indicateWorking keeps Telegram's typing indicator visible while a turn | |||
| 193 | + | // submitted from Telegram is waiting or running. Telegram clears chat actions | |||
| 194 | + | // after a few seconds, so refresh it until the reply is ready or the bridge | |||
| 195 | + | // stops. | |||
| 196 | + | func (b *Bridge) indicateWorking(ctx context.Context, chatID int64) { | |||
| 197 | + | workingCtx, cancel := context.WithCancel(ctx) | |||
| 198 | + | ||||
| 199 | + | b.mu.Lock() | |||
| 200 | + | previous := b.workingCancel | |||
| 201 | + | b.workingCancel = cancel | |||
| 202 | + | b.mu.Unlock() | |||
| 203 | + | if previous != nil { | |||
| 204 | + | previous() | |||
| 205 | + | } | |||
| 206 | + | ||||
| 207 | + | go func() { | |||
| 208 | + | ticker := time.NewTicker(4 * time.Second) | |||
| 209 | + | defer ticker.Stop() | |||
| 210 | + | for { | |||
| 211 | + | _ = b.Client.SendChatAction(workingCtx, chatID, "typing") | |||
| 212 | + | select { | |||
| 213 | + | case <-workingCtx.Done(): | |||
| 214 | + | return | |||
| 215 | + | case <-ticker.C: | |||
| 216 | + | } | |||
| 217 | + | } | |||
| 218 | + | }() | |||
| 219 | + | } | |||
| 220 | + | ||||
| 221 | + | // stopWorking clears the local typing-indicator refresh loop. Telegram removes | |||
| 222 | + | // the visible indicator itself when the next message arrives or it expires. | |||
| 223 | + | func (b *Bridge) stopWorking() { | |||
| 224 | + | b.mu.Lock() | |||
| 225 | + | cancel := b.workingCancel | |||
| 226 | + | b.workingCancel = nil | |||
| 227 | + | b.mu.Unlock() | |||
| 228 | + | if cancel != nil { | |||
| 229 | + | cancel() | |||
| 230 | + | } | |||
| 231 | + | } | |||
| 232 | + | ||||
| 181 | // sendToPaired writes text (with an optional prefix, chunked to | 233 | // sendToPaired writes text (with an optional prefix, chunked to | ||
| 182 | // Telegram's 4096-char cap) to the paired chat. No-op when the | 234 | // Telegram's 4096-char cap) to the paired chat. No-op when the | ||
| 183 | // bridge is stopped or before the paired chat id is known. | 235 | // bridge is stopped or before the paired chat id is known. | ||
| @@ -342,12 +394,14 @@ func (b *Bridge) handleUpdate(ctx context.Context, u Update) { | |||||
| 342 | return | 394 | return | ||
| 343 | case "/stop": | 395 | case "/stop": | ||
| 344 | b.Host.CancelTurn() | 396 | b.Host.CancelTurn() | ||
| 397 | + | b.stopWorking() | |||
| 345 | _ = b.Client.SendMessage(ctx, msg.Chat.ID, | 398 | _ = b.Client.SendMessage(ctx, msg.Chat.ID, | ||
| 346 | "cancelled the current turn.", msg.MessageID) | 399 | "cancelled the current turn.", msg.MessageID) | ||
| 347 | return | 400 | return | ||
| 348 | } | 401 | } | ||
| 349 | if isStopCommand(text) { | 402 | if isStopCommand(text) { | ||
| 350 | b.Host.CancelTurn() | 403 | b.Host.CancelTurn() | ||
| 404 | + | b.stopWorking() | |||
| 351 | _ = b.Client.SendMessage(ctx, msg.Chat.ID, | 405 | _ = b.Client.SendMessage(ctx, msg.Chat.ID, | ||
| 352 | "cancelled the current turn.", msg.MessageID) | 406 | "cancelled the current turn.", msg.MessageID) | ||
| 353 | return | 407 | return | ||
| @@ -382,6 +436,7 @@ func (b *Bridge) handleUpdate(ctx context.Context, u Update) { | |||||
| 382 | b.mu.Lock() | 436 | b.mu.Lock() | ||
| 383 | b.nextReplyFromTelegram = true | 437 | b.nextReplyFromTelegram = true | ||
| 384 | b.mu.Unlock() | 438 | b.mu.Unlock() | ||
| 439 | + | b.indicateWorking(ctx, msg.Chat.ID) | |||
| 385 | b.Host.SubmitOrQueue(prompt, images) | 440 | b.Host.SubmitOrQueue(prompt, images) | ||
| 386 | } | 441 | } | ||
| 387 | 442 | ||||