feat(exec): implement AsyncExecutor for ExecTool with bus-based background delivery

Signed-off-by: Boris Bliznioukov <blib@mail.com>
This commit is contained in:
Boris Bliznioukov 2026-03-05 10:46:26 +01:00
parent 4bb4200ca0
commit bf1b07d407
No known key found for this signature in database
8 changed files with 179 additions and 135 deletions

View file

@ -140,15 +140,15 @@
12. The cron tool (`pkg/tools/cron.go`) MUST use the same `ExecTool` with the same guard system.
13. The `ExecTool` MUST implement the `AsyncTool` interface (`SetCallback(AsyncCallback)`). When the LLM passes `background=true` and a callback is registered, the command MUST be launched in a goroutine; the result is delivered via the callback. When `background=true` but no callback is registered, execution falls through to synchronous mode. Compile-time interface check: `var _ AsyncTool = (*ExecTool)(nil)`.
13. The `ExecTool` MUST implement the `AsyncExecutor` interface (`ExecuteAsync(ctx, args, cb)`). When the LLM passes `background=true` and a `bus.MessageBus` was injected at construction, the command MUST be launched in a goroutine; the result is delivered via `bus.PublishInbound` with `Channel: "system"`, `SenderID: "exec:<cmd>"`, `ChatID: "<channel>:<chatID>"`. When `background=true` but no bus is available (e.g. cron-created instances), execution falls through to synchronous mode. Compile-time interface check: `var _ AsyncExecutor = (*ExecTool)(nil)`.
14. The implementation MUST use a subpackage structure: `pkg/tools/shell/` contains the core logic (risk classifier, env sanitizer, sandbox, runner) and `pkg/tools/shell_tool.go` is the thin adapter implementing the `Tool` + `AsyncTool` interfaces. Tests live alongside their source in both locations.
14. The implementation MUST use a subpackage structure: `pkg/tools/shell/` contains the core logic (risk classifier, env sanitizer, sandbox, runner) and `pkg/tools/shell_tool.go` is the thin adapter implementing the `Tool` + `AsyncExecutor` interfaces. Tests live alongside their source in both locations.
### 2.3 Migration
- **Config**: Old fields are silently ignored with a logged warning. No config version bump required. Add a migration note to `docs/tools_configuration.md`.
- **Behavioral**: Commands that previously passed regex checks but are genuinely dangerous (variable indirection bypasses) will now be blocked. This is intentional and constitutes the security fix.
- **Backward compatibility**: The `Tool` interface (`Name`, `Description`, `Parameters`, `Execute`) is unchanged. Callers (`ToolRegistry`, `RunToolLoop`, agent instance) require no changes.
- **Backward compatibility**: The `Tool` interface (`Name`, `Description`, `Parameters`, `Execute`) is unchanged. `ExecTool` construction moved from `NewAgentInstance` to `registerSharedTools` (in `loop.go`) where the message bus is available for constructor injection.
---
@ -205,7 +205,7 @@
| Who / what | Impact |
| ------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------- |
| `pkg/tools/shell_tool.go` | Thin adapter: `ExecTool` struct implementing `Tool` + `AsyncTool`. Delegates to `pkg/tools/shell/` subpackage. |
| `pkg/tools/shell_tool.go` | Thin adapter: `ExecTool` struct implementing `Tool` + `AsyncExecutor`. Delegates to `pkg/tools/shell/` subpackage. Bus-injected for background result delivery. |
| `pkg/tools/shell_tool_test.go` | Tests for `ExecTool` sync/async behavior and interface compliance. |
| `pkg/tools/shell_process_unix.go`, `shell_process_windows.go` | Removed. Interpreter manages process lifecycle. |
| `pkg/config/config.go` (`ExecConfig`) | Three fields removed, four fields added. |
@ -229,7 +229,7 @@
| `pkg/tools/shell/sandbox_test.go` | Redirect inside/outside workspace, symlink escape, safe-path exemption. |
| `pkg/tools/shell/runner.go` | `Run` function: parser + interpreter + `ExecHandlers` middleware integration. |
| `pkg/tools/shell/runner_test.go` | End-to-end runner tests: timeout, working dir, env sanitization, pipelines. |
| `pkg/tools/shell_tool.go` | Adapter: `ExecTool` struct, `NewExecToolWithConfig`, `AsyncTool` impl, arg modifier wiring. |
| `pkg/tools/shell_tool.go` | Adapter: `ExecTool` struct, `NewExecToolWithConfig(workDir, restrict, cfg, bus)`, `AsyncExecutor` impl, arg modifier wiring. |
| `pkg/tools/shell_tool_test.go` | `ExecTool` sync/async tests, interface compliance checks. |
| `pkg/tools/cron_exec_test.go` | AC-8: cron-originated `ExecTool` blocks dangerous commands identically to agent-created one; safe commands pass. |
@ -243,7 +243,7 @@
| 4 | Rewrite shell tool: parser + interpreter + middleware (`shell/runner.go` + `shell_tool.go`) | Maintainer | 5, 6 | **Done** |
| 5 | Port all existing test cases + add bypass tests | Maintainer | 7 | **Done** |
| 6 | Update `ExecConfig`, defaults, migration warning | Maintainer | 7 | **Done** |
| 7 | Implement `AsyncTool` on `ExecTool` | Maintainer | — | **Done** |
| 7 | Implement `AsyncExecutor` on `ExecTool` (bus-based background delivery) | Maintainer | — | **Done** |
| 8 | Implement configurable `ArgModifiers` (user-defined, highest-match-wins) | Maintainer | — | **Done** |
| 9 | Fix runner_test PATH resolution for external binaries in sandboxed interpreter | Maintainer | — | **Done** |
| 10 | Update cron tool, docs, config example | Maintainer | — | **Done** |
@ -266,7 +266,7 @@
### Rollback plan
- The change is contained within `pkg/tools/` and `pkg/config/`. Git revert of the implementation commits restores the regex-based system.
- Assumption: no other packages depend on `ExecTool` internals (only the `Tool` interface is public contract). Verified: only `pkg/agent/instance.go` and `pkg/tools/cron.go` call `NewExecToolWithConfig`.
- Assumption: no other packages depend on `ExecTool` internals (only the `Tool` interface is public contract). Verified: only `pkg/agent/loop.go` (via `registerSharedTools`) and `pkg/tools/cron.go` call `NewExecToolWithConfig`.
- If the interpreter causes widespread command failures post-deploy, revert and reconsider the hybrid approach (Alternative #4).
### Decision review date

View file

@ -128,9 +128,30 @@ When `restrict_to_workspace` is enabled (the default), the interpreter's
### Cron Integration
The cron tool creates its own `ExecTool` via `NewExecToolWithConfig`, so
scheduled commands go through the same risk classifier, env sanitization, and
sandbox as agent-originated commands.
The cron tool creates its own `ExecTool` via `NewExecToolWithConfig` (with `nil`
bus), so scheduled commands go through the same risk classifier, env
sanitization, and sandbox as agent-originated commands. Because there is no bus,
cron-executed commands always run synchronously regardless of the `background`
parameter.
### Background Execution
When the LLM passes `background: true`, the exec tool launches the command in a
goroutine and immediately returns a confirmation to the agent. The result is
delivered asynchronously via `bus.PublishInbound` as a system-channel inbound
message:
| Field | Value |
| ---------- | -------------------------------------- |
| `Channel` | `"system"` |
| `SenderID` | `"exec:<command>"` |
| `ChatID` | `"<originChannel>:<originChatID>"` |
| `Content` | stdout/stderr output (or error string) |
This is the same delivery mechanism used by `SpawnTool` for subagent results.
If no message bus is available (e.g. cron-created instances), `background: true`
falls through to synchronous execution.
### Configuration Example
@ -281,7 +302,7 @@ All configuration options can be overridden via environment variables with the f
For example:
- `PICOCLAW_TOOLS_WEB_BRAVE_ENABLED=true`
- `PICOCLAW_TOOLS_EXEC_ENABLE_DENY_PATTERNS=false`
- `PICOCLAW_TOOLS_EXEC_RISK_THRESHOLD=high`
- `PICOCLAW_TOOLS_CRON_EXEC_TIMEOUT_MINUTES=10`
- `PICOCLAW_TOOLS_MCP_ENABLED=true`

View file

@ -2,7 +2,6 @@ package agent
import (
"fmt"
"log"
"os"
"path/filepath"
"regexp"
@ -78,13 +77,6 @@ func NewAgentInstance(
if cfg.Tools.IsToolEnabled("list_dir") {
toolsRegistry.Register(tools.NewListDirTool(workspace, readRestrict, allowReadPaths))
}
if cfg.Tools.IsToolEnabled("exec") {
execTool, err := tools.NewExecToolWithConfig(workspace, restrict, cfg)
if err != nil {
log.Fatalf("Critical error: unable to initialize exec tool: %v", err)
}
toolsRegistry.Register(execTool)
}
if cfg.Tools.IsToolEnabled("edit_file") {
toolsRegistry.Register(tools.NewEditFileTool(workspace, restrict, allowWritePaths))

View file

@ -168,6 +168,18 @@ func registerSharedTools(
agent.Tools.Register(tools.NewSPITool())
}
// Exec tool — created here (not in NewAgentInstance) because it
// needs bus access for background command result delivery.
if cfg.Tools.IsToolEnabled("exec") {
restrict := cfg.Agents.Defaults.RestrictToWorkspace
execTool, err := tools.NewExecToolWithConfig(agent.Workspace, restrict, cfg, msgBus)
if err != nil {
logger.ErrorCF("agent", "Failed to create exec tool", map[string]any{"error": err.Error()})
} else {
agent.Tools.Register(execTool)
}
}
// Message tool
if cfg.Tools.IsToolEnabled("message") {
messageTool := tools.NewMessageTool()

View file

@ -609,6 +609,7 @@ type ArgModifierConfig struct {
}
type ExecConfig struct {
ToolConfig ` envPrefix:"PICOCLAW_TOOLS_EXEC_"`
RiskThreshold string ` json:"risk_threshold" env:"PICOCLAW_TOOLS_EXEC_RISK_THRESHOLD"` // "low"|"medium"|"high"|"critical"; default "medium"
RiskOverrides map[string]string ` json:"risk_overrides" env:"PICOCLAW_TOOLS_EXEC_RISK_OVERRIDES"` // command → level override
ArgModifiers map[string][]ArgModifierConfig ` json:"arg_modifiers" env:"PICOCLAW_TOOLS_EXEC_ARG_MODIFIERS"` // command → argument-aware risk adjustments (extends built-ins)

View file

@ -31,7 +31,7 @@ func NewCronTool(
cronService *cron.CronService, executor JobExecutor, msgBus *bus.MessageBus, workspace string, restrict bool,
execTimeout time.Duration, config *config.Config,
) (*CronTool, error) {
execTool, err := NewExecToolWithConfig(workspace, restrict, config)
execTool, err := NewExecToolWithConfig(workspace, restrict, config, nil)
if err != nil {
return nil, fmt.Errorf("unable to configure exec tool: %w", err)
}

View file

@ -6,22 +6,25 @@ import (
"os"
"time"
"github.com/sipeed/picoclaw/pkg/bus"
"github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/tools/shell"
"github.com/sipeed/picoclaw/pkg/utils"
)
// Compile-time interface checks.
var (
_ Tool = (*ExecTool)(nil)
_ AsyncTool = (*ExecTool)(nil)
_ AsyncExecutor = (*ExecTool)(nil)
)
// ExecTool executes shell commands using an in-process interpreter
// with AST-based risk classification, env sanitization, and file-access sandboxing.
//
// ExecTool implements AsyncTool. When the LLM passes background=true the
// command runs in a goroutine and the result is delivered via the callback
// injected by the tool registry.
// ExecTool implements AsyncExecutor. When the LLM passes background=true,
// the command runs in a goroutine and the result is delivered via
// bus.PublishInbound (re-entering the agent loop). Without a bus,
// background=true falls back to synchronous execution.
type ExecTool struct {
workingDir string
timeout time.Duration
@ -33,19 +36,25 @@ type ExecTool struct {
envAllowlist []string
envSet map[string]string
callback AsyncCallback
bus *bus.MessageBus
}
func NewExecTool(workingDir string, restrict bool) (*ExecTool, error) {
return NewExecToolWithConfig(workingDir, restrict, nil)
return NewExecToolWithConfig(workingDir, restrict, nil, nil)
}
func NewExecToolWithConfig(workingDir string, restrict bool, cfg *config.Config) (*ExecTool, error) {
func NewExecToolWithConfig(
workingDir string,
restrict bool,
cfg *config.Config,
msgBus *bus.MessageBus,
) (*ExecTool, error) {
t := &ExecTool{
workingDir: workingDir,
timeout: 60 * time.Second,
restrictToWorkspace: restrict,
riskThreshold: shell.RiskMedium,
bus: msgBus,
}
if cfg != nil {
@ -101,11 +110,6 @@ func (t *ExecTool) Description() string {
return "Execute a shell command and return its output. Use with caution."
}
// SetCallback implements AsyncTool.
func (t *ExecTool) SetCallback(cb AsyncCallback) {
t.callback = cb
}
func (t *ExecTool) Parameters() map[string]any {
return map[string]any{
"type": "object",
@ -128,9 +132,66 @@ func (t *ExecTool) Parameters() map[string]any {
}
func (t *ExecTool) Execute(ctx context.Context, args map[string]any) *ToolResult {
cfg, err := t.buildConfig(args)
if err != nil {
return err
}
result := shell.Run(ctx, cfg)
return &ToolResult{
ForLLM: result.Output,
ForUser: result.Output,
IsError: result.IsError,
}
}
// ExecuteAsync implements AsyncExecutor. The registry calls this for
// parallel tool dispatch. When background=true and a bus is configured,
// the command runs in a goroutine and the result is delivered via
// bus.PublishInbound (re-entering the agent loop). Without a bus,
// background=true falls back to synchronous execution.
func (t *ExecTool) ExecuteAsync(ctx context.Context, args map[string]any, cb AsyncCallback) *ToolResult {
background, _ := args["background"].(bool)
if !background || t.bus == nil {
return t.Execute(ctx, args)
}
cfg, errResult := t.buildConfig(args)
if errResult != nil {
return errResult
}
channel := ToolChannel(ctx)
chatID := ToolChatID(ctx)
go func() {
result := shell.Run(ctx, cfg)
content := fmt.Sprintf("Background command completed: `%s`\n\n%s",
utils.Truncate(cfg.Command, 100), result.Output)
if result.IsError {
content = fmt.Sprintf("Background command failed: `%s`\n\n%s",
utils.Truncate(cfg.Command, 100), result.Output)
}
pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer pubCancel()
_ = t.bus.PublishInbound(pubCtx, bus.InboundMessage{
Channel: "system",
SenderID: fmt.Sprintf("exec:%s", utils.Truncate(cfg.Command, 50)),
ChatID: fmt.Sprintf("%s:%s", channel, chatID),
Content: content,
})
}()
return AsyncResult(fmt.Sprintf("Running `%s` in background", utils.Truncate(cfg.Command, 100)))
}
// buildConfig extracts args and returns a RunConfig. Returns a *ToolResult
// error if validation fails (bad command, sandbox violation, etc.).
func (t *ExecTool) buildConfig(args map[string]any) (shell.RunConfig, *ToolResult) {
command, ok := args["command"].(string)
if !ok {
return ErrorResult("command is required")
return shell.RunConfig{}, ErrorResult("command is required")
}
cwd := t.workingDir
@ -138,7 +199,7 @@ func (t *ExecTool) Execute(ctx context.Context, args map[string]any) *ToolResult
if t.restrictToWorkspace && t.workingDir != "" {
resolvedWD, err := validatePath(wd, t.workingDir, true)
if err != nil {
return ErrorResult("Command blocked by safety guard (" + err.Error() + ")")
return shell.RunConfig{}, ErrorResult("Command blocked by safety guard (" + err.Error() + ")")
}
cwd = resolvedWD
} else {
@ -153,7 +214,7 @@ func (t *ExecTool) Execute(ctx context.Context, args map[string]any) *ToolResult
}
}
cfg := shell.RunConfig{
return shell.RunConfig{
Command: command,
Dir: cwd,
Timeout: t.timeout,
@ -164,35 +225,7 @@ func (t *ExecTool) Execute(ctx context.Context, args map[string]any) *ToolResult
ExtraArgModifiers: t.argModifiers,
EnvAllowlist: t.envAllowlist,
EnvSet: t.envSet,
}
background, _ := args["background"].(bool)
if background && t.callback != nil {
return t.executeAsync(ctx, cfg)
}
result := shell.Run(ctx, cfg)
return &ToolResult{
ForLLM: result.Output,
ForUser: result.Output,
IsError: result.IsError,
}
}
// executeAsync launches the command in a goroutine and delivers the result
// through the AsyncCallback. The parent ctx is used for cancellation so the
// goroutine respects agent shutdown.
func (t *ExecTool) executeAsync(ctx context.Context, cfg shell.RunConfig) *ToolResult {
cb := t.callback // capture before goroutine
go func() {
result := shell.Run(ctx, cfg)
cb(ctx, &ToolResult{
ForLLM: result.Output,
ForUser: result.Output,
IsError: result.IsError,
})
}()
return AsyncResult(fmt.Sprintf("Running `%s` in background", cfg.Command))
}, nil
}
func (t *ExecTool) SetTimeout(timeout time.Duration) {

View file

@ -5,10 +5,9 @@ import (
"context"
"os"
"strings"
"sync"
"testing"
"time"
"github.com/sipeed/picoclaw/pkg/bus"
"github.com/sipeed/picoclaw/pkg/config"
)
@ -30,123 +29,109 @@ func TestExecTool_SyncExecution(t *testing.T) {
}
}
func TestExecTool_BackgroundWithoutCallback(t *testing.T) {
func TestExecTool_BackgroundWithoutBus(t *testing.T) {
tool, err := NewExecTool(t.TempDir(), false)
if err != nil {
t.Fatal(err)
}
// background=true but no callback set → falls through to sync
result := tool.Execute(context.Background(), map[string]any{
// background=true but no bus → ExecuteAsync falls back to synchronous
cb := func(_ context.Context, _ *ToolResult) {
t.Error("callback should not be invoked for sync fallback")
}
result := tool.ExecuteAsync(context.Background(), map[string]any{
"command": "echo fallback",
"background": true,
})
}, cb)
if result.Async {
t.Error("should fall back to sync when no callback is set")
t.Error("should fall back to sync when no bus is configured")
}
if result.IsError {
t.Fatalf("expected success: %s", result.ForLLM)
}
}
func TestExecTool_BackgroundWithCallback(t *testing.T) {
tool, err := NewExecTool(t.TempDir(), false)
func TestExecTool_BackgroundWithBus(t *testing.T) {
msgBus := bus.NewMessageBus()
defer msgBus.Close()
tool, err := NewExecToolWithConfig(t.TempDir(), false, nil, msgBus)
if err != nil {
t.Fatal(err)
}
var (
mu sync.Mutex
received *ToolResult
)
done := make(chan struct{})
ctx := WithToolContext(context.Background(), "telegram", "chat-123")
tool.SetCallback(func(_ context.Context, r *ToolResult) {
mu.Lock()
received = r
mu.Unlock()
close(done)
})
result := tool.Execute(context.Background(), map[string]any{
"command": "echo async_output",
cb := func(_ context.Context, _ *ToolResult) {}
result := tool.ExecuteAsync(ctx, map[string]any{
"command": "echo bg_output",
"background": true,
})
}, cb)
if !result.Async {
t.Fatal("expected async result")
}
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("timed out waiting for async callback")
// Consume the inbound message published by the background goroutine.
msg, ok := msgBus.ConsumeInbound(ctx)
if !ok {
t.Fatal("expected inbound message from background exec")
}
mu.Lock()
defer mu.Unlock()
if received == nil {
t.Fatal("callback was never invoked")
if msg.Channel != "system" {
t.Errorf("expected channel=system, got %q", msg.Channel)
}
if received.IsError {
t.Fatalf("async command failed: %s", received.ForLLM)
if msg.ChatID != "telegram:chat-123" {
t.Errorf("expected chatID=telegram:chat-123, got %q", msg.ChatID)
}
if !strings.Contains(msg.Content, "bg_output") {
t.Errorf("expected output in message content: %s", msg.Content)
}
if !strings.Contains(msg.Content, "completed") {
t.Errorf("expected 'completed' in message content: %s", msg.Content)
}
}
func TestExecTool_BackgroundBlockedCommand(t *testing.T) {
tool, err := NewExecTool(t.TempDir(), false)
msgBus := bus.NewMessageBus()
defer msgBus.Close()
tool, err := NewExecToolWithConfig(t.TempDir(), false, nil, msgBus)
if err != nil {
t.Fatal(err)
}
var (
mu sync.Mutex
received *ToolResult
)
done := make(chan struct{})
ctx := WithToolContext(context.Background(), "telegram", "chat-123")
tool.SetCallback(func(_ context.Context, r *ToolResult) {
mu.Lock()
received = r
mu.Unlock()
close(done)
})
result := tool.Execute(context.Background(), map[string]any{
cb := func(_ context.Context, _ *ToolResult) {}
result := tool.ExecuteAsync(ctx, map[string]any{
"command": "sudo rm -rf /",
"background": true,
})
}, cb)
if !result.Async {
t.Fatal("expected async result even for blocked commands")
}
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("timed out waiting for async callback")
msg, ok := msgBus.ConsumeInbound(ctx)
if !ok {
t.Fatal("expected inbound message from background exec")
}
mu.Lock()
defer mu.Unlock()
if received == nil {
t.Fatal("callback was never invoked")
}
if !received.IsError {
t.Error("blocked command should report error via callback")
if !strings.Contains(msg.Content, "failed") {
t.Errorf("expected 'failed' in message content: %s", msg.Content)
}
}
func TestExecTool_ImplementsAsyncTool(t *testing.T) {
func TestExecTool_ImplementsAsyncExecutor(t *testing.T) {
tool, err := NewExecTool(t.TempDir(), false)
if err != nil {
t.Fatal(err)
}
var _ AsyncTool = tool // compile-time check
var _ AsyncExecutor = tool // compile-time check
}
// captureStdout runs fn and returns whatever it wrote to os.Stdout.
@ -260,7 +245,7 @@ func TestNewExecToolWithConfig_EnableDenyPatternsFalseWarning(t *testing.T) {
out := captureStdout(t, func() {
cfg := &config.Config{}
cfg.Tools.Exec.EnableDenyPatterns = boolPtr(false)
_, err := NewExecToolWithConfig(t.TempDir(), false, cfg)
_, err := NewExecToolWithConfig(t.TempDir(), false, cfg, nil)
if err != nil {
t.Fatal(err)
}