diff --git a/cmd/root/root.go b/cmd/root/root.go index c94ff70b1..a5a1b0832 100644 --- a/cmd/root/root.go +++ b/cmd/root/root.go @@ -130,9 +130,11 @@ func Execute() { } // printError logs a failed command's error as JSON for a collecting parent, -// and otherwise prints it styled for a person. +// and otherwise prints it styled for a person. The text keeps its case, as fang +// would title-case its first word and turn a flag like --model into --Model. func printError(w io.Writer, styles fang.Styles, err error) { if !logger.ReportExitError(os.Stderr, err) { + styles.ErrorText = styles.ErrorText.UnsetTransform() fang.DefaultErrorHandler(w, styles, err) } } diff --git a/cmd/root/root_test.go b/cmd/root/root_test.go index 779f845b0..7c92036e8 100644 --- a/cmd/root/root_test.go +++ b/cmd/root/root_test.go @@ -1,11 +1,17 @@ package root import ( + "bytes" + "errors" "slices" + "strings" "testing" require "github.com/stretchr/testify/require" + fang "charm.land/fang/v2" + lipgloss "charm.land/lipgloss/v2" + output "github.com/inference-gateway/cli/cmd/output" ) @@ -49,3 +55,12 @@ func TestNewCommandReturnsIndependentTrees(t *testing.T) { require.NoError(t, err) require.NotSame(t, firstChat, secondChat) } + +func TestPrintError_KeepsTheCaseOfTheMessage(t *testing.T) { + var out bytes.Buffer + styles := fang.Styles{ErrorText: lipgloss.NewStyle().Transform(strings.ToUpper)} + + printError(&out, styles, errors.New("--model is required")) + + require.Contains(t, out.String(), "--model is required") +} diff --git a/internal/agent/loop/agent.go b/internal/agent/loop/agent.go index a28080c1e..611bc67ec 100644 --- a/internal/agent/loop/agent.go +++ b/internal/agent/loop/agent.go @@ -295,7 +295,7 @@ func (p *eventPublisher) publishBashOutputChunk(callID string, output string, is select { case p.chatEvents <- event: default: - logger.Warn("bash output chunk dropped - channel full") + logger.Debug("bash output chunk dropped - channel full") } } diff --git a/internal/agent/loop/testdata/tool_definitions.golden.json b/internal/agent/loop/testdata/tool_definitions.golden.json index 0470cc320..c96429a0f 100644 --- a/internal/agent/loop/testdata/tool_definitions.golden.json +++ b/internal/agent/loop/testdata/tool_definitions.golden.json @@ -742,7 +742,7 @@ }, { "function": { - "description": "A powerful search tool with configurable backend (ripgrep or Go implementation)\n\n Usage:\n - ALWAYS use Grep for search tasks. NEVER invoke `grep` or `rg` as a Bash command. The Grep tool has been optimized for correct permissions and access.\n - Supports full regex syntax (e.g., \"log.*Error\", \"function\\s+\\w+\")\n - Filter files with glob parameter (e.g., \"*.js\", \"**/*.tsx\") or type parameter (e.g., \"js\", \"py\", \"rust\")\n - Output modes: \"content\" shows matching lines, \"files_with_matches\" shows only file paths (default), \"count\" shows match counts\n - Use the Agent tool for open-ended searches requiring multiple rounds\n - Pattern syntax: When using ripgrep backend - literal braces need escaping (use `interface\\{\\}` to find `any` in Go code)\n - Multiline matching: By default patterns match within single lines only. For cross-line patterns like `struct \\{[\\s\\S]*?field`, use `multiline: true`\n", + "description": "A powerful search tool with configurable backend (ripgrep or Go implementation)\n\n Usage:\n - ALWAYS use Grep for search tasks. NEVER invoke `grep` or `rg` as a Bash command. The Grep tool has been optimized for correct permissions and access.\n - Regex syntax is RE2 (e.g., \"log.*Error\", \"function\\s+\\w+\") - no lookahead, lookbehind or backreferences\n - Filter files with glob parameter (e.g., \"*.js\", \"**/*.tsx\") or type parameter (e.g., \"js\", \"py\", \"rust\")\n - Output modes: \"content\" shows matching lines, \"files_with_matches\" shows only file paths (default), \"count\" shows match counts\n - Use the Agent tool for open-ended searches requiring multiple rounds\n - Pattern syntax: When using ripgrep backend - literal braces need escaping (use `interface\\{\\}` to find `any` in Go code)\n - Multiline matching: By default patterns match within single lines only. For cross-line patterns like `struct \\{[\\s\\S]*?field`, use `multiline: true`\n", "name": "Grep", "parameters": { "$schema": "http://json-schema.org/draft-07/schema#", @@ -1348,7 +1348,7 @@ "type": "string" }, "submit": { - "description": "Interactive subagents only. Press Enter to submit a prompt and wait for the subagent to finish (default true). Set false to only send keys for TUI navigation - then inspect with ReadSubagentScreen.", + "description": "Interactive subagents only. Press Enter to submit the prompt (default true). Set false to only send keys for TUI navigation - then inspect with ReadSubagentScreen.", "type": "boolean" }, "text": { diff --git a/internal/gateway/supervisor.go b/internal/gateway/supervisor.go index b5ae6d312..d0fc21b1f 100644 --- a/internal/gateway/supervisor.go +++ b/internal/gateway/supervisor.go @@ -162,17 +162,17 @@ func (gm *Supervisor) startContainer(ctx context.Context) error { } } + if err := gm.pullImage(ctx); err != nil { + logger.Warn("failed to pull image, attempting to use local image", "error", err) + fmt.Println("• Could not pull latest image, using cached version") + } + if gm.containerRuntime != nil { if err := gm.containerRuntime.EnsureNetwork(ctx); err != nil { logger.Warn("failed to create Docker network", "session", gm.sessionID, "error", err) } } - if err := gm.pullImage(ctx); err != nil { - logger.Warn("failed to pull image, attempting to use local image", "error", err) - fmt.Println("• Could not pull latest image, using cached version") - } - if gm.config.Gateway.Debug { fmt.Println("• Debug mode enabled - Gateway is running in development mode with detailed logging") } diff --git a/internal/insights/insights.go b/internal/insights/insights.go index 337a28be3..9748f06f0 100644 --- a/internal/insights/insights.go +++ b/internal/insights/insights.go @@ -127,7 +127,7 @@ func (g *Generator) Generate(ctx context.Context, since time.Time, progress io.W memory := g.memoryIndex() - _, _ = fmt.Fprintf(progress, "Read %d sessions and %d log failure groups\n", len(sessions), len(logs.Groups)) + _, _ = fmt.Fprintf(progress, "Read %d sessions and %d log groups\n", len(sessions), len(logs.Groups)) _, _ = fmt.Fprintf(progress, "Analyzing with %s, can take minutes...\n", model) analysis, usage, err := g.analyze(ctx, model, buildDigest(sessions, failures, tools, memory, logs)) if err != nil { @@ -370,17 +370,18 @@ func buildDigest(sessions []sessionDigest, failures []toolFailure, tools []telem } if len(logs.Groups) > 0 { - b.WriteString("\nLOG FAILURES (from the log files, most frequent first)\n") - b.WriteString("These are failures the sessions above never recorded - crashes, startup and\n") - b.WriteString("background errors. xN is how many times that same line recurred: a high N\n") - b.WriteString("over a short span is a retry loop that never succeeded, the same N spread\n") - b.WriteString("over days is a chronic fault. Near-identical lines are already folded together.\n") + b.WriteString("\nLOG WARNINGS AND ERRORS (from the log files, most frequent first)\n") + b.WriteString("A line naming a tool repeats a failed tool call the sessions above already\n") + b.WriteString("counted. The rest never reached a session - crashes, startup and background\n") + b.WriteString("problems. xN is how many times that same line recurred: a high N over a short\n") + b.WriteString("span is a retry loop that never succeeded, the same N spread over days is a\n") + b.WriteString("chronic fault. Near-identical lines are already folded together.\n") for _, g := range logs.Groups { tool := "" if g.Tool != "" { tool = "tool " + g.Tool + ": " } - fmt.Fprintf(&b, "- x%d [%s .. %s] %s%s\n", g.Count, + fmt.Fprintf(&b, "- x%d %s [%s .. %s] %s%s\n", g.Count, g.Level, g.First.Format(time.RFC3339), g.Last.Format(time.RFC3339), tool, g.Sample) } } @@ -421,11 +422,13 @@ For each recurring failure: what is actually going wrong and the concrete fix Skip tools whose failures look incidental rather than systematic. ### Failures that never reached a session -Recurring failures from LOG FAILURES below, which no saved session recorded - crashes, -startup and background errors. For each: what is actually going wrong and the concrete -fix. Read the count with the time span: a high count over seconds is a retry loop that -never succeeded, the same count spread over days is a chronic fault. Skip anything that -looks incidental. If there are no log failures, say so in one line. +Recurring problems from LOG WARNINGS AND ERRORS below that no saved session recorded - +crashes, startup and background problems. Lines naming a tool repeat the failed tool +calls above, so do not count them again. A warn is a recoverable problem, so report one +only when it costs the user something. For each: what is actually going wrong and the +concrete fix. Read the count with the time span: a high count over seconds is a retry +loop that never succeeded, the same count spread over days is a chronic fault. Skip +anything that looks incidental. If there is nothing to report, say so in one line. PERSISTENT MEMORY below, when present, is what the agent has already learned about this user. Use it as context so you do not suggest what they already do; never @@ -516,11 +519,11 @@ func renderReport(meta reportMeta, failures []toolFailure, tools []telemetry.Too } if len(logs.Groups) > 0 { - b.WriteString("## Log failures\n\n") - b.WriteString("| Count | First | Last | Tool | Message | Trace |\n") - b.WriteString("|-------|-------|------|------|---------|-------|\n") + b.WriteString("## Log warnings and errors\n\n") + b.WriteString("| Count | Level | First | Last | Tool | Message | Trace |\n") + b.WriteString("|-------|-------|-------|------|------|---------|-------|\n") for _, g := range logs.Groups { - fmt.Fprintf(&b, "| %d | %s | %s | %s | %s | %s |\n", g.Count, + fmt.Fprintf(&b, "| %d | %s | %s | %s | %s | %s | %s |\n", g.Count, g.Level, g.First.Format(time.RFC3339), g.Last.Format(time.RFC3339), g.Tool, strings.ReplaceAll(g.Sample, "|", "\\|"), g.TraceID) } diff --git a/internal/insights/insights_logs.go b/internal/insights/insights_logs.go index 6b839e59b..e791f8e32 100644 --- a/internal/insights/insights_logs.go +++ b/internal/insights/insights_logs.go @@ -35,6 +35,7 @@ var logNoise = regexp.MustCompile(`(/[\w.\-/]+)|(\b[0-9a-f]{8,}\b)`) // same count spread over days is a chronic fault. type logGroup struct { Template string + Level string Count int First time.Time Last time.Time @@ -172,7 +173,7 @@ func addLogRecord(groups map[string]*logGroup, rec logRecord, when time.Time) { if len(groups) >= maxLogTemplates { return } - g = &logGroup{Template: key, First: when, Last: when, Sample: oneLine(raw, maxLogSampleChars), Tool: rec.Tool} + g = &logGroup{Template: key, Level: rec.Level, First: when, Last: when, Sample: oneLine(raw, maxLogSampleChars), Tool: rec.Tool} groups[key] = g } if g.TraceID == "" { diff --git a/internal/insights/insights_logs_test.go b/internal/insights/insights_logs_test.go index 833a1c484..2cfdcf6cc 100644 --- a/internal/insights/insights_logs_test.go +++ b/internal/insights/insights_logs_test.go @@ -190,14 +190,14 @@ func TestLogFailureWithoutConversationReachesDigest(t *testing.T) { } prompt := buildDigest(nil, nil, nil, "", digest) - for _, want := range []string{"LOG FAILURES", "x400", "failed to start gateway container", "port 8080 already in use"} { + for _, want := range []string{"LOG WARNINGS AND ERRORS", "x400", "failed to start gateway container", "port 8080 already in use"} { if !strings.Contains(prompt, want) { t.Errorf("digest missing %q:\n%s", want, prompt) } } report := renderReport(reportMeta{Generated: now, LogRecords: digest.Scanned, LogGroups: len(digest.Groups)}, nil, nil, digest, "x") - for _, want := range []string{"log_records: 400", "log_groups: 1", "## Log failures", "| 400 |"} { + for _, want := range []string{"log_records: 400", "log_groups: 1", "## Log warnings and errors", "| 400 | error |"} { if !strings.Contains(report, want) { t.Errorf("report missing %q:\n%s", want, report) } diff --git a/internal/insights/insights_test.go b/internal/insights/insights_test.go index fd22c4495..263033961 100644 --- a/internal/insights/insights_test.go +++ b/internal/insights/insights_test.go @@ -440,7 +440,7 @@ func TestLogSectionSurvivesManySessions(t *testing.T) { if len(digest) > maxDigestChars+3 { t.Errorf("digest not bounded: %d chars", len(digest)) } - for _, want := range []string{"LOG FAILURES", "x400", "failed to start gateway container"} { + for _, want := range []string{"LOG WARNINGS AND ERRORS", "x400", "failed to start gateway container"} { if !strings.Contains(digest, want) { t.Errorf("a busy session store must not crowd out %q:\n...%s", want, digest[max(0, len(digest)-400):]) } @@ -459,11 +459,11 @@ func TestPromptCoversEveryDigestSection(t *testing.T) { ) headings := map[string]string{ - "PERSISTENT MEMORY": "PERSISTENT MEMORY", - "SESSIONS": "sessions", - "TOOL CALL TOTALS": "Tool calls that keep failing", - "FAILED TOOL CALLS": "Tool calls that keep failing", - "LOG FAILURES": "LOG FAILURES", + "PERSISTENT MEMORY": "PERSISTENT MEMORY", + "SESSIONS": "sessions", + "TOOL CALL TOTALS": "Tool calls that keep failing", + "FAILED TOOL CALLS": "Tool calls that keep failing", + "LOG WARNINGS AND ERRORS": "LOG WARNINGS AND ERRORS", } for section, instruction := range headings { if !strings.Contains(digest, section) { diff --git a/internal/platform/container/docker_runtime.go b/internal/platform/container/docker_runtime.go index 1ddf16fec..2f14820db 100644 --- a/internal/platform/container/docker_runtime.go +++ b/internal/platform/container/docker_runtime.go @@ -5,6 +5,7 @@ import ( "fmt" "os/exec" "strings" + "sync/atomic" logger "github.com/inference-gateway/cli/internal/platform/logger" ) @@ -13,7 +14,7 @@ import ( type DockerRuntime struct { sessionID string networkName string - networkCreated bool + networkCreated atomic.Bool } // NewDockerRuntime creates a new Docker runtime manager @@ -30,16 +31,12 @@ func (dr *DockerRuntime) GetNetworkName() string { } // EnsureNetwork creates the shared Docker network if it doesn't exist. The -// network is reused across sessions, so at most one ever exists. If creation -// fails because the IPAM address pools are exhausted (leaked networks from -// prior sessions), it prunes those and retries once. +// network is reused across sessions and any of them may remove it, so every +// call checks again. If creation fails because the IPAM address pools are +// exhausted (leaked networks from prior sessions), it prunes those and retries once. func (dr *DockerRuntime) EnsureNetwork(ctx context.Context) error { - if dr.networkCreated { - return nil - } - if err := exec.CommandContext(ctx, "docker", "network", "inspect", dr.networkName).Run(); err == nil { - dr.networkCreated = true + dr.networkCreated.Store(true) return nil } @@ -54,7 +51,7 @@ func (dr *DockerRuntime) EnsureNetwork(ctx context.Context) error { } } - dr.networkCreated = true + dr.networkCreated.Store(true) logger.Info("docker network ready", "session", dr.sessionID, "network", dr.networkName) return nil } @@ -79,20 +76,20 @@ func (dr *DockerRuntime) createNetwork(ctx context.Context) error { // (keeping networkCreated set so a later call retries once the network frees // up). It is never an error to fail here - shutdown must not block. func (dr *DockerRuntime) CleanupNetwork(ctx context.Context) error { - if !dr.networkCreated { + if !dr.networkCreated.Load() { return nil } output, err := exec.CommandContext(ctx, "docker", "network", "rm", dr.networkName).CombinedOutput() if err == nil { - dr.networkCreated = false + dr.networkCreated.Store(false) logger.Info("docker network removed successfully", "network", dr.networkName) return nil } switch gone, inUse := interpretNetworkRm(string(output)); { case gone: - dr.networkCreated = false + dr.networkCreated.Store(false) case inUse: logger.Debug("docker network still in use by another session; leaving in place", "network", dr.networkName) default: diff --git a/internal/platform/container/podman_runtime.go b/internal/platform/container/podman_runtime.go index be01b2ef1..8b341eac1 100644 --- a/internal/platform/container/podman_runtime.go +++ b/internal/platform/container/podman_runtime.go @@ -5,6 +5,7 @@ import ( "fmt" "os/exec" "strings" + "sync/atomic" logger "github.com/inference-gateway/cli/internal/platform/logger" ) @@ -13,7 +14,7 @@ import ( type PodmanRuntime struct { sessionID string networkName string - networkCreated bool + networkCreated atomic.Bool } // NewPodmanRuntime creates a new Podman runtime manager @@ -30,16 +31,12 @@ func (pr *PodmanRuntime) GetNetworkName() string { } // EnsureNetwork creates the shared Podman network if it doesn't exist. The -// network is reused across sessions, so at most one ever exists. If creation -// fails because the IPAM address pools are exhausted (leaked networks from -// prior sessions), it prunes those and retries once. +// network is reused across sessions and any of them may remove it, so every +// call checks again. If creation fails because the IPAM address pools are +// exhausted (leaked networks from prior sessions), it prunes those and retries once. func (pr *PodmanRuntime) EnsureNetwork(ctx context.Context) error { - if pr.networkCreated { - return nil - } - if err := exec.CommandContext(ctx, "podman", "network", "inspect", pr.networkName).Run(); err == nil { - pr.networkCreated = true + pr.networkCreated.Store(true) return nil } @@ -54,7 +51,7 @@ func (pr *PodmanRuntime) EnsureNetwork(ctx context.Context) error { } } - pr.networkCreated = true + pr.networkCreated.Store(true) logger.Info("podman network ready", "session", pr.sessionID, "network", pr.networkName) return nil } @@ -79,20 +76,20 @@ func (pr *PodmanRuntime) createNetwork(ctx context.Context) error { // (keeping networkCreated set so a later call retries once the network frees // up). It is never an error to fail here - shutdown must not block. func (pr *PodmanRuntime) CleanupNetwork(ctx context.Context) error { - if !pr.networkCreated { + if !pr.networkCreated.Load() { return nil } output, err := exec.CommandContext(ctx, "podman", "network", "rm", pr.networkName).CombinedOutput() if err == nil { - pr.networkCreated = false + pr.networkCreated.Store(false) logger.Info("podman network removed successfully", "network", pr.networkName) return nil } switch gone, inUse := interpretNetworkRm(string(output)); { case gone: - pr.networkCreated = false + pr.networkCreated.Store(false) case inUse: logger.Debug("podman network still in use by another session; leaving in place", "network", pr.networkName) default: diff --git a/internal/protocols/a2a/infrastructure/address.go b/internal/protocols/a2a/infrastructure/address.go new file mode 100644 index 000000000..6030d6eca --- /dev/null +++ b/internal/protocols/a2a/infrastructure/address.go @@ -0,0 +1,74 @@ +package infrastructure + +import ( + "net/url" + "sync" + + config "github.com/inference-gateway/cli/config" +) + +// agentRoute sends requests for an agent's configured origin to the host its +// container actually listens on. +type agentRoute struct { + origin string + host string +} + +// agentRoutes holds the local agents the supervisor had to start on another +// host port than agents.yaml names. The configured URL stays the agent's identity. +// ponytail: one process-wide table, as one supervisor runs per process. Pass it +// through NewClient if a process ever runs several. +var agentRoutes = struct { + sync.RWMutex + byAgent map[string]agentRoute +}{byAgent: map[string]agentRoute{}} + +// RouteAgent sends requests for the named agent, configured at configuredURL, +// to the host of runningAt. Running on the configured origin clears the route. +func RouteAgent(name, configuredURL, runningAt string) { + running, err := url.Parse(runningAt) + origin := config.URLOrigin(configuredURL) + if err != nil || origin == "" || config.URLOrigin(runningAt) == origin { + UnrouteAgent(name) + return + } + agentRoutes.Lock() + defer agentRoutes.Unlock() + agentRoutes.byAgent[name] = agentRoute{origin: origin, host: running.Host} +} + +// UnrouteAgent sends requests for the named agent back to its configured URL. +func UnrouteAgent(name string) { + agentRoutes.Lock() + defer agentRoutes.Unlock() + delete(agentRoutes.byAgent, name) +} + +// runningURL returns agentURL on the host its agent actually runs on, or +// agentURL itself when the agent runs where it is configured. +func runningURL(agentURL string) string { + host, routed := routedHost(config.URLOrigin(agentURL)) + if !routed { + return agentURL + } + u, err := url.Parse(agentURL) + if err != nil { + return agentURL + } + u.Host = host + return u.String() +} + +func routedHost(origin string) (string, bool) { + if origin == "" { + return "", false + } + agentRoutes.RLock() + defer agentRoutes.RUnlock() + for _, route := range agentRoutes.byAgent { + if route.origin == origin { + return route.host, true + } + } + return "", false +} diff --git a/internal/protocols/a2a/infrastructure/address_test.go b/internal/protocols/a2a/infrastructure/address_test.go new file mode 100644 index 000000000..14ee24dba --- /dev/null +++ b/internal/protocols/a2a/infrastructure/address_test.go @@ -0,0 +1,34 @@ +package infrastructure + +import ( + "fmt" + "net/http/httptest" + "testing" +) + +func TestNewClient_ReachesAnAgentMovedToAnotherPort(t *testing.T) { + vacated := httptest.NewServer(nil) + configuredURL := vacated.URL + vacated.Close() + + agent := &authRecorder{} + srv := httptest.NewServer(agent) + defer srv.Close() + writeAgentsYAML(t, fmt.Sprintf("agents:\n - name: research\n url: %s\n auth:\n token_env: RESEARCH_TOKEN\n", configuredURL)) + t.Setenv("RESEARCH_TOKEN", "s3cret") + + RouteAgent("research", configuredURL, srv.URL) + t.Cleanup(func() { UnrouteAgent("research") }) + + if _, err := NewClient(configuredURL).GetAgentCard(t.Context()); err != nil { + t.Fatalf("request to the routed agent failed: %v", err) + } + if got := agent.headers(); len(got) != 1 || got[0] != "Bearer s3cret" { + t.Fatalf("Authorization headers = %q, want the agent's credentials on its running origin", got) + } + + UnrouteAgent("research") + if _, err := NewClient(configuredURL).GetAgentCard(t.Context()); err == nil { + t.Fatal("an unrouted agent was still reached on its old port") + } +} diff --git a/internal/protocols/a2a/infrastructure/auth.go b/internal/protocols/a2a/infrastructure/auth.go index c014a3904..96635faa1 100644 --- a/internal/protocols/a2a/infrastructure/auth.go +++ b/internal/protocols/a2a/infrastructure/auth.go @@ -95,13 +95,29 @@ func PolicyRefusal(err error) (string, bool) { return "The request was refused by a guardrail policy: " + envelope.Error.Message, true } +// methodNotFound is how the ADK client ends the error for a JSON-RPC +// "method not found" answer. +const methodNotFound = "(code: -32601)" + +// MethodUnsupported describes err for the model when the agent does not +// implement the A2A method that was called, which no retry can change. +func MethodUnsupported(agentURL string, err error) (string, bool) { + if err == nil || !strings.Contains(err.Error(), methodNotFound) { + return "", false + } + return fmt.Sprintf("A2A agent %q does not implement the A2A method that was called, so it likely speaks another A2A protocol version than this CLI (v1.0). Retrying will not help.", agentDisplayName(agentURL)), true +} + // Rejection describes err for the model when the agent turned the request -// down for good, so retrying is pointless: a policy refusal or an -// authentication failure. +// down for good, so retrying is pointless: a policy refusal, an +// authentication failure or a method the agent does not implement. func Rejection(agentURL string, err error) (string, bool) { if message, ok := PolicyRefusal(err); ok { return message, true } + if message, ok := MethodUnsupported(agentURL, err); ok { + return message, true + } return AuthFailure(agentURL, err) } @@ -151,7 +167,7 @@ type authTransport struct { // authenticates with the credentials agents.yaml gives the agent, or with the // gateway credential when the agent is the gateway. Any other agent gets the default. func newAuthTransport(agentURL string) http.RoundTripper { - origin := config.URLOrigin(agentURL) + origin := config.URLOrigin(runningURL(agentURL)) if agent, ok := configuredAgent(agentURL); ok && agent.Auth != nil { return authTransport{base: http.DefaultTransport, origin: origin, token: func() (string, error) { return bearerToken(agent) }} } @@ -282,7 +298,7 @@ func declaredTokenEndpoint(agentURL, pinnedIssuer string) (tokenEndpoint, error) ctx, cancel := context.WithTimeout(context.Background(), oidcHTTPClient.Timeout) defer cancel() - card, err := client.NewClient(agentURL).GetAgentCard(ctx) + card, err := client.NewClient(runningURL(agentURL)).GetAgentCard(ctx) if err != nil { return tokenEndpoint{}, fmt.Errorf("reading the security schemes of the agent card: %w", err) } diff --git a/internal/protocols/a2a/infrastructure/auth_test.go b/internal/protocols/a2a/infrastructure/auth_test.go index e3046ab51..88424be2b 100644 --- a/internal/protocols/a2a/infrastructure/auth_test.go +++ b/internal/protocols/a2a/infrastructure/auth_test.go @@ -432,6 +432,13 @@ func TestRejection_ReadsTheGatewayResponses(t *testing.T) { want: []string{"refused by a guardrail policy", "request contains sensitive payment information"}, wantAbsent: "Authentication failed", }, + { + name: "method not found says retrying will not help", + status: http.StatusOK, + body: `{"jsonrpc":"2.0","id":"1","error":{"code":-32601,"message":"method not found: GetTask"}}`, + want: []string{"does not implement the A2A method", "Retrying will not help"}, + wantAbsent: "Authentication failed", + }, { name: "403 without a JSON-RPC error stays an authentication failure", status: http.StatusForbidden, diff --git a/internal/protocols/a2a/infrastructure/client.go b/internal/protocols/a2a/infrastructure/client.go index 17b50a959..8cf1cac5a 100644 --- a/internal/protocols/a2a/infrastructure/client.go +++ b/internal/protocols/a2a/infrastructure/client.go @@ -9,10 +9,10 @@ import ( // NewClient returns an ADK client for the agent at agentURL whose requests // carry the session's trace context, so remote spans nest under ours, and the -// credentials agents.yaml gives the agent. Every request activates the usage -// extension, so tasks come back with their usage. +// credentials agents.yaml gives the agent. They go to the host the agent +// actually runs on. Every request activates the usage extension. func NewClient(agentURL string) client.A2AClient { - cfg := client.DefaultConfig(agentURL) + cfg := client.DefaultConfig(runningURL(agentURL)) cfg.Transport = telemetry.PropagationTransport(newAuthTransport(agentURL)) cfg.Headers["A2A-Extensions"] = adk.UsageExtensionURI return client.NewClientWithConfig(cfg) diff --git a/internal/protocols/a2a/submit_task.go b/internal/protocols/a2a/submit_task.go index a8b07043c..b71cfd82b 100644 --- a/internal/protocols/a2a/submit_task.go +++ b/internal/protocols/a2a/submit_task.go @@ -216,6 +216,9 @@ func (t *SubmitTaskTool) Execute(ctx context.Context, args map[string]any) (*age return t.errorResult(args, startTime, "Failed to parse task submission response") } + if sendResponse.Task == nil && sendResponse.Message != nil { + return t.directReply(args, startTime, agentURL, *sendResponse.Message) + } if sendResponse.Task == nil || sendResponse.Task.ID == "" { return t.errorResult(args, startTime, "Task submitted but no task ID received") } @@ -877,6 +880,29 @@ func (t *SubmitTaskTool) ShouldAlwaysExpand() bool { } // errorResult creates an error result +// directReply is the result when the agent answers with a message instead of +// a task. The answer is final, so there is nothing to track or poll. +func (t *SubmitTaskTool) directReply(args map[string]any, startTime time.Time, agentURL string, reply adk.Message) (*agentdomain.ToolExecutionResult, error) { + var contextID string + if reply.ContextID != nil { + contextID = *reply.ContextID + } + return &agentdomain.ToolExecutionResult{ + ToolName: ToolSubmitTask, + Arguments: args, + Success: true, + Duration: time.Since(startTime), + Data: SubmitTaskResult{ + ContextID: contextID, + AgentURL: agentURL, + State: string(adk.TaskStateCompleted), + Success: true, + Message: fmt.Sprintf("%s answered directly", agentURL), + TaskResult: textFromParts(reply.Parts), + }, + }, nil +} + func (t *SubmitTaskTool) errorResult(args map[string]any, startTime time.Time, errorMsg string) (*agentdomain.ToolExecutionResult, error) { agentURL, _ := args["agent_url"].(string) diff --git a/internal/protocols/a2a/submit_task_test.go b/internal/protocols/a2a/submit_task_test.go index 8726a9e5d..b8c2d030f 100644 --- a/internal/protocols/a2a/submit_task_test.go +++ b/internal/protocols/a2a/submit_task_test.go @@ -132,6 +132,41 @@ func TestSubmitTaskTool_Execute_RequestsImmediateReturn(t *testing.T) { assert.True(t, *sendRequest.Configuration.ReturnImmediately, "A2A_SubmitTask returns after the task is created and the background poller takes over") } +func TestSubmitTaskTool_Execute_DirectMessageReplyIsTheAnswer(t *testing.T) { + cfg := &config.Config{ + A2A: config.A2AConfig{ + Enabled: true, + Tools: config.A2AToolsConfig{ + SubmitTask: config.SubmitTaskToolConfig{Enabled: true}, + }, + }, + } + + reply := adk.Message{ + MessageID: "msg-1", + ContextID: ptrString("context-1"), + Role: adk.RoleAgent, + Parts: []adk.Part{{Text: ptrString("the answer is 42")}}, + } + mockClient := &adkmocks.FakeA2AClient{} + mockClient.SendTaskReturns(&adk.JSONRPCSuccessResponse{Result: adk.SendMessageResponse{Message: &reply}}, nil) + + tool := NewSubmitTaskToolWithClient(cfg, nil, nil, nil, mockClient) + + result, err := tool.Execute(context.Background(), map[string]any{ + "agent_url": "http://test-agent", + "task_description": "What is the answer?", + }) + + require.NoError(t, err) + require.True(t, result.Success, result.Error) + data, ok := result.Data.(SubmitTaskResult) + require.True(t, ok) + assert.Equal(t, "the answer is 42", data.TaskResult) + assert.Equal(t, "context-1", data.ContextID) + assert.Empty(t, data.TaskID, "a message reply starts no task to poll") +} + func TestSubmitTaskTool_Execute_Tenant(t *testing.T) { const gatewayURL = "http://gateway:8080" pausedTaskID := "mock-b:task-b" diff --git a/internal/protocols/a2a/supervisor.go b/internal/protocols/a2a/supervisor.go index a9d6cff5e..b10692cef 100644 --- a/internal/protocols/a2a/supervisor.go +++ b/internal/protocols/a2a/supervisor.go @@ -2,7 +2,9 @@ package a2a import ( "context" + "errors" "fmt" + "io/fs" "maps" "net/http" "net/url" @@ -11,6 +13,7 @@ import ( "path/filepath" "reflect" "slices" + "strconv" "strings" "sync" "time" @@ -29,6 +32,9 @@ import ( const ( // AgentContainerPrefix is the naming prefix for agent containers AgentContainerPrefix = "inference-agent-" + + // readinessLogLines is how much of a container's log a failed start reports. + readinessLogLines = 20 ) // AgentSupervisor manages the lifecycle of A2A agent containers (local and external) @@ -248,6 +254,7 @@ func (s *AgentSupervisor) stopRun(name string) { delete(s.agentStates, name) delete(s.assignedPorts, name) s.containersMutex.Unlock() + a2ainfra.UnrouteAgent(name) } // livenessInterval is how often a ready agent is probed again. @@ -429,6 +436,9 @@ func (s *AgentSupervisor) StartAgent(ctx context.Context, agent config.AgentEntr s.notifyStatus(agent.Name, a2adomain.AgentStateWaitingReady, "Waiting for health check", agent.URL, agent.OCI) if err := s.waitForReady(ctx, agent); err != nil { + if diagnosis := s.readinessDiagnosis(ctx, agent.Name); diagnosis != "" { + err = fmt.Errorf("%w, %s", err, diagnosis) + } if stopErr := s.StopAgent(context.WithoutCancel(ctx), agent.Name); stopErr != nil { logger.Warn("failed to stop agent during error cleanup", "name", agent.Name, "error", stopErr) } @@ -451,7 +461,7 @@ func (s *AgentSupervisor) StartAgent(ctx context.Context, agent config.AgentEntr // startLocalAgentProbe starts a periodic health check for a local (docker) agent func (s *AgentSupervisor) startLocalAgentProbe(ctx context.Context, agent config.AgentEntry) { - healthURL := strings.TrimSuffix(agent.URL, "/") + "/health" + healthURL := strings.TrimSuffix(s.localURL(agent), "/") + "/health" s.goAgent(ctx, agent.Name, func(ctx context.Context) { ticker := time.NewTicker(s.livenessInterval()) @@ -669,6 +679,12 @@ func (s *AgentSupervisor) startContainer(ctx context.Context, agent config.Agent args = append(args, agent.OCI) + if s.containerRuntime != nil { + if err := s.containerRuntime.EnsureNetwork(ctx); err != nil { + return err + } + } + cmd := exec.CommandContext(ctx, "docker", args...) var outputBuf, errBuf strings.Builder cmd.Stdout = &outputBuf @@ -682,6 +698,7 @@ func (s *AgentSupervisor) startContainer(ctx context.Context, agent config.Agent s.containersMutex.Lock() s.containers[agent.Name] = containerID s.containersMutex.Unlock() + a2ainfra.RouteAgent(agent.Name, agent.URL, setURLPort(agent.URL, assignedPort)) return nil } @@ -762,7 +779,8 @@ func resolveAgentEnv(env, dotEnvVars, authKeys map[string]string) map[string]str return resolvedEnv } -// loadDotEnvFile loads environment variables from .env file in the current directory +// loadDotEnvFile loads the optional .env file in the current directory. A +// missing file yields no variables and no error. func (s *AgentSupervisor) loadDotEnvFile() (map[string]string, error) { cwd, err := os.Getwd() if err != nil { @@ -770,8 +788,8 @@ func (s *AgentSupervisor) loadDotEnvFile() (map[string]string, error) { } dotEnvPath := filepath.Join(cwd, ".env") - if _, err := os.Stat(dotEnvPath); os.IsNotExist(err) { - return nil, fmt.Errorf(".env file not found at %s", dotEnvPath) + if _, err := os.Stat(dotEnvPath); errors.Is(err, fs.ErrNotExist) { + return nil, nil } envMap, err := gotenv.Read(dotEnvPath) @@ -814,9 +832,37 @@ func (s *AgentSupervisor) runningContainerID(expectedName string) string { return "" } +// localURL returns the URL this session's container for agent listens on: the +// configured URL on the host port the agent was actually given. +func (s *AgentSupervisor) localURL(agent config.AgentEntry) string { + s.containersMutex.Lock() + port, assigned := s.assignedPorts[agent.Name] + s.containersMutex.Unlock() + if !assigned { + return agent.URL + } + return setURLPort(agent.URL, port) +} + +// readinessDiagnosis says why the agent's container never became ready, before +// --rm deletes it with its logs: the tail of its log, or that it already exited. +func (s *AgentSupervisor) readinessDiagnosis(ctx context.Context, agentName string) string { + s.containersMutex.Lock() + containerID := s.containers[agentName] + s.containersMutex.Unlock() + if containerID == "" || ctx.Err() != nil { + return "" + } + output, err := exec.CommandContext(ctx, "docker", "logs", "--tail", strconv.Itoa(readinessLogLines), containerID).CombinedOutput() + if err != nil { + return "the container exited before it became ready" + } + return "last container logs:\n" + strings.TrimSpace(string(output)) +} + // waitForReady waits for an agent to become ready func (s *AgentSupervisor) waitForReady(ctx context.Context, agent config.AgentEntry) error { - healthURL := strings.TrimSuffix(agent.URL, "/") + "/health" + healthURL := strings.TrimSuffix(s.localURL(agent), "/") + "/health" timeout := 30 * time.Second deadline := time.Now().Add(timeout) diff --git a/internal/protocols/a2a/supervisor_test.go b/internal/protocols/a2a/supervisor_test.go index 53787b055..b8b5b7463 100644 --- a/internal/protocols/a2a/supervisor_test.go +++ b/internal/protocols/a2a/supervisor_test.go @@ -102,9 +102,8 @@ func TestAgentSupervisor_loadDotEnvFile_NotFound(t *testing.T) { supervisor := NewAgentSupervisor(sessionID, cfg, agentsConfig, nil, nil) envMap, err := supervisor.loadDotEnvFile() - require.Error(t, err) + require.NoError(t, err, "a project without a .env file is normal") require.Nil(t, envMap) - require.Contains(t, err.Error(), ".env file not found") } func TestAgentSupervisor_loadDotEnvFile_InvalidFormat(t *testing.T) { @@ -222,6 +221,16 @@ func TestAgentTelemetryEnv(t *testing.T) { require.Equal(t, "https://otel.example.com:4318", agentTelemetryEnv("https://otel.example.com:4318")["A2A_OTEL_EXPORTER_OTLP_ENDPOINT"]) } +func TestAgentSupervisor_localURL(t *testing.T) { + supervisor := NewAgentSupervisor(convdomain.GenerateSessionID(), &config.Config{}, &config.AgentsConfig{}, nil, nil) + agent := config.AgentEntry{Name: "research", URL: "http://localhost:8081"} + + require.Equal(t, "http://localhost:8081", supervisor.localURL(agent), "an agent without a container keeps its configured URL") + + supervisor.assignedPorts[agent.Name] = 8082 + require.Equal(t, "http://localhost:8082", supervisor.localURL(agent), "probes must reach the port the container was given") +} + func TestSetURLPort(t *testing.T) { cases := []struct { in string diff --git a/internal/protocols/mcp/supervisor.go b/internal/protocols/mcp/supervisor.go index 598eb18f2..2489a737b 100644 --- a/internal/protocols/mcp/supervisor.go +++ b/internal/protocols/mcp/supervisor.go @@ -711,6 +711,9 @@ func (s *Supervisor) startContainer(ctx context.Context, server config.MCPServer env[key] = os.ExpandEnv(value) } + if err := s.containerRuntime.EnsureNetwork(ctx); err != nil { + return err + } containerID, err := s.containerRuntime.RunContainer(ctx, containerruntime.RunContainerOptions{ Name: containerName, Image: server.OCI, diff --git a/internal/tools/Grep.yaml b/internal/tools/Grep.yaml index e03549c5e..bec96be44 100644 --- a/internal/tools/Grep.yaml +++ b/internal/tools/Grep.yaml @@ -4,7 +4,7 @@ description: | Usage: - ALWAYS use Grep for search tasks. NEVER invoke `grep` or `rg` as a Bash command. The Grep tool has been optimized for correct permissions and access. - - Supports full regex syntax (e.g., "log.*Error", "function\s+\w+") + - Regex syntax is RE2 (e.g., "log.*Error", "function\s+\w+") - no lookahead, lookbehind or backreferences - Filter files with glob parameter (e.g., "*.js", "**/*.tsx") or type parameter (e.g., "js", "py", "rust") - Output modes: "content" shows matching lines, "files_with_matches" shows only file paths (default), "count" shows match counts - Use the Agent tool for open-ended searches requiring multiple rounds diff --git a/internal/tools/SendSubagentInput.yaml b/internal/tools/SendSubagentInput.yaml index cd05b318e..2e4d49f58 100644 --- a/internal/tools/SendSubagentInput.yaml +++ b/internal/tools/SendSubagentInput.yaml @@ -14,7 +14,7 @@ parameters: description: The subagent id from ListSubagents or the Agent result submit: type: boolean - description: Interactive subagents only. Press Enter to submit a prompt and wait for the subagent to finish (default true). Set false to only send keys for TUI navigation - then inspect with ReadSubagentScreen. + description: Interactive subagents only. Press Enter to submit the prompt (default true). Set false to only send keys for TUI navigation - then inspect with ReadSubagentScreen. text: type: string description: The message to send (a follow-up prompt, new information, a correction). Typed literally into an interactive pane. diff --git a/internal/tools/grep.go b/internal/tools/grep.go index 435ecc219..4b3bd2db6 100644 --- a/internal/tools/grep.go +++ b/internal/tools/grep.go @@ -299,7 +299,8 @@ func (t *GrepTool) performRipgrepSearch(ctx context.Context, pattern string, arg rgArgs := t.buildRipgrepArgs(outputMode, args) rgArgs = append(rgArgs, "-e", pattern, "--", searchPath) - result, err := t.executeRipgrep(ctx, rgArgs, outputMode, pattern, start) + headLimit, _ := args["head_limit"].(float64) + result, err := t.executeRipgrep(ctx, rgArgs, outputMode, pattern, int(headLimit), start) if err != nil { return nil, err } @@ -383,17 +384,11 @@ func (t *GrepTool) addSearchOptions(rgArgs []string, args map[string]any) []stri } } - if headLimit, exists := args["head_limit"]; exists { - if headLimitFloat, ok := headLimit.(float64); ok { - rgArgs = append(rgArgs, "--max-count", strconv.Itoa(int(headLimitFloat))) - } - } - return rgArgs } // executeRipgrep runs the ripgrep command and processes the output -func (t *GrepTool) executeRipgrep(ctx context.Context, rgArgs []string, outputMode, pattern string, start time.Time) (*GrepResult, error) { +func (t *GrepTool) executeRipgrep(ctx context.Context, rgArgs []string, outputMode, pattern string, headLimit int, start time.Time) (*GrepResult, error) { cmd := exec.CommandContext(ctx, t.ripgrepPath, rgArgs...) output, err := cmd.Output() if err != nil { @@ -414,13 +409,14 @@ func (t *GrepTool) executeRipgrep(ctx context.Context, rgArgs []string, outputMo return nil, fmt.Errorf("ripgrep execution failed: %w", err) } - result := t.parseRipgrepOutput(string(output), outputMode, pattern) + result := t.parseRipgrepOutput(string(output), outputMode, pattern, headLimit) result.Duration = time.Since(start).String() return result, nil } -// parseRipgrepOutput parses ripgrep output into GrepResult -func (t *GrepTool) parseRipgrepOutput(output, outputMode, pattern string) *GrepResult { +// parseRipgrepOutput parses ripgrep output into GrepResult, keeping only the +// first headLimit lines when it is set, like "| head -N". +func (t *GrepTool) parseRipgrepOutput(output, outputMode, pattern string, headLimit int) *GrepResult { result := &GrepResult{ Pattern: pattern, OutputMode: outputMode, @@ -433,6 +429,9 @@ func (t *GrepTool) parseRipgrepOutput(output, outputMode, pattern string) *GrepR if len(lines) == 1 && lines[0] == "" { return result } + if headLimit > 0 { + lines = lines[:min(len(lines), headLimit)] + } switch outputMode { case "files_with_matches": diff --git a/internal/tools/grep_test.go b/internal/tools/grep_test.go index e3bb8a681..aa858798c 100644 --- a/internal/tools/grep_test.go +++ b/internal/tools/grep_test.go @@ -648,6 +648,17 @@ func TestGrepTool_RipgrepDetection(t *testing.T) { t.Logf("Ripgrep detected at: %s", tool.ripgrepPath) } +func TestGrepTool_RipgrepHeadLimitCapsTheTotal(t *testing.T) { + tool := &GrepTool{} + output := "a.go:1:x\na.go:2:y\nb.go:1:z\nb.go:2:w\n" + + result := tool.parseRipgrepOutput(output, "content", "x", 3) + + if len(result.Matches) != 3 || result.Total != 3 { + t.Fatalf("got %d matches (total %d), want 3 across both files like | head -3", len(result.Matches), result.Total) + } +} + func TestGrepTool_HybridSearch(t *testing.T) { cfg := &config.Config{ Tools: config.ToolsConfig{