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 subagent manager tools")
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 == "subagent_spawn" || name == "subagent_status" || name == "subagent_stop" || name == "subagent_resume" {
515 continue
516 }
517 next[name] = t
518 }
519 if i.autoSubagentsAvailable() {
520 if i.autoSubagentsToolAllowed() {
521 canonical := &tools.SubagentSpawnTool{
522 ResidentManager: i.cfg.ResidentManager,
523 BuildResidentSpec: i.cfg.BuildResidentSpec,
524 Enabled: func() bool { return true },
525 DefaultModel: func() string { return i.cfg.Model },
526 DefaultProvider: func() string { return i.cfg.Provider },
527 DefaultReasoning: func() string { return i.cfg.Reasoning },
528 ResolveSubagent: i.cfg.ResolveSubagent,
529 }
530 next[canonical.Name()] = canonical
531 }
532 if i.autoSubagentsStatusToolAllowed() {
533 statusTool := &tools.SubagentStatusTool{
534 ResidentManager: i.cfg.ResidentManager,
535 Enabled: func() bool { return true },
536 }
537 next[statusTool.Name()] = statusTool
538 }
539 if i.autoSubagentsStopToolAllowed() {
540 stopTool := &tools.SubagentStopTool{
541 ResidentManager: i.cfg.ResidentManager,
542 Enabled: func() bool { return true },
543 }
544 next[stopTool.Name()] = stopTool
545 }
546 if i.autoSubagentsResumeToolAllowed() {
547 resumeTool := &tools.SubagentResumeTool{
548 ResidentManager: i.cfg.ResidentManager,
549 Enabled: func() bool { return true },
550 }
551 next[resumeTool.Name()] = resumeTool
552 }
553 }
554 i.agent.SetTools(next)
555}
556func (i *Interactive) applyTelegramTools(active bool) {
557 if active {
558 // The child ceiling must be in place even when no live main agent exists,
559 // and before Bridge.Start can accept an external prompt.
560 i.setWebSearchAvailable(false)
561 }
562 i.agentMu.Lock()
563 defer i.agentMu.Unlock()
564 i.mu.Lock()
565 defer i.mu.Unlock()
566 if i.agent == nil {
567 return
568 }
569 current := i.agent.ToolsSnapshot()
570 next := core.Registry{}
571 for name, t := range current {
572 if name == "telegram_send_image" || name == "telegram_send_file" {
573 continue
574 }
575 if active && tools.IsWebCapabilityName(name) {
576 // External Telegram prompts have no per-request confirmation
577 // surface, so V1 does not expose public-web navigation while paired.
578 continue
579 }
580 next[name] = t
581 }
582 if active {
583 sender := telegramSenderAdapter{bridge: i.telegramBridge}
584 next["telegram_send_image"] = &tools.TelegramSendImageTool{
585 CWD: i.cfg.CWD, Sandbox: i.cfg.Sandbox, Sender: sender,
586 }
587 next["telegram_send_file"] = &tools.TelegramSendFileTool{
588 CWD: i.cfg.CWD, Sandbox: i.cfg.Sandbox, Sender: sender,
589 }
590 }
591 i.agent.SetTools(next)
592 _, webSearchAvailable := next["web_search"]
593 i.setWebSearchAvailable(webSearchAvailable && !active)
594}
595func (i *Interactive) telegramStatus() {
596 var msg string
597 if i.telegramBridge != nil && i.telegramBridge.Active() {
598 s := i.telegramBridge.State()
599 msg = "telegram: connected (tui bridge)"
600 if s.Username != "" {
601 msg += " as @" + s.Username
602 }
603 if s.PairedID != 0 {
604 msg += fmt.Sprintf(" - paired with user %d", s.PairedID)
605 } else {
606 msg += " - awaiting pairing"
607 }
608 } else if pid, alive, _ := telegram.IsRunning(i.cfg.ZutHome); alive && pid > 0 {
609 msg = fmt.Sprintf("telegram: background daemon running (pid %d) - /telegram connect won't work until you stop it", pid)
610 } else {
611 cfg, _ := telegram.LoadConfig(i.cfg.ZutHome)
612 if cfg.BotToken == "" {
613 msg = "telegram: not configured. run `zut telegram-bot setup` first."
614 } else {
615 msg = "telegram: disconnected"
616 if cfg.BotUsername != "" {
617 msg += " (@" + cfg.BotUsername + " ready to connect)"
618 }
619 }
620 }
621 i.mu.Lock()
622 i.statusOK = msg
623 i.statusErr = ""
624 i.mu.Unlock()
625 i.invalidate()
626}
627func (h *telegramHost) SubmitOrQueue(prompt string, images []provider.ImageBlock) {
628 h.iv.SubmitOrQueue(prompt, images)
629}
630func (h *telegramHost) CancelTurn() { h.iv.CancelTurn() }
631func (h *telegramHost) Status() string {
632 h.iv.mu.Lock()
633 providerName := h.iv.cfg.Provider
634 model := h.iv.cfg.Model
635 cwd := h.iv.cfg.CWD
636 usage := h.iv.cumUsage
637 subscription := h.iv.cfg.AuthMethod == "oauth"
638 ctxUsed := h.iv.lastCtxInput
639 busy := h.iv.busy
640 queued := len(h.iv.queued)
641 h.iv.mu.Unlock()
642
643 ctxMax := 0
644 if m, err := provider.FindModel(providerName, model); err == nil {
645 ctxMax = m.ContextWindow
646 }
647 return telegram.FormatStatus(telegram.StatusSnapshot{
648 Provider: providerName,
649 Model: model,
650 CWD: cwd,
651 Usage: usage,
652 Subscription: subscription,
653 ContextUsed: ctxUsed,
654 ContextMax: ctxMax,
655 Busy: busy,
656 Queued: queued,
657 })
658}
659func (h *telegramHost) Notify(level, message string) {
660 h.iv.mu.Lock()
661 switch level {
662 case "error", "warn":
663 h.iv.statusErr = message
664 h.iv.statusOK = ""
665 default:
666 h.iv.statusOK = message
667 h.iv.statusErr = ""
668 }
669 h.iv.mu.Unlock()
670 h.iv.invalidate()
671}
672func (i *Interactive) openSessionOpsDialog() {
673 items := []sessionOpsItem{
674 {label: "export", action: "export", hint: "write the current session to a .zutsession file"},
675 {label: "import", action: "import", hint: "load a .zutsession file into this directory"},
676 {label: "fork", action: "fork", hint: "branch from a past user message into a new session"},
677 {label: "tree", action: "tree", hint: "switch between branches in this directory"},
678 }
679 i.sessionOpsDialog.Open(items)
680 i.invalidate()
681}
682func (i *Interactive) doSessionOp(action, arg string) {
683 switch action {
684 case "export":
685 i.doSessionExport(arg)
686 case "import":
687 i.doSessionImport(arg)
688 case "fork":
689 i.doSessionFork()
690 case "tree":
691 i.doSessionTree()
692 default:
693 i.mu.Lock()
694 i.statusErr = "unknown /session action: " + action + " (use export, import, fork, or tree)"
695 i.mu.Unlock()
696 i.invalidate()
697 }
698}
699func (i *Interactive) doSessionExport(dst string) {
700 src := i.currentSessionPath()
701 if src == "" {
702 i.mu.Lock()
703 i.statusErr = "export: no session is active (running with --no-session?)"
704 i.mu.Unlock()
705 i.invalidate()
706 return
707 }
708 // Persist any in-memory agent messages to the session file so
709 // the export carries the full conversation. Without this, the
710 // default lazy-flush-at-exit strategy leaves most of a running
711 // session unwritten and the export ends up with only the meta.
712 if i.cfg.FlushSession != nil {
713 i.cfg.FlushSession()
714 }
715 dst = unquotePath(dst)
716 if dst == "" {
717 dst = defaultExportDir()
718 } else {
719 dst = expandTilde(dst)
720 }
721 out, err := core.ExportSession(src, dst)
722 if err != nil {
723 i.mu.Lock()
724 i.statusErr = "export: " + err.Error()
725 i.mu.Unlock()
726 i.invalidate()
727 return
728 }
729 i.mu.Lock()
730 i.statusOK = "exported session to " + friendlyPath(out)
731 i.statusErr = ""
732 i.mu.Unlock()
733 i.invalidate()
734}
735func (i *Interactive) doSessionImport(src string) {
736 src = unquotePath(src)
737 if src == "" {
738 i.mu.Lock()
739 i.statusErr = "import: pass a path — e.g. /session import ~/Downloads/work.zutsession"
740 i.mu.Unlock()
741 i.invalidate()
742 return
743 }
744 src = expandTilde(src)
745 if _, err := os.Stat(src); err != nil {
746 i.mu.Lock()
747 i.statusErr = "import: " + err.Error()
748 i.mu.Unlock()
749 i.invalidate()
750 return
751 }
752 newPath, err := core.ImportSession(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version)
753 if err != nil {
754 i.mu.Lock()
755 i.statusErr = "import: " + err.Error()
756 i.mu.Unlock()
757 i.invalidate()
758 return
759 }
760 if i.cfg.LoadSession == nil {
761 i.mu.Lock()
762 i.statusOK = "imported session at " + friendlyPath(newPath) + " (run /sessions to resume it)"
763 i.statusErr = ""
764 i.mu.Unlock()
765 i.invalidate()
766 return
767 }
768 i.markSessionTitleSwitching()
769 if err := i.cfg.LoadSession(newPath); err != nil {
770 i.restoreFailedSessionTitle()
771 i.mu.Lock()
772 i.statusErr = "import: load failed: " + err.Error()
773 i.mu.Unlock()
774 i.invalidate()
775 return
776 }
777 i.restoreLoadedSessionTitle()
778 state := i.restoreCurrentCompactHandoff()
779 i.mu.Lock()
780 i.statusOK = "imported and switched to session " + friendlyPath(newPath)
781 i.statusErr = ""
782 if i.agent != nil {
783 i.view.Messages = filterHiddenTranscriptMessages(i.agent.Messages())
784 i.cumUsage = i.agent.Cost()
785 last := i.agent.LastTurnUsage()
786 i.lastCtxInput = last.InputTokens + last.CacheReadTokens + last.CacheWriteTokens
787 if len(i.view.Messages) > initialResumeTailLimit {
788 i.view.TailLimit = initialResumeTailLimit
789 } else {
790 i.view.TailLimit = 0
791 }
792 i.view.InvalidateRenderCache()
793 }
794 i.mu.Unlock()
795 i.invalidate()
796 if state.reason != compactContinuationNone {
797 i.startRestoredCompactHandoff(i.runCtx)
798 }
799}
800func defaultExportDir() string {
801 home, err := os.UserHomeDir()
802 if err != nil || home == "" {
803 return os.TempDir()
804 }
805 downloads := filepath.Join(home, "Downloads")
806 if fi, err := os.Stat(downloads); err == nil && fi.IsDir() {
807 return downloads
808 }
809 return home
810}
811func expandTilde(p string) string {
812 if p == "" || p[0] != '~' {
813 return p
814 }
815 home, err := os.UserHomeDir()
816 if err != nil || home == "" {
817 return p
818 }
819 if len(p) == 1 {
820 return home
821 }
822 if p[1] == '/' || p[1] == filepath.Separator {
823 return filepath.Join(home, p[2:])
824 }
825 return p
826}
827func unquotePath(p string) string {
828 p = strings.TrimSpace(p)
829 if len(p) >= 2 {
830 first := p[0]
831 last := p[len(p)-1]
832 if (first == '\'' && last == '\'') || (first == '"' && last == '"') {
833 p = p[1 : len(p)-1]
834 }
835 }
836 return p
837}
838func friendlyPath(p string) string {
839 home, err := os.UserHomeDir()
840 if err != nil || home == "" {
841 return p
842 }
843 if strings.HasPrefix(p, home+string(filepath.Separator)) {
844 return "~" + p[len(home):]
845 }
846 return p
847}
848func (i *Interactive) doSessionFork() {
849 path := i.currentSessionPath()
850 if path == "" {
851 i.mu.Lock()
852 i.statusErr = "fork: no session is active (running with --no-session?)"
853 i.mu.Unlock()
854 i.invalidate()
855 return
856 }
857 i.mu.Lock()
858 busy := i.busy || i.shellRunning || i.compacting || i.autoCompacting || i.sessionLoading
859 queued := len(i.queued) != 0
860 ag := i.agent
861 i.mu.Unlock()
862 if busy || queued || ag == nil || ag.QueuedMessageCount() != 0 {
863 i.mu.Lock()
864 i.statusErr = "fork: wait until the current turn and queued messages finish"
865 i.mu.Unlock()
866 i.invalidate()
867 return
868 }
869 if i.cfg.FlushSession != nil {
870 i.cfg.FlushSession()
871 }
872 snapshot, err := core.ReadSessionSnapshot(path)
873 if err != nil {
874 i.mu.Lock()
875 i.statusErr = "fork: " + err.Error()
876 i.mu.Unlock()
877 i.invalidate()
878 return
879 }
880 msgs := snapshot.Messages
881 if len(msgs) == 0 {
882 i.mu.Lock()
883 i.statusErr = "fork: transcript is empty; nothing to fork from"
884 i.mu.Unlock()
885 i.invalidate()
886 return
887 }
888 i.pendingFork = true
889 i.jumpDialog.Open(msgs, "")
890 i.invalidate()
891}
892func (i *Interactive) doSessionTree() {
893 i.openSessionTree()
894}
895func (i *Interactive) applySessionTreeMessageSelection(src string, msgIdx, turnNo int, role provider.Role, prompt string) {
896 target := sessionTreeTarget{
897 SourcePath: src,
898 EffectiveIndex: msgIdx,
899 SelectionBoundary: msgIdx + 1,
900 Role: role,
901 UserDraft: prompt,
902 Boundary: sessionTreeMessageBoundary,
903 }
904 if role == provider.RoleUser {
905 target.SelectionBoundary = msgIdx
906 }
907 i.applySessionTreeTarget(target, turnNo)
908}
909func (i *Interactive) applySessionTreeTarget(target sessionTreeTarget, turnNo int) {
910 // Selection is a state-changing operation too. Use the selection gate so
911 // a queued/busy turn cannot race the hidden branch or accidentally submit
912 // a provider turn while the session is being swapped. The open gate cannot
913 // be reused here because the tree dialog itself owns the keyboard.
914 if !i.canCommitSessionTreeSelection() {
915 i.setSessionTreeError("tree: session branching is not available in this build")
916 return
917 }
918 src := target.SourcePath
919 if src == "" {
920 src = i.currentSessionPath()
921 }
922 if src == "" {
923 i.setSessionTreeError("tree: no session is active")
924 return
925 }
926
927 // Read and validate the selected row before flushing or creating anything.
928 // This gives tool-call/result pairs an atomic boundary and prevents a
929 // malformed selection from changing the active session. Historical rows
930 // come from a pre-compaction segment; current rows use the effective
931 // snapshot that the running agent resumes from.
932 var msgs []provider.Message
933 var historicalSegment core.SessionHistorySegment
934 if target.Historical {
935 history, err := core.ReadSessionHistory(src)
936 if err != nil {
937 i.setSessionTreeError("tree: read selection: " + err.Error())
938 return
939 }
940 if target.HistorySegment < 0 || target.HistorySegment >= len(history.Segments) {
941 i.setSessionTreeError("tree: selected history segment is unavailable")
942 return
943 }
944 historicalSegment = history.Segments[target.HistorySegment]
945 msgs = historicalSegment.Messages
946 } else {
947 sess, current, err := core.OpenSession(src)
948 if err != nil {
949 i.setSessionTreeError("tree: read selection: " + err.Error())
950 return
951 }
952 if err := sess.Close(); err != nil {
953 i.setSessionTreeError("tree: close selection: " + err.Error())
954 return
955 }
956 msgs = current
957 }
958 selection, err := sessionTreeSelection(msgs, target)
959 if err != nil {
960 i.setSessionTreeError("tree: " + err.Error())
961 return
962 }
963 if i.cfg.FlushSession != nil {
964 i.cfg.FlushSession()
965 }
966 var newPath string
967 if target.Historical {
968 newPath, err = core.BranchSessionHiddenFromHistory(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version, historicalSegment, selection.upTo)
969 } else {
970 newPath, err = core.BranchSessionHidden(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version, selection.upTo)
971 }
972 if err != nil {
973 i.setSessionTreeError("tree: " + err.Error())
974 return
975 }
976 i.markSessionTitleSwitching()
977 if err := i.cfg.LoadSession(newPath); err != nil {
978 i.restoreFailedSessionTitle()
979 i.setSessionTreeError("tree: checkout failed: " + err.Error())
980 return
981 }
982 state := i.restoreCurrentCompactHandoff()
983
984 // LoadSession is intentionally the old func(string) error callback: the
985 // CLI and embedders already own agent/session swapping. Refresh the view
986 // from the swapped agent here instead of requiring a new callback result.
987 i.resetSessionTitleForFreshBranch()
988 i.scrollToBottom()
989 i.mu.Lock()
990 i.lastCtxInput = 0
991 if i.agent != nil {
992 i.view.Messages = filterHiddenTranscriptMessages(i.agent.Messages())
993 i.cumUsage = i.agent.Cost()
994 last := i.agent.LastTurnUsage()
995 i.lastCtxInput = last.InputTokens + last.CacheReadTokens + last.CacheWriteTokens
996 i.view.InvalidateRenderCache()
997 }
998 i.toolCalls = map[string]*tui.ToolCallView{}
999 i.toolOrder = nil
1000 i.toolGate = map[string]int{}
1001 i.extNotes = nil
1002 if selection.restoreDraft {
1003 i.ed.SetValue(selection.draftText)
1004 i.clipboardImages = selection.images
1005 i.statusOK = fmt.Sprintf("checked out before turn %d; edit and send to branch", turnNo)
1006 } else {
1007 i.ed.Clear()
1008 i.clipboardImages = nil
1009 i.statusOK = fmt.Sprintf("checked out turn %d into a new branch", turnNo)
1010 }
1011 i.statusErr = ""
1012 i.mu.Unlock()
1013 i.scrollToBottom()
1014 // Selection does not submit, start, or queue the restored user draft; only
1015 // a later explicit Enter does that. An inherited compact handoff may resume
1016 // its hidden continuation below when this branch has the complete effective
1017 // transcript, without submitting the restored draft.
1018 i.invalidate()
1019 if state.reason != compactContinuationNone {
1020 i.startRestoredCompactHandoff(i.runCtx)
1021 }
1022}
1023func (i *Interactive) applyForkSelection(msgIdx int) {
1024 i.pendingFork = false
1025 src := i.currentSessionPath()
1026 if src == "" {
1027 i.mu.Lock()
1028 i.statusErr = "fork: no session is active"
1029 i.mu.Unlock()
1030 i.invalidate()
1031 return
1032 }
1033 if i.cfg.FlushSession != nil {
1034 i.cfg.FlushSession()
1035 }
1036 // msgIdx is 0-indexed message position; copy msgIdx+1 rows so
1037 // the selected user message is included.
1038 upTo := msgIdx + 1
1039 newPath, err := core.BranchSession(src, i.sessionsRoot(), i.cfg.CWD, i.cfg.Version, upTo)
1040 if err != nil {
1041 i.mu.Lock()
1042 i.statusErr = "fork: " + err.Error()
1043 i.mu.Unlock()
1044 i.invalidate()
1045 return
1046 }
1047 if i.cfg.LoadSession == nil {
1048 i.mu.Lock()
1049 i.statusOK = "forked at message " + formatInt(upTo) + " (run /sessions to resume)"
1050 i.statusErr = ""
1051 i.mu.Unlock()
1052 i.invalidate()
1053 return
1054 }
1055 i.markSessionTitleSwitching()
1056 if err := i.cfg.LoadSession(newPath); err != nil {
1057 i.restoreFailedSessionTitle()
1058 i.mu.Lock()
1059 i.statusErr = "fork: switch failed: " + err.Error()
1060 i.mu.Unlock()
1061 i.invalidate()
1062 return
1063 }
1064 state := i.restoreCurrentCompactHandoff()
1065 i.resetSessionTitleForFreshBranch()
1066 i.mu.Lock()
1067 i.statusOK = "forked and switched to new branch at " + friendlyPath(newPath)
1068 i.statusErr = ""
1069 i.mu.Unlock()
1070 i.invalidate()
1071 if state.reason != compactContinuationNone {
1072 i.startRestoredCompactHandoff(i.runCtx)
1073 }
1074}
1075func formatInt(n int) string {
1076 return fmt.Sprintf("%d", n)
1077}
1078func assistantText(m provider.Message) string {
1079 var sb strings.Builder
1080 for _, c := range m.Content {
1081 if tb, ok := c.(provider.TextBlock); ok {
1082 if sb.Len() > 0 {
1083 sb.WriteByte('\n')
1084 }
1085 sb.WriteString(tb.Text)
1086 }
1087 }
1088 return sb.String()
1089}
1090func (i *Interactive) resetTranscriptRenderLocked() {
1091 i.view.InvalidateRenderCache()
1092 i.chatCacheValid = false
1093 i.prevChatLen = 0
1094 i.prevChatCols = 0
1095 i.prevChatRows = 0
1096 i.prevScrollOffset = 0
1097 i.requestRendererInvalidate()
1098}
1099func (i *Interactive) resetStreamingStateLocked() {
1100 i.streaming.Reset()
1101 i.streamPending = i.streamPending[:0]
1102 i.streamFlushPending = false
1103 i.streamOn = false
1104 i.openAllToolGatesLocked()
1105}
1106func (i *Interactive) openAllToolGatesLocked() {
1107 for id := range i.toolGate {
1108 i.toolGate[id] = 0
1109 }
1110}
1111func (i *Interactive) gateToolLocked(id string) {
1112 if _, ok := i.toolGate[id]; ok {
1113 return
1114 }
1115 if !i.streamOn {
1116 i.toolGate[id] = 0
1117 return
1118 }
1119 i.toolGate[id] = i.streaming.Len() + len(i.streamPending)
1120}
1121func (i *Interactive) toolGateOpenLocked(id string) bool {
1122 gate, ok := i.toolGate[id]
1123 if !ok || gate == 0 {
1124 return true
1125 }
1126 return i.streaming.Len() >= gate
1127}
1128func (i *Interactive) assistantMessageSideEffects(m provider.Message) {
1129 if i.cfg.OnAssistant != nil {
1130 i.cfg.OnAssistant(m)
1131 }
1132 if i.telegramBridge != nil && i.telegramBridge.Active() {
1133 var sb strings.Builder
1134 for _, c := range m.Content {
1135 if tb, ok := c.(provider.TextBlock); ok {
1136 if sb.Len() > 0 {
1137 sb.WriteString("\n")
1138 }
1139 sb.WriteString(tb.Text)
1140 }
1141 }
1142 if text := sb.String(); strings.TrimSpace(text) != "" {
1143 go i.telegramBridge.OnAssistantText(text)
1144 }
1145 }
1146}
1147func (i *Interactive) runStreamPacer(ctx context.Context) {
1148 t := time.NewTicker(paintPaceInterval)
1149 defer t.Stop()
1150 for {
1151 select {
1152 case <-ctx.Done():
1153 return
1154 case <-t.C:
1155 i.mu.Lock()
1156 if len(i.streamPending) == 0 {
1157 // EvAssistantMessage already fired but the pacer
1158 // was still draining a tick ago. Everything is now
1159 // painted; clear the streaming flags so the next
1160 // redraw shows the finalised transcript message
1161 // and hides the streaming overlay.
1162 if i.streamFlushPending {
1163 i.streamFlushPending = false
1164 i.streaming.Reset()
1165 i.streamOn = false
1166 i.openAllToolGatesLocked()
1167 i.mu.Unlock()
1168 i.invalidate()
1169 continue
1170 }
1171 i.mu.Unlock()
1172 continue
1173 }
1174 n := paintPaceRate
1175 if n > len(i.streamPending) {
1176 n = len(i.streamPending)
1177 }
1178 i.streaming.WriteString(string(i.streamPending[:n]))
1179 i.streamPending = i.streamPending[n:]
1180 i.mu.Unlock()
1181 i.invalidate()
1182 }
1183 }
1184}