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