opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge 56fe6ac596 into 223ebdf0c7

Browse source
This commit is contained in:
Anton Bogdanovich 2026-05-12 20:50:02 +08:00 • committed by GitHub
commit 3c1ebe65fa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 414 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

88
pkg/agent/eventbus_test.go
View file

@ -699,6 +699,7 @@ type asyncFollowUpTool struct {
name string
followUpText string
completionSig chan struct{}
deliveryMode tools.AsyncDeliveryMode
}
func (t *asyncFollowUpTool) Name() string {
@ -726,7 +727,11 @@ func (t *asyncFollowUpTool) ExecuteAsync(
cb tools.AsyncCallback,
) *tools.ToolResult {
go func() {
cb(ctx, &tools.ToolResult{ForLLM: t.followUpText})
res := &tools.ToolResult{ForLLM: t.followUpText}
if t.deliveryMode != "" {
res.WithAsyncDelivery(t.deliveryMode)
}
cb(ctx, res)
if t.completionSig != nil {
close(t.completionSig)
}
@ -738,3 +743,84 @@ var (
_ tools.Tool = (*mockCustomTool)(nil)
_ tools.AsyncExecutor = (*asyncFollowUpTool)(nil)
)
func TestAgentLoop_AsyncToolUserOnly_DoesNotEmitFollowUpQueued(t *testing.T) {
tmpDir := t.TempDir()
cfg := &config.Config{
Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{
Workspace: tmpDir,
ModelName: "test-model",
MaxTokens: 4096,
MaxToolIterations: 10,
},
},
}
provider := &toolCallProvider{
toolCalls: []providers.ToolCall{
{
ID: "call_async_1",
Type: "function",
Name: "async_followup_user_only",
Function: &providers.FunctionCall{
Name: "async_followup_user_only",
Arguments: "{}",
},
Arguments: map[string]any{},
},
},
finalResp: "async launched",
}
msgBus := bus.NewMessageBus()
al := NewAgentLoop(cfg, msgBus, provider)
doneCh := make(chan struct{})
al.RegisterTool(&asyncFollowUpTool{
name: "async_followup_user_only",
followUpText: "background result",
completionSig: doneCh,
deliveryMode: tools.AsyncDeliveryUserOnly,
})
defaultAgent := al.registry.GetDefaultAgent()
if defaultAgent == nil {
t.Fatal("expected default agent")
}
runtimeCh, closeRuntimeEvents := subscribeRuntimeEventsForTest(
t,
al,
8,
runtimeevents.KindAgentFollowUpQueued,
)
defer closeRuntimeEvents()
resp, err := al.runAgentLoop(context.Background(), defaultAgent, processOptions{
SessionKey: "session-1",
Channel: "cli",
ChatID: "direct",
UserMessage: "run async tool",
DefaultResponse: defaultResponse,
EnableSummary: false,
SendResponse: false,
})
if err != nil {
t.Fatalf("runAgentLoop failed: %v", err)
}
if resp != "async launched" {
t.Fatalf("expected final response 'async launched', got %q", resp)
}
select {
case <-doneCh:
case <-time.After(2 * time.Second):
t.Fatal("timeout waiting for async tool completion")
}
select {
case evt := <-runtimeCh:
t.Fatalf("unexpected follow-up queued event: %+v", evt)
case <-time.After(200 * time.Millisecond):
}
}

42
pkg/agent/pipeline_execute.go
View file

@ -103,6 +103,42 @@ func inferSkillNamesFromToolCall(ts *turnState, toolName string, toolArgs map[st
return names
}
func effectiveAsyncToolResultDelivery(result *tools.ToolResult) tools.AsyncDeliveryMode {
if result == nil || result.AsyncDelivery == "" {
return tools.AsyncDeliveryUserAndParent
}
return result.AsyncDelivery
}
func shouldPublishAsyncToolResultToUser(result *tools.ToolResult) bool {
if result == nil {
return false
}
switch effectiveAsyncToolResultDelivery(result) {
case tools.AsyncDeliveryParentOnly:
return false
default:
return !result.Silent && result.ForUser != ""
}
}
func shouldQueueAsyncToolResultForParent(result *tools.ToolResult) bool {
if result == nil {
return false
}
content := result.ContentForLLM()
if content == "" {
return false
}
switch effectiveAsyncToolResultDelivery(result) {
case tools.AsyncDeliveryUserOnly:
return false
case tools.AsyncDeliveryParentOnly, tools.AsyncDeliveryUserAndParent:
return true
}
return true
}
// ExecuteTools executes the tool loop, handling BeforeTool/ApproveTool/AfterTool hooks,
// tool execution with async callbacks, media delivery, and steering injection.
// Returns ToolControl indicating what the coordinator should do next:
@ -474,17 +510,17 @@ toolLoop:
toolCallID := tc.ID
asyncToolName := toolName
asyncCallback := func(_ context.Context, result *tools.ToolResult) {
if !result.Silent && result.ForUser != "" {
if shouldPublishAsyncToolResultToUser(result) {
outCtx, outCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer outCancel()
_ = al.bus.PublishOutbound(outCtx, outboundMessageForTurn(ts, result.ForUser))
}
content := result.ContentForLLM()
if content == "" {
if !shouldQueueAsyncToolResultForParent(result) {
return
}
content := result.ContentForLLM()
content = al.cfg.FilterSensitiveData(content)
logger.InfoCF("agent", "Async tool completed, publishing result",

15
pkg/agent/steering_test.go
View file

@ -935,11 +935,24 @@ func TestAgentLoop_Run_QueuedVoiceMessageIsTranscribedBeforeSteering(t *testing.
subCtx, subCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer subCancel()
var out1 bus.OutboundMessage
select {
case <-msgBus.OutboundChan():
case out1 = <-msgBus.OutboundChan():
case <-subCtx.Done():
t.Fatal("expected outbound response")
}
if out1.Content != "continued response" {
t.Fatalf("expected continued response, got %q", out1.Content)
}
noExtraCtx, cancelNoExtra := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer cancelNoExtra()
select {
case out2 := <-msgBus.OutboundChan():
t.Fatalf("expected stale direct response to be suppressed, got extra outbound %q", out2.Content)
case <-noExtraCtx.Done():
}
cancelRun()
select {

11
pkg/tools/result_test.go
View file

@ -56,6 +56,17 @@ func TestAsyncResult(t *testing.T) {
if !result.Async {
t.Error("Expected Async to be true")
}
if result.AsyncDelivery != "" {
t.Errorf("Expected empty AsyncDelivery by default, got %q", result.AsyncDelivery)
}
}
func TestToolResultWithAsyncDelivery(t *testing.T) {
result := AsyncResult("async task started").WithAsyncDelivery(AsyncDeliveryUserOnly)
if result.AsyncDelivery != AsyncDeliveryUserOnly {
t.Fatalf("AsyncDelivery = %q, want %q", result.AsyncDelivery, AsyncDeliveryUserOnly)
}
}
func TestErrorResult(t *testing.T) {

24
pkg/tools/shared/result.go
View file

@ -12,6 +12,14 @@ const (
ArtifactPathsLLMNote = "Use `send_file` with one of these paths to send it to the user, or use file/exec tools to save it inside the workspace if requested."
)
type AsyncDeliveryMode string
const (
AsyncDeliveryUserOnly AsyncDeliveryMode = "user_only"
AsyncDeliveryParentOnly AsyncDeliveryMode = "parent_only"
AsyncDeliveryUserAndParent AsyncDeliveryMode = "user_and_parent"
)
// ToolResult represents the structured return value from tool execution.
// It provides clear semantics for different types of results and supports
// async operations, user-facing messages, and error handling.
@ -37,6 +45,16 @@ type ToolResult struct {
// When true, the tool will complete later and notify via callback.
Async bool `json:"async"`
// AsyncDelivery controls how the final async result should be routed when
// the background work completes.
//
// Empty means "use runtime default behavior".
// Supported values:
// - user_only
// - parent_only
// - user_and_parent
AsyncDelivery AsyncDeliveryMode `json:"async_delivery,omitempty"`
// Err is the underlying error (not JSON serialized).
// Used for internal error handling and logging.
Err error `json:"-"`
@ -221,3 +239,9 @@ func (tr *ToolResult) WithResponseHandled() *ToolResult {
tr.ResponseHandled = true
return tr
}
// WithAsyncDelivery sets the async delivery policy for this tool result.
func (tr *ToolResult) WithAsyncDelivery(mode AsyncDeliveryMode) *ToolResult {
tr.AsyncDelivery = mode
return tr
}

6
pkg/tools/shared_facade.go
View file

@ -36,6 +36,10 @@ const (
ToolPromptSlotMCP = toolshared.ToolPromptSlotMCP
ToolPromptSourceRegistry = toolshared.ToolPromptSourceRegistry
ToolPromptSourceDiscovery = toolshared.ToolPromptSourceDiscovery
AsyncDeliveryUserOnly = toolshared.AsyncDeliveryUserOnly
AsyncDeliveryParentOnly = toolshared.AsyncDeliveryParentOnly
AsyncDeliveryUserAndParent = toolshared.AsyncDeliveryUserAndParent
)
func WithToolContext(ctx context.Context, channel, chatID string) context.Context {
@ -101,6 +105,8 @@ func SilentResult(forLLM string) *ToolResult {
return toolshared.SilentResult(forLLM)
}
type AsyncDeliveryMode = toolshared.AsyncDeliveryMode
func AsyncResult(forLLM string) *ToolResult {
return toolshared.AsyncResult(forLLM)
}

133
pkg/tools/spawn.go
View file

@ -7,6 +7,7 @@ import (
)
type SpawnTool struct {
manager *SubagentManager
spawner SubTurnSpawner
defaultModel string
maxTokens int
@ -22,6 +23,7 @@ func NewSpawnTool(manager *SubagentManager) *SpawnTool {
return &SpawnTool{}
}
return &SpawnTool{
manager: manager,
defaultModel: manager.defaultModel,
maxTokens: manager.maxTokens,
temperature: manager.temperature,
@ -31,6 +33,27 @@ func NewSpawnTool(manager *SubagentManager) *SpawnTool {
// SetSpawner sets the SubTurnSpawner for direct sub-turn execution.
func (t *SpawnTool) SetSpawner(spawner SubTurnSpawner) {
t.spawner = spawner
if t.manager != nil && spawner != nil {
t.manager.SetSpawner(func(
ctx context.Context,
task, label, agentID string,
tools *ToolRegistry,
maxTokens int,
temperature float64,
hasMaxTokens, hasTemperature bool,
) (*ToolResult, error) {
return spawner.SpawnSubTurn(ctx, SubTurnConfig{
TargetAgentID: strings.TrimSpace(agentID),
Model: t.defaultModel,
Tools: nil,
SystemPrompt: buildSpawnSystemPrompt(task, label),
MaxTokens: maxTokens,
Temperature: temperature,
Async: false,
Critical: true,
})
})
}
}
func (t *SpawnTool) Name() string {
@ -38,7 +61,7 @@ func (t *SpawnTool) Name() string {
}
func (t *SpawnTool) Description() string {
return "Spawn a subagent to handle a task in the background. Use this for complex or time-consuming tasks that can run independently. The subagent will complete the task and report back when done."
return "Spawn a subagent to handle a task in the background. Use this for complex or time-consuming tasks that can run independently. The subagent will complete the task and report back when done. Optional delivery_mode controls whether the final async result goes to the user, the parent agent, or both."
}
func (t *SpawnTool) Parameters() map[string]any {
@ -57,6 +80,15 @@ func (t *SpawnTool) Parameters() map[string]any {
"type": "string",
"description": "Optional target agent ID to delegate the task to",
},
"delivery_mode": map[string]any{
"type": "string",
"description": "Optional async result routing policy: user_only, parent_only, or user_and_parent. Defaults to user_only.",
"enum": []string{
string(AsyncDeliveryUserOnly),
string(AsyncDeliveryParentOnly),
string(AsyncDeliveryUserAndParent),
},
},
},
"required": []string{"task"},
}
@ -93,6 +125,10 @@ func (t *SpawnTool) execute(
label, _ := args["label"].(string)
agentID, _ := args["agent_id"].(string)
targetAgentID := strings.TrimSpace(agentID)
deliveryMode, err := parseSpawnDeliveryMode(args["delivery_mode"])
if err != nil {
return ErrorResult(err.Error()).WithError(err)
}
// Check allowlist if targeting a specific agent
if targetAgentID != "" && t.allowlistCheck != nil {
@ -101,16 +137,58 @@ func (t *SpawnTool) execute(
}
}
// Build system prompt for spawned subagent
systemPrompt := fmt.Sprintf(
`You are a spawned subagent running in the background. Complete the given task independently and report back when done.
// Preferred path: route through SubagentManager so spawn_status and
// background execution share the same task registry.
if t.manager != nil {
wrappedCallback := cb
if cb != nil {
wrappedCallback = func(cbCtx context.Context, res *ToolResult) {
if res != nil {
res.WithAsyncDelivery(deliveryMode)
}
cb(cbCtx, res)
}
}
ack, err := t.manager.Spawn(
ctx,
task,
label,
strings.TrimSpace(agentID),
ToolChannel(ctx),
ToolChatID(ctx),
wrappedCallback,
)
if err != nil {
return ErrorResult(fmt.Sprintf("Spawn failed: %v", err)).WithError(err)
}
return AsyncResult(ack)
}
Task: %s`,
task,
)
// Fallback: manager not configured
return ErrorResult("Subagent manager not configured")
}
func parseSpawnDeliveryMode(raw any) (AsyncDeliveryMode, error) {
if raw == nil {
return AsyncDeliveryUserOnly, nil
}
value, ok := raw.(string)
if !ok {
return "", fmt.Errorf("delivery_mode must be a string")
}
switch AsyncDeliveryMode(strings.TrimSpace(value)) {
case AsyncDeliveryUserOnly, AsyncDeliveryParentOnly, AsyncDeliveryUserAndParent:
return AsyncDeliveryMode(strings.TrimSpace(value)), nil
case "":
return AsyncDeliveryUserOnly, nil
default:
return "", fmt.Errorf("delivery_mode must be one of: user_only, parent_only, user_and_parent")
}
}
func buildSpawnSystemPrompt(task, label string) string {
if label != "" {
systemPrompt = fmt.Sprintf(
return fmt.Sprintf(
`You are a spawned subagent labeled "%s" running in the background. Complete the given task independently and report back when done.
Task: %s`,
@ -118,38 +196,9 @@ Task: %s`,
task,
)
}
// Use spawner if available (direct SpawnSubTurn call)
if t.spawner != nil {
// Launch async sub-turn in goroutine
go func() {
result, err := t.spawner.SpawnSubTurn(ctx, SubTurnConfig{
Model: t.defaultModel,
Tools: nil, // Will inherit from parent via context
SystemPrompt: systemPrompt,
MaxTokens: t.maxTokens,
Temperature: t.temperature,
Async: true, // Async execution
Critical: true, // Background spawn should survive parent turn completion
TargetAgentID: targetAgentID,
})
if err != nil {
result = ErrorResult(fmt.Sprintf("Spawn failed: %v", err)).WithError(err)
}
// Call callback if provided
if cb != nil {
cb(ctx, result)
}
}()
// Return immediate acknowledgment
if label != "" {
return AsyncResult(fmt.Sprintf("Spawned subagent '%s' for task: %s", label, task))
}
return AsyncResult(fmt.Sprintf("Spawned subagent for task: %s", task))
}
// Fallback: spawner not configured
return ErrorResult("Subagent manager not configured")
return fmt.Sprintf(
`You are a spawned subagent running in the background. Complete the given task independently and report back when done.
Task: %s`,
task,
)
}

138
pkg/tools/spawn_test.go
View file

@ -4,6 +4,7 @@ import (
"context"
"strings"
"testing"
"time"
)
// mockSpawner implements SubTurnSpawner for testing.
@ -113,3 +114,140 @@ func TestSpawnTool_Execute_NilManager(t *testing.T) {
t.Errorf("Error message should mention manager not configured, got: %s", result.ForLLM)
}
}
func TestSpawnTool_SpawnStatusSeesSpawnedTask(t *testing.T) {
provider := &MockLLMProvider{}
manager := NewSubagentManager(provider, "test-model", "/tmp/test")
spawnTool := NewSpawnTool(manager)
spawner := &mockSpawner{done: make(chan struct{})}
spawnTool.SetSpawner(spawner)
statusTool := NewSpawnStatusTool(manager)
ctx := WithToolContext(context.Background(), "telegram", "chat-1")
args := map[string]any{
"task": "Write a haiku about coding",
"label": "haiku-task",
"agent_id": "deep-research",
}
result := spawnTool.Execute(ctx, args)
if result == nil {
t.Fatal("Result should not be nil")
}
if result.IsError {
t.Fatalf("Expected success for valid task, got error: %s", result.ForLLM)
}
if !result.Async {
t.Fatal("SpawnTool should return async result")
}
deadline := time.Now().Add(2 * time.Second)
for {
status := statusTool.Execute(ctx, map[string]any{})
if status == nil {
t.Fatal("status result should not be nil")
}
if status.IsError {
t.Fatalf("spawn_status returned error: %s", status.ForLLM)
}
if strings.Contains(status.ForLLM, "subagent-1") {
if !strings.Contains(status.ForLLM, "haiku-task") {
t.Fatalf("expected label in status output, got: %s", status.ForLLM)
}
break
}
if time.Now().After(deadline) {
t.Fatalf("spawn_status never observed spawned task; last output: %s", status.ForLLM)
}
time.Sleep(10 * time.Millisecond)
}
<-spawner.done
}
func TestSpawnTool_ExecuteAsync_MarksCallbackResultUserOnly(t *testing.T) {
provider := &MockLLMProvider{}
manager := NewSubagentManager(provider, "test-model", "/tmp/test")
tool := NewSpawnTool(manager)
spawner := &mockSpawner{}
tool.SetSpawner(spawner)
done := make(chan *ToolResult, 1)
result := tool.ExecuteAsync(context.Background(), map[string]any{
"task": "Write a haiku about coding",
}, func(_ context.Context, res *ToolResult) {
done <- res
})
if result == nil || !result.Async {
t.Fatal("expected async acknowledgment result")
}
select {
case cbResult := <-done:
if cbResult == nil {
t.Fatal("expected callback result")
}
if cbResult.AsyncDelivery != AsyncDeliveryUserOnly {
t.Fatalf("AsyncDelivery = %q, want %q", cbResult.AsyncDelivery, AsyncDeliveryUserOnly)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for spawn callback result")
}
}
func TestSpawnTool_ExecuteAsync_RespectsExplicitDeliveryMode(t *testing.T) {
provider := &MockLLMProvider{}
manager := NewSubagentManager(provider, "test-model", "/tmp/test")
tool := NewSpawnTool(manager)
spawner := &mockSpawner{}
tool.SetSpawner(spawner)
done := make(chan *ToolResult, 1)
result := tool.ExecuteAsync(context.Background(), map[string]any{
"task": "Write a haiku about coding",
"delivery_mode": string(AsyncDeliveryUserAndParent),
}, func(_ context.Context, res *ToolResult) {
done <- res
})
if result == nil || !result.Async {
t.Fatal("expected async acknowledgment result")
}
select {
case cbResult := <-done:
if cbResult == nil {
t.Fatal("expected callback result")
}
if cbResult.AsyncDelivery != AsyncDeliveryUserAndParent {
t.Fatalf("AsyncDelivery = %q, want %q", cbResult.AsyncDelivery, AsyncDeliveryUserAndParent)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for spawn callback result")
}
}
func TestSpawnTool_Execute_InvalidDeliveryMode(t *testing.T) {
provider := &MockLLMProvider{}
manager := NewSubagentManager(provider, "test-model", "/tmp/test")
tool := NewSpawnTool(manager)
tests := []map[string]any{
{"task": "test", "delivery_mode": 123},
{"task": "test", "delivery_mode": "wrong"},
}
for _, args := range tests {
result := tool.Execute(context.Background(), args)
if result == nil {
t.Fatal("expected result")
}
if !result.IsError {
t.Fatalf("expected error for args=%v", args)
}
if !strings.Contains(result.ForLLM, "delivery_mode") {
t.Fatalf("expected delivery_mode error, got: %s", result.ForLLM)
}
}
}

7
pkg/tools/subagent.go
View file

@ -2,6 +2,7 @@ package tools
import (
"context"
"errors"
"fmt"
"sync"
"sync/atomic"
@ -273,14 +274,14 @@ After completing the task, provide a clear summary of what was done.`
if err != nil {
task.Status = "failed"
task.Result = fmt.Sprintf("Error: %v", err)
// Check if it was canceled
if ctx.Err() != nil {
// Only report cancellation when cancellation is the actual cause.
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
task.Status = "canceled"
task.Result = "Task canceled during execution"
}
result = &ToolResult{
ForLLM: task.Result,
ForUser: "",
ForUser: task.Result,
Silent: false,
IsError: true,
Async: false,