diff --git a/docs/changelog.mdx b/docs/changelog.mdx index cf4216d0..efd7c303 100644 --- a/docs/changelog.mdx +++ b/docs/changelog.mdx @@ -10,6 +10,11 @@ upgrades with `berth upgrade BOX`, or by running the again. ## Unreleased +- **A resumed chat agent no longer looks busy forever.** An agent carried on + without a new prompt (after its box restarted, or its conversation was + opened again) showed as working, with a timer counting up, until someone + sent it something. It now shows as idle at its prompt. Needs the box + upgraded. - **Boxes are ready for the Shipyard phone app.** With phone access on, a box now also lets a paired phone read an agent's recent conversation (never tool output or files), follow which agents and runs change as it diff --git a/internal/box/acpdriver.go b/internal/box/acpdriver.go index 27dc1bb0..01d9f8fb 100644 --- a/internal/box/acpdriver.go +++ b/internal/box/acpdriver.go @@ -122,6 +122,7 @@ type acpConv struct { next int lastSeq int64 exited bool + atPrompt bool // ready at its prompt since its host last started initialized bool changed chan struct{} following bool @@ -252,7 +253,7 @@ func (c *acpConv) follow() { func (c *acpConv) reconcile() { c.mu.Lock() - busy, pending, exited, greeted, lastSeq, bypass := c.busy, append([]acpRequest(nil), c.pending...), c.exited, c.initialized, c.lastSeq, c.bypass + busy, pending, exited, greeted, lastSeq, bypass, atPrompt := c.busy, append([]acpRequest(nil), c.pending...), c.exited, c.initialized, c.lastSeq, c.bypass, c.atPrompt c.mu.Unlock() if exited { return @@ -277,6 +278,11 @@ func (c *acpConv) reconcile() { c.event(adapters.Started, map[string]any{"signal": "tool"}, nil) case !busy && (st.State == "running" || st.State == "waiting"): c.event(adapters.Finished, nil, nil) + case !busy && atPrompt && (st.State == "" || st.State == "exited"): + // Started again with no prompt (resumed, or after berthd restarted): + // it said it was ready before the log was read, so nothing did. It + // sits at its prompt, and would otherwise list as working. + c.event(adapters.Ready, nil, nil) } } @@ -302,13 +308,13 @@ func (c *acpConv) apply(e agenthost.Entry, live bool) { case "host": switch e.Event { case "started": - c.busy, c.pending, c.since, c.exited, c.initialized = false, nil, time.Time{}, false, false + c.busy, c.pending, c.since, c.exited, c.initialized, c.atPrompt = false, nil, time.Time{}, false, false, false c.prompts = map[string]bool{} if live { say = append(say, func() { go c.handshake() }) } case "exited": - c.exited, c.busy, c.pending = true, false, nil + c.exited, c.busy, c.pending, c.atPrompt = true, false, nil, false say = append(say, func() { c.event(adapters.Exited, nil, nil) }) } return @@ -396,6 +402,9 @@ func (c *acpConv) apply(e agenthost.Entry, live bool) { if r.SessionID != "" { c.sessionID = r.SessionID } + if m.Error == nil && c.sessionID != "" { + c.atPrompt = true + } if live && m.Error == nil && c.sessionID != "" { sid, dir := c.sessionID, c.conv say = append(say, func() { noteConversation(dir, func(x *Conversation) { x.AgentSessionID = sid }) }, func() { c.event(adapters.Ready, nil, nil) }) diff --git a/internal/box/claudedriver.go b/internal/box/claudedriver.go index e6773e10..50b209f1 100644 --- a/internal/box/claudedriver.go +++ b/internal/box/claudedriver.go @@ -221,6 +221,7 @@ type claudeConv struct { next int lastSeq int64 exited bool + atPrompt bool // ready at its prompt since its host last started changed chan struct{} following bool // caught is closed once the log as it was is read back: what comes @@ -421,10 +422,10 @@ func (c *claudeConv) apply(e agenthost.Entry, live bool) { switch e.Event { case "started": // A host started again (resumed, or back from its terminal). - c.busy, c.pending, c.since, c.exited = false, nil, time.Time{}, false + c.busy, c.pending, c.since, c.exited, c.atPrompt = false, nil, time.Time{}, false, true say = append(say, func() { c.event(adapters.Ready, nil, nil) }) case "exited": - c.exited, c.busy, c.pending = true, false, nil + c.exited, c.busy, c.pending, c.atPrompt = true, false, nil, false say = append(say, func() { c.event(adapters.Exited, nil, nil) }) } return @@ -610,7 +611,7 @@ func (c *claudeConv) drop(id string) bool { // it back: a turn the agent ended, or one it still works on or waits in. func (c *claudeConv) reconcile() { c.mu.Lock() - busy, pending, exited, bypass := c.busy, append([]claudeRequest(nil), c.pending...), c.exited, c.bypass + busy, pending, exited, bypass, atPrompt := c.busy, append([]claudeRequest(nil), c.pending...), c.exited, c.bypass, c.atPrompt c.mu.Unlock() if exited { return @@ -638,6 +639,11 @@ func (c *claudeConv) reconcile() { c.event(adapters.Started, map[string]any{"signal": "tool"}, nil) case !busy && (st.State == "running" || st.State == "waiting"): c.event(adapters.Finished, nil, nil) + case !busy && atPrompt && (st.State == "" || st.State == "exited"): + // Started again with no prompt (resumed, or after berthd restarted): + // it said it was ready before the log was read, so nothing did. It + // sits at its prompt, and would otherwise list as working. + c.event(adapters.Ready, nil, nil) } } diff --git a/internal/box/claudedriver_test.go b/internal/box/claudedriver_test.go index 367bdc85..a015d48c 100644 --- a/internal/box/claudedriver_test.go +++ b/internal/box/claudedriver_test.go @@ -981,3 +981,101 @@ func TestATurnClaudeStartsByItselfShowsAsWorking(t *testing.T) { } } } + +// A conversation carried on with no prompt (its agent started again on its +// log, or the conversation resumed in a new session, as after the box +// restarted) sits at its prompt: idle, not working. Its host said it +// started before berthd read the log, so only the catch-up can say so. +func TestAResumedConversationWithNoPromptIsIdle(t *testing.T) { + if _, err := exec.LookPath("python3"); err != nil { + t.Skip("the stand-in Claude needs python3") + } + bin := t.TempDir() + os.WriteFile(filepath.Join(bin, "claude"), []byte(fakeStreamClaude), 0o755) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("CLAUDE_CONFIG_DIR", t.TempDir()) + exe, _ := os.Executable() + states := &Turns{} + var bx *Box + c, bus := servedBox(t, func(b *Box) { + bx = b + b.Turns = states + b.ConversationsDir = conversationsDir(t) + b.AgentHost = []string{"env", "BERTH_TEST_AGENT_HOST=1", exe} + }) + states.Attach(bus) + repo := gitRepo(t) + call(t, c, "POST", "/v1/locations", "", map[string]string{"name": "acme", "path": repo}, nil) + var sess Session + if status := call(t, c, "POST", "/v1/sessions", "", map[string]any{"location": "acme", "agent": "claude", "driver": "claude", "prompt": "hello"}, &sess); status != 200 { + t.Fatalf("start: %d", status) + } + names := []string{sess.Name} + t.Cleanup(func() { + for _, n := range names { + call(t, c, "DELETE", "/v1/sessions/"+n, "", nil, nil) + } + }) + // listed is the session as the app sees it, once ok holds. + listed := func(what, name string, ok func(Session) bool) Session { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + for { + var list []Session + call(t, c, "GET", "/v1/sessions", "", nil, &list) + for _, s := range list { + if s.Name == name && ok(s) { + return s + } + } + if time.Now().After(deadline) { + st, _ := states.State(name) + t.Fatalf("%s never happened: %+v (turns %+v)", what, list, st) + } + time.Sleep(50 * time.Millisecond) + } + } + listed("the first turn's end", sess.Name, func(s Session) bool { return s.AgentState == "finished" }) + + // Its agent ends and is started again on its own log: at its prompt. + live, err := bx.Sessions.Get(t.Context(), sess.Name) + if err != nil { + t.Fatal(err) + } + if err := agenthost.Stop(t.Context(), live.sock); err != nil { + t.Fatal(err) + } + listed("the session's end", sess.Name, func(s Session) bool { return s.Exited }) + if status := call(t, c, "POST", "/v1/sessions/"+sess.Name+"/resume", "", nil, nil); status != 200 { + t.Fatalf("resume: %d", status) + } + listed("idle once resumed", sess.Name, func(s Session) bool { return !s.Exited && s.AgentState == "idle" }) + + // Its session gone, the conversation carried on in a new one: idle too. + if status := call(t, c, "DELETE", "/v1/sessions/"+sess.Name, "", nil, nil); status != 200 { + t.Fatalf("delete: %d", status) + } + var convs []Conversation + deadline := time.Now().Add(10 * time.Second) + for { + convs = nil + call(t, c, "GET", "/v1/conversations", "", nil, &convs) + if len(convs) == 1 && !convs[0].Live { + break + } + if time.Now().After(deadline) { + t.Fatalf("the conversation is still %+v", convs) + } + time.Sleep(100 * time.Millisecond) + } + var again Session + if status := call(t, c, "POST", "/v1/conversations/"+convs[0].ID+"/resume", "", nil, &again); status != 200 { + t.Fatalf("resume the conversation: %d", status) + } + names = append(names, again.Name) + listed("idle in the new session", again.Name, func(s Session) bool { return s.AgentState == "idle" }) + + // And a prompt still starts and ends a turn there. + call(t, c, "POST", "/v1/sessions/"+again.Name+"/send", "", map[string]any{"text": "again", "when": "now"}, nil) + listed("the new turn's end", again.Name, func(s Session) bool { return s.AgentState == "finished" }) +} diff --git a/internal/box/codexdriver.go b/internal/box/codexdriver.go index a4e5ccb5..a4d66225 100644 --- a/internal/box/codexdriver.go +++ b/internal/box/codexdriver.go @@ -119,6 +119,7 @@ type codexConv struct { next int lastSeq int64 exited bool + atPrompt bool // ready at its prompt since its host last started initialized bool changed chan struct{} following bool @@ -253,7 +254,7 @@ func (c *codexConv) follow() { // of a host berthd hasn't greeted yet. func (c *codexConv) reconcile() { c.mu.Lock() - busy, pending, exited, greeted, lastSeq := c.busy, len(c.pending), c.exited, c.initialized, c.lastSeq + busy, pending, exited, greeted, lastSeq, atPrompt := c.busy, len(c.pending), c.exited, c.initialized, c.lastSeq, c.atPrompt c.mu.Unlock() if exited { return @@ -272,6 +273,11 @@ func (c *codexConv) reconcile() { c.event(adapters.Started, map[string]any{"signal": "tool"}, nil) case !busy && (st.State == "running" || st.State == "waiting"): c.event(adapters.Finished, nil, nil) + case !busy && atPrompt && (st.State == "" || st.State == "exited"): + // Started again with no prompt (resumed, or after berthd restarted): + // it said it was ready before the log was read, so nothing did. It + // sits at its prompt, and would otherwise list as working. + c.event(adapters.Ready, nil, nil) } } @@ -307,12 +313,12 @@ func (c *codexConv) apply(e agenthost.Entry, live bool) { case "host": switch e.Event { case "started": - c.busy, c.pending, c.since, c.exited, c.initialized, c.turnID = false, nil, time.Time{}, false, false, "" + c.busy, c.pending, c.since, c.exited, c.initialized, c.turnID, c.atPrompt = false, nil, time.Time{}, false, false, "", false if live { say = append(say, func() { go c.handshake() }) } case "exited": - c.exited, c.busy, c.pending = true, false, nil + c.exited, c.busy, c.pending, c.atPrompt = true, false, nil, false say = append(say, func() { c.event(adapters.Exited, nil, nil) }) } return @@ -351,11 +357,18 @@ func (c *codexConv) apply(e agenthost.Entry, live bool) { ID string `json:"id"` } `json:"thread"` } - if json.Unmarshal(m.Result, &r) == nil && r.Thread.ID != "" && r.Thread.ID != c.threadID { - c.threadID = r.Thread.ID + if json.Unmarshal(m.Result, &r) == nil && r.Thread.ID != "" { + c.atPrompt = true + if r.Thread.ID != c.threadID { + c.threadID = r.Thread.ID + if live { + tid, dir := r.Thread.ID, c.conv + say = append(say, func() { noteConversation(dir, func(x *Conversation) { x.AgentSessionID = tid }) }) + } + } + // At its prompt, a new thread or the same one resumed. if live { - tid, dir := r.Thread.ID, c.conv - say = append(say, func() { noteConversation(dir, func(x *Conversation) { x.AgentSessionID = tid }) }, func() { c.event(adapters.Ready, nil, nil) }) + say = append(say, func() { c.event(adapters.Ready, nil, nil) }) } } } diff --git a/internal/box/codexdriver_test.go b/internal/box/codexdriver_test.go index 404d439e..4aa29e8d 100644 --- a/internal/box/codexdriver_test.go +++ b/internal/box/codexdriver_test.go @@ -275,6 +275,12 @@ func TestAStructuredCodexSessionIsDrivenThroughItsAppServer(t *testing.T) { if status := call(t, c, "POST", "/v1/sessions/"+name+"/resume", "", nil, nil); status != 200 { t.Fatalf("resume: %d", status) } + // At its prompt, as the app lists it: not working. + until("idle once resumed", func() bool { + var list []Session + call(t, c, "GET", "/v1/sessions", "", nil, &list) + return len(list) == 1 && !list[0].Exited && list[0].AgentState == "idle" + }) call(t, c, "POST", "/v1/sessions/"+name+"/send", "", map[string]any{"text": "again", "when": "now"}, nil) until("a reply once resumed", said("echo: again")) var controls Controls