1package extensions
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "time"
8
9 "github.com/patriceckhart/zot/internal/extproto"
10)
11
12// EmitEvent fires a one-way lifecycle event to every extension that
13// subscribed to it via SubscribeFromExt.events. Non-blocking: each
14// extension's pipe write happens on a per-call goroutine so a slow
15// extension can't stall the agent loop.
16//
17// Event names are documented on extproto.EventFromHost. Unknown event
18// names are still routed (subscribers can use any string they want).
19func (m *Manager) EmitEvent(ev extproto.EventFromHost) {
20 ev.Type = "event"
21 frame, err := extproto.Encode(ev)
22 if err != nil {
23 return
24 }
25 m.mu.RLock()
26 subs := make([]*Extension, 0, len(m.ext))
27 for _, ext := range m.ext {
28 ext.mu.Lock()
29 _, subscribed := ext.eventSubs[ev.Event]
30 ext.mu.Unlock()
31 if subscribed {
32 subs = append(subs, ext)
33 }
34 }
35 m.mu.RUnlock()
36
37 for _, ext := range subs {
38 go func(ext *Extension) {
39 defer func() {
40 // A panicking write to a closed pipe shouldn't kill
41 // the calling goroutine.
42 _ = recover()
43 }()
44 _, _ = ext.stdin.Write(frame)
45 }(ext)
46 }
47}
48
49// InterceptToolCall asks every extension that subscribed to tool_call
50// interception whether the call may proceed. Subscribers are invoked
51// serially; the first one to return Block: true wins, with its Reason
52// surfaced as the refusal text. allowed=true means proceed.
53//
54// Each subscriber gets up to interceptTimeout to reply; missing the
55// deadline counts as "allow" (i.e. an unresponsive interceptor never
56// stalls the agent).
57func (m *Manager) InterceptToolCall(ctx context.Context, toolID, toolName string, args json.RawMessage) (allowed bool, reason string) {
58 m.mu.RLock()
59 subs := make([]*Extension, 0, len(m.ext))
60 for _, ext := range m.ext {
61 ext.mu.Lock()
62 _, subscribed := ext.interceptSubs["tool_call"]
63 ext.mu.Unlock()
64 if subscribed {
65 subs = append(subs, ext)
66 }
67 }
68 m.mu.RUnlock()
69 if len(subs) == 0 {
70 return true, ""
71 }
72
73 for _, ext := range subs {
74 ok, reason := m.askIntercept(ctx, ext, toolID, toolName, args)
75 if !ok {
76 return false, reason
77 }
78 }
79 return true, ""
80}
81
82const interceptTimeout = 5 * time.Second
83
84func (m *Manager) askIntercept(ctx context.Context, ext *Extension, toolID, toolName string, args json.RawMessage) (allowed bool, reason string) {
85 id := newCorrelationID()
86 ch := make(chan extproto.EventInterceptResponseFromExt, 1)
87 ext.mu.Lock()
88 ext.pendingIntercept[id] = ch
89 ext.mu.Unlock()
90
91 frame, _ := extproto.Encode(extproto.EventInterceptFromHost{
92 Type: "event_intercept",
93 ID: id,
94 Event: "tool_call",
95 ToolID: toolID,
96 ToolName: toolName,
97 ToolArgs: args,
98 })
99 if _, err := ext.stdin.Write(frame); err != nil {
100 ext.mu.Lock()
101 delete(ext.pendingIntercept, id)
102 ext.mu.Unlock()
103 fmt.Fprintf(ext.logFile, "[zot] intercept write failed: %v\n", err)
104 return true, ""
105 }
106
107 select {
108 case resp := <-ch:
109 return !resp.Block, resp.Reason
110 case <-time.After(interceptTimeout):
111 ext.mu.Lock()
112 delete(ext.pendingIntercept, id)
113 ext.mu.Unlock()
114 fmt.Fprintf(ext.logFile, "[zot] intercept %s timed out; allowing\n", toolName)
115 return true, ""
116 case <-ctx.Done():
117 ext.mu.Lock()
118 delete(ext.pendingIntercept, id)
119 ext.mu.Unlock()
120 return true, ""
121 }
122}