Cletus plugin kit: rules for coding agents, specification, host API, style guide, worker SDK and cletus-plugin tool

This commit is contained in:
Eidonix committed 2026-11-10 19:50:09 -05:00
commit 9b19d3a8be
41 files changed
+4121

No files matched your search

+90
View File
@@ -0,0 +1,90 @@
package pluginsdk
import (
"context"
"encoding/json"
"fmt"
"time"
)
// Host makes calls to Cletus on behalf of the request being handled. Use
// it only inside the handler it was passed to.
type Host struct {
call func(ctx context.Context, method string, params any) (json.RawMessage, error)
}
// NewTestHost returns a Host whose calls are answered by fn, for unit tests
// of handlers. fn receives the method and the params as JSON.
func NewTestHost(fn func(method string, params json.RawMessage) (any, error)) *Host {
return &Host{call: func(_ context.Context, method string, params any) (json.RawMessage, error) {
raw, err := json.Marshal(params)
if err != nil {
return nil, err
}
result, err := fn(method, raw)
if err != nil {
return nil, err
}
return json.Marshal(result)
}}
}
// Call makes any host call in docs/host-api.md and decodes its result into
// out (which may be nil).
func (h *Host) Call(ctx context.Context, method string, params, out any) error {
if params == nil {
params = struct{}{}
}
raw, err := h.call(ctx, method, params)
if err != nil {
return err
}
if out == nil || len(raw) == 0 {
return nil
}
if err := json.Unmarshal(raw, out); err != nil {
return fmt.Errorf("%s: unexpected result: %w", method, err)
}
return nil
}
// ResolvePath returns the real path of a file the user chose
// (host.files.resolve; needs the files permission). Resolve every path
// before reading it.
func (h *Host) ResolvePath(ctx context.Context, path string) (string, error) {
var resolved string
err := h.Call(ctx, "host.files.resolve", map[string]string{"path": path}, &resolved)
return resolved, err
}
// CanWrite reports whether the user's current rules allow writing path
// (host.files.can_write; needs the files permission).
func (h *Host) CanWrite(ctx context.Context, path string) (bool, error) {
var allowed bool
err := h.Call(ctx, "host.files.can_write", map[string]string{"path": path}, &allowed)
return allowed, err
}
// Event sends an event to the plugin's screen. name must start with the
// plugin id and a colon ("acme.gauge-check:progress").
func (h *Host) Event(ctx context.Context, name string, args ...any) error {
if args == nil {
args = []any{}
}
return h.Call(ctx, "host.event", map[string]any{"name": name, "args": args}, nil)
}
// Published is a link to a file in the data folder that the screen can fetch.
type Published struct {
Token string `json:"token"`
URL string `json:"url"`
ExpiresAt string `json:"expiresAt"`
}
// Publish makes a file in the data folder fetchable by the screen for ttl
// (1 second to 1 hour). The file must not change while published.
func (h *Host) Publish(ctx context.Context, path, contentType string, ttl time.Duration) (Published, error) {
var out Published
err := h.Call(ctx, "host.data.publish", map[string]any{"path": path, "contentType": contentType, "ttl": int(ttl / time.Second)}, &out)
return out, err
}
+337
View File
@@ -0,0 +1,337 @@
// Package pluginsdk is the worker side of a Cletus plugin: the message
// protocol, chat tools, screen and service handlers, and calls to the host.
// See docs/specification.md and docs/host-api.md.
package pluginsdk
import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"strings"
"sync"
)
// APIVersion is the worker protocol version the host expects at handshake.
const APIVersion = 1
// MaxMessageBytes is the largest message (one line) either side may send.
const MaxMessageBytes = 64 << 20
// Func handles one screen (ui.*) request. params is the JSON array of the
// screen's arguments; decode it with Args.
type Func func(ctx context.Context, h *Host, params json.RawMessage) (any, error)
// ServiceFunc handles one service method another plugin calls.
type ServiceFunc func(ctx context.Context, h *Host, call ServiceCall) (any, error)
// ServiceCall is a service request the host routed from another plugin.
type ServiceCall struct {
Caller string `json:"caller"`
Mutating bool `json:"mutating"`
Args json.RawMessage `json:"args"`
}
// Handlers is everything a worker answers.
type Handlers struct {
// Tools are the chat tools. Each must also be listed in plugin.json.
Tools []Tool
// UI maps a screen method name ("check", answered as ui.check) to its handler.
UI map[string]Func
// Services maps "<service>.<method>" ("gauge.check") to its handler.
// Each must be declared under "provides" in plugin.json.
Services map[string]ServiceFunc
// Busy reports whether work is running that must not be interrupted.
// Nil means never busy.
Busy func() bool
}
// ID returns the plugin id the host started the worker as.
func ID() string { return os.Getenv("CLETUS_PLUGIN_ID") }
// DataDir returns the worker's data folder for the signed-in account. It is
// also the worker's working directory. Empty when not started by a host.
func DataDir() string { return os.Getenv("CLETUS_PLUGIN_DATA") }
// Serve runs the worker on standard input and output until the host shuts
// it down or closes the pipe. Standard output belongs to the protocol, so
// Serve points os.Stdout at standard error: a stray fmt.Println lands in
// the worker log instead of breaking the connection.
func Serve(h Handlers) error {
out := os.Stdout
os.Stdout = os.Stderr
return ServeIO(context.Background(), os.Stdin, out, h)
}
type message struct {
ID string `json:"id,omitempty"`
ParentID string `json:"parent_id,omitempty"`
Method string `json:"method,omitempty"`
Params json.RawMessage `json:"params,omitempty"`
Result json.RawMessage `json:"result,omitempty"`
Error *wireError `json:"error,omitempty"`
}
type wireError struct {
Message string `json:"message"`
}
type server struct {
handlers Handlers
tools map[string]Tool
out io.Writer
writeMu sync.Mutex
mu sync.Mutex
active map[string]context.CancelFunc
pending map[string]chan message
next int
wg sync.WaitGroup
}
// ServeIO runs the worker on the given streams. It returns when ctx ends,
// the input closes, or the host sends plugin.shutdown.
func ServeIO(ctx context.Context, in io.Reader, out io.Writer, h Handlers) error {
s := &server{handlers: h, tools: map[string]Tool{}, out: out,
active: map[string]context.CancelFunc{}, pending: map[string]chan message{}}
for _, t := range h.Tools {
if _, dup := s.tools[t.Name]; dup {
return fmt.Errorf("tool %q is registered twice", t.Name)
}
s.tools[t.Name] = t
}
ctx, cancel := context.WithCancel(ctx)
defer cancel()
defer s.wg.Wait()
defer s.cancelAll()
lines := make(chan []byte)
readErr := make(chan error, 1)
go func() {
reader := bufio.NewReaderSize(in, 64<<10)
for {
line, err := readLine(reader)
if len(line) > 0 {
select {
case lines <- line:
case <-ctx.Done():
return
}
}
if err != nil {
readErr <- err
return
}
}
}()
for {
select {
case <-ctx.Done():
return ctx.Err()
case err := <-readErr:
if errors.Is(err, io.EOF) {
return nil
}
return err
case line := <-lines:
var m message
if err := json.Unmarshal(line, &m); err != nil {
return fmt.Errorf("invalid message from host: %w", err)
}
if stop := s.dispatch(ctx, m); stop {
return nil
}
}
}
}
func readLine(r *bufio.Reader) ([]byte, error) {
var line []byte
for {
chunk, err := r.ReadSlice('\n')
line = append(line, chunk...)
if len(line) > MaxMessageBytes+1 {
return nil, errors.New("message from host is too large")
}
if err == bufio.ErrBufferFull {
continue
}
return []byte(strings.TrimSpace(string(line))), err
}
}
// dispatch handles one incoming message and reports whether to stop.
func (s *server) dispatch(ctx context.Context, m message) bool {
switch {
case m.Method == "$/cancel":
var id string
if json.Unmarshal(m.Params, &id) == nil {
s.mu.Lock()
if cancel := s.active[id]; cancel != nil {
cancel()
}
s.mu.Unlock()
}
case m.Method == "plugin.shutdown":
s.reply(m.ID, map[string]bool{"ok": true}, nil)
return true
case m.Method != "":
if m.ID == "" {
return false // a notification this worker doesn't use
}
reqCtx, cancel := context.WithCancel(ctx)
s.mu.Lock()
s.active[m.ID] = cancel
s.mu.Unlock()
s.wg.Add(1)
go func() {
defer s.wg.Done()
defer func() {
s.mu.Lock()
delete(s.active, m.ID)
s.mu.Unlock()
cancel()
}()
result, err := s.handle(reqCtx, m)
s.reply(m.ID, result, err)
}()
case m.ID != "":
s.mu.Lock()
ch := s.pending[m.ID]
delete(s.pending, m.ID)
s.mu.Unlock()
if ch != nil {
ch <- m
}
}
return false
}
func (s *server) handle(ctx context.Context, m message) (result any, err error) {
defer func() {
if r := recover(); r != nil {
result, err = nil, fmt.Errorf("the plugin failed while handling %s: %v", m.Method, r)
}
}()
h := &Host{call: func(ctx context.Context, method string, params any) (json.RawMessage, error) {
return s.callHost(ctx, m.ID, method, params)
}}
switch {
case m.Method == "plugin.handshake":
return map[string]int{"api_version": APIVersion}, nil
case m.Method == "plugin.busy":
return s.handlers.Busy != nil && s.handlers.Busy(), nil
case m.Method == "tools.list":
return s.listTools(), nil
case m.Method == "tools.invoke":
return s.invokeTool(ctx, h, m.Params)
case strings.HasPrefix(m.Method, "ui."):
fn := s.handlers.UI[strings.TrimPrefix(m.Method, "ui.")]
if fn == nil {
return nil, fmt.Errorf("unknown method %s", m.Method)
}
return fn(ctx, h, m.Params)
case strings.HasPrefix(m.Method, "svc."):
fn := s.handlers.Services[strings.TrimPrefix(m.Method, "svc.")]
if fn == nil {
return nil, fmt.Errorf("unknown method %s", m.Method)
}
var call ServiceCall
if err := json.Unmarshal(m.Params, &call); err != nil {
return nil, fmt.Errorf("invalid service request: %w", err)
}
return fn(ctx, h, call)
}
return nil, fmt.Errorf("unknown method %s", m.Method)
}
func (s *server) reply(id string, result any, err error) {
if id == "" {
return
}
m := message{ID: id}
if err != nil {
m.Error = &wireError{Message: err.Error()}
} else {
raw, marshalErr := json.Marshal(result)
if marshalErr != nil {
m.Error = &wireError{Message: "the plugin returned a result that is not JSON: " + marshalErr.Error()}
} else {
m.Result = raw
}
}
s.send(m)
}
func (s *server) send(m message) {
data, err := json.Marshal(m)
if err != nil || len(data) > MaxMessageBytes {
if m.ID != "" && m.Method == "" {
data, _ = json.Marshal(message{ID: m.ID, Error: &wireError{Message: "the plugin's reply is too large"}})
} else {
return
}
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
_, _ = s.out.Write(append(data, '\n'))
}
func (s *server) callHost(ctx context.Context, parent, method string, params any) (json.RawMessage, error) {
raw, err := json.Marshal(params)
if err != nil {
return nil, fmt.Errorf("%s: %w", method, err)
}
ch := make(chan message, 1)
s.mu.Lock()
s.next++
id := fmt.Sprintf("w%d", s.next)
s.pending[id] = ch
s.mu.Unlock()
s.send(message{ID: id, ParentID: parent, Method: method, Params: raw})
select {
case reply := <-ch:
if reply.Error != nil {
return nil, errors.New(reply.Error.Message)
}
return reply.Result, nil
case <-ctx.Done():
s.mu.Lock()
delete(s.pending, id)
s.mu.Unlock()
cancel, _ := json.Marshal(id)
s.send(message{Method: "$/cancel", Params: cancel})
return nil, ctx.Err()
}
}
func (s *server) cancelAll() {
s.mu.Lock()
defer s.mu.Unlock()
for _, cancel := range s.active {
cancel()
}
}
// Args decodes a screen request's argument array into targets, in order.
// Missing trailing arguments leave their targets unchanged.
func Args(params json.RawMessage, targets ...any) error {
var items []json.RawMessage
if len(params) > 0 && string(params) != "null" {
if err := json.Unmarshal(params, &items); err != nil {
return fmt.Errorf("screen arguments must be a JSON array: %w", err)
}
}
for i, target := range targets {
if i >= len(items) {
break
}
if err := json.Unmarshal(items[i], target); err != nil {
return fmt.Errorf("argument %d: %w", i+1, err)
}
}
return nil
}
+168
View File
@@ -0,0 +1,168 @@
package pluginsdk
import (
"bufio"
"context"
"encoding/json"
"io"
"strings"
"testing"
"time"
)
// pipeHost drives a worker over in-memory pipes, as Cletus does.
type pipeHost struct {
t *testing.T
toW *io.PipeWriter
fromW *bufio.Scanner
done chan error
cancel context.CancelFunc
}
func startWorker(t *testing.T, h Handlers) *pipeHost {
t.Helper()
inR, inW := io.Pipe()
outR, outW := io.Pipe()
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- ServeIO(ctx, inR, outW, h); outW.Close() }()
sc := bufio.NewScanner(outR)
sc.Buffer(make([]byte, 1<<20), MaxMessageBytes)
p := &pipeHost{t: t, toW: inW, fromW: sc, done: done, cancel: cancel}
t.Cleanup(func() { cancel(); inW.Close() })
return p
}
func (p *pipeHost) send(m message) {
p.t.Helper()
data, _ := json.Marshal(m)
if _, err := p.toW.Write(append(data, '\n')); err != nil {
p.t.Fatal(err)
}
}
func (p *pipeHost) read() message {
p.t.Helper()
if !p.fromW.Scan() {
p.t.Fatalf("worker closed: %v", p.fromW.Err())
}
var m message
if err := json.Unmarshal(p.fromW.Bytes(), &m); err != nil {
p.t.Fatal(err)
}
return m
}
func TestHandshakeAndShutdown(t *testing.T) {
p := startWorker(t, Handlers{})
p.send(message{ID: "1", Method: "plugin.handshake"})
if got := p.read(); string(got.Result) != `{"api_version":1}` {
t.Fatalf("handshake = %s", got.Result)
}
p.send(message{ID: "2", Method: "plugin.shutdown"})
if got := p.read(); got.ID != "2" || got.Error != nil {
t.Fatalf("shutdown reply = %+v", got)
}
select {
case err := <-p.done:
if err != nil {
t.Fatalf("ServeIO = %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("worker did not stop after plugin.shutdown")
}
}
func TestToolWithHostCall(t *testing.T) {
tool := Tool{Name: "demo_read", Description: "Reads a file.", Run: func(ctx context.Context, h *Host, call ToolCall) ToolResult {
var args struct{ Path string }
if err := call.Decode(&args); err != nil {
return Fail("%v", err)
}
path, err := h.ResolvePath(ctx, args.Path)
if err != nil {
return Fail("%v", err)
}
return Text("resolved %s", path)
}}
p := startWorker(t, Handlers{Tools: []Tool{tool}})
p.send(message{ID: "1", Method: "tools.list"})
var list []toolDescriptor
if err := json.Unmarshal(p.read().Result, &list); err != nil || len(list) != 1 || list[0].Def.Function.Name != "demo_read" {
t.Fatalf("tools.list = %+v, %v", list, err)
}
params, _ := json.Marshal(toolInvocation{Name: "demo_read", Args: `{"Path":"a.txt"}`})
p.send(message{ID: "2", Method: "tools.invoke", Params: params})
call := p.read()
if call.Method != "host.files.resolve" || call.ParentID != "2" || !strings.Contains(string(call.Params), "a.txt") {
t.Fatalf("host call = %+v", call)
}
p.send(message{ID: call.ID, Result: json.RawMessage(`"/work/a.txt"`)})
var result ToolResult
if err := json.Unmarshal(p.read().Result, &result); err != nil || result.Text != "resolved /work/a.txt" {
t.Fatalf("result = %+v, %v", result, err)
}
}
func TestCancelStopsHandler(t *testing.T) {
started := make(chan struct{})
p := startWorker(t, Handlers{UI: map[string]Func{"wait": func(ctx context.Context, h *Host, params json.RawMessage) (any, error) {
close(started)
<-ctx.Done()
return nil, ctx.Err()
}}})
p.send(message{ID: "9", Method: "ui.wait", Params: json.RawMessage(`[]`)})
<-started
cancelID, _ := json.Marshal("9")
p.send(message{Method: "$/cancel", Params: cancelID})
if got := p.read(); got.ID != "9" || got.Error == nil {
t.Fatalf("cancelled reply = %+v", got)
}
}
func TestPanicBecomesError(t *testing.T) {
p := startWorker(t, Handlers{UI: map[string]Func{"boom": func(context.Context, *Host, json.RawMessage) (any, error) {
panic("bad input")
}}})
p.send(message{ID: "1", Method: "ui.boom"})
if got := p.read(); got.Error == nil || !strings.Contains(got.Error.Message, "bad input") {
t.Fatalf("reply = %+v", got)
}
}
func TestServiceAndUnknownMethod(t *testing.T) {
p := startWorker(t, Handlers{Services: map[string]ServiceFunc{"gauge.check": func(_ context.Context, _ *Host, call ServiceCall) (any, error) {
return map[string]string{"caller": call.Caller}, nil
}}})
p.send(message{ID: "1", Method: "svc.gauge.check", Params: json.RawMessage(`{"caller":"acme.other","args":{}}`)})
if got := p.read(); string(got.Result) != `{"caller":"acme.other"}` {
t.Fatalf("service = %+v", got)
}
p.send(message{ID: "2", Method: "nope"})
if got := p.read(); got.Error == nil {
t.Fatalf("unknown method answered %+v", got)
}
}
func TestArgs(t *testing.T) {
var name string
var count int
if err := Args(json.RawMessage(`["x", 3]`), &name, &count); err != nil || name != "x" || count != 3 {
t.Fatalf("Args = %q %d %v", name, count, err)
}
if err := Args(json.RawMessage(`{"a":1}`), &name); err == nil {
t.Fatal("Args accepted an object")
}
}
func TestNewTestHost(t *testing.T) {
h := NewTestHost(func(method string, params json.RawMessage) (any, error) {
return method == "host.files.can_write", nil
})
ok, err := h.CanWrite(context.Background(), "x")
if err != nil || !ok {
t.Fatalf("CanWrite = %v %v", ok, err)
}
}
+116
View File
@@ -0,0 +1,116 @@
package pluginsdk
import (
"context"
"encoding/json"
"fmt"
"sort"
)
// Tool is a chat tool: a function Cletus's AI can call. Name, Mutating and
// RequiresApproval must match the plugin's entry in plugin.json "tools".
type Tool struct {
Name string
Description string
// Parameters is the JSON schema of the arguments (an object schema).
Parameters map[string]any
// Mutating is true if the tool writes anything anywhere.
Mutating bool
// RequiresApproval makes Cletus ask the user before every call.
RequiresApproval bool
Run func(ctx context.Context, h *Host, call ToolCall) ToolResult
}
// ToolCall is one call of a chat tool.
type ToolCall struct {
// Args is the JSON object of arguments the AI sent.
Args json.RawMessage
// WorkingDir is the user's working folder ("" when none is open).
WorkingDir string
// EditMode is false when the user has Cletus read-only: a mutating tool
// must check Host.CanWrite before each write and refuse when it's false.
EditMode bool
}
// Decode decodes the arguments into v.
func (c ToolCall) Decode(v any) error {
if len(c.Args) == 0 {
return json.Unmarshal([]byte("{}"), v)
}
if err := json.Unmarshal(c.Args, v); err != nil {
return fmt.Errorf("the arguments don't match the tool's parameters: %w", err)
}
return nil
}
// ToolResult is what the AI reads back. Set Text on success, Error on
// failure (a plain sentence with the next step). Images are optional
// (PNG or JPEG bytes).
type ToolResult struct {
Text string `json:"Text,omitempty"`
Error string `json:"Error,omitempty"`
Images [][]byte `json:"Images,omitempty"`
}
// Text returns a successful result.
func Text(format string, args ...any) ToolResult {
return ToolResult{Text: fmt.Sprintf(format, args...)}
}
// Fail returns a failed result.
func Fail(format string, args ...any) ToolResult {
return ToolResult{Error: fmt.Sprintf(format, args...)}
}
type toolDescriptor struct {
Def toolDef `json:"def"`
Mutating bool `json:"mutating"`
RequiresApproval bool `json:"requires_approval"`
}
type toolDef struct {
Type string `json:"type"`
Function functionDef `json:"function"`
}
type functionDef struct {
Name string `json:"name"`
Description string `json:"description"`
Parameters map[string]any `json:"parameters"`
}
func (s *server) listTools() []toolDescriptor {
out := make([]toolDescriptor, 0, len(s.tools))
for _, t := range s.tools {
params := t.Parameters
if params == nil {
params = map[string]any{"type": "object", "properties": map[string]any{}}
}
out = append(out, toolDescriptor{
Def: toolDef{Type: "function", Function: functionDef{Name: t.Name, Description: t.Description, Parameters: params}},
Mutating: t.Mutating,
RequiresApproval: t.RequiresApproval,
})
}
sort.Slice(out, func(i, j int) bool { return out[i].Def.Function.Name < out[j].Def.Function.Name })
return out
}
type toolInvocation struct {
Name string `json:"name"`
Args string `json:"args"`
WorkingDir string `json:"working_dir"`
EditMode bool `json:"edit_mode"`
}
func (s *server) invokeTool(ctx context.Context, h *Host, params json.RawMessage) (ToolResult, error) {
var inv toolInvocation
if err := json.Unmarshal(params, &inv); err != nil {
return ToolResult{}, fmt.Errorf("invalid tool call: %w", err)
}
t, ok := s.tools[inv.Name]
if !ok || t.Run == nil {
return Fail("This plugin has no tool named %q.", inv.Name), nil
}
return t.Run(ctx, h, ToolCall{Args: json.RawMessage(inv.Args), WorkingDir: inv.WorkingDir, EditMode: inv.EditMode}), nil
}