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)
This commit is contained in:
ZanzyTHEbar
parent
61e9c34165
commit
938b77f9b5
4 changed files with 434 additions and 0 deletions
143
pkg/runtime/bootstrap.go
Normal file
143
pkg/runtime/bootstrap.go
Normal file
|
|
@ -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
|
||||
}
|
||||
102
pkg/runtime/config.go
Normal file
102
pkg/runtime/config.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
45
pkg/runtime/execute.go
Normal file
45
pkg/runtime/execute.go
Normal file
|
|
@ -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())
|
||||
}
|
||||
144
pkg/runtime/runtime_test.go
Normal file
144
pkg/runtime/runtime_test.go
Normal file
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue