fix(stream): bind replay to actor generations
New actors get fresh replay tickets. Keep generation and sequence together through resets and reconnects.
333 unmodified linesExpand all33433533633733833934034134134234234334434634534634734834935035135240 unmodified lines333 unmodified linesExpand all away: str(elsewhere), busy: str(self.app.workspace), }, ) cursor = self.app.stream_page(busy) release.set() self.app.idle(busy) self.assertEqual(workspaces()[busy], str(moved)) with self.app.api(f"/sessions/{busy}/stream?after_seq=0") as response: frame = next(line for line in response if line.startswith(b"data: ")) self.assertIn( "workspace moved with parent; python variables were cleared, " "the transcript is intact", [event.get("text") for event in json.loads(frame[6:])["events"]], [ event.get("text") for event in self.app.stream_page(busy, cursor)["events"] ], )
def test_qualified_and_inferred_cross_provider_models(self): parent = self.app.session()40 unmodified lines220 unmodified linesExpand all221222223224225226227228229230230231232233231232233234235290 unmodified linesExpand all52652752852953053153253353412 unmodified lines547548549550552553554555551552553554555336 unmodified linesExpand all220 unmodified linesExpand all ) removed.unlink()
def test_oversized_agent_instructions_warn_and_allow_turns(self): before = self.app.stream_page(self.sid) (self.app.workspace / "AGENTS.md").write_text("X" * (1024 * 1024 + 1)) (self.app.workspace / "CLAUDE.md").write_text("USABLE_CONVENTION") (request,) = self.turn("continue without oversized instructions") self.assertIn("USABLE_CONVENTION", request["instructions"]) self.assertNotIn("X" * 128, request["instructions"]) with self.app.api(f"/sessions/{self.sid}/stream?after_seq=0") as stream: events = json.loads( next(line[6:] for line in stream if line.startswith(b"data: ")) )["events"] events = self.app.stream_page(self.sid, before)["events"] warnings = [ event["text"] for event in events if event.get("type") == "note" and "AGENTS.md" in event.get("text", "")290 unmodified linesExpand all self.assertNotIn("second workspace item", json.dumps(invoke(sessions[0], {}))) self.assertIn("second workspace item", json.dumps(invoke(sessions[1], {})))
def test_manual_compaction_keeps_tree_and_reuses_summary(self): cursor = self.app.stream_page(self.sid) self.turn("first turn") self.turn("second turn") self.turn("third turn") self.turn("fourth turn")12 unmodified lines context = self.get(f"/sessions/{self.sid}/context") self.assertEqual( (context["state"], context["compaction"]["status"]), ("ready", "compacted") ) with self.app.api(f"/sessions/{self.sid}/stream?after_seq=0") as stream: events = json.loads( next(line[6:] for line in stream if line.startswith(b"data: ")) )["events"] events = self.app.stream_page(self.sid, cursor)["events"] compacted = next(event for event in events if event.get("type") == "compacted") self.assertGreater(compacted["evicted"], 0) self.assertIn("older conversation summary", compacted["summary"]) history = self.get(f"/sessions/{self.sid}/context/history/0")["content"]336 unmodified linesExpand all52 unmodified lines535455565758596061860 unmodified linesExpand all9229239249259259269269279289299309319329339349359369379389299309399409419429439449459469479483 unmodified lines52 unmodified linesimport tempfileimport threadingimport timeimport urllib.errorimport urllib.parseimport uuid
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "test"))860 unmodified linesExpand all return status time.sleep(0.05) self._fail(f"session did not settle: {status}")
def events(self, session_id): with self.api(f"/sessions/{session_id}/stream") as response: def stream_page(self, session_id, after=None, *, tail=None): query = {} if after is not None: query = { "after_generation": after["generation"], "after_seq": after["cursor"], } if tail is not None: query["tail"] = tail suffix = "?" + urllib.parse.urlencode(query) if query else "" with self.api(f"/sessions/{session_id}/stream{suffix}") as response: for line in response: if line.startswith(b"data: "): return json.loads(line[6:])["events"] return [] return json.loads(line[6:]) self._fail("stream closed before its first batch")
def events(self, session_id): return self.stream_page(session_id)["events"]
def history(self, session_id): with self.api(f"/sessions/{session_id}/history") as response: return json.load(response)
3 unmodified lines1233456789111213141516171819101112131415201617181956 unmodified lines7677787980818283845 unmodified lines909192939894959697986 unmodified lines105106107108113109110111112"""Idle kernel reaping persists Python state and explains restoration to the model."""
import jsonimport subprocessimport timeimport unittest
from harness import Albedo, Provider, exclusive, python, text
def notes_for(app, session): with app.api(f"/sessions/{session}/stream?after_seq=0") as response: for line in response: if line.startswith(b"data: "): return [ event["text"] for event in json.loads(line[6:])["events"] if event.get("type") == "note" ]def notes_for(app, session, cursor): return [ event["text"] for event in app.stream_page(session, cursor)["events"] if event.get("type") == "note" ] return []
# Short enough that the test waits seconds, long enough to outlast a turn.IDLE_SECONDS = 256 unmodified lines app = self.app daemon = app.connection["pid"] existing = set(kernels(daemon)) session = app.session() cursor = app.stream_page(session) self.assertEqual( set(kernels(daemon)) - existing, set(), "session must not eagerly open a kernel",5 unmodified lines time.sleep(0.1) self.assertEqual( set(kernels(daemon)) - existing, set(), "idle kernel outlived its limit" ) notes = notes_for(app, session) notes = notes_for(app, session, cursor) self.assertIn("released", notes[-1]) self.assertIn("2 variables saved to disk", notes[-1]) self.assertTrue((app.home / "kernels" / f"{session}.state").exists())
6 unmodified lines "recalled 7 [1, 2, 3]" in item.get("output", "") for item in self.provider.requests[-1]["request"]["input"] ) ) notes = notes_for(app, session) notes = notes_for(app, session, cursor) self.assertTrue( any("restored 2 variables from disk" in note for note in notes), notes )123345652 unmodified lines5960616264656667686963716465666768697870717273747576778678798081823 unmodified lines8687888998909192939417 unmodified lines"""Unread background job results wake idle sessions once, even across idle reaping."""
import jsonimport timeimport unittest
from harness import Albedo, Provider, exclusive, python, text52 unmodified lines
def users(self): return [user_text(record["request"]) for record in self.provider.requests]
def stream(self, session, query): with self.app.api(f"/sessions/{session}/stream{query}") as response: for line in response: if line.startswith(b"data: "): return json.loads(line[6:]) self.fail("no stream snapshot")
class JobWakeTests(JobWakeCase): def test_unread_job_wakes_once_with_live_source_and_durable_turn(self): session = self.app.session() self.app.prompt(session, "start a slow job").close() self.app.idle(session) self.assertEqual(len(self.users()), 2) before = self.stream(session, "?after_seq=-1") before = self.app.stream_page(session) wait_for(lambda: len(self.users()) >= 3) self.app.idle(session) self.assertEqual(len(self.users()), 3) wake = self.users()[2] self.assertIn("background job finished", wake) self.assertIn("jobs[", wake) self.assertIn("output.read", wake) live = self.stream(session, f"?after_seq={before['cursor']}")["events"] live = self.app.stream_page(session, before)["events"] notices = [ event for event in live if event.get("type") == "user" and "job finished" in event.get("text", "")3 unmodified lines self.assertEqual(notices[0]["clientId"], "job") self.assertIn("exit_code=0", notices[0]["text"]) durable = [ event for event in self.stream(session, "?after_seq=-1")["events"] for event in self.app.events(session) if event.get("type") == "user" and "job finished" in event.get("text", "") ] self.assertEqual(len(durable), 1) time.sleep(2.5)17 unmodified lines1234567891011121314151617181920212223242526345678910111213141516171819202122232425262728293031323334{ "malformed_batches": { "malformed JSON": "{\"cursor\":8,", "missing cursor": "{\"events\":[]}", "null cursor": "{\"cursor\":null,\"events\":[]}", "fractional cursor": "{\"cursor\":8.5,\"events\":[]}", "negative cursor": "{\"cursor\":-1,\"events\":[]}", "overflowing cursor": "{\"cursor\":9223372036854775808,\"events\":[]}", "regressing cursor": "{\"cursor\":6,\"events\":[]}", "missing events": "{\"cursor\":8}", "null events": "{\"cursor\":8,\"events\":null}", "missing event type": "{\"cursor\":8,\"events\":[{\"text\":\"not an assistant\"}]}", "empty event type": "{\"cursor\":8,\"events\":[{\"type\":\"\",\"text\":\"not an assistant\"}]}", "nonstring event type": "{\"cursor\":8,\"events\":[{\"type\":42}]}", "partial batch": "{\"cursor\":8,\"events\":[{\"type\":\"text\",\"text\":\"must not escape\"},{\"type\":\"message\",\"role\":\"assistant\",\"text\":false}]}", "missing known field": "{\"cursor\":8,\"events\":[{\"type\":\"message\",\"role\":\"assistant\"}]}", "missing identity": "{\"cursor\":8,\"events\":[{\"type\":\"turn_completed\",\"turnId\":\"\"}]}", "negative transcript offset": "{\"cursor\":8,\"events\":[{\"type\":\"committed\",\"seq\":-1}]}", "tool args nonstring": "{\"cursor\":8,\"events\":[{\"type\":\"tool\",\"callId\":\"c\",\"name\":\"python\",\"args\":false,\"result\":\"\"}]}", "missing user source": "{\"cursor\":8,\"events\":[{\"type\":\"user\",\"text\":\"hello\",\"triggeredAt\":\"\"}]}", "missing user trigger": "{\"cursor\":8,\"events\":[{\"type\":\"user\",\"text\":\"hello\",\"source\":\"chat\"}]}", "invalid usage count": "{\"cursor\":8,\"events\":[{\"type\":\"usage\",\"model\":\"test\",\"recordedAt\":1,\"promptTokens\":-1}]}", "missing progress": "{\"cursor\":8,\"events\":[{\"type\":\"tool_progress\"}]}", "invalid member identity": "{\"cursor\":8,\"events\":[{\"type\":\"turn_membership\",\"turnId\":\"turn\",\"submissionIds\":[\"\"]}]}", "invalid thinking elapsed": "{\"cursor\":8,\"events\":[{\"type\":\"thinking\",\"text\":\"thought\",\"elapsedMs\":1.5}]}", "compaction missing summary": "{\"cursor\":8,\"events\":[{\"type\":\"compacted\",\"evicted\":2}]}" "malformed JSON": "{\"generation\":\"generation-a\",\"cursor\":8,", "missing cursor": "{\"generation\":\"generation-a\",\"events\":[]}", "null cursor": "{\"generation\":\"generation-a\",\"cursor\":null,\"events\":[]}", "fractional cursor": "{\"generation\":\"generation-a\",\"cursor\":8.5,\"events\":[]}", "negative cursor": "{\"generation\":\"generation-a\",\"cursor\":-1,\"events\":[]}", "overflowing cursor": "{\"generation\":\"generation-a\",\"cursor\":9223372036854775808,\"events\":[]}", "regressing cursor": "{\"generation\":\"generation-a\",\"cursor\":6,\"events\":[]}", "missing events": "{\"generation\":\"generation-a\",\"cursor\":8}", "null events": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":null}", "missing event type": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"text\":\"not an assistant\"}]}", "empty event type": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"\",\"text\":\"not an assistant\"}]}", "nonstring event type": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":42}]}", "partial batch": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"text\",\"text\":\"must not escape\"},{\"type\":\"message\",\"role\":\"assistant\",\"text\":false}]}", "missing known field": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"message\",\"role\":\"assistant\"}]}", "missing identity": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"turn_completed\",\"turnId\":\"\"}]}", "negative transcript offset": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"committed\",\"seq\":-1}]}", "tool args nonstring": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"tool\",\"callId\":\"c\",\"name\":\"python\",\"args\":false,\"result\":\"\"}]}", "missing user source": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"user\",\"text\":\"hello\",\"triggeredAt\":\"\"}]}", "missing user trigger": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"user\",\"text\":\"hello\",\"source\":\"chat\"}]}", "invalid usage count": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"usage\",\"model\":\"test\",\"recordedAt\":1,\"promptTokens\":-1}]}", "missing progress": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"tool_progress\"}]}", "invalid member identity": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"turn_membership\",\"turnId\":\"turn\",\"submissionIds\":[\"\"]}]}", "invalid thinking elapsed": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"thinking\",\"text\":\"thought\",\"elapsedMs\":1.5}]}", "compaction missing summary": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"compacted\",\"evicted\":2}]}", "missing generation": "{\"cursor\":8,\"events\":[]}", "null generation": "{\"cursor\":8,\"events\":[],\"generation\":null}", "empty generation": "{\"cursor\":8,\"events\":[],\"generation\":\"\"}", "nonstring generation": "{\"cursor\":8,\"events\":[],\"generation\":42}", "changed generation without reset": "{\"cursor\":8,\"events\":[],\"generation\":\"generation-b\"}", "nonleading reset": "{\"generation\":\"generation-a\",\"cursor\":8,\"events\":[{\"type\":\"future_output\"},{\"type\":\"reset\"}]}" }}134 unmodified linesExpand all135136137138139139140141142143112 unmodified linesExpand all134 unmodified linesExpand all raise AssertionError(f"session never settled: {state}")
def outputs(session): """The first stream page after reset is the full history snapshot.""" with api(f"/sessions/{session}/stream?after_seq=-1") as response: with api(f"/sessions/{session}/stream") as response: response.fp.raw._sock.settimeout(60) while True: line = response.readline() if not line:112 unmodified linesExpand all164 unmodified linesExpand all165166167168169169170171172173174175176177178179818 unmodified linesExpand all164 unmodified linesExpand all
def send_chunk(events, cursor): payload = ( "data: " + json.dumps({"cursor": cursor, "events": events}) + json.dumps( { "generation": "smoke-generation", "cursor": cursor, "events": events, } ) + "\n\n" ) self.wfile.write(payload.encode()) self.wfile.flush()818 unmodified linesExpand all122 unmodified linesExpand all12312412512612712712812913013137 unmodified lines122 unmodified linesExpand all )
# Hold the stream open: no status polling, no python calls, nothing that # could carry the completion back except the wake itself. with api(f"/sessions/{session_id}/stream?after_seq=-1") as response: with api(f"/sessions/{session_id}/stream") as response: response.fp.raw._sock.settimeout(150) seen = [] wake_at = None while True:37 unmodified lines55 unmodified lines565758596060616263646 unmodified lines7172737475757677787934 unmodified lines11411511611711811811912012112224 unmodified lines55 unmodified lines writer.WriteHeader(http.StatusAccepted) _ = json.NewEncoder(writer).Encode(map[string]any{"ok": true, "queued": true, "operationId": payload.OperationID}) case "/sessions/test/stream": writer.Header().Set("Content-Type", "text/event-stream") _, _ = fmt.Fprint(writer, "data: {\"cursor\":0,\"events\":[{\"type\":\"reset\"}]}\n\n") _, _ = fmt.Fprint(writer, "data: {\"generation\":\"generation-a\",\"cursor\":0,\"events\":[{\"type\":\"reset\"}]}\n\n") writer.(http.Flusher).Flush() id := <-submitted events := []map[string]any{ {"type": "turn_membership", "turnId": "combined", "submissionIds": []string{id, "other"}},6 unmodified lines {"type": "turn_completed", "turnId": "combined"}, {"type": "turn_membership", "turnId": "later", "submissionIds": []string{"other"}}, {"type": "message", "role": "assistant", "text": "later answer"}, } page, _ := json.Marshal(map[string]any{"cursor": 1, "events": events}) page, _ := json.Marshal(map[string]any{"generation": "generation-a", "cursor": 1, "events": events}) _, _ = fmt.Fprintf(writer, "data: %s\n\n", page) writer.(http.Flusher).Flush() <-request.Context().Done() default:34 unmodified lines _ = json.NewEncoder(writer).Encode(map[string]any{"ok": true, "queued": true, "operationId": payload.OperationID}) close(submitted) case "/sessions/test/stream": writer.Header().Set("Content-Type", "text/event-stream") _, _ = fmt.Fprint(writer, "data: {\"cursor\":0,\"events\":[{\"type\":\"reset\"}]}\n\n") _, _ = fmt.Fprint(writer, "data: {\"generation\":\"generation-a\",\"cursor\":0,\"events\":[{\"type\":\"reset\"}]}\n\n") writer.(http.Flusher).Flush() <-request.Context().Done() case "/sessions/test/cancel-submission": if request.Context().Err() != nil {24 unmodified lines63 unmodified lines64656667686869707172381 unmodified linesExpand all45445545645745845845946046146246346446546646746846947047146246347347446446546646746846941 unmodified lines51151251351452451551651752751851952052152252352952452552652752852953053149 unmodified lines58158258358458959058558658758858959459059159259359459559659759859960060160260360560460560760860660760861060961061161361461561661761261361461539 unmodified lines65565665765866565966066166266350 unmodified lines63 unmodified lines argumentsByCall map[string]*strings.Builder agentID string clientID string afterSeq int64 requireReset bool afterGeneration string mu sync.Mutex}
// NewChatClient shares conn's HTTP transport. A nil connection panics.381 unmodified linesExpand allfunc (c *ChatClient) ResetStream() { c.mu.Lock() c.argumentsByCall = make(map[string]*strings.Builder) c.afterSeq = -1 c.requireReset = false c.afterGeneration = "" c.mu.Unlock()}
// RequireStreamReset requests durable history before accepting further events.// Call it after the subscription has stopped writing.func (c *ChatClient) RequireStreamReset() { c.mu.Lock() c.argumentsByCall = make(map[string]*strings.Builder) c.afterSeq = -1 c.requireReset = true c.mu.Unlock()}
type streamBatch struct { Generation string `json:"generation"` Cursor *int64 `json:"cursor"` Events []json.RawMessage `json:"events"` Cursor *int64 `json:"cursor"` Events []json.RawMessage `json:"events"`}
func streamFailure(kind StreamFailureKind, cause error) error { return &StreamError{Kind: kind, Cause: cause}41 unmodified lines } }()
c.mu.Lock() afterSeq := c.afterSeq afterSeq, afterGeneration := c.afterSeq, c.afterGeneration c.mu.Unlock()
route := fmt.Sprintf("/stream?after_seq=%d", afterSeq) query := url.Values{} if afterGeneration != "" { query.Set("after_generation", afterGeneration) query.Set("after_seq", fmt.Sprint(afterSeq)) } if tail > 0 { route += fmt.Sprintf("&tail=%d", tail) query.Set("tail", fmt.Sprint(tail)) }
route := "/stream?" + query.Encode() operation := Operation{Name: "stream session", Method: http.MethodGet, Path: c.agentPath(route), Policy: ReadRecovery} err := scanEventStream(ctx, c.conn, operation, streamLimits{requireSSE: true, lineBytes: 10 * 1024 * 1024, errorBytes: 64 * 1024}, func(scanner *bufio.Scanner) error { return c.readStream(ctx, scanner, onBatchConsumed, func(event StreamEvent) error { if err := onEvent(event); err != nil {49 unmodified lines var batch streamBatch if err := json.Unmarshal([]byte(payload), &batch); err != nil { return streamFailure(StreamProtocol, fmt.Errorf("invalid stream batch: %w", err)) } if batch.Cursor == nil || *batch.Cursor < 0 || batch.Events == nil { return streamFailure(StreamProtocol, errors.New("invalid stream batch: missing cursor or events")) if batch.Generation == "" || batch.Cursor == nil || *batch.Cursor < 0 || batch.Events == nil { return streamFailure(StreamProtocol, errors.New("invalid stream batch: missing generation, cursor, or events")) } // Validate the whole batch before delivering events or changing preview state. events := make([]wireChatEvent, 0, len(batch.Events)) for _, raw := range batch.Events { for index, raw := range batch.Events { event, err := decodeChatEvent(raw) if err != nil { return streamFailure(StreamProtocol, err) } if event != nil { if event.Type == EventReset && index != 0 { return streamFailure(StreamProtocol, errors.New("reset must begin its batch")) } events = append(events, *event) } } reset := len(events) > 0 && events[0].Type == EventReset c.mu.Lock() previous, requiresReset := c.afterSeq, c.requireReset previous, generation := c.afterSeq, c.afterGeneration c.mu.Unlock() if requiresReset && !reset { return streamFailure(StreamProtocol, errors.New("recovery stream must begin with reset")) if generation != batch.Generation && !reset { return streamFailure(StreamProtocol, errors.New("initial or changed generation stream must begin with reset")) } if *batch.Cursor < previous && !reset { if generation == batch.Generation && *batch.Cursor < previous && !reset { return streamFailure(StreamProtocol, errors.New("stream cursor regressed without reset")) } for i, event := range events { if event.Type == EventReset && i != 0 { return streamFailure(StreamProtocol, errors.New("reset must begin its batch")) } } if err := c.validateArgumentBatch(events); err != nil { return streamFailure(StreamProtocol, err) } for _, event := range events {39 unmodified lines } } c.mu.Lock() c.afterSeq = *batch.Cursor c.requireReset = false c.afterGeneration = batch.Generation c.mu.Unlock() if onBatchConsumed != nil { onBatchConsumed() }50 unmodified lines