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