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