The daemon owns the root agent in ``agent/`` in the data directory and
listens on the Unix socket ``onechat.sock`` there. In the client, each
line is a message to the agent, the line ``/stop`` interrupts the
-running turn, and Ctrl-D ends the client. The client shows the replies,
-the tool calls, and the first lines of each tool result as they happen.
-It also shows each message that the user sends from another client, with
-the client as a mark, such as ``[telegram] text``. The daemon is the
-only process that uses the token file while it runs, so ``onechat
-login`` and ``onechat probe`` refuse to start until it stops.
+running turn, the line ``/stop NAME`` stops the subagent NAME, and
+Ctrl-D ends the client. The client shows the replies, the tool calls,
+and the first lines of each tool result as they happen. It also shows
+each message that the user sends from another client, with the client as
+a mark, such as ``[telegram] text``. The daemon is the only process that
+uses the token file while it runs, so ``onechat login`` and ``onechat
+probe`` refuse to start until it stops.
The agent runs commands with bash as the user in its working directory,
with no sandbox. It has the rights of the user, also on the token file.
back to the agent as one message that starts with ``[NAME]``. The agent
can send a subagent more messages: a running subagent reads them between
its tool calls, and a finished one goes on from its log, and its reply
-comes back in the same way. Each model call of a subagent has the name
-in the ``agent`` field of ``agent/usage.jsonl``.
+comes back in the same way. When the user stops a subagent, the chat
+gets a note, and the report of the subagent is logged, but it starts no
+turn. Each model call of a subagent has the name in the ``agent`` field
+of ``agent/usage.jsonl``.
``agent/config.json`` holds the settings of the agent. The first start
writes it with the defaults and a random ID. Do not change the ID: the
{"token": "123456:ABC-token-from-BotFather", "owner": 123456789}
The bot reads only the private chat of the owner and ignores all other
-updates. Each text message goes to the agent as a user message, and the
-line ``/stop`` interrupts the running turn. The bot sends the replies in
-parts of at most 4096 characters, with their Markdown as Telegram
-formats: bold, italic, strikethrough, code, code blocks, links, and
-headings as bold text. When Telegram refuses the formats of a part, the
-bot sends the part again as plain text. Each error goes as a short
-message in plain text. It shows the typing action while a turn runs. It
-does not send the tool calls. It sends each message that the user types
-in the terminal client, with the mark ``[terminal]``.
+updates. Each text message goes to the agent as a user message, the line
+``/stop`` interrupts the running turn, and the line ``/stop NAME`` stops
+the subagent NAME. The bot sends the replies in parts of at most 4096
+characters, with their Markdown as Telegram formats: bold, italic,
+strikethrough, code, code blocks, links, and headings as bold text. When
+Telegram refuses the formats of a part, the bot sends the part again as
+plain text. Each error goes as a short message in plain text. It shows
+the typing action while a turn runs. It does not send the tool calls. It
+sends each message that the user types in the terminal client, with the
+mark ``[terminal]``.
``agent/telegram-offset.json`` holds the ID of the next update. The bot
moves it past an update only after the message is durable in the log, so
}
return nil
+ case "stop":
+ found, err := l.StopSubagent(request.Text)
+ if err == nil && !found {
+ err = fmt.Errorf("no subagent %s runs", request.Text)
+ }
+
+ return err
default:
return fmt.Errorf("unknown request %q", request.Type)
}
first.Type(stopCommand)
first.WaitFor(t, "[error] no turn runs")
+ first.Type(stopCommand + " BlueFox")
+ first.WaitFor(t, "[error] no subagent BlueFox runs")
require.NotContains(t, second.Output(), "no turn runs")
require.NoError(t, first.Close())
}
}
// SocketRequest is one line from a client: "send" with the text of a user
-// message, or "interrupt".
+// message, "interrupt", or "stop" with the name of a subagent in Text.
type SocketRequest struct {
Type string `json:"type"`
Text string `json:"text,omitempty"`
package main
-// stopCommand is the input line that interrupts the running turn.
+import (
+ "strings"
+)
+
+// stopCommand is the input line that interrupts the running turn. The line
+// "/stop NAME" stops the subagent NAME.
const stopCommand = "/stop"
+
+// stopSubagentName returns NAME of a line "/stop NAME", or false for any other
+// line.
+func stopSubagentName(line string) (string, bool) {
+ name, found := strings.CutPrefix(strings.TrimSpace(line), stopCommand+" ")
+ name = strings.TrimSpace(name)
+ if !found || name == "" {
+ return "", false
+ }
+
+ return name, true
+}
--- /dev/null
+package main
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestStopNameReadsTheNameOfASubagent(t *testing.T) {
+ testCases := []struct {
+ name string
+ line string
+ found bool
+ agent string
+ }{
+ {
+ name: "a name after the command",
+ line: " /stop GreenOtter ",
+ found: true,
+ agent: "GreenOtter",
+ },
+ {
+ name: "the command alone",
+ line: "/stop",
+ },
+ {
+ name: "the command with spaces only",
+ line: "/stop ",
+ },
+ {
+ name: "a word that starts with the command",
+ line: "/stopGreenOtter",
+ },
+ {
+ name: "a message",
+ line: "please /stop GreenOtter",
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ agent, found := stopSubagentName(tc.line)
+ require.Equal(t, tc.found, found)
+ require.Equal(t, tc.agent, agent)
+ })
+ }
+}
// errSubagentClosed tells that the runs of the subagents ended with the loop.
var errSubagentClosed = errors.New("the daemon stops, so no subagent starts")
-// SubagentRuns runs the subagents of one agent in the background. Each run is
-// a goroutine with its own context, which only Close ends. A run is one or more
-// request chains. The first request of each chain holds the view of the agent
-// at the start of the run, then the state of the turn, then the whole log of
-// the subagent. A message of Tell that comes during a chain joins its next
+// SubagentRuns runs the subagents of one agent in the background. Each run is a
+// goroutine with its own context, which only Stop and Close end. A run is one
+// or more request chains. The first request of each chain holds the view of the
+// agent at the start of the run, then the state of the turn, then the whole log
+// of the subagent. A message of Tell that comes during a chain joins its next
// request after the tool results. One that comes after the final answer starts
// one more chain, which reads the log again. The report of the run goes to
-// report: the final answer of each chain, or the error of a failed call. At the
-// stop of the daemon the runs end with no report.
+// report: the final answer of each chain, or the error of a failed call. Stop
+// ends one run, and its report says that it stopped. At the stop of the daemon
+// the runs end with no report.
//
// Spawn and Tell split the view with the layout of the agent, which has no
// lock. They are tools, so they run in the goroutine of the turns, as each
client ChainClient
state func() string
report func(text string, stopped bool)
+ note func(text string) error
notify func(Event)
mutex sync.Mutex
return "Sent. " + name + "'s reply will reach you as a message.", nil
}
+// Stop ends the run of the subagent name and tells if it ran. The chat gets the
+// note "Stopped by the user: NAME." under the mutex, so the note comes before
+// the report of the run.
+func (r *SubagentRuns) Stop(name string) (bool, error) {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+
+ run, found := r.running[name]
+ if !found {
+ return false, nil
+ }
+
+ err := r.note("Stopped by the user: " + name + ".")
+ if err != nil {
+ return true, err
+ }
+
+ run.stopped = true
+ run.cancel()
+
+ return true, nil
+}
+
// Close ends every run and waits for them. No subagent starts after Close.
func (r *SubagentRuns) Close() {
r.mutex.Lock()
}
// start runs run in the background. Its context keeps the values of ctx, but
-// only Close ends it.
+// only Stop and Close end it.
func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error {
ctx, run.cancel = context.WithCancel(context.WithoutCancel(ctx))
// run runs the chains of run and reports its end, unless Close ended it. A
// chain follows while messages wait after the final answer of the chain before
-// it, and no chain failed.
+// it, and no chain failed. A chain that Stop ends gives the reply "stopped".
func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) {
defer r.workers.Done()
defer run.cancel()
var replies []string
+ stopped := false
for more := true; more; {
reply, err := r.chain(ctx, run)
- if err != nil {
+
+ r.mutex.Lock()
+ stopped = run.stopped
+ switch {
+ case err != nil && stopped:
+ reply = "stopped"
+ case err != nil:
reply = "failed: " + err.Error()
}
- replies = append(replies, reply)
-
- r.mutex.Lock()
- more = err == nil && len(run.pending) > 0
+ more = err == nil && ctx.Err() == nil && len(run.pending) > 0
run.pending = nil
if !more {
delete(r.running, run.subagent.name)
}
r.mutex.Unlock()
+
+ replies = append(replies, reply)
}
- if ctx.Err() != nil {
+ if ctx.Err() != nil && !stopped {
return
}
- r.report(subagentReport(run.subagent.name, replies), false)
+ r.report(subagentReport(run.subagent.name, replies), stopped)
}
// chain runs one request chain of run and returns its final answer.
}
// SubagentRun is one run of a subagent: the view that its requests get, its
-// effort, the end of its context, and the messages of Tell that wait for a
-// request. The mutex of SubagentRuns protects the messages.
+// effort, the end of its context, the messages of Tell that wait for a request,
+// and whether Stop ended it. The mutex of SubagentRuns protects the messages
+// and the stop.
type SubagentRun struct {
subagent *Subagent
view []string
effort string
cancel context.CancelFunc
pending []Message
+ stopped bool
}
// subagentReport returns the report of the subagent name with its replies: the
require.EqualError(t, err, "no agent BlueFox")
}
+func TestStopEndsASubagentAndItsCommand(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "shell", `{"command":"touch started && sleep 30"}`)}, 100, 0, 5))
+ }
+
+ test := newTestTurnLoop(t, server)
+ runs := test.loop.runs
+ _, err := runs.Spawn(t.Context(), []string{"Wait long."}, nil)
+ require.NoError(t, err)
+
+ name := test.events.Last("subagent")
+ subagent, err := runs.subagents.Get(t.Context(), name)
+ require.NoError(t, err)
+
+ require.Eventually(t, func() bool {
+ _, err := os.Stat(filepath.Join(subagent.WorkDir(), "started"))
+ return err == nil
+ }, 10*time.Second, time.Millisecond)
+
+ start := time.Now()
+ found, err := test.loop.StopSubagent(name)
+ require.NoError(t, err)
+ require.True(t, found)
+
+ test.events.WaitFor(t, "work", 1)
+ require.Less(t, time.Since(start), 10*time.Second)
+ require.Equal(t, []string{
+ "note: Stopped by the user: " + name + ".",
+ "work: " + subagentReport(name, []string{"stopped"}),
+ }, test.Log(t))
+
+ log, err := subagent.Show()
+ require.NoError(t, err)
+ require.Contains(t, log, "echo: \n[interrupted]")
+
+ found, err = test.loop.StopSubagent(name)
+ require.NoError(t, err)
+ require.False(t, found)
+
+ time.Sleep(50 * time.Millisecond)
+ require.Zero(t, test.events.Count("busy"))
+}
+
func TestCloseEndsTheSubagentsWithNoReport(t *testing.T) {
server := newModelFake(t)
arrived := make(chan struct{})
// handle runs one update. A text message of the owner in the private chat goes
// to the agent, and handle returns when it is durable in the log. The line
-// /stop interrupts the turn. Any other message of the owner gets the answer
-// that only text is supported. Every other update is ignored, and its content
-// is not logged.
+// /stop interrupts the turn, and the line /stop NAME stops the subagent NAME.
+// Any other message of the owner gets the answer that only text is supported.
+// Every other update is ignored, and its content is not logged.
func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, update TelegramUpdate) error {
message := update.Message
if message == nil || message.Chat.Type != "private" || message.Chat.ID != b.owner || message.From == nil || message.From.ID != b.owner {
return nil
}
+ name, stop := stopSubagentName(message.Text)
switch {
case message.Text == "":
b.notice(telegramTextOnly)
if !l.Interrupt() {
b.notice("No turn runs.")
}
+ case stop:
+ found, err := l.StopSubagent(name)
+ if err != nil {
+ return fmt.Errorf("stop subagent %s: %w", name, err)
+ }
+
+ if !found {
+ b.notice("No subagent " + name + " runs.")
+ }
default:
_, err := l.Submit(telegramSource, message.Text)
if err != nil {
})
telegramWaitForSent(t, telegram, "Error: interrupted", "No turn runs.")
+
+ telegram.Add(TelegramUpdate{
+ UpdateID: 4,
+ Message: telegramOwnerMessage(stopCommand + " BlueFox"),
+ })
+
+ telegramWaitForSent(t, telegram, "No turn runs.", "No subagent BlueFox runs.")
}
func TestBotStartsAtTheStoredOffset(t *testing.T) {
var errTerminalDaemonClosed = errors.New("the daemon closed the connection")
// terminal runs "onechat chat": the terminal client of the daemon. Each line of
-// input is a user message, and the line /stop interrupts the running turn. The
-// client shows the replies, the tool calls, and their results as they happen.
-// It ends at the end of the input or when the daemon closes the connection.
+// input is a user message, the line /stop interrupts the running turn, and the
+// line /stop NAME stops the subagent NAME. The client shows the replies, the
+// tool calls, and their results as they happen. It ends at the end of the input
+// or when the daemon closes the connection.
func terminal(ctx context.Context, in io.Reader, out io.Writer) error {
dir, err := dataDir()
if err != nil {
Text: line,
}
+ name, stop := stopSubagentName(line)
switch strings.TrimSpace(line) {
case "":
continue
request = SocketRequest{
Type: "interrupt",
}
+ default:
+ if stop {
+ request = SocketRequest{
+ Type: "stop",
+ Text: name,
+ }
+ }
}
err := encoder.Encode(request)
client: client,
state: l.state,
report: l.Report,
+ note: l.note,
notify: notify,
running: make(map[string]*SubagentRun),
}
}
}
+// StopSubagent ends the run of the subagent name and tells if it ran. Its
+// report is logged, but it starts no turn.
+func (l *TurnLoop) StopSubagent(name string) (bool, error) {
+ return l.runs.Stop(name)
+}
+
// Interrupt ends the running turn. A running tool is killed, and the turn sends
// no further request.
func (l *TurnLoop) Interrupt() bool {
}
}
+// note logs a note and tells the clients. A note starts no turn.
+func (l *TurnLoop) note(text string) error {
+ _, err := l.record("note", "", text)
+
+ return err
+}
+
// record logs a message and tells the clients, with the client source of a user
// message.
func (l *TurnLoop) record(kind, source, text string) (Message, error) {
projects / onechat / commitdiff