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