From 938b77f9b5c4f57091eaf92ecde900ef8ddd4fae Mon Sep 17 00:00:00 2001 From: ZanzyTHEbar Date: Sat, 21 Feb 2026 00:39:10 +0000 Subject: [PATCH] feat(runtime): extract Bootstrap/RuntimeHandle/RunPrompt into pkg/runtime Introduces a reusable runtime package that encapsulates the full lifecycle of a picoclaw agent session: provider creation, model wiring, message bus setup, outbound drain, and agent loop initialization. pkg/runtime/bootstrap.go - Bootstrap() creates a RuntimeHandle from a Config and BootstrapOptions - BootstrapOptions.WrapModel allows callers (e.g. eval-runner) to inject an instrumented model wrapper without coupling to agent internals - OutboundMode enum (None/Consume/Drop/Callback) controls how outbound messages are drained so producers never block - RuntimeHandle.Close() tears down agentLoop, cancels context, and waits for the outbound goroutine to exit pkg/runtime/config.go - ResolveBaseConfigPath() prefers XDG (~/.config/picoclaw/config.json) then falls back to legacy (~/.picoclaw/config.json) - LoadResolvedConfig() loads base config then applies an optional overlay - LoadEvalConfig() convenience wrapper for eval-runner (reads PICOCLAW_EVAL_CONFIG env var as overlay path) - EnsureMinProviderTimeout() sets a floor on all provider timeout fields pkg/runtime/execute.go - RunPrompt() runs a single prompt through an existing RuntimeHandle and returns a RunResult with output, error, duration, and session key - NewSessionKey() generates deterministic session keys from a prefix + time pkg/runtime/runtime_test.go - Tests for XDG vs legacy path resolution - Tests for overlay config merging and base value preservation - Tests for provider timeout floor enforcement - Tests for outbound drain modes (drop/consume/callback) --- pkg/runtime/bootstrap.go | 143 +++++++++++++++++++++++++++++++++++ pkg/runtime/config.go | 102 +++++++++++++++++++++++++ pkg/runtime/execute.go | 45 +++++++++++ pkg/runtime/runtime_test.go | 144 ++++++++++++++++++++++++++++++++++++ 4 files changed, 434 insertions(+) create mode 100644 pkg/runtime/bootstrap.go create mode 100644 pkg/runtime/config.go create mode 100644 pkg/runtime/execute.go create mode 100644 pkg/runtime/runtime_test.go diff --git a/pkg/runtime/bootstrap.go b/pkg/runtime/bootstrap.go new file mode 100644 index 000000000..f42e4b98b --- /dev/null +++ b/pkg/runtime/bootstrap.go @@ -0,0 +1,143 @@ +package runtime + +import ( + "context" + "fmt" + "time" + + fantasy "charm.land/fantasy" + "github.com/sipeed/picoclaw/pkg/agent" + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" + picofantasy "github.com/sipeed/picoclaw/pkg/fantasy" +) + +type OutboundMode string + +const ( + OutboundModeNone OutboundMode = "none" + OutboundModeConsume OutboundMode = "consume" + OutboundModeDrop OutboundMode = "drop" + OutboundModeCallback OutboundMode = "callback" +) + +type BootstrapOptions struct { + Timeout time.Duration + OutboundMode OutboundMode + OutboundCallback func(bus.OutboundMessage) + WrapModel func(fantasy.LanguageModel) fantasy.LanguageModel +} + +type RuntimeHandle struct { + ctx context.Context + cancel context.CancelFunc + agentLoop *agent.AgentLoop + msgBus *bus.MessageBus + outDone chan struct{} +} + +func (h *RuntimeHandle) Context() context.Context { + return h.ctx +} + +func (h *RuntimeHandle) AgentLoop() *agent.AgentLoop { + return h.agentLoop +} + +func (h *RuntimeHandle) MessageBus() *bus.MessageBus { + return h.msgBus +} + +func (h *RuntimeHandle) Close() { + if h.agentLoop != nil { + h.agentLoop.Stop() + } + if h.cancel != nil { + h.cancel() + } + if h.outDone != nil { + <-h.outDone + } +} + +func Bootstrap(parent context.Context, cfg *config.Config, opts BootstrapOptions) (*RuntimeHandle, error) { + if parent == nil { + parent = context.Background() + } + + ctx, cancel := withExecutionContext(parent, opts.Timeout) + + provider, err := picofantasy.CreateProvider(cfg) + if err != nil { + cancel() + return nil, fmt.Errorf("provider error: %w", err) + } + + model, err := provider.LanguageModel(ctx, picofantasy.ModelID(cfg)) + if err != nil { + cancel() + return nil, fmt.Errorf("model error: %w", err) + } + + if opts.WrapModel != nil { + model = opts.WrapModel(model) + } + + msgBus := bus.NewMessageBus() + outDone := startOutbound(msgBus, ctx, opts) + + agentLoop, err := agent.NewAgentLoop(ctx, cfg, msgBus, model) + if err != nil { + cancel() + if outDone != nil { + <-outDone + } + return nil, fmt.Errorf("agent loop init error: %w", err) + } + + return &RuntimeHandle{ + ctx: ctx, + cancel: cancel, + agentLoop: agentLoop, + msgBus: msgBus, + outDone: outDone, + }, nil +} + +func withExecutionContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { + if timeout > 0 { + return context.WithTimeout(parent, timeout) + } + return context.WithCancel(parent) +} + +func startOutbound(msgBus *bus.MessageBus, ctx context.Context, opts BootstrapOptions) chan struct{} { + mode := opts.OutboundMode + if mode == "" || mode == OutboundModeNone { + return nil + } + + done := make(chan struct{}) + go func() { + defer close(done) + for { + msg, ok := msgBus.SubscribeOutbound(ctx) + if !ok { + return + } + + switch mode { + case OutboundModeConsume, OutboundModeDrop: + // Intentionally no-op: consume and discard so producers never block. + case OutboundModeCallback: + if opts.OutboundCallback != nil { + opts.OutboundCallback(msg) + } + default: + // Unknown modes degrade safely to consume-and-discard. + } + } + }() + + return done +} diff --git a/pkg/runtime/config.go b/pkg/runtime/config.go new file mode 100644 index 000000000..b4db2fd8a --- /dev/null +++ b/pkg/runtime/config.go @@ -0,0 +1,102 @@ +package runtime + +import ( + "fmt" + "os" + "path/filepath" + "time" + + "github.com/sipeed/picoclaw/pkg/config" +) + +const EvalConfigEnvVar = "PICOCLAW_EVAL_CONFIG" + +type LoadConfigOptions struct { + BaseConfigPath string + OverlayConfigPath string + MinProviderTimeout time.Duration +} + +func ResolveBaseConfigPath() string { + // Prefer XDG standard path (~/.config/picoclaw/config.json) when present. + if xdgPath, err := config.DefaultConfigPath(); err == nil { + if _, statErr := os.Stat(xdgPath); statErr == nil { + return xdgPath + } + } + + home, _ := os.UserHomeDir() + legacy := filepath.Join(home, ".picoclaw", "config.json") + if _, err := os.Stat(legacy); err == nil { + return legacy + } + + // Neither exists; return XDG path if resolvable so defaults still load. + if xdgPath, err := config.DefaultConfigPath(); err == nil { + return xdgPath + } + return legacy +} + +func LoadResolvedConfig(opts LoadConfigOptions) (*config.Config, error) { + basePath := opts.BaseConfigPath + if basePath == "" { + basePath = ResolveBaseConfigPath() + } + + cfg, err := config.LoadConfig(basePath) + if err != nil { + return nil, fmt.Errorf("load base config: %w", err) + } + + if opts.OverlayConfigPath != "" { + if err := config.OverlayConfigFile(cfg, opts.OverlayConfigPath); err != nil { + return nil, fmt.Errorf("load config overlay: %w", err) + } + } + + EnsureMinProviderTimeout(cfg, opts.MinProviderTimeout) + return cfg, nil +} + +func LoadEvalConfig(minTimeout time.Duration) (*config.Config, error) { + return LoadResolvedConfig(LoadConfigOptions{ + OverlayConfigPath: os.Getenv(EvalConfigEnvVar), + MinProviderTimeout: minTimeout, + }) +} + +func EnsureMinProviderTimeout(cfg *config.Config, minTimeout time.Duration) { + if cfg == nil || minTimeout <= 0 { + return + } + minSeconds := int(minTimeout.Seconds()) + if minSeconds <= 0 { + return + } + + set := func(p *config.ProviderConfig) { + if p.Timeout == 0 || p.Timeout < minSeconds { + p.Timeout = minSeconds + } + } + + providers := []*config.ProviderConfig{ + &cfg.Providers.Anthropic, + &cfg.Providers.OpenAI.ProviderConfig, + &cfg.Providers.OpenRouter, + &cfg.Providers.Groq, + &cfg.Providers.Zhipu, + &cfg.Providers.VLLM, + &cfg.Providers.Gemini, + &cfg.Providers.Nvidia, + &cfg.Providers.Ollama, + &cfg.Providers.Moonshot, + &cfg.Providers.ShengSuanYun, + &cfg.Providers.DeepSeek, + &cfg.Providers.GitHubCopilot, + } + for _, p := range providers { + set(p) + } +} diff --git a/pkg/runtime/execute.go b/pkg/runtime/execute.go new file mode 100644 index 000000000..247d19f2b --- /dev/null +++ b/pkg/runtime/execute.go @@ -0,0 +1,45 @@ +package runtime + +import ( + "context" + "fmt" + "time" +) + +type RunResult struct { + Output string + Error string + SessionKey string + Duration time.Duration +} + +func RunPrompt(ctx context.Context, handle *RuntimeHandle, prompt, sessionKey string) RunResult { + start := time.Now() + result := RunResult{SessionKey: sessionKey} + + if handle == nil || handle.AgentLoop() == nil { + result.Error = "runtime handle is not initialized" + result.Duration = time.Since(start) + return result + } + + runCtx := ctx + if runCtx == nil { + runCtx = handle.Context() + } + + output, err := handle.AgentLoop().ProcessDirect(runCtx, prompt, sessionKey) + result.Output = output + if err != nil { + result.Error = err.Error() + } + result.Duration = time.Since(start) + return result +} + +func NewSessionKey(prefix string, now time.Time) string { + if prefix == "" { + prefix = "session" + } + return fmt.Sprintf("%s:%d", prefix, now.UnixNano()) +} diff --git a/pkg/runtime/runtime_test.go b/pkg/runtime/runtime_test.go new file mode 100644 index 000000000..ff8b98bde --- /dev/null +++ b/pkg/runtime/runtime_test.go @@ -0,0 +1,144 @@ +package runtime + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestResolveBaseConfigPath_PrefersXDGOverLegacy(t *testing.T) { + home := t.TempDir() + xdg := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", xdg) + + xdgPath := filepath.Join(xdg, "picoclaw", "config.json") + legacyPath := filepath.Join(home, ".picoclaw", "config.json") + require.NoError(t, os.MkdirAll(filepath.Dir(xdgPath), 0o755)) + require.NoError(t, os.MkdirAll(filepath.Dir(legacyPath), 0o755)) + require.NoError(t, os.WriteFile(xdgPath, []byte(`{}`), 0o644)) + require.NoError(t, os.WriteFile(legacyPath, []byte(`{}`), 0o644)) + + got := ResolveBaseConfigPath() + assert.Equal(t, xdgPath, got) +} + +func TestResolveBaseConfigPath_FallsBackToLegacyWhenXDGMissing(t *testing.T) { + home := t.TempDir() + xdg := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("XDG_CONFIG_HOME", xdg) + + legacyPath := filepath.Join(home, ".picoclaw", "config.json") + require.NoError(t, os.MkdirAll(filepath.Dir(legacyPath), 0o755)) + require.NoError(t, os.WriteFile(legacyPath, []byte(`{}`), 0o644)) + + got := ResolveBaseConfigPath() + assert.Equal(t, legacyPath, got) +} + +func TestLoadResolvedConfig_AppliesOverlayAndKeepsBaseValues(t *testing.T) { + dir := t.TempDir() + basePath := filepath.Join(dir, "base.json") + overlayPath := filepath.Join(dir, "overlay.json") + + base := []byte(`{ + "providers": {"openai": {"api_key": "base-key"}}, + "agents": {"defaults": {"restrict_to_sandbox": false}} + }`) + overlay := []byte(`{ + "agents": {"defaults": {"restrict_to_sandbox": true}} + }`) + require.NoError(t, os.WriteFile(basePath, base, 0o644)) + require.NoError(t, os.WriteFile(overlayPath, overlay, 0o644)) + + cfg, err := LoadResolvedConfig(LoadConfigOptions{ + BaseConfigPath: basePath, + OverlayConfigPath: overlayPath, + }) + require.NoError(t, err) + assert.Equal(t, "base-key", cfg.Providers.OpenAI.APIKey) + assert.True(t, cfg.Agents.Defaults.RestrictToSandbox) +} + +func TestEnsureMinProviderTimeout_SetsFloor(t *testing.T) { + dir := t.TempDir() + basePath := filepath.Join(dir, "base.json") + require.NoError(t, os.WriteFile(basePath, []byte(`{"providers":{"openai":{"timeout":0}}}`), 0o644)) + + cfg, err := LoadResolvedConfig(LoadConfigOptions{ + BaseConfigPath: basePath, + MinProviderTimeout: 180 * time.Second, + }) + require.NoError(t, err) + assert.Equal(t, 180, cfg.Providers.OpenAI.Timeout) +} + +func TestStartOutbound_DropAndConsumeDoNotBlockPublishers(t *testing.T) { + modes := []OutboundMode{OutboundModeDrop, OutboundModeConsume} + for _, mode := range modes { + t.Run(string(mode), func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + msgBus := bus.NewMessageBus() + done := startOutbound(msgBus, ctx, BootstrapOptions{OutboundMode: mode}) + require.NotNil(t, done) + + publishDone := make(chan struct{}) + go func() { + defer close(publishDone) + for i := 0; i < 300; i++ { + msgBus.PublishOutbound(bus.OutboundMessage{ + Channel: "test", + ChatID: "test", + Content: "payload", + }) + } + }() + + select { + case <-publishDone: + case <-time.After(2 * time.Second): + t.Fatalf("publishing outbound messages blocked under mode=%s", mode) + } + }) + } +} + +func TestStartOutbound_CallbackReceivesMessages(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + msgBus := bus.NewMessageBus() + received := make(chan bus.OutboundMessage, 1) + done := startOutbound(msgBus, ctx, BootstrapOptions{ + OutboundMode: OutboundModeCallback, + OutboundCallback: func(msg bus.OutboundMessage) { + select { + case received <- msg: + default: + } + }, + }) + require.NotNil(t, done) + + msgBus.PublishOutbound(bus.OutboundMessage{ + Channel: "cli", + ChatID: "direct", + Content: "hello", + }) + + select { + case got := <-received: + assert.Equal(t, "hello", got.Content) + case <-time.After(2 * time.Second): + t.Fatal("did not receive callback outbound message") + } +}