From 105defc9259267fb4dc949d9f3e62a754fb3da52 Mon Sep 17 00:00:00 2001 From: Connor Moss Date: Fri, 2 Oct 2026 10:58:43 -0400 Subject: [PATCH 1/8] feat: add an interactive chat command --- internal/README.md | 17 ++++ internal/cli/chat.go | 161 ++++++++++++++++++++++++++++++ internal/cli/chat_test.go | 203 ++++++++++++++++++++++++++++++++++++++ internal/cli/root.go | 9 +- 4 files changed, 388 insertions(+), 2 deletions(-) create mode 100644 internal/cli/chat.go create mode 100644 internal/cli/chat_test.go diff --git a/internal/README.md b/internal/README.md index e8d4cf0..e1c28d5 100644 --- a/internal/README.md +++ b/internal/README.md @@ -205,6 +205,23 @@ 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. An `AUTH_REQUIRED` task ends +the session with an explanation. Type `/quit` or press Ctrl-D to exit; the context +and task ids are printed so you can resume with `send`. + +```bash +a2a chat -a +a2a chat -a "Hello, what can you do?" +``` + +`chat` is interactive only. When standard input or output is not a terminal, it +sends the positional message once, like `send`, and reads no input. Without a +message it is a usage error. + ### `task get` - Get Task Details ```bash diff --git a/internal/cli/chat.go b/internal/cli/chat.go new file mode 100644 index 0000000..f0f419c --- /dev/null +++ b/internal/cli/chat.go @@ -0,0 +1,161 @@ +// 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 ( + "bufio" + "context" + "fmt" + "io" + "os" + "strings" + + "github.com/spf13/cobra" + + "github.com/a2aproject/a2a-cli/internal/clierr" + "github.com/a2aproject/a2a-go/v2/a2a" + "github.com/a2aproject/a2a-go/v2/a2aclient" +) + +const chatQuit = "/quit" + +func newChatCmd(cfg *globalConfig, isTerminal func() bool) *cobra.Command { + return &cobra.Command{ + Use: "chat [message]", + Short: "Chat with an agent, carrying the conversation across turns", + Long: "Chat with an agent interactively. Each reply continues the same conversation: the context id is carried " + + "across turns, and an input-required task is continued with your next message. Type " + chatQuit + " or press " + + "Ctrl-D to exit; the context and task ids are printed so you can resume with send.\n\n" + + "When standard input or output is not a terminal, chat sends [message] once, like send, and reads no input.", + Args: cobra.MaximumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + interactive := isTerminal() + if !interactive && len(args) == 0 { + return clierr.Usage("chat needs a terminal; pass a message to send it once, or use send") + } + + 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 := &chatSession{cfg: cfg, client: client} + if !interactive { + _, err := session.turn(baseCtx, args[0]) + return err + } + return session.run(baseCtx, cmd.InOrStdin(), args) + }, + } +} + +// chatSession carries a conversation's identifiers from one turn to the next. +type chatSession struct { + cfg *globalConfig + client *a2aclient.Client + contextID string + taskID a2a.TaskID +} + +func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) error { + defer s.printResume() + + if len(args) == 1 { + if done, err := s.turn(ctx, args[0]); err != nil || done { + return err + } + } + + lines := bufio.NewScanner(in) + for { + _, _ = fmt.Fprint(s.cfg.stderr(), "> ") + if !lines.Scan() { + _, _ = fmt.Fprintln(s.cfg.stderr()) + return lines.Err() + } + text := strings.TrimSpace(lines.Text()) + if text == "" { + continue + } + if text == chatQuit { + return nil + } + if done, err := s.turn(ctx, text); err != nil || done { + return err + } + } +} + +// turn sends one message and reports done when the conversation cannot continue. +func (s *chatSession) turn(ctx context.Context, text string) (done bool, err error) { + msg := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart(text)) + msg.ContextID = s.contextID + msg.TaskID = s.taskID + + ctx, cancel := context.WithTimeout(ctx, s.cfg.timeout) + defer cancel() + + result, err := s.client.SendMessage(ctx, &a2a.SendMessageRequest{Message: msg, Tenant: s.cfg.tenant}) + if err != nil { + return false, fmt.Errorf("failed to send message: %w", err) + } + if err := s.cfg.handleEvent(result); err != nil { + return false, err + } + + switch r := result.(type) { + case *a2a.Task: + s.contextID = r.ContextID + s.taskID = "" + switch r.Status.State { + case a2a.TaskStateInputRequired: + s.taskID = r.ID + case a2a.TaskStateAuthRequired: + s.taskID = r.ID + _, _ = fmt.Fprintln(s.cfg.stderr(), "The agent needs authentication to continue this task. "+ + "Provide credentials with --auth or --svc-param, then resume it with send.") + return true, nil + } + case *a2a.Message: + if r.ContextID != "" { + s.contextID = r.ContextID + } + s.taskID = "" + } + return false, nil +} + +func (s *chatSession) printResume() { + if s.contextID != "" { + _, _ = fmt.Fprintf(s.cfg.stderr(), "contextId: %s\n", s.contextID) + } + if s.taskID != "" { + _, _ = fmt.Fprintf(s.cfg.stderr(), "taskId: %s\n", s.taskID) + } +} + +// 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..01e6aae --- /dev/null +++ b/internal/cli/chat_test.go @@ -0,0 +1,203 @@ +// 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" + "iter" + "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" +) + +// chatRequest is the message the agent received on one turn. +type chatRequest struct { + Text string + TaskID string + ContextID string +} + +// 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 +} + +func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) { + t.Helper() + agent := &chatAgent{states: states} + server := httptest.NewServer(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, + }) + task := &a2a.Task{ + ID: ec.TaskID, + ContextID: ec.ContextID, + Status: a2a.TaskStatus{State: agent.states[turn]}, + } + agent.tasks = append(agent.tasks, task) + agent.mu.Unlock() + yield(task, nil) + } + }), + ))) + t.Cleanup(server.Close) + return agent, server.URL +} + +func runChat(t *testing.T, url string, terminal bool, stdin string, args ...string) (stderr string, err error) { + t.Helper() + var out, errBuf bytes.Buffer + cfg := &globalConfig{ + Printer: output.NewPrinter(&out, output.ModeText), + svcParams: &flagparse.ServiceParams{}, + errOut: &errBuf, + } + root := mustNewRoot(t, cfg, deps{ + poller: polling.Stream, + cfgLoader: clicfg.LoadEmpty, + isTerminal: func() bool { return terminal }, + }) + root.SetIn(strings.NewReader(stdin)) + root.SetArgs(append([]string{"chat", "-e", url, "--transport", "rest"}, args...)) + err = root.Execute() + return errBuf.String(), err +} + +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("chat error = %v, want nil", err) + } + + first := agent.tasks[0] + want := []chatRequest{ + {Text: "one"}, + // The first task asked for input, so the reply continues it. + {Text: "two", TaskID: string(first.ID), ContextID: first.ContextID}, + // That task completed, so the next turn starts a new task in the same context. + {Text: "three", ContextID: first.ContextID}, + } + if diff := cmp.Diff(want, agent.requests); diff != "" { + t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + } + if !strings.Contains(stderr, "contextId: "+first.ContextID) { + t.Fatalf("chat did not print the context id to resume with:\n%s", 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: "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("chat error = %v, want nil", err) + } + if got := len(agent.requests); got != tt.wantRequests { + t.Errorf("chat sent %d messages, want %d", got, tt.wantRequests) + } + if !strings.Contains(stderr, tt.wantStderr) { + t.Errorf("chat stderr missing %q:\n%s", tt.wantStderr, stderr) + } + }) + } +} + +func TestChatWithoutATerminal(t *testing.T) { + t.Parallel() + + t.Run("sends the message once and reads no input", func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, a2a.TaskStateCompleted) + + if _, err := runChat(t, url, false, "two\n", "one"); err != nil { + t.Fatalf("chat error = %v, want nil", err) + } + if diff := cmp.Diff([]chatRequest{{Text: "one"}}, agent.requests); diff != "" { + t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + } + }) + + t.Run("without a message is a usage error", func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t) + + _, err := runChat(t, url, false, "one\n") + if ce := clierr.Classify(err); ce == nil || ce.Code != clierr.CodeUsage { + t.Fatalf("chat error = %v, want a usage error", err) + } + if got := len(agent.requests); got != 0 { + t.Fatalf("chat sent %d messages, want 0", got) + } + }) +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 5554d83..e709994 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -38,8 +38,9 @@ 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 } func (d *deps) setDefaults() { @@ -49,6 +50,9 @@ func (d *deps) setDefaults() { if d.cfgLoader == nil { d.cfgLoader = clicfg.Load } + if d.isTerminal == nil { + d.isTerminal = stdioIsTerminal + } } type globalConfig struct { @@ -181,6 +185,7 @@ func newRootCmd(cfg *globalConfig, deps deps) (*cobra.Command, error) { cmd.AddCommand( newCardCmd(cfg), newSendCmd(cfg, deps.poller), + newChatCmd(cfg, deps.isTerminal), newTaskCmd(cfg), newConfigCmd(cfg), newServeCmd(cfg), From 2554dae15876052f0f14280ab0ace758b1d6fca0 Mon Sep 17 00:00:00 2001 From: Connor Moss Date: Tue, 20 Oct 2026 12:36:15 -0400 Subject: [PATCH 2/8] feat: resume a task in chat with --context-id and --task-id --- internal/README.md | 10 +++++++++- internal/cli/chat.go | 19 ++++++++++++++++--- internal/cli/chat_test.go | 20 ++++++++++++++++++++ 3 files changed, 45 insertions(+), 4 deletions(-) diff --git a/internal/README.md b/internal/README.md index e1c28d5..8b584c1 100644 --- a/internal/README.md +++ b/internal/README.md @@ -211,13 +211,21 @@ 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. An `AUTH_REQUIRED` task ends the session with an explanation. Type `/quit` or press Ctrl-D to exit; the context -and task ids are printed so you can resume with `send`. +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?" + +# Resume a conversation (same task) +a2a chat -a --context-id --task-id ``` +| Flag | Description | +|---|---| +| `--task-id ` | Continue an existing task with the first message. | +| `--context-id ` | Start the conversation in an existing context. | + `chat` is interactive only. When standard input or output is not a terminal, it sends the positional message once, like `send`, and reads no input. Without a message it is a usage error. diff --git a/internal/cli/chat.go b/internal/cli/chat.go index f0f419c..3b33359 100644 --- a/internal/cli/chat.go +++ b/internal/cli/chat.go @@ -31,13 +31,20 @@ import ( const chatQuit = "/quit" +type chatFlags struct { + taskID string + contextID string +} + func newChatCmd(cfg *globalConfig, isTerminal func() bool) *cobra.Command { - return &cobra.Command{ + var flags chatFlags + + cmd := &cobra.Command{ Use: "chat [message]", Short: "Chat with an agent, carrying the conversation across turns", Long: "Chat with an agent interactively. Each reply continues the same conversation: the context id is carried " + "across turns, and an input-required task is continued with your next message. Type " + chatQuit + " or press " + - "Ctrl-D to exit; the context and task ids are printed so you can resume with send.\n\n" + + "Ctrl-D to exit; the context and task ids are printed so you can resume with send, or with chat --context-id and --task-id.\n\n" + "When standard input or output is not a terminal, chat sends [message] once, like send, and reads no input.", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { @@ -55,7 +62,7 @@ func newChatCmd(cfg *globalConfig, isTerminal func() bool) *cobra.Command { } defer destroyClient(cfg, client) - session := &chatSession{cfg: cfg, client: client} + session := &chatSession{cfg: cfg, client: client, contextID: flags.contextID, taskID: a2a.TaskID(flags.taskID)} if !interactive { _, err := session.turn(baseCtx, args[0]) return err @@ -63,6 +70,12 @@ func newChatCmd(cfg *globalConfig, isTerminal func() bool) *cobra.Command { return session.run(baseCtx, cmd.InOrStdin(), args) }, } + + f := cmd.Flags() + f.StringVar(&flags.taskID, "task-id", "", "Task ID to continue an existing task") + f.StringVar(&flags.contextID, "context-id", "", "Context ID to group this turn under") + + return cmd } // chatSession carries a conversation's identifiers from one turn to the next. diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go index 01e6aae..b3d686b 100644 --- a/internal/cli/chat_test.go +++ b/internal/cli/chat_test.go @@ -201,3 +201,23 @@ func TestChatWithoutATerminal(t *testing.T) { } }) } + +func TestChatResumesATask(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, a2a.TaskStateInputRequired, a2a.TaskStateCompleted) + + if _, err := runChat(t, url, false, "", "one"); err != nil { + t.Fatalf("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("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("chat did not resume the task (-want +got):\n%s", diff) + } +} From 7d78c4cba9c670bac65029b16b7199c29ea2075f Mon Sep 17 00:00:00 2001 From: Connor Moss Date: Tue, 20 Oct 2026 12:37:08 -0400 Subject: [PATCH 3/8] fix: keep the chat session open when a turn fails --- internal/README.md | 5 +++-- internal/cli/chat.go | 20 ++++++++++++++------ internal/cli/chat_test.go | 35 ++++++++++++++++++++++++++++++++++- internal/cli/root.go | 6 +++--- 4 files changed, 54 insertions(+), 12 deletions(-) diff --git a/internal/README.md b/internal/README.md index 8b584c1..f6c1c52 100644 --- a/internal/README.md +++ b/internal/README.md @@ -209,8 +209,9 @@ a2a send -a --context-id "Related question" 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. An `AUTH_REQUIRED` task ends -the session with an explanation. Type `/quit` or press Ctrl-D to exit; the context +`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. Type `/quit` or press Ctrl-D to exit; the context and task ids are printed so you can resume with `send` or `chat`. ```bash diff --git a/internal/cli/chat.go b/internal/cli/chat.go index 3b33359..adc5228 100644 --- a/internal/cli/chat.go +++ b/internal/cli/chat.go @@ -89,10 +89,8 @@ type chatSession struct { func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) error { defer s.printResume() - if len(args) == 1 { - if done, err := s.turn(ctx, args[0]); err != nil || done { - return err - } + if len(args) == 1 && s.reply(ctx, args[0]) { + return nil } lines := bufio.NewScanner(in) @@ -109,12 +107,22 @@ func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) erro if text == chatQuit { return nil } - if done, err := s.turn(ctx, text); err != nil || done { - return err + if s.reply(ctx, text) { + return nil } } } +// reply runs one turn and reports whether the session should end. A failed turn is +// printed the way send prints its errors, and the conversation goes on. +func (s *chatSession) reply(ctx context.Context, text string) (done bool) { + done, err := s.turn(ctx, text) + if err != nil { + s.cfg.renderError(err) + } + return done +} + // turn sends one message and reports done when the conversation cannot continue. func (s *chatSession) turn(ctx context.Context, text string) (done bool, err error) { msg := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart(text)) diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go index b3d686b..00ed5ff 100644 --- a/internal/cli/chat_test.go +++ b/internal/cli/chat_test.go @@ -17,6 +17,7 @@ package cli import ( "bytes" "context" + "errors" "iter" "net/http/httptest" "strings" @@ -41,6 +42,9 @@ type chatRequest struct { ContextID string } +// chatTurnFails scripts a turn that the agent answers with an error. +const chatTurnFails = a2a.TaskStateUnspecified + // chatAgent answers each turn with a task in the next scripted state. type chatAgent struct { mu sync.Mutex @@ -62,13 +66,18 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) 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: agent.states[turn]}, + Status: a2a.TaskStatus{State: state}, } agent.tasks = append(agent.tasks, task) agent.mu.Unlock() + if state == chatTurnFails { + yield(nil, errors.New("agent failed")) + return + } yield(task, nil) } }), @@ -221,3 +230,27 @@ func TestChatResumesATask(t *testing.T) { t.Fatalf("chat did not resume the task (-want +got):\n%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("chat error = %v, want nil", err) + } + + contextID := agent.tasks[0].ContextID + want := []chatRequest{ + {Text: "one"}, + {Text: "two", ContextID: contextID}, + // The failed turn is reported and the conversation goes on in the same context. + {Text: "three", ContextID: contextID}, + } + if diff := cmp.Diff(want, agent.requests); diff != "" { + t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + } + if !strings.Contains(stderr, "Error: failed to send message") { + t.Fatalf("chat did not print the send error:\n%s", stderr) + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 9a5c01b..f581c7a 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -122,13 +122,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 } From e5288f12fff969ff232e0a10af14a27a2668192c Mon Sep 17 00:00:00 2001 From: Connor Moss Date: Tue, 20 Oct 2026 12:38:53 -0400 Subject: [PATCH 4/8] feat: stream chat replies with --stream --- internal/README.md | 4 ++ internal/cli/chat.go | 100 +++++++++++++++++++++++++++----------- internal/cli/chat_test.go | 48 +++++++++++++----- internal/cli/root.go | 2 +- internal/cli/send.go | 46 ++++++++++-------- 5 files changed, 139 insertions(+), 61 deletions(-) diff --git a/internal/README.md b/internal/README.md index f6c1c52..213008c 100644 --- a/internal/README.md +++ b/internal/README.md @@ -218,12 +218,16 @@ and task ids are printed so you can resume with `send` or `chat`. 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. | diff --git a/internal/cli/chat.go b/internal/cli/chat.go index adc5228..872ad98 100644 --- a/internal/cli/chat.go +++ b/internal/cli/chat.go @@ -21,6 +21,7 @@ import ( "io" "os" "strings" + "time" "github.com/spf13/cobra" @@ -32,11 +33,13 @@ import ( const chatQuit = "/quit" type chatFlags struct { - taskID string - contextID string + stream bool + taskID string + contextID string + pollInterval time.Duration } -func newChatCmd(cfg *globalConfig, isTerminal func() bool) *cobra.Command { +func newChatCmd(cfg *globalConfig, poller pollerFunc, isTerminal func() bool) *cobra.Command { var flags chatFlags cmd := &cobra.Command{ @@ -62,7 +65,15 @@ func newChatCmd(cfg *globalConfig, isTerminal func() bool) *cobra.Command { } defer destroyClient(cfg, client) - session := &chatSession{cfg: cfg, client: client, contextID: flags.contextID, taskID: a2a.TaskID(flags.taskID)} + session := &chatSession{ + cfg: cfg, + client: client, + flags: &flags, + poller: poller, + contextID: flags.contextID, + taskID: a2a.TaskID(flags.taskID), + } + cfg.handlers = append(cfg.handlers, session.track) if !interactive { _, err := session.turn(baseCtx, args[0]) return err @@ -72,18 +83,23 @@ func newChatCmd(cfg *globalConfig, isTerminal func() bool) *cobra.Command { } f := cmd.Flags() + f.BoolVar(&flags.stream, "stream", false, "Use streaming response. Falls back to polling and synthetic events if a server does not support streaming.") f.StringVar(&flags.taskID, "task-id", "", "Task ID to continue an existing task") f.StringVar(&flags.contextID, "context-id", "", "Context ID to group this turn under") + f.DurationVar(&flags.pollInterval, "poll-interval", 2*time.Second, "Duration between GetTask requests in polling fallback mode.") return cmd } // chatSession carries a conversation's identifiers from one turn to the next. type chatSession struct { - cfg *globalConfig - client *a2aclient.Client - contextID string - taskID a2a.TaskID + cfg *globalConfig + client *a2aclient.Client + flags *chatFlags + poller pollerFunc + contextID string + taskID a2a.TaskID + authRequired bool } func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) error { @@ -116,6 +132,8 @@ func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) erro // reply runs one turn and reports whether the session should end. A failed turn is // printed the way send prints its errors, and the conversation goes on. func (s *chatSession) reply(ctx context.Context, text string) (done bool) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() // ends a streamed turn's inactivity timer along with the turn done, err := s.turn(ctx, text) if err != nil { s.cfg.renderError(err) @@ -128,38 +146,62 @@ func (s *chatSession) turn(ctx context.Context, text string) (done bool, err err msg := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart(text)) msg.ContextID = s.contextID msg.TaskID = s.taskID + req := &a2a.SendMessageRequest{Message: msg, Tenant: s.cfg.tenant} + + if s.flags.stream { + err = sendStreaming(ctx, s.cfg, s.client, req, s.poller, s.flags.pollInterval) + } else { + err = s.send(ctx, req) + } + if err != nil { + return false, err + } + + if s.authRequired { + _, _ = fmt.Fprintln(s.cfg.stderr(), "The agent needs authentication to continue this task. "+ + "Provide credentials with --auth or --svc-param, then resume it with send.") + return true, nil + } + return false, nil +} +func (s *chatSession) send(ctx context.Context, req *a2a.SendMessageRequest) error { ctx, cancel := context.WithTimeout(ctx, s.cfg.timeout) defer cancel() - result, err := s.client.SendMessage(ctx, &a2a.SendMessageRequest{Message: msg, Tenant: s.cfg.tenant}) + result, err := s.client.SendMessage(ctx, req) if err != nil { - return false, fmt.Errorf("failed to send message: %w", err) - } - if err := s.cfg.handleEvent(result); err != nil { - return false, err + return fmt.Errorf("failed to send message: %w", err) } + return s.cfg.handleEvent(result) +} - switch r := result.(type) { +// track follows the conversation's identifiers through every event a turn prints. +func (s *chatSession) track(event a2a.Event) error { + switch e := event.(type) { case *a2a.Task: - s.contextID = r.ContextID - s.taskID = "" - switch r.Status.State { - case a2a.TaskStateInputRequired: - s.taskID = r.ID - case a2a.TaskStateAuthRequired: - s.taskID = r.ID - _, _ = fmt.Fprintln(s.cfg.stderr(), "The agent needs authentication to continue this task. "+ - "Provide credentials with --auth or --svc-param, then resume it with send.") - return true, nil - } + 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 r.ContextID != "" { - s.contextID = r.ContextID + if e.ContextID != "" { + s.contextID = e.ContextID } - s.taskID = "" + s.taskID, s.authRequired = "", false + } + return nil +} + +// follow keeps a task's id for the next turn only while the task waits on the user. +func (s *chatSession) follow(contextID string, taskID a2a.TaskID, state a2a.TaskState) { + s.contextID = contextID + s.taskID, s.authRequired = "", false + switch state { + case a2a.TaskStateInputRequired: + s.taskID = taskID + case a2a.TaskStateAuthRequired: + s.taskID, s.authRequired = taskID, true } - return false, nil } func (s *chatSession) printResume() { diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go index 00ed5ff..61e9bca 100644 --- a/internal/cli/chat_test.go +++ b/internal/cli/chat_test.go @@ -70,7 +70,7 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) task := &a2a.Task{ ID: ec.TaskID, ContextID: ec.ContextID, - Status: a2a.TaskStatus{State: state}, + Status: a2a.TaskStatus{State: a2a.TaskStateWorking}, } agent.tasks = append(agent.tasks, task) agent.mu.Unlock() @@ -78,7 +78,10 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) yield(nil, errors.New("agent failed")) return } - yield(task, nil) + // A working task followed by its final status, so a stream has more than one event. + if yield(task, nil) { + yield(a2a.NewStatusUpdateEvent(task, state, nil), nil) + } } }), ))) @@ -86,7 +89,7 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) return agent, server.URL } -func runChat(t *testing.T, url string, terminal bool, stdin string, args ...string) (stderr string, err error) { +func runChat(t *testing.T, url string, terminal bool, stdin string, args ...string) (stdout, stderr string, err error) { t.Helper() var out, errBuf bytes.Buffer cfg := &globalConfig{ @@ -102,14 +105,14 @@ func runChat(t *testing.T, url string, terminal bool, stdin string, args ...stri root.SetIn(strings.NewReader(stdin)) root.SetArgs(append([]string{"chat", "-e", url, "--transport", "rest"}, args...)) err = root.Execute() - return errBuf.String(), err + return out.String(), errBuf.String(), err } 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") + _, stderr, err := runChat(t, url, true, "two\nthree\n/quit\nnever sent\n", "one") if err != nil { t.Fatalf("chat error = %v, want nil", err) } @@ -168,7 +171,7 @@ func TestChatEnds(t *testing.T) { t.Parallel() agent, url := startChatAgent(t, tt.states...) - stderr, err := runChat(t, url, true, tt.stdin) + _, stderr, err := runChat(t, url, true, tt.stdin) if err != nil { t.Fatalf("chat error = %v, want nil", err) } @@ -189,7 +192,7 @@ func TestChatWithoutATerminal(t *testing.T) { t.Parallel() agent, url := startChatAgent(t, a2a.TaskStateCompleted) - if _, err := runChat(t, url, false, "two\n", "one"); err != nil { + if _, _, err := runChat(t, url, false, "two\n", "one"); err != nil { t.Fatalf("chat error = %v, want nil", err) } if diff := cmp.Diff([]chatRequest{{Text: "one"}}, agent.requests); diff != "" { @@ -201,7 +204,7 @@ func TestChatWithoutATerminal(t *testing.T) { t.Parallel() agent, url := startChatAgent(t) - _, err := runChat(t, url, false, "one\n") + _, _, err := runChat(t, url, false, "one\n") if ce := clierr.Classify(err); ce == nil || ce.Code != clierr.CodeUsage { t.Fatalf("chat error = %v, want a usage error", err) } @@ -215,12 +218,12 @@ func TestChatResumesATask(t *testing.T) { t.Parallel() agent, url := startChatAgent(t, a2a.TaskStateInputRequired, a2a.TaskStateCompleted) - if _, err := runChat(t, url, false, "", "one"); err != nil { + if _, _, err := runChat(t, url, false, "", "one"); err != nil { t.Fatalf("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)) + _, _, err := runChat(t, url, true, "two\n", "--context-id", first.ContextID, "--task-id", string(first.ID)) if err != nil { t.Fatalf("chat error = %v, want nil", err) } @@ -235,7 +238,7 @@ 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") + _, stderr, err := runChat(t, url, true, "two\nthree\n", "one") if err != nil { t.Fatalf("chat error = %v, want nil", err) } @@ -254,3 +257,26 @@ func TestChatContinuesAfterAnError(t *testing.T) { t.Fatalf("chat did not print the send error:\n%s", 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("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("chat requests wrong (-want +got):\n%s", diff) + } + // Each turn streams the working task and then its final status update. + if got := strings.Count(stdout, "[status] "); got != 2 { + t.Fatalf("chat printed %d status updates, want 2:\n%s", got, stdout) + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index f581c7a..76d13eb 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -188,7 +188,7 @@ func newRootCmd(cfg *globalConfig, deps deps) (*cobra.Command, error) { cmd.AddCommand( newCardCmd(cfg), newSendCmd(cfg, deps.poller), - newChatCmd(cfg, deps.isTerminal), + newChatCmd(cfg, deps.poller, deps.isTerminal), 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) From 936c233bcb52ee238185c61542d8ead417901029 Mon Sep 17 00:00:00 2001 From: Connor Moss Date: Tue, 20 Oct 2026 12:46:34 -0400 Subject: [PATCH 5/8] feat: cancel a running chat turn with Ctrl-C --- internal/README.md | 5 +- internal/cli/chat.go | 73 ++++++++++++++++++++++----- internal/cli/chat_test.go | 102 +++++++++++++++++++++++++++++++++++--- internal/cli/root.go | 19 ++++--- 4 files changed, 170 insertions(+), 29 deletions(-) diff --git a/internal/README.md b/internal/README.md index 213008c..4935c5a 100644 --- a/internal/README.md +++ b/internal/README.md @@ -211,8 +211,9 @@ 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. Type `/quit` or press Ctrl-D to exit; the context -and task ids are printed so you can resume with `send` or `chat`. +explanation. Ctrl-C cancels a running turn and returns to the prompt. Type `/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 diff --git a/internal/cli/chat.go b/internal/cli/chat.go index 872ad98..f25014b 100644 --- a/internal/cli/chat.go +++ b/internal/cli/chat.go @@ -39,7 +39,7 @@ type chatFlags struct { pollInterval time.Duration } -func newChatCmd(cfg *globalConfig, poller pollerFunc, isTerminal func() bool) *cobra.Command { +func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { var flags chatFlags cmd := &cobra.Command{ @@ -47,11 +47,12 @@ func newChatCmd(cfg *globalConfig, poller pollerFunc, isTerminal func() bool) *c Short: "Chat with an agent, carrying the conversation across turns", Long: "Chat with an agent interactively. Each reply continues the same conversation: the context id is carried " + "across turns, and an input-required task is continued with your next message. Type " + chatQuit + " or press " + - "Ctrl-D to exit; the context and task ids are printed so you can resume with send, or with chat --context-id and --task-id.\n\n" + + "Ctrl-D to exit; the context and task ids are printed so you can resume with send, or with chat --context-id and --task-id. " + + "Ctrl-C cancels a running turn, or exits at the prompt.\n\n" + "When standard input or output is not a terminal, chat sends [message] once, like send, and reads no input.", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { - interactive := isTerminal() + interactive := deps.isTerminal() if !interactive && len(args) == 0 { return clierr.Usage("chat needs a terminal; pass a message to send it once, or use send") } @@ -69,7 +70,8 @@ func newChatCmd(cfg *globalConfig, poller pollerFunc, isTerminal func() bool) *c cfg: cfg, client: client, flags: &flags, - poller: poller, + poller: deps.poller, + notify: deps.notifyInterrupt, contextID: flags.contextID, taskID: a2a.TaskID(flags.taskID), } @@ -97,6 +99,7 @@ type chatSession struct { client *a2aclient.Client flags *chatFlags poller pollerFunc + notify func(context.Context) (context.Context, context.CancelFunc) contextID string taskID a2a.TaskID authRequired bool @@ -109,14 +112,17 @@ func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) erro return nil } - lines := bufio.NewScanner(in) + stop := make(chan struct{}) + defer close(stop) + lines, scanErr := scanLines(in, stop) for { _, _ = fmt.Fprint(s.cfg.stderr(), "> ") - if !lines.Scan() { + line, ok, err := s.next(ctx, lines, scanErr) + if !ok { _, _ = fmt.Fprintln(s.cfg.stderr()) - return lines.Err() + return err } - text := strings.TrimSpace(lines.Text()) + text := strings.TrimSpace(line) if text == "" { continue } @@ -129,13 +135,54 @@ func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) erro } } -// reply runs one turn and reports whether the session should end. A failed turn is -// printed the way send prints its errors, and the conversation goes on. +// scanLines reads lines on a goroutine so an interrupt is noticed while chat waits 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) + 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 *chatSession) next(ctx context.Context, lines <-chan string, scanErr <-chan error) (string, bool, error) { + ctx, stop := s.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. Ctrl-C cancels the +// turn, and a failed turn is printed the way send prints its errors; either way the +// conversation goes on. func (s *chatSession) reply(ctx context.Context, text string) (done bool) { - ctx, cancel := context.WithCancel(ctx) - defer cancel() // ends a streamed turn's inactivity timer along with the turn + ctx, stop := s.notify(ctx) + defer stop() done, err := s.turn(ctx, text) - if err != nil { + switch { + case err != nil && ctx.Err() != nil: + _, _ = fmt.Fprintln(s.cfg.stderr(), "Interrupted.") + case err != nil: s.cfg.renderError(err) } return done diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go index 61e9bca..886b197 100644 --- a/internal/cli/chat_test.go +++ b/internal/cli/chat_test.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "errors" + "io" "iter" "net/http/httptest" "strings" @@ -42,8 +43,12 @@ type chatRequest struct { ContextID string } -// chatTurnFails scripts a turn that the agent answers with an error. -const chatTurnFails = a2a.TaskStateUnspecified +const ( + // chatTurnFails scripts a turn that the agent answers with an error. + chatTurnFails a2a.TaskState = "fail" + // chatTurnInterrupted scripts a turn that the user interrupts while the agent works. + chatTurnInterrupted a2a.TaskState = "interrupt" +) // chatAgent answers each turn with a task in the next scripted state. type chatAgent struct { @@ -51,11 +56,25 @@ type chatAgent struct { states []a2a.TaskState requests []chatRequest tasks []*a2a.Task + + // interrupt stands in for Ctrl-C: it cancels the context of the latest interrupt + // scope chat opened, which during a turn is that turn. + interrupt context.CancelFunc +} + +// notifyInterrupt replaces signal.NotifyContext so a test can interrupt chat. +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 startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) { t.Helper() agent := &chatAgent{states: states} + release := make(chan struct{}) server := httptest.NewServer(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) { @@ -78,6 +97,16 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) yield(nil, errors.New("agent failed")) return } + if state == chatTurnInterrupted { + agent.mu.Lock() + agent.interrupt() + agent.mu.Unlock() + select { + case <-ctx.Done(): + case <-release: + } + return + } // A working task followed by its final status, so a stream has more than one event. if yield(task, nil) { yield(a2a.NewStatusUpdateEvent(task, state, nil), nil) @@ -86,10 +115,16 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) }), ))) 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() + return runChatWith(t, deps{isTerminal: func() bool { return terminal }}, url, strings.NewReader(stdin), args...) +} + +func runChatWith(t *testing.T, d deps, url string, stdin io.Reader, args ...string) (stdout, stderr string, err error) { t.Helper() var out, errBuf bytes.Buffer cfg := &globalConfig{ @@ -97,12 +132,10 @@ func runChat(t *testing.T, url string, terminal bool, stdin string, args ...stri svcParams: &flagparse.ServiceParams{}, errOut: &errBuf, } - root := mustNewRoot(t, cfg, deps{ - poller: polling.Stream, - cfgLoader: clicfg.LoadEmpty, - isTerminal: func() bool { return terminal }, - }) - root.SetIn(strings.NewReader(stdin)) + 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 out.String(), errBuf.String(), err @@ -280,3 +313,56 @@ func TestChatStreams(t *testing.T) { t.Fatalf("chat printed %d status updates, want 2:\n%s", got, stdout) } } + +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) + d := deps{isTerminal: func() bool { return true }, notifyInterrupt: agent.notifyInterrupt} + + _, stderr, err := runChatWith(t, d, url, strings.NewReader("two\nthree\n"), "one") + if err != nil { + t.Fatalf("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("chat requests wrong (-want +got):\n%s", diff) + } + if !strings.Contains(stderr, "Interrupted") { + t.Fatalf("chat did not report the interrupted turn:\n%s", 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} + // Input that never arrives, so only the interrupt can end the wait at the prompt. + stdin, w := io.Pipe() + t.Cleanup(func() { _ = w.Close() }) + + _, stderr, err := runChatWith(t, d, url, stdin, "--context-id", "ctx-1") + if err != nil { + t.Fatalf("chat error = %v, want nil", err) + } + if got := len(agent.requests); got != 0 { + t.Fatalf("chat sent %d messages, want 0", got) + } + if !strings.Contains(stderr, "contextId: ctx-1") { + t.Fatalf("chat did not print the context id to resume with:\n%s", stderr) + } + }) +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 76d13eb..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,9 +40,10 @@ const pluginsEnabledKey = "plugins-enabled" type cfgLoaderFunc func(clicfg.LoadOpts) (*clicfg.Store, error) type deps struct { - poller pollerFunc - cfgLoader cfgLoaderFunc - isTerminal func() bool + poller pollerFunc + cfgLoader cfgLoaderFunc + isTerminal func() bool + notifyInterrupt func(context.Context) (context.Context, context.CancelFunc) } func (d *deps) setDefaults() { @@ -54,6 +56,11 @@ func (d *deps) setDefaults() { 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 { @@ -126,9 +133,9 @@ func (g *globalConfig) renderError(err error) int { } return ce.Exit } - fmt.Fprintf(g.stderr(), "Error: %s\n", ce.Message) + _, _ = fmt.Fprintf(g.stderr(), "Error: %s\n", ce.Message) if ce.Hint != "" { - fmt.Fprintf(g.stderr(), "hint: %s\n", ce.Hint) + _, _ = fmt.Fprintf(g.stderr(), "hint: %s\n", ce.Hint) } return ce.Exit } @@ -188,7 +195,7 @@ func newRootCmd(cfg *globalConfig, deps deps) (*cobra.Command, error) { cmd.AddCommand( newCardCmd(cfg), newSendCmd(cfg, deps.poller), - newChatCmd(cfg, deps.poller, deps.isTerminal), + newChatCmd(cfg, deps), newTaskCmd(cfg), newConfigCmd(cfg), newServeCmd(cfg), From 663dd60f5099c3d27df3fb78e29fa2085872887e Mon Sep 17 00:00:00 2001 From: Connor Moss Date: Tue, 20 Oct 2026 13:59:40 -0400 Subject: [PATCH 6/8] feat: cancel the agent's task when a chat turn is interrupted --- internal/README.md | 7 ++-- internal/cli/chat.go | 35 ++++++++++++++++--- internal/cli/chat_test.go | 72 ++++++++++++++++++++++++++++++++++----- 3 files changed, 97 insertions(+), 17 deletions(-) diff --git a/internal/README.md b/internal/README.md index 4935c5a..131b3e8 100644 --- a/internal/README.md +++ b/internal/README.md @@ -211,9 +211,10 @@ 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 and returns to the prompt. Type `/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`. +explanation. Ctrl-C cancels a running turn, asks the agent to cancel the task it +started, and returns to the prompt. Type `/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 diff --git a/internal/cli/chat.go b/internal/cli/chat.go index f25014b..3220a3e 100644 --- a/internal/cli/chat.go +++ b/internal/cli/chat.go @@ -48,7 +48,7 @@ func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { Long: "Chat with an agent interactively. Each reply continues the same conversation: the context id is carried " + "across turns, and an input-required task is continued with your next message. Type " + chatQuit + " or press " + "Ctrl-D to exit; the context and task ids are printed so you can resume with send, or with chat --context-id and --task-id. " + - "Ctrl-C cancels a running turn, or exits at the prompt.\n\n" + + "Ctrl-C cancels a running turn and asks the agent to cancel its task, or exits at the prompt.\n\n" + "When standard input or output is not a terminal, chat sends [message] once, like send, and reads no input.", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { @@ -103,6 +103,8 @@ type chatSession struct { 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 } func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) error { @@ -173,23 +175,42 @@ func (s *chatSession) next(ctx context.Context, lines <-chan string, scanErr <-c } // reply runs one turn and reports whether the session should end. Ctrl-C cancels the -// turn, and a failed turn is printed the way send prints its errors; either way the -// conversation goes on. +// turn and the task it left running, and a failed turn is printed the way send prints +// its errors; either way the conversation goes on. func (s *chatSession) reply(ctx context.Context, text string) (done bool) { ctx, stop := s.notify(ctx) defer stop() done, err := s.turn(ctx, text) switch { case err != nil && ctx.Err() != nil: - _, _ = fmt.Fprintln(s.cfg.stderr(), "Interrupted.") + s.cancelRunning(ctx) case err != nil: s.cfg.renderError(err) } return done } +// cancelRunning asks the agent to stop the task an interrupted turn left running. The +// turn's context is already canceled, so the request gets a fresh timeout. +func (s *chatSession) cancelRunning(ctx context.Context) { + id := s.running + if id == "" { + _, _ = fmt.Fprintln(s.cfg.stderr(), "Interrupted.") + return + } + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), s.cfg.timeout) + defer cancel() + if _, err := s.client.CancelTask(ctx, &a2a.CancelTaskRequest{ID: id, Tenant: s.cfg.tenant}); err != nil { + _, _ = fmt.Fprintf(s.cfg.stderr(), "Interrupted. Could not cancel task %s: %v\n", id, err) + return + } + _, _ = fmt.Fprintf(s.cfg.stderr(), "Interrupted. Canceled task %s.\n", id) + s.taskID, s.running = "", "" +} + // turn sends one message and reports done when the conversation cannot continue. func (s *chatSession) 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 @@ -234,7 +255,7 @@ func (s *chatSession) track(event a2a.Event) error { if e.ContextID != "" { s.contextID = e.ContextID } - s.taskID, s.authRequired = "", false + s.taskID, s.running, s.authRequired = "", "", false } return nil } @@ -243,6 +264,10 @@ func (s *chatSession) track(event a2a.Event) error { func (s *chatSession) 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 diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go index 886b197..8e967ef 100644 --- a/internal/cli/chat_test.go +++ b/internal/cli/chat_test.go @@ -20,6 +20,7 @@ import ( "errors" "io" "iter" + "net/http" "net/http/httptest" "strings" "sync" @@ -48,6 +49,9 @@ const ( chatTurnFails a2a.TaskState = "fail" // chatTurnInterrupted scripts a turn that the user interrupts while the agent works. chatTurnInterrupted a2a.TaskState = "interrupt" + // chatTurnInterruptedWorking scripts a turn that reports the task as working and then + // waits; the test interrupts it once chat has printed the task. + chatTurnInterruptedWorking a2a.TaskState = "interrupt-working" ) // chatAgent answers each turn with a task in the next scripted state. @@ -56,6 +60,7 @@ type chatAgent struct { states []a2a.TaskState requests []chatRequest tasks []*a2a.Task + canceled []string // interrupt stands in for Ctrl-C: it cancels the context of the latest interrupt // scope chat opened, which during a turn is that turn. @@ -75,7 +80,7 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) t.Helper() agent := &chatAgent{states: states} release := make(chan struct{}) - server := httptest.NewServer(a2asrv.NewRESTHandler(a2asrv.NewHandler( + 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() @@ -97,10 +102,15 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) yield(nil, errors.New("agent failed")) return } - if state == chatTurnInterrupted { - agent.mu.Lock() - agent.interrupt() - agent.mu.Unlock() + if state == chatTurnInterruptedWorking && !yield(task, nil) { + return + } + if state == chatTurnInterrupted || state == chatTurnInterruptedWorking { + if state == chatTurnInterrupted { + agent.mu.Lock() + agent.interrupt() + agent.mu.Unlock() + } select { case <-ctx.Done(): case <-release: @@ -113,7 +123,15 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) } } }), - ))) + )) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if id, ok := strings.CutSuffix(strings.TrimPrefix(r.URL.Path, "/tasks/"), ":cancel"); ok { + agent.mu.Lock() + agent.canceled = append(agent.canceled, id) + agent.mu.Unlock() + } + handler.ServeHTTP(w, r) + })) t.Cleanup(server.Close) t.Cleanup(func() { close(release) }) return agent, server.URL @@ -126,9 +144,16 @@ func runChat(t *testing.T, url string, terminal bool, stdin string, args ...stri func runChatWith(t *testing.T, d deps, url string, stdin io.Reader, args ...string) (stdout, stderr string, err error) { t.Helper() - var out, errBuf bytes.Buffer + var out bytes.Buffer + stderr, err = runChatTo(t, d, url, 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), + Printer: output.NewPrinter(out, output.ModeText), svcParams: &flagparse.ServiceParams{}, errOut: &errBuf, } @@ -138,7 +163,17 @@ func runChatWith(t *testing.T, d deps, url string, stdin io.Reader, args ...stri root.SetIn(stdin) root.SetArgs(append([]string{"chat", "-e", url, "--transport", "rest"}, args...)) err = root.Execute() - return out.String(), errBuf.String(), err + return errBuf.String(), err +} + +// interruptOnPrint stands in for a user who presses Ctrl-C once chat prints something. +type interruptOnPrint struct{ agent *chatAgent } + +func (w interruptOnPrint) Write(p []byte) (int, error) { + w.agent.mu.Lock() + defer w.agent.mu.Unlock() + w.agent.interrupt() + return len(p), nil } func TestChatCarriesTheConversationAcrossTurns(t *testing.T) { @@ -341,6 +376,25 @@ func TestChatInterrupt(t *testing.T) { } }) + t.Run("cancels the task the agent is working on", func(t *testing.T) { + t.Parallel() + agent, url := startChatAgent(t, chatTurnInterruptedWorking) + d := deps{isTerminal: func() bool { return true }, notifyInterrupt: agent.notifyInterrupt} + + stderr, err := runChatTo(t, d, url, strings.NewReader(""), interruptOnPrint{agent}, "--stream", "one") + if err != nil { + t.Fatalf("chat error = %v, want nil", err) + } + + want := []string{string(agent.tasks[0].ID)} + if diff := cmp.Diff(want, agent.canceled); diff != "" { + t.Fatalf("chat canceled the wrong tasks (-want +got):\n%s\n%s", diff, stderr) + } + if !strings.Contains(stderr, "Canceled task "+string(agent.tasks[0].ID)) { + t.Fatalf("chat did not report the canceled task:\n%s", stderr) + } + }) + t.Run("at the prompt ends the session", func(t *testing.T) { t.Parallel() agent, url := startChatAgent(t) From 01671d1e97c75e2e2b192880310173ee71a50285 Mon Sep 17 00:00:00 2001 From: Yaroslav Shevchuk Date: Wed, 28 Oct 2026 14:56:19 +0000 Subject: [PATCH 7/8] refactor: move the chat session into internal/chat and cancel timed-out turns - A turn that exceeds --timeout now cancels the task it left running, instead of leaving it working while the next message starts a new task. - The cancel request after an interrupt gets its own interrupt scope, so a second Ctrl-C abandons it rather than waiting up to --timeout. - Without a terminal, chat prints the context and task ids after the single send so the conversation can be resumed. - The session logic lives in internal/chat; the cli package only wires the client into it. Input lines may be up to 1 MiB. - Tests follow the repository's conventions. --- internal/README.md | 7 +- internal/chat/session.go | 243 ++++++++++++++++++++++++++++++++++++++ internal/cli/chat.go | 240 ++++++------------------------------- internal/cli/chat_test.go | 229 ++++++++++++++++++++++------------- 4 files changed, 426 insertions(+), 293 deletions(-) create mode 100644 internal/chat/session.go diff --git a/internal/README.md b/internal/README.md index 131b3e8..58ac146 100644 --- a/internal/README.md +++ b/internal/README.md @@ -212,7 +212,8 @@ 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. Type `/quit`, press Ctrl-D, or press Ctrl-C at +started, and returns to the prompt; a second Ctrl-C abandons that cancel request. +A turn that exceeds `--timeout` also cancels its task. Type `/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`. @@ -234,8 +235,8 @@ a2a chat -a --context-id --task-id | `--context-id ` | Start the conversation in an existing context. | `chat` is interactive only. When standard input or output is not a terminal, it -sends the positional message once, like `send`, and reads no input. Without a -message it is a usage error. +sends the positional message once, prints the context and task ids, and reads no +input. Without a message it is a usage error. ### `task get` - Get Task Details diff --git a/internal/chat/session.go b/internal/chat/session.go new file mode 100644 index 0000000..cfad6ca --- /dev/null +++ b/internal/chat/session.go @@ -0,0 +1,243 @@ +// 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" +) + +// Quit is the input that ends a session. +const Quit = "/quit" + +// 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} +} + +// Once sends a single message and prints the identifiers to resume the conversation with. +func (s *Session) Once(ctx context.Context, text string) error { + defer s.printResume() + _, err := s.turn(ctx, text) + return err +} + +// Run reads messages from in until Quit, 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 { + 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 index 3220a3e..07e84a3 100644 --- a/internal/cli/chat.go +++ b/internal/cli/chat.go @@ -15,23 +15,19 @@ package cli import ( - "bufio" "context" "fmt" - "io" "os" - "strings" "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" ) -const chatQuit = "/quit" - type chatFlags struct { stream bool taskID string @@ -46,10 +42,12 @@ func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { Use: "chat [message]", Short: "Chat with an agent, carrying the conversation across turns", Long: "Chat with an agent interactively. Each reply continues the same conversation: the context id is carried " + - "across turns, and an input-required task is continued with your next message. Type " + chatQuit + " or press " + + "across turns, and an input-required task is continued with your next message. Type " + chat.Quit + " or press " + "Ctrl-D to exit; the context and task ids are printed so you can resume with send, or with chat --context-id and --task-id. " + - "Ctrl-C cancels a running turn and asks the agent to cancel its task, or exits at the prompt.\n\n" + - "When standard input or output is not a terminal, chat sends [message] once, like send, and reads no input.", + "Ctrl-C cancels a running turn and asks the agent to cancel its task, or exits at the prompt. " + + "A turn that exceeds --timeout also cancels its task.\n\n" + + "When standard input or output is not a terminal, chat sends [message] once, prints the context and task ids, " + + "and reads no input.", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { interactive := deps.isTerminal() @@ -66,21 +64,30 @@ func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { } defer destroyClient(cfg, client) - session := &chatSession{ - cfg: cfg, - client: client, - flags: &flags, - poller: deps.poller, - notify: deps.notifyInterrupt, - contextID: flags.contextID, - taskID: a2a.TaskID(flags.taskID), - } - cfg.handlers = append(cfg.handlers, session.track) + 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) + if !interactive { - _, err := session.turn(baseCtx, args[0]) - return err + return session.Once(baseCtx, args[0]) } - return session.run(baseCtx, cmd.InOrStdin(), args) + var first string + if len(args) == 1 { + first = args[0] + } + return session.Run(baseCtx, cmd.InOrStdin(), first) }, } @@ -93,196 +100,19 @@ func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { return cmd } -// chatSession carries a conversation's identifiers from one turn to the next. -type chatSession struct { - cfg *globalConfig - client *a2aclient.Client - flags *chatFlags - poller pollerFunc - notify func(context.Context) (context.Context, context.CancelFunc) - 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 -} - -func (s *chatSession) run(ctx context.Context, in io.Reader, args []string) error { - defer s.printResume() - - if len(args) == 1 && s.reply(ctx, args[0]) { - return nil - } - - stop := make(chan struct{}) - defer close(stop) - lines, scanErr := scanLines(in, stop) - for { - _, _ = fmt.Fprint(s.cfg.stderr(), "> ") - line, ok, err := s.next(ctx, lines, scanErr) - if !ok { - _, _ = fmt.Fprintln(s.cfg.stderr()) - return err - } - text := strings.TrimSpace(line) - if text == "" { - continue - } - if text == chatQuit { - return nil - } - if s.reply(ctx, text) { - return nil - } +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) } -} -// scanLines reads lines on a goroutine so an interrupt is noticed while chat waits 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) - 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 *chatSession) next(ctx context.Context, lines <-chan string, scanErr <-chan error) (string, bool, error) { - ctx, stop := s.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. Ctrl-C cancels the -// turn and the task it left running, and a failed turn is printed the way send prints -// its errors; either way the conversation goes on. -func (s *chatSession) reply(ctx context.Context, text string) (done bool) { - ctx, stop := s.notify(ctx) - defer stop() - done, err := s.turn(ctx, text) - switch { - case err != nil && ctx.Err() != nil: - s.cancelRunning(ctx) - case err != nil: - s.cfg.renderError(err) - } - return done -} - -// cancelRunning asks the agent to stop the task an interrupted turn left running. The -// turn's context is already canceled, so the request gets a fresh timeout. -func (s *chatSession) cancelRunning(ctx context.Context) { - id := s.running - if id == "" { - _, _ = fmt.Fprintln(s.cfg.stderr(), "Interrupted.") - return - } - ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), s.cfg.timeout) + ctx, cancel := context.WithTimeout(ctx, cfg.timeout) defer cancel() - if _, err := s.client.CancelTask(ctx, &a2a.CancelTaskRequest{ID: id, Tenant: s.cfg.tenant}); err != nil { - _, _ = fmt.Fprintf(s.cfg.stderr(), "Interrupted. Could not cancel task %s: %v\n", id, err) - return - } - _, _ = fmt.Fprintf(s.cfg.stderr(), "Interrupted. Canceled task %s.\n", id) - s.taskID, s.running = "", "" -} - -// turn sends one message and reports done when the conversation cannot continue. -func (s *chatSession) 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 - req := &a2a.SendMessageRequest{Message: msg, Tenant: s.cfg.tenant} - - if s.flags.stream { - err = sendStreaming(ctx, s.cfg, s.client, req, s.poller, s.flags.pollInterval) - } else { - err = s.send(ctx, req) - } - if err != nil { - return false, err - } - - if s.authRequired { - _, _ = fmt.Fprintln(s.cfg.stderr(), "The agent needs authentication to continue this task. "+ - "Provide credentials with --auth or --svc-param, then resume it with send.") - return true, nil - } - return false, nil -} - -func (s *chatSession) send(ctx context.Context, req *a2a.SendMessageRequest) error { - ctx, cancel := context.WithTimeout(ctx, s.cfg.timeout) - defer cancel() - - result, err := s.client.SendMessage(ctx, req) + result, err := client.SendMessage(ctx, req) if err != nil { return fmt.Errorf("failed to send message: %w", err) } - return s.cfg.handleEvent(result) -} - -// track follows the conversation's identifiers through every event a turn prints. -func (s *chatSession) 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 -} - -// follow keeps a task's id for the next turn only while the task waits on the user. -func (s *chatSession) 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 *chatSession) printResume() { - if s.contextID != "" { - _, _ = fmt.Fprintf(s.cfg.stderr(), "contextId: %s\n", s.contextID) - } - if s.taskID != "" { - _, _ = fmt.Fprintf(s.cfg.stderr(), "taskId: %s\n", s.taskID) - } + return cfg.handleEvent(result) } // stdioIsTerminal reports whether both standard input and output are terminals. diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go index 8e967ef..52895d5 100644 --- a/internal/cli/chat_test.go +++ b/internal/cli/chat_test.go @@ -37,21 +37,19 @@ import ( "github.com/a2aproject/a2a-go/v2/a2asrv" ) -// chatRequest is the message the agent received on one turn. type chatRequest struct { Text string TaskID string ContextID string } +// Scripted turns that are not task states. const ( - // chatTurnFails scripts a turn that the agent answers with an error. chatTurnFails a2a.TaskState = "fail" - // chatTurnInterrupted scripts a turn that the user interrupts while the agent works. + // chatTurnInterrupted interrupts the turn before the agent names a task. chatTurnInterrupted a2a.TaskState = "interrupt" - // chatTurnInterruptedWorking scripts a turn that reports the task as working and then - // waits; the test interrupts it once chat has printed the task. - chatTurnInterruptedWorking a2a.TaskState = "interrupt-working" + // 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. @@ -62,12 +60,13 @@ type chatAgent struct { tasks []*a2a.Task canceled []string - // interrupt stands in for Ctrl-C: it cancels the context of the latest interrupt - // scope chat opened, which during a turn is that turn. + // 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 } -// notifyInterrupt replaces signal.NotifyContext so a test can interrupt chat. func (a *chatAgent) notifyInterrupt(ctx context.Context) (context.Context, context.CancelFunc) { ctx, cancel := context.WithCancel(ctx) a.mu.Lock() @@ -76,6 +75,16 @@ func (a *chatAgent) notifyInterrupt(ctx context.Context) (context.Context, conte 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} @@ -98,37 +107,44 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) } agent.tasks = append(agent.tasks, task) agent.mu.Unlock() - if state == chatTurnFails { + + switch state { + case chatTurnFails: yield(nil, errors.New("agent failed")) return - } - if state == chatTurnInterruptedWorking && !yield(task, nil) { - return - } - if state == chatTurnInterrupted || state == chatTurnInterruptedWorking { - if state == chatTurnInterrupted { - agent.mu.Lock() - agent.interrupt() - agent.mu.Unlock() + case chatTurnInterrupted: + agent.pressCtrlC() + case chatTurnHangs: + if !yield(task, nil) { + return } - select { - case <-ctx.Done(): - case <-release: + default: + if yield(task, nil) { + yield(a2a.NewStatusUpdateEvent(task, state, nil), nil) } return } - // A working task followed by its final status, so a stream has more than one event. - if yield(task, nil) { - yield(a2a.NewStatusUpdateEvent(task, state, nil), nil) + select { + case <-ctx.Done(): + case <-release: } } }), )) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if id, ok := strings.CutSuffix(strings.TrimPrefix(r.URL.Path, "/tasks/"), ":cancel"); ok { - agent.mu.Lock() - agent.canceled = append(agent.canceled, id) - agent.mu.Unlock() + 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) })) @@ -138,14 +154,9 @@ func startChatAgent(t *testing.T, states ...a2a.TaskState) (*chatAgent, string) } func runChat(t *testing.T, url string, terminal bool, stdin string, args ...string) (stdout, stderr string, err error) { - t.Helper() - return runChatWith(t, deps{isTerminal: func() bool { return terminal }}, url, strings.NewReader(stdin), args...) -} - -func runChatWith(t *testing.T, d deps, url string, stdin io.Reader, args ...string) (stdout, stderr string, err error) { t.Helper() var out bytes.Buffer - stderr, err = runChatTo(t, d, url, stdin, &out, args...) + stderr, err = runChatTo(t, deps{isTerminal: func() bool { return terminal }}, url, strings.NewReader(stdin), &out, args...) return out.String(), stderr, err } @@ -166,13 +177,11 @@ func runChatTo(t *testing.T, d deps, url string, stdin io.Reader, out io.Writer, return errBuf.String(), err } -// interruptOnPrint stands in for a user who presses Ctrl-C once chat prints something. +// interruptOnPrint presses Ctrl-C whenever chat prints to stdout. type interruptOnPrint struct{ agent *chatAgent } func (w interruptOnPrint) Write(p []byte) (int, error) { - w.agent.mu.Lock() - defer w.agent.mu.Unlock() - w.agent.interrupt() + w.agent.pressCtrlC() return len(p), nil } @@ -182,22 +191,20 @@ func TestChatCarriesTheConversationAcrossTurns(t *testing.T) { _, stderr, err := runChat(t, url, true, "two\nthree\n/quit\nnever sent\n", "one") if err != nil { - t.Fatalf("chat error = %v, want nil", err) + t.Fatalf("a2a chat error = %v, want nil", err) } first := agent.tasks[0] want := []chatRequest{ {Text: "one"}, - // The first task asked for input, so the reply continues it. {Text: "two", TaskID: string(first.ID), ContextID: first.ContextID}, - // That task completed, so the next turn starts a new task in the same context. {Text: "three", ContextID: first.ContextID}, } if diff := cmp.Diff(want, agent.requests); diff != "" { - t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) } if !strings.Contains(stderr, "contextId: "+first.ContextID) { - t.Fatalf("chat did not print the context id to resume with:\n%s", stderr) + t.Fatalf("a2a chat stderr = %q, want the context id", stderr) } } @@ -241,13 +248,13 @@ func TestChatEnds(t *testing.T) { _, stderr, err := runChat(t, url, true, tt.stdin) if err != nil { - t.Fatalf("chat error = %v, want nil", err) + t.Fatalf("a2a chat error = %v, want nil", err) } if got := len(agent.requests); got != tt.wantRequests { - t.Errorf("chat sent %d messages, want %d", got, tt.wantRequests) + t.Fatalf("a2a chat sent %d messages, want %d", got, tt.wantRequests) } if !strings.Contains(stderr, tt.wantStderr) { - t.Errorf("chat stderr missing %q:\n%s", tt.wantStderr, stderr) + t.Fatalf("a2a chat stderr = %q, want it to contain %q", stderr, tt.wantStderr) } }) } @@ -256,15 +263,22 @@ func TestChatEnds(t *testing.T) { func TestChatWithoutATerminal(t *testing.T) { t.Parallel() - t.Run("sends the message once and reads no input", func(t *testing.T) { + t.Run("sends the message once and prints the ids", func(t *testing.T) { t.Parallel() - agent, url := startChatAgent(t, a2a.TaskStateCompleted) + agent, url := startChatAgent(t, a2a.TaskStateInputRequired) - if _, _, err := runChat(t, url, false, "two\n", "one"); err != nil { - t.Fatalf("chat error = %v, want nil", err) + _, stderr, err := runChat(t, url, false, "two\n", "one") + if err != nil { + t.Fatalf("a2a chat error = %v, want nil", err) } if diff := cmp.Diff([]chatRequest{{Text: "one"}}, agent.requests); diff != "" { - t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) + } + task := agent.tasks[0] + for _, want := range []string{"contextId: " + task.ContextID, "taskId: " + string(task.ID)} { + if !strings.Contains(stderr, want) { + t.Fatalf("a2a chat stderr = %q, want it to contain %q", stderr, want) + } } }) @@ -274,10 +288,10 @@ func TestChatWithoutATerminal(t *testing.T) { _, _, err := runChat(t, url, false, "one\n") if ce := clierr.Classify(err); ce == nil || ce.Code != clierr.CodeUsage { - t.Fatalf("chat error = %v, want a usage error", err) + t.Fatalf("a2a chat error = %v, want a usage error", err) } if got := len(agent.requests); got != 0 { - t.Fatalf("chat sent %d messages, want 0", got) + t.Fatalf("a2a chat sent %d messages, want 0", got) } }) } @@ -287,18 +301,18 @@ func TestChatResumesATask(t *testing.T) { agent, url := startChatAgent(t, a2a.TaskStateInputRequired, a2a.TaskStateCompleted) if _, _, err := runChat(t, url, false, "", "one"); err != nil { - t.Fatalf("chat error = %v, want nil", err) + 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("chat error = %v, want nil", err) + 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("chat did not resume the task (-want +got):\n%s", diff) + t.Fatalf("a2a chat resumed request wrong result (-want +got) diff = %s", diff) } } @@ -308,21 +322,20 @@ func TestChatContinuesAfterAnError(t *testing.T) { _, stderr, err := runChat(t, url, true, "two\nthree\n", "one") if err != nil { - t.Fatalf("chat error = %v, want nil", err) + t.Fatalf("a2a chat error = %v, want nil", err) } contextID := agent.tasks[0].ContextID want := []chatRequest{ {Text: "one"}, {Text: "two", ContextID: contextID}, - // The failed turn is reported and the conversation goes on in the same context. {Text: "three", ContextID: contextID}, } if diff := cmp.Diff(want, agent.requests); diff != "" { - t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) } if !strings.Contains(stderr, "Error: failed to send message") { - t.Fatalf("chat did not print the send error:\n%s", stderr) + t.Fatalf("a2a chat stderr = %q, want the send error", stderr) } } @@ -332,7 +345,7 @@ func TestChatStreams(t *testing.T) { stdout, _, err := runChat(t, url, true, "two\n", "--stream", "one") if err != nil { - t.Fatalf("chat error = %v, want nil", err) + t.Fatalf("a2a chat error = %v, want nil", err) } first := agent.tasks[0] @@ -341,11 +354,37 @@ func TestChatStreams(t *testing.T) { {Text: "two", TaskID: string(first.ID), ContextID: first.ContextID}, } if diff := cmp.Diff(want, agent.requests); diff != "" { - t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) } - // Each turn streams the working task and then its final status update. if got := strings.Count(stdout, "[status] "); got != 2 { - t.Fatalf("chat printed %d status updates, want 2:\n%s", got, stdout) + 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) + } } } @@ -355,11 +394,10 @@ func TestChatInterrupt(t *testing.T) { 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) - d := deps{isTerminal: func() bool { return true }, notifyInterrupt: agent.notifyInterrupt} - _, stderr, err := runChatWith(t, d, url, strings.NewReader("two\nthree\n"), "one") + stderr, err := runChatTo(t, agent.deps(), url, strings.NewReader("two\nthree\n"), &bytes.Buffer{}, "one") if err != nil { - t.Fatalf("chat error = %v, want nil", err) + t.Fatalf("a2a chat error = %v, want nil", err) } contextID := agent.tasks[0].ContextID @@ -369,29 +407,46 @@ func TestChatInterrupt(t *testing.T) { {Text: "three", ContextID: contextID}, } if diff := cmp.Diff(want, agent.requests); diff != "" { - t.Fatalf("chat requests wrong (-want +got):\n%s", diff) + t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) } if !strings.Contains(stderr, "Interrupted") { - t.Fatalf("chat did not report the interrupted turn:\n%s", stderr) + 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, chatTurnInterruptedWorking) - d := deps{isTerminal: func() bool { return true }, notifyInterrupt: agent.notifyInterrupt} + agent, url := startChatAgent(t, chatTurnHangs) - stderr, err := runChatTo(t, d, url, strings.NewReader(""), interruptOnPrint{agent}, "--stream", "one") + stderr, err := runChatTo(t, agent.deps(), url, strings.NewReader(""), interruptOnPrint{agent}, "--stream", "one") if err != nil { - t.Fatalf("chat error = %v, want nil", err) + t.Fatalf("a2a chat error = %v, want nil", err) } - want := []string{string(agent.tasks[0].ID)} - if diff := cmp.Diff(want, agent.canceled); diff != "" { - t.Fatalf("chat canceled the wrong tasks (-want +got):\n%s\n%s", diff, stderr) + 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 "+string(agent.tasks[0].ID)) { - t.Fatalf("chat did not report the canceled task:\n%s", stderr) + 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) } }) @@ -404,19 +459,23 @@ func TestChatInterrupt(t *testing.T) { return ctx, cancel } d := deps{isTerminal: func() bool { return true }, notifyInterrupt: interrupted} - // Input that never arrives, so only the interrupt can end the wait at the prompt. + // The pipe never delivers input, so only the interrupt can end the wait at the prompt. stdin, w := io.Pipe() - t.Cleanup(func() { _ = w.Close() }) + t.Cleanup(func() { + if err := w.Close(); err != nil { + t.Errorf("io.PipeWriter.Close() error = %v", err) + } + }) - _, stderr, err := runChatWith(t, d, url, stdin, "--context-id", "ctx-1") + stderr, err := runChatTo(t, d, url, stdin, &bytes.Buffer{}, "--context-id", "ctx-1") if err != nil { - t.Fatalf("chat error = %v, want nil", err) + t.Fatalf("a2a chat error = %v, want nil", err) } if got := len(agent.requests); got != 0 { - t.Fatalf("chat sent %d messages, want 0", got) + t.Fatalf("a2a chat sent %d messages, want 0", got) } if !strings.Contains(stderr, "contextId: ctx-1") { - t.Fatalf("chat did not print the context id to resume with:\n%s", stderr) + t.Fatalf("a2a chat stderr = %q, want the context id", stderr) } }) } From 4a567a5905d1318bd31262ee020b052396e4a4ef Mon Sep 17 00:00:00 2001 From: Yaroslav Shevchuk Date: Wed, 28 Oct 2026 15:23:00 +0000 Subject: [PATCH 8/8] feat: require a terminal for chat, accept /exit, and reword its help Without a terminal chat was a send with fewer options, so it is now a usage error that points to send. Help and flag text follow the review suggestions, and /exit ends a session like /quit. --- internal/README.md | 9 ++++--- internal/chat/session.go | 18 ++++++-------- internal/cli/chat.go | 30 ++++++++++------------- internal/cli/chat_test.go | 50 +++++++++++++-------------------------- 4 files changed, 41 insertions(+), 66 deletions(-) diff --git a/internal/README.md b/internal/README.md index 58ac146..07738f0 100644 --- a/internal/README.md +++ b/internal/README.md @@ -213,8 +213,8 @@ same conversation: the context id is carried across turns, and a task waiting on 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 `/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 +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 @@ -234,9 +234,8 @@ a2a chat -a --context-id --task-id | `--task-id ` | Continue an existing task with the first message. | | `--context-id ` | Start the conversation in an existing context. | -`chat` is interactive only. When standard input or output is not a terminal, it -sends the positional message once, prints the context and task ids, and reads no -input. Without a message it is a usage error. +`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 diff --git a/internal/chat/session.go b/internal/chat/session.go index cfad6ca..57d1c36 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -26,8 +26,11 @@ import ( "github.com/a2aproject/a2a-go/v2/a2a" ) -// Quit is the input that ends a session. -const Quit = "/quit" +// 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 @@ -63,14 +66,7 @@ func NewSession(cfg Config, contextID string, taskID a2a.TaskID) *Session { return &Session{cfg: cfg, contextID: contextID, taskID: taskID} } -// Once sends a single message and prints the identifiers to resume the conversation with. -func (s *Session) Once(ctx context.Context, text string) error { - defer s.printResume() - _, err := s.turn(ctx, text) - return err -} - -// Run reads messages from in until Quit, the end of input, an interrupt at the prompt, +// 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() @@ -93,7 +89,7 @@ func (s *Session) Run(ctx context.Context, in io.Reader, first string) error { if text == "" { continue } - if text == Quit { + if text == Quit || text == Exit { return nil } if s.reply(ctx, text) { diff --git a/internal/cli/chat.go b/internal/cli/chat.go index 07e84a3..5ea372e 100644 --- a/internal/cli/chat.go +++ b/internal/cli/chat.go @@ -40,19 +40,18 @@ func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { cmd := &cobra.Command{ Use: "chat [message]", - Short: "Chat with an agent, carrying the conversation across turns", - Long: "Chat with an agent interactively. Each reply continues the same conversation: the context id is carried " + - "across turns, and an input-required task is continued with your next message. Type " + chat.Quit + " or press " + - "Ctrl-D to exit; the context and task ids are printed so you can resume with send, or with chat --context-id and --task-id. " + - "Ctrl-C cancels a running turn and asks the agent to cancel its task, or exits at the prompt. " + - "A turn that exceeds --timeout also cancels its task.\n\n" + - "When standard input or output is not a terminal, chat sends [message] once, prints the context and task ids, " + - "and reads no input.", + 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 { - interactive := deps.isTerminal() - if !interactive && len(args) == 0 { - return clierr.Usage("chat needs a terminal; pass a message to send it once, or use send") + if !deps.isTerminal() { + return clierr.Usage("chat needs a terminal; use send to send a single message") } baseCtx := withServiceParams(cmd.Context(), cfg) @@ -80,9 +79,6 @@ func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { }, flags.contextID, a2a.TaskID(flags.taskID)) cfg.handlers = append(cfg.handlers, session.Track) - if !interactive { - return session.Once(baseCtx, args[0]) - } var first string if len(args) == 1 { first = args[0] @@ -92,10 +88,10 @@ func newChatCmd(cfg *globalConfig, deps deps) *cobra.Command { } f := cmd.Flags() - f.BoolVar(&flags.stream, "stream", false, "Use streaming response. Falls back to polling and synthetic events if a server does not support streaming.") + 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 group this turn under") - f.DurationVar(&flags.pollInterval, "poll-interval", 2*time.Second, "Duration between GetTask requests in polling fallback mode.") + 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 } diff --git a/internal/cli/chat_test.go b/internal/cli/chat_test.go index 52895d5..823c900 100644 --- a/internal/cli/chat_test.go +++ b/internal/cli/chat_test.go @@ -232,6 +232,13 @@ func TestChatEnds(t *testing.T) { 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}, @@ -260,47 +267,24 @@ func TestChatEnds(t *testing.T) { } } -func TestChatWithoutATerminal(t *testing.T) { +func TestChatWithoutATerminalIsAUsageError(t *testing.T) { t.Parallel() + agent, url := startChatAgent(t) - t.Run("sends the message once and prints the ids", func(t *testing.T) { - t.Parallel() - agent, url := startChatAgent(t, a2a.TaskStateInputRequired) - - _, stderr, err := runChat(t, url, false, "two\n", "one") - if err != nil { - t.Fatalf("a2a chat error = %v, want nil", err) - } - if diff := cmp.Diff([]chatRequest{{Text: "one"}}, agent.requests); diff != "" { - t.Fatalf("a2a chat requests wrong result (-want +got) diff = %s", diff) - } - task := agent.tasks[0] - for _, want := range []string{"contextId: " + task.ContextID, "taskId: " + string(task.ID)} { - if !strings.Contains(stderr, want) { - t.Fatalf("a2a chat stderr = %q, want it to contain %q", stderr, want) - } - } - }) - - t.Run("without a message is a usage error", func(t *testing.T) { - t.Parallel() - agent, url := startChatAgent(t) - - _, _, err := runChat(t, url, false, "one\n") - 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) - } - }) + _, _, 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, false, "", "one"); err != nil { + if _, _, err := runChat(t, url, true, "", "one"); err != nil { t.Fatalf("a2a chat error = %v, want nil", err) } first := agent.tasks[0]