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