Refuse lead tools that would restart an interrupted or errored agent
This commit is contained in:
+2
-2
File diff suppressed because one or more lines are too long
@@ -45,6 +45,10 @@ func leadActor(login string) string {
|
|||||||
return login + " (lead)"
|
return login + " (lead)"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func isLeadActor(actor string) bool {
|
||||||
|
return strings.HasSuffix(actor, " (lead)")
|
||||||
|
}
|
||||||
|
|
||||||
func humanActor(actor string) bool {
|
func humanActor(actor string) bool {
|
||||||
return actor != "byre" && actor != agentActor && actor != boardActor && !strings.HasSuffix(actor, leadActor(""))
|
return actor != "byre" && actor != agentActor && actor != boardActor && !strings.HasSuffix(actor, leadActor(""))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -111,7 +111,7 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
tool("answer_question",
|
tool("answer_question",
|
||||||
"Answer an open question on the human's behalf. Only when the answer is clear or the human told you it. "+
|
"Answer an open question on the human's behalf. Only when the answer is clear or the human told you it. "+
|
||||||
"type option: options holds the chosen choices' exact text (one for pick-one), and text may add notes."+
|
"type option: options holds the chosen choices' exact text (one for pick-one), and text may add notes."+
|
||||||
"type other: free text.",
|
"type other: free text."+stoppedRefusal,
|
||||||
`{"type": "object", "properties": {
|
`{"type": "object", "properties": {
|
||||||
"item_id": {"type": "string"},
|
"item_id": {"type": "string"},
|
||||||
"type": {"type": "string", "enum": ["option", "other"]},
|
"type": {"type": "string", "enum": ["option", "other"]},
|
||||||
@@ -124,6 +124,18 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
|
item, err := m.store.GetInboxItem(ctx, c.ItemId)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
sess, err := m.store.GetSession(ctx, item.AgentSessionId)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
err = m.checkNotStopped(ctx, sess)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
m.leadAnswered[c.ItemId] = true
|
m.leadAnswered[c.ItemId] = true
|
||||||
m.mu.Unlock()
|
m.mu.Unlock()
|
||||||
@@ -141,7 +153,8 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
"Send a message to a task's agent; it starts a turn. Starts the agent when the task has none. "+
|
"Send a message to a task's agent; it starts a turn. Starts the agent when the task has none. "+
|
||||||
"If the agent is mid-turn, the message is queued like the human's: the agent reads it with its next tool call's result, "+
|
"If the agent is mid-turn, the message is queued like the human's: the agent reads it with its next tool call's result, "+
|
||||||
"or as its next turn if it makes no further tool call, and gets it only once. "+
|
"or as its next turn if it makes no further tool call, and gets it only once. "+
|
||||||
"On a team task, the message is posted to the team's board instead, so every agent of the team gets it the same way; model is refused there.",
|
"On a team task, the message is posted to the team's board instead, so every agent of the team gets it the same way; model is refused there, "+
|
||||||
|
"and so is the post while any agent of the team is stopped."+stoppedRefusal,
|
||||||
`{"type": "object", "properties": {
|
`{"type": "object", "properties": {
|
||||||
"task_id": {"type": "string"},
|
"task_id": {"type": "string"},
|
||||||
"text": {"type": "string"},
|
"text": {"type": "string"},
|
||||||
@@ -172,6 +185,10 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
if args.Model != "" {
|
if args.Model != "" {
|
||||||
return "", fmt.Errorf("%w: model can't be given for a team task; its agents keep their own models", ErrInvalid)
|
return "", fmt.Errorf("%w: model can't be given for a team task; its agents keep their own models", ErrInvalid)
|
||||||
}
|
}
|
||||||
|
err = m.checkTeamNotStopped(ctx, task.Id)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
err = m.Post(ctx, task, leadActor(user.Login), args.Text)
|
err = m.Post(ctx, task, leadActor(user.Login), args.Text)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
@@ -359,7 +376,7 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
return "Interrupted.", nil
|
return "Interrupted.", nil
|
||||||
}),
|
}),
|
||||||
|
|
||||||
tool("compact_task", "Have a task's idle agent write a handoff and continue in a fresh session.",
|
tool("compact_task", "Have a task's idle agent write a handoff and continue in a fresh session."+stoppedRefusal,
|
||||||
`{"type": "object", "properties": {
|
`{"type": "object", "properties": {
|
||||||
"task_id": {"type": "string"},
|
"task_id": {"type": "string"},
|
||||||
"model": {"type": "string", "description": "Start the fresh session on this model, on either vendor; the current agent still writes the handoff. Defaults to the current model."}
|
"model": {"type": "string", "description": "Start the fresh session on this model, on either vendor; the current agent still writes the handoff. Defaults to the current model."}
|
||||||
@@ -376,6 +393,10 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
|
err = m.checkNotStopped(ctx, sess)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
_, err = m.Compact(ctx, sess.Id, args.Model)
|
_, err = m.Compact(ctx, sess.Id, args.Model)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
@@ -387,7 +408,7 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
"Merge a task as the human: its agent merges the default branch in, then byre squashes it into one commit, pushes it to origin and closes the task. "+
|
"Merge a task as the human: its agent merges the default branch in, then byre squashes it into one commit, pushes it to origin and closes the task. "+
|
||||||
"While another task in the project is merging, byre holds the merge and starts it when that one is done. "+
|
"While another task in the project is merging, byre holds the merge and starts it when that one is done. "+
|
||||||
"For a subtask, target parent instead asks its parent task's agent to merge it into the parent's branch, and the subtask closes once the parent has. "+
|
"For a subtask, target parent instead asks its parent task's agent to merge it into the parent's branch, and the subtask closes once the parent has. "+
|
||||||
"Refused for a task with active subtasks. Only when the human asks. You get a signal when it is merged; failures show as errors in the task's transcript, and you get a signal for each.",
|
"Refused for a task with active subtasks. Only when the human asks. You get a signal when it is merged; failures show as errors in the task's transcript, and you get a signal for each."+stoppedRefusal,
|
||||||
`{"type": "object", "properties": {
|
`{"type": "object", "properties": {
|
||||||
"task_id": {"type": "string"},
|
"task_id": {"type": "string"},
|
||||||
"target": {"type": "string", "enum": ["main", "parent"], "description": "Default main."}
|
"target": {"type": "string", "enum": ["main", "parent"], "description": "Default main."}
|
||||||
@@ -406,7 +427,17 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
}
|
}
|
||||||
switch args.Target {
|
switch args.Target {
|
||||||
case "", "main":
|
case "", "main":
|
||||||
|
err = m.checkTaskNotStopped(ctx, task.Id)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
case "parent":
|
case "parent":
|
||||||
|
if task.ParentTaskId != nil {
|
||||||
|
err = m.checkTaskNotStopped(ctx, *task.ParentTaskId)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
}
|
||||||
_, err = m.MergeIntoParent(ctx, task, user)
|
_, err = m.MergeIntoParent(ctx, task, user)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
@@ -440,13 +471,17 @@ func (m *Manager) leadTools(l *live) []mcp.Tool {
|
|||||||
|
|
||||||
tool("discard_task",
|
tool("discard_task",
|
||||||
"Discard a task: its agent gets one turn to write the task summary, then byre stops it and deletes its worktree and branch without merging. This destroys its unmerged work. Refused for a task with active subtasks. Only when the human asks. "+
|
"Discard a task: its agent gets one turn to write the task summary, then byre stops it and deletes its worktree and branch without merging. This destroys its unmerged work. Refused for a task with active subtasks. Only when the human asks. "+
|
||||||
"If the agent is mid-turn, the summary turn starts when that turn ends; no other turn runs in between.",
|
"If the agent is mid-turn, the summary turn starts when that turn ends; no other turn runs in between."+stoppedRefusal,
|
||||||
taskOnly,
|
taskOnly,
|
||||||
func(ctx context.Context, user api.User, raw json.RawMessage) (string, error) {
|
func(ctx context.Context, user api.User, raw json.RawMessage) (string, error) {
|
||||||
task, _, err := m.taskArg(ctx, raw)
|
task, _, err := m.taskArg(ctx, raw)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
|
err = m.checkTaskNotStopped(ctx, task.Id)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
_, err = m.Discard(ctx, task, user.Login)
|
_, err = m.Discard(ctx, task, user.Login)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
@@ -784,6 +819,38 @@ func (m *Manager) liveSession(ctx context.Context, taskID string) (api.AgentSess
|
|||||||
return sess, true, nil
|
return sess, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const stoppedRefusal = " Refused while the agent is stopped: the human interrupted its last turn, or that turn ended in an error, and no turn has started since. Then put it to the human in chat instead."
|
||||||
|
|
||||||
|
func (m *Manager) checkTaskNotStopped(ctx context.Context, taskID string) error {
|
||||||
|
sess, found, err := m.liveSession(ctx, taskID)
|
||||||
|
if err != nil || !found {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return m.checkNotStopped(ctx, sess)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *Manager) checkTeamNotStopped(ctx context.Context, taskID string) error {
|
||||||
|
sessions, err := m.store.ListSessions(ctx, taskID)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
for _, s := range teamSessions(sessions) {
|
||||||
|
err = m.checkNotStopped(ctx, s)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *Manager) checkNotStopped(ctx context.Context, sess api.AgentSession) error {
|
||||||
|
stopped, err := m.store.SessionStopped(ctx, sess.TaskId, sess.Id)
|
||||||
|
if err != nil || !stopped {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return fmt.Errorf("%w: the human interrupted task %s's agent (or its last turn ended in an error) and hasn't messaged it since, so byre won't start a turn on it for you; put what you wanted to send or do to the human in chat instead", ErrConflict, sess.TaskId)
|
||||||
|
}
|
||||||
|
|
||||||
func (m *Manager) taskArg(ctx context.Context, raw json.RawMessage) (api.Task, api.Project, error) {
|
func (m *Manager) taskArg(ctx context.Context, raw json.RawMessage) (api.Task, api.Project, error) {
|
||||||
args := struct {
|
args := struct {
|
||||||
TaskID string `json:"task_id"`
|
TaskID string `json:"task_id"`
|
||||||
|
|||||||
@@ -881,7 +881,7 @@ func (m *Manager) pump(l *live) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
m.endMergeTurn(l.id, true)
|
m.endMergeTurn(l.id, true)
|
||||||
next, _, err := m.store.EndTurn(ctx, l.id, agentActor, map[string]any{"error": "agent process exited mid-turn"}, nil, false, false)
|
next, _, err := m.store.EndTurn(ctx, l.id, agentActor, map[string]any{"error": "agent process exited mid-turn", "stopped": true}, nil, false, false)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
slog.Error("record session exit", "session", l.id, "err", err)
|
slog.Error("record session exit", "session", l.id, "err", err)
|
||||||
}
|
}
|
||||||
@@ -1186,8 +1186,12 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
|
|||||||
unlock()
|
unlock()
|
||||||
|
|
||||||
held := len(l.subagents) > 0 || interrupted
|
held := len(l.subagents) > 0 || interrupted
|
||||||
|
var state map[string]any
|
||||||
|
if interrupted || res.isError {
|
||||||
|
state = map[string]any{"stopped": true}
|
||||||
|
}
|
||||||
turn := m.event(l, agentActor, "turn.ended", payload)
|
turn := m.event(l, agentActor, "turn.ended", payload)
|
||||||
next, finalizeBy, err := m.store.EndTurn(ctx, l.id, agentActor, nil, &turn, review && !held, finished && !handoff)
|
next, finalizeBy, err := m.store.EndTurn(ctx, l.id, agentActor, state, &turn, review && !held, finished && !handoff)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -47,6 +47,12 @@ func (m *Manager) Message(ctx context.Context, sessionID, actor, text, model str
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return api.AgentSession{}, false, err
|
return api.AgentSession{}, false, err
|
||||||
}
|
}
|
||||||
|
if sess.State == api.AgentSessionStateIdle && isLeadActor(actor) {
|
||||||
|
err = m.checkNotStopped(ctx, sess)
|
||||||
|
if err != nil {
|
||||||
|
return api.AgentSession{}, false, err
|
||||||
|
}
|
||||||
|
}
|
||||||
held := sess.State == api.AgentSessionStateIdle && sess.Role == api.Implementer && automatedActor(actor) && m.interrupted(sess.Id)
|
held := sess.State == api.AgentSessionStateIdle && sess.Role == api.Implementer && automatedActor(actor) && m.interrupted(sess.Id)
|
||||||
if !held && (sess.State != api.AgentSessionStateRunning || sess.Role != api.Implementer) {
|
if !held && (sess.State != api.AgentSessionStateRunning || sess.Role != api.Implementer) {
|
||||||
sess, err = m.Send(ctx, sessionID, actor, text, model, images...)
|
sess, err = m.Send(ctx, sessionID, actor, text, model, images...)
|
||||||
|
|||||||
@@ -1061,6 +1061,24 @@ func (s *Store) TaskHandoffs(ctx context.Context, taskID string) ([]Handoff, err
|
|||||||
return handoffs, rows.Err()
|
return handoffs, rows.Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Store) SessionStopped(ctx context.Context, taskID, sessionID string) (bool, error) {
|
||||||
|
stopped := false
|
||||||
|
err := s.db.QueryRowContext(ctx, `
|
||||||
|
SELECT coalesce(json_extract(e.payload, '$.stopped'), 0) = 1 AND NOT EXISTS (
|
||||||
|
SELECT 1 FROM events a
|
||||||
|
WHERE a.task_id = e.task_id AND a.agent_session_id = e.agent_session_id AND a.seq > e.seq
|
||||||
|
AND a.kind = 'inbox.updated' AND json_extract(a.payload, '$.state') = 'answered'
|
||||||
|
)
|
||||||
|
FROM events e
|
||||||
|
WHERE e.task_id = ? AND e.agent_session_id = ? AND e.kind = 'session.state'
|
||||||
|
ORDER BY e.seq DESC LIMIT 1
|
||||||
|
`, taskID, sessionID).Scan(&stopped)
|
||||||
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
return stopped, err
|
||||||
|
}
|
||||||
|
|
||||||
func (s *Store) LeadEvents(ctx context.Context, owner string, after int64, limit int) ([]api.Event, error) {
|
func (s *Store) LeadEvents(ctx context.Context, owner string, after int64, limit int) ([]api.Event, error) {
|
||||||
rows, err := s.db.QueryContext(ctx, `
|
rows, err := s.db.QueryContext(ctx, `
|
||||||
SELECT `+eventColumns+`
|
SELECT `+eventColumns+`
|
||||||
|
|||||||
Reference in New Issue
Block a user