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