From fd5f355f1a070101841946d67e142f439d644fcf Mon Sep 17 00:00:00 2001 From: enowdev Date: Wed, 4 Nov 2026 20:20:24 +0700 Subject: [PATCH] Deliver queued GOA tasks to sessions on other hosts and sessions already done Only a change of a session on GOA's own host gave a queued task. The app's word for another host's session (goa.worker) moved the board and stopped there, and the stall watch skipped any session that said it was done, so a task queued for another host's session never went in. Neither did one for a session whose turn ended before the task was queued, or one brought back by a restart with no state. Tasks now go one at a time, the next only once the one before stops running. What can't go in fails on the board and tells GOA, which a failed typing didn't do, and a task queued on a session later hibernated brings it back. Co-Authored-By: Claude Opus 5.5 --- daemon/goa.go | 21 ++- daemon/goa_queue_test.go | 279 +++++++++++++++++++++++++++++++++++++++ daemon/goa_stall.go | 87 ++++++++++-- daemon/goa_stall_test.go | 6 + daemon/goa_tasks.go | 59 +++++++-- daemon/rpc_goa.go | 5 +- docs/goa.md | 2 +- 7 files changed, 429 insertions(+), 30 deletions(-) create mode 100644 daemon/goa_queue_test.go diff --git a/daemon/goa.go b/daemon/goa.go index bd848400..6f9c9fe7 100644 --- a/daemon/goa.go +++ b/daemon/goa.go @@ -25,6 +25,8 @@ import ( // mu guards the rest. type goaState struct { board *goa.Board + // delivering holds a queued task's delivery, from its claim until it is typed. + delivering sync.Mutex mu sync.Mutex tell []goaLine // lines for GOA until it takes them, kept in tellFile @@ -197,18 +199,23 @@ func (s *Server) goaSeen(info proto.SessionInfo) { } if s.goa.board.Has(info.ID) { s.goaRanOn(info) - go func() { - s.goaReportSeen(info) - s.goaWorkerChanged(info) - if info.State == proto.StateDone || info.Exited && !info.Hibernated { - s.goaDeliver(info) - } - }() + go s.goaWorkerSeen(info) return } go s.goaToldChanged(info) } +// goaWorkerSeen follows a change of worker info, heard from this host's sessions or +// from the app for another host's: the board, then its next queued task if it can +// take one. +func (s *Server) goaWorkerSeen(info proto.SessionInfo) { + s.goaReportSeen(info) + s.goaWorkerChanged(info) + if info.State == proto.StateDone || info.Exited && !info.Hibernated { + s.goaDeliver(info) + } +} + // goaRanOn records the model a task's session says it runs on, each change, on // the tasks not settled. func (s *Server) goaRanOn(info proto.SessionInfo) { diff --git a/daemon/goa_queue_test.go b/daemon/goa_queue_test.go new file mode 100644 index 00000000..dc505197 --- /dev/null +++ b/daemon/goa_queue_test.go @@ -0,0 +1,279 @@ +package daemon + +import ( + "encoding/json" + "os" + "path/filepath" + "runtime" + "slices" + "strings" + "testing" + "time" + + "github.com/enowx/enowx/client" + "github.com/enowx/enowx/core/session" + "github.com/enowx/enowx/proto" +) + +// queueServerAgent is a stand-in agent that says what it hears, each line once. +func queueServerAgent(t *testing.T) { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("uses sh") + } + bin := t.TempDir() + script := "#!/bin/sh\necho '❯ '\nwhile read -r line; do echo \"heard: $line\"; done\n" + if err := os.WriteFile(filepath.Join(bin, "claude"), []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) +} + +func heardTimes(sess *session.Session, what string) int { + return strings.Count(sess.Terminal().Text(), "heard: "+what) +} + +func queuedTask(t *testing.T, c *client.Client, p proto.GOADelegateParams) proto.GOATask { + t.Helper() + p.NoRoute = true + var task proto.GOATask + if err := c.Call(proto.MethodGOADelegate, p, &task); err != nil { + t.Fatal(err) + } + if task.Status != proto.GOATaskQueued { + t.Fatalf("not queued: %+v", task) + } + return task +} + +func taskStatus(t *testing.T, srv *Server, id string) proto.GOATask { + t.Helper() + tasks := srv.goa.board.List() + i := slices.IndexFunc(tasks, func(x proto.GOATask) bool { return x.ID == id }) + if i < 0 { + t.Fatalf("no task %s", id) + } + return tasks[i] +} + +// busySession starts a claude session in project, at work. +func busySession(t *testing.T, srv *Server, c *client.Client, project string) *session.Session { + t.Helper() + var info proto.SessionInfo + if err := c.Call(proto.MethodSessionCreate, proto.SessionCreateParams{Agent: "claude", Project: project, BranchMode: proto.BranchNone, Cols: 80, Rows: 24}, &info); err != nil { + t.Fatal(err) + } + sess, err := srv.sessions.Get(info.ID) + if err != nil { + t.Fatal(err) + } + if err := srv.sessions.SetState(sess, proto.StateWorking); err != nil { + t.Fatal(err) + } + return sess +} + +// GOA on host A, its worker on host B: the app telling A of B's turn ending is the +// only word A gets, and has to give it its queued task, once, the next one waiting +// for its turn. This is what never reached a session on another host. +func TestGOAQueuedTaskReachesAnotherHost(t *testing.T) { + queueServerAgent(t) + a, b := startServer(t), startServer(t) + standInApp(t, a, map[string]*client.Client{b.daemonID: connect(t, b, nil), a.daemonID: connect(t, a, nil)}, func(proto.HostCallParams) bool { return true }) + ca, cb := connect(t, a, nil), connect(t, b, nil) + project := t.TempDir() + sess := busySession(t, b, cb, project) + p := proto.GOADelegateParams{Host: b.daemonID, Session: sess.ID(), Title: "One"} + p.Prompt = "first thing" + first := queuedTask(t, ca, p) + p.Prompt, p.Title = "second thing", "Two" + second := queuedTask(t, ca, p) + + done := func() { + t.Helper() + if err := b.sessions.SetState(sess, proto.StateDone); err != nil { + t.Fatal(err) + } + if err := ca.Call(proto.MethodGOAWorker, proto.GOAWorkerParams{Host: b.daemonID, Session: sess.Info()}, nil); err != nil { + t.Fatal(err) + } + } + done() + waitFor(t, "the first went in", func() bool { return heardTimes(sess, "first thing") == 1 }) + if got := taskStatus(t, a, first.ID).Status; got != proto.GOATaskWorking { + t.Fatalf("the first: %s", got) + } + // Told again of the turn before, as polls and other changes do: nothing more goes in. + done() + time.Sleep(500 * time.Millisecond) + if heardTimes(sess, "first thing") != 1 || heardTimes(sess, "second thing") != 0 { + t.Fatalf("delivered on top of a turn: %q", sess.Terminal().Text()) + } + time.Sleep(goaTurnLeast) // what is told from here on is the first's own turn ending + done() + waitFor(t, "the second went in", func() bool { return heardTimes(sess, "second thing") == 1 }) + if heardTimes(sess, "first thing") != 1 || taskStatus(t, a, second.ID).Status != proto.GOATaskWorking { + t.Fatalf("%q, %+v", sess.Terminal().Text(), taskStatus(t, a, second.ID)) + } +} + +// Many changes of a session still done at once give it one task, not all of them. +func TestGOAQueuedTasksGoOneAtATime(t *testing.T) { + queueServerAgent(t) + srv := startServer(t) + c := connect(t, srv, nil) + sess := busySession(t, srv, c, t.TempDir()) + for _, prompt := range []string{"one", "two", "three"} { + queuedTask(t, c, proto.GOADelegateParams{Session: sess.ID(), Prompt: prompt}) + } + if err := srv.sessions.SetState(sess, proto.StateDone); err != nil { + t.Fatal(err) + } + info := sess.Info() + for range 8 { + go srv.goaDeliver(info) + } + waitFor(t, "the first went in", func() bool { return heardTimes(sess, "one") == 1 }) + time.Sleep(500 * time.Millisecond) + if heardTimes(sess, "two") != 0 || heardTimes(sess, "three") != 0 { + t.Fatalf("more than one went in: %q", sess.Terminal().Text()) + } +} + +// A session done before its task was queued hears no turn's end again; the watch +// gives it the task once the session has been quiet, once. +func TestGOAQueuedTaskReachesASessionAlreadyDone(t *testing.T) { + queueServerAgent(t) + srv := startServer(t) + c := connect(t, srv, nil) + sess := busySession(t, srv, c, t.TempDir()) + if err := srv.sessions.SetState(sess, proto.StateDone); err != nil { + t.Fatal(err) + } + time.Sleep(200 * time.Millisecond) // its turn's end has been heard, with nothing queued + task, err := srv.goa.board.Add(proto.GOATask{Session: sess.ID(), Prompt: "after the turn", Status: proto.GOATaskQueued}) + if err != nil { + t.Fatal(err) + } + queuedPassUntil(t, srv, func() bool { return heardTimes(sess, "after the turn") == 1 }) + if got := taskStatus(t, srv, task.ID).Status; got != proto.GOATaskWorking { + t.Fatalf("the task: %s", got) + } +} + +// queuedPassUntil runs the stall watch's passes, its queued tasks' screens long +// quiet, until ok. +func queuedPassUntil(t *testing.T, srv *Server, ok func() bool) { + t.Helper() + w := &stallWatch{follows: map[string]*stallFollow{}, sampled: map[string]bool{}} + deadline := time.Now().Add(20 * time.Second) + for !ok() { + if time.Now().After(deadline) { + t.Fatal("never") + } + srv.stallPass(w) + for _, f := range w.follows { + f.since = time.Now().Add(-2 * queuedQuiet) + } + time.Sleep(minSample) + } +} + +// A queued task survives a daemon's restart, whole, and goes in once to the session +// brought back, which has no state to say its turn ended. +func TestGOAQueuedTaskSurvivesARestart(t *testing.T) { + queueServerAgent(t) + t.Setenv("SHELL", "/bin/sh") + dir := t.TempDir() + srv, err := NewServer(dir) + if err != nil { + t.Fatal(err) + } + c := connect(t, srv, nil) + project := t.TempDir() + sess := busySession(t, srv, c, project) + id := sess.ID() + prompt := "all of this " + strings.Repeat("word ", 40) + "end" + task := queuedTask(t, c, proto.GOADelegateParams{Session: id, Prompt: prompt}) + srv.Close() + + again, err := NewServer(dir) + if err != nil { + t.Fatal(err) + } + t.Cleanup(again.Close) + kept := taskStatus(t, again, task.ID) + if kept.Status != proto.GOATaskQueued || !strings.HasPrefix(kept.Prompt, prompt) { + t.Fatalf("after the restart: %+v", kept) + } + back, err := again.sessions.Get(id) + if err != nil { + t.Fatal(err) + } + queuedPassUntil(t, again, func() bool { + return strings.Contains(strings.Join(strings.Fields(back.Terminal().Text()), ""), "heard:allofthis") + }) + if got := taskStatus(t, again, task.ID).Status; got != proto.GOATaskWorking { + t.Fatalf("the task: %s", got) + } + // Later passes and turn's-end words don't give it again. + w := &stallWatch{follows: map[string]*stallFollow{}, sampled: map[string]bool{}} + again.stallPass(w) + again.stallPass(w) + again.goaDeliver(back.Info()) + time.Sleep(200 * time.Millisecond) + if n := strings.Count(strings.Join(strings.Fields(back.Terminal().Text()), ""), "heard:allofthis"); n != 1 { + t.Fatalf("went in %d times: %q", n, back.Terminal().Text()) + } +} + +// What can't go in says so, on the board and to GOA, rather than stay queued. +func TestGOAQueuedTaskThatCantGoIn(t *testing.T) { + queueServerAgent(t) + srv := startServer(t) + c := connect(t, srv, nil) + g, err := c.CreateSession(proto.SessionCreateParams{Agent: "claude", GOA: true, Cols: 100, Rows: 30}) + if err != nil { + t.Fatal(err) + } + // Told: waiting for it, or typed in already, as its stand-in's prompt is free. + told := func(what string) bool { + goa, err := srv.sessions.Get(g.ID) + if err != nil { + t.Fatal(err) + } + if strings.Contains(strings.Join(strings.Fields(goa.Terminal().Text()), ""), strings.Join(strings.Fields(what), "")) { + return true + } + return slices.ContainsFunc(srv.goaWaitingLines(), func(l string) bool { return strings.Contains(l, what) }) + } + + // Its session is closed: told as it ends, or found gone by the watch. + gone := busySession(t, srv, c, t.TempDir()) + first := queuedTask(t, c, proto.GOADelegateParams{From: g.ID, Session: gone.ID(), Prompt: "lost"}) + if err := srv.deleteSession(gone.ID()); err != nil { + t.Fatal(err) + } + queuedPassUntil(t, srv, func() bool { return taskStatus(t, srv, first.ID).Status == proto.GOATaskFailed }) + if !strings.Contains(taskStatus(t, srv, first.ID).Result, "its session") { + t.Fatalf("%+v", taskStatus(t, srv, first.ID)) + } + waitFor(t, "GOA told of the first", func() bool { return told("Task " + first.ID) }) + + // Typing it in fails: its host's app no longer answers, say. + busy := busySession(t, srv, c, t.TempDir()) + second := queuedTask(t, c, proto.GOADelegateParams{From: g.ID, Session: busy.ID(), Prompt: "undelivered"}) + _, _ = srv.goa.board.Update(second.ID, func(t *proto.GOATask) { t.Host = "no-such-host" }) + if err := srv.sessions.SetState(busy, proto.StateDone); err != nil { + t.Fatal(err) + } + waitFor(t, "failed", func() bool { return taskStatus(t, srv, second.ID).Status == proto.GOATaskFailed }) + waitFor(t, "GOA told of the second", func() bool { return told("Task " + second.ID) }) + if heardTimes(busy, "undelivered") != 0 { + t.Fatal("it went in") + } + raw, _ := json.Marshal(taskStatus(t, srv, second.ID)) + if !strings.Contains(string(raw), "couldn't be delivered") { + t.Fatalf("%s", raw) + } +} diff --git a/daemon/goa_stall.go b/daemon/goa_stall.go index 71131a2d..61c1e935 100644 --- a/daemon/goa_stall.go +++ b/daemon/goa_stall.go @@ -109,14 +109,28 @@ func stallCase(l stallLook, lim proto.GOAStall) (string, time.Duration) { // before a task queued for it goes in although its turn was never seen to end. const queuedQuiet = 20 * time.Second -// queuedCanGo says a task queued for a session can go in now, its turn's end -// unheard: an interrupt ends a turn with no hook, and a command typed at the -// prompt (/model) says nothing either, so its state stays the working it last -// was, gone stale. Only such a session counts: one with no state may be a plain -// shell, in a command that reads what it is typed. +// queuedCanGo says a task queued for a session can go in now, though no turn's end +// gave it: its session says it is done, as one does whose turn ended before the task +// was queued; or its turn's end went unheard, as an interrupt's, or a command typed at +// the prompt (/model), which leave the working it last was, gone stale, or as after a +// daemon's restart, which leaves an agent no state. No state and no agent may be a +// plain shell, in a command that reads what it is typed. func queuedCanGo(l stallLook, screen string) bool { - return l.Info.StateStale && !l.Info.Exited && !l.Info.Queued && !l.Gone && - l.Still >= queuedQuiet && (l.NoCPU || l.CPU < stallCPUIdle) && agent.PromptFree(screen) + if l.Info.Exited || l.Info.Queued || l.Gone || l.Still < queuedQuiet { + return false + } + switch { + case l.Info.State == proto.StateDone: + return l.NoCPU || l.CPU < turnOverCPU // an agent at rest isn't at 0 + case l.Info.StateStale: + case l.Info.State == "": + if _, isAgent := agent.Find(l.Info.Agent); !isAgent { + return false + } + default: + return false + } + return (l.NoCPU || l.CPU < stallCPUIdle) && agent.PromptFree(screen) } const ( @@ -143,6 +157,7 @@ type stallFollow struct { since time.Time told time.Time // its last line fired bool // the hook ran for this stall + waking bool // its queued task's session is being brought back } // stallHost is what a host says of its sessions, for one look. @@ -275,7 +290,14 @@ func (s *Server) stallHostLost(w *stallWatch, key string, t proto.GOATask, lim p return } limit := stallSeconds(lim.Idle) - if limit <= 0 || time.Since(since) < limit || t.Status != proto.GOATaskWorking { + if limit <= 0 || time.Since(since) < limit { + return + } + if t.Status == proto.GOATaskQueued { + s.queuedHostLost(w, t) + return + } + if t.Status != proto.GOATaskWorking { return } if s.stallMark(t.ID, proto.GOATaskWorking, proto.GOATaskStalled) { @@ -283,6 +305,24 @@ func (s *Server) stallHostLost(w *stallWatch, key string, t proto.GOATask, lim p } } +// queuedHostLost says once, on the board and to GOA, that queued task t waits for its +// host. It stays queued, and goes in when the host is back. +func (s *Server) queuedHostLost(w *stallWatch, t proto.GOATask) { + f := w.follows[t.ID] + if f == nil { + f = &stallFollow{since: time.Now()} + w.follows[t.ID] = f + } + if !f.told.IsZero() { + return + } + f.told = time.Now() + const why = "Its host can't be reached (the app that reaches it is closed, or the connection dropped): it goes in when the host is back." + _, _ = s.goa.board.Update(t.ID, func(t *proto.GOATask) { t.Why = strings.TrimSpace(t.Why + " " + why) }) + s.goaChanged() + s.goaTellLine(goaLine{Task: t.ID, Session: t.Session, Text: fmt.Sprintf("[enowX GOA] Task %s %q is still queued. %s", t.ID, t.Title, why)}) +} + // queuedCheck gives task t's session its oldest queued task if its prompt is // free, on host h, and says whether it did. func (s *Server) queuedCheck(w *stallWatch, t proto.GOATask, h stallHost) bool { @@ -293,8 +333,23 @@ func (s *Server) queuedCheck(w *stallWatch, t proto.GOATask, h stallHost) bool { w.follows[t.ID] = f } info, there := h.infos[t.Session] - if !there { + switch { + case !there: + // Its host answers and has no such session, for longer than a restart takes. + if now.Sub(f.since) >= queuedQuiet { + s.goaQueuedFailed(t, "its session is gone") + return true + } + return false + case info.Exited && info.Hibernated: + if !f.waking { // once: waking it takes a while, and each pass comes back here + f.waking = true + go s.queuedWake(t, info) + } return false + case info.Exited: + s.goaDeliver(info) // says it can't go + return true } text, err := h.screen(t.Session) if err != nil { @@ -312,6 +367,20 @@ func (s *Server) queuedCheck(w *stallWatch, t proto.GOATask, h stallHost) bool { return true } +// queuedWake brings back the hibernated session of queued task t and gives it its task. +func (s *Server) queuedWake(t proto.GOATask, info proto.SessionInfo) { + from := "" + if g, ok := s.goaSession(); ok { + from = g.ID + } + if _, err := s.goaReady(proto.GOADelegateParams{From: from, Host: t.Host}, info); err != nil { + s.goaQueuedFailed(t, "its session couldn't be brought back: "+err.Error()) + return + } + info.Exited, info.Hibernated = false, false + s.goaDeliver(info) +} + // stallCheck tells what's new of task t, on host h. func (s *Server) stallCheck(w *stallWatch, t proto.GOATask, h stallHost, lim proto.GOAStall) { now := time.Now() diff --git a/daemon/goa_stall_test.go b/daemon/goa_stall_test.go index bf4af8a3..2c5ba0da 100644 --- a/daemon/goa_stall_test.go +++ b/daemon/goa_stall_test.go @@ -61,6 +61,12 @@ func TestQueuedCanGo(t *testing.T) { {"it asks", stallLook{Info: stale, Still: time.Minute}, "Do you want to proceed?\n ❯ 1. Yes\n 2. No\n", false}, {"a blank screen", stallLook{Info: stale, Still: time.Minute}, " \n", false}, {"no state: maybe a shell", stallLook{Still: time.Minute}, free, false}, + {"no state, an agent: after a restart", stallLook{Info: proto.SessionInfo{Agent: "claude"}, Still: time.Minute}, free, true}, + {"no state, an agent at work", stallLook{Info: proto.SessionInfo{Agent: "claude"}, Still: time.Minute}, "✶ Thinking… (esc to interrupt)\n", false}, + {"done, its turn over before the task came", stallLook{Info: proto.SessionInfo{State: proto.StateDone}, Still: time.Minute, CPU: 2}, "anything\n", true}, + {"done, but its screen moves", stallLook{Info: proto.SessionInfo{State: proto.StateDone}, Still: time.Second}, free, false}, + {"done, but something runs", stallLook{Info: proto.SessionInfo{State: proto.StateDone}, Still: time.Minute, CPU: 80}, free, false}, + {"asking", stallLook{Info: proto.SessionInfo{State: proto.StateWaiting}, Still: time.Minute}, free, false}, {"really working", stallLook{Info: proto.SessionInfo{State: proto.StateWorking}, Still: time.Minute}, free, false}, {"its own input queued", stallLook{Info: proto.SessionInfo{StateStale: true, Queued: true}, Still: time.Minute}, free, false}, {"ended", stallLook{Info: proto.SessionInfo{StateStale: true, Exited: true}, Still: time.Minute}, free, false}, diff --git a/daemon/goa_tasks.go b/daemon/goa_tasks.go index d69e4648..ec5718d7 100644 --- a/daemon/goa_tasks.go +++ b/daemon/goa_tasks.go @@ -171,15 +171,47 @@ func (s *Server) goaSkipOn(p proto.GOADelegateParams, info proto.SessionInfo) { // goaDeliver gives a session that ended its turn its oldest queued task, or, // when it ended for good, says those tasks can't go. func (s *Server) goaDeliver(info proto.SessionInfo) { + // One at a time, from claiming a task until it has been typed: a turn's end, an + // app's word and the stall watch may all come at once, and each change of a session + // still done from its last turn says so again. + s.goa.delivering.Lock() + defer s.goa.delivering.Unlock() + for s.goaDeliverNext(info) { + } +} + +// goaDeliverNext gives or fails session info's oldest queued task, and says whether +// the next one is to be looked at too: all of them fail with a session that ended. +func (s *Server) goaDeliverNext(info proto.SessionInfo) bool { var next *proto.GOATask + running := false tasks := s.goa.board.List() for i := range tasks { - if t := &tasks[i]; t.Session == info.ID && t.Status == proto.GOATaskQueued && (next == nil || t.Created.Before(next.Created)) { - next = t + t := &tasks[i] + if t.Session != info.ID { + continue + } + switch t.Status { + case proto.GOATaskQueued: + if next == nil || t.Created.Before(next.Created) { + next = t + } + case proto.GOATaskWorking, proto.GOATaskWaiting, proto.GOATaskStalled: + running = true } } if next == nil { - return + return false + } + if !info.Exited { + if running { + return false // the task before has its turn: this one follows it + } + if sess, err := s.sessions.Get(info.ID); err == nil { + if now := sess.Info(); now.State == proto.StateWorking && !now.StateStale { + return false // at work again since this change was heard + } + } } if info.Exited { t, _ := s.goa.board.Update(next.ID, func(t *proto.GOATask) { @@ -187,8 +219,7 @@ func (s *Server) goaDeliver(info proto.SessionInfo) { }) s.goaChanged() s.goaTell(fmt.Sprintf("[enowX GOA] Task %s %q failed: its session %s ended before it could take it. Delegate it again.", t.ID, t.Title, t.Session)) - s.goaDeliver(info) - return + return true } from := "" if g, ok := s.goaSession(); ok { @@ -203,15 +234,25 @@ func (s *Server) goaDeliver(info proto.SessionInfo) { } }) if !claimed { - return + return false } err := s.goaCall(from, next.Host, proto.MethodSessionInput, proto.SessionInputParams{ID: next.Session, Text: next.Prompt, Paste: true, Enter: true}, nil) if err != nil { - _, _ = s.goa.board.Update(next.ID, func(t *proto.GOATask) { - t.Status, t.Result = proto.GOATaskFailed, "it couldn't be delivered: "+err.Error() - }) + // Not tried again: it may have gone in, and a task must not go in twice. + s.goaQueuedFailed(*next, "it couldn't be delivered, and may not have reached its session: "+err.Error()) + return false } + s.logger().Info("goa: a queued task went in", "session", next.Session, "task", next.ID, "host", next.Host) + s.goaChanged() + return false +} + +// goaQueuedFailed says on the board and to GOA that queued task t won't go in. +func (s *Server) goaQueuedFailed(t proto.GOATask, why string) { + s.logger().Warn("goa: a queued task can't go in", "session", t.Session, "task", t.ID, "host", t.Host, "why", why) + _, _ = s.goa.board.Update(t.ID, func(t *proto.GOATask) { t.Status, t.Result = proto.GOATaskFailed, why }) s.goaChanged() + s.goaTell(fmt.Sprintf("[enowX GOA] Task %s %q failed: %s. Check session %s, then delegate it again.", t.ID, t.Title, why, t.Session)) } // goaSpawn starts a new session for task t on p's host, as GOA's worker. diff --git a/daemon/rpc_goa.go b/daemon/rpc_goa.go index f5788ab2..045a441c 100644 --- a/daemon/rpc_goa.go +++ b/daemon/rpc_goa.go @@ -17,10 +17,7 @@ var goaHandlers = map[string]handler{ }), proto.MethodGOAWorker: with(func(c *conn, p proto.GOAWorkerParams) (any, error) { if c.srv.goa.board.Has(p.Session.ID) { - go func() { - c.srv.goaReportSeen(p.Session) - c.srv.goaWorkerChanged(p.Session) - }() + go c.srv.goaWorkerSeen(p.Session) } else { go c.srv.goaToldChanged(p.Session) } diff --git a/docs/goa.md b/docs/goa.md index abfe669e..bd37dc72 100644 --- a/docs/goa.md +++ b/docs/goa.md @@ -132,7 +132,7 @@ A task shows its project's name, path and host, kept from when it was delegated A session is also read for what stops it, every few seconds, whatever started it, a shell tab with an agent run by hand included (`core/agent/prompts.go`): prompts in bypass mode, startup dialogs, usage limits, signing in, updates, listed by their words; and, for what nobody listed, a screen still for 20 seconds, not saying it works, that ends on a question or a menu. It goes to waiting, GOA's line quotes what's asked and any time it gives, and `session_ask` says the same; when a timed prompt goes, GOA is told to read what happened. -A task for a project goes to that project's own session, even busy, asking, hibernated or started with other permissions (`goa delegate --project`, `core/goa.PickSession`). A busy one has it queued, shown so on the board, delivered at its turn's end, oldest first; if the session ends first the task fails with that said. A new session starts only when the project has none, or on `--branch` or a named `--config`, what the user asked for, and delegate says so. Worktrees and shell tabs don't take a project's tasks. A queued task goes in when its session's turn ends, or, its turn's end unheard as after an interrupt, once its prompt is really free: hooks gone stale, its screen still for 20 seconds, nothing running, and it neither works nor asks by what it shows. +A task for a project goes to that project's own session, even busy, asking, hibernated or started with other permissions (`goa delegate --project`, `core/goa.PickSession`). A busy one has it queued, shown so on the board and kept over a daemon's restart, delivered at its turn's end, oldest first and one at a time; a queued task goes in only once the task before it has stopped running. Its turn's end is heard from its hooks, or for another host's session from the app. A session whose end went unheard (an interrupt, a daemon's restart, a turn over before the task came) gets it from the stall watch, once its screen has been still for 20 seconds with nothing running. A hibernated session is brought back first. If it can't go in, because its session ended or is gone, can't be brought back, or typing failed, the task fails, the board says why and GOA is told; a failed typing isn't tried again, as it may have gone in. A host that can't be reached keeps it queued, and GOA is told once. A new session starts only when the project has none, or on `--branch` or a named `--config`, what the user asked for, and delegate says so. Worktrees and shell tabs don't take a project's tasks. A queued task goes in when its session's turn ends, or, its turn's end unheard as after an interrupt, once its prompt is really free: hooks gone stale, its screen still for 20 seconds, nothing running, and it neither works nor asks by what it shows. A task that isn't moving goes to `stalled` on the board until its screen moves, and GOA gets one line naming how, for how long and what the screen shows; a further one comes only after three times the limit, and again each time that passes while it stays stalled. The hook event `session.stalled` fires once per stall, with no hook needed for GOA to be told.