diff --git a/internal/README.md b/internal/README.md index e8d4cf0..07738f0 100644 --- a/internal/README.md +++ b/internal/README.md @@ -205,6 +205,38 @@ a2a send -a --context-id "Related question" | `--context-id ` | Group this turn under an existing context (new task). | | `--history ` | Request `n` history messages in the response. | +### `chat` - Chat with an Agent + +Hold a back-and-forth conversation with one agent. Each message continues the +same conversation: the context id is carried across turns, and a task waiting on +`INPUT_REQUIRED` is continued with your next message. A failed turn prints the +error and the session stays open. An `AUTH_REQUIRED` task ends the session with an +explanation. Ctrl-C cancels a running turn, asks the agent to cancel the task it +started, and returns to the prompt; a second Ctrl-C abandons that cancel request. +A turn that exceeds `--timeout` also cancels its task. Type `/exit` or `/quit`, +press Ctrl-D, or press Ctrl-C at the prompt to exit; the context and task ids are printed so you can resume with +`send` or `chat`. + +```bash +a2a chat -a +a2a chat -a "Hello, what can you do?" + +# Stream each reply as it arrives +a2a chat -a --stream + +# Resume a conversation (same task) +a2a chat -a --context-id --task-id +``` + +| Flag | Description | +|---|---| +| `--stream` | Use `SendStreamingMessage` for each turn. Events are printed incrementally. Falls back to polling if the server does not support streaming. | +| `--task-id ` | Continue an existing task with the first message. | +| `--context-id ` | Start the conversation in an existing context. | + +`chat` requires a terminal on standard input and output; otherwise it is a usage +error. Use `send` to send a single message from a script. + ### `task get` - Get Task Details ```bash diff --git a/internal/chat/session.go b/internal/chat/session.go new file mode 100644 index 0000000..57d1c36 --- /dev/null +++ b/internal/chat/session.go @@ -0,0 +1,239 @@ +// Copyright 2026 The A2A Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package chat runs a conversation with an agent, carrying its identifiers from one turn to the next. +package chat + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "strings" + + "github.com/a2aproject/a2a-go/v2/a2a" +) + +// Inputs that end a session. +const ( + Quit = "/quit" + Exit = "/exit" +) + +// maxLineSize bounds a single line of input, so a large paste does not end the session. +const maxLineSize = 1 << 20 + +// NotifyFunc returns a context canceled when the user interrupts the process. +type NotifyFunc func(context.Context) (context.Context, context.CancelFunc) + +// Config is how a Session talks to the agent and the user. +type Config struct { + // Send sends one message and must pass every event it receives to Session.Track. + Send func(ctx context.Context, msg *a2a.Message) error + // Cancel asks the agent to cancel a task. + Cancel func(ctx context.Context, id a2a.TaskID) error + // RenderError reports a failed turn to the user. + RenderError func(error) + Notify NotifyFunc + // Out receives the prompt and session messages. + Out io.Writer +} + +// Session is a conversation with one agent. +type Session struct { + cfg Config + contextID string + taskID a2a.TaskID + authRequired bool + // running is the task the current turn is working on, once the agent has named it. + running a2a.TaskID +} + +// NewSession starts a conversation, optionally in an existing context or task. +func NewSession(cfg Config, contextID string, taskID a2a.TaskID) *Session { + return &Session{cfg: cfg, contextID: contextID, taskID: taskID} +} + +// Run reads messages from in until Quit or Exit, the end of input, an interrupt at the prompt, +// or a task that needs authentication. A non-empty first is sent before reading input. +func (s *Session) Run(ctx context.Context, in io.Reader, first string) error { + defer s.printResume() + + if first != "" && s.reply(ctx, first) { + return nil + } + + stop := make(chan struct{}) + defer close(stop) + lines, scanErr := scanLines(in, stop) + for { + s.printf("> ") + line, ok, err := s.next(ctx, lines, scanErr) + if !ok { + s.printf("\n") + return err + } + text := strings.TrimSpace(line) + if text == "" { + continue + } + if text == Quit || text == Exit { + return nil + } + if s.reply(ctx, text) { + return nil + } + } +} + +// Track follows the conversation's identifiers through the events of a turn. +func (s *Session) Track(event a2a.Event) error { + switch e := event.(type) { + case *a2a.Task: + s.follow(e.ContextID, e.ID, e.Status.State) + case *a2a.TaskStatusUpdateEvent: + s.follow(e.ContextID, e.TaskID, e.Status.State) + case *a2a.Message: + if e.ContextID != "" { + s.contextID = e.ContextID + } + s.taskID, s.running, s.authRequired = "", "", false + } + return nil +} + +// scanLines reads lines on a goroutine so an interrupt is noticed while waiting for +// input. It sends the scan error, if any, before closing lines, and quits once stop is +// closed; a read already blocked on a terminal only returns when the process exits. +func scanLines(in io.Reader, stop <-chan struct{}) (<-chan string, <-chan error) { + lines := make(chan string) + scanErr := make(chan error, 1) + go func() { + defer close(lines) + scanner := bufio.NewScanner(in) + scanner.Buffer(nil, maxLineSize) + for scanner.Scan() { + select { + case lines <- scanner.Text(): + case <-stop: + return + } + } + scanErr <- scanner.Err() + }() + return lines, scanErr +} + +// next waits for a line of input. It reports !ok at the end of input, or when +// interrupted at the prompt. +func (s *Session) next(ctx context.Context, lines <-chan string, scanErr <-chan error) (string, bool, error) { + ctx, stop := s.cfg.Notify(ctx) + defer stop() + select { + case line, ok := <-lines: + if !ok { + return "", false, <-scanErr + } + return line, true, nil + case <-ctx.Done(): + return "", false, nil + } +} + +// reply runs one turn and reports whether the session should end. A turn that is +// interrupted or times out cancels the task it left running, and a failed turn is +// reported; either way the conversation goes on. +func (s *Session) reply(ctx context.Context, text string) (done bool) { + turnCtx, stop := s.cfg.Notify(ctx) + defer stop() + done, err := s.turn(turnCtx, text) + switch { + case err == nil: + case turnCtx.Err() != nil: + s.printf("Interrupted.\n") + s.cancelRunning(ctx) + case errors.Is(err, context.DeadlineExceeded): + s.cfg.RenderError(err) + s.cancelRunning(ctx) + default: + s.cfg.RenderError(err) + } + return done +} + +// cancelRunning asks the agent to stop the task an abandoned turn left running. The +// request gets its own interrupt scope, so another interrupt abandons it. +func (s *Session) cancelRunning(ctx context.Context) { + id := s.running + if id == "" { + return + } + ctx, stop := s.cfg.Notify(ctx) + defer stop() + if err := s.cfg.Cancel(ctx, id); err != nil { + s.printf("Could not cancel task %s: %v\n", id, err) + return + } + s.printf("Canceled task %s.\n", id) + s.taskID, s.running = "", "" +} + +// turn sends one message and reports done when the conversation cannot continue. +func (s *Session) turn(ctx context.Context, text string) (done bool, err error) { + s.running = s.taskID + msg := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart(text)) + msg.ContextID = s.contextID + msg.TaskID = s.taskID + if err := s.cfg.Send(ctx, msg); err != nil { + return false, err + } + if s.authRequired { + s.printf("The agent needs authentication to continue this task. " + + "Provide credentials with --auth or --svc-param, then resume it with send.\n") + return true, nil + } + return false, nil +} + +// follow keeps a task's id for the next turn only while the task waits on the user. +func (s *Session) follow(contextID string, taskID a2a.TaskID, state a2a.TaskState) { + s.contextID = contextID + s.taskID, s.authRequired = "", false + s.running = "" + if !state.Terminal() { + s.running = taskID + } + switch state { + case a2a.TaskStateInputRequired: + s.taskID = taskID + case a2a.TaskStateAuthRequired: + s.taskID, s.authRequired = taskID, true + } +} + +func (s *Session) printResume() { + if s.contextID != "" { + s.printf("contextId: %s\n", s.contextID) + } + if s.taskID != "" { + s.printf("taskId: %s\n", s.taskID) + } +} + +// printf writes a session message. A failed write to the user's terminal leaves +// nothing better to report it to, so it is dropped. +func (s *Session) printf(format string, args ...any) { + _, _ = fmt.Fprintf(s.cfg.Out, format, args...) +} diff --git a/internal/cli/chat.go b/internal/cli/chat.go new file mode 100644 index 0000000..5ea372e --- /dev/null +++ b/internal/cli/chat.go @@ -0,0 +1,122 @@ +// Copyright 2026 The A2A Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cli + +import ( + "context" + "fmt" + "os" + "time" + + "github.com/spf13/cobra" + + "github.com/a2aproject/a2a-cli/internal/chat" + "github.com/a2aproject/a2a-cli/internal/clierr" + "github.com/a2aproject/a2a-go/v2/a2a" + "github.com/a2aproject/a2a-go/v2/a2aclient" +) + +type chatFlags struct { + stream bool + taskID string + contextID string + pollInterval time.Duration +} + +func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { + var flags chatFlags + + cmd := &cobra.Command{ + Use: "chat [message]", + Short: "Chat interactively with an agent", + Long: "Start an interactive chat session with an agent. Chat reuses the same context ID " + + "across turns and continues tasks that require input.\n\n" + + "Controls:\n" + + " " + chat.Exit + ", " + chat.Quit + ", Ctrl-D Exit and print the context and task IDs to resume later\n" + + " Ctrl-C Cancel an active turn (and its agent task), or exit at the prompt\n\n" + + "Chat requires a terminal; use send to send a single message. " + + "Use chat --context-id and --task-id to resume a conversation.", + Args: cobra.MaximumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if !deps.isTerminal() { + return clierr.Usage("chat needs a terminal; use send to send a single message") + } + + baseCtx := withServiceParams(cmd.Context(), cfg) + ctx, cancel := context.WithTimeout(baseCtx, cfg.timeout) + client, err := newAgentClient(ctx, cfg) + cancel() + if err != nil { + return fmt.Errorf("failed to create a client: %w", err) + } + defer destroyClient(cfg, client) + + session := chat.NewSession(chat.Config{ + Send: func(ctx context.Context, msg *a2a.Message) error { + return sendChatMessage(ctx, cfg, client, &flags, deps.poller, msg) + }, + Cancel: func(ctx context.Context, id a2a.TaskID) error { + ctx, cancel := context.WithTimeout(ctx, cfg.timeout) + defer cancel() + _, err := client.CancelTask(ctx, &a2a.CancelTaskRequest{ID: id, Tenant: cfg.tenant}) + return err + }, + RenderError: func(err error) { cfg.renderError(err) }, + Notify: deps.notifyInterrupt, + Out: cfg.stderr(), + }, flags.contextID, a2a.TaskID(flags.taskID)) + cfg.handlers = append(cfg.handlers, session.Track) + + var first string + if len(args) == 1 { + first = args[0] + } + return session.Run(baseCtx, cmd.InOrStdin(), first) + }, + } + + f := cmd.Flags() + f.BoolVar(&flags.stream, "stream", false, "Stream responses (falls back to polling if unsupported)") + f.StringVar(&flags.taskID, "task-id", "", "Task ID to continue an existing task") + f.StringVar(&flags.contextID, "context-id", "", "Context ID to continue an existing conversation") + f.DurationVar(&flags.pollInterval, "poll-interval", 2*time.Second, "Interval between GetTask requests when polling") + + return cmd +} + +func sendChatMessage(ctx context.Context, cfg *globalConfig, client *a2aclient.Client, flags *chatFlags, poller pollerFunc, msg *a2a.Message) error { + req := &a2a.SendMessageRequest{Message: msg, Tenant: cfg.tenant} + if flags.stream { + return sendStreaming(ctx, cfg, client, req, poller, flags.pollInterval) + } + + ctx, cancel := context.WithTimeout(ctx, cfg.timeout) + defer cancel() + result, err := client.SendMessage(ctx, req) + if err != nil { + return fmt.Errorf("failed to send message: %w", err) + } + return cfg.handleEvent(result) +} + +// stdioIsTerminal reports whether both standard input and output are terminals. +func stdioIsTerminal() bool { + return isCharDevice(os.Stdin) && isCharDevice(os.Stdout) +} + +func isCharDevice(f *os.File) bool { + info, err := f.Stat() + return err == nil && info.Mode()&os.ModeCharDevice != 0 +} diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go new file mode 100644 index 0000000..823c900 --- /dev/null +++ b/internal/cli/chat_test.go @@ -0,0 +1,465 @@ +// Copyright 2026 The A2A Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cli + +import ( + "bytes" + "context" + "errors" + "io" + "iter" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/google/go-cmp/cmp" + + "github.com/a2aproject/a2a-cli/internal/clicfg" + "github.com/a2aproject/a2a-cli/internal/clierr" + "github.com/a2aproject/a2a-cli/internal/flagparse" + "github.com/a2aproject/a2a-cli/internal/output" + "github.com/a2aproject/a2a-cli/internal/polling" + "github.com/a2aproject/a2a-go/v2/a2a" + "github.com/a2aproject/a2a-go/v2/a2asrv" +) + +type chatRequest struct { + Text string + TaskID string + ContextID string +} + +// Scripted turns that are not task states. +const ( + chatTurnFails a2a.TaskState = "fail" + // chatTurnInterrupted interrupts the turn before the agent names a task. + chatTurnInterrupted a2a.TaskState = "interrupt" + // chatTurnHangs reports a working task and never finishes it. + chatTurnHangs a2a.TaskState = "hang" +) + +// chatAgent answers each turn with a task in the next scripted state. +type chatAgent struct { + mu sync.Mutex + states []a2a.TaskState + requests []chatRequest + tasks []*a2a.Task + canceled []string + + // hangCancel makes a cancel request interrupt chat and then wait for the client to give up. + hangCancel bool + + // interrupt stands in for Ctrl-C: it cancels the latest interrupt scope chat opened. + interrupt context.CancelFunc +} + +func (a *chatAgent) notifyInterrupt(ctx context.Context) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(ctx) + a.mu.Lock() + a.interrupt = cancel + a.mu.Unlock() + return ctx, cancel +} + +func (a *chatAgent) pressCtrlC() { + a.mu.Lock() + defer a.mu.Unlock() + a.interrupt() +} + +func (a *chatAgent) deps() deps { + return deps{isTerminal: func() bool { return true }, notifyInterrupt: a.notifyInterrupt} +} + +func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) { + t.Helper() + agent := &chatAgent{states: states} + release := make(chan struct{}) + handler := a2asrv.NewRESTHandler(a2asrv.NewHandler( + a2asrv.AgentExecutorFunc(func(ctx context.Context, ec *a2asrv.ExecutorContext) iter.Seq2[a2a.Event, error] { + return func(yield func(a2a.Event, error) bool) { + agent.mu.Lock() + turn := len(agent.requests) + agent.requests = append(agent.requests, chatRequest{ + Text: ec.Message.Parts[0].Text(), + TaskID: string(ec.Message.TaskID), + ContextID: ec.Message.ContextID, + }) + state := agent.states[turn] + task := &a2a.Task{ + ID: ec.TaskID, + ContextID: ec.ContextID, + Status: a2a.TaskStatus{State: a2a.TaskStateWorking}, + } + agent.tasks = append(agent.tasks, task) + agent.mu.Unlock() + + switch state { + case chatTurnFails: + yield(nil, errors.New("agent failed")) + return + case chatTurnInterrupted: + agent.pressCtrlC() + case chatTurnHangs: + if !yield(task, nil) { + return + } + default: + if yield(task, nil) { + yield(a2a.NewStatusUpdateEvent(task, state, nil), nil) + } + return + } + select { + case <-ctx.Done(): + case <-release: + } + } + }), + )) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + id, isCancel := strings.CutSuffix(strings.TrimPrefix(r.URL.Path, "/tasks/"), ":cancel") + if !isCancel { + handler.ServeHTTP(w, r) + return + } + agent.mu.Lock() + agent.canceled = append(agent.canceled, id) + hang := agent.hangCancel + agent.mu.Unlock() + if hang { + agent.pressCtrlC() + <-r.Context().Done() + return + } + handler.ServeHTTP(w, r) + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(release) }) + return agent, server.URL +} + +func runChat(t *testing.T, url string, terminal bool, stdin string, args ...string) (stdout, stderr string, err error) { + t.Helper() + var out bytes.Buffer + stderr, err = runChatTo(t, deps{isTerminal: func() bool { return terminal }}, url, strings.NewReader(stdin), &out, args...) + return out.String(), stderr, err +} + +func runChatTo(t *testing.T, d deps, url string, stdin io.Reader, out io.Writer, args ...string) (stderr string, err error) { + t.Helper() + var errBuf bytes.Buffer + cfg := &globalConfig{ + Printer: output.NewPrinter(out, output.ModeText), + svcParams: &flagparse.ServiceParams{}, + errOut: &errBuf, + } + d.poller = polling.Stream + d.cfgLoader = clicfg.LoadEmpty + root := mustNewRoot(t, cfg, d) + root.SetIn(stdin) + root.SetArgs(append([]string{"chat", "-e", url, "--transport", "rest"}, args...)) + err = root.Execute() + return errBuf.String(), err +} + +// interruptOnPrint presses Ctrl-C whenever chat prints to stdout. +type interruptOnPrint struct{ agent *chatAgent } + +func (w interruptOnPrint) Write(p []byte) (int, error) { + w.agent.pressCtrlC() + return len(p), nil +} + +func TestChatCarriesTheConversationAcrossTurns(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, a2a.TaskStateInputRequired, a2a.TaskStateCompleted, a2a.TaskStateCompleted) + + _, stderr, err := runChat(t, url, true, "two\nthree\n/quit\nnever sent\n", "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + first := agent.tasks[0] + want := []chatRequest{ + {Text: "one"}, + {Text: "two", TaskID: string(first.ID), ContextID: first.ContextID}, + {Text: "three", ContextID: first.ContextID}, + } + if diff := cmp.Diff(want, agent.requests); diff != "" { + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) + } + if !strings.Contains(stderr, "contextId: "+first.ContextID) { + t.Fatalf("a2a chat stderr = %q, want the context id", stderr) + } +} + +func TestChatEnds(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + states []a2a.TaskState + stdin string + wantRequests int + wantStderr string + }{ + { + name: "on EOF", + states: []a2a.TaskState{a2a.TaskStateInputRequired}, + stdin: "one\n", + wantRequests: 1, + wantStderr: "taskId: ", + }, + { + name: "on /quit", + states: []a2a.TaskState{a2a.TaskStateCompleted}, + stdin: "one\n/quit\ntwo\n", + wantRequests: 1, + wantStderr: "contextId: ", + }, + { + name: "on /exit", + states: []a2a.TaskState{a2a.TaskStateCompleted}, + stdin: "one\n/exit\ntwo\n", + wantRequests: 1, + wantStderr: "contextId: ", + }, + { + name: "when the agent needs authentication", + states: []a2a.TaskState{a2a.TaskStateAuthRequired}, + stdin: "one\ntwo\n", + wantRequests: 1, + wantStderr: "authentication", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, tt.states...) + + _, stderr, err := runChat(t, url, true, tt.stdin) + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + if got := len(agent.requests); got != tt.wantRequests { + t.Fatalf("a2a chat sent %d messages, want %d", got, tt.wantRequests) + } + if !strings.Contains(stderr, tt.wantStderr) { + t.Fatalf("a2a chat stderr = %q, want it to contain %q", stderr, tt.wantStderr) + } + }) + } +} + +func TestChatWithoutATerminalIsAUsageError(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t) + + _, _, err := runChat(t, url, false, "two\n", "one") + if ce := clierr.Classify(err); ce == nil || ce.Code != clierr.CodeUsage { + t.Fatalf("a2a chat error = %v, want a usage error", err) + } + if got := len(agent.requests); got != 0 { + t.Fatalf("a2a chat sent %d messages, want 0", got) + } +} + +func TestChatResumesATask(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, a2a.TaskStateInputRequired, a2a.TaskStateCompleted) + + if _, _, err := runChat(t, url, true, "", "one"); err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + first := agent.tasks[0] + + _, _, err := runChat(t, url, true, "two\n", "--context-id", first.ContextID, "--task-id", string(first.ID)) + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + want := chatRequest{Text: "two", TaskID: string(first.ID), ContextID: first.ContextID} + if diff := cmp.Diff(want, agent.requests[1]); diff != "" { + t.Fatalf("a2a chat resumed request wrong result (-want +got) diff = %s", diff) + } +} + +func TestChatContinuesAfterAnError(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, a2a.TaskStateCompleted, chatTurnFails, a2a.TaskStateCompleted) + + _, stderr, err := runChat(t, url, true, "two\nthree\n", "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + contextID := agent.tasks[0].ContextID + want := []chatRequest{ + {Text: "one"}, + {Text: "two", ContextID: contextID}, + {Text: "three", ContextID: contextID}, + } + if diff := cmp.Diff(want, agent.requests); diff != "" { + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) + } + if !strings.Contains(stderr, "Error: failed to send message") { + t.Fatalf("a2a chat stderr = %q, want the send error", stderr) + } +} + +func TestChatStreams(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, a2a.TaskStateInputRequired, a2a.TaskStateCompleted) + + stdout, _, err := runChat(t, url, true, "two\n", "--stream", "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + first := agent.tasks[0] + want := []chatRequest{ + {Text: "one"}, + {Text: "two", TaskID: string(first.ID), ContextID: first.ContextID}, + } + if diff := cmp.Diff(want, agent.requests); diff != "" { + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) + } + if got := strings.Count(stdout, "[status] "); got != 2 { + t.Fatalf("a2a chat printed %d status updates, want 2:\n%s", got, stdout) + } +} + +func TestChatTimeoutCancelsTheTask(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, chatTurnHangs, a2a.TaskStateCompleted) + + _, stderr, err := runChat(t, url, true, "two\n", "--stream", "--timeout", "500ms", "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + hung := agent.tasks[0] + if diff := cmp.Diff([]string{string(hung.ID)}, agent.canceled); diff != "" { + t.Fatalf("a2a chat canceled tasks wrong result (-want +got) diff = %s", diff) + } + want := []chatRequest{ + {Text: "one"}, + {Text: "two", ContextID: hung.ContextID}, + } + if diff := cmp.Diff(want, agent.requests); diff != "" { + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) + } + for _, want := range []string{"hint: increase --timeout", "Canceled task " + string(hung.ID)} { + if !strings.Contains(stderr, want) { + t.Fatalf("a2a chat stderr = %q, want it to contain %q", stderr, want) + } + } +} + +func TestChatInterrupt(t *testing.T) { + t.Parallel() + + t.Run("cancels the running turn and keeps the session open", func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, a2a.TaskStateCompleted, chatTurnInterrupted, a2a.TaskStateCompleted) + + stderr, err := runChatTo(t, agent.deps(), url, strings.NewReader("two\nthree\n"), &bytes.Buffer{}, "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + contextID := agent.tasks[0].ContextID + want := []chatRequest{ + {Text: "one"}, + {Text: "two", ContextID: contextID}, + {Text: "three", ContextID: contextID}, + } + if diff := cmp.Diff(want, agent.requests); diff != "" { + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) + } + if !strings.Contains(stderr, "Interrupted") { + t.Fatalf("a2a chat stderr = %q, want it to report the interrupted turn", stderr) + } + }) + + t.Run("cancels the task the agent is working on", func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, chatTurnHangs) + + stderr, err := runChatTo(t, agent.deps(), url, strings.NewReader(""), interruptOnPrint{agent}, "--stream", "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + id := string(agent.tasks[0].ID) + if diff := cmp.Diff([]string{id}, agent.canceled); diff != "" { + t.Fatalf("a2a chat canceled tasks wrong result (-want +got) diff = %s", diff) + } + if !strings.Contains(stderr, "Canceled task "+id) { + t.Fatalf("a2a chat stderr = %q, want it to report the canceled task", stderr) + } + }) + + t.Run("again abandons a pending cancel", func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, chatTurnHangs) + agent.mu.Lock() + agent.hangCancel = true + agent.mu.Unlock() + + stderr, err := runChatTo(t, agent.deps(), url, strings.NewReader(""), interruptOnPrint{agent}, "--stream", "--timeout", "5s", "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + + id := string(agent.tasks[0].ID) + if !strings.Contains(stderr, "Could not cancel task "+id) || !strings.Contains(stderr, "context canceled") { + t.Fatalf("a2a chat stderr = %q, want it to report the abandoned cancel", stderr) + } + }) + + t.Run("at the prompt ends the session", func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t) + interrupted := func(ctx context.Context) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(ctx) + cancel() + return ctx, cancel + } + d := deps{isTerminal: func() bool { return true }, notifyInterrupt: interrupted} + // The pipe never delivers input, so only the interrupt can end the wait at the prompt. + stdin, w := io.Pipe() + t.Cleanup(func() { + if err := w.Close(); err != nil { + t.Errorf("io.PipeWriter.Close() error = %v", err) + } + }) + + stderr, err := runChatTo(t, d, url, stdin, &bytes.Buffer{}, "--context-id", "ctx-1") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) + } + if got := len(agent.requests); got != 0 { + t.Fatalf("a2a chat sent %d messages, want 0", got) + } + if !strings.Contains(stderr, "contextId: ctx-1") { + t.Fatalf("a2a chat stderr = %q, want the context id", stderr) + } + }) +} diff --git a/internal/cli/root.go b/internal/cli/root.go index ee29bf6..33d96c7 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -21,6 +21,7 @@ import ( "io" "net/http" "os" + "os/signal" "sync" "time" @@ -39,8 +40,10 @@ const pluginsEnabledKey = "plugins-enabled" type cfgLoaderFunc func(clicfg.LoadOpts) (*clicfg.Store, error) type deps struct { - poller pollerFunc - cfgLoader cfgLoaderFunc + poller pollerFunc + cfgLoader cfgLoaderFunc + isTerminal func() bool + notifyInterrupt func(context.Context) (context.Context, context.CancelFunc) } func (d *deps) setDefaults() { @@ -50,6 +53,14 @@ func (d *deps) setDefaults() { if d.cfgLoader == nil { d.cfgLoader = clicfg.Load } + if d.isTerminal == nil { + d.isTerminal = stdioIsTerminal + } + if d.notifyInterrupt == nil { + d.notifyInterrupt = func(ctx context.Context) (context.Context, context.CancelFunc) { + return signal.NotifyContext(ctx, os.Interrupt) + } + } } type globalConfig struct { @@ -118,13 +129,13 @@ func (g *globalConfig) renderError(err error) int { ce := clierr.Classify(err) if g.IsJSON() { if perr := g.PrintJSON(ce); perr != nil { - _, _ = fmt.Fprintf(os.Stderr, "Error: %v\n", err) + _, _ = fmt.Fprintf(g.stderr(), "Error: %v\n", err) } return ce.Exit } - fmt.Fprintf(os.Stderr, "Error: %s\n", ce.Message) + _, _ = fmt.Fprintf(g.stderr(), "Error: %s\n", ce.Message) if ce.Hint != "" { - fmt.Fprintf(os.Stderr, "hint: %s\n", ce.Hint) + _, _ = fmt.Fprintf(g.stderr(), "hint: %s\n", ce.Hint) } return ce.Exit } @@ -184,6 +195,7 @@ func newRootCmd(cfg *globalConfig, deps deps) (*cobra.Command, error) { cmd.AddCommand( newCardCmd(cfg), newSendCmd(cfg, deps.poller), + newChatCmd(cfg, deps), newTaskCmd(cfg), newConfigCmd(cfg), newServeCmd(cfg), diff --git a/internal/cli/send.go b/internal/cli/send.go index 5af9e64..cd0b20d 100644 --- a/internal/cli/send.go +++ b/internal/cli/send.go @@ -72,26 +72,7 @@ func newSendCmd(cfg *globalConfig, poller pollerFunc) *cobra.Command { defer destroyClient(cfg, client) if flags.stream { - defer func() { _ = cfg.Printer.Flush() }() - ctx, debounceTimeout := utils.WithInactivityTimeout(baseCtx, cfg.timeout) - - err = handleStreaming(ctx, cfg, client, req, debounceTimeout) - if err == nil { - return nil - } - if !errors.Is(err, a2a.ErrUnsupportedOperation) { - return utils.UnpackCause(ctx, err) - } - - cfg.logf("falling back to polling (%v): %v", flags.pollInterval, err) - - for event, err := range poller(ctx, client, req, flags.pollInterval) { - debounceTimeout() - if err := handleStreamEntry(cfg, event, err); err != nil { - return utils.UnpackCause(ctx, err) - } - } - return nil + return sendStreaming(baseCtx, cfg, client, req, poller, flags.pollInterval) } ctx, cancel = context.WithTimeout(baseCtx, cfg.timeout) @@ -156,6 +137,31 @@ func buildRequest(cmd *cobra.Command, cfg *globalConfig, flags *sendFlags, args return req, nil } +// sendStreaming streams the reply to req, falling back to polling when the agent does +// not support streaming. --timeout bounds the time between events. +func sendStreaming(ctx context.Context, cfg *globalConfig, client *a2aclient.Client, req *a2a.SendMessageRequest, poller pollerFunc, pollInterval time.Duration) error { + defer func() { _ = cfg.Printer.Flush() }() + ctx, debounceTimeout := utils.WithInactivityTimeout(ctx, cfg.timeout) + + err := handleStreaming(ctx, cfg, client, req, debounceTimeout) + if err == nil { + return nil + } + if !errors.Is(err, a2a.ErrUnsupportedOperation) { + return utils.UnpackCause(ctx, err) + } + + cfg.logf("falling back to polling (%v): %v", pollInterval, err) + + for event, err := range poller(ctx, client, req, pollInterval) { + debounceTimeout() + if err := handleStreamEntry(cfg, event, err); err != nil { + return utils.UnpackCause(ctx, err) + } + } + return nil +} + func handleStreaming(ctx context.Context, cfg *globalConfig, client *a2aclient.Client, req *a2a.SendMessageRequest, debounceTimeout utils.DebounceFunc) error { if card := client.Card(); card != nil && !card.Capabilities.Streaming { return fmt.Errorf("streaming not listed in agent capabilities: %w", a2a.ErrUnsupportedOperation)