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