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