Deliver inbox answers mid-turn through the PostToolUse hook
This commit is contained in:
@@ -1,8 +1,5 @@
|
||||
# Bugs
|
||||
|
||||
- Question answers given while the agent is working get lost
|
||||
Answering an inbox question while the task's agent is mid-turn sometimes never reaches the agent. Answers are meant to be held and delivered as one message once the session is idle with no blocking question still waiting (`Deliver` in `internal/agents/inbox.go`, called when a turn ends). Find the path where a held answer is dropped. Byre restarts often while agents deploy, and a restart cuts the turn off and resumes it with the restart notice (`Recover` in `internal/agents/manager.go`), so check that path first. Add debugging output before changing code.
|
||||
|
||||
# M1 — One agent end to end
|
||||
|
||||
- Subagent views
|
||||
|
||||
+3
-1
@@ -69,7 +69,7 @@ Compaction is vendor-agnostic: `POST /api/sessions/{id}/compact` (the transcript
|
||||
|
||||
Agents spawn parallel work with the `create_task` MCP tool (`internal/agents/tasks.go`): `title`, `prompt`, `base` and optional `model` (default: the caller's session model). It creates a task in the caller's project exactly as `POST /api/projects/{id}/tasks` does, and starts a session there with `prompt` as the first message (actor `agent`), created by the caller's session creator so commits use that user's git identity. `base` is the agent's choice: `default_branch` starts from the project's default branch, `this_task` from the caller's worktree `HEAD`, which leaves out uncommitted edits from the current turn. The tool returns the new task's id at once; the new session is independent of the caller.
|
||||
|
||||
Agents don't commit; byre does. Each `user.message` event records `head`, the worktree's `HEAD` when the turn started. When a turn ends, byre runs `git add --all` and commits if anything changed (`git.CommitAll`), as the session's creator, with the first sentence of the turn's final message as the message (`commitMessage` in `internal/agents/manager.go`); `turn.ended` then carries the new sha as `commit`. A failed commit is logged and the turn still ends.
|
||||
Agents don't commit; byre does. Each `user.message` event that starts a turn records `head`, the worktree's `HEAD` when the turn started. When a turn ends, byre runs `git add --all` and commits if anything changed (`git.CommitAll`), as the session's creator, with the first sentence of the turn's final message as the message (`commitMessage` in `internal/agents/manager.go`); `turn.ended` then carries the new sha as `commit`. A failed commit is logged and the turn still ends.
|
||||
|
||||
byre chooses the Claude session ID (`vendor_session_id`) up front, so every session can be resumed. Processes start on demand: the first message after a restart resumes the session with `--resume`. Sessions are `running` from the moment a turn is sent until its `result` arrives; on startup, every session still marked `running` was cut off mid-turn, and byre resumes it and sends a message telling the agent to re-check the working tree. Stopping the server kills all agent processes. Merging or discarding a task kills its sessions and marks them `completed`.
|
||||
|
||||
@@ -139,6 +139,8 @@ Agents ask questions with the `ask_human` MCP tool. Each question carries a `con
|
||||
| `other` (free text in `text`: an answer, a question back, or new instructions) | questions, errors | `answered` |
|
||||
| `dismiss` | reports, errors | `dismissed` |
|
||||
|
||||
Answers reach the agent as one message holding every undelivered `option` and `other` response of the session (`deliveryMessage` in `internal/agents/inbox.go`), never while one of its blocking questions is still open; until then they stay held (`inbox_responses.delivered_at` empty). While a turn is running, the task session's PostToolUse hook (`afterTool` in `internal/agents/snapshots.go`, through `deliverMidTurn`) hands them over after the agent's next tool call as the hook's `additionalContext`, so the agent reads them alongside that tool's result without being interrupted; the `user.message` event recording it carries `mid_turn: true` and no `head`, and the transcript doesn't treat it as a turn start. A turn that makes no further tool call leaves them to `Deliver`, which runs after each submission and when a turn ends, and sends them as a new turn once the session is `idle`. The log traces each answer: `inbox response recorded` when it is stored, `deliver held` with the reason, `deliver sent mid-turn`, `deliver sent` with the session's state and whether its agent process had exited or byre was closing, and `undelivered responses` with a reason wherever a session has held answers when its turn ends in a handoff or merge, its agent process ends, startup resumes it, or startup finds it idle. `cancelled queued turn carried inbox responses` logs the text of a delivery that a cancelled queued turn discards. `sqlite3 byre.db "SELECT * FROM inbox_responses WHERE delivered_at IS NULL"` lists the answers held now.
|
||||
|
||||
## Notifications
|
||||
|
||||
New inbox items are sent as web push notifications to every browser a user has turned them on in. Web push needs a secure origin and a service worker; byre gets the origin from the `tailscale serve` certificate, and serves the worker at `/sw.js` (compiled from `web/src/sw.ts` with its own `web/tsconfig.sw.json`, since it needs the WebWorker library instead of the DOM one) so its scope is the whole SPA.
|
||||
|
||||
+69
-10
@@ -304,11 +304,21 @@ func (m *Manager) Deliver(ctx context.Context, sessionID string) error {
|
||||
m.deliverMu.Lock()
|
||||
defer m.deliverMu.Unlock()
|
||||
|
||||
pending, err := m.store.Undelivered(ctx, sessionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(pending) == 0 {
|
||||
return nil
|
||||
}
|
||||
ids := responseIDs(pending)
|
||||
|
||||
sess, err := m.store.GetSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if sess.State != api.AgentSessionStateIdle {
|
||||
slog.Info("deliver held: session not idle", "session", sessionID, "state", sess.State, "responses", ids)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -317,14 +327,7 @@ func (m *Manager) Deliver(ctx context.Context, sessionID string) error {
|
||||
return err
|
||||
}
|
||||
if waiting > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
pending, err := m.store.Undelivered(ctx, sessionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(pending) == 0 {
|
||||
slog.Info("deliver held: blocking question waiting", "session", sessionID, "waiting", waiting, "responses", ids)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -333,19 +336,75 @@ func (m *Manager) Deliver(ctx context.Context, sessionID string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = m.turn(ctx, sess, task, pending[0].Response.User, deliveryMessage(pending), nil)
|
||||
after, err := m.turn(ctx, sess, task, pending[0].Response.User, deliveryMessage(pending), nil)
|
||||
if errors.Is(err, ErrConflict) {
|
||||
slog.Info("deliver held: session changed state", "session", sessionID, "err", err, "responses", ids)
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
slog.Error("deliver failed", "session", sessionID, "err", err, "responses", ids)
|
||||
return err
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
l := m.live[sessionID]
|
||||
closing := m.closing
|
||||
m.mu.Unlock()
|
||||
pid := 0
|
||||
exited := false
|
||||
if l != nil {
|
||||
pid = l.sess.Pid()
|
||||
exited = l.sess.Exited()
|
||||
}
|
||||
slog.Info("deliver sent", "session", sessionID, "state", after.State, "pid", pid, "exited", exited, "closing", closing, "responses", ids)
|
||||
return m.store.MarkDelivered(ctx, ids)
|
||||
}
|
||||
|
||||
func (m *Manager) deliverMidTurn(ctx context.Context, l *live) string {
|
||||
m.deliverMu.Lock()
|
||||
defer m.deliverMu.Unlock()
|
||||
|
||||
pending, err := m.store.Undelivered(ctx, l.id)
|
||||
if err != nil {
|
||||
slog.Error("list undelivered responses", "session", l.id, "err", err)
|
||||
return ""
|
||||
}
|
||||
if len(pending) == 0 {
|
||||
return ""
|
||||
}
|
||||
ids := responseIDs(pending)
|
||||
|
||||
waiting, err := m.store.AwaitingHuman(ctx, l.id)
|
||||
if err != nil {
|
||||
slog.Error("count blocking questions", "session", l.id, "err", err)
|
||||
return ""
|
||||
}
|
||||
if waiting > 0 {
|
||||
slog.Info("deliver held mid-turn: blocking question waiting", "session", l.id, "waiting", waiting, "responses", ids)
|
||||
return ""
|
||||
}
|
||||
|
||||
err = m.store.MarkDelivered(ctx, ids)
|
||||
if err != nil {
|
||||
slog.Error("mark responses delivered", "session", l.id, "err", err, "responses", ids)
|
||||
return ""
|
||||
}
|
||||
|
||||
text := deliveryMessage(pending)
|
||||
err = m.emit(ctx, l, pending[0].Response.User, "user.message", map[string]any{"text": text, "mid_turn": true})
|
||||
if err != nil {
|
||||
slog.Error("record mid-turn delivery", "session", l.id, "err", err)
|
||||
}
|
||||
slog.Info("deliver sent mid-turn", "session", l.id, "responses", ids)
|
||||
return text
|
||||
}
|
||||
|
||||
func responseIDs(pending []store.Pending) []string {
|
||||
ids := []string{}
|
||||
for _, p := range pending {
|
||||
ids = append(ids, p.Response.Id)
|
||||
}
|
||||
return m.store.MarkDelivered(ctx, ids)
|
||||
return ids
|
||||
}
|
||||
|
||||
func deliveryMessage(pending []store.Pending) string {
|
||||
|
||||
@@ -275,6 +275,7 @@ func (m *Manager) Recover(ctx context.Context) error {
|
||||
l, err := m.ensureLive(ctx, sess, task)
|
||||
if err != nil {
|
||||
slog.Error("resume session", "session", sess.Id, "err", err)
|
||||
m.logUndelivered(ctx, sess.Id, "resume failed")
|
||||
err = m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, "byre", map[string]any{"error": err.Error()})
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -290,12 +291,33 @@ func (m *Manager) Recover(ctx context.Context) error {
|
||||
return err
|
||||
}
|
||||
slog.Info("resumed session", "session", sess.Id, "task", task.Id, "owner", sess.Owner)
|
||||
m.logUndelivered(ctx, sess.Id, "resumed with restart notice")
|
||||
}
|
||||
|
||||
idle, err := m.store.SessionsInState(ctx, api.AgentSessionStateIdle)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, sess := range idle {
|
||||
m.logUndelivered(ctx, sess.Id, "idle at startup")
|
||||
}
|
||||
|
||||
go m.deliverAllLeads()
|
||||
return m.drain(ctx)
|
||||
}
|
||||
|
||||
func (m *Manager) logUndelivered(ctx context.Context, sessionID, reason string) {
|
||||
pending, err := m.store.Undelivered(ctx, sessionID)
|
||||
if err != nil {
|
||||
slog.Error("list undelivered responses", "session", sessionID, "err", err)
|
||||
return
|
||||
}
|
||||
if len(pending) == 0 {
|
||||
return
|
||||
}
|
||||
slog.Info("undelivered responses", "session", sessionID, "reason", reason, "responses", responseIDs(pending))
|
||||
}
|
||||
|
||||
func (m *Manager) Close() {
|
||||
close(m.watchStop)
|
||||
|
||||
@@ -510,7 +532,17 @@ func (m *Manager) cancelQueued(ctx context.Context, sess api.AgentSession) (api.
|
||||
m.capMu.Lock()
|
||||
defer m.capMu.Unlock()
|
||||
|
||||
err := m.store.DeleteQueuedTurn(ctx, sess.Id)
|
||||
queued, err := m.store.QueuedTurns(ctx)
|
||||
if err != nil {
|
||||
return api.AgentSession{}, err
|
||||
}
|
||||
for _, q := range queued {
|
||||
if q.SessionID == sess.Id && strings.HasPrefix(q.Text, "Responses from your inbox:") {
|
||||
slog.Info("cancelled queued turn carried inbox responses", "session", sess.Id, "text", q.Text)
|
||||
}
|
||||
}
|
||||
|
||||
err = m.store.DeleteQueuedTurn(ctx, sess.Id)
|
||||
if errors.Is(err, store.ErrNotFound) {
|
||||
return api.AgentSession{}, fmt.Errorf("%w: session %s has no queued turn", ErrConflict, sess.Id)
|
||||
}
|
||||
@@ -607,7 +639,7 @@ func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task ap
|
||||
hooks := map[string][]claude.HookMatcher{
|
||||
"UserPromptSubmit": {{Func: remindGuidance(g)}},
|
||||
"Stop": {{Func: m.bounceProseQuestion(l)}},
|
||||
"PostToolUse": {{Func: m.snapshotAfterTool(l)}},
|
||||
"PostToolUse": {{Func: m.afterTool(l)}},
|
||||
}
|
||||
if l.owner != "" {
|
||||
l.dir = filepath.Join(m.leadRoot, url.PathEscape(l.owner))
|
||||
@@ -682,6 +714,7 @@ func (m *Manager) pump(l *live) {
|
||||
m.mu.Unlock()
|
||||
|
||||
if quiet {
|
||||
m.logUndelivered(ctx, l.id, "agent process ended while closing or stopping")
|
||||
return
|
||||
}
|
||||
|
||||
@@ -695,6 +728,7 @@ func (m *Manager) pump(l *live) {
|
||||
return
|
||||
}
|
||||
|
||||
m.logUndelivered(ctx, l.id, "agent process exited mid-turn")
|
||||
err = m.store.SetSessionState(ctx, l.id, api.AgentSessionStateIdle, agentActor, map[string]any{"error": "agent process exited mid-turn"})
|
||||
if err != nil {
|
||||
slog.Error("record session exit", "session", l.id, "err", err)
|
||||
@@ -822,11 +856,13 @@ func (m *Manager) handle(ctx context.Context, l *live, ev claude.Event) error {
|
||||
slog.Error("start queued turns", "err", err)
|
||||
}
|
||||
if handoff {
|
||||
m.logUndelivered(ctx, l.id, "turn ended in handoff")
|
||||
go m.continueFromHandoff(l.id, res.Result)
|
||||
return nil
|
||||
}
|
||||
mergeBy := m.takeMerge(l.taskID)
|
||||
if mergeBy != "" && finished {
|
||||
m.logUndelivered(ctx, l.id, "turn ended in merge")
|
||||
go m.finishMerge(l, mergeBy)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -503,6 +503,64 @@ func TestAskHumanRoundTrip(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
const shapePrompt = `Ask me with ask_human (not blocking) whether shape.txt should say circle or square, recommending circle. ` +
|
||||
"Then, without waiting for my answer, run these Bash commands one at a time, each as its own tool call: `sleep 8`, then `echo second`, then `echo third`. " +
|
||||
`After those, write shape.txt containing exactly my answer if you have received it by then, or exactly unknown if not.`
|
||||
|
||||
func TestAnswerDeliveredMidTurn(t *testing.T) {
|
||||
t.Parallel()
|
||||
e := newEnv(t)
|
||||
m := e.manager()
|
||||
|
||||
sess, err := m.Create(e.ctx, e.task, e.user, shapePrompt, testModel)
|
||||
if err != nil {
|
||||
t.Fatalf("create session: %v", err)
|
||||
}
|
||||
e.waitFor("question filed", func(ev api.Event) bool {
|
||||
return ev.Kind == "inbox.created" && ev.Payload["kind"] == string(api.Question)
|
||||
})
|
||||
|
||||
questions := e.openQuestions(sess.Id)
|
||||
if len(questions) != 1 {
|
||||
t.Fatalf("want one open question, got %+v", questions)
|
||||
}
|
||||
square := ""
|
||||
options, _ := questions[0].Payload["options"].([]any)
|
||||
for _, o := range options {
|
||||
if strings.Contains(fmt.Sprint(o), "square") {
|
||||
square = fmt.Sprint(o)
|
||||
}
|
||||
}
|
||||
e.respond(m, questions[0], api.Option, square, "", api.Answered)
|
||||
events := e.waitIdle()
|
||||
|
||||
midTurn := false
|
||||
for _, ev := range events {
|
||||
if ev.Kind == "user.message" && ev.Payload["mid_turn"] == true && strings.Contains(ev.Payload["text"].(string), "chose: "+square) {
|
||||
midTurn = true
|
||||
}
|
||||
if ev.Kind == "turn.ended" && !midTurn {
|
||||
t.Fatalf("turn ended before the answer was delivered mid-turn: %v", kinds(events))
|
||||
}
|
||||
}
|
||||
if !midTurn {
|
||||
t.Fatalf("answer was not delivered mid-turn: %v", kinds(events))
|
||||
}
|
||||
|
||||
out, err := os.ReadFile(filepath.Join(e.task.WorktreePath, "shape.txt"))
|
||||
if err != nil {
|
||||
t.Fatalf("agent didn't write shape.txt: %v", err)
|
||||
}
|
||||
if strings.TrimSpace(string(out)) != "square" {
|
||||
t.Fatalf("shape.txt contains %q, want square", out)
|
||||
}
|
||||
|
||||
pending, err := e.store.Undelivered(e.ctx, sess.Id)
|
||||
if err != nil || len(pending) != 0 {
|
||||
t.Fatalf("responses left undelivered: %+v %v", pending, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompactCarriesHandoff(t *testing.T) {
|
||||
t.Parallel()
|
||||
e := newEnv(t)
|
||||
|
||||
@@ -62,8 +62,12 @@ func (m *Manager) Respond(ctx context.Context, user string, creates []api.InboxR
|
||||
|
||||
err := m.store.Respond(ctx, user, resolutions)
|
||||
if err != nil {
|
||||
slog.Error("record inbox responses", "user", user, "err", err)
|
||||
return nil, err
|
||||
}
|
||||
for _, res := range resolutions {
|
||||
slog.Info("inbox response recorded", "item", res.Response.ItemId, "response", res.Response.Id, "type", res.Response.Type, "user", user)
|
||||
}
|
||||
|
||||
for _, id := range sessions {
|
||||
go func() {
|
||||
|
||||
@@ -129,10 +129,19 @@ func (m *Manager) dropSnapshots(repo, taskID string) error {
|
||||
return git.DeleteSnapshotRefs(repo, taskID)
|
||||
}
|
||||
|
||||
func (m *Manager) snapshotAfterTool(l *live) claude.HookFunc {
|
||||
func (m *Manager) afterTool(l *live) claude.HookFunc {
|
||||
return func(ctx context.Context, input json.RawMessage) (map[string]any, error) {
|
||||
m.snapshotSoon(l, api.SnapshotTriggerToolCall)
|
||||
return map[string]any{}, nil
|
||||
text := m.deliverMidTurn(ctx, l)
|
||||
if text == "" {
|
||||
return map[string]any{}, nil
|
||||
}
|
||||
return map[string]any{
|
||||
"hookSpecificOutput": map[string]any{
|
||||
"hookEventName": "PostToolUse",
|
||||
"additionalContext": text,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Vendored
+1
-1
@@ -409,7 +409,7 @@ export function mountChat(root, src, makeHead) {
|
||||
continue;
|
||||
}
|
||||
put(el);
|
||||
if (ev.kind === "user.message") {
|
||||
if (ev.kind === "user.message" && p["mid_turn"] !== true) {
|
||||
ctx.turnStart = el;
|
||||
}
|
||||
if (ev.kind === "turn.ended" || ev.kind === "inbox.created") {
|
||||
|
||||
Vendored
+1
-1
File diff suppressed because one or more lines are too long
+1
-1
@@ -492,7 +492,7 @@ export function mountChat(root: HTMLElement, src: ChatSource, makeHead: (rerende
|
||||
continue;
|
||||
}
|
||||
put(el);
|
||||
if (ev.kind === "user.message") {
|
||||
if (ev.kind === "user.message" && p["mid_turn"] !== true) {
|
||||
ctx.turnStart = el;
|
||||
}
|
||||
if (ev.kind === "turn.ended" || ev.kind === "inbox.created") {
|
||||
|
||||
Reference in New Issue
Block a user