feat(plugin): stop idle host plugins and start them on the next call

Every enabled host plugin kept a process running on every node, about
10 MB for a Go plugin and 16 MB for a Python one, though most of them
(connectors, parsers) are called now and then.

With WEKNORA_PLUGIN_IDLE_TIMEOUT set (e.g. 10m), the host stops a plugin
that has had no calls for that long and starts it again on the next one.
The process record stays, in a new idle state: routing, the plugin hosts'
announcements (idle counts as running) and the node status are
unchanged. A call to an idle plugin wakes it and waits for it, bounded by
the caller's context; callers arriving together share one start.

Calls are counted in the plugin client's transport from the request
until the answer's body is closed, streams included; health and manifest
checks do not count, and handing out a client counts as use, so a plugin
is not stopped between a caller getting its client and sending the call.
A stop signal left over from before a crash and restart is ignored.

Off by default; the desktop app sets 10 minutes. Plugins that work
between calls declare runtime.keepAlive (singleton plugins are resident
too). Upgrades still start the new version to check it before it idles.
Measured on the examples: a cold call to a Go plugin took about 11 ms,
a Python plugin reaches its handshake in about 70 ms.

.env.example

@@ -832,6 +832,9 @@ CONCURRENCY_POOL_SIZE=5
 # 插件停用、升级或重启时，留给旧进程处理完进行中调用（解析、同步）的时间，超时后强制结束。
 # 默认 65s，略长于 SDK 收到 SIGTERM 后的 60 秒排空时间。
 # WEKNORA_PLUGIN_STOP_GRACE=65s
+# 宿主插件多久没有调用就停掉进程（如 10m），下一次调用时再启动，首次调用多等一次启动时间。
+# 默认不停（桌面端默认 10m）。声明 runtime.keepAlive 或 singleton 的插件始终常驻。
+# WEKNORA_PLUGIN_IDLE_TIMEOUT=10m
 # 委托给 WeKnora 的 cgroup v2 目录（Linux，需可写）。设置后每个插件进程进入各自的子 cgroup，
 # 按 runtime.resources 设置 memory.max 与 cpu.max；不设置时只限内存（rlimit），不限 CPU。
 # WEKNORA_PLUGIN_CGROUP=/sys/fs/cgroup/weknora-plugins

cmd/desktop/plugins.go

