Merge pull request #207 from m1lian/perf/swarm-detached-log-tail
Read only the event log tail for detached swarm agents
7 files · +251 -5
M README.md +1 -1
| @@ -448,7 +448,7 @@ Background subagents that run alongside your main session. Each one is a separat | |||
| 448 | 448 | ||
| 449 | 449 | **Session scoping** — each agent is stamped with the host session that spawned it and only shows up in that session's dashboard. Swap sessions with `/sessions` and the dashboard re-narrows accordingly. Agents from other sessions keep running in the background and reappear when you switch back. | |
| 450 | 450 | ||
| 451 | - | **Persistence across zot restarts** — every spawn writes a `meta.json` next to its event log and session file under `$ZOT_HOME/swarm/agents/<id>/`. On the next `zot` launch they show up in the dashboard as **detached**; press `R` (or `/swarm resume <id>`) to bring one back. Resumed agents reattach to the same session and inbox socket, so the conversation continues from where it left off. | |
| 451 | + | **Persistence across zot restarts** — every spawn writes a `meta.json` next to its event log and session file under `$ZOT_HOME/swarm/agents/<id>/`. On the next `zot` launch they show up in the dashboard as **detached**; press `R` (or `/swarm resume <id>`) to bring one back. Startup reads only the last 4 MiB of each large event log. Opening `/swarm` or resuming an agent loads its full log on demand, which can briefly delay that action for very large logs. The displayed transcript still retains at most 2000 lines. Resumed agents reattach to the same session and inbox socket, so the conversation continues from where it left off. | |
| 452 | 452 | ||
| 453 | 453 | **Where state lives** — everything per-agent (session file, events log, inbox socket, meta) lives under `$ZOT_HOME/swarm/agents/<id>/`. The agent's actual code edits land directly in your repo; track them with normal `git status` / `git diff`. | |
| 454 | 454 | ||
| @@ -448,7 +448,7 @@ Background subagents that run alongside your main session. Each one is a separat | |||||
| 448 | 448 | ||||
| 449 | **Session scoping** — each agent is stamped with the host session that spawned it and only shows up in that session's dashboard. Swap sessions with `/sessions` and the dashboard re-narrows accordingly. Agents from other sessions keep running in the background and reappear when you switch back. | 449 | **Session scoping** — each agent is stamped with the host session that spawned it and only shows up in that session's dashboard. Swap sessions with `/sessions` and the dashboard re-narrows accordingly. Agents from other sessions keep running in the background and reappear when you switch back. | ||
| 450 | 450 | ||||
| 451 | - | **Persistence across zot restarts** — every spawn writes a `meta.json` next to its event log and session file under `$ZOT_HOME/swarm/agents/<id>/`. On the next `zot` launch they show up in the dashboard as **detached**; press `R` (or `/swarm resume <id>`) to bring one back. Resumed agents reattach to the same session and inbox socket, so the conversation continues from where it left off. | 451 | + | **Persistence across zot restarts** — every spawn writes a `meta.json` next to its event log and session file under `$ZOT_HOME/swarm/agents/<id>/`. On the next `zot` launch they show up in the dashboard as **detached**; press `R` (or `/swarm resume <id>`) to bring one back. Startup reads only the last 4 MiB of each large event log. Opening `/swarm` or resuming an agent loads its full log on demand, which can briefly delay that action for very large logs. The displayed transcript still retains at most 2000 lines. Resumed agents reattach to the same session and inbox socket, so the conversation continues from where it left off. |
| 452 | 452 | ||||
| 453 | **Where state lives** — everything per-agent (session file, events log, inbox socket, meta) lives under `$ZOT_HOME/swarm/agents/<id>/`. The agent's actual code edits land directly in your repo; track them with normal `git status` / `git diff`. | 453 | **Where state lives** — everything per-agent (session file, events log, inbox socket, meta) lives under `$ZOT_HOME/swarm/agents/<id>/`. The agent's actual code edits land directly in your repo; track them with normal `git status` / `git diff`. | ||
| 454 | 454 | ||||
M packages/agent/swarm/agent.go +4 -2
| @@ -69,8 +69,10 @@ type Agent struct { | |||
| 69 | 69 | activity string | |
| 70 | 70 | transcript []string | |
| 71 | 71 | lastAssistant string | |
| 72 | - | finished time.Time | |
| 73 | - | lastErr error | |
| 72 | + | // Large detached logs are loaded in full only when their transcript is viewed or resumed. | |
| 73 | + | needsFullReplay bool | |
| 74 | + | finished time.Time | |
| 75 | + | lastErr error | |
| 74 | 76 | ||
| 75 | 77 | // OnTurnEnd, if set, fires once per prompt-level turn_end event | |
| 76 | 78 | // emitted by the swarm daemon wrapper. Provider/tool-loop | |
| @@ -69,8 +69,10 @@ type Agent struct { | |||||
| 69 | activity string | 69 | activity string | ||
| 70 | transcript []string | 70 | transcript []string | ||
| 71 | lastAssistant string | 71 | lastAssistant string | ||
| 72 | - | finished time.Time | 72 | + | // Large detached logs are loaded in full only when their transcript is viewed or resumed. |
| 73 | - | lastErr error | 73 | + | needsFullReplay bool |
| 74 | + | finished time.Time | |||
| 75 | + | lastErr error | |||
| 74 | 76 | ||||
| 75 | // OnTurnEnd, if set, fires once per prompt-level turn_end event | 77 | // OnTurnEnd, if set, fires once per prompt-level turn_end event | ||
| 76 | // emitted by the swarm daemon wrapper. Provider/tool-loop | 78 | // emitted by the swarm daemon wrapper. Provider/tool-loop | ||
M packages/agent/swarm/event.go +38 -0
| @@ -154,6 +154,44 @@ func ReadEventLog(path string) ([]Event, error) { | |||
| 154 | 154 | return readEvents(f) | |
| 155 | 155 | } | |
| 156 | 156 | ||
| 157 | + | // ReadEventLogTail parses only the last maxBytes of the log. | |
| 158 | + | // A partial first line is skipped; a complete line on the boundary is kept. | |
| 159 | + | func ReadEventLogTail(path string, maxBytes int64) ([]Event, error) { | |
| 160 | + | f, err := os.Open(path) | |
| 161 | + | if err != nil { | |
| 162 | + | if os.IsNotExist(err) { | |
| 163 | + | return nil, nil | |
| 164 | + | } | |
| 165 | + | return nil, err | |
| 166 | + | } | |
| 167 | + | defer f.Close() | |
| 168 | + | fi, err := f.Stat() | |
| 169 | + | if err != nil { | |
| 170 | + | return nil, err | |
| 171 | + | } | |
| 172 | + | if maxBytes <= 0 || fi.Size() <= maxBytes { | |
| 173 | + | return readEvents(f) | |
| 174 | + | } | |
| 175 | + | start := fi.Size() - maxBytes | |
| 176 | + | var prev [1]byte | |
| 177 | + | if _, err := f.ReadAt(prev[:], start-1); err != nil { | |
| 178 | + | return nil, err | |
| 179 | + | } | |
| 180 | + | if _, err := f.Seek(start, io.SeekStart); err != nil { | |
| 181 | + | return nil, err | |
| 182 | + | } | |
| 183 | + | br := bufio.NewReader(f) | |
| 184 | + | if prev[0] != '\n' { | |
| 185 | + | if _, err := br.ReadBytes('\n'); err != nil { | |
| 186 | + | if errors.Is(err, io.EOF) { | |
| 187 | + | return nil, nil | |
| 188 | + | } | |
| 189 | + | return nil, err | |
| 190 | + | } | |
| 191 | + | } | |
| 192 | + | return readEvents(br) | |
| 193 | + | } | |
| 194 | + | ||
| 157 | 195 | func readEvents(r io.Reader) ([]Event, error) { | |
| 158 | 196 | br := bufio.NewReader(r) | |
| 159 | 197 | var out []Event | |
| @@ -154,6 +154,44 @@ func ReadEventLog(path string) ([]Event, error) { | |||||
| 154 | return readEvents(f) | 154 | return readEvents(f) | ||
| 155 | } | 155 | } | ||
| 156 | 156 | ||||
| 157 | + | // ReadEventLogTail parses only the last maxBytes of the log. | |||
| 158 | + | // A partial first line is skipped; a complete line on the boundary is kept. | |||
| 159 | + | func ReadEventLogTail(path string, maxBytes int64) ([]Event, error) { | |||
| 160 | + | f, err := os.Open(path) | |||
| 161 | + | if err != nil { | |||
| 162 | + | if os.IsNotExist(err) { | |||
| 163 | + | return nil, nil | |||
| 164 | + | } | |||
| 165 | + | return nil, err | |||
| 166 | + | } | |||
| 167 | + | defer f.Close() | |||
| 168 | + | fi, err := f.Stat() | |||
| 169 | + | if err != nil { | |||
| 170 | + | return nil, err | |||
| 171 | + | } | |||
| 172 | + | if maxBytes <= 0 || fi.Size() <= maxBytes { | |||
| 173 | + | return readEvents(f) | |||
| 174 | + | } | |||
| 175 | + | start := fi.Size() - maxBytes | |||
| 176 | + | var prev [1]byte | |||
| 177 | + | if _, err := f.ReadAt(prev[:], start-1); err != nil { | |||
| 178 | + | return nil, err | |||
| 179 | + | } | |||
| 180 | + | if _, err := f.Seek(start, io.SeekStart); err != nil { | |||
| 181 | + | return nil, err | |||
| 182 | + | } | |||
| 183 | + | br := bufio.NewReader(f) | |||
| 184 | + | if prev[0] != '\n' { | |||
| 185 | + | if _, err := br.ReadBytes('\n'); err != nil { | |||
| 186 | + | if errors.Is(err, io.EOF) { | |||
| 187 | + | return nil, nil | |||
| 188 | + | } | |||
| 189 | + | return nil, err | |||
| 190 | + | } | |||
| 191 | + | } | |||
| 192 | + | return readEvents(br) | |||
| 193 | + | } | |||
| 194 | + | ||||
| 157 | func readEvents(r io.Reader) ([]Event, error) { | 195 | func readEvents(r io.Reader) ([]Event, error) { | ||
| 158 | br := bufio.NewReader(r) | 196 | br := bufio.NewReader(r) | ||
| 159 | var out []Event | 197 | var out []Event | ||
A packages/agent/swarm/event_tail_test.go +121 -0
| @@ -0,0 +1,121 @@ | |||
| 1 | + | package swarm | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "bytes" | |
| 5 | + | "fmt" | |
| 6 | + | "os" | |
| 7 | + | "path/filepath" | |
| 8 | + | "testing" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | // writeTailLog writes n events "e0".."e<n-1>" and returns the path and | |
| 12 | + | // the file size. | |
| 13 | + | func writeTailLog(t *testing.T, n int) (string, int64) { | |
| 14 | + | t.Helper() | |
| 15 | + | path := filepath.Join(t.TempDir(), "events.jsonl") | |
| 16 | + | log, err := OpenEventLog(path) | |
| 17 | + | if err != nil { | |
| 18 | + | t.Fatalf("open: %v", err) | |
| 19 | + | } | |
| 20 | + | for i := 0; i < n; i++ { | |
| 21 | + | if err := log.Append(NewEvent(fmt.Sprintf("e%d", i), map[string]any{"pad": "0123456789"})); err != nil { | |
| 22 | + | t.Fatalf("append: %v", err) | |
| 23 | + | } | |
| 24 | + | } | |
| 25 | + | _ = log.Close() | |
| 26 | + | info, err := os.Stat(path) | |
| 27 | + | if err != nil { | |
| 28 | + | t.Fatalf("stat: %v", err) | |
| 29 | + | } | |
| 30 | + | return path, info.Size() | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | // lineStarts returns the byte offset of every line in the file. Event | |
| 34 | + | // lines differ in length (timestamps drop trailing zeros), so windows | |
| 35 | + | // are computed from real offsets rather than an average line size. | |
| 36 | + | func lineStarts(t *testing.T, path string) []int64 { | |
| 37 | + | t.Helper() | |
| 38 | + | data, err := os.ReadFile(path) | |
| 39 | + | if err != nil { | |
| 40 | + | t.Fatalf("read: %v", err) | |
| 41 | + | } | |
| 42 | + | starts := []int64{0} | |
| 43 | + | for i, b := range data[:len(data)-1] { | |
| 44 | + | if b == '\n' { | |
| 45 | + | starts = append(starts, int64(i+1)) | |
| 46 | + | } | |
| 47 | + | } | |
| 48 | + | if !bytes.HasSuffix(data, []byte("\n")) { | |
| 49 | + | t.Fatalf("log does not end with a newline") | |
| 50 | + | } | |
| 51 | + | return starts | |
| 52 | + | } | |
| 53 | + | ||
| 54 | + | func eventTypes(evs []Event) []string { | |
| 55 | + | out := make([]string, len(evs)) | |
| 56 | + | for i, ev := range evs { | |
| 57 | + | out[i] = ev.Type | |
| 58 | + | } | |
| 59 | + | return out | |
| 60 | + | } | |
| 61 | + | ||
| 62 | + | func TestReadEventLogTailSmallFileReadsEverything(t *testing.T) { | |
| 63 | + | path, size := writeTailLog(t, 5) | |
| 64 | + | for _, limit := range []int64{0, size, size + 100} { | |
| 65 | + | got, err := ReadEventLogTail(path, limit) | |
| 66 | + | if err != nil { | |
| 67 | + | t.Fatalf("limit %d: %v", limit, err) | |
| 68 | + | } | |
| 69 | + | if len(got) != 5 || got[0].Type != "e0" || got[4].Type != "e4" { | |
| 70 | + | t.Fatalf("limit %d: got %v", limit, eventTypes(got)) | |
| 71 | + | } | |
| 72 | + | } | |
| 73 | + | } | |
| 74 | + | ||
| 75 | + | // The tail window usually starts mid-line; that partial line must be | |
| 76 | + | // dropped rather than parsed, and every later event returned in order. | |
| 77 | + | func TestReadEventLogTailSkipsPartialFirstLine(t *testing.T) { | |
| 78 | + | path, size := writeTailLog(t, 50) | |
| 79 | + | starts := lineStarts(t, path) | |
| 80 | + | // Start the window in the middle of line 46: e46 is partial. | |
| 81 | + | mid := (starts[46] + starts[47]) / 2 | |
| 82 | + | got, err := ReadEventLogTail(path, size-mid) | |
| 83 | + | if err != nil { | |
| 84 | + | t.Fatalf("read: %v", err) | |
| 85 | + | } | |
| 86 | + | want := []string{"e47", "e48", "e49"} | |
| 87 | + | if fmt.Sprint(eventTypes(got)) != fmt.Sprint(want) { | |
| 88 | + | t.Fatalf("got %v, want %v", eventTypes(got), want) | |
| 89 | + | } | |
| 90 | + | } | |
| 91 | + | ||
| 92 | + | // A window starting on a line boundary must keep its first event. | |
| 93 | + | // In particular, that event might be the only lifecycle marker in the tail. | |
| 94 | + | func TestReadEventLogTailBoundaryWindow(t *testing.T) { | |
| 95 | + | path, size := writeTailLog(t, 10) | |
| 96 | + | starts := lineStarts(t, path) | |
| 97 | + | // The window starts exactly at the beginning of e6. | |
| 98 | + | got, err := ReadEventLogTail(path, size-starts[6]) | |
| 99 | + | if err != nil { | |
| 100 | + | t.Fatalf("read: %v", err) | |
| 101 | + | } | |
| 102 | + | want := []string{"e6", "e7", "e8", "e9"} | |
| 103 | + | if fmt.Sprint(eventTypes(got)) != fmt.Sprint(want) { | |
| 104 | + | t.Fatalf("got %v, want %v", eventTypes(got), want) | |
| 105 | + | } | |
| 106 | + | } | |
| 107 | + | ||
| 108 | + | func TestReadEventLogTailWindowWithoutNewline(t *testing.T) { | |
| 109 | + | path, _ := writeTailLog(t, 10) | |
| 110 | + | got, err := ReadEventLogTail(path, 5) // inside the last line | |
| 111 | + | if err != nil || len(got) != 0 { | |
| 112 | + | t.Fatalf("got %v, %v; want no events and no error", eventTypes(got), err) | |
| 113 | + | } | |
| 114 | + | } | |
| 115 | + | ||
| 116 | + | func TestReadEventLogTailMissingFile(t *testing.T) { | |
| 117 | + | got, err := ReadEventLogTail(filepath.Join(t.TempDir(), "missing.jsonl"), 1024) | |
| 118 | + | if err != nil || got != nil { | |
| 119 | + | t.Fatalf("got %v, %v; want nil, nil", got, err) | |
| 120 | + | } | |
| 121 | + | } | |
| @@ -0,0 +1,121 @@ | |||||
| 1 | + | package swarm | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "bytes" | |||
| 5 | + | "fmt" | |||
| 6 | + | "os" | |||
| 7 | + | "path/filepath" | |||
| 8 | + | "testing" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | // writeTailLog writes n events "e0".."e<n-1>" and returns the path and | |||
| 12 | + | // the file size. | |||
| 13 | + | func writeTailLog(t *testing.T, n int) (string, int64) { | |||
| 14 | + | t.Helper() | |||
| 15 | + | path := filepath.Join(t.TempDir(), "events.jsonl") | |||
| 16 | + | log, err := OpenEventLog(path) | |||
| 17 | + | if err != nil { | |||
| 18 | + | t.Fatalf("open: %v", err) | |||
| 19 | + | } | |||
| 20 | + | for i := 0; i < n; i++ { | |||
| 21 | + | if err := log.Append(NewEvent(fmt.Sprintf("e%d", i), map[string]any{"pad": "0123456789"})); err != nil { | |||
| 22 | + | t.Fatalf("append: %v", err) | |||
| 23 | + | } | |||
| 24 | + | } | |||
| 25 | + | _ = log.Close() | |||
| 26 | + | info, err := os.Stat(path) | |||
| 27 | + | if err != nil { | |||
| 28 | + | t.Fatalf("stat: %v", err) | |||
| 29 | + | } | |||
| 30 | + | return path, info.Size() | |||
| 31 | + | } | |||
| 32 | + | ||||
| 33 | + | // lineStarts returns the byte offset of every line in the file. Event | |||
| 34 | + | // lines differ in length (timestamps drop trailing zeros), so windows | |||
| 35 | + | // are computed from real offsets rather than an average line size. | |||
| 36 | + | func lineStarts(t *testing.T, path string) []int64 { | |||
| 37 | + | t.Helper() | |||
| 38 | + | data, err := os.ReadFile(path) | |||
| 39 | + | if err != nil { | |||
| 40 | + | t.Fatalf("read: %v", err) | |||
| 41 | + | } | |||
| 42 | + | starts := []int64{0} | |||
| 43 | + | for i, b := range data[:len(data)-1] { | |||
| 44 | + | if b == '\n' { | |||
| 45 | + | starts = append(starts, int64(i+1)) | |||
| 46 | + | } | |||
| 47 | + | } | |||
| 48 | + | if !bytes.HasSuffix(data, []byte("\n")) { | |||
| 49 | + | t.Fatalf("log does not end with a newline") | |||
| 50 | + | } | |||
| 51 | + | return starts | |||
| 52 | + | } | |||
| 53 | + | ||||
| 54 | + | func eventTypes(evs []Event) []string { | |||
| 55 | + | out := make([]string, len(evs)) | |||
| 56 | + | for i, ev := range evs { | |||
| 57 | + | out[i] = ev.Type | |||
| 58 | + | } | |||
| 59 | + | return out | |||
| 60 | + | } | |||
| 61 | + | ||||
| 62 | + | func TestReadEventLogTailSmallFileReadsEverything(t *testing.T) { | |||
| 63 | + | path, size := writeTailLog(t, 5) | |||
| 64 | + | for _, limit := range []int64{0, size, size + 100} { | |||
| 65 | + | got, err := ReadEventLogTail(path, limit) | |||
| 66 | + | if err != nil { | |||
| 67 | + | t.Fatalf("limit %d: %v", limit, err) | |||
| 68 | + | } | |||
| 69 | + | if len(got) != 5 || got[0].Type != "e0" || got[4].Type != "e4" { | |||
| 70 | + | t.Fatalf("limit %d: got %v", limit, eventTypes(got)) | |||
| 71 | + | } | |||
| 72 | + | } | |||
| 73 | + | } | |||
| 74 | + | ||||
| 75 | + | // The tail window usually starts mid-line; that partial line must be | |||
| 76 | + | // dropped rather than parsed, and every later event returned in order. | |||
| 77 | + | func TestReadEventLogTailSkipsPartialFirstLine(t *testing.T) { | |||
| 78 | + | path, size := writeTailLog(t, 50) | |||
| 79 | + | starts := lineStarts(t, path) | |||
| 80 | + | // Start the window in the middle of line 46: e46 is partial. | |||
| 81 | + | mid := (starts[46] + starts[47]) / 2 | |||
| 82 | + | got, err := ReadEventLogTail(path, size-mid) | |||
| 83 | + | if err != nil { | |||
| 84 | + | t.Fatalf("read: %v", err) | |||
| 85 | + | } | |||
| 86 | + | want := []string{"e47", "e48", "e49"} | |||
| 87 | + | if fmt.Sprint(eventTypes(got)) != fmt.Sprint(want) { | |||
| 88 | + | t.Fatalf("got %v, want %v", eventTypes(got), want) | |||
| 89 | + | } | |||
| 90 | + | } | |||
| 91 | + | ||||
| 92 | + | // A window starting on a line boundary must keep its first event. | |||
| 93 | + | // In particular, that event might be the only lifecycle marker in the tail. | |||
| 94 | + | func TestReadEventLogTailBoundaryWindow(t *testing.T) { | |||
| 95 | + | path, size := writeTailLog(t, 10) | |||
| 96 | + | starts := lineStarts(t, path) | |||
| 97 | + | // The window starts exactly at the beginning of e6. | |||
| 98 | + | got, err := ReadEventLogTail(path, size-starts[6]) | |||
| 99 | + | if err != nil { | |||
| 100 | + | t.Fatalf("read: %v", err) | |||
| 101 | + | } | |||
| 102 | + | want := []string{"e6", "e7", "e8", "e9"} | |||
| 103 | + | if fmt.Sprint(eventTypes(got)) != fmt.Sprint(want) { | |||
| 104 | + | t.Fatalf("got %v, want %v", eventTypes(got), want) | |||
| 105 | + | } | |||
| 106 | + | } | |||
| 107 | + | ||||
| 108 | + | func TestReadEventLogTailWindowWithoutNewline(t *testing.T) { | |||
| 109 | + | path, _ := writeTailLog(t, 10) | |||
| 110 | + | got, err := ReadEventLogTail(path, 5) // inside the last line | |||
| 111 | + | if err != nil || len(got) != 0 { | |||
| 112 | + | t.Fatalf("got %v, %v; want no events and no error", eventTypes(got), err) | |||
| 113 | + | } | |||
| 114 | + | } | |||
| 115 | + | ||||
| 116 | + | func TestReadEventLogTailMissingFile(t *testing.T) { | |||
| 117 | + | got, err := ReadEventLogTail(filepath.Join(t.TempDir(), "missing.jsonl"), 1024) | |||
| 118 | + | if err != nil || got != nil { | |||
| 119 | + | t.Fatalf("got %v, %v; want nil, nil", got, err) | |||
| 120 | + | } | |||
| 121 | + | } | |||
M packages/agent/swarm/persist.go +29 -1
| @@ -180,6 +180,10 @@ func (f *Swarm) Reload() (loaded int, errs []error) { | |||
| 180 | 180 | return loaded, errs | |
| 181 | 181 | } | |
| 182 | 182 | ||
| 183 | + | // detachedReplayBytes bounds startup replay; full history is loaded | |
| 184 | + | // when the dashboard is opened or the agent is resumed. | |
| 185 | + | const detachedReplayBytes = 4 << 20 | |
| 186 | + | ||
| 183 | 187 | // buildDetachedAgent constructs an Agent from a meta.json with no | |
| 184 | 188 | // running Runner. The agent's transcript is populated from the tail | |
| 185 | 189 | // of its event log so the dashboard immediately shows recent output; | |
| @@ -223,13 +227,36 @@ func (f *Swarm) buildDetachedAgent(m agentMeta) *Agent { | |||
| 223 | 227 | // effort: a missing or unreadable log just leaves the agent | |
| 224 | 228 | // detached with an empty transcript. | |
| 225 | 229 | if a.EventLogPath != "" { | |
| 226 | - | if evs, err := ReadEventLog(a.EventLogPath); err == nil { | |
| 230 | + | if evs, err := ReadEventLogTail(a.EventLogPath, detachedReplayBytes); err == nil { | |
| 227 | 231 | replayEventsIntoAgent(a, evs) | |
| 228 | 232 | } | |
| 233 | + | if fi, err := os.Stat(a.EventLogPath); err == nil && fi.Size() > detachedReplayBytes { | |
| 234 | + | a.needsFullReplay = true | |
| 235 | + | } | |
| 229 | 236 | } | |
| 230 | 237 | return a | |
| 231 | 238 | } | |
| 232 | 239 | ||
| 240 | + | // loadFullTranscript backfills a detached agent on demand. Keep the | |
| 241 | + | // status inferred from the newest lifecycle event in the startup tail. | |
| 242 | + | // The agent's mutex serializes concurrent dashboard refreshes and resume. | |
| 243 | + | func (a *Agent) loadFullTranscript() { | |
| 244 | + | a.mu.Lock() | |
| 245 | + | defer a.mu.Unlock() | |
| 246 | + | if !a.needsFullReplay { | |
| 247 | + | return | |
| 248 | + | } | |
| 249 | + | evs, err := ReadEventLog(a.EventLogPath) | |
| 250 | + | if err != nil { | |
| 251 | + | return | |
| 252 | + | } | |
| 253 | + | full := &Agent{status: StatusDetached, activity: "detached"} | |
| 254 | + | replayEventsIntoAgent(full, evs) | |
| 255 | + | a.transcript = full.transcript | |
| 256 | + | a.lastAssistant = full.lastAssistant | |
| 257 | + | a.needsFullReplay = false | |
| 258 | + | } | |
| 259 | + | ||
| 233 | 260 | // replayEventsIntoAgent re-derives an agent's transcript and last | |
| 234 | 261 | // known status hint from its event log. Mirrors applyEventToSink in | |
| 235 | 262 | // runner.go but writes directly to the Agent fields (no Sink because | |
| @@ -333,6 +360,7 @@ func (f *Swarm) Resume(ctx context.Context, id string) (*Agent, error) { | |||
| 333 | 360 | if st == StatusRunning || st == StatusPending { | |
| 334 | 361 | return nil, fmt.Errorf("swarm: agent %s is still %s; stop it first", existing.ID, st) | |
| 335 | 362 | } | |
| 363 | + | existing.loadFullTranscript() | |
| 336 | 364 | ||
| 337 | 365 | // Rebuild from the meta record so we don't carry stale runner | |
| 338 | 366 | // state from a previous incarnation. The inbox is transient and | |
| @@ -180,6 +180,10 @@ func (f *Swarm) Reload() (loaded int, errs []error) { | |||||
| 180 | return loaded, errs | 180 | return loaded, errs | ||
| 181 | } | 181 | } | ||
| 182 | 182 | ||||
| 183 | + | // detachedReplayBytes bounds startup replay; full history is loaded | |||
| 184 | + | // when the dashboard is opened or the agent is resumed. | |||
| 185 | + | const detachedReplayBytes = 4 << 20 | |||
| 186 | + | ||||
| 183 | // buildDetachedAgent constructs an Agent from a meta.json with no | 187 | // buildDetachedAgent constructs an Agent from a meta.json with no | ||
| 184 | // running Runner. The agent's transcript is populated from the tail | 188 | // running Runner. The agent's transcript is populated from the tail | ||
| 185 | // of its event log so the dashboard immediately shows recent output; | 189 | // of its event log so the dashboard immediately shows recent output; | ||
| @@ -223,13 +227,36 @@ func (f *Swarm) buildDetachedAgent(m agentMeta) *Agent { | |||||
| 223 | // effort: a missing or unreadable log just leaves the agent | 227 | // effort: a missing or unreadable log just leaves the agent | ||
| 224 | // detached with an empty transcript. | 228 | // detached with an empty transcript. | ||
| 225 | if a.EventLogPath != "" { | 229 | if a.EventLogPath != "" { | ||
| 226 | - | if evs, err := ReadEventLog(a.EventLogPath); err == nil { | 230 | + | if evs, err := ReadEventLogTail(a.EventLogPath, detachedReplayBytes); err == nil { |
| 227 | replayEventsIntoAgent(a, evs) | 231 | replayEventsIntoAgent(a, evs) | ||
| 228 | } | 232 | } | ||
| 233 | + | if fi, err := os.Stat(a.EventLogPath); err == nil && fi.Size() > detachedReplayBytes { | |||
| 234 | + | a.needsFullReplay = true | |||
| 235 | + | } | |||
| 229 | } | 236 | } | ||
| 230 | return a | 237 | return a | ||
| 231 | } | 238 | } | ||
| 232 | 239 | ||||
| 240 | + | // loadFullTranscript backfills a detached agent on demand. Keep the | |||
| 241 | + | // status inferred from the newest lifecycle event in the startup tail. | |||
| 242 | + | // The agent's mutex serializes concurrent dashboard refreshes and resume. | |||
| 243 | + | func (a *Agent) loadFullTranscript() { | |||
| 244 | + | a.mu.Lock() | |||
| 245 | + | defer a.mu.Unlock() | |||
| 246 | + | if !a.needsFullReplay { | |||
| 247 | + | return | |||
| 248 | + | } | |||
| 249 | + | evs, err := ReadEventLog(a.EventLogPath) | |||
| 250 | + | if err != nil { | |||
| 251 | + | return | |||
| 252 | + | } | |||
| 253 | + | full := &Agent{status: StatusDetached, activity: "detached"} | |||
| 254 | + | replayEventsIntoAgent(full, evs) | |||
| 255 | + | a.transcript = full.transcript | |||
| 256 | + | a.lastAssistant = full.lastAssistant | |||
| 257 | + | a.needsFullReplay = false | |||
| 258 | + | } | |||
| 259 | + | ||||
| 233 | // replayEventsIntoAgent re-derives an agent's transcript and last | 260 | // replayEventsIntoAgent re-derives an agent's transcript and last | ||
| 234 | // known status hint from its event log. Mirrors applyEventToSink in | 261 | // known status hint from its event log. Mirrors applyEventToSink in | ||
| 235 | // runner.go but writes directly to the Agent fields (no Sink because | 262 | // runner.go but writes directly to the Agent fields (no Sink because | ||
| @@ -333,6 +360,7 @@ func (f *Swarm) Resume(ctx context.Context, id string) (*Agent, error) { | |||||
| 333 | if st == StatusRunning || st == StatusPending { | 360 | if st == StatusRunning || st == StatusPending { | ||
| 334 | return nil, fmt.Errorf("swarm: agent %s is still %s; stop it first", existing.ID, st) | 361 | return nil, fmt.Errorf("swarm: agent %s is still %s; stop it first", existing.ID, st) | ||
| 335 | } | 362 | } | ||
| 363 | + | existing.loadFullTranscript() | |||
| 336 | 364 | ||||
| 337 | // Rebuild from the meta record so we don't carry stale runner | 365 | // Rebuild from the meta record so we don't carry stale runner | ||
| 338 | // state from a previous incarnation. The inbox is transient and | 366 | // state from a previous incarnation. The inbox is transient and | ||
M packages/agent/swarm/persist_test.go +55 -0
| @@ -221,6 +221,61 @@ func TestReloadReplaysTranscriptFromEventLog(t *testing.T) { | |||
| 221 | 221 | } | |
| 222 | 222 | } | |
| 223 | 223 | ||
| 224 | + | func TestDetachedReplayKeepsLastAnswerAcrossLargeEvents(t *testing.T) { | |
| 225 | + | root := t.TempDir() | |
| 226 | + | path := filepath.Join(root, "events.jsonl") | |
| 227 | + | log, err := OpenEventLog(path) | |
| 228 | + | if err != nil { | |
| 229 | + | t.Fatal(err) | |
| 230 | + | } | |
| 231 | + | for _, ev := range []Event{ | |
| 232 | + | NewEvent("assistant_message", map[string]any{ | |
| 233 | + | "content": []any{map[string]any{"type": "text", "text": "final answer"}}, | |
| 234 | + | }), | |
| 235 | + | NewEvent("tool_result", map[string]any{"data": strings.Repeat("x", detachedReplayBytes)}), | |
| 236 | + | NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": 0}), | |
| 237 | + | } { | |
| 238 | + | if err := log.Append(ev); err != nil { | |
| 239 | + | t.Fatal(err) | |
| 240 | + | } | |
| 241 | + | } | |
| 242 | + | if err := log.Close(); err != nil { | |
| 243 | + | t.Fatal(err) | |
| 244 | + | } | |
| 245 | + | ||
| 246 | + | f := New(Config{Root: root, RepoRoot: root}) | |
| 247 | + | a := f.buildDetachedAgent(agentMeta{ID: "large", EventLogPath: path}) | |
| 248 | + | if got := a.Snapshot(); got.LastAssistant != "" || len(got.Lines) != 0 || got.Status != StatusDone { | |
| 249 | + | t.Fatalf("startup should use a bounded tail: assistant=%q, lines=%v, status=%s", got.LastAssistant, got.Lines, got.Status) | |
| 250 | + | } | |
| 251 | + | f.agents[a.ID] = a | |
| 252 | + | f.order = append(f.order, a.ID) | |
| 253 | + | got := f.SnapshotAll() | |
| 254 | + | if len(got) != 1 || got[0].LastAssistant != "final answer" || | |
| 255 | + | len(got[0].Lines) != 1 || got[0].Lines[0] != "final answer" || got[0].Status != StatusDone { | |
| 256 | + | t.Fatalf("dashboard snapshot: %v", got) | |
| 257 | + | } | |
| 258 | + | ||
| 259 | + | // Resume must backfill even when the dashboard was never opened. | |
| 260 | + | f2 := New(Config{Root: root, RepoRoot: root, NewRunner: func(*Agent) Runner { | |
| 261 | + | return RunnerFunc(func(ctx context.Context, _ Sink) error { | |
| 262 | + | <-ctx.Done() | |
| 263 | + | return ctx.Err() | |
| 264 | + | }) | |
| 265 | + | }}) | |
| 266 | + | defer f2.StopAll() | |
| 267 | + | b := f2.buildDetachedAgent(agentMeta{ID: "large", EventLogPath: path}) | |
| 268 | + | f2.agents[b.ID] = b | |
| 269 | + | f2.order = append(f2.order, b.ID) | |
| 270 | + | resumed, err := f2.Resume(context.Background(), b.ID) | |
| 271 | + | if err != nil { | |
| 272 | + | t.Fatal(err) | |
| 273 | + | } | |
| 274 | + | if lines := resumed.Transcript(); len(lines) != 1 || lines[0] != "final answer" { | |
| 275 | + | t.Fatalf("resumed transcript: %v", lines) | |
| 276 | + | } | |
| 277 | + | } | |
| 278 | + | ||
| 224 | 279 | // TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json | |
| 225 | 280 | // doesn't blow up the whole reload. Directories with no meta.json | |
| 226 | 281 | // at all are silently ignored (Spawn that failed mid-way leaves | |
| @@ -221,6 +221,61 @@ func TestReloadReplaysTranscriptFromEventLog(t *testing.T) { | |||||
| 221 | } | 221 | } | ||
| 222 | } | 222 | } | ||
| 223 | 223 | ||||
| 224 | + | func TestDetachedReplayKeepsLastAnswerAcrossLargeEvents(t *testing.T) { | |||
| 225 | + | root := t.TempDir() | |||
| 226 | + | path := filepath.Join(root, "events.jsonl") | |||
| 227 | + | log, err := OpenEventLog(path) | |||
| 228 | + | if err != nil { | |||
| 229 | + | t.Fatal(err) | |||
| 230 | + | } | |||
| 231 | + | for _, ev := range []Event{ | |||
| 232 | + | NewEvent("assistant_message", map[string]any{ | |||
| 233 | + | "content": []any{map[string]any{"type": "text", "text": "final answer"}}, | |||
| 234 | + | }), | |||
| 235 | + | NewEvent("tool_result", map[string]any{"data": strings.Repeat("x", detachedReplayBytes)}), | |||
| 236 | + | NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": 0}), | |||
| 237 | + | } { | |||
| 238 | + | if err := log.Append(ev); err != nil { | |||
| 239 | + | t.Fatal(err) | |||
| 240 | + | } | |||
| 241 | + | } | |||
| 242 | + | if err := log.Close(); err != nil { | |||
| 243 | + | t.Fatal(err) | |||
| 244 | + | } | |||
| 245 | + | ||||
| 246 | + | f := New(Config{Root: root, RepoRoot: root}) | |||
| 247 | + | a := f.buildDetachedAgent(agentMeta{ID: "large", EventLogPath: path}) | |||
| 248 | + | if got := a.Snapshot(); got.LastAssistant != "" || len(got.Lines) != 0 || got.Status != StatusDone { | |||
| 249 | + | t.Fatalf("startup should use a bounded tail: assistant=%q, lines=%v, status=%s", got.LastAssistant, got.Lines, got.Status) | |||
| 250 | + | } | |||
| 251 | + | f.agents[a.ID] = a | |||
| 252 | + | f.order = append(f.order, a.ID) | |||
| 253 | + | got := f.SnapshotAll() | |||
| 254 | + | if len(got) != 1 || got[0].LastAssistant != "final answer" || | |||
| 255 | + | len(got[0].Lines) != 1 || got[0].Lines[0] != "final answer" || got[0].Status != StatusDone { | |||
| 256 | + | t.Fatalf("dashboard snapshot: %v", got) | |||
| 257 | + | } | |||
| 258 | + | ||||
| 259 | + | // Resume must backfill even when the dashboard was never opened. | |||
| 260 | + | f2 := New(Config{Root: root, RepoRoot: root, NewRunner: func(*Agent) Runner { | |||
| 261 | + | return RunnerFunc(func(ctx context.Context, _ Sink) error { | |||
| 262 | + | <-ctx.Done() | |||
| 263 | + | return ctx.Err() | |||
| 264 | + | }) | |||
| 265 | + | }}) | |||
| 266 | + | defer f2.StopAll() | |||
| 267 | + | b := f2.buildDetachedAgent(agentMeta{ID: "large", EventLogPath: path}) | |||
| 268 | + | f2.agents[b.ID] = b | |||
| 269 | + | f2.order = append(f2.order, b.ID) | |||
| 270 | + | resumed, err := f2.Resume(context.Background(), b.ID) | |||
| 271 | + | if err != nil { | |||
| 272 | + | t.Fatal(err) | |||
| 273 | + | } | |||
| 274 | + | if lines := resumed.Transcript(); len(lines) != 1 || lines[0] != "final answer" { | |||
| 275 | + | t.Fatalf("resumed transcript: %v", lines) | |||
| 276 | + | } | |||
| 277 | + | } | |||
| 278 | + | ||||
| 224 | // TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json | 279 | // TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json | ||
| 225 | // doesn't blow up the whole reload. Directories with no meta.json | 280 | // doesn't blow up the whole reload. Directories with no meta.json | ||
| 226 | // at all are silently ignored (Spawn that failed mid-way leaves | 281 | // at all are silently ignored (Spawn that failed mid-way leaves | ||
M packages/agent/swarm/swarm.go +3 -1
| @@ -540,7 +540,8 @@ func (a *Agent) Snapshot() AgentSnapshot { | |||
| 540 | 540 | } | |
| 541 | 541 | ||
| 542 | 542 | // SnapshotAll returns snapshots of every agent in creation order, | |
| 543 | - | // scoped to the active session when one is set. | |
| 543 | + | // scoped to the active session when one is set. Opening the dashboard | |
| 544 | + | // backfills large detached logs once, rather than delaying startup. | |
| 544 | 545 | // | |
| 545 | 546 | // Scoping rules: | |
| 546 | 547 | // - activeSession == "": no filter, every agent is returned | |
| @@ -561,6 +562,7 @@ func (f *Swarm) SnapshotAll() []AgentSnapshot { | |||
| 561 | 562 | if active != "" && a.SessionID != "" && a.SessionID != active { | |
| 562 | 563 | continue | |
| 563 | 564 | } | |
| 565 | + | a.loadFullTranscript() | |
| 564 | 566 | out = append(out, a.Snapshot()) | |
| 565 | 567 | } | |
| 566 | 568 | // Sort by start time for a stable, deterministic listing. | |
| @@ -540,7 +540,8 @@ func (a *Agent) Snapshot() AgentSnapshot { | |||||
| 540 | } | 540 | } | ||
| 541 | 541 | ||||
| 542 | // SnapshotAll returns snapshots of every agent in creation order, | 542 | // SnapshotAll returns snapshots of every agent in creation order, | ||
| 543 | - | // scoped to the active session when one is set. | 543 | + | // scoped to the active session when one is set. Opening the dashboard |
| 544 | + | // backfills large detached logs once, rather than delaying startup. | |||
| 544 | // | 545 | // | ||
| 545 | // Scoping rules: | 546 | // Scoping rules: | ||
| 546 | // - activeSession == "": no filter, every agent is returned | 547 | // - activeSession == "": no filter, every agent is returned | ||
| @@ -561,6 +562,7 @@ func (f *Swarm) SnapshotAll() []AgentSnapshot { | |||||
| 561 | if active != "" && a.SessionID != "" && a.SessionID != active { | 562 | if active != "" && a.SessionID != "" && a.SessionID != active { | ||
| 562 | continue | 563 | continue | ||
| 563 | } | 564 | } | ||
| 565 | + | a.loadFullTranscript() | |||
| 564 | out = append(out, a.Snapshot()) | 566 | out = append(out, a.Snapshot()) | ||
| 565 | } | 567 | } | ||
| 566 | // Sort by start time for a stable, deterministic listing. | 568 | // Sort by start time for a stable, deterministic listing. | ||