feat: add mobile chat view for live agent sessions

internal/agentstore/index.go

@@ -594,6 +594,38 @@ func (x *Index) Resolve(tool Tool, id string) (string, Summary, bool) {
 	return "", Summary{}, false
 }
 
+// MatchSession finds the newest live transcript in cwd whose file was
+// written since the shell was created. It is the fallback link between a PTY
+// session and an agent conversation when no agent hook supplied an exact id.
+//
+// The match deliberately uses file mtime rather than EndedAt. Timestamps
+// inside vendor-owned files are less trustworthy than the filesystem fact
+// that the agent just appended to it. The index sweep is poked first; callers
+// receive the best snapshot available now and naturally see a newly
+// discovered transcript on their next session-list poll.
+func (x *Index) MatchSession(cwd string, since time.Time) (Tool, string, bool) {
+	if cwd == "" {
+		return "", "", false
+	}
+	x.poke()
+	x.mu.Lock()
+	defer x.mu.Unlock()
+	var best *entry
+	for _, e := range x.entries {
+		if e == nil || e.ParentPath != "" || e.Missing || e.Summary.MessageCount == 0 ||
+			e.Summary.Cwd != cwd || e.MtimeNs < since.UnixNano() {
+			continue
+		}
+		if best == nil || e.MtimeNs > best.MtimeNs {
+			best = e
+		}
+	}
+	if best == nil {
+		return "", "", false
+	}
+	return best.Summary.Tool, best.Summary.ID, true
+}
+
 func toolAllowed(tool Tool, filter []string) bool {
 	if len(filter) == 0 {
 		return true

internal/agentstore/index_test.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 )
 
 const (
@@ -111,6 +112,33 @@ func TestSnapshotFilters(t *testing.T) {
 	}
 }
 
+func TestMatchSessionPicksNewestTranscriptSinceShellStart(t *testing.T) {
+	home, paths := fakeHome(t)
+	base := time.Now().Add(-time.Hour).Truncate(time.Second)
+	for tool, at := range map[Tool]time.Time{
+		ToolClaude: base.Add(10 * time.Minute),
+		ToolCodex:  base.Add(20 * time.Minute),
+		ToolPi:     base.Add(30 * time.Minute),
+	} {
+		if err := os.Chtimes(paths[tool], at, at); err != nil {
+			t.Fatalf("Chtimes(%s): %v", tool, err)
+		}
+	}
+	x := New(t.TempDir(), home)
+	x.sweep()
+
+	tool, id, ok := x.MatchSession("/home/dev/proj", base.Add(15*time.Minute))
+	if !ok || tool != ToolPi || id != piFixtureID {
+		t.Fatalf("MatchSession = (%s, %s, %v), want newest pi transcript", tool, id, ok)
+	}
+	if _, _, ok := x.MatchSession("/home/dev/proj", base.Add(40*time.Minute)); ok {
+		t.Fatal("MatchSession linked a transcript older than the shell")
+	}
+	if _, _, ok := x.MatchSession("/somewhere/else", base); ok {
+		t.Fatal("MatchSession linked a transcript from another cwd")
+	}
+}
+
 // TestSnapshotHidesEmptySessions pins the stub filter: a transcript whose
 // lines are all bookkeeping (a real store is full of them — hook-only
 // sessions, bare last-prompt stubs) is indexed but never listed. Its zero

internal/daemon/agents.go

@@ -2,8 +2,10 @@ package daemon
 
 import (
 	"errors"
+	"time"
 
 	"github.com/karnstack/flue/internal/agentstore"
+	"github.com/karnstack/flue/internal/session"
 	"github.com/karnstack/flue/internal/wire"
 )
 
@@ -46,6 +48,17 @@ import (
 // goroutine to cap.
 const maxAgentWork = 4
 
+// maxAgentFollows bounds the durable file tails one connection can hold.
+// Eight covers a densely split workspace while preventing a malformed client
+// from turning one socket into an unbounded transcript poller.
+const maxAgentFollows = 8
+
+// agentFollowPoll is deliberately much faster than the old viewer's five
+// second timer while still coalescing the burst of filesystem writes that
+// makes one completed agent message. The read itself advances by byte offset,
+// so an unchanged file costs an open and stat rather than a reparse.
+const agentFollowPoll = 400 * time.Millisecond
+
 // SetAgentIndex installs the transcript index. Nil — a Server nobody wired,
 // which is every test's default and a daemon whose home directory could not
 // be resolved — leaves the verbs answering bad_message and keeps the
@@ -63,6 +76,22 @@ func (s *Server) agentIndex() *agentstore.Index {
 	return s.agentIdx
 }
 
+// withAgent enriches one session-list row with the best live transcript link
+// the index can prove from cwd and recency. Hooks can make this exact later;
+// keeping the fallback here means both WebSocket and loopback HTTP lists use
+// the same answer and no session implementation has to know agent stores.
+func (s *Server) withAgent(info session.Info) session.Info {
+	idx := s.agentIndex()
+	if idx == nil || info.State != "running" {
+		return info
+	}
+	tool, id, ok := idx.MatchSession(info.Cwd, info.CreatedAt)
+	if ok {
+		info.Agent, info.AgentSession = string(tool), id
+	}
+	return info
+}
+
 // handleAgents answers the agents verb. A snapshot of the in-memory index —
 // no file work — so it runs on the read loop like list does.
 func (c *conn) handleAgents(m wire.Agents) {
@@ -155,6 +184,99 @@ func (c *conn) handleAgentRead(m wire.AgentRead) {
 	}()
 }
 
+func agentFollowKey(tool, id string) string { return tool + "\x00" + id }
+
+// handleAgentFollow starts (or moves) a pushed transcript tail. Pages carry
+// no reqId: they are events rather than answers, and the client routes them by
+// (tool, id). Reads remain bounded by agentstore's ordinary page limits; a
+// large append drains page by page without waiting another tick.
+func (c *conn) handleAgentFollow(m wire.AgentFollow) {
+	idx := c.srv.agentIndex()
+	if idx == nil {
+		c.sendError("bad_message", "agents unavailable")
+		return
+	}
+	tool := agentstore.Tool(m.Tool)
+	if _, _, ok := idx.Resolve(tool, m.ID); !ok {
+		c.sendError("not_found", "cannot follow that transcript")
+		return
+	}
+	if m.Offset < 0 {
+		m.Offset = 0
+	}
+	key := agentFollowKey(m.Tool, m.ID)
+	done := make(chan struct{})
+	c.mu.Lock()
+	previous := c.agentFollows[key]
+	if previous == nil && len(c.agentFollows) >= maxAgentFollows {
+		c.mu.Unlock()
+		c.sendError("busy", "too many followed transcripts")
+		return
+	}
+	c.agentFollows[key] = done
+	c.mu.Unlock()
+	if previous != nil {
+		close(previous)
+	}
+
+	c.pumps.Add(1)
+	go func(offset int64) {
+		defer c.pumps.Done()
+		defer func() {
+			c.mu.Lock()
+			if c.agentFollows[key] == done {
+				delete(c.agentFollows, key)
+			}
+			c.mu.Unlock()
+		}()
+
+		timer := time.NewTimer(0)
+		defer timer.Stop()
+		for {
+			select {
+			case <-c.ctx.Done():
+				return
+			case <-done:
+				return
+			case <-timer.C:
+			}
+
+			for {
+				page, err := idx.ReadPage(tool, m.ID, offset, "forward", 100)
+				if err != nil {
+					return
+				}
+				moved := page.Next != offset
+				offset = page.Next
+				if len(page.Messages) > 0 {
+					if err := c.sendControl(wire.AgentPage{
+						Tool: m.Tool, ID: m.ID, Messages: page.Messages,
+						Start: page.Start, Next: page.Next, Eof: page.Eof,
+						FileSize: page.FileSize,
+					}); err != nil {
+						return
+					}
+				}
+				if page.Eof || !moved {
+					break
+				}
+			}
+			timer.Reset(agentFollowPoll)
+		}
+	}(m.Offset)
+}
+
+func (c *conn) handleAgentUnfollow(m wire.AgentUnfollow) {
+	key := agentFollowKey(m.Tool, m.ID)
+	c.mu.Lock()
+	done := c.agentFollows[key]
+	delete(c.agentFollows, key)
+	c.mu.Unlock()
+	if done != nil {
+		close(done)
+	}
+}
+
 // handleAgentSearch answers agentSearch on its own goroutine, for the reason
 // agentRead gets one with less argument: a search's budget is sixty-four
 // megabytes of other tools' files.

internal/daemon/agents_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/coder/websocket"
 	"github.com/karnstack/flue/internal/agentstore"
+	"github.com/karnstack/flue/internal/session"
 	"github.com/karnstack/flue/internal/wire"
 )
 
@@ -174,6 +175,41 @@ func TestAgentsListsTranscriptsNewestFirst(t *testing.T) {
 	})
 }
 
+func TestSessionInfoIsEnrichedWithLiveAgentTranscript(t *testing.T) {
+	ts, _, srv := newTestServerUI(t, http.NotFoundHandler())
+	home := agentFixtureHome(t)
+	claudePath := filepath.Join(home, ".claude", "projects", "-home-dev-proj", agentClaudeID+".jsonl")
+	codexPath := filepath.Join(home, ".codex", "sessions", "2026", "07", "01", "rollout-2026-07-01T10-00-00-"+agentCodexID+".jsonl")
+	piPath := filepath.Join(home, ".pi", "agent", "sessions", "--home-dev-proj--", "2026-03-27T12-35-33_"+agentPiID+".jsonl")
+	older := time.Now().Add(-2 * time.Minute)
+	for _, path := range []string{codexPath, piPath} {
+		if err := os.Chtimes(path, older, older); err != nil {
+			t.Fatalf("Chtimes older transcript: %v", err)
+		}
+	}
+	newest := time.Now().Add(-time.Minute)
+	if err := os.Chtimes(claudePath, newest, newest); err != nil {
+		t.Fatalf("Chtimes claude transcript: %v", err)
+	}
+	idx := agentstore.New(t.TempDir(), home)
+	srv.SetAgentIndex(idx)
+	c := dial(t, ts)
+	settleAgents(t, c)
+
+	got := srv.withAgent(session.Info{
+		State: "running", Cwd: "/home/dev/proj", CreatedAt: time.Now().Add(-time.Hour),
+	})
+	if got.Agent != "claude" || got.AgentSession != agentClaudeID {
+		t.Fatalf("withAgent = (%s, %s), want newest claude transcript", got.Agent, got.AgentSession)
+	}
+	exited := srv.withAgent(session.Info{
+		State: "exited", Cwd: "/home/dev/proj", CreatedAt: time.Now().Add(-time.Hour),
+	})
+	if exited.Agent != "" || exited.AgentSession != "" {
+		t.Fatalf("exited session was linked: %+v", exited)
+	}
+}
+
 func TestAgentReadPagesForwardAndBackward(t *testing.T) {
 	ts := newAgentTestServer(t)
 	c := dial(t, ts)
@@ -235,6 +271,53 @@ func TestAgentReadPagesForwardAndBackward(t *testing.T) {
 	})
 }
 
+func TestAgentFollowPushesCompletedMessages(t *testing.T) {
+	ts, _, srv := newTestServerUI(t, http.NotFoundHandler())
+	home := agentFixtureHome(t)
+	idx := agentstore.New(t.TempDir(), home)
+	srv.SetAgentIndex(idx)
+	c := dial(t, ts)
+	writeControl(t, c, wire.Hello{Ver: "test"})
+	settleAgents(t, c)
+
+	path, summary, ok := idx.Resolve(agentstore.ToolClaude, agentClaudeID)
+	if !ok {
+		t.Fatal("Resolve missed claude fixture")
+	}
+	writeControl(t, c, wire.AgentFollow{
+		Tool: "claude", ID: agentClaudeID, Offset: summary.FileSize,
+	})
+
+	f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	line := `{"parentUuid":"x","isSidechain":false,"cwd":"/home/dev/proj","sessionId":"` + agentClaudeID + `","type":"assistant","requestId":"follow","message":{"id":"follow","type":"message","role":"assistant","model":"claude-fable-5","content":[{"type":"text","text":"Followed live."}],"usage":{"input_tokens":1,"output_tokens":1}},"uuid":"follow","timestamp":"2026-08-10T09:00:30.000Z"}` + "\n"
+	if _, err := f.WriteString(line); err != nil {
+		f.Close()
+		t.Fatal(err)
+	}
+	if err := f.Close(); err != nil {
+		t.Fatal(err)
+	}
+	direct, err := idx.ReadPage(agentstore.ToolClaude, agentClaudeID, summary.FileSize, "forward", 100)
+	if err != nil || len(direct.Messages) != 1 {
+		t.Fatalf("direct read after append = (%+v, %v), want one message", direct, err)
+	}
+
+	readUntil(t, c, func(msg any, _ []byte) bool {
+		page, ok := msg.(wire.AgentPage)
+		if !ok || page.ReqID != 0 || page.ID != agentClaudeID {
+			return false
+		}
+		if len(page.Messages) != 1 || page.Messages[0].Text != "Followed live." {
+			t.Errorf("pushed page = %+v", page.Messages)
+		}
+		return true
+	})
+	writeControl(t, c, wire.AgentUnfollow{Tool: "claude", ID: agentClaudeID})
+}
+
 func TestAgentSearchFindsAndBounds(t *testing.T) {
 	ts := newAgentTestServer(t)
 	c := dial(t, ts)