@@ -17,8 +17,12 @@ import (
 // directory, which macOS purges while the app runs. And a Finder-launched
 // app only has launchd's PATH, where python3 is at best Apple's stub that
 // offers to install the developer tools; python plugins get an interpreter
-// found in the usual places, or are left off.
+// found in the usual places, or are left off. Plugins stop after 10
+// minutes without calls, so a laptop does not keep them all running.
 func configureDesktopPlugins(home string) {
+	if _, set := os.LookupEnv("WEKNORA_PLUGIN_IDLE_TIMEOUT"); !set {
+		_ = os.Setenv("WEKNORA_PLUGIN_IDLE_TIMEOUT", desktopIdleTimeout)
+	}
 	if strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_CACHE_DIR")) == "" {
 		dir := filepath.Join(home, ".weknora", "data", "plugins")
 		if err := os.MkdirAll(dir, 0o755); err != nil {
@@ -39,6 +43,9 @@ func configureDesktopPlugins(home string) {
 	_ = os.Setenv("WEKNORA_PLUGIN_EMBEDDED_KINDS", "binary")
 }
 
+// desktopIdleTimeout is how long a desktop plugin runs without calls.
+const desktopIdleTimeout = "10m"
+
 // desktopPython finds a python3 a desktop app can run: on PATH, then where
 // Homebrew, python.org and pyenv put it. On macOS /usr/bin/python3 counts
 // only when the command line tools behind it are installed.

cmd/desktop/plugins_test.go

@@ -40,7 +40,11 @@ func TestConfigureDesktopPluginsKeepsExplicitSettings(t *testing.T) {
 	t.Setenv("WEKNORA_PLUGIN_CACHE_DIR", "")
 	t.Setenv("WEKNORA_PLUGIN_PYTHON", "/custom/python3")
 	t.Setenv("WEKNORA_PLUGIN_EMBEDDED_KINDS", "")
+	t.Setenv("WEKNORA_PLUGIN_IDLE_TIMEOUT", "0")
 	configureDesktopPlugins(home)
+	if got := os.Getenv("WEKNORA_PLUGIN_IDLE_TIMEOUT"); got != "0" {
+		t.Fatalf("an explicit idle timeout was replaced: %q", got)
+	}
 	if got := os.Getenv("WEKNORA_PLUGIN_CACHE_DIR"); got != filepath.Join(home, ".weknora", "data", "plugins") {
 		t.Fatalf("cache dir = %q", got)
 	}
@@ -51,3 +55,14 @@ func TestConfigureDesktopPluginsKeepsExplicitSettings(t *testing.T) {
 		t.Fatalf("kinds changed despite an explicit interpreter: %q", got)
 	}
 }
+
+func TestConfigureDesktopPluginsStopsIdlePlugins(t *testing.T) {
+	t.Setenv("WEKNORA_PLUGIN_CACHE_DIR", t.TempDir())
+	t.Setenv("WEKNORA_PLUGIN_PYTHON", "/custom/python3")
+	t.Setenv("WEKNORA_PLUGIN_IDLE_TIMEOUT", "")
+	_ = os.Unsetenv("WEKNORA_PLUGIN_IDLE_TIMEOUT")
+	configureDesktopPlugins(t.TempDir())
+	if got := os.Getenv("WEKNORA_PLUGIN_IDLE_TIMEOUT"); got != desktopIdleTimeout {
+		t.Fatalf("idle timeout = %q", got)
+	}
+}

internal/container/plugin_runtime.go

@@ -272,7 +272,7 @@ func (c pluginClients) Client(ctx context.Context, m *manifest.Manifest) (*clien
 	case c.remote.Owns(m.ID):
 		return c.remote.Client(m.ID)
 	case c.host.Local(m.ID) || c.pool == nil || m.Runtime.Type != manifest.RuntimeHost:
-		return c.host.Client(m.ID)
+		return c.host.Client(ctx, m.ID)
 	default:
 		return c.pool.Client(ctx, m.ID, m.Version)
 	}

internal/plugin/host/gateway.go

@@ -77,7 +77,7 @@ func (m *Manager) relay(w http.ResponseWriter, r *http.Request, key []byte) {
 		gatewayError(w, pluginapi.Errorf(pluginapi.CodeUnavailable, "this host does not run %s@%s", id, version))
 		return
 	}
-	c, err := p.Client()
+	c, err := p.Client(r.Context())
 	if err != nil {
 		if pe, ok := pluginapi.AsError(err); ok {
 			gatewayError(w, pe)

internal/plugin/host/host.go

@@ -58,6 +58,12 @@ type Manager struct {
 	// hosts only).
 	handingOver map[string]*process
 	onChange    func()
+
+	// idleTimeout stops plugins without calls for that long; 0 keeps them
+	// running. The reaper starts with the first plugin.
+	idleTimeout time.Duration
+	reaperOnce  sync.Once
+	reaperStop  chan struct{}
 }
 
 // handoverWindow is how long a standalone host keeps serving the previous
@@ -68,7 +74,7 @@ var handoverWindow = reconcile.DefaultInterval + 15*time.Second
 
 // NewManager creates an empty host that runs every kind this machine can.
 func NewManager() *Manager {
-	m := &Manager{procs: map[string]*process{}}
+	m := &Manager{procs: map[string]*process{}, idleTimeout: IdleTimeoutFromEnv(), reaperStop: make(chan struct{})}
 	m.SetKinds(AvailableKinds())
 	return m
 }
@@ -76,7 +82,10 @@ func NewManager() *Manager {
 // NewStandaloneManager creates the host of a standalone plugin host
 // (weknora plugin-host): it runs the given kinds and refuses the others.
 func NewStandaloneManager(kinds []string) *Manager {
-	m := &Manager{procs: map[string]*process{}, standalone: true, handingOver: map[string]*process{}}
+	m := &Manager{
+		procs: map[string]*process{}, standalone: true, handingOver: map[string]*process{},
+		idleTimeout: IdleTimeoutFromEnv(), reaperStop: make(chan struct{}),
+	}
 	m.SetKinds(kinds)
 	return m
 }
@@ -150,6 +159,14 @@ func (m *Manager) SetDirectHosts(hosts ...string) {
 	m.mu.Unlock()
 }
 
+// SetIdleTimeout stops plugins that go without calls for d until their
+// next call; 0 keeps them running. Set it before the first plugin starts.
+func (m *Manager) SetIdleTimeout(d time.Duration) {
+	m.mu.Lock()
+	m.idleTimeout = d
+	m.mu.Unlock()
+}
+
 // SetHostAPIAddr names this node's Host API on loopback ("127.0.0.1:8080"),
 // which sandboxed plugins (see WEKNORA_PLUGIN_NETNS) reach through a relay.
 func (m *Manager) SetHostAPIAddr(addr string) {
@@ -229,6 +246,7 @@ func (m *Manager) Stage(ctx context.Context, _, l *reconcile.Loaded) (reconcile.
 // it replaces.
 func (m *Manager) serve(ctx context.Context, mf *manifest.Manifest, p *process) {
 	id := mf.ID
+	m.startReaper()
 	m.mu.Lock()
 	old := m.procs[id]
 	m.procs[id] = p
@@ -250,6 +268,21 @@ func (m *Manager) serve(ctx context.Context, mf *manifest.Manifest, p *process)
 	logger.Infof(ctx, "[plugin] host started %s %s", id, mf.Version)
 }
 
+// startReaper starts stopping idle plugins, once, if the host is set to.
+func (m *Manager) startReaper() {
+	m.mu.Lock()
+	timeout := m.idleTimeout
+	m.mu.Unlock()
+	if timeout <= 0 {
+		return
+	}
+	m.reaperOnce.Do(func() {
+		logger.Infof(context.Background(), "[plugin] host plugins stop after %s without calls (%s)",
+			timeout, envIdleTimeout)
+		go m.idleReaper(timeout, m.reaperStop)
+	})
+}
+
 // endHandover stops serving a plugin's previous version, unless it was
 // stopped already.
 func (m *Manager) endHandover(id string, p *process) {
@@ -314,9 +347,10 @@ func (m *Manager) report(pluginID string, p *process, s State, err error) {
 	}
 }
 
-// Client returns a client for a running plugin. The error is a retryable
+// Client returns a client for a running plugin, starting it if it was
+// stopped for being idle (ctx bounds the wait). The error is a retryable
 // pluginapi unavailable error while the plugin is starting or restarting.
-func (m *Manager) Client(pluginID string) (*client.Client, error) {
+func (m *Manager) Client(ctx context.Context, pluginID string) (*client.Client, error) {
 	m.mu.Lock()
 	p := m.procs[pluginID]
 	m.mu.Unlock()
@@ -325,7 +359,7 @@ func (m *Manager) Client(pluginID string) (*client.Client, error) {
 			Code: pluginapi.CodeUnavailable, Message: fmt.Sprintf("plugin %s is not running on this node", pluginID),
 		}
 	}
-	return p.Client()
+	return p.Client(ctx)
 }
 
 // Local reports whether a plugin runs on this host.
@@ -391,7 +425,13 @@ func (m *Manager) InFlight() int64 { return m.inFlight.Load() }
 
 // Close stops every plugin, for shutdown.
 func (m *Manager) Close() {
+	m.reaperOnce.Do(func() {}) // no reaper starts after this
 	m.mu.Lock()
+	select {
+	case <-m.reaperStop:
+	default:
+		close(m.reaperStop)
+	}
 	procs := make([]*process, 0, len(m.procs)+len(m.handingOver))
 	for _, p := range m.procs {
 		procs = append(procs, p)

internal/plugin/host/host_test.go

@@ -112,7 +112,7 @@ func fastTimings(t *testing.T) {
 
 func search(t *testing.T, m *Manager, q string) (string, error) {
 	t.Helper()
-	c, err := m.Client("acme.echo")
+	c, err := m.Client(context.Background(), "acme.echo")
 	if err != nil {
 		return "", err
 	}
@@ -175,7 +175,7 @@ func TestHostRunsRestartsAndStopsAPlugin(t *testing.T) {
 	if err := m.Deactivate(ctx, "acme.echo"); err != nil {
 		t.Fatal(err)
 	}
-	if _, err := m.Client("acme.echo"); err == nil {
+	if _, err := m.Client(context.Background(), "acme.echo"); err == nil {
 		t.Fatal("a stopped plugin must have no client")
 	}
 	if runtime.GOOS != "windows" {

internal/plugin/host/idle.go

@@ -0,0 +1,204 @@
+package host
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/Tencent/WeKnora/pluginsdk/client"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// envIdleTimeout is how long a host plugin may go without calls before the
+// host stops it, e.g. "10m"; the next call starts it again. Unset or zero:
+// plugins keep running. Plugins with runtime.keepAlive (or singleton) are
+// never stopped for being idle.
+const envIdleTimeout = "WEKNORA_PLUGIN_IDLE_TIMEOUT"
+
+// IdleTimeoutFromEnv reads WEKNORA_PLUGIN_IDLE_TIMEOUT; 0 when unset or
+// not a positive duration.
+func IdleTimeoutFromEnv() time.Duration {
+	d, err := time.ParseDuration(strings.TrimSpace(os.Getenv(envIdleTimeout)))
+	if err != nil || d <= 0 {
+		return 0
+	}
+	return d
+}
+
+// errParked ends a watch when the host stops an idle plugin.
+var errParked = errors.New("stopped while idle")
+
+// activity is what a plugin's callers have been doing: calls in flight and
+// when the last one started or ended. Health checks do not count.
+type activity struct {
+	inFlight atomic.Int64
+	last     atomic.Int64 // unix nanoseconds
+}
+
+func newActivity() *activity {
+	a := &activity{}
+	a.touch()
+	return a
+}
+
+func (a *activity) touch() { a.last.Store(time.Now().UnixNano()) }
+
+func (a *activity) done() {
+	a.inFlight.Add(-1)
+	a.touch()
+}
+
+// idleSince reports how long the plugin has been without calls; false
+// while one is in flight.
+func (a *activity) idleSince(now time.Time) (time.Duration, bool) {
+	if a.inFlight.Load() > 0 {
+		return 0, false
+	}
+	return now.Sub(time.Unix(0, a.last.Load())), true
+}
+
+// countingTransport counts a plugin's calls from the request until its
+// answer's body is closed, streams included.
+type countingTransport struct {
+	next http.RoundTripper
+	a    *activity
+}
+
+// quietPaths are the host's own checks, which say nothing about use.
+var quietPaths = map[string]bool{"/v1/health": true, "/v1/manifest": true}
+
+func (t countingTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	if quietPaths[req.URL.Path] {
+		return t.next.RoundTrip(req)
+	}
+	t.a.inFlight.Add(1)
+	t.a.touch()
+	resp, err := t.next.RoundTrip(req)
+	if err != nil {
+		t.a.done()
+		return nil, err
+	}
+	resp.Body = &countedBody{ReadCloser: resp.Body, a: t.a}
+	return resp, nil
+}
+
+// CloseIdleConnections lets client.Close reach the wrapped transport.
+func (t countingTransport) CloseIdleConnections() {
+	if c, ok := t.next.(interface{ CloseIdleConnections() }); ok {
+		c.CloseIdleConnections()
+	}
+}
+
+type countedBody struct {
+	io.ReadCloser
+	a    *activity
+	once sync.Once
+}
+
+func (b *countedBody) Close() error {
+	err := b.ReadCloser.Close()
+	b.once.Do(b.a.done)
+	return err
+}
+
+// processClient reaches a plugin process at the address its handshake
+// names, counting its calls in a.
+func processClient(hs pluginapi.Handshake, auth client.Auth, a *activity) *client.Client {
+	if hs.Network == "unix" {
+		tr := &http.Transport{
+			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+				var d net.Dialer
+				return d.DialContext(ctx, "unix", hs.Address)
+			},
+			MaxIdleConnsPerHost: 16,
+			IdleConnTimeout:     90 * time.Second,
+		}
+		return client.New("http://plugin", &http.Client{Transport: countingTransport{tr, a}}, auth)
+	}
+	tr := http.DefaultTransport.(*http.Transport).Clone()
+	return client.New("http://"+hs.Address, &http.Client{Transport: countingTransport{tr, a}}, auth)
+}
+
+// parkIfIdle stops the plugin if it has had no calls for timeout; the next
+// Client call starts it again. It reports whether it did.
+func (p *process) parkIfIdle(now time.Time, timeout time.Duration) bool {
+	p.mu.Lock()
+	idle, ok := p.activity.idleSince(now)
+	if p.state != StateReady || !ok || idle < timeout {
+		p.mu.Unlock()
+		return false
+	}
+	// Callers from here on see the plugin idle and wake it; the ones
+	// before touched the activity, so it would not be idle.
+	c := p.client
+	p.client = nil
+	p.setStateLocked(StateIdle, nil)
+	p.mu.Unlock()
+	if p.onState != nil {
+		p.onState(p, StateIdle, nil)
+	}
+	if c != nil {
+		c.Close()
+	}
+	select {
+	case p.park <- struct{}{}:
+	default:
+	}
+	return true
+}
+
+// requestWake asks the supervisor to start an idle plugin.
+func (p *process) requestWake() {
+	select {
+	case p.wake <- struct{}{}:
+	default:
+	}
+}
+
+// parkIntended reports whether a park signal still stands: a plugin that
+// crashed and restarted since is not idle any more.
+func (p *process) parkIntended() bool {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return p.state == StateIdle
+}
+
+// idleReaper stops the idle plugins of a manager every tick until stop.
+func (m *Manager) idleReaper(timeout time.Duration, stop <-chan struct{}) {
+	tick := min(timeout/4, 30*time.Second)
+	if tick <= 0 {
+		tick = timeout
+	}
+	t := time.NewTicker(tick)
+	defer t.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case now := <-t.C:
+			m.parkIdle(now, timeout)
+		}
+	}
+}
+
+// parkIdle stops the plugins idle for timeout, except resident ones.
+func (m *Manager) parkIdle(now time.Time, timeout time.Duration) {
+	m.mu.Lock()
+	procs := make([]*process, 0, len(m.procs))
+	for _, p := range m.procs {
+		if !p.spec.m.Runtime.Resident() {
+			procs = append(procs, p)
+		}
+	}
+	m.mu.Unlock()
+	for _, p := range procs {
+		p.parkIfIdle(now, timeout)
+	}
+}

internal/plugin/host/idle_test.go

@@ -0,0 +1,159 @@
+package host
+
+import (
+	"context"
+	"os"
+	"runtime"
+	"strconv"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/Tencent/WeKnora/internal/plugin/reconcile"
+)
+
+func state(m *Manager, id string) State {
+	for _, r := range m.Running() {
+		if r.ID == id {
+			return r.State
+		}
+	}
+	return ""
+}
+
+func exited(pid string) bool {
+	n, _ := strconv.Atoi(pid)
+	proc, err := os.FindProcess(n)
+	return err != nil || proc.Signal(syscall.Signal(0)) != nil
+}
+
+// A plugin without calls is stopped after the idle timeout and started
+// again by the next call, which waits for it; the host still runs it as
+// far as routing and plugin hosts are concerned.
+func TestIdlePluginStopsAndStartsOnTheNextCall(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("process checks use signal 0")
+	}
+	fastTimings(t)
+	ctx := context.Background()
+	m := NewManager()
+	m.SetIdleTimeout(300 * time.Millisecond)
+	rep := &reports{}
+	m.SetReporter(rep)
+	defer m.Close()
+	if err := reconcile.Activate(ctx, m, install(t, "1.0.0", "")); err != nil {
+		t.Fatal(err)
+	}
+	pid1, err := search(t, m, "pid")
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitFor(t, "the idle plugin to stop", func() bool { return state(m, "acme.echo") == StateIdle })
+	waitFor(t, "the idle process to exit", func() bool { return exited(pid1) })
+	if !m.Local("acme.echo") {
+		t.Fatal("an idle plugin must still route here")
+	}
+
+	pid2, err := search(t, m, "pid")
+	if err != nil || pid2 == pid1 {
+		t.Fatalf("after waking pid = %s (was %s), %v", pid2, pid1, err)
+	}
+	if state(m, "acme.echo") != StateReady {
+		t.Fatalf("state after waking = %s", state(m, "acme.echo"))
+	}
+	if got := rep.String(); got != "acme.echo:true" {
+		t.Fatalf("stopping and starting an idle plugin is not a health change: %s", got)
+	}
+}
+
+// A call in flight, a stream included, keeps the plugin running past the
+// timeout; health checks do not.
+func TestCallsInFlightKeepAPluginRunning(t *testing.T) {
+	fastTimings(t)
+	ctx := context.Background()
+	m := NewManager()
+	m.SetIdleTimeout(200 * time.Millisecond)
+	defer m.Close()
+	if err := reconcile.Activate(ctx, m, install(t, "1.0.0", "")); err != nil {
+		t.Fatal(err)
+	}
+	if got, err := search(t, m, "sleep:900ms"); err != nil || got != "slept" {
+		t.Fatalf("slow call = %q, %v", got, err)
+	}
+	if st := state(m, "acme.echo"); st != StateReady {
+		t.Fatalf("a plugin was stopped with a call in flight: %s", st)
+	}
+	// healthInterval is 100ms here: checks run, yet it goes idle.
+	waitFor(t, "the idle plugin to stop", func() bool { return state(m, "acme.echo") == StateIdle })
+}
+
+// A plugin that asks to keep running is never stopped for being idle.
+func TestKeepAlivePluginsStayRunning(t *testing.T) {
+	fastTimings(t)
+	ctx := context.Background()
+	m := NewManager()
+	m.SetIdleTimeout(100 * time.Millisecond)
+	defer m.Close()
+	l := install(t, "1.0.0", "")
+	l.Manifest.Runtime.KeepAlive = true
+	if err := reconcile.Activate(ctx, m, l); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(600 * time.Millisecond)
+	if st := state(m, "acme.echo"); st != StateReady {
+		t.Fatalf("a keepAlive plugin went %s", st)
+	}
+}
+
+// Callers that find a plugin idle at once share one start.
+func TestConcurrentCallsWakeAPluginOnce(t *testing.T) {
+	fastTimings(t)
+	ctx := context.Background()
+	m := NewManager()
+	m.SetIdleTimeout(200 * time.Millisecond)
+	defer m.Close()
+	if err := reconcile.Activate(ctx, m, install(t, "1.0.0", "")); err != nil {
+		t.Fatal(err)
+	}
+	waitFor(t, "the idle plugin to stop", func() bool { return state(m, "acme.echo") == StateIdle })
+	var wg sync.WaitGroup
+	pids := make([]string, 8)
+	for i := range pids {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			pids[i], _ = search(t, m, "pid")
+		}()
+	}
+	wg.Wait()
+	for _, pid := range pids {
+		if pid == "" || pid != pids[0] {
+			t.Fatalf("pids = %v, want one process", pids)
+		}
+	}
+}
+
+// A caller gives up on a plugin that takes too long to start, with an
+// error to retry.
+func TestWakingIsBoundedByTheCaller(t *testing.T) {
+	fastTimings(t)
+	ctx := context.Background()
+	m := NewManager()
+	m.SetIdleTimeout(200 * time.Millisecond)
+	defer m.Close()
+	if err := reconcile.Activate(ctx, m, install(t, "1.0.0", "")); err != nil {
+		t.Fatal(err)
+	}
+	waitFor(t, "the idle plugin to stop", func() bool { return state(m, "acme.echo") == StateIdle })
+	cctx, cancel := context.WithTimeout(ctx, time.Nanosecond)
+	defer cancel()
+	time.Sleep(time.Millisecond)
+	if _, err := m.Client(cctx, "acme.echo"); err == nil {
+		t.Fatal("want an error once the caller's deadline passed")
+	}
+	// The start it asked for goes on; the next caller gets the plugin.
+	if _, err := search(t, m, "hello"); err != nil {
+		t.Fatal(err)
+	}
+}

internal/plugin/host/process.go

@@ -62,6 +62,9 @@ const (
 	StateReady    State = "ready"
 	StateDegraded State = "degraded"
 	StateStopped  State = "stopped"
+	// StateIdle: stopped after going without calls; the next call starts
+	// it again (WEKNORA_PLUGIN_IDLE_TIMEOUT).
+	StateIdle State = "idle"
 )
 
 // spec is everything needed to start one plugin version.
@@ -139,6 +142,14 @@ type process struct {
 	client  *client.Client
 	state   State
 	lastErr error
+	// changed is closed and replaced on every state change, for callers
+	// waiting for an idle plugin to start.
+	changed chan struct{}
+
+	activity *activity
+	// park asks the supervisor to stop the idle child; wake to start it.
+	park chan struct{}
+	wake chan struct{}
 
 	cancel  context.CancelFunc
 	done    chan struct{}
@@ -177,7 +188,8 @@ func startProcess(sp spec, onState func(*process, State, error)) (*process, erro
 	ctx, cancel := context.WithCancel(context.Background())
 	p := &process{
 		spec: sp, onState: onState, cancel: cancel, done: make(chan struct{}), proxy: proxy, sockDir: sockDir,
-		sandboxed: useSandbox(),
+		sandboxed: useSandbox(), changed: make(chan struct{}), activity: newActivity(),
+		park: make(chan struct{}, 1), wake: make(chan struct{}, 1),
 	}
 	p.setState(StateStarting, nil)
 	first, err := p.launch(ctx, entry)
@@ -373,7 +385,7 @@ func (p *process) launch(ctx context.Context, entry string) (*launched, error) {
 		kill()
 		return nil, fmt.Errorf("plugin listens on %s %s, not where it was told", hs.Network, hs.Address)
 	}
-	c := client.ForHandshake(hs, client.Bearer(token))
+	c := processClient(hs, client.Bearer(token), p.activity)
 	cctx, cancel := context.WithTimeout(ctx, healthTimeout)
 	defer cancel()
 	if err := c.Health(cctx); err != nil {
@@ -483,14 +495,25 @@ func readLogLine(br *bufio.Reader) (line string, truncated bool, err error) {
 
 func (p *process) setState(s State, err error) {
 	p.mu.Lock()
-	changed := p.state != s || (err != nil) != (p.lastErr != nil)
-	p.state, p.lastErr = s, err
+	changed := p.setStateLocked(s, err)
 	p.mu.Unlock()
 	if changed && p.onState != nil {
 		p.onState(p, s, err)
 	}
 }
 
+// setStateLocked records a state with p.mu held and wakes the callers
+// waiting on it; it reports whether anything changed.
+func (p *process) setStateLocked(s State, err error) bool {
+	changed := p.state != s || (err != nil) != (p.lastErr != nil)
+	p.state, p.lastErr = s, err
+	if changed {
+		close(p.changed)
+		p.changed = make(chan struct{})
+	}
+	return changed
+}
+
 func (p *process) setClient(c *client.Client) {
 	p.mu.Lock()
 	old := p.client
@@ -509,18 +532,41 @@ func (p *process) egress() driver.EgressMode {
 	return driver.EgressProxy
 }
 
-// Client returns the client of the running child, or an unavailable error.
-func (p *process) Client() (*client.Client, error) {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	if p.client == nil || p.state != StateReady {
-		msg := fmt.Sprintf("plugin %s is %s", p.spec.m.ID, p.state)
-		if p.lastErr != nil {
-			msg += ": " + p.lastErr.Error()
+// Client returns the client of the running child, starting an idle one
+// and waiting for it (bounded by ctx), or an unavailable error.
+func (p *process) Client(ctx context.Context) (*client.Client, error) {
+	for {
+		p.mu.RLock()
+		st, c, changed, lastErr := p.state, p.client, p.changed, p.lastErr
+		if st == StateReady && c != nil {
+			// Counts as use, so the plugin is not stopped for being
+			// idle before the call goes out.
+			p.activity.touch()
+			p.mu.RUnlock()
+			return c, nil
+		}
+		p.mu.RUnlock()
+		switch st {
+		case StateIdle:
+			p.requestWake()
+		case StateStarting:
+		default:
+			msg := fmt.Sprintf("plugin %s is %s", p.spec.m.ID, st)
+			if lastErr != nil {
+				msg += ": " + lastErr.Error()
+			}
+			return nil, &pluginapi.Error{Code: pluginapi.CodeUnavailable, Message: msg, Retryable: true}
+		}
+		select {
+		case <-changed:
+		case <-ctx.Done():
+			return nil, &pluginapi.Error{
+				Code:      pluginapi.CodeUnavailable,
+				Message:   fmt.Sprintf("plugin %s is starting: %v", p.spec.m.ID, ctx.Err()),
+				Retryable: true,
+			}
 		}
-		return nil, &pluginapi.Error{Code: pluginapi.CodeUnavailable, Message: msg, Retryable: true}
 	}
-	return p.client, nil
 }
 
 // supervise keeps the plugin running: health checks, restart with backoff
@@ -537,6 +583,27 @@ func (p *process) supervise(ctx context.Context, entry string, cur *launched) {
 			p.setState(StateStopped, nil)
 			return
 		}
+		if errors.Is(exitErr, errParked) {
+			logger.Infof(ctx, "[plugin] %s stopped while idle; the next call starts it", p.spec.m.ID)
+			select {
+			case <-ctx.Done():
+				p.setState(StateStopped, nil)
+				return
+			case <-p.wake:
+			}
+			p.setState(StateStarting, nil)
+			next, err := p.launch(ctx, entry)
+			if err == nil {
+				cur = next
+				logger.Infof(ctx, "[plugin] %s started for a call", p.spec.m.ID)
+				continue
+			}
+			if ctx.Err() != nil {
+				p.setState(StateStopped, nil)
+				return
+			}
+			exitErr = err
+		}
 		logger.Warnf(ctx, "[plugin] %s stopped unexpectedly: %v; restarting in %s", p.spec.m.ID, exitErr, backoff)
 		p.setState(StateDegraded, exitErr)
 		for {
@@ -580,6 +647,12 @@ func (p *process) watch(ctx context.Context, cur *launched) error {
 		case <-ctx.Done():
 			stopChild(cur.cmd, cur.exited)
 			return nil
+		case <-p.park:
+			if !p.parkIntended() {
+				continue // a stale signal: the plugin restarted since
+			}
+			stopChild(cur.cmd, cur.exited)
+			return errParked
 		case <-t.C:
 			hctx, cancel := context.WithTimeout(ctx, healthTimeout)
 			err := cur.client.Health(hctx)

internal/plugin/hostpool/hostpool.go

@@ -103,7 +103,8 @@ type Info struct {
 // runs reports whether the host serves a plugin version right now.
 func (i Info) runs(pluginID, version string) bool {
 	for _, p := range i.Plugins {
-		if p.ID == pluginID && p.Version == version && p.State == host.StateReady {
+		// An idle plugin starts on its first call.
+		if p.ID == pluginID && p.Version == version && (p.State == host.StateReady || p.State == host.StateIdle) {
 			return true
 		}
 	}

internal/plugin/hostpool/hostpool_test.go

@@ -258,3 +258,16 @@ func waitFor(t *testing.T, cond func() bool) {
 		time.Sleep(10 * time.Millisecond)
 	}
 }
+
+// A plugin the host stopped for being idle still counts: the gateway starts
+// it on the first call.
+func TestIdlePluginsCountAsRunning(t *testing.T) {
+	i := Info{Plugins: []host.Running{{ID: "acme.py", Version: "1.0.0", State: host.StateIdle}}}
+	if !i.runs("acme.py", "1.0.0") {
+		t.Fatal("an idle plugin must count as running")
+	}
+	i.Plugins[0].State = host.StateDegraded
+	if i.runs("acme.py", "1.0.0") {
+		t.Fatal("a degraded plugin must not")
+	}
+}

internal/plugin/manifest/manifest.go

@@ -106,7 +106,11 @@ type Runtime struct {
 	Entry string `json:"entry,omitempty"     yaml:"entry"`
 	// Singleton allows only one live instance across the cluster (long-lived
 	// connections such as an IM WebSocket).
-	Singleton bool       `json:"singleton,omitempty" yaml:"singleton"`
+	Singleton bool `json:"singleton,omitempty" yaml:"singleton"`
+	// KeepAlive keeps a host plugin running while it is idle, where the
+	// platform stops idle plugins (WEKNORA_PLUGIN_IDLE_TIMEOUT): for
+	// plugins that work between calls, such as background polling.
+	KeepAlive bool       `json:"keepAlive,omitempty" yaml:"keepAlive"`
 	Resources *Resources `json:"resources,omitempty" yaml:"resources"`
 	// Image is the container image of a kubernetes plugin; Port is where
 	// it serves the protocol (default 8080).
@@ -114,6 +118,9 @@ type Runtime struct {
 	Port  int    `json:"port,omitempty"  yaml:"port"`
 }
 
+// Resident reports whether a host plugin must keep running while idle.
+func (r Runtime) Resident() bool { return r.KeepAlive || r.Singleton }
+
 // Resources caps a host plugin process.
 type Resources struct {
 	CPU    string `json:"cpu,omitempty"    yaml:"cpu"`

pluginsdk/README.md

@@ -440,6 +440,12 @@ A few rules the manifest and host enforce:
   caps the process (Linux). Memory is always capped. CPU is capped only
   when the platform delegates a cgroup to WeKnora
   (`WEKNORA_PLUGIN_CGROUP`).
+- **Idle plugins.** A platform may stop a plugin that has had no calls for
+  a while (`WEKNORA_PLUGIN_IDLE_TIMEOUT`; the desktop app does after 10
+  minutes) and start it again on the next call. Keep no state between
+  calls that you cannot rebuild, and do no work in the background. If the
+  plugin must keep running (polling, a long-lived connection), declare
+  `runtime: { ..., keepAlive: true }`.
 - **IDs.** Contribution IDs are qualified with the plugin ID
   (`acme.notes/notes`); for connectors and web search that must stay within
   50 characters.

website-docs/03-features/25-plugins.md

@@ -156,6 +156,10 @@ weknora-plugin verify -pubkey ed25519:... acme-search-1.0.0.wkp
 - 出网流量经宿主的出口代理，只放行插件在 `permissions.egress` 中声明的域名，内网地址一律拒绝。
 - 宿主检查运行的进程与安装的包一致，健康检查失败时按退避重启，并在插件详情中显示为「异常」。
 - Python 插件使用本机的 `python3`（可用 `WEKNORA_PLUGIN_PYTHON` 指定解释器）。Docker app 镜像已带 Python。
+- 空闲回收：设置 `WEKNORA_PLUGIN_IDLE_TIMEOUT`（如 `10m`）后，插件在这段时间内没有任何调用（健康检查不算）就停掉进程，下一次调用时再启动，这次调用会多等一次启动时间（示例插件实测：Go 约 10 ms，Python 约 70 ms，视插件自身的初始化而定）。有调用进行中（包括连接器的流式同步）时不会回收。
+  - 默认不回收；桌面端默认 `10m`。
+  - 停掉的插件仍在「插件管理」中显示为运行中，路由、独立宿主的通告都不受影响；升级时照常先启动新版本校验，再按空闲时间回收。
+  - 插件在两次调用之间要做事（如后台轮询、长连接）时，在 `plugin.yaml` 中声明 `runtime.keepAlive: true`，始终常驻；声明了 `singleton` 的插件同样常驻。
 - 在 Linux 上，插件进程按 `runtime.resources` 限制资源：
   - 内存总是受限。插件没有声明时，按 `WEKNORA_PLUGIN_MEMORY_DEFAULT`（如 `1Gi`）限制，未设置则不限。
   - CPU 需要把一个可写的 cgroup v2 目录委托给 WeKnora，并通过 `WEKNORA_PLUGIN_CGROUP` 指定。设置后，每个插件进程进入各自的子 cgroup。