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