interactive_resident_subagents.go RawHistoryFolder
1package modes
2
3import (
4 "context"
5 "fmt"
6 "os"
7 "path/filepath"
8 "strings"
9 "time"
10
11 "github.com/bnema/zut/packages/agent/internal/orchestration"
12 "github.com/bnema/zut/packages/agent/modes/telegram"
13 "github.com/bnema/zut/packages/agent/subagents"
14 "github.com/bnema/zut/packages/agent/tools"
15 "github.com/bnema/zut/packages/core"
16 "github.com/bnema/zut/packages/provider"
17 "github.com/bnema/zut/packages/tui"
18)
19
20func (i *Interactive) TrackResidentSubagent(childID, turnID string) {
21 if i == nil {
22 return
23 }
24 if !i.ensureCompletionTracker().TrackResident(childID, turnID) {
25 return
26 }
27 i.registerCoordinatorWorker(childID)
28 i.invalidate()
29 i.requestCompletionDelivery()
30}
31func (i *Interactive) ReportResidentSubagent(completion subagents.ResidentCompletion) {
32 if i == nil {
33 return
34 }
35 if !i.ensureCompletionTracker().Report(completion.Completion()) {
36 return
37 }
38 i.reloadOpenResidentChildSession(completion.ChildID)
39 i.invalidate()
40 i.requestCompletionDelivery()
41}
42func (i *Interactive) reloadOpenResidentChildSession(childID string) {
43 if i == nil {
44 return
45 }
46 i.mu.Lock()
47 session := i.residentChildSession
48 i.mu.Unlock()
49 if session == nil || session.childID != childID || !session.RequestRecentReload() {
50 return
51 }
52 i.reloadResidentChildSession(session)
53}
54func (i *Interactive) reloadResidentChildSession(session *residentChildSession) {
55 if i == nil || session == nil {
56 return
57 }
58 go func() {
59 err := session.ReloadAll(200, i.invalidate)
60 if session.FinishLoad(err) {
61 if session.RequestRecentReload() {
62 i.reloadResidentChildSession(session)
63 }
64 }
65 i.invalidate()
66 }()
67}
68func (i *Interactive) ensureCompletionTracker() *subagents.CompletionTracker {
69 // Tests and lightweight embedders may construct Interactive with a struct
70 // literal instead of NewInteractive, so retain this lazy fallback.
71 i.completionDeliveryMu.Lock()
72 defer i.completionDeliveryMu.Unlock()
73 if i.completionTracker == nil {
74 i.completionTracker = subagents.NewCompletionTracker()
75 }
76 return i.completionTracker
77}
78func (i *Interactive) ensureCoordinatorLocked() *orchestration.Coordinator {
79 if i.turnCoordinator == nil {
80 i.turnCoordinator = orchestration.New()
81 }
82 return i.turnCoordinator
83}
84func (i *Interactive) coordinatorAcceptsUserInput() bool {
85 if i == nil {
86 return false
87 }
88 i.completionDeliveryMu.Lock()
89 accepts := i.ensureCoordinatorLocked().AcceptsUserInput()
90 i.completionDeliveryMu.Unlock()
91 return accepts
92}
93func (i *Interactive) coordinatorHasPendingWorkers() bool {
94 if i == nil {
95 return false
96 }
97 i.completionDeliveryMu.Lock()
98 pending := i.ensureCoordinatorLocked().HasPendingWorkers()
99 i.completionDeliveryMu.Unlock()
100 return pending
101}
102func (i *Interactive) applyCoordinator(event orchestration.Event) []orchestration.Action {
103 if i == nil {
104 return nil
105 }
106 i.completionDeliveryMu.Lock()
107 result := i.ensureCoordinatorLocked().Apply(event)
108 i.completionDeliveryMu.Unlock()
109 return result.Actions
110}
111func (i *Interactive) registerCoordinatorWorker(workerID string) {
112 if i == nil || workerID == "" {
113 return
114 }
115 i.completionDeliveryMu.Lock()
116 coordinator := i.ensureCoordinatorLocked()
117 implicitWave := i.completionDeliveryHolds == 0
118 if implicitWave {
119 // A one-worker sealed wave replaces whatever wave an older release may
120 // still be holding a sampled snapshot for. Bump the generation so that
121 // release cannot seal this newer wave when it relocks.
122 i.completionDeliveryGeneration++
123 coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
124 }
125 i.coordinatorWorkerSeq++
126 registrationID := fmt.Sprintf("%s#%d", workerID, i.coordinatorWorkerSeq)
127 if i.coordinatorWorkerIDs == nil {
128 i.coordinatorWorkerIDs = make(map[string][]string)
129 }
130 i.coordinatorWorkerIDs[workerID] = append(i.coordinatorWorkerIDs[workerID], registrationID)
131 coordinator.Apply(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: registrationID, AgentID: workerID})
132 var actions []orchestration.Action
133 if implicitWave {
134 actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
135 }
136 i.completionDeliveryMu.Unlock()
137 i.executeCoordinatorActions(actions)
138}
139func (i *Interactive) cancelCoordinator() {
140 if i == nil {
141 return
142 }
143 i.completionDeliveryMu.Lock()
144 coordinator := i.ensureCoordinatorLocked()
145 actions := coordinator.Apply(orchestration.Event{Kind: orchestration.EventCancelled}).Actions
146 // A later user turn starts a fresh wave. Drop resident registrations as
147 // well as coordinator identities so late outcomes cannot match a new child.
148 if i.completionTracker != nil {
149 i.completionTracker.Reset()
150 }
151 // Cancellation also invalidates any snapshot a release sampled before it.
152 // Bump the wave generation rather than zeroing the hold count: outstanding
153 // release closures must still balance that count, and the generation is
154 // what stops a stale snapshot from sealing the replacement coordinator.
155 i.completionDeliveryGeneration++
156 i.turnCoordinator = orchestration.New()
157 i.coordinatorWorkerIDs = nil
158 i.completionDeliveryMu.Unlock()
159 i.executeCoordinatorActions(actions)
160}
161func (i *Interactive) takeCoordinatorWorkerID(agentID string) string {
162 if i == nil || agentID == "" {
163 return ""
164 }
165 i.completionDeliveryMu.Lock()
166 defer i.completionDeliveryMu.Unlock()
167 ids := i.coordinatorWorkerIDs[agentID]
168 if len(ids) == 0 {
169 return ""
170 }
171 workerID := ids[0]
172 if len(ids) == 1 {
173 delete(i.coordinatorWorkerIDs, agentID)
174 } else {
175 i.coordinatorWorkerIDs[agentID] = ids[1:]
176 }
177 return workerID
178}
179func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action) {
180 for _, action := range actions {
181 if action.Kind != orchestration.ActionRunManager && action.Kind != orchestration.ActionQueueManager {
182 continue
183 }
184 prompt := action.Text
185 if len(action.Completions) != 0 {
186 instruction := "Briefly summarise the collective outcome for the user. Reference the agents by id. If any failed, suggest a follow-up; otherwise confirm completion. Do not spawn new sub-agents unless the user asks."
187 if action.Kind == orchestration.ActionQueueManager || len(action.ActiveAgentIDs) != 0 {
188 instruction = "Treat these completion reports as worker evidence, not a new user request. Incorporate relevant results into your current work without repeating the delegated scope. Other workers may still be running; do not poll solely to wait for them."
189 }
190 update := subagents.FormatCompletionUpdateWithActive(action.Completions, action.ActiveAgentIDs, instruction)
191 if prompt != "" {
192 prompt = update + "\n\nQueued user request:\n" + prompt
193 } else {
194 prompt = update
195 }
196 }
197 if prompt != "" || len(action.Images) != 0 {
198 i.submitOrQueueMessage(core.QueuedMessage{Text: prompt, Images: action.Images, HostEvent: len(action.Completions) != 0}, false)
199 } else if action.Reason == orchestration.WakeGoal {
200 // Esc-armed reassessment keeps the next turn for user input.
201 if i.interruptedPromptPending() {
202 continue
203 }
204 parent := i.runCtx
205 if parent == nil {
206 parent = context.Background()
207 }
208 i.requestGoalContinuationIfIdle(parent)
209 }
210 }
211}
212func (i *Interactive) requestCompletionDelivery() {
213 if i == nil {
214 return
215 }
216 i.ensureCompletionTracker()
217 i.completionDeliveryMu.Lock()
218 i.completionDeliveryRequest = true
219 if i.completionDeliveryRunning {
220 i.completionDeliveryMu.Unlock()
221 return
222 }
223 i.completionDeliveryRunning = true
224 i.completionDeliveryMu.Unlock()
225
226 go i.deliverCompletionUpdates()
227}
228func (i *Interactive) beginCompletionDeliveryHold() func() {
229 if i == nil {
230 return func() {}
231 }
232 i.completionDeliveryMu.Lock()
233 if i.completionDeliveryHolds == 0 {
234 // A wave starts only from zero holds. Bumping the generation here lets a
235 // release that drained an older wave detect that a newer wave replaced
236 // it while it sampled goal state.
237 i.completionDeliveryGeneration++
238 }
239 i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
240 i.completionDeliveryHolds++
241 i.completionDeliveryMu.Unlock()
242 return i.releaseCompletionDeliveryHold
243}
244func (i *Interactive) releaseCompletionDeliveryHold() {
245 if i == nil {
246 return
247 }
248 i.completionDeliveryMu.Lock()
249 sealing := false
250 var sealingGeneration uint64
251 if i.completionDeliveryHolds > 0 {
252 i.completionDeliveryHolds--
253 if i.completionDeliveryHolds == 0 {
254 sealing = true
255 sealingGeneration = i.completionDeliveryGeneration
256 }
257 }
258 if !sealing {
259 // A nested or duplicate release only returns a hold; it must not sample
260 // goal state or seal a wave it does not own. Starting pending delivery
261 // stays available because that is the existing behaviour whenever the
262 // wave is already idle.
263 start := i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning
264 if start {
265 i.completionDeliveryRunning = true
266 }
267 i.completionDeliveryMu.Unlock()
268 if start {
269 go i.deliverCompletionUpdates()
270 }
271 return
272 }
273 i.completionDeliveryMu.Unlock()
274
275 // Only the release that drained the last hold seals the wave, so this
276 // snapshot belongs to exactly that wave. Read the goal and interruption
277 // state before taking completionDeliveryMu: both helpers acquire i.mu, and
278 // callers legitimately hold i.mu while they read coordinator state, so
279 // acquiring i.mu while holding completionDeliveryMu would invert the lock
280 // order and deadlock.
281 //
282 // The coordinator must observe the goal before sealing the wave, so a
283 // pending worker wins the wake decision instead of a direct continuation.
284 // An Esc-armed reassessment prompt suppresses the wake: the next ordinary
285 // user message owns the turn, not autonomous work.
286 _, goalActive := i.goalContinuationMessage()
287 if i.interruptedPromptPending() {
288 goalActive = false
289 }
290
291 start := false
292 var actions []orchestration.Action
293 i.completionDeliveryMu.Lock()
294 // A newer wave may have replaced this one while goal state was sampled, so
295 // the hold count alone is not enough: a newer hold can begin and fully
296 // drain, leaving the count at zero, and sealing then would re-apply this
297 // stale snapshot over the wave that already sealed. The captured generation
298 // identifies exactly the wave this release drained.
299 if i.completionDeliveryHolds == 0 && i.completionDeliveryGeneration == sealingGeneration {
300 coordinator := i.ensureCoordinatorLocked()
301 coordinator.Apply(orchestration.Event{Kind: orchestration.EventGoalChanged, GoalActive: goalActive})
302 actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
303 }
304 if i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning {
305 i.completionDeliveryRunning = true
306 start = true
307 }
308 i.completionDeliveryMu.Unlock()
309 i.executeCoordinatorActions(actions)
310 if start {
311 go i.deliverCompletionUpdates()
312 }
313}
314func (i *Interactive) completionWaitContext() context.Context {
315 i.mu.Lock()
316 ctx := i.runCtx
317 i.mu.Unlock()
318 if ctx == nil {
319 return context.Background()
320 }
321 return ctx
322}
323func (i *Interactive) deliverCompletionUpdates() {
324 for {
325 i.completionDeliveryMu.Lock()
326 i.completionDeliveryRequest = false
327 tracker := i.completionTracker
328 i.completionDeliveryMu.Unlock()
329
330 batch, err := tracker.WaitReady(i.completionWaitContext())
331 if err == nil && len(batch) != 0 {
332 terminalAgentIDs := make(map[string]struct{}, len(batch))
333 for _, completion := range batch {
334 terminalAgentIDs[completion.AgentID] = struct{}{}
335 }
336 var actions []orchestration.Action
337 for _, completion := range batch {
338 workerID := i.takeCoordinatorWorkerID(completion.AgentID)
339 if workerID == "" {
340 continue
341 }
342 actions = append(actions, i.applyCoordinator(orchestration.Event{
343 Kind: orchestration.EventWorkerFinished,
344 WorkerID: workerID,
345 Completion: completion,
346 })...)
347 }
348 for index := range actions {
349 active := actions[index].ActiveAgentIDs[:0]
350 for _, agentID := range actions[index].ActiveAgentIDs {
351 if _, terminal := terminalAgentIDs[agentID]; !terminal {
352 active = append(active, agentID)
353 }
354 }
355 actions[index].ActiveAgentIDs = active
356 }
357 i.executeCoordinatorActions(actions)
358 }
359
360 i.completionDeliveryMu.Lock()
361 if i.completionDeliveryRequest {
362 i.completionDeliveryMu.Unlock()
363 continue
364 }
365 i.completionDeliveryRunning = false
366 i.completionDeliveryMu.Unlock()
367 return
368 }
369}
370func autoSubagentsAddenda(cfg InteractiveConfig, orchestrating bool) []string {
371 addenda := make([]string, 0, 2)
372 if cfg.ResidentManager != nil && autoSubagentsToolAllowedConfig(cfg) {
373 if addendum := strings.TrimSpace(cfg.SubagentsSystemAddendum); addendum != "" {
374 addenda = append(addenda, addendum)
375 }
376 }
377 if !orchestrating {
378 if cfg.ResidentManager != nil && autoSubagentsAnyToolAllowedConfig(cfg) {
379 if addendum := strings.TrimSpace(cfg.OnDemandSubagentsSystemAddendum); addendum != "" {
380 addenda = append(addenda, addendum)
381 }
382 }
383 return addenda
384 }
385
386 if addendum := strings.TrimSpace(cfg.AutoSubagentsSystemAddendum); addendum != "" {
387 addenda = append(addenda, addendum)
388 }
389 return addenda
390}
391func removeLastAutoSubagentsAddendum(system, addendum string) (string, bool) {
392 idx := strings.LastIndex(system, addendum)
393 if idx < 0 {
394 return system, false
395 }
396 return system[:idx] + system[idx+len(addendum):], true
397}
398func (i *Interactive) applyAutoSubagentsSystemPrompt(orchestrating bool) {
399 // i.mu is sufficient here: this path updates the existing agent's system
400 // prompt and managed addenda in place; it does not replace the agent or
401 // its tool registry, so agentMu is not required.
402 i.mu.Lock()
403 defer i.mu.Unlock()
404 if i.agent == nil {
405 return
406 }
407
408 sys, _ := i.agent.PromptConfig()
409 changed := false
410 if guidance := strings.TrimSpace(i.cfg.WritingGuidance); guidance != "" {
411 var removed bool
412 sys, removed = removeLastAutoSubagentsAddendum(sys, guidance)
413 changed = changed || removed
414 }
415 for idx := len(i.managedAutoSubagentsAddenda) - 1; idx >= 0; idx-- {
416 var removed bool
417 sys, removed = removeLastAutoSubagentsAddendum(sys, i.managedAutoSubagentsAddenda[idx])
418 changed = changed || removed
419 }
420 i.managedAutoSubagentsAddenda = nil
421
422 addenda := autoSubagentsAddenda(i.cfg, orchestrating)
423 if changed {
424 sys = strings.TrimRight(sys, "\n")
425 }
426 for _, addendum := range addenda {
427 if sys != "" {
428 sys += "\n\n"
429 }
430 sys += addendum
431 i.managedAutoSubagentsAddenda = append(i.managedAutoSubagentsAddenda, addendum)
432 changed = true
433 }
434 if guidance := strings.TrimSpace(i.cfg.WritingGuidance); guidance != "" {
435 if sys != "" {
436 sys += "\n\n"
437 }
438 sys += guidance
439 changed = true
440 }
441 if !changed {
442 return
443 }
444 i.agent.SetSystemPrompt(sys)
445}
446func (i *Interactive) autoSubagentsAvailable() bool {
447 return i.cfg.ResidentManager != nil && autoSubagentsAnyToolAllowedConfig(i.cfg)
448}
449func (i *Interactive) autoSubagentsUnavailableHint() string {
450 var hints []string
451 if i.cfg.ResidentManager == nil {
452 hints = append(hints, "resident subagent manager is unavailable in this mode")
453 }
454 if !autoSubagentsAnyToolAllowedConfig(i.cfg) {
455 hints = append(hints, "launch-time tool policy excludes every subagent action")
456 }
457 return strings.Join(hints, "; ")
458}
459func autoSubagentsToolAllowedConfig(cfg InteractiveConfig) bool {
460 return cfg.AutoSubagentsToolAllowed == nil || *cfg.AutoSubagentsToolAllowed
461}
462func autoSubagentsStatusToolAllowedConfig(cfg InteractiveConfig) bool {
463 if cfg.AutoSubagentsStatusToolAllowed != nil {
464 return *cfg.AutoSubagentsStatusToolAllowed
465 }
466 // Preserve the old single-boolean embedding contract: when callers have
467 // not supplied the new field, status follows the existing delegation gate.
468 return autoSubagentsToolAllowedConfig(cfg)
469}
470func autoSubagentsStopToolAllowedConfig(cfg InteractiveConfig) bool {
471 if cfg.AutoSubagentsStopToolAllowed != nil {
472 return *cfg.AutoSubagentsStopToolAllowed
473 }
474 return autoSubagentsToolAllowedConfig(cfg)
475}
476func autoSubagentsResumeToolAllowedConfig(cfg InteractiveConfig) bool {
477 if cfg.AutoSubagentsResumeToolAllowed != nil {
478 return *cfg.AutoSubagentsResumeToolAllowed
479 }
480 return autoSubagentsToolAllowedConfig(cfg)
481}
482func autoSubagentsAnyToolAllowedConfig(cfg InteractiveConfig) bool {
483 return autoSubagentsToolAllowedConfig(cfg) ||
484 autoSubagentsStatusToolAllowedConfig(cfg) ||
485 autoSubagentsStopToolAllowedConfig(cfg) ||
486 autoSubagentsResumeToolAllowedConfig(cfg)
487}
488func (i *Interactive) autoSubagentsToolAllowed() bool {
489 return autoSubagentsToolAllowedConfig(i.cfg)
490}
491func (i *Interactive) autoSubagentsStatusToolAllowed() bool {
492 return autoSubagentsStatusToolAllowedConfig(i.cfg)
493}
494func (i *Interactive) autoSubagentsStopToolAllowed() bool {
495 return autoSubagentsStopToolAllowedConfig(i.cfg)
496}
497func (i *Interactive) autoSubagentsResumeToolAllowed() bool {
498 return autoSubagentsResumeToolAllowedConfig(i.cfg)
499}
500func (i *Interactive) autoSubagentsEnabledLocked() bool {
501 return i.cfg.AutoSubagentsEnabled != nil && *i.cfg.AutoSubagentsEnabled && i.autoSubagentsAvailable()
502}
503func (i *Interactive) applyAutoSubagentsTool() {
504 i.agentMu.Lock()
505 defer i.agentMu.Unlock()
506 i.mu.Lock()
507 defer i.mu.Unlock()
508 if i.agent == nil {
509 return
510 }
511 current := i.agent.ToolsSnapshot()
512 next := core.Registry{}
513 for name, t := range current {
514 if name == tools.SubagentToolName {
515 continue
516 }
517 next[name] = t
518 }
519 if i.autoSubagentsAvailable() {
520 facade := &tools.SubagentTool{}
521 if i.autoSubagentsToolAllowed() {
522 facade.Spawn = &tools.SubagentSpawnTool{
523 ResidentManager: i.cfg.ResidentManager,
524 BuildResidentSpec: i.cfg.BuildResidentSpec,
525 Enabled: func() bool { return true },
526 DefaultModel: func() string { return i.cfg.Model },
527 DefaultProvider: func() string { return i.cfg.Provider },
528 DefaultReasoning: func() string { return i.cfg.Reasoning },
529 ResolveSubagent: i.cfg.ResolveSubagent,
530 }
531 }
532 if i.autoSubagentsStatusToolAllowed() {
533 facade.Status = &tools.SubagentStatusTool{
534 ResidentManager: i.cfg.ResidentManager,
535 Enabled: func() bool { return true },
536 }
537 }
538 if i.autoSubagentsStopToolAllowed() {
539 facade.Stop = &tools.SubagentStopTool{
540 ResidentManager: i.cfg.ResidentManager,
541 Enabled: func() bool { return true },
542 }
543 }
544 if i.autoSubagentsResumeToolAllowed() {
545 facade.Resume = &tools.SubagentResumeTool{
546 ResidentManager: i.cfg.ResidentManager,
547 Enabled: func() bool { return true },
548 }
549 }
550 next[facade.Name()] = facade
551 }
552 i.agent.SetTools(next)
553}
554func (i *Interactive) applyTelegramTools(active bool) {
555 if active {
556 // The child ceiling must be in place even when no live main agent exists,
557 // and before Bridge.Start can accept an external prompt.
558 i.setWebSearchAvailable(false)
559 }
560 i.agentMu.Lock()
561 defer i.agentMu.Unlock()
562 i.mu.Lock()
563 defer i.mu.Unlock()
564 if i.agent == nil {
565 return
566 }
567 current := i.agent.ToolsSnapshot()
568 next := core.Registry{}
569 for name, t := range current {
570 if name == "telegram_send_image" || name == "telegram_send_file" {
571 continue
572 }
573 if active && tools.IsWebCapabilityName(name) {
574 // External Telegram prompts have no per-request confirmation
575 // surface, so V1 does not expose public-web navigation while paired.
576 continue
577 }
578 next[name] = t
579 }
580 if active {
581 sender := telegramSenderAdapter{bridge: i.telegramBridge}
582 next["telegram_send_image"] = &tools.TelegramSendImageTool{
583 CWD: i.cfg.CWD, Sandbox: i.cfg.Sandbox, Sender: sender,
584 }
585 next["telegram_send_file"] = &tools.TelegramSendFileTool{
586 CWD: i.cfg.CWD, Sandbox: i.cfg.Sandbox, Sender: sender,
587 }
588 }
589 i.agent.SetTools(next)
590 _, webSearchAvailable := next["web_search"]
591 i.setWebSearchAvailable(webSearchAvailable && !active)
592}
593func (i *Interactive) telegramStatus() {
594 var msg string
595 if i.telegramBridge != nil && i.telegramBridge.Active() {
596 s := i.telegramBridge.State()
597 msg = "telegram: connected (tui bridge)"
598 if s.Username != "" {
599 msg += " as @" + s.Username
600 }
601 if s.PairedID != 0 {
602 msg += fmt.Sprintf(" - paired with user %d", s.PairedID)
603 } else {
604 msg += " - awaiting pairing"
605 }
606 } else if pid, alive, _ := telegram.IsRunning(i.cfg.ZutHome); alive && pid > 0 {
607 msg = fmt.Sprintf("telegram: background daemon running (pid %d) - /telegram connect won't work until you stop it", pid)
608 } else {
609 cfg, _ := telegram.LoadConfig(i.cfg.ZutHome)
610 if cfg.BotToken == "" {
611 msg = "telegram: not configured. run `zut telegram-bot setup` first."
612 } else {
613 msg = "telegram: disconnected"
614 if cfg.BotUsername != "" {
615 msg += " (@" + cfg.BotUsername + " ready to connect)"
616 }
617 }
618 }
619 i.mu.Lock()
620 i.statusOK = msg
621 i.statusErr = ""
622 i.mu.Unlock()
623 i.invalidate()
624}
625func (h *telegramHost) SubmitOrQueue(prompt string, images []provider.ImageBlock) {
626 h.iv.SubmitOrQueue(prompt, images)
627}
628func (h *telegramHost) CancelTurn() { h.iv.CancelTurn() }
629func (h *telegramHost) Status() string {
630 h.iv.mu.Lock()
631 providerName := h.iv.cfg.Provider
632 model := h.iv.cfg.Model
633 cwd := h.iv.cfg.CWD
634 usage := h.iv.cumUsage
635 subscription := h.iv.cfg.AuthMethod == "oauth"
636 ctxUsed := h.iv.lastCtxInput
637 busy := h.iv.busy
638 queued := len(h.iv.queued)
639 h.iv.mu.Unlock()
640
641 ctxMax := 0
642 if m, err := provider.FindModel(providerName, model); err == nil {
643 ctxMax = m.ContextWindow
644 }
645 return telegram.FormatStatus(telegram.StatusSnapshot{
646 Provider: providerName,
647 Model: model,
648 CWD: cwd,
649 Usage: usage,
650 Subscription: subscription,
651 ContextUsed: ctxUsed,
652 ContextMax: ctxMax,
653 Busy: busy,
654 Queued: queued,
655 })
656}
657func (h *telegramHost) Notify(level, message string) {
658 h.iv.mu.Lock()
659 switch level {
660 case "error", "warn":
661 h.iv.statusErr = message
662 h.iv.statusOK = ""
663 default:
664 h.iv.statusOK = message
665 h.iv.statusErr = ""
666 }
667 h.iv.mu.Unlock()
668 h.iv.invalidate()
669}
670func (i *Interactive) openSessionOpsDialog() {
671 items := []sessionOpsItem{
672 {label: "export", action: "export", hint: "write the current session to a .zutsession file"},
673 {label: "import", action: "import", hint: "load a .zutsession file into this directory"},
674 {label: "fork", action: "fork", hint: "branch from a past user message into a new session"},
675 {label: "tree", action: "tree", hint: "switch between branches in this directory"},
676 }
677 i.sessionOpsDialog.Open(items)
678 i.invalidate()
679}
680func (i *Interactive) doSessionOp(action, arg string) {
681 switch action {
682 case "export":
683 i.doSessionExport(arg)
684 case "import":
685 i.doSessionImport(arg)
686 case "fork":
687 i.doSessionFork()
688 case "tree":
689 i.doSessionTree()
690 default:
691 i.mu.Lock()
692 i.statusErr = "unknown /session action: " + action + " (use export, import, fork, or tree)"
693 i.mu.Unlock()
694 i.invalidate()
695 }
696}
697func (i *Interactive) doSessionExport(dst string) {
698 src := i.currentSessionPath()
699 if src == "" {
700 i.mu.Lock()
701 i.statusErr = "export: no session is active (running with --no-session?)"
702 i.mu.Unlock()
703 i.invalidate()
704 return
705 }
706 // Persist any in-memory agent messages to the session file so
707 // the export carries the full conversation. Without this, the
708 // default lazy-flush-at-exit strategy leaves most of a running
709 // session unwritten and the export ends up with only the meta.
710 if i.cfg.FlushSession != nil {
711 i.cfg.FlushSession()
712 }
713 dst = unquotePath(dst)
714 if dst == "" {
715 dst = defaultExportDir()
716 } else {
717 dst = expandTilde(dst)
718 }
719 out, err := core.ExportSession(src, dst)
720 if err != nil {
721 i.mu.Lock()
722 i.statusErr = "export: " + err.Error()
723 i.mu.Unlock()
724 i.invalidate()
725 return
726 }
727 i.mu.Lock()
728 i.statusOK = "exported session to " + friendlyPath(out)
729 i.statusErr = ""
730 i.mu.Unlock()
731 i.invalidate()
732}
733func (i *Interactive) doSessionImport(src string) {
734 src = unquotePath(src)
735 if src == "" {
736 i.mu.Lock()
737 i.statusErr = "import: pass a path — e.g. /session import ~/Downloads/work.zutsession"
738 i.mu.Unlock()
739 i.invalidate()
740 return
741 }
742 src = expandTilde(src)
743 if _, err := os.Stat(src); err != nil {
744 i.mu.Lock()
745 i.statusErr = "import: " + err.Error()
746 i.mu.Unlock()
747 i.invalidate()
748 return
749 }
750 newPath, err := core.ImportSession(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version)
751 if err != nil {
752 i.mu.Lock()
753 i.statusErr = "import: " + err.Error()
754 i.mu.Unlock()
755 i.invalidate()
756 return
757 }
758 if i.cfg.LoadSession == nil {
759 i.mu.Lock()
760 i.statusOK = "imported session at " + friendlyPath(newPath) + " (run /sessions to resume it)"
761 i.statusErr = ""
762 i.mu.Unlock()
763 i.invalidate()
764 return
765 }
766 i.markSessionTitleSwitching()
767 if err := i.cfg.LoadSession(newPath); err != nil {
768 i.restoreFailedSessionTitle()
769 i.mu.Lock()
770 i.statusErr = "import: load failed: " + err.Error()
771 i.mu.Unlock()
772 i.invalidate()
773 return
774 }
775 i.restoreLoadedSessionTitle()
776 state := i.restoreCurrentCompactHandoff()
777 i.mu.Lock()
778 i.statusOK = "imported and switched to session " + friendlyPath(newPath)
779 i.statusErr = ""
780 if i.agent != nil {
781 i.view.Messages = filterHiddenTranscriptMessages(i.agent.Messages())
782 i.cumUsage = i.agent.Cost()
783 last := i.agent.LastTurnUsage()
784 i.lastCtxInput = last.InputTokens + last.CacheReadTokens + last.CacheWriteTokens
785 if len(i.view.Messages) > initialResumeTailLimit {
786 i.view.TailLimit = initialResumeTailLimit
787 } else {
788 i.view.TailLimit = 0
789 }
790 i.view.InvalidateRenderCache()
791 }
792 i.mu.Unlock()
793 i.invalidate()
794 if state.reason != compactContinuationNone {
795 i.startRestoredCompactHandoff(i.runCtx)
796 }
797}
798func defaultExportDir() string {
799 home, err := os.UserHomeDir()
800 if err != nil || home == "" {
801 return os.TempDir()
802 }
803 downloads := filepath.Join(home, "Downloads")
804 if fi, err := os.Stat(downloads); err == nil && fi.IsDir() {
805 return downloads
806 }
807 return home
808}
809func expandTilde(p string) string {
810 if p == "" || p[0] != '~' {
811 return p
812 }
813 home, err := os.UserHomeDir()
814 if err != nil || home == "" {
815 return p
816 }
817 if len(p) == 1 {
818 return home
819 }
820 if p[1] == '/' || p[1] == filepath.Separator {
821 return filepath.Join(home, p[2:])
822 }
823 return p
824}
825func unquotePath(p string) string {
826 p = strings.TrimSpace(p)
827 if len(p) >= 2 {
828 first := p[0]
829 last := p[len(p)-1]
830 if (first == '\'' && last == '\'') || (first == '"' && last == '"') {
831 p = p[1 : len(p)-1]
832 }
833 }
834 return p
835}
836func friendlyPath(p string) string {
837 home, err := os.UserHomeDir()
838 if err != nil || home == "" {
839 return p
840 }
841 if strings.HasPrefix(p, home+string(filepath.Separator)) {
842 return "~" + p[len(home):]
843 }
844 return p
845}
846func (i *Interactive) doSessionFork() {
847 path := i.currentSessionPath()
848 if path == "" {
849 i.mu.Lock()
850 i.statusErr = "fork: no session is active (running with --no-session?)"
851 i.mu.Unlock()
852 i.invalidate()
853 return
854 }
855 i.mu.Lock()
856 busy := i.busy || i.shellRunning || i.compacting || i.autoCompacting || i.sessionLoading
857 queued := len(i.queued) != 0
858 ag := i.agent
859 i.mu.Unlock()
860 if busy || queued || ag == nil || ag.QueuedMessageCount() != 0 {
861 i.mu.Lock()
862 i.statusErr = "fork: wait until the current turn and queued messages finish"
863 i.mu.Unlock()
864 i.invalidate()
865 return
866 }
867 if i.cfg.FlushSession != nil {
868 i.cfg.FlushSession()
869 }
870 snapshot, err := core.ReadSessionSnapshot(path)
871 if err != nil {
872 i.mu.Lock()
873 i.statusErr = "fork: " + err.Error()
874 i.mu.Unlock()
875 i.invalidate()
876 return
877 }
878 msgs := snapshot.Messages
879 if len(msgs) == 0 {
880 i.mu.Lock()
881 i.statusErr = "fork: transcript is empty; nothing to fork from"
882 i.mu.Unlock()
883 i.invalidate()
884 return
885 }
886 i.pendingFork = true
887 i.jumpDialog.Open(msgs, "")
888 i.invalidate()
889}
890func (i *Interactive) doSessionTree() {
891 i.openSessionTree()
892}
893func (i *Interactive) applySessionTreeMessageSelection(src string, msgIdx, turnNo int, role provider.Role, prompt string) {
894 target := sessionTreeTarget{
895 SourcePath: src,
896 EffectiveIndex: msgIdx,
897 SelectionBoundary: msgIdx + 1,
898 Role: role,
899 UserDraft: prompt,
900 Boundary: sessionTreeMessageBoundary,
901 }
902 if role == provider.RoleUser {
903 target.SelectionBoundary = msgIdx
904 }
905 i.applySessionTreeTarget(target, turnNo)
906}
907func (i *Interactive) applySessionTreeTarget(target sessionTreeTarget, turnNo int) {
908 // Selection is a state-changing operation too. Use the selection gate so
909 // a queued/busy turn cannot race the hidden branch or accidentally submit
910 // a provider turn while the session is being swapped. The open gate cannot
911 // be reused here because the tree dialog itself owns the keyboard.
912 if !i.canCommitSessionTreeSelection() {
913 i.setSessionTreeError("tree: session branching is not available in this build")
914 return
915 }
916 src := target.SourcePath
917 if src == "" {
918 src = i.currentSessionPath()
919 }
920 if src == "" {
921 i.setSessionTreeError("tree: no session is active")
922 return
923 }
924
925 // Read and validate the selected row before flushing or creating anything.
926 // This gives tool-call/result pairs an atomic boundary and prevents a
927 // malformed selection from changing the active session. Historical rows
928 // come from a pre-compaction segment; current rows use the effective
929 // snapshot that the running agent resumes from.
930 var msgs []provider.Message
931 var historicalSegment core.SessionHistorySegment
932 if target.Historical {
933 history, err := core.ReadSessionHistory(src)
934 if err != nil {
935 i.setSessionTreeError("tree: read selection: " + err.Error())
936 return
937 }
938 if target.HistorySegment < 0 || target.HistorySegment >= len(history.Segments) {
939 i.setSessionTreeError("tree: selected history segment is unavailable")
940 return
941 }
942 historicalSegment = history.Segments[target.HistorySegment]
943 msgs = historicalSegment.Messages
944 } else {
945 sess, current, err := core.OpenSession(src)
946 if err != nil {
947 i.setSessionTreeError("tree: read selection: " + err.Error())
948 return
949 }
950 if err := sess.Close(); err != nil {
951 i.setSessionTreeError("tree: close selection: " + err.Error())
952 return
953 }
954 msgs = current
955 }
956 selection, err := sessionTreeSelection(msgs, target)
957 if err != nil {
958 i.setSessionTreeError("tree: " + err.Error())
959 return
960 }
961 if i.cfg.FlushSession != nil {
962 i.cfg.FlushSession()
963 }
964 var newPath string
965 if target.Historical {
966 newPath, err = core.BranchSessionHiddenFromHistory(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version, historicalSegment, selection.upTo)
967 } else {
968 newPath, err = core.BranchSessionHidden(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version, selection.upTo)
969 }
970 if err != nil {
971 i.setSessionTreeError("tree: " + err.Error())
972 return
973 }
974 i.markSessionTitleSwitching()
975 if err := i.cfg.LoadSession(newPath); err != nil {
976 i.restoreFailedSessionTitle()
977 i.setSessionTreeError("tree: checkout failed: " + err.Error())
978 return
979 }
980 state := i.restoreCurrentCompactHandoff()
981
982 // LoadSession is intentionally the old func(string) error callback: the
983 // CLI and embedders already own agent/session swapping. Refresh the view
984 // from the swapped agent here instead of requiring a new callback result.
985 i.resetSessionTitleForFreshBranch()
986 i.scrollToBottom()
987 i.mu.Lock()
988 i.lastCtxInput = 0
989 if i.agent != nil {
990 i.view.Messages = filterHiddenTranscriptMessages(i.agent.Messages())
991 i.cumUsage = i.agent.Cost()
992 last := i.agent.LastTurnUsage()
993 i.lastCtxInput = last.InputTokens + last.CacheReadTokens + last.CacheWriteTokens
994 i.view.InvalidateRenderCache()
995 }
996 i.toolCalls = map[string]*tui.ToolCallView{}
997 i.toolOrder = nil
998 i.toolGate = map[string]int{}
999 i.extNotes = nil
1000 if selection.restoreDraft {
1001 i.ed.SetValue(selection.draftText)
1002 i.clipboardImages = selection.images
1003 i.statusOK = fmt.Sprintf("checked out before turn %d; edit and send to branch", turnNo)
1004 } else {
1005 i.ed.Clear()
1006 i.clipboardImages = nil
1007 i.statusOK = fmt.Sprintf("checked out turn %d into a new branch", turnNo)
1008 }
1009 i.statusErr = ""
1010 i.mu.Unlock()
1011 i.scrollToBottom()
1012 // Selection does not submit, start, or queue the restored user draft; only
1013 // a later explicit Enter does that. An inherited compact handoff may resume
1014 // its hidden continuation below when this branch has the complete effective
1015 // transcript, without submitting the restored draft.
1016 i.invalidate()
1017 if state.reason != compactContinuationNone {
1018 i.startRestoredCompactHandoff(i.runCtx)
1019 }
1020}
1021func (i *Interactive) applyForkSelection(msgIdx int) {
1022 i.pendingFork = false
1023 src := i.currentSessionPath()
1024 if src == "" {
1025 i.mu.Lock()
1026 i.statusErr = "fork: no session is active"
1027 i.mu.Unlock()
1028 i.invalidate()
1029 return
1030 }
1031 if i.cfg.FlushSession != nil {
1032 i.cfg.FlushSession()
1033 }
1034 // msgIdx is 0-indexed message position; copy msgIdx+1 rows so
1035 // the selected user message is included.
1036 upTo := msgIdx + 1
1037 newPath, err := core.BranchSession(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version, upTo)
1038 if err != nil {
1039 i.mu.Lock()
1040 i.statusErr = "fork: " + err.Error()
1041 i.mu.Unlock()
1042 i.invalidate()
1043 return
1044 }
1045 if i.cfg.LoadSession == nil {
1046 i.mu.Lock()
1047 i.statusOK = "forked at message " + formatInt(upTo) + " (run /sessions to resume)"
1048 i.statusErr = ""
1049 i.mu.Unlock()
1050 i.invalidate()
1051 return
1052 }
1053 i.markSessionTitleSwitching()
1054 if err := i.cfg.LoadSession(newPath); err != nil {
1055 i.restoreFailedSessionTitle()
1056 i.mu.Lock()
1057 i.statusErr = "fork: switch failed: " + err.Error()
1058 i.mu.Unlock()
1059 i.invalidate()
1060 return
1061 }
1062 state := i.restoreCurrentCompactHandoff()
1063 i.resetSessionTitleForFreshBranch()
1064 i.mu.Lock()
1065 i.statusOK = "forked and switched to new branch at " + friendlyPath(newPath)
1066 i.statusErr = ""
1067 i.mu.Unlock()
1068 i.invalidate()
1069 if state.reason != compactContinuationNone {
1070 i.startRestoredCompactHandoff(i.runCtx)
1071 }
1072}
1073func formatInt(n int) string {
1074 return fmt.Sprintf("%d", n)
1075}
1076func assistantText(m provider.Message) string {
1077 var sb strings.Builder
1078 for _, c := range m.Content {
1079 if tb, ok := c.(provider.TextBlock); ok {
1080 if sb.Len() > 0 {
1081 sb.WriteByte('\n')
1082 }
1083 sb.WriteString(tb.Text)
1084 }
1085 }
1086 return sb.String()
1087}
1088func (i *Interactive) resetTranscriptRenderLocked() {
1089 i.view.InvalidateRenderCache()
1090 i.chatCacheValid = false
1091 i.prevChatLen = 0
1092 i.prevChatCols = 0
1093 i.prevChatRows = 0
1094 i.prevScrollOffset = 0
1095 i.requestRendererInvalidate()
1096}
1097func (i *Interactive) resetStreamingStateLocked() {
1098 i.streaming.Reset()
1099 i.streamPending = i.streamPending[:0]
1100 i.streamFlushPending = false
1101 i.streamOn = false
1102 i.openAllToolGatesLocked()
1103}
1104func (i *Interactive) openAllToolGatesLocked() {
1105 for id := range i.toolGate {
1106 i.toolGate[id] = 0
1107 }
1108}
1109func (i *Interactive) gateToolLocked(id string) {
1110 if _, ok := i.toolGate[id]; ok {
1111 return
1112 }
1113 if !i.streamOn {
1114 i.toolGate[id] = 0
1115 return
1116 }
1117 i.toolGate[id] = i.streaming.Len() + len(i.streamPending)
1118}
1119func (i *Interactive) toolGateOpenLocked(id string) bool {
1120 gate, ok := i.toolGate[id]
1121 if !ok || gate == 0 {
1122 return true
1123 }
1124 return i.streaming.Len() >= gate
1125}
1126func (i *Interactive) assistantMessageSideEffects(m provider.Message) {
1127 if i.cfg.OnAssistant != nil {
1128 i.cfg.OnAssistant(m)
1129 }
1130 if i.telegramBridge != nil && i.telegramBridge.Active() {
1131 var sb strings.Builder
1132 for _, c := range m.Content {
1133 if tb, ok := c.(provider.TextBlock); ok {
1134 if sb.Len() > 0 {
1135 sb.WriteString("\n")
1136 }
1137 sb.WriteString(tb.Text)
1138 }
1139 }
1140 if text := sb.String(); strings.TrimSpace(text) != "" {
1141 go i.telegramBridge.OnAssistantText(text)
1142 }
1143 }
1144}
1145func (i *Interactive) runStreamPacer(ctx context.Context) {
1146 t := time.NewTicker(paintPaceInterval)
1147 defer t.Stop()
1148 for {
1149 select {
1150 case <-ctx.Done():
1151 return
1152 case <-t.C:
1153 i.mu.Lock()
1154 if len(i.streamPending) == 0 {
1155 // EvAssistantMessage already fired but the pacer
1156 // was still draining a tick ago. Everything is now
1157 // painted; clear the streaming flags so the next
1158 // redraw shows the finalised transcript message
1159 // and hides the streaming overlay.
1160 if i.streamFlushPending {
1161 i.streamFlushPending = false
1162 i.streaming.Reset()
1163 i.streamOn = false
1164 i.openAllToolGatesLocked()
1165 i.mu.Unlock()
1166 i.invalidate()
1167 continue
1168 }
1169 i.mu.Unlock()
1170 continue
1171 }
1172 n := paintPaceRate
1173 if n > len(i.streamPending) {
1174 n = len(i.streamPending)
1175 }
1176 i.streaming.WriteString(string(i.streamPending[:n]))
1177 i.streamPending = i.streamPending[n:]
1178 i.mu.Unlock()
1179 i.invalidate()
1180 }
1181 }
1182}