Tell GOA of a worker's turn in one line, with its real last message

core/session/session.go

@@ -85,6 +85,8 @@ type Meta struct {
 	// Note is what the session is at, as its agent or the user last said
 	// (SessionInfo.Note).
 	Note string
+	// TurnText is the message its last turn ended on, when its agent's hook says.
+	TurnText string
 	// Report, ReportOutcome and ReportAt are its agent's last report of a turn
 	// (SessionInfo). They don't outlast the daemon, their news having gone.
 	Report        string
@@ -207,6 +209,7 @@ func (s *Session) Info() proto.SessionInfo {
 		Report:          s.meta.Report,
 		ReportOutcome:   s.meta.ReportOutcome,
 		ReportAt:        s.meta.ReportAt,
+		TurnText:        s.meta.TurnText,
 		Group:           s.meta.Group,
 		Account:         s.meta.Account,
 		Config:          cloneConfig(s.meta.Config),

core/session/work.go

@@ -104,6 +104,21 @@ func (m *Manager) SetNote(s *Session, note string) {
 	m.notify(s)
 }
 
+// turnTextMost bounds what a session keeps of its turn's last message: what GOA's line
+// holds, as it goes with every update of the session.
+const turnTextMost = 2000
+
+// NoteTurnText keeps the message s's turn ended on, or "" as a turn starts. It says
+// nothing by itself: a change of state follows.
+func (m *Manager) NoteTurnText(s *Session, text string) {
+	if r := []rune(text); len(r) > turnTextMost {
+		text = string(r[:turnTextMost]) + "…"
+	}
+	s.mu.Lock()
+	s.meta.TurnText = text
+	s.mu.Unlock()
+}
+
 // SetReport keeps its agent's report of a turn.
 func (m *Manager) SetReport(s *Session, outcome, text string) {
 	s.mu.Lock()

daemon/goa.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"slices"
 	"sync"
+	"time"
 
 	"github.com/enowx/enowx/core/agent"
 	"github.com/enowx/enowx/core/goa"
@@ -198,6 +199,9 @@ func (s *Server) goaSeen(info proto.SessionInfo) {
 	if s.goa.board.Has(info.ID) {
 		s.goaRanOn(info)
 		go func() {
+			if info.State == proto.StateWorking {
+				s.goaTurnStarted(info.ID)
+			}
 			s.goaReportSeen(info)
 			s.goaWorkerChanged(info)
 			if info.State == proto.StateDone || info.Exited && !info.Hibernated {
@@ -242,6 +246,13 @@ type goaLine struct {
 	// Soft: it only says to look. It never keeps GOA's turn going, and reading the
 	// session takes it back.
 	Soft bool `json:"soft,omitempty"`
+	// Turn says it tells of a turn (goa_turnline.go): a report (goaLineReport) or its
+	// end (goaLineTurn). Those of one turn become one line.
+	Turn string `json:"turn,omitempty"`
+	// Closed: its turn is over and told, so nothing more merges into it.
+	Closed bool `json:"closed,omitempty"`
+	// HoldUntil keeps a report back for its turn's end to join it.
+	HoldUntil time.Time `json:"holdUntil,omitzero"`
 }
 
 // goaTellMax bounds what waits for a GOA that never comes back, the oldest going.
@@ -256,6 +267,9 @@ func (s *Server) goaTellLine(line goaLine) {
 		return
 	}
 	s.goa.mu.Lock()
+	if line.Turn == "" && line.Session != "" {
+		releaseHeld(s.goa.tell, line.Session) // what it says of the session comes first
+	}
 	s.goa.tell = append(s.goa.tell, line)
 	if over := len(s.goa.tell) - goaTellMax; over > 0 {
 		s.goa.tell = slices.Delete(s.goa.tell, 0, over)
@@ -320,16 +334,23 @@ func (s *Server) goaFlushIdle() {
 
 // goaLinesToSay takes what waits for GOA, less what its task's end made moot.
 func (s *Server) goaLinesToSay() []goaLine {
+	now := time.Now()
 	s.goa.mu.Lock()
 	lines := s.goa.tell
 	s.goa.tell = nil
-	s.goa.mu.Unlock()
+	// Of a session, nothing goes past what it holds back; another session's go on.
+	held := map[string]bool{}
 	var out []goaLine
 	for _, l := range lines {
-		if l.Task == "" || !s.goa.board.IsSettled(l.Task) {
+		switch {
+		case l.HoldUntil.After(now) || held[l.Session] && l.Session != "":
+			held[l.Session] = true
+			s.goa.tell = append(s.goa.tell, l)
+		case l.Task == "" || !s.goa.board.IsSettled(l.Task):
 			out = append(out, l)
 		}
 	}
+	s.goa.mu.Unlock()
 	return out
 }
 

daemon/goa_tasks.go

@@ -312,9 +312,12 @@ func (s *Server) goaWorkerChanged(info proto.SessionInfo) {
 		head := fmt.Sprintf("[enowX GOA] Task %s %q (%s)", t.ID, t.Title, where)
 		switch status {
 		case proto.GOATaskTurnEnded:
-			said := s.goaResult(t)
+			said := nonEmpty(info.TurnText, "")
+			if said == "" {
+				said = s.goaResult(t)
+			}
 			t, _ = s.goa.board.Update(t.ID, func(t *proto.GOATask) { t.Said = said })
-			s.goaTellLine(goaLine{Task: t.ID, Session: t.Session, Text: head + " finished a turn, which may be mid-work: " + clip(oneLine(said), goaResultMax) + " Check it (chat " + t.Session + hostFlag + "), then settle it: goa task " + t.ID + " --status done|failed --result \"…\"."})
+			s.goaTellTurnEnd(goaLine{Task: t.ID, Session: t.Session, Text: head + " finished a turn, which may be mid-work: " + clip(oneLine(said), goaResultMax) + " Check it (chat " + t.Session + hostFlag + "), then settle it: goa task " + t.ID + " --status done|failed --result \"…\"."}, said)
 		case proto.GOATaskWaiting:
 			ask := s.goaAskOf(t)
 			t, _ = s.goa.board.Update(t.ID, func(t *proto.GOATask) { t.Asks, t.Urgent = askLine(ask), ask.Urgent })
@@ -412,7 +415,7 @@ func (s *Server) goaReportSeen(info proto.SessionInfo) {
 	})
 	for _, t := range changed {
 		where, hostFlag := goaTaskPlace(t)
-		s.goaTellLine(goaLine{Task: t.ID, Session: t.Session, Text: fmt.Sprintf("[enowX GOA] Task %s %q (%s) reports %s: %s Check it (chat %s%s), then settle it: goa task %s --status done|failed --result \"…\".", t.ID, t.Title, where, t.Outcome, clip(oneLine(t.Said), goaResultMax), t.Session, hostFlag, t.ID)})
+		s.goaTellReport(info, goaLine{Task: t.ID, Session: t.Session, Text: fmt.Sprintf("[enowX GOA] Task %s %q (%s) reports %s: %s Check it (chat %s%s), then settle it: goa task %s --status done|failed --result \"…\".", t.ID, t.Title, where, t.Outcome, clip(oneLine(t.Said), goaResultMax), t.Session, hostFlag, t.ID)})
 	}
 	if len(changed) > 0 {
 		s.goaChanged()
@@ -458,13 +461,24 @@ func (s *Server) goaResult(t proto.GOATask) string {
 	if g, ok := s.goaSession(); ok {
 		from = g.ID
 	}
-	var tr proto.Transcript
-	if err := s.goaCall(from, t.Host, proto.MethodSessionTranscript, proto.TranscriptParams{ID: t.Session, Tail: 20}, &tr); err == nil {
-		for i := len(tr.Entries) - 1; i >= 0; i-- {
-			if e := tr.Entries[i]; e.Kind == proto.TranscriptAssistant && strings.TrimSpace(e.Text) != "" {
-				return clip(strings.TrimSpace(e.Text), goaResultMax)
+	// Never a message from before the turn's end: wait for the transcript to have it.
+	deadline := time.Now().Add(turnEndWait)
+	for {
+		var tr proto.Transcript
+		err := s.goaCall(from, t.Host, proto.MethodSessionTranscript, proto.TranscriptParams{ID: t.Session, Tail: 20}, &tr)
+		if err != nil {
+			break // it has none: its screen says
+		}
+		if text, ok := turnEndIn(tr.Entries); ok {
+			return clip(text, goaResultMax)
+		}
+		if time.Now().After(deadline) {
+			if len(tr.Entries) == 0 {
+				break
 			}
+			return "(its last message isn't in its transcript yet: chat " + t.Session + " shows it)"
 		}
+		time.Sleep(turnEndEvery)
 	}
 	var sc proto.SessionScreen
 	if err := s.goaCall(from, t.Host, proto.MethodSessionScreen, proto.SessionRef{ID: t.Session}, &sc); err != nil {

daemon/goa_told.go

@@ -157,6 +157,9 @@ func (s *Server) goaMarkTold(from, host, id string) {
 // goaToldChanged tells GOA of a session it sent something to, with no task, as it
 // is told of a worker: its report, its turn's end, what it asks, its end.
 func (s *Server) goaToldChanged(info proto.SessionInfo) {
+	if info.State == proto.StateWorking {
+		s.goaTurnStarted(info.ID)
+	}
 	s.goa.mu.Lock()
 	was, told := s.goa.told[info.ID]
 	if !told {
@@ -185,14 +188,21 @@ func (s *Server) goaToldChanged(info proto.SessionInfo) {
 	s.saveGOATold()
 	t := proto.GOATask{Session: info.ID}
 	if info.ReportAt > was.ReportAt {
-		s.goaTellAbout(info.ID, fmt.Sprintf("[enowX GOA] Session %s, which you sent something to, reports %s: %s", info.ID, info.ReportOutcome, clip(oneLine(info.Report), goaResultMax)))
-		if status == was.Status || status == proto.GOATaskTurnEnded {
-			return // that was its turn's end
+		s.goaTellReport(info, goaLine{Session: info.ID, Text: fmt.Sprintf("[enowX GOA] Session %s, which you sent something to, reports %s: %s", info.ID, info.ReportOutcome, clip(oneLine(info.Report), goaResultMax))})
+		if status == was.Status {
+			return // what it reports is all there is to say
+		}
+		if status != proto.GOATaskTurnEnded {
+			return // its turn goes on: its end joins the report
 		}
 	}
 	switch status {
 	case proto.GOATaskTurnEnded:
-		s.goaTellAbout(info.ID, fmt.Sprintf("[enowX GOA] Session %s, which you sent something to, finished a turn: %s", info.ID, clip(oneLine(s.goaResult(t)), goaResultMax)))
+		said := info.TurnText
+		if said == "" {
+			said = s.goaResult(t)
+		}
+		s.goaTellTurnEnd(goaLine{Session: info.ID, Text: fmt.Sprintf("[enowX GOA] Session %s, which you sent something to, finished a turn: %s", info.ID, clip(oneLine(said), goaResultMax))}, said)
 	case proto.GOATaskWaiting:
 		s.goaTellAbout(info.ID, fmt.Sprintf("[enowX GOA] Session %s, which you sent something to, asks something: %s", info.ID, clip(oneLine(s.goaResult(t)), goaResultMax)))
 	case proto.GOATaskFailed, proto.GOATaskStopped:

daemon/goa_turnline.go

@@ -0,0 +1,145 @@
+package daemon
+
+import (
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/enowx/enowx/proto"
+)
+
+// A worker's turn says two things of itself: its agent's goa report, a tool call made
+// before the turn is over, and the turn's end. GOA gets them as one line. A report made
+// at work waits for its turn's end, for a while, and the end joins it; a turn's end still
+// waiting for GOA takes a report that comes after it. Lines never change places, a
+// question or a failure is never merged, and a later turn has its own line.
+
+const (
+	goaLineReport = "report"
+	goaLineTurn   = "turn"
+	// goaReportHold is how long a report waits for its turn's end: past it, it goes alone,
+	// as a turn may go on long after, or never be heard to end.
+	goaReportHold = 30 * time.Second
+)
+
+// releaseHeld lets what lines hold back of session go. The caller holds goa.mu.
+func releaseHeld(lines []goaLine, session string) {
+	for i := range lines {
+		if lines[i].Session == session {
+			lines[i].HoldUntil = time.Time{}
+		}
+	}
+}
+
+// goaTellReport tells of a report of session info, as line: joined to its turn's end if
+// that waits for GOA, else held for it if the session is at work.
+func (s *Server) goaTellReport(info proto.SessionInfo, line goaLine) {
+	line.Turn = goaLineReport
+	s.goa.mu.Lock()
+	i := s.goaOpenLine(line, goaLineTurn)
+	if i >= 0 {
+		said := s.goa.tell[i].Text
+		s.goa.tell[i].Text, s.goa.tell[i].Turn, s.goa.tell[i].Closed = line.Text+lastMessageOf(said), goaLineTurn, true
+	}
+	s.goa.mu.Unlock()
+	if i >= 0 {
+		s.goaLinesChanged()
+		return
+	}
+	if info.State == proto.StateWorking && !info.StateStale {
+		line.HoldUntil = time.Now().Add(goaReportHold)
+	} else {
+		line.Closed = true
+	}
+	s.goaTellLine(line)
+}
+
+// goaTellTurnEnd tells of a turn's end, as line, saying said last: into its report if
+// that waits for GOA. A report already gone leaves it a line of its own.
+func (s *Server) goaTellTurnEnd(line goaLine, said string) {
+	line.Turn = goaLineTurn
+	s.goa.mu.Lock()
+	i := s.goaOpenLine(line, goaLineReport)
+	if i >= 0 {
+		l := &s.goa.tell[i]
+		l.HoldUntil, l.Closed = time.Time{}, true
+		if text := strings.TrimSpace(said); text != "" && !strings.Contains(oneLine(l.Text), oneLine(text)) {
+			l.Text += turnEndMark + oneLine(text)
+		}
+	}
+	s.goa.mu.Unlock()
+	if i >= 0 {
+		s.goaLinesChanged()
+		return
+	}
+	s.goaTellLine(line)
+}
+
+// turnEndMark comes before the turn's last message, in a line that holds its report.
+const turnEndMark = " Its turn ended on: "
+
+// lastMessageOf is a turn's end line as what follows its report in one line.
+func lastMessageOf(line string) string {
+	if _, said, ok := strings.Cut(line, ": "); ok {
+		return turnEndMark + said
+	}
+	return ""
+}
+
+// goaOpenLine is the last line waiting for GOA of line's session and task, of kind turn,
+// still open to joining; -1 for none. The caller holds goa.mu.
+func (s *Server) goaOpenLine(line goaLine, turn string) int {
+	return slices.IndexFunc(s.goa.tell, func(l goaLine) bool {
+		return l.Turn == turn && !l.Closed && l.Session == line.Session && l.Task == line.Task
+	})
+}
+
+// goaTurnStarted closes what waits of session's turns: what comes now is of another.
+func (s *Server) goaTurnStarted(session string) {
+	s.goa.mu.Lock()
+	changed := false
+	for i := range s.goa.tell {
+		if l := &s.goa.tell[i]; l.Session == session && l.Turn == goaLineTurn && !l.Closed {
+			l.Closed, changed = true, true
+		}
+	}
+	s.goa.mu.Unlock()
+	if changed {
+		s.saveGOATell()
+	}
+}
+
+// goaLinesChanged saves what waits, and gives it to an idle GOA.
+func (s *Server) goaLinesChanged() {
+	s.saveGOATell()
+	if g, ok := s.goaSession(); ok && !g.Exited && g.State != proto.StateWorking {
+		s.goaFlush(g.ID)
+	}
+	s.goaChanged()
+}
+
+// How long a turn's last message may take to reach its transcript, for an agent whose
+// hook doesn't say it. Claude Code's transcript was seen behind its Stop hook by over a
+// second (2.1.318).
+const (
+	turnEndWait  = 6 * time.Second
+	turnEndEvery = 400 * time.Millisecond
+)
+
+// turnEndIn is the message transcript entries end on, once it holds the turn's end: the
+// last there is but thinking and notices. What follows a tool's call or a prompt isn't
+// written yet, and a message before them isn't the turn's last.
+func turnEndIn(entries []proto.TranscriptEntry) (string, bool) {
+	for i := len(entries) - 1; i >= 0; i-- {
+		switch e := entries[i]; e.Kind {
+		case proto.TranscriptThinking, proto.TranscriptNotice:
+		case proto.TranscriptAssistant:
+			if text := strings.TrimSpace(e.Text); text != "" {
+				return text, true
+			}
+		default:
+			return "", false
+		}
+	}
+	return "", false
+}