Refactor agent call handling and enhance request structures
- Update `forceSkipForSubAgent` to respect caller's `skip.output` setting, allowing for silent internal worker agent execution. - Refactor error handling in `callAgentWithContext` to utilize a new `NewResult` function for consistent result construction. - Introduce `ProcessCallRequest` structure for improved agent call parameters, including a default timeout constant for process calls.
This commit is contained in:
Max
parent
19c2d6b547
commit
342c2f11a3
7 changed files with 858 additions and 27 deletions
47
agent/caller/context.go
Normal file
47
agent/caller/context.go
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
package caller
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
"github.com/yaoapp/yao/openapi/oauth/types"
|
||||
)
|
||||
|
||||
// NewHeadlessContext creates a headless agent context from a ProcessCallRequest.
|
||||
// This is the Process equivalent of openapi.GetCompletionRequest — constructs
|
||||
// a Context + Options without HTTP dependencies (no Writer, no Interrupt).
|
||||
//
|
||||
// Key behaviors:
|
||||
// - parent context controls timeout/cancellation (caller is responsible)
|
||||
// - skip.output = true (forced): no Writer available, must skip output
|
||||
// - skip.history = true (forced): Process calls don't save chat history
|
||||
// - authorized info is passed in (from authorized.ProcessAuthInfo by caller)
|
||||
// - chatID is auto-generated if not provided
|
||||
func NewHeadlessContext(parent context.Context, authInfo *types.AuthorizedInfo, req *ProcessCallRequest) (*agentContext.Context, *agentContext.Options) {
|
||||
chatID := req.ChatID
|
||||
if chatID == "" {
|
||||
chatID = agentContext.GenChatID()
|
||||
}
|
||||
|
||||
ctx := agentContext.New(parent, authInfo, chatID)
|
||||
ctx.AssistantID = req.AssistantID
|
||||
ctx.Referer = agentContext.RefererProcess
|
||||
ctx.Locale = req.Locale
|
||||
ctx.Route = req.Route
|
||||
ctx.Metadata = req.Metadata
|
||||
|
||||
// Force skip for headless context — no Writer, no chat history
|
||||
skip := req.Skip
|
||||
if skip == nil {
|
||||
skip = &agentContext.Skip{}
|
||||
}
|
||||
skip.Output = true // no Writer available
|
||||
skip.History = true // Process calls don't save chat history
|
||||
|
||||
opts := &agentContext.Options{Skip: skip}
|
||||
if req.Model != "" {
|
||||
opts.Connector = req.Model
|
||||
}
|
||||
|
||||
return ctx, opts
|
||||
}
|
||||
|
|
@ -110,23 +110,27 @@ func (api *JSAPI) RaceWithHandler(requests []interface{}, globalHandler agentCon
|
|||
}
|
||||
|
||||
// forceSkipForSubAgent ensures proper A2A call behavior:
|
||||
// - skip.history = true: A2A messages are not saved to chat history
|
||||
// - skip.output = false: Sub-agents output normally with ThreadID for SSE stream isolation
|
||||
//
|
||||
// IMPORTANT: skip.output is explicitly set to false to override any user settings.
|
||||
// This ensures ThreadID mechanism works correctly for concurrent sub-agent calls.
|
||||
// Users can use the onChunk callback to receive streaming messages if needed.
|
||||
// - skip.history = true: always set — A2A messages are not saved to chat history
|
||||
// - skip.output: defaults to false (sub-agents output with ThreadID for SSE stream isolation),
|
||||
// but if the caller explicitly sets skip.output = true, it is respected.
|
||||
// This allows internal worker agents (e.g. classifiers) to run silently.
|
||||
func (api *JSAPI) forceSkipForSubAgent(req *Request) {
|
||||
if req.Options == nil {
|
||||
req.Options = &CallOptions{}
|
||||
}
|
||||
|
||||
// Preserve caller's explicit skip.output = true before overwriting Skip struct
|
||||
callerSkipOutput := req.Options.Skip != nil && req.Options.Skip.Output
|
||||
|
||||
if req.Options.Skip == nil {
|
||||
req.Options.Skip = &agentContext.Skip{}
|
||||
}
|
||||
req.Options.Skip.History = true
|
||||
// Force output to be enabled - this overrides any user settings
|
||||
// Sub-agents MUST output with ThreadID for proper SSE stream isolation
|
||||
req.Options.Skip.Output = false
|
||||
|
||||
if callerSkipOutput {
|
||||
req.Options.Skip.Output = true
|
||||
}
|
||||
// else: skip.output remains false (default zero value) — sub-agent outputs normally
|
||||
}
|
||||
|
||||
// parseRequestsWithHandlers parses requests and attaches handlers
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package caller
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
|
|
@ -225,20 +226,14 @@ func (o *Orchestrator) callAgentWithContext(ctx *agentContext.Context, req *Requ
|
|||
return &Result{Error: "nil request"}
|
||||
}
|
||||
|
||||
result := &Result{
|
||||
AgentID: req.AgentID,
|
||||
}
|
||||
|
||||
// Get the agent using the getter function
|
||||
if AgentGetterFunc == nil {
|
||||
result.Error = "agent getter not initialized"
|
||||
return result
|
||||
return NewResult(req.AgentID, nil, fmt.Errorf("agent getter not initialized"))
|
||||
}
|
||||
|
||||
agent, err := AgentGetterFunc(req.AgentID)
|
||||
if err != nil {
|
||||
result.Error = "failed to get agent: " + err.Error()
|
||||
return result
|
||||
return NewResult(req.AgentID, nil, fmt.Errorf("failed to get agent: %w", err))
|
||||
}
|
||||
|
||||
// Mark this as an agent-to-agent fork call for proper source tracking
|
||||
|
|
@ -266,18 +261,10 @@ func (o *Orchestrator) callAgentWithContext(ctx *agentContext.Context, req *Requ
|
|||
// The agent.Stream method will use the context's Writer for output
|
||||
resp, err := agent.Stream(ctx, req.Messages, ctxOpts)
|
||||
if err != nil {
|
||||
result.Error = "agent call failed: " + err.Error()
|
||||
return result
|
||||
return NewResult(req.AgentID, nil, fmt.Errorf("agent call failed: %w", err))
|
||||
}
|
||||
|
||||
result.Response = resp
|
||||
|
||||
// Extract content from completion if available
|
||||
if resp != nil && resp.Completion != nil {
|
||||
result.Content = extractContentFromCompletion(resp.Completion)
|
||||
}
|
||||
|
||||
return result
|
||||
return NewResult(req.AgentID, resp, nil)
|
||||
}
|
||||
|
||||
// extractContentFromCompletion extracts the text content from a completion response
|
||||
|
|
|
|||
159
agent/caller/process.go
Normal file
159
agent/caller/process.go
Normal file
|
|
@ -0,0 +1,159 @@
|
|||
package caller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/yaoapp/gou/process"
|
||||
"github.com/yaoapp/kun/exception"
|
||||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
"github.com/yaoapp/yao/openapi/oauth/authorized"
|
||||
)
|
||||
|
||||
func init() {
|
||||
process.Register("agent.Call", processAgentCall)
|
||||
}
|
||||
|
||||
// processAgentCall implements the agent.Call Process handler.
|
||||
// Enables agent-to-agent calls from contexts without agent.Context (e.g., YaoJob).
|
||||
//
|
||||
// Usage: Process("agent.Call", { assistant_id, messages, model?, ... })
|
||||
// Returns: *Result (same structure as ctx.agent.Call in JSAPI)
|
||||
func processAgentCall(p *process.Process) interface{} {
|
||||
|
||||
// 1. Parse parameters via struct — fail fast on invalid input
|
||||
if len(p.Args) == 0 {
|
||||
exception.New("agent.Call: argument is required", 400).Throw()
|
||||
}
|
||||
|
||||
var req ProcessCallRequest
|
||||
raw, err := json.Marshal(p.Args[0])
|
||||
if err != nil {
|
||||
exception.New("agent.Call: invalid argument: %s", 400, err.Error()).Throw()
|
||||
}
|
||||
if err := json.Unmarshal(raw, &req); err != nil {
|
||||
exception.New("agent.Call: failed to parse request: %s", 400, err.Error()).Throw()
|
||||
}
|
||||
|
||||
if req.AssistantID == "" {
|
||||
exception.New("agent.Call: assistant_id is required", 400).Throw()
|
||||
}
|
||||
if len(req.Messages) == 0 {
|
||||
exception.New("agent.Call: messages is required", 400).Throw()
|
||||
}
|
||||
|
||||
// 2. Auto-inject authorization info from process context
|
||||
authInfo := authorized.ProcessAuthInfo(p)
|
||||
|
||||
// 3. Build timeout context — LLM calls can take minutes (tool use, multi-turn)
|
||||
// Default: 10 minutes (DefaultProcessTimeout). Caller can override via `timeout` field.
|
||||
timeoutSec := req.Timeout
|
||||
if timeoutSec <= 0 {
|
||||
timeoutSec = DefaultProcessTimeout
|
||||
}
|
||||
parent := p.Context
|
||||
if parent == nil {
|
||||
parent = context.Background()
|
||||
}
|
||||
timeoutCtx, cancel := context.WithTimeout(parent, time.Duration(timeoutSec)*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// 4. Build headless context + options (encapsulated in context.go)
|
||||
ctx, opts := NewHeadlessContext(timeoutCtx, authInfo, &req)
|
||||
defer ctx.Release()
|
||||
|
||||
// 5. Parse messages from []map[string]interface{} to []agentContext.Message
|
||||
messages := ParseMessages(req.Messages)
|
||||
|
||||
// 6. Get agent and execute
|
||||
if AgentGetterFunc == nil {
|
||||
return NewResult(req.AssistantID, nil, fmt.Errorf("agent getter not initialized"))
|
||||
}
|
||||
|
||||
agent, err := AgentGetterFunc(req.AssistantID)
|
||||
if err != nil {
|
||||
return NewResult(req.AssistantID, nil, fmt.Errorf("failed to get agent: %w", err))
|
||||
}
|
||||
|
||||
resp, err := agent.Stream(ctx, messages, opts)
|
||||
if err != nil {
|
||||
return NewResult(req.AssistantID, nil, fmt.Errorf("agent call failed: %w", err))
|
||||
}
|
||||
|
||||
// 7. Return *Result — shared with ctx.agent.Call() via NewResult()
|
||||
return NewResult(req.AssistantID, resp, nil)
|
||||
}
|
||||
|
||||
// ParseMessages converts []map[string]interface{} to []agentContext.Message.
|
||||
// Extracted as a package-level function so it can be reused by both
|
||||
// processAgentCall and JSAPI.parseMessages.
|
||||
func ParseMessages(raw []map[string]interface{}) []agentContext.Message {
|
||||
result := make([]agentContext.Message, 0, len(raw))
|
||||
for _, msg := range raw {
|
||||
ctxMsg := agentContext.Message{}
|
||||
|
||||
// Parse role
|
||||
if role, ok := msg["role"].(string); ok {
|
||||
ctxMsg.Role = agentContext.MessageRole(role)
|
||||
}
|
||||
|
||||
// Parse content (can be string or array of content parts)
|
||||
ctxMsg.Content = msg["content"]
|
||||
|
||||
// Parse name
|
||||
if name, ok := msg["name"].(string); ok {
|
||||
ctxMsg.Name = &name
|
||||
}
|
||||
|
||||
// Parse tool_call_id
|
||||
if toolCallID, ok := msg["tool_call_id"].(string); ok {
|
||||
ctxMsg.ToolCallID = &toolCallID
|
||||
}
|
||||
|
||||
// Parse tool_calls
|
||||
if toolCalls, ok := msg["tool_calls"].([]interface{}); ok {
|
||||
ctxMsg.ToolCalls = parseToolCalls(toolCalls)
|
||||
}
|
||||
|
||||
// Parse refusal
|
||||
if refusal, ok := msg["refusal"].(string); ok {
|
||||
ctxMsg.Refusal = &refusal
|
||||
}
|
||||
|
||||
result = append(result, ctxMsg)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// parseToolCalls converts []interface{} to []agentContext.ToolCall
|
||||
func parseToolCalls(toolCalls []interface{}) []agentContext.ToolCall {
|
||||
result := make([]agentContext.ToolCall, 0, len(toolCalls))
|
||||
for _, tc := range toolCalls {
|
||||
tcMap, ok := tc.(map[string]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
toolCall := agentContext.ToolCall{}
|
||||
|
||||
if id, ok := tcMap["id"].(string); ok {
|
||||
toolCall.ID = id
|
||||
}
|
||||
if tcType, ok := tcMap["type"].(string); ok {
|
||||
toolCall.Type = agentContext.ToolCallType(tcType)
|
||||
}
|
||||
if fn, ok := tcMap["function"].(map[string]interface{}); ok {
|
||||
if name, ok := fn["name"].(string); ok {
|
||||
toolCall.Function.Name = name
|
||||
}
|
||||
if args, ok := fn["arguments"].(string); ok {
|
||||
toolCall.Function.Arguments = args
|
||||
}
|
||||
}
|
||||
|
||||
result = append(result, toolCall)
|
||||
}
|
||||
return result
|
||||
}
|
||||
385
agent/caller/process_e2e_test.go
Normal file
385
agent/caller/process_e2e_test.go
Normal file
|
|
@ -0,0 +1,385 @@
|
|||
package caller_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/yaoapp/gou/process"
|
||||
"github.com/yaoapp/yao/agent/caller"
|
||||
"github.com/yaoapp/yao/agent/testutils"
|
||||
)
|
||||
|
||||
// newLLMProcess creates a process.Process with a 120s outer timeout for LLM calls.
|
||||
// agent.Call has its own internal default timeout (DefaultProcessTimeout = 600s),
|
||||
// but the outer context (120s) takes precedence via context.WithTimeout chaining.
|
||||
func newLLMProcess(t *testing.T, name string, args ...interface{}) *process.Process {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
return process.NewWithContext(ctx, name, args...)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// A. Pure LLM scenarios (tests.simple-greeting — no hooks)
|
||||
// ============================================================================
|
||||
|
||||
func TestProcessCall_LLM_Basic(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM (source env.local.sh)")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.simple-greeting",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "Hello!"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err)
|
||||
|
||||
val := proc.Value()
|
||||
require.NotNil(t, val, "process should return a value")
|
||||
|
||||
result, ok := val.(*caller.Result)
|
||||
require.True(t, ok, "value should be *caller.Result, got %T", val)
|
||||
|
||||
assert.Equal(t, "tests.simple-greeting", result.AgentID)
|
||||
assert.Empty(t, result.Error, "should not have error")
|
||||
assert.NotEmpty(t, result.Content, "should have LLM content")
|
||||
assert.NotNil(t, result.Response, "should have full response")
|
||||
t.Logf("LLM response: %s", result.Content)
|
||||
}
|
||||
|
||||
func TestProcessCall_LLM_MultipleMessages(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.simple-greeting",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "system", "content": "Always reply in JSON format."},
|
||||
map[string]interface{}{"role": "user", "content": "Say hello"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err)
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.NotEmpty(t, result.Content)
|
||||
t.Logf("Multi-message response: %s", result.Content)
|
||||
}
|
||||
|
||||
func TestProcessCall_LLM_WithMetadata(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.simple-greeting",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "Hi there!"},
|
||||
},
|
||||
"metadata": map[string]interface{}{"source": "e2e-test", "mode": "task"},
|
||||
"locale": "zh-CN",
|
||||
"route": "/test/e2e",
|
||||
"chat_id": "e2e-test-chat-001",
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err)
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.NotEmpty(t, result.Content)
|
||||
t.Logf("With-metadata response: %s", result.Content)
|
||||
}
|
||||
|
||||
func TestProcessCall_LLM_SkipOutputForced(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Explicitly pass skip.output=false — headless context MUST force it to true
|
||||
// If the force logic fails, this would panic (nil Writer).
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.simple-greeting",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "Hello!"},
|
||||
},
|
||||
"skip": map[string]interface{}{
|
||||
"output": false,
|
||||
"history": false,
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err, "should NOT panic even with skip.output=false — headless forces true")
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.NotEmpty(t, result.Content)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// B. Create Hook scenarios (tests.create)
|
||||
// ============================================================================
|
||||
|
||||
func TestProcessCall_CreateHook_Default(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Send a generic message — Create Hook routes to scenarioDefault
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.create",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "hello world"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err)
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "tests.create", result.AgentID)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.NotEmpty(t, result.Content, "Create Hook should still produce LLM response")
|
||||
t.Logf("CreateHook default response: %s", result.Content)
|
||||
}
|
||||
|
||||
func TestProcessCall_CreateHook_ReturnFull(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Send "return_full" — Create Hook returns full HookCreateResponse with custom messages
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.create",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "return_full"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err)
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "tests.create", result.AgentID)
|
||||
assert.Empty(t, result.Error)
|
||||
// The Create Hook overrides messages with system + user, then LLM responds
|
||||
assert.NotEmpty(t, result.Content)
|
||||
t.Logf("CreateHook return_full response: %s", result.Content)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// C. Next Hook scenarios (tests.next)
|
||||
// ============================================================================
|
||||
|
||||
func TestProcessCall_NextHook_Standard(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Send "standard" — Next Hook returns null, standard LLM response is used
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.next",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "standard"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err)
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "tests.next", result.AgentID)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.NotEmpty(t, result.Content, "standard scenario should return LLM content")
|
||||
t.Logf("NextHook standard response: %s", result.Content)
|
||||
}
|
||||
|
||||
func TestProcessCall_NextHook_CustomData(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Send "return_custom_data" — Next Hook returns custom data
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.next",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "return_custom_data"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err)
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "tests.next", result.AgentID)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.NotNil(t, result.Response, "should have response")
|
||||
|
||||
// Next Hook custom data is available in response.Next
|
||||
if result.Response != nil && result.Response.Next != nil {
|
||||
t.Logf("NextHook custom data: %+v", result.Response.Next)
|
||||
nextMap, ok := result.Response.Next.(map[string]interface{})
|
||||
if ok {
|
||||
// The Next Hook returns { data: { message, test, timestamp } }
|
||||
if dataMap, ok := nextMap["data"].(map[string]interface{}); ok {
|
||||
assert.Equal(t, "Custom response from Next Hook", dataMap["message"])
|
||||
assert.Equal(t, true, dataMap["test"])
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// D. Timeout scenarios
|
||||
// ============================================================================
|
||||
|
||||
func TestProcessCall_Timeout_Short(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires real LLM (source env.local.sh)")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Set timeout=2 seconds — LLM round-trip will certainly exceed this.
|
||||
// Verifies that the timeout parameter is respected and produces an error.
|
||||
proc := newLLMProcess(t, "agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.simple-greeting",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "Tell me a very long story about the history of computing."},
|
||||
},
|
||||
"timeout": 2,
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
if err != nil {
|
||||
// Timeout may surface as a process-level error (context deadline exceeded)
|
||||
t.Logf("Process error (expected timeout): %s", err.Error())
|
||||
assert.Contains(t, err.Error(), "deadline exceeded",
|
||||
"error should indicate context deadline exceeded")
|
||||
return
|
||||
}
|
||||
|
||||
// Or the agent.Stream may catch the timeout and return it in Result.Error
|
||||
val := proc.Value()
|
||||
require.NotNil(t, val, "process should return a value")
|
||||
result, ok := val.(*caller.Result)
|
||||
require.True(t, ok, "value should be *caller.Result, got %T", val)
|
||||
assert.NotEmpty(t, result.Error, "should have timeout error in result")
|
||||
t.Logf("Timeout error in result: %s", result.Error)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// E. Error / validation scenarios
|
||||
// ============================================================================
|
||||
|
||||
func TestProcessCall_Error_MissingAssistantID(t *testing.T) {
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
proc := process.New("agent.call", map[string]interface{}{
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "Hello"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.Error(t, err, "should fail: assistant_id is required")
|
||||
t.Logf("Expected error: %s", err.Error())
|
||||
assert.Contains(t, err.Error(), "assistant_id")
|
||||
}
|
||||
|
||||
func TestProcessCall_Error_EmptyMessages(t *testing.T) {
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
proc := process.New("agent.call", map[string]interface{}{
|
||||
"assistant_id": "tests.simple-greeting",
|
||||
"messages": []interface{}{},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.Error(t, err, "should fail: messages is required")
|
||||
t.Logf("Expected error: %s", err.Error())
|
||||
assert.Contains(t, err.Error(), "messages")
|
||||
}
|
||||
|
||||
func TestProcessCall_Error_InvalidArgument(t *testing.T) {
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Pass a string instead of map — json.Marshal will succeed but Unmarshal will fail
|
||||
proc := process.New("agent.call", "not-a-map")
|
||||
|
||||
err := proc.Execute()
|
||||
require.Error(t, err, "should fail: argument must be a map")
|
||||
t.Logf("Expected error: %s", err.Error())
|
||||
}
|
||||
|
||||
func TestProcessCall_Error_NoArgument(t *testing.T) {
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
proc := process.New("agent.call")
|
||||
|
||||
err := proc.Execute()
|
||||
require.Error(t, err, "should fail: argument is required")
|
||||
t.Logf("Expected error: %s", err.Error())
|
||||
}
|
||||
|
||||
func TestProcessCall_Error_NonexistentAgent(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping: requires environment")
|
||||
}
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
proc := process.New("agent.call", map[string]interface{}{
|
||||
"assistant_id": "does.not.exist.agent",
|
||||
"messages": []interface{}{
|
||||
map[string]interface{}{"role": "user", "content": "Hello"},
|
||||
},
|
||||
})
|
||||
|
||||
err := proc.Execute()
|
||||
require.NoError(t, err, "process should not error — error is in Result")
|
||||
|
||||
result, ok := proc.Value().(*caller.Result)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "does.not.exist.agent", result.AgentID)
|
||||
assert.NotEmpty(t, result.Error, "should have error for nonexistent agent")
|
||||
assert.Contains(t, result.Error, "failed to get agent")
|
||||
t.Logf("Expected error in result: %s", result.Error)
|
||||
}
|
||||
214
agent/caller/process_test.go
Normal file
214
agent/caller/process_test.go
Normal file
|
|
@ -0,0 +1,214 @@
|
|||
package caller_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/yaoapp/yao/agent/caller"
|
||||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
"github.com/yaoapp/yao/openapi/oauth/types"
|
||||
)
|
||||
|
||||
// --- NewHeadlessContext tests ---
|
||||
|
||||
func TestNewHeadlessContext_Basic(t *testing.T) {
|
||||
authInfo := &types.AuthorizedInfo{
|
||||
TeamID: "team-123",
|
||||
UserID: "user-456",
|
||||
}
|
||||
req := &caller.ProcessCallRequest{
|
||||
AssistantID: "yao.keeper.classify",
|
||||
Messages: []map[string]interface{}{
|
||||
{"role": "user", "content": "hello"},
|
||||
},
|
||||
Locale: "zh-CN",
|
||||
}
|
||||
|
||||
ctx, opts := caller.NewHeadlessContext(context.Background(), authInfo, req)
|
||||
defer ctx.Release()
|
||||
|
||||
assert.Equal(t, "yao.keeper.classify", ctx.AssistantID)
|
||||
assert.Equal(t, agentContext.RefererProcess, ctx.Referer)
|
||||
assert.Equal(t, "zh-CN", ctx.Locale)
|
||||
assert.NotEmpty(t, ctx.ChatID) // auto-generated
|
||||
|
||||
require.NotNil(t, opts)
|
||||
require.NotNil(t, opts.Skip)
|
||||
assert.True(t, opts.Skip.Output, "skip.output must be forced true for headless context")
|
||||
assert.True(t, opts.Skip.History, "skip.history must be forced true for headless context")
|
||||
assert.Empty(t, opts.Connector)
|
||||
}
|
||||
|
||||
func TestNewHeadlessContext_WithModel(t *testing.T) {
|
||||
req := &caller.ProcessCallRequest{
|
||||
AssistantID: "test.agent",
|
||||
Messages: []map[string]interface{}{{"role": "user", "content": "hi"}},
|
||||
Model: "deepseek.v3",
|
||||
}
|
||||
|
||||
ctx, opts := caller.NewHeadlessContext(context.Background(), nil, req)
|
||||
defer ctx.Release()
|
||||
|
||||
assert.Equal(t, "deepseek.v3", opts.Connector)
|
||||
}
|
||||
|
||||
func TestNewHeadlessContext_WithChatID(t *testing.T) {
|
||||
req := &caller.ProcessCallRequest{
|
||||
AssistantID: "test.agent",
|
||||
Messages: []map[string]interface{}{{"role": "user", "content": "hi"}},
|
||||
ChatID: "custom-chat-id",
|
||||
}
|
||||
|
||||
ctx, _ := caller.NewHeadlessContext(context.Background(), nil, req)
|
||||
defer ctx.Release()
|
||||
|
||||
assert.Equal(t, "custom-chat-id", ctx.ChatID)
|
||||
}
|
||||
|
||||
func TestNewHeadlessContext_ForceSkipOverridesUserSkip(t *testing.T) {
|
||||
req := &caller.ProcessCallRequest{
|
||||
AssistantID: "test.agent",
|
||||
Messages: []map[string]interface{}{{"role": "user", "content": "hi"}},
|
||||
Skip: &agentContext.Skip{Output: false, History: false, Trace: true},
|
||||
}
|
||||
|
||||
_, opts := caller.NewHeadlessContext(context.Background(), nil, req)
|
||||
|
||||
// Output and History must be forced true regardless of user input
|
||||
assert.True(t, opts.Skip.Output, "skip.output must be forced true")
|
||||
assert.True(t, opts.Skip.History, "skip.history must be forced true")
|
||||
// User-specified skip.trace should be preserved
|
||||
assert.True(t, opts.Skip.Trace, "skip.trace should be preserved from user input")
|
||||
}
|
||||
|
||||
func TestNewHeadlessContext_WithMetadata(t *testing.T) {
|
||||
req := &caller.ProcessCallRequest{
|
||||
AssistantID: "test.agent",
|
||||
Messages: []map[string]interface{}{{"role": "user", "content": "hi"}},
|
||||
Metadata: map[string]interface{}{"key": "value"},
|
||||
Route: "/test",
|
||||
}
|
||||
|
||||
ctx, _ := caller.NewHeadlessContext(context.Background(), nil, req)
|
||||
defer ctx.Release()
|
||||
|
||||
assert.Equal(t, "value", ctx.Metadata["key"])
|
||||
assert.Equal(t, "/test", ctx.Route)
|
||||
}
|
||||
|
||||
func TestNewHeadlessContext_WithTimeout(t *testing.T) {
|
||||
req := &caller.ProcessCallRequest{
|
||||
AssistantID: "test.agent",
|
||||
Messages: []map[string]interface{}{{"role": "user", "content": "hi"}},
|
||||
Timeout: 30,
|
||||
}
|
||||
|
||||
// Pass a context with timeout to verify it propagates
|
||||
ctx, _ := caller.NewHeadlessContext(context.Background(), nil, req)
|
||||
defer ctx.Release()
|
||||
|
||||
// Timeout field is consumed by processAgentCall, not NewHeadlessContext.
|
||||
// Here we just verify the field is correctly set in the struct.
|
||||
assert.Equal(t, 30, req.Timeout)
|
||||
}
|
||||
|
||||
func TestProcessCallRequest_DefaultTimeout(t *testing.T) {
|
||||
req := &caller.ProcessCallRequest{
|
||||
AssistantID: "test.agent",
|
||||
Messages: []map[string]interface{}{{"role": "user", "content": "hi"}},
|
||||
}
|
||||
|
||||
// When Timeout is 0 (zero value), the default should be used
|
||||
assert.Equal(t, 0, req.Timeout, "zero value means use default")
|
||||
assert.Equal(t, 600, caller.DefaultProcessTimeout, "default timeout should be 600 seconds")
|
||||
}
|
||||
|
||||
// --- ParseMessages tests ---
|
||||
|
||||
func TestParseMessages_Basic(t *testing.T) {
|
||||
raw := []map[string]interface{}{
|
||||
{"role": "user", "content": "hello"},
|
||||
{"role": "assistant", "content": "hi there"},
|
||||
}
|
||||
|
||||
messages := caller.ParseMessages(raw)
|
||||
require.Len(t, messages, 2)
|
||||
|
||||
assert.Equal(t, agentContext.MessageRole("user"), messages[0].Role)
|
||||
assert.Equal(t, "hello", messages[0].Content)
|
||||
assert.Equal(t, agentContext.MessageRole("assistant"), messages[1].Role)
|
||||
assert.Equal(t, "hi there", messages[1].Content)
|
||||
}
|
||||
|
||||
func TestParseMessages_WithOptionalFields(t *testing.T) {
|
||||
name := "test-name"
|
||||
raw := []map[string]interface{}{
|
||||
{
|
||||
"role": "tool",
|
||||
"content": "result",
|
||||
"name": name,
|
||||
"tool_call_id": "tc-1",
|
||||
},
|
||||
}
|
||||
|
||||
messages := caller.ParseMessages(raw)
|
||||
require.Len(t, messages, 1)
|
||||
|
||||
msg := messages[0]
|
||||
assert.Equal(t, agentContext.MessageRole("tool"), msg.Role)
|
||||
require.NotNil(t, msg.Name)
|
||||
assert.Equal(t, name, *msg.Name)
|
||||
require.NotNil(t, msg.ToolCallID)
|
||||
assert.Equal(t, "tc-1", *msg.ToolCallID)
|
||||
}
|
||||
|
||||
func TestParseMessages_Empty(t *testing.T) {
|
||||
messages := caller.ParseMessages(nil)
|
||||
assert.Empty(t, messages)
|
||||
}
|
||||
|
||||
// --- NewResult tests ---
|
||||
|
||||
func TestNewResult_Success(t *testing.T) {
|
||||
resp := &agentContext.Response{
|
||||
Completion: &agentContext.CompletionResponse{
|
||||
Content: "answer text",
|
||||
},
|
||||
}
|
||||
|
||||
result := caller.NewResult("test.agent", resp, nil)
|
||||
|
||||
assert.Equal(t, "test.agent", result.AgentID)
|
||||
assert.Equal(t, "answer text", result.Content)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.NotNil(t, result.Response)
|
||||
}
|
||||
|
||||
func TestNewResult_WithError(t *testing.T) {
|
||||
result := caller.NewResult("test.agent", nil, errors.New("something failed"))
|
||||
|
||||
assert.Equal(t, "test.agent", result.AgentID)
|
||||
assert.Equal(t, "something failed", result.Error)
|
||||
assert.Empty(t, result.Content)
|
||||
assert.Nil(t, result.Response)
|
||||
}
|
||||
|
||||
func TestNewResult_NilResponse(t *testing.T) {
|
||||
result := caller.NewResult("test.agent", nil, nil)
|
||||
|
||||
assert.Equal(t, "test.agent", result.AgentID)
|
||||
assert.Empty(t, result.Content)
|
||||
assert.Empty(t, result.Error)
|
||||
assert.Nil(t, result.Response)
|
||||
}
|
||||
|
||||
func TestNewResult_NilCompletion(t *testing.T) {
|
||||
resp := &agentContext.Response{Completion: nil}
|
||||
result := caller.NewResult("test.agent", resp, nil)
|
||||
|
||||
assert.Empty(t, result.Content, "content should be empty when completion is nil")
|
||||
assert.NotNil(t, result.Response)
|
||||
}
|
||||
|
|
@ -5,6 +5,10 @@ import (
|
|||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
)
|
||||
|
||||
// DefaultProcessTimeout is the default timeout (in seconds) for agent.Call Process.
|
||||
// LLM calls with tool use can take minutes; 10 minutes provides safe headroom.
|
||||
const DefaultProcessTimeout = 600
|
||||
|
||||
// Request represents a request to call an agent
|
||||
type Request struct {
|
||||
AgentID string `json:"agent"` // Target agent ID
|
||||
|
|
@ -29,6 +33,37 @@ type Result struct {
|
|||
Error string `json:"error,omitempty"` // Error message if call failed
|
||||
}
|
||||
|
||||
// ProcessCallRequest is the parameter structure for the agent.Call Process.
|
||||
// Fields mirror CompletionRequest + HTTP header semantics, enabling headless
|
||||
// agent calls from contexts without agent.Context (e.g., YaoJob async tasks).
|
||||
type ProcessCallRequest struct {
|
||||
AssistantID string `json:"assistant_id"` // Required: target assistant ID (maps to X-Yao-Assistant header)
|
||||
Messages []map[string]interface{} `json:"messages"` // Required: message list (maps to CompletionRequest.Messages)
|
||||
Model string `json:"model,omitempty"` // Optional: connector ID override (maps to CompletionRequest.Model)
|
||||
Skip *agentContext.Skip `json:"skip,omitempty"` // Optional: skip config (maps to CompletionRequest.Skip)
|
||||
Metadata map[string]interface{} `json:"metadata,omitempty"` // Optional: passed to hooks (maps to CompletionRequest.Metadata)
|
||||
Locale string `json:"locale,omitempty"` // Optional (maps to locale query param)
|
||||
Route string `json:"route,omitempty"` // Optional (maps to CompletionRequest.Route)
|
||||
ChatID string `json:"chat_id,omitempty"` // Optional: auto-generated if empty (maps to chat_id query/header)
|
||||
Timeout int `json:"timeout,omitempty"` // Optional: timeout in seconds (default: DefaultProcessTimeout = 600)
|
||||
}
|
||||
|
||||
// NewResult builds a Result from an agent call response.
|
||||
// Used by both ctx.agent.Call (orchestrator) and Process("agent.Call") to
|
||||
// ensure consistent result construction.
|
||||
func NewResult(agentID string, resp *agentContext.Response, err error) *Result {
|
||||
result := &Result{AgentID: agentID}
|
||||
if err != nil {
|
||||
result.Error = err.Error()
|
||||
return result
|
||||
}
|
||||
result.Response = resp
|
||||
if resp != nil && resp.Completion != nil {
|
||||
result.Content = extractContentFromCompletion(resp.Completion)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// ToContextOptions converts CallOptions to context.Options for the agent call
|
||||
func (o *CallOptions) ToContextOptions() *agentContext.Options {
|
||||
if o == nil {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue