01a4224 Preserve detached swarm replies when replaying log tails · zut

M packages/agent/swarm/event.go +14 -9


@@ -154,10 +154,8 @@ 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. Startup
158		-	// replay of detached agents keeps a bounded transcript anyway, so
159		-	// reading multi-gigabyte logs in full only delays the first frame.
160		-	// The first, possibly partial, line of the window is skipped.
	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.
161	159		func ReadEventLogTail(path string, maxBytes int64) ([]Event, error) {
162	160		f, err := os.Open(path)
163	161		if err != nil {

@@ -174,15 +172,22 @@ func ReadEventLogTail(path string, maxBytes int64) ([]Event, error) {
174	172		if maxBytes <= 0 \|\| fi.Size() <= maxBytes {
175	173		return readEvents(f)
176	174		}
177		-	if _, err := f.Seek(fi.Size()-maxBytes, io.SeekStart); err != nil {
	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 {
178	181		return nil, err
179	182		}
180	183		br := bufio.NewReader(f)
181		-	if _, err := br.ReadBytes('\n'); err != nil {
182		-	if errors.Is(err, io.EOF) {
183		-	return nil, nil
	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
184	190		}
185		-	return nil, err
186	191		}
187	192		return readEvents(br)
188	193		}

 	return readEvents(f)
 }
-// ReadEventLogTail parses only the last maxBytes of the log. Startup
+// ReadEventLogTail parses only the last maxBytes of the log.
-// replay of detached agents keeps a bounded transcript anyway, so
+// A partial first line is skipped; a complete line on the boundary is kept.
-// reading multi-gigabyte logs in full only delays the first frame.
-// The first, possibly partial, line of the window is skipped.
 func ReadEventLogTail(path string, maxBytes int64) ([]Event, error) {
 	f, err := os.Open(path)
 	if err != nil {
 	if maxBytes <= 0 || fi.Size() <= maxBytes {
 		return readEvents(f)
 	}
-	if _, err := f.Seek(fi.Size()-maxBytes, io.SeekStart); err != nil {
+	start := fi.Size() - maxBytes
+	var prev [1]byte
+	if _, err := f.ReadAt(prev[:], start-1); err != nil {
+		return nil, err
+	}
+	if _, err := f.Seek(start, io.SeekStart); err != nil {
 		return nil, err
 	}
 	br := bufio.NewReader(f)
-	if _, err := br.ReadBytes('\n'); err != nil {
+	if prev[0] != '\n' {
-		if errors.Is(err, io.EOF) {
+		if _, err := br.ReadBytes('\n'); err != nil {
-			return nil, nil
+			if errors.Is(err, io.EOF) {
+				return nil, nil
+			}
+			return nil, err
 		}
-		return nil, err
 	}
 	return readEvents(br)
 }

M packages/agent/swarm/event_tail_test.go +3 -4


@@ -89,9 +89,8 @@ func TestReadEventLogTailSkipsPartialFirstLine(t *testing.T) {
89	89		}
90	90		}
91	91
92		-	// A window that falls exactly on a line boundary still drops its first
93		-	// line: it cannot tell a boundary from a partial line, and losing one
94		-	// event of a bounded transcript is harmless.
	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.
95	94		func TestReadEventLogTailBoundaryWindow(t *testing.T) {
96	95		path, size := writeTailLog(t, 10)
97	96		starts := lineStarts(t, path)

@@ -100,7 +99,7 @@ func TestReadEventLogTailBoundaryWindow(t *testing.T) {
100	99		if err != nil {
101	100		t.Fatalf("read: %v", err)
102	101		}
103		-	want := []string{"e7", "e8", "e9"}
	102	+	want := []string{"e6", "e7", "e8", "e9"}
104	103		if fmt.Sprint(eventTypes(got)) != fmt.Sprint(want) {
105	104		t.Fatalf("got %v, want %v", eventTypes(got), want)
106	105		}

 	}
 }
-// A window that falls exactly on a line boundary still drops its first
+// A window starting on a line boundary must keep its first event.
-// line: it cannot tell a boundary from a partial line, and losing one
+// In particular, that event might be the only lifecycle marker in the tail.
-// event of a bounded transcript is harmless.
 func TestReadEventLogTailBoundaryWindow(t *testing.T) {
 	path, size := writeTailLog(t, 10)
 	starts := lineStarts(t, path)
 	if err != nil {
 		t.Fatalf("read: %v", err)
 	}
-	want := []string{"e7", "e8", "e9"}
+	want := []string{"e6", "e7", "e8", "e9"}
 	if fmt.Sprint(eventTypes(got)) != fmt.Sprint(want) {
 		t.Fatalf("got %v, want %v", eventTypes(got), want)
 	}

M packages/agent/swarm/persist.go +36 -6


@@ -180,10 +180,8 @@ func (f *Swarm) Reload() (loaded int, errs []error) {
180	180		return loaded, errs
181	181		}
182	182
183		-	// detachedReplayBytes bounds how much of each event log is replayed for
184		-	// a detached agent. The transcript keeps at most 2000 lines and the
185		-	// terminal lifecycle event sits at the end of the log, so the tail is
186		-	// enough; full logs are still read on demand by the dashboard.
	183	+	// detachedReplayBytes is the initial replay window. Expand it when a
	184	+	// sparse transcript or a large event leaves relevant history outside it.
187	185		const detachedReplayBytes = 4 << 20
188	186
189	187		// buildDetachedAgent constructs an Agent from a meta.json with no

@@ -229,11 +227,43 @@ func (f Swarm) buildDetachedAgent(m agentMeta) Agent {
229	227		// effort: a missing or unreadable log just leaves the agent
230	228		// detached with an empty transcript.
231	229		if a.EventLogPath != "" {
232		-	if evs, err := ReadEventLogTail(a.EventLogPath, detachedReplayBytes); err == nil {
	230	+	replayDetachedEventLog(a)
	231	+	}
	232	+	return a
	233	+	}
	234	+
	235	+	// replayDetachedEventLog starts at the end of the file, expanding the
	236	+	// window only when the bounded transcript or latest assistant reply may
	237	+	// still depend on earlier events. In sparse logs, preserving those fields
	238	+	// can require reading the entire file.
	239	+	func replayDetachedEventLog(a *Agent) {
	240	+	fi, err := os.Stat(a.EventLogPath)
	241	+	if err != nil {
	242	+	return
	243	+	}
	244	+	size := fi.Size()
	245	+	limit := int64(detachedReplayBytes)
	246	+	for {
	247	+	evs, err := ReadEventLogTail(a.EventLogPath, limit)
	248	+	if err != nil {
	249	+	return
	250	+	}
	251	+	if limit >= size {
	252	+	replayEventsIntoAgent(a, evs)
	253	+	return
	254	+	}
	255	+	trial := &Agent{status: StatusDetached, activity: "detached"}
	256	+	replayEventsIntoAgent(trial, evs)
	257	+	if len(trial.transcript) == 2000 && trial.lastAssistant != "" {
233	258		replayEventsIntoAgent(a, evs)
	259	+	return
	260	+	}
	261	+	if limit >= size/2 {
	262	+	limit = size
	263	+	} else {
	264	+	limit *= 2
234	265		}
235	266		}
236		-	return a
237	267		}
238	268
239	269		// replayEventsIntoAgent re-derives an agent's transcript and last


@@ -180,10 +180,8 @@ func (f *Swarm) Reload() (loaded int, errs []error) {
180		return loaded, errs	180		return loaded, errs
181		}	181		}
182			182
183	-	// detachedReplayBytes bounds how much of each event log is replayed for	183	+	// detachedReplayBytes is the initial replay window. Expand it when a
184	-	// a detached agent. The transcript keeps at most 2000 lines and the	184	+	// sparse transcript or a large event leaves relevant history outside it.
185	-	// terminal lifecycle event sits at the end of the log, so the tail is
186	-	// enough; full logs are still read on demand by the dashboard.
187		const detachedReplayBytes = 4 << 20	185		const detachedReplayBytes = 4 << 20
188			186
189		// buildDetachedAgent constructs an Agent from a meta.json with no	187		// buildDetachedAgent constructs an Agent from a meta.json with no

@@ -229,11 +227,43 @@ func (f Swarm) buildDetachedAgent(m agentMeta) Agent {
229		// effort: a missing or unreadable log just leaves the agent	227		// effort: a missing or unreadable log just leaves the agent
230		// detached with an empty transcript.	228		// detached with an empty transcript.
231		if a.EventLogPath != "" {	229		if a.EventLogPath != "" {
232	-	if evs, err := ReadEventLogTail(a.EventLogPath, detachedReplayBytes); err == nil {	230	+	replayDetachedEventLog(a)
			231	+	}
			232	+	return a
			233	+	}
			234	+
			235	+	// replayDetachedEventLog starts at the end of the file, expanding the
			236	+	// window only when the bounded transcript or latest assistant reply may
			237	+	// still depend on earlier events. In sparse logs, preserving those fields
			238	+	// can require reading the entire file.
			239	+	func replayDetachedEventLog(a *Agent) {
			240	+	fi, err := os.Stat(a.EventLogPath)
			241	+	if err != nil {
			242	+	return
			243	+	}
			244	+	size := fi.Size()
			245	+	limit := int64(detachedReplayBytes)
			246	+	for {
			247	+	evs, err := ReadEventLogTail(a.EventLogPath, limit)
			248	+	if err != nil {
			249	+	return
			250	+	}
			251	+	if limit >= size {
			252	+	replayEventsIntoAgent(a, evs)
			253	+	return
			254	+	}
			255	+	trial := &Agent{status: StatusDetached, activity: "detached"}
			256	+	replayEventsIntoAgent(trial, evs)
			257	+	if len(trial.transcript) == 2000 && trial.lastAssistant != "" {
233		replayEventsIntoAgent(a, evs)	258		replayEventsIntoAgent(a, evs)
			259	+	return
			260	+	}
			261	+	if limit >= size/2 {
			262	+	limit = size
			263	+	} else {
			264	+	limit *= 2
234		}	265		}
235		}	266		}
236	-	return a
237		}	267		}
238			268
239		// replayEventsIntoAgent re-derives an agent's transcript and last	269		// replayEventsIntoAgent re-derives an agent's transcript and last

M packages/agent/swarm/persist_test.go +31 -0


@@ -221,6 +221,37 @@ 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	+	a := New(Config{Root: root, RepoRoot: root}).buildDetachedAgent(agentMeta{
	247	+	ID: "large", EventLogPath: path,
	248	+	})
	249	+	if got := a.Snapshot(); got.LastAssistant != "final answer" \|\|
	250	+	len(got.Lines) != 1 \|\| got.Lines[0] != "final answer" \|\| got.Status != StatusDone {
	251	+	t.Fatalf("replayed snapshot: assistant=%q, lines=%v, status=%s", got.LastAssistant, got.Lines, got.Status)
	252	+	}
	253	+	}
	254	+
224	255		// TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json
225	256		// doesn't blow up the whole reload. Directories with no meta.json
226	257		// at all are silently ignored (Spawn that failed mid-way leaves


@@ -221,6 +221,37 @@ 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	+	a := New(Config{Root: root, RepoRoot: root}).buildDetachedAgent(agentMeta{
		247	+	ID: "large", EventLogPath: path,
		248	+	})
		249	+	if got := a.Snapshot(); got.LastAssistant != "final answer" \|\|
		250	+	len(got.Lines) != 1 \|\| got.Lines[0] != "final answer" \|\| got.Status != StatusDone {
		251	+	t.Fatalf("replayed snapshot: assistant=%q, lines=%v, status=%s", got.LastAssistant, got.Lines, got.Status)
		252	+	}
		253	+	}
		254	+
224	// TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json	255		// TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json
225	// doesn't blow up the whole reload. Directories with no meta.json	256		// doesn't blow up the whole reload. Directories with no meta.json
226	// at all are silently ignored (Spawn that failed mid-way leaves	257		// at all are silently ignored (Spawn that failed mid-way leaves