1// Package extensions implements the host side of zut's subprocess
2// extension protocol. The Manager discovers extensions in well-known
3// directories, spawns each one, completes the hello handshake, and
4// routes slash commands to the right extension.
5//
6// Each extension is its own process, communicating with zut over its
7// stdin/stdout in newline-delimited JSON. Stderr is redirected to a
8// per-extension log file under $ZUT_HOME/logs/. Crashing one
9// extension does not affect the others or the host.
10//
11// See docs/extensions.md for the user-facing reference and
12// packages/agent/extproto for the wire format.
13package extensions
14
15import (
16 "bufio"
17 "context"
18 "encoding/json"
19 "errors"
20 "fmt"
21 "io"
22 "os"
23 "os/exec"
24 "path/filepath"
25 "slices"
26 "sort"
27 "strings"
28 "sync"
29 "sync/atomic"
30 "syscall"
31 "time"
32
33 "github.com/bnema/zut/packages/agent/extproto"
34 "github.com/bnema/zut/packages/agent/skills"
35 "github.com/bnema/zut/packages/tui"
36)
37
38// Manifest is the extension.json file shipped alongside an
39// extension's executable. It tells zut how to launch the extension
40// and provides display metadata.
41type Manifest struct {
42 Name string `json:"name"`
43 Version string `json:"version,omitempty"`
44 Exec string `json:"exec"` // executable path, relative to manifest dir
45 Args []string `json:"args,omitempty"` // extra argv passed to exec
46 Language string `json:"language,omitempty"` // informational ("go", "python", "typescript", ...)
47 Enabled *bool `json:"enabled,omitempty"` // nil = enabled
48 Description string `json:"description,omitempty"`
49 Skills []string `json:"skills,omitempty"` // bundled skill directories
50}
51
52// IsEnabled returns the manifest's effective enabled state. Default
53// is true so adding a new extension folder Just Works without an
54// extra zut ext enable command.
55func (m Manifest) IsEnabled() bool {
56 return m.Enabled == nil || *m.Enabled
57}
58
59func loadManifestSkills(extDir string, mf Manifest) ([]*skills.Skill, []error) {
60 if len(mf.Skills) == 0 {
61 return nil, nil
62 }
63 root, err := filepath.Abs(extDir)
64 if err != nil {
65 return nil, []error{fmt.Errorf("resolve extension directory: %w", err)}
66 }
67 root, err = filepath.EvalSymlinks(root)
68 if err != nil {
69 return nil, []error{fmt.Errorf("resolve extension directory: %w", err)}
70 }
71 declared := make([]skills.BundledSkillDir, 0, len(mf.Skills))
72 var errs []error
73 for _, raw := range mf.Skills {
74 rel := strings.TrimSpace(raw)
75 if rel == "" || filepath.IsAbs(rel) {
76 errs = append(errs, fmt.Errorf("skill path %q must be a relative directory", raw))
77 continue
78 }
79 clean := filepath.Clean(rel)
80 if clean == "." || clean == ".." || strings.HasPrefix(clean, ".."+string(filepath.Separator)) {
81 errs = append(errs, fmt.Errorf("skill path %q escapes the extension directory", raw))
82 continue
83 }
84 candidate := filepath.Join(root, clean)
85 resolved, err := filepath.EvalSymlinks(candidate)
86 if err != nil {
87 errs = append(errs, fmt.Errorf("skill path %q: %w", raw, err))
88 continue
89 }
90 relResolved, err := filepath.Rel(root, resolved)
91 if err != nil || relResolved == ".." || strings.HasPrefix(relResolved, ".."+string(filepath.Separator)) {
92 errs = append(errs, fmt.Errorf("skill path %q escapes the extension directory through a symlink", raw))
93 continue
94 }
95 declared = append(declared, skills.BundledSkillDir{Dir: resolved, Root: root, Source: "extension " + mf.Name})
96 }
97 loaded, loadErrs := skills.LoadBundled(declared)
98 return loaded, append(errs, loadErrs...)
99}
100
101// Extension is a running extension subprocess and the metadata zut
102// tracks about it.
103type Extension struct {
104 Manifest Manifest
105 Dir string // absolute path to extension directory
106 LogPath string
107
108 cmd *exec.Cmd
109 stdin io.WriteCloser
110 stdout io.ReadCloser
111 logFile *os.File
112 helloAck bool
113 // toolCancel records whether the extension advertised the "tool_cancel"
114 // capability in hello. Set during the handshake, before publication.
115 toolCancel bool
116 commands []extproto.RegisterCommandFromExt
117 tools []extproto.RegisterToolFromExt
118 skills []*skills.Skill
119
120 // writeMu serializes concurrent host→extension frames on plain writers.
121 // Spawned extensions use an orderedPipe, which serializes internally and
122 // bounds every synchronous write; writeMu is then only used by the
123 // lifecycle writer's lock probe.
124 writeMu sync.Mutex
125
126 // lifecycleFrames decouples session-event delivery from the host's
127 // synchronous control paths. The queue preserves lifecycle ordering while
128 // bounding backpressure from an extension that stops reading stdin.
129 lifecycleFrames chan []byte
130 lifecycleStop chan struct{}
131 lifecycleStopOnce sync.Once
132
133 // readyCh is closed when the extension sends a ReadyFromExt
134 // frame, or when the host gives up waiting (registrationGrace).
135 readyCh chan struct{}
136 readyOnce sync.Once
137
138 // pending command invocations waiting on a CommandResponseFromExt
139 // keyed by the id we sent in CommandInvokedFromHost.
140 // pendingTool is the same idea for tool calls.
141 // pendingIntercept is the same idea for event_intercept calls.
142 mu sync.Mutex
143 pending map[string]chan extproto.CommandResponseFromExt
144 pendingTool map[string]chan extproto.ToolResultFromExt
145 pendingIntercept map[string]chan extproto.EventInterceptResponseFromExt
146
147 // lastFrameTime is updated by the read loop on every frame it
148 // processes. Used by the auto-ready idle watchdog so legacy
149 // extensions (no `ready` frame) don't pin the WaitForReady wait
150 // to its full grace.
151 lastFrameTime time.Time
152
153 // eventSubs and interceptSubs are the sets of event names this
154 // extension subscribed to via SubscribeFromExt. Used by
155 // EmitEvent / InterceptToolCall to filter recipients.
156 eventSubs map[string]struct{}
157 interceptSubs map[string]struct{}
158
159 autoReady bool
160 readyTimedOut bool
161 diagnostics []string
162}
163
164// HostHooks is the small interface the manager calls back into the
165// running TUI through. Decouples extensions from interactive.go.
166type HostHooks interface {
167 // Notify pushes an ext-originated status message into the chat.
168 // Level is one of "info", "warn", "error", "success".
169 Notify(extName, level, message string)
170
171 // Submit feeds text as if the user had typed and pressed enter,
172 // running it through the agent loop.
173 Submit(text string)
174
175 // SubmitSlash runs text as a slash command in the TUI as if the
176 // user had typed it (text must start with '/'). Unlike Submit it
177 // does NOT run text through the model. Wired to the spontaneous
178 // submit_slash frame from extensions; ignored when the host is
179 // not interactive.
180 SubmitSlash(text string)
181
182 // Insert places text at the cursor in the editor.
183 Insert(text string)
184
185 // Display appends a one-shot styled note to the chat without
186 // invoking the model and without writing to the transcript.
187 Display(extName, text string)
188
189 // ClearNotes removes any notes previously pushed by extName via
190 // Notify/Display so transient status lines do not stack forever.
191 ClearNotes(extName string)
192
193 OpenPanel(extName string, spec extproto.PanelSpec)
194 UpdatePanel(extName, panelID, title string, lines []string, footer string)
195 ClosePanel(extName, panelID string)
196
197 // SetStatus and SetWidget maintain persistent extension-owned UI state.
198 // Hosts without a persistent UI may ignore these calls.
199 SetStatus(extName, key, level, text string)
200 SetWidget(extName, id, position, title string, lines []string)
201 ClearWidget(extName, id string)
202}
203
204// InteractiveToolHostHooks is an optional extension to HostHooks. A host
205// that can actually show extension panels and wait for a person returns true
206// from SupportsInteractiveTools. Hosts that do not implement it, or return
207// false, are headless: interactive tool calls fail closed with a tool error
208// instead of waiting forever on an invisible prompt.
209type InteractiveToolHostHooks interface {
210 SupportsInteractiveTools() bool
211}
212
213// supportsInteractiveTools reports whether the current host opted in.
214func (m *Manager) supportsInteractiveTools() bool {
215 if m == nil {
216 return false
217 }
218 h, ok := m.hooks.(InteractiveToolHostHooks)
219 return ok && h.SupportsInteractiveTools()
220}
221
222// AlertHostHooks is an optional extension to HostHooks. Hosts that support
223// structured alerts implement it without breaking older embedders that only
224// implement the original HostHooks surface.
225type AlertHostHooks interface {
226 Alert(extName string, alert extproto.AlertRequest)
227}
228
229// ExtensionChromeHooks is an optional lifecycle hook for hosts that retain
230// persistent extension-owned UI. The manager calls it when an extension exits
231// so a crash, disable, or reload cannot leave stale widgets or statuses.
232type ExtensionChromeHooks interface {
233 ClearExtensionChrome(extName string)
234}
235
236type commandRegistration struct {
237 ext *Extension
238 name string
239}
240
241// Manager owns every extension subprocess for the lifetime of zut.
242type Manager struct {
243 zutHome string
244 cwd string
245 zutVersion string
246 provider string
247 model string
248 hooks HostHooks
249
250 mu sync.RWMutex
251 ext map[string]*Extension // keyed by manifest name
252
253 // commandIndex maps a lower-cased slash-command name (without the
254 // leading /) to its canonical registration. First-come-first-served:
255 // later registrations that differ only by case are dropped with a warning.
256 commandIndex map[string]commandRegistration
257
258 // toolIndex maps an extension-defined tool name to its owning
259 // extension. Same first-come-first-served rule as commandIndex.
260 toolIndex map[string]*Extension
261
262 // explicitPaths remembers ad-hoc paths passed via --ext so
263 // Reload can respawn them alongside the discovered set.
264 explicitPaths []string
265
266 // onReload, if set, is invoked after a successful Reload. Used
267 // by the host so it can rebuild the agent's tool registry with
268 // the freshly-registered extension tools.
269 onReload func()
270
271 // activeSession is replayed to extensions loaded after the current
272 // session was opened, such as after /reload-ext.
273 activeSession *sessionEventState
274 sessionEmitMu sync.Mutex
275}
276
277type sessionEventState struct {
278 event extproto.EventFromHost
279 states map[string]json.RawMessage
280}
281
282// New constructs an empty Manager. Call Discover to populate it from
283// the on-disk extension directories.
284func New(zutHome, cwd, zutVersion, provider, model string, hooks HostHooks) *Manager {
285 return &Manager{
286 zutHome: zutHome,
287 cwd: cwd,
288 zutVersion: zutVersion,
289 provider: provider,
290 model: model,
291 hooks: hooks,
292 ext: map[string]*Extension{},
293 commandIndex: map[string]commandRegistration{},
294 toolIndex: map[string]*Extension{},
295 }
296}
297
298// Discover scans the global and project extension dirs and starts
299// every extension whose manifest is enabled. Spawns happen in
300// parallel so a slow runtime (e.g. `npx tsx` cold-start, ~1.5s)
301// doesn't block other extensions from starting. Returns a slice of
302// errors encountered (one per extension); a single bad extension
303// doesn't abort the rest.
304func (m *Manager) Discover(ctx context.Context) []error {
305 type loadJob struct {
306 dir string
307 }
308 var jobs []loadJob
309 seenDirs := map[string]bool{} // dedup by basename so project wins
310 for _, dir := range m.searchDirs() {
311 entries, err := os.ReadDir(dir)
312 if err != nil {
313 continue // missing directory is fine
314 }
315 for _, e := range entries {
316 if !e.IsDir() || strings.HasPrefix(e.Name(), ".") {
317 continue
318 }
319 extDir := filepath.Join(dir, e.Name())
320 if _, err := os.Stat(filepath.Join(extDir, "extension.json")); errors.Is(err, os.ErrNotExist) {
321 continue
322 }
323 if seenDirs[e.Name()] {
324 continue // higher-priority location already queued
325 }
326 seenDirs[e.Name()] = true
327 jobs = append(jobs, loadJob{dir: extDir})
328 }
329 }
330
331 var wg sync.WaitGroup
332 errCh := make(chan error, len(jobs))
333 for _, j := range jobs {
334 wg.Add(1)
335 go func(extDir string) {
336 defer wg.Done()
337 if err := m.loadOne(ctx, extDir); err != nil {
338 errCh <- err
339 }
340 }(j.dir)
341 }
342 wg.Wait()
343 close(errCh)
344
345 var errs []error
346 for e := range errCh {
347 errs = append(errs, e)
348 }
349 return errs
350}
351
352// searchDirs returns the directories the discoverer walks, in
353// priority order: project-local first (so a project can override
354// global behavior), then global.
355func (m *Manager) searchDirs() []string {
356 var dirs []string
357 if m.cwd != "" {
358 dirs = append(dirs, filepath.Join(m.cwd, ".zut", "extensions"))
359 }
360 if m.zutHome != "" {
361 dirs = append(dirs, filepath.Join(m.zutHome, "extensions"))
362 }
363 return dirs
364}
365
366// loadOne reads a single extension's manifest and, if enabled,
367// spawns its subprocess + completes the hello handshake.
368func (m *Manager) loadOne(ctx context.Context, dir string) error {
369 manifestPath := filepath.Join(dir, "extension.json")
370 raw, err := os.ReadFile(manifestPath)
371 if err != nil {
372 return fmt.Errorf("%s: read manifest: %w", dir, err)
373 }
374 var mf Manifest
375 if err := json.Unmarshal(raw, &mf); err != nil {
376 return fmt.Errorf("%s: parse manifest: %w", dir, err)
377 }
378 if mf.Name == "" {
379 return fmt.Errorf("%s: manifest: name is required", dir)
380 }
381 hasTheme := HasExtensionTheme(dir)
382 if mf.Exec == "" && !hasTheme {
383 return fmt.Errorf("%s: manifest: exec is required", dir)
384 }
385 if !mf.IsEnabled() {
386 // Quietly skip disabled extensions; zut ext list will show them.
387 return nil
388 }
389
390 m.mu.RLock()
391 _, dup := m.ext[mf.Name]
392 m.mu.RUnlock()
393 if dup {
394 // Project-local copy already won; ignore the global one.
395 return nil
396 }
397
398 bundledSkills, skillErrs := loadManifestSkills(dir, mf)
399 ext := &Extension{
400 Manifest: mf,
401 Dir: dir,
402 readyCh: make(chan struct{}),
403 pending: map[string]chan extproto.CommandResponseFromExt{},
404 pendingTool: map[string]chan extproto.ToolResultFromExt{},
405 pendingIntercept: map[string]chan extproto.EventInterceptResponseFromExt{},
406 eventSubs: map[string]struct{}{},
407 interceptSubs: map[string]struct{}{},
408 skills: bundledSkills,
409 lifecycleFrames: make(chan []byte, 64),
410 lifecycleStop: make(chan struct{}),
411 }
412 for _, skillErr := range skillErrs {
413 ext.diagnostics = append(ext.diagnostics, "skill: "+skillErr.Error())
414 }
415 if mf.Exec != "" {
416 if err := m.spawn(ctx, ext); err != nil {
417 language := ""
418 if mf.Language != "" {
419 language = fmt.Sprintf(" (declared language: %q)", mf.Language)
420 }
421 logDetail := ""
422 if ext.LogPath != "" {
423 logDetail = "\n stderr log: " + ext.LogPath
424 }
425 return fmt.Errorf("Extension %s failed to start.\n\n exec: %q%s\n error: %w\n extension directory: %s%s", mf.Name, mf.Exec, language, err, dir, logDetail)
426 }
427 } else {
428 ext.readyOnce.Do(func() { close(ext.readyCh) })
429 }
430
431 m.mu.Lock()
432 m.ext[mf.Name] = ext
433 // Note: ext.commands and ext.tools may be empty here — they're
434 // populated by the read loop as register_* frames arrive after
435 // hello. Indexing happens in the read loop too. Discover()'s
436 // caller can WaitForReady() before relying on the registries.
437 m.mu.Unlock()
438 return nil
439}
440
441// LoadExplicit loads each path as an ad-hoc extension. Used for
442// `zut --ext <path>` so extension authors can iterate on a working
443// copy without having to `zut ext install` after every change.
444//
445// Loaded BEFORE Discover so explicit paths win on name conflicts
446// against installed extensions. Spawns happen in parallel like the
447// regular discovery path; errors are returned per path.
448// HasExtensionTheme reports whether an extension directory contains a theme
449// file in one of the supported extension-owned locations.
450func HasExtensionTheme(dir string) bool {
451 for _, file := range []string{"theme.json", filepath.Join("themes", "theme.json")} {
452 if _, err := os.Stat(filepath.Join(dir, file)); err == nil {
453 return true
454 }
455 }
456 return false
457}
458
459func (m *Manager) ThemeOptions() []tui.ThemeOption {
460 m.mu.RLock()
461 defer m.mu.RUnlock()
462 names := make([]string, 0, len(m.ext))
463 for name := range m.ext {
464 names = append(names, name)
465 }
466 sort.Strings(names)
467 var out []tui.ThemeOption
468 for _, name := range names {
469 ext := m.ext[name]
470 for _, file := range []string{"theme.json", "themes/theme.json"} {
471 path := filepath.Join(ext.Dir, file)
472 value := path
473 if opt, ok := tui.ThemeOptionFromFile(path, value, "extension "+ext.Manifest.Name); ok {
474 out = append(out, opt)
475 }
476 }
477 }
478 return out
479}
480
481func (m *Manager) LoadExplicit(ctx context.Context, paths []string) []error {
482 if len(paths) == 0 {
483 return nil
484 }
485
486 var wg sync.WaitGroup
487 errCh := make(chan error, len(paths))
488 absPaths := make([]string, 0, len(paths))
489 for _, p := range paths {
490 abs, err := filepath.Abs(p)
491 if err != nil {
492 errCh <- fmt.Errorf("%s: %w", p, err)
493 continue
494 }
495 absPaths = append(absPaths, abs)
496 wg.Add(1)
497 go func(extDir string) {
498 defer wg.Done()
499 if err := m.loadOne(ctx, extDir); err != nil {
500 errCh <- err
501 }
502 }(abs)
503 }
504 wg.Wait()
505 close(errCh)
506
507 m.mu.Lock()
508 m.explicitPaths = append(m.explicitPaths, absPaths...)
509 m.mu.Unlock()
510
511 var errs []error
512 for e := range errCh {
513 errs = append(errs, e)
514 }
515 return errs
516}
517
518// SetOnReload registers a callback fired after a successful Reload.
519// Hosts use it to rebuild the agent's tool registry with freshly-
520// registered extension tools.
521func (m *Manager) SetOnReload(fn func()) {
522 m.mu.Lock()
523 m.onReload = fn
524 m.mu.Unlock()
525}
526
527// ReloadStats summarises the outcome of Reload.
528type ReloadStats struct {
529 Stopped int // how many old processes were torn down
530 Loaded int // how many new processes reached spawn
531 Ready int // how many of those signalled ready in time
532 Errors []error // non-fatal per-extension errors
533}
534
535// Reload tears down every running extension, re-reads the manifests
536// from disk, respawns everyone (including the --ext paths remembered
537// from LoadExplicit), waits up to grace for ready signals, and
538// invokes the SetOnReload callback so the host can rebuild its tool
539// registry. The manager's internal maps are cleared before the
540// new load to ensure a clean slate.
541//
542// Safe to call concurrently with normal host operations: the lock is
543// released between stop and respawn so pending InvokeTool / Invoke
544// calls on the old processes get a clean error as their stdin
545// closes.
546func (m *Manager) Reload(ctx context.Context, grace time.Duration) ReloadStats {
547 stats := ReloadStats{}
548
549 // Snapshot and remember the explicit paths before we wipe state.
550 m.mu.Lock()
551 old := m.ext
552 explicit := append([]string(nil), m.explicitPaths...)
553 stats.Stopped = len(old)
554 m.ext = map[string]*Extension{}
555 m.commandIndex = map[string]commandRegistration{}
556 m.toolIndex = map[string]*Extension{}
557 m.explicitPaths = nil
558 callback := m.onReload
559 m.mu.Unlock()
560
561 // Graceful stop of the old set (reuses Stop's shutdown logic,
562 // but Stop re-reads m.ext which is now empty, so we replicate
563 // the small shutdown loop here on the snapshot).
564 oldExts := make([]*Extension, 0, len(old))
565 for _, ext := range old {
566 oldExts = append(oldExts, ext)
567 }
568 stopExtensions(oldExts, grace)
569
570 // Fresh load. Explicit paths first (they still win on conflict).
571 if errs := m.LoadExplicit(ctx, explicit); len(errs) > 0 {
572 stats.Errors = append(stats.Errors, errs...)
573 }
574 if errs := m.Discover(ctx); len(errs) > 0 {
575 stats.Errors = append(stats.Errors, errs...)
576 }
577
578 m.mu.RLock()
579 stats.Loaded = len(m.ext)
580 m.mu.RUnlock()
581
582 // Wait for ready frames. Use the same 3s grace zut uses at
583 // startup so the reload feels no slower than a cold boot.
584 readyDeadline := time.Now().Add(grace)
585 if time.Until(readyDeadline) < 3*time.Second {
586 readyDeadline = time.Now().Add(3 * time.Second)
587 }
588 m.WaitForReady(time.Until(readyDeadline))
589
590 m.mu.RLock()
591 for _, ext := range m.ext {
592 select {
593 case <-ext.readyCh:
594 stats.Ready++
595 default:
596 }
597 }
598 m.mu.RUnlock()
599
600 if callback != nil {
601 callback()
602 }
603
604 return stats
605}
606
607// WaitForReady blocks until every loaded extension has signalled
608// ReadyFromExt, or the grace period expires for the slowest one.
609//
610// Waits run in parallel: total time is max(per-extension wait), not
611// sum. Without this, a single slow extension (e.g. `npx tsx` cold)
612// would gate every other extension's wait too and zut startup would
613// scale linearly with the number of slow runtimes installed.
614//
615// Call after Discover and before relying on tool registrations.
616func (m *Manager) WaitForReady(grace time.Duration) {
617 m.mu.RLock()
618 exts := make([]*Extension, 0, len(m.ext))
619 for _, e := range m.ext {
620 exts = append(exts, e)
621 }
622 m.mu.RUnlock()
623
624 deadline := time.After(grace)
625 var wg sync.WaitGroup
626 for _, ext := range exts {
627 wg.Add(1)
628 go func(ext *Extension) {
629 defer wg.Done()
630 select {
631 case <-ext.readyCh:
632 case <-deadline:
633 ext.mu.Lock()
634 ext.readyTimedOut = true
635 ext.diagnostics = append(ext.diagnostics, "timed out waiting for ready frame")
636 ext.mu.Unlock()
637 fmt.Fprintf(ext.logFile, "[zut] timed out waiting for ready frame; proceeding\n")
638 ext.readyOnce.Do(func() { close(ext.readyCh) })
639 }
640 }(ext)
641 }
642 wg.Wait()
643}
644
645// extensionExitGrace bounds how long startup waits for a process to exit
646// after it closes stdout before completing the hello handshake.
647const extensionExitGrace = time.Second
648
649// ResolveExecPath applies the manifest exec resolution rules. The bool is
650// true when exec names an absolute or extension-relative path; false means
651// the value is a bare command resolved through PATH.
652func ResolveExecPath(extDir, execName string) (string, bool) {
653 normalized := filepath.FromSlash(execName)
654 if filepath.IsAbs(normalized) {
655 return normalized, true
656 }
657 if normalized == "." || normalized == ".." ||
658 strings.HasPrefix(normalized, "."+string(filepath.Separator)) ||
659 strings.HasPrefix(normalized, ".."+string(filepath.Separator)) ||
660 strings.ContainsRune(normalized, filepath.Separator) {
661 return filepath.Join(extDir, normalized), true
662 }
663 return normalized, false
664}
665
666// spawn launches the subprocess, hooks up pipes, logs stderr, and
667// runs the synchronous portion of the hello handshake. Asynchronous
668// frames are processed in a goroutine started here.
669func (m *Manager) spawn(ctx context.Context, ext *Extension) error {
670 logsDir := filepath.Join(m.zutHome, "logs")
671 _ = os.MkdirAll(logsDir, 0o755)
672 logPath := filepath.Join(logsDir, "ext-"+ext.Manifest.Name+".log")
673 logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
674 if err != nil {
675 return fmt.Errorf("open log: %w", err)
676 }
677 ext.LogPath = logPath
678 ext.logFile = logFile
679 fmt.Fprintf(logFile, "\n[zut] starting %s/%s at %s\n", ext.Manifest.Name, ext.Manifest.Version, time.Now().Format(time.RFC3339))
680
681 // Bare names (node, npx, python3, tsx, etc.) are looked up via PATH;
682 // absolute and extension-relative paths are resolved by the helper.
683 execPath, _ := ResolveExecPath(ext.Dir, ext.Manifest.Exec)
684 cmd := exec.CommandContext(ctx, execPath, ext.Manifest.Args...)
685 cmd.Dir = ext.Dir
686 cmd.Stderr = logFile
687 isolateExtensionProcess(cmd)
688
689 var stdin io.WriteCloser
690 var stdout io.ReadCloser
691 started := false
692 cleanup := true
693 defer func() {
694 if !cleanup {
695 return
696 }
697 if stdin != nil {
698 _ = stdin.Close()
699 }
700 if stdout != nil {
701 _ = stdout.Close()
702 }
703 if started {
704 _ = cmd.Process.Kill()
705 _ = cmd.Wait()
706 }
707 _ = logFile.Close()
708 }()
709
710 stdin, err = cmd.StdinPipe()
711 if err != nil {
712 return fmt.Errorf("stdin pipe (stderr log: %s): %w", logPath, err)
713 }
714 stdout, err = cmd.StdoutPipe()
715 if err != nil {
716 return fmt.Errorf("stdout pipe (stderr log: %s): %w", logPath, err)
717 }
718 if err := cmd.Start(); err != nil {
719 return err
720 }
721 started = true
722 ext.cmd = cmd
723 ext.stdin = stdin
724 ext.stdout = stdout
725
726 // Hello handshake. Read the extension's HelloFromExt synchronously
727 // so we can fail fast on a broken extension; everything after is
728 // processed in the read goroutine.
729 scanner := bufio.NewScanner(stdout)
730 scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
731 if !scanner.Scan() {
732 if err := scanner.Err(); err != nil {
733 return fmt.Errorf("read hello (stderr log: %s): %w", logPath, err)
734 }
735 status, exited := waitForExtensionExit(cmd, extensionExitGrace)
736 started = false // waitForExtensionExit always reaps the process.
737 if exited {
738 return fmt.Errorf("extension exited before hello: %s (stderr log: %s)", status, logPath)
739 }
740 return fmt.Errorf("extension closed stdout before hello without exiting (stderr log: %s)", logPath)
741 }
742 var hello extproto.HelloFromExt
743 if err := json.Unmarshal(scanner.Bytes(), &hello); err != nil {
744 return fmt.Errorf("parse hello (stderr log: %s): %w", logPath, err)
745 }
746 if hello.Type != "hello" || hello.Name == "" {
747 return fmt.Errorf("first frame must be hello (got %q; stderr log: %s)", hello.Type, logPath)
748 }
749 if hello.ProtocolVersion != extproto.ProtocolVersion {
750 return fmt.Errorf("unsupported extension protocol version %d (want %d; stderr log: %s)", hello.ProtocolVersion, extproto.ProtocolVersion, logPath)
751 }
752 // Trust the manifest's name; ignore mismatch from the hello.
753 ext.helloAck = true
754 ext.toolCancel = slices.Contains(hello.Capabilities, "tool_cancel")
755
756 m.mu.RLock()
757 var activeSession *extproto.SessionContext
758 if m.activeSession != nil {
759 activeSession = cloneSessionContext(m.activeSession.event.Session)
760 }
761 m.mu.RUnlock()
762 ack, _ := extproto.Encode(extproto.HelloAckFromHost{
763 Type: "hello_ack",
764 ProtocolVersion: extproto.ProtocolVersion,
765 ZutVersion: m.zutVersion,
766 Provider: m.provider,
767 Model: m.model,
768 CWD: m.cwd,
769 ExtensionDir: ext.Dir,
770 DataDir: ext.Dir,
771 Session: activeSession,
772 })
773 if _, err := stdin.Write(ack); err != nil {
774 return fmt.Errorf("send hello_ack (stderr log: %s): %w", logPath, err)
775 }
776
777 // Every frame after the handshake goes through the bounded, ordered
778 // transport so a subprocess that stops reading stdin cannot pin host
779 // goroutines indefinitely.
780 ext.stdin = newOrderedPipe(stdin)
781
782 cleanup = false
783
784 // Spin up the lifecycle writer and read loop now that the handshake is done.
785 ext.startLifecycleWriter()
786 go m.readLoop(ext, scanner)
787
788 // Compatibility shim: extensions built against the phase-1 SDK
789 // don't send a ready frame. Watch the read loop's frame arrival
790 // rate; if nothing's arrived for readyIdleWindow we treat the
791 // extension as ready so WaitForReady doesn't burn the full grace
792 // on every startup. Newer extensions still trigger the explicit
793 // path on their own ready frame.
794 go m.assumeReadyAfterIdle(ext)
795
796 return nil
797}
798
799// waitForExtensionExit returns the process status when it exits promptly.
800// A process that closes stdout but remains alive has already violated the
801// protocol, so kill and reap it rather than blocking extension discovery.
802func waitForExtensionExit(cmd *exec.Cmd, grace time.Duration) (string, bool) {
803 done := make(chan error, 1)
804 go func() { done <- cmd.Wait() }()
805
806 timer := time.NewTimer(grace)
807 defer timer.Stop()
808 select {
809 case err := <-done:
810 if cmd.ProcessState != nil {
811 return cmd.ProcessState.String(), true
812 }
813 if err != nil {
814 return err.Error(), true
815 }
816 return "exit status 0", true
817 case <-timer.C:
818 _ = cmd.Process.Kill()
819 <-done
820 return "", false
821 }
822}
823
824// readyIdleWindow is how long the manager waits for a frame after
825// hello before assuming an extension that doesn't send `ready` is
826// nevertheless ready. 250ms is enough for any well-behaved native
827// binary to flush its register frames; slow runtimes (npx tsx) flush
828// even faster once they've started, so this rarely affects them.
829const readyIdleWindow = 250 * time.Millisecond
830
831func (m *Manager) assumeReadyAfterIdle(ext *Extension) {
832 ext.mu.Lock()
833 last := ext.lastFrameTime
834 ext.mu.Unlock()
835 for {
836 select {
837 case <-ext.readyCh:
838 return
839 case <-time.After(readyIdleWindow):
840 }
841 ext.mu.Lock()
842 current := ext.lastFrameTime
843 ext.mu.Unlock()
844 if current.Equal(last) {
845 // No new frame in the idle window. Treat as ready.
846 ext.readyOnce.Do(func() {
847 ext.mu.Lock()
848 ext.autoReady = true
849 ext.diagnostics = append(ext.diagnostics, "no ready frame; auto-ready after idle")
850 ext.mu.Unlock()
851 fmt.Fprintf(ext.logFile, "[zut] no ready frame; auto-readying after idle (legacy SDK?)\n")
852 close(ext.readyCh)
853 })
854 return
855 }
856 last = current
857 }
858}
859
860// readLoop processes every frame the extension sends after hello.
861// Returns when stdout closes.
862func (m *Manager) readLoop(ext *Extension, scanner *bufio.Scanner) {
863 defer func() {
864 // The peer is gone: disconnect the transport so every pending
865 // synchronous call (including deadline-free interactive tools)
866 // observes the disconnect instead of waiting for a reply that
867 // can never arrive.
868 if ext.stdin != nil {
869 _ = ext.stdin.Close()
870 }
871 // On close, drop every command + tool this extension owned so
872 // future invocations don't dangle. The subprocess is gone; we
873 // won't hear back about its commands or tool calls anymore.
874 m.mu.Lock()
875 for name, registration := range m.commandIndex {
876 if registration.ext == ext {
877 delete(m.commandIndex, name)
878 }
879 }
880 for name, owner := range m.toolIndex {
881 if owner == ext {
882 delete(m.toolIndex, name)
883 }
884 }
885 m.mu.Unlock()
886 if hooks, ok := m.hooks.(ExtensionChromeHooks); ok {
887 hooks.ClearExtensionChrome(ext.Manifest.Name)
888 }
889 ext.stopLifecycleWriter()
890 ext.readyOnce.Do(func() { close(ext.readyCh) })
891 fmt.Fprintf(ext.logFile, "[zut] extension %s read loop exited at %s\n", ext.Manifest.Name, time.Now().Format(time.RFC3339))
892 }()
893
894 for scanner.Scan() {
895 line := scanner.Bytes()
896 ext.mu.Lock()
897 ext.lastFrameTime = time.Now()
898 ext.mu.Unlock()
899 var frame extproto.Frame
900 if err := json.Unmarshal(line, &frame); err != nil {
901 ext.recordDiagnostic(fmt.Sprintf("malformed json frame: %v", err))
902 fmt.Fprintf(ext.logFile, "[zut] malformed json from extension: %v\n", err)
903 continue
904 }
905 switch frame.Type {
906 case "register_command":
907 var rc extproto.RegisterCommandFromExt
908 if err := json.Unmarshal(line, &rc); err == nil {
909 // Both ext.commands and m.commandIndex are read by
910 // the public Commands() / HasCommand() helpers under
911 // m.mu, so the writes have to take the same lock to
912 // keep the race detector happy.
913 m.mu.Lock()
914 ext.commands = append(ext.commands, rc)
915 key := strings.ToLower(rc.Name)
916 _, shadowed := m.commandIndex[key]
917 if !shadowed {
918 m.commandIndex[key] = commandRegistration{ext: ext, name: rc.Name}
919 }
920 m.mu.Unlock()
921 if shadowed {
922 ext.recordDiagnostic(fmt.Sprintf("command /%s conflicts with another extension registering the same name", rc.Name))
923 }
924 }
925 case "register_tool":
926 var rt extproto.RegisterToolFromExt
927 if err := json.Unmarshal(line, &rt); err != nil {
928 fmt.Fprintf(ext.logFile, "[zut] bad register_tool frame: %v\n", err)
929 continue
930 }
931 // Validate the schema parses as JSON. If not, refuse to
932 // register — a broken schema confuses the model.
933 if len(rt.Schema) > 0 {
934 var tmp any
935 if err := json.Unmarshal(rt.Schema, &tmp); err != nil {
936 ext.recordDiagnostic(fmt.Sprintf("tool %q schema is not valid json: %v", rt.Name, err))
937 fmt.Fprintf(ext.logFile, "[zut] tool %q: schema is not valid json (%v); skipped\n", rt.Name, err)
938 continue
939 }
940 }
941 m.mu.Lock()
942 ext.tools = append(ext.tools, rt)
943 _, shadowed := m.toolIndex[rt.Name]
944 if !shadowed {
945 m.toolIndex[rt.Name] = ext
946 }
947 m.mu.Unlock()
948 if shadowed {
949 ext.recordDiagnostic(fmt.Sprintf("tool %s conflicts with another extension registering the same name", rt.Name))
950 }
951 case "ready":
952 ext.readyOnce.Do(func() { close(ext.readyCh) })
953 m.replaySessionEvent(ext)
954 case "subscribe":
955 var sub extproto.SubscribeFromExt
956 if err := json.Unmarshal(line, &sub); err == nil {
957 ext.mu.Lock()
958 for _, ev := range sub.Events {
959 ext.eventSubs[ev] = struct{}{}
960 }
961 for _, ev := range sub.Intercept {
962 switch ev {
963 case "tool_call", "turn_start", "assistant_message":
964 ext.interceptSubs[ev] = struct{}{}
965 }
966 }
967 ext.mu.Unlock()
968 }
969 case "event_intercept_response":
970 var er extproto.EventInterceptResponseFromExt
971 if err := json.Unmarshal(line, &er); err == nil {
972 ext.mu.Lock()
973 ch, ok := ext.pendingIntercept[er.ID]
974 if ok {
975 delete(ext.pendingIntercept, er.ID)
976 }
977 ext.mu.Unlock()
978 if ok {
979 select {
980 case ch <- er:
981 default:
982 }
983 }
984 }
985 case "tool_result":
986 var tr extproto.ToolResultFromExt
987 if err := json.Unmarshal(line, &tr); err == nil {
988 ext.mu.Lock()
989 ch, ok := ext.pendingTool[tr.ID]
990 if ok {
991 delete(ext.pendingTool, tr.ID)
992 }
993 ext.mu.Unlock()
994 if ok {
995 select {
996 case ch <- tr:
997 default:
998 }
999 }
1000 }
1001 case "notify":
1002 var n extproto.NotifyFromExt
1003 if err := json.Unmarshal(line, &n); err == nil && m.hooks != nil {
1004 m.hooks.Notify(ext.Manifest.Name, n.Level, n.Message)
1005 }
1006 case "status":
1007 var s extproto.StatusFromExt
1008 if err := json.Unmarshal(line, &s); err == nil && m.hooks != nil {
1009 m.hooks.SetStatus(ext.Manifest.Name, s.Key, s.Level, s.Text)
1010 }
1011 case "widget":
1012 var w extproto.WidgetFromExt
1013 if err := json.Unmarshal(line, &w); err == nil && m.hooks != nil {
1014 m.hooks.SetWidget(ext.Manifest.Name, w.ID, w.Position, w.Title, w.Lines)
1015 }
1016 case "widget_clear":
1017 var w extproto.ClearWidgetFromExt
1018 if err := json.Unmarshal(line, &w); err == nil && m.hooks != nil {
1019 m.hooks.ClearWidget(ext.Manifest.Name, w.ID)
1020 }
1021 case "alert":
1022 var a extproto.AlertFromExt
1023 if err := json.Unmarshal(line, &a); err == nil && m.hooks != nil {
1024 if hooks, ok := m.hooks.(AlertHostHooks); ok {
1025 hooks.Alert(ext.Manifest.Name, a.AlertRequest)
1026 }
1027 }
1028 case "clear_notes":
1029 if m.hooks != nil {
1030 m.hooks.ClearNotes(ext.Manifest.Name)
1031 }
1032 case "submit":
1033 var s extproto.SubmitFromExt
1034 if err := json.Unmarshal(line, &s); err == nil && m.hooks != nil {
1035 text := strings.TrimSpace(s.Text)
1036 if text != "" {
1037 m.hooks.Submit(text)
1038 }
1039 }
1040 case "submit_slash":
1041 // Spontaneous request to invoke a slash command in the
1042 // TUI. Refused unless the payload looks like a slash
1043 // command so a misbehaving extension can't sneak a model
1044 // prompt through this path.
1045 var s extproto.SubmitSlashFromExt
1046 if err := json.Unmarshal(line, &s); err == nil {
1047 text := strings.TrimSpace(s.Text)
1048 if strings.HasPrefix(text, "/") && m.hooks != nil {
1049 m.hooks.SubmitSlash(text)
1050 } else {
1051 fmt.Fprintf(ext.logFile, "[zut] submit_slash refused (not a slash command): %q\n", s.Text)
1052 }
1053 }
1054 case "command_response":
1055 var cr extproto.CommandResponseFromExt
1056 if err := json.Unmarshal(line, &cr); err == nil {
1057 ext.mu.Lock()
1058 ch, ok := ext.pending[cr.ID]
1059 if ok {
1060 delete(ext.pending, cr.ID)
1061 }
1062 ext.mu.Unlock()
1063 if ok {
1064 select {
1065 case ch <- cr:
1066 default:
1067 }
1068 }
1069 }
1070 case "open_panel":
1071 var op extproto.OpenPanelFromExt
1072 if err := json.Unmarshal(line, &op); err == nil && m.hooks != nil {
1073 m.hooks.OpenPanel(ext.Manifest.Name, op.Panel)
1074 }
1075 case "panel_render":
1076 var pr extproto.PanelRenderFromExt
1077 if err := json.Unmarshal(line, &pr); err == nil && m.hooks != nil {
1078 m.hooks.UpdatePanel(ext.Manifest.Name, pr.PanelID, pr.Title, pr.Lines, pr.Footer)
1079 }
1080 case "panel_close":
1081 var pc extproto.PanelCloseFromExt
1082 if err := json.Unmarshal(line, &pc); err == nil && m.hooks != nil {
1083 m.hooks.ClosePanel(ext.Manifest.Name, pc.PanelID)
1084 }
1085 case "shutdown_ack":
1086 // Caller of Stop is waiting on the process exit, not this frame.
1087 default:
1088 ext.recordDiagnostic(fmt.Sprintf("unknown frame type %q", frame.Type))
1089 fmt.Fprintf(ext.logFile, "[zut] unknown frame type %q\n", frame.Type)
1090 }
1091 }
1092}
1093
1094const lifecycleWriteGrace = 250 * time.Millisecond
1095
1096// writeFrame sends one synchronous host frame. On the ordered transport the
1097// write is serialized and bounded by transportWriteTimeout; on a plain
1098// writer (tests, legacy embedders) writeMu provides the serialization.
1099func (ext *Extension) writeFrame(frame []byte) (int, error) {
1100 if ext == nil || ext.stdin == nil {
1101 return 0, fmt.Errorf("extension stdin is closed")
1102 }
1103 if pipe, ok := ext.stdin.(*orderedPipe); ok {
1104 return pipe.Write(frame)
1105 }
1106 ext.writeMu.Lock()
1107 defer ext.writeMu.Unlock()
1108 return ext.stdin.Write(frame)
1109}
1110
1111// orderedStdin returns the bounded transport when the extension was spawned
1112// by this manager. The bool is false for plain writers.
1113func (ext *Extension) orderedStdin() (*orderedPipe, bool) {
1114 if ext == nil || ext.stdin == nil {
1115 return nil, false
1116 }
1117 pipe, ok := ext.stdin.(*orderedPipe)
1118 return pipe, ok
1119}
1120
1121// tryWriteLifecycleFrame bounds a lifecycle writer's wait for writeMu; a
1122// snapshot that cannot acquire it is dropped. Frames are serialized and bounded
1123// by the ordered transport: a blocked pipe write still triggers its timeout
1124// and disconnects the extension.
1125func (ext *Extension) tryWriteLifecycleFrame(frame []byte) error {
1126 if ext == nil || ext.stdin == nil {
1127 return fmt.Errorf("extension stdin is closed")
1128 }
1129 if !ext.writeMu.TryLock() {
1130 timer := time.NewTimer(lifecycleWriteGrace)
1131 defer timer.Stop()
1132 ticker := time.NewTicker(5 * time.Millisecond)
1133 defer ticker.Stop()
1134 for {
1135 select {
1136 case <-timer.C:
1137 return fmt.Errorf("extension lifecycle write is blocked")
1138 case <-ticker.C:
1139 if ext.writeMu.TryLock() {
1140 goto locked
1141 }
1142 }
1143 }
1144 }
1145locked:
1146 defer ext.writeMu.Unlock()
1147 _, err := ext.stdin.Write(frame)
1148 return err
1149}
1150
1151func (ext *Extension) startLifecycleWriter() {
1152 if ext.lifecycleFrames == nil {
1153 ext.lifecycleFrames = make(chan []byte, 64)
1154 }
1155 if ext.lifecycleStop == nil {
1156 ext.lifecycleStop = make(chan struct{})
1157 }
1158 go ext.lifecycleWriter()
1159}
1160
1161func (ext *Extension) stopLifecycleWriter() {
1162 if ext == nil || ext.lifecycleStop == nil {
1163 return
1164 }
1165 ext.lifecycleStopOnce.Do(func() { close(ext.lifecycleStop) })
1166}
1167
1168func (ext *Extension) enqueueLifecycleFrame(frame []byte) bool {
1169 if ext == nil {
1170 return false
1171 }
1172 if ext.lifecycleFrames == nil || ext.lifecycleStop == nil {
1173 _, err := ext.writeFrame(frame)
1174 return err == nil
1175 }
1176 copyFrame := append([]byte(nil), frame...)
1177 select {
1178 case <-ext.lifecycleStop:
1179 return false
1180 default:
1181 }
1182 select {
1183 case ext.lifecycleFrames <- copyFrame:
1184 return true
1185 default:
1186 // Prefer the newest lifecycle snapshot when a broken extension stops
1187 // draining its pipe. The latest event still carries the current state.
1188 select {
1189 case <-ext.lifecycleFrames:
1190 default:
1191 }
1192 select {
1193 case ext.lifecycleFrames <- copyFrame:
1194 return true
1195 default:
1196 return false
1197 }
1198 }
1199}
1200
1201func (ext *Extension) lifecycleWriter() {
1202 for {
1203 select {
1204 case <-ext.lifecycleStop:
1205 return
1206 case frame := <-ext.lifecycleFrames:
1207 if err := ext.tryWriteLifecycleFrame(frame); err != nil {
1208 return
1209 }
1210 }
1211 }
1212}
1213
1214func (ext *Extension) recordDiagnostic(msg string) {
1215 ext.mu.Lock()
1216 defer ext.mu.Unlock()
1217 ext.diagnostics = append(ext.diagnostics, msg)
1218}
1219
1220// RegisteredCommandDiagnostic describes one command an extension attempted
1221// to register and whether it owns the active dispatch slot.
1222type RegisteredCommandDiagnostic struct {
1223 Name string
1224 Description string
1225 Active bool
1226}
1227
1228// RegisteredToolDiagnostic describes one tool an extension attempted to
1229// register and whether it owns the active dispatch slot.
1230type RegisteredToolDiagnostic struct {
1231 Name string
1232 Description string
1233 Active bool
1234}
1235
1236// ExtensionDiagnostic is a stable snapshot for diagnostic commands.
1237type ExtensionDiagnostic struct {
1238 Name string
1239 Version string
1240 Dir string
1241 LogPath string
1242 ThemeOnly bool
1243 Ready bool
1244 AutoReady bool
1245 ReadyTimedOut bool
1246 Commands []RegisteredCommandDiagnostic
1247 Tools []RegisteredToolDiagnostic
1248 Messages []string
1249}
1250
1251// Diagnostics returns a snapshot of loaded extension state for commands such
1252// as `zut ext doctor`. It is read-only and does not change manager behavior.
1253func (m *Manager) Diagnostics() []ExtensionDiagnostic {
1254 m.mu.RLock()
1255 defer m.mu.RUnlock()
1256 names := make([]string, 0, len(m.ext))
1257 for name := range m.ext {
1258 names = append(names, name)
1259 }
1260 sort.Strings(names)
1261 out := make([]ExtensionDiagnostic, 0, len(names))
1262 for _, name := range names {
1263 ext := m.ext[name]
1264 ext.mu.Lock()
1265 msgs := append([]string(nil), ext.diagnostics...)
1266 autoReady := ext.autoReady
1267 readyTimedOut := ext.readyTimedOut
1268 ext.mu.Unlock()
1269
1270 diag := ExtensionDiagnostic{
1271 Name: ext.Manifest.Name,
1272 Version: ext.Manifest.Version,
1273 Dir: ext.Dir,
1274 LogPath: ext.LogPath,
1275 ThemeOnly: ext.Manifest.Exec == "",
1276 AutoReady: autoReady,
1277 ReadyTimedOut: readyTimedOut,
1278 Messages: msgs,
1279 }
1280 select {
1281 case <-ext.readyCh:
1282 diag.Ready = true
1283 default:
1284 }
1285 for _, c := range ext.commands {
1286 registration, ok := m.commandIndex[strings.ToLower(c.Name)]
1287 diag.Commands = append(diag.Commands, RegisteredCommandDiagnostic{
1288 Name: c.Name,
1289 Description: c.Description,
1290 Active: ok && registration.ext == ext && registration.name == c.Name,
1291 })
1292 }
1293 for _, t := range ext.tools {
1294 diag.Tools = append(diag.Tools, RegisteredToolDiagnostic{
1295 Name: t.Name,
1296 Description: t.Description,
1297 Active: m.toolIndex[t.Name] == ext,
1298 })
1299 }
1300 out = append(out, diag)
1301 }
1302 return out
1303}
1304
1305// Commands returns a snapshot of every (extension, command) pair
1306// currently registered. Used by the slash autocomplete + /help.
1307func (m *Manager) Commands() []CommandInfo {
1308 m.mu.RLock()
1309 defer m.mu.RUnlock()
1310 var out []CommandInfo
1311 for _, ext := range m.ext {
1312 for _, c := range ext.commands {
1313 registration, ok := m.commandIndex[strings.ToLower(c.Name)]
1314 if !ok || registration.ext != ext || registration.name != c.Name {
1315 continue
1316 }
1317 out = append(out, CommandInfo{
1318 Extension: ext.Manifest.Name,
1319 Name: c.Name,
1320 Description: c.Description,
1321 })
1322 }
1323 }
1324 return out
1325}
1326
1327// CommandInfo is one extension-registered slash command, surfaced to
1328// the rest of zut for display purposes.
1329type CommandInfo struct {
1330 Extension string
1331 Name string
1332 Description string
1333}
1334
1335// ToolInfo is one extension-registered tool. Used by the agent's
1336// build step to materialise core.Tool wrappers.
1337type ToolInfo struct {
1338 Extension string
1339 Name string
1340 Description string
1341 Schema json.RawMessage
1342 Deferred bool
1343 // Interactive tools wait for user input and carry no reply deadline.
1344 Interactive bool
1345}
1346
1347// Tools returns a snapshot of every (extension, tool) pair currently
1348// registered. Used at agent-build time to fold extension tools into
1349// the runtime tool registry.
1350func (m *Manager) Tools() []ToolInfo {
1351 m.mu.RLock()
1352 defer m.mu.RUnlock()
1353 var out []ToolInfo
1354 for _, ext := range m.ext {
1355 for _, t := range ext.tools {
1356 out = append(out, ToolInfo{
1357 Extension: ext.Manifest.Name,
1358 Name: t.Name,
1359 Description: t.Description,
1360 Schema: t.Schema,
1361 Deferred: t.Deferred,
1362 Interactive: t.Interactive,
1363 })
1364 }
1365 }
1366 return out
1367}
1368
1369// Skills returns extension-bundled skills in deterministic extension-name
1370// order. The build layer merges them after user skills and before built-ins.
1371func (m *Manager) Skills() []*skills.Skill {
1372 m.mu.RLock()
1373 names := make([]string, 0, len(m.ext))
1374 for name := range m.ext {
1375 names = append(names, name)
1376 }
1377 sort.Strings(names)
1378 var out []*skills.Skill
1379 for _, name := range names {
1380 ext := m.ext[name]
1381 out = append(out, ext.skills...)
1382 }
1383 m.mu.RUnlock()
1384 return out
1385}
1386
1387// HasTool reports whether name is registered by any extension.
1388func (m *Manager) HasTool(name string) bool {
1389 m.mu.RLock()
1390 defer m.mu.RUnlock()
1391 _, ok := m.toolIndex[name]
1392 return ok
1393}
1394
1395// InvokeTool sends a tool_call to the owning extension and waits for
1396// the matching tool_result. Used by the core.Tool wrapper that the
1397// agent registers per extension-defined tool.
1398//
1399// timeout bounds the wait for the reply; timeout <= 0 means no reply
1400// deadline (interactive tools). Regardless of timeout, the request write is
1401// bounded by the transport, and context cancellation, context deadlines, and
1402// an extension disconnect always end the call. When the call is abandoned
1403// after the request was delivered, a best-effort tool_cancel is sent to
1404// extensions that advertised the "tool_cancel" capability. A late
1405// tool_result for an abandoned id is ignored.
1406func (m *Manager) InvokeTool(ctx context.Context, name string, args json.RawMessage, timeout time.Duration) (extproto.ToolResultFromExt, error) {
1407 if err := ctx.Err(); err != nil {
1408 return extproto.ToolResultFromExt{}, err
1409 }
1410 m.mu.RLock()
1411 ext, ok := m.toolIndex[name]
1412 m.mu.RUnlock()
1413 if !ok {
1414 return extproto.ToolResultFromExt{}, fmt.Errorf("no extension registered for tool %q", name)
1415 }
1416
1417 id := newCorrelationID()
1418 ch := make(chan extproto.ToolResultFromExt, 1)
1419 ext.mu.Lock()
1420 ext.pendingTool[id] = ch
1421 ext.mu.Unlock()
1422 defer func() {
1423 ext.mu.Lock()
1424 delete(ext.pendingTool, id)
1425 ext.mu.Unlock()
1426 }()
1427
1428 frame, _ := extproto.Encode(extproto.ToolCallFromHost{
1429 Type: "tool_call",
1430 ID: id,
1431 Name: name,
1432 Args: args,
1433 })
1434 var disconnected <-chan struct{}
1435 var err error
1436 pipe, ordered := ext.orderedStdin()
1437 if ordered {
1438 disconnected = pipe.Done()
1439 _, err = pipe.writeContext(ctx, frame, transportWriteTimeout)
1440 } else {
1441 _, err = ext.writeFrame(frame)
1442 }
1443 if err != nil {
1444 return extproto.ToolResultFromExt{}, fmt.Errorf("write: %w", err)
1445 }
1446
1447 // cancelRemote tells a capable extension to abandon the delivered
1448 // invocation. Enqueue only: the caller is already leaving and must not
1449 // block on a stalled pipe.
1450 cancelRemote := func() {
1451 if !ordered || !ext.toolCancel {
1452 return
1453 }
1454 cancelFrame, encErr := extproto.Encode(extproto.ToolCancelFromHost{Type: "tool_cancel", ID: id})
1455 if encErr != nil {
1456 return
1457 }
1458 _, _ = pipe.enqueue(cancelFrame, nil)
1459 }
1460
1461 var deadline <-chan time.Time
1462 if timeout > 0 {
1463 timer := time.NewTimer(timeout)
1464 defer timer.Stop()
1465 deadline = timer.C
1466 }
1467 select {
1468 case resp := <-ch:
1469 return resp, nil
1470 case <-deadline:
1471 cancelRemote()
1472 return extproto.ToolResultFromExt{}, fmt.Errorf("timeout waiting for %s/%s: %w", ext.Manifest.Name, name, context.DeadlineExceeded)
1473 case <-ctx.Done():
1474 cancelRemote()
1475 return extproto.ToolResultFromExt{}, ctx.Err()
1476 case <-disconnected:
1477 // readLoop may have delivered a result immediately before EOF. Prefer
1478 // that completed reply over the simultaneously ready disconnect.
1479 select {
1480 case resp := <-ch:
1481 return resp, nil
1482 default:
1483 }
1484 return extproto.ToolResultFromExt{}, fmt.Errorf("extension %s disconnected", ext.Manifest.Name)
1485 }
1486}
1487
1488// HasCommand reports whether name is registered by any extension.
1489func (m *Manager) HasCommand(name string) bool {
1490 m.mu.RLock()
1491 defer m.mu.RUnlock()
1492 _, ok := m.commandIndex[strings.ToLower(name)]
1493 return ok
1494}
1495
1496func (m *Manager) CommandOwner(name string) string {
1497 m.mu.RLock()
1498 defer m.mu.RUnlock()
1499 if registration, ok := m.commandIndex[strings.ToLower(name)]; ok && registration.ext != nil {
1500 return registration.ext.Manifest.Name
1501 }
1502 return ""
1503}
1504
1505// Invoke fires the named slash command's handler in the owning
1506// extension and waits up to timeout for the response. Returns the
1507// extension's CommandResponse so the caller can act on the action
1508// (prompt / insert / display).
1509func (m *Manager) Invoke(ctx context.Context, name, args string, timeout time.Duration) (extproto.CommandResponseFromExt, error) {
1510 m.mu.RLock()
1511 registration, ok := m.commandIndex[strings.ToLower(name)]
1512 m.mu.RUnlock()
1513 if !ok {
1514 return extproto.CommandResponseFromExt{}, fmt.Errorf("no extension registered for /%s", name)
1515 }
1516 ext := registration.ext
1517 name = registration.name
1518
1519 id := newCorrelationID()
1520 ch := make(chan extproto.CommandResponseFromExt, 1)
1521 ext.mu.Lock()
1522 ext.pending[id] = ch
1523 ext.mu.Unlock()
1524
1525 frame, _ := extproto.Encode(extproto.CommandInvokedFromHost{
1526 Type: "command_invoked",
1527 ID: id,
1528 Name: name,
1529 Args: args,
1530 })
1531 if _, err := ext.writeFrame(frame); err != nil {
1532 ext.mu.Lock()
1533 delete(ext.pending, id)
1534 ext.mu.Unlock()
1535 return extproto.CommandResponseFromExt{}, fmt.Errorf("write: %w", err)
1536 }
1537
1538 select {
1539 case resp := <-ch:
1540 return resp, nil
1541 case <-time.After(timeout):
1542 ext.mu.Lock()
1543 delete(ext.pending, id)
1544 ext.mu.Unlock()
1545 return extproto.CommandResponseFromExt{}, fmt.Errorf("timeout waiting for %s/%s", ext.Manifest.Name, name)
1546 case <-ctx.Done():
1547 ext.mu.Lock()
1548 delete(ext.pending, id)
1549 ext.mu.Unlock()
1550 return extproto.CommandResponseFromExt{}, ctx.Err()
1551 }
1552}
1553
1554// Stop cleanly terminates every extension. Sends ShutdownFromHost,
1555// waits up to gracePeriod for each subprocess to exit, then SIGTERMs
1556// (and SIGKILLs after another second) the holdouts.
1557func (m *Manager) SendPanelKey(extName, panelID, key, text string) error {
1558 m.mu.RLock()
1559 ext, ok := m.ext[extName]
1560 m.mu.RUnlock()
1561 if !ok {
1562 return fmt.Errorf("no extension %q", extName)
1563 }
1564 frame, _ := extproto.Encode(extproto.PanelKeyFromHost{Type: "panel_key", PanelID: panelID, Key: key, Text: text})
1565 _, err := ext.writeFrame(frame)
1566 return err
1567}
1568
1569func (m *Manager) SendPanelClose(extName, panelID string) error {
1570 m.mu.RLock()
1571 ext, ok := m.ext[extName]
1572 m.mu.RUnlock()
1573 if !ok {
1574 return fmt.Errorf("no extension %q", extName)
1575 }
1576 frame, _ := extproto.Encode(extproto.PanelCloseFromHost{Type: "panel_close", PanelID: panelID})
1577 _, err := ext.writeFrame(frame)
1578 return err
1579}
1580
1581func (m *Manager) Stop(gracePeriod time.Duration) {
1582 m.mu.RLock()
1583 exts := make([]*Extension, 0, len(m.ext))
1584 for _, e := range m.ext {
1585 exts = append(exts, e)
1586 }
1587 m.mu.RUnlock()
1588 stopExtensions(exts, gracePeriod)
1589}
1590
1591func stopExtensions(exts []*Extension, gracePeriod time.Duration) {
1592 deadline := time.Now().Add(gracePeriod)
1593 for _, ext := range exts {
1594 if ext.stdin == nil {
1595 continue
1596 }
1597 if frame, err := extproto.Encode(extproto.ShutdownFromHost{Type: "shutdown"}); err == nil {
1598 if pipe, ok := ext.orderedStdin(); ok {
1599 // Never wait longer than the shutdown grace for a peer that
1600 // stopped reading; the process is killed below anyway.
1601 wait := time.Until(deadline)
1602 if wait > transportWriteTimeout {
1603 wait = transportWriteTimeout
1604 }
1605 if wait < 50*time.Millisecond {
1606 wait = 50 * time.Millisecond
1607 }
1608 _, _ = pipe.writeContext(context.Background(), frame, wait)
1609 } else {
1610 _, _ = ext.writeFrame(frame)
1611 }
1612 }
1613 _ = ext.stdin.Close()
1614 }
1615
1616 for _, ext := range exts {
1617 if ext.cmd == nil {
1618 if ext.logFile != nil {
1619 _ = ext.logFile.Close()
1620 }
1621 continue
1622 }
1623 remaining := time.Until(deadline)
1624 if remaining <= 0 {
1625 remaining = 100 * time.Millisecond
1626 }
1627 done := make(chan struct{})
1628 go func() { _ = ext.cmd.Wait(); close(done) }()
1629 select {
1630 case <-done:
1631 case <-time.After(remaining):
1632 _ = ext.cmd.Process.Signal(syscall.SIGTERM)
1633 select {
1634 case <-done:
1635 case <-time.After(time.Second):
1636 _ = ext.cmd.Process.Kill()
1637 <-done
1638 }
1639 }
1640 if ext.logFile != nil {
1641 _ = ext.logFile.Close()
1642 }
1643 }
1644}
1645
1646// All returns every extension currently tracked, enabled or not.
1647// Used by `zut ext list`.
1648func (m *Manager) All() []*Extension {
1649 m.mu.RLock()
1650 defer m.mu.RUnlock()
1651 out := make([]*Extension, 0, len(m.ext))
1652 for _, e := range m.ext {
1653 out = append(out, e)
1654 }
1655 return out
1656}
1657
1658var correlationSequence atomic.Uint64
1659
1660// newCorrelationID returns an id that is unique within this host process.
1661// Two invocations issued in the same microsecond must never share a pending
1662// slot, so a monotonic counter replaces the former clock-derived id.
1663func newCorrelationID() string {
1664 return fmt.Sprintf("c%d", correlationSequence.Add(1))
1665}