diff --git a/sessions_export.go b/sessions_export.go index c3b8bd7..7fe1a3b 100644 --- a/sessions_export.go +++ b/sessions_export.go @@ -27,14 +27,22 @@ import ( // // API: POST /safari/session/export (session-read-export). func (s *SessionsService) Export(ctx context.Context, req *SessionExportRequest) (io.ReadCloser, *Response, error) { - httpReq, err := s.client.newRequest(ctx, http.MethodPost, "/safari/session/export", req) + // The success body is a stream, not a JSON envelope; ask for NDJSON. + return s.client.openStream(ctx, s.client.client, "/safari/session/export", "application/x-ndjson", req) +} + +// openStream POSTs req and returns the live success body for the caller to +// close. Error responses are JSON envelopes even on a streaming endpoint: the +// (bounded) body is drained and closed and surfaced as a typed error. hc is the +// http.Client to send with, so a long-lived stream can drop the overall timeout. +func (c *Client) openStream(ctx context.Context, hc *http.Client, path, accept string, req any) (io.ReadCloser, *Response, error) { + httpReq, err := c.newRequest(ctx, http.MethodPost, path, req) if err != nil { return nil, nil, err } - // The success body is a stream, not a JSON envelope; ask for NDJSON. - httpReq.Header.Set("Accept", "application/x-ndjson") + httpReq.Header.Set("Accept", accept) - httpResp, err := s.client.client.Do(httpReq) + httpResp, err := hc.Do(httpReq) if err != nil { return nil, nil, fmt.Errorf("flashduty: request to %s failed: %v", sanitizeURL(httpReq.URL), sanitizeError(err)) } @@ -43,8 +51,6 @@ func (s *SessionsService) Export(ctx context.Context, req *SessionExportRequest) resp.RateLimit = parseRateLimit(httpResp.Header) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { - // Error responses are JSON envelopes, even on a streaming endpoint. Drain - // (bounded) and close the body, then surface a typed error. defer func() { _ = httpResp.Body.Close() }() raw, _ := io.ReadAll(io.LimitReader(httpResp.Body, maxResponseBodySize)) var env envelope diff --git a/sessions_message.go b/sessions_message.go new file mode 100644 index 0000000..3c1bd3c --- /dev/null +++ b/sessions_message.go @@ -0,0 +1,161 @@ +package flashduty + +import ( + "context" + "fmt" + "io" +) + +// The session message, create, attachment-upload and subscribe endpoints are +// hand-written: they are not part of the vendored OpenAPI, and subscribe is a +// long-lived SSE stream the typed JSON path cannot model. + +// SessionCreateRequest creates an empty session. It is needed before the first +// message when attachments must be uploaded first, because an attachment is +// stored under an existing session. +type SessionCreateRequest struct { + // AppName is the agent app, e.g. "ai-sre". Required. + AppName string `json:"app_name"` + // EntryKind is how the session was initiated: web, im, api or automation. + EntryKind string `json:"entry_kind,omitempty"` + // TeamID binds the session to a team (0 = account scope). Immutable. + TeamID int64 `json:"team_id,omitempty"` +} + +// SessionCreateResponse is the created session. +type SessionCreateResponse struct { + SessionID string `json:"session_id"` + AppName string `json:"app_name"` + EntryKind string `json:"entry_kind"` +} + +// WriteCreate creates a session. +// +// API: POST /safari/session/create. +func (s *SessionsService) WriteCreate(ctx context.Context, req *SessionCreateRequest) (*SessionCreateResponse, *Response, error) { + out := new(SessionCreateResponse) + resp, err := s.client.do(ctx, "/safari/session/create", req, out) + if err != nil { + return nil, resp, err + } + return out, resp, nil +} + +// SessionMessageRequest is one user turn sent to a session. +type SessionMessageRequest struct { + // SessionID continues an existing session. Omit it to start a new one, in + // which case AppName and EntryKind are required. + SessionID string `json:"session_id,omitempty"` + // AppName is the agent app ("ai-sre"); required when SessionID is empty. + AppName string `json:"app_name,omitempty"` + // EntryKind is how the run was initiated (web, im, api); required when + // SessionID is empty. + EntryKind string `json:"entry_kind,omitempty"` + // Query is the message text (max 32000 characters). + Query string `json:"query,omitempty"` + // Parts carries additional message parts, such as uploaded files. + Parts []SessionMessagePart `json:"parts,omitempty"` + // Surface tags where this turn originated, e.g. "cli". + Surface string `json:"surface,omitempty"` + // DeliveryMode negotiates how replies are delivered. "reply_only" makes the + // agent answer through its reply tool only; check the ack to see whether + // the server accepted it. + DeliveryMode string `json:"delivery_mode,omitempty"` + // TeamID binds a new session to a team (0 = account scope). + TeamID int64 `json:"team_id,omitempty"` + // EnvironmentKind selects the execution environment: "cloud" or "byoc". + EnvironmentKind string `json:"environment_kind,omitempty"` + // EnvironmentID selects a specific environment of that kind. + EnvironmentID string `json:"environment_id,omitempty"` +} + +// SessionMessagePart is one part of a message: text or a stored file. +type SessionMessagePart struct { + // Type is "text" or "file". + Type string `json:"type"` + // Text is set when Type is "text". + Text string `json:"text,omitempty"` + // File is set when Type is "file". + File *SessionFilePart `json:"file,omitempty"` +} + +// SessionFilePart references a file already stored by UploadAttachment. +type SessionFilePart struct { + Mime string `json:"mime"` + Filename string `json:"filename,omitempty"` + ObjectKey string `json:"object_key"` + Size int64 `json:"size,omitempty"` +} + +// SessionMessageAck is the acknowledgement of an accepted message. The turn +// runs asynchronously; follow it with Subscribe. +type SessionMessageAck struct { + SessionID string `json:"session_id"` + InvocationID string `json:"invocation_id"` + RootInvocationID string `json:"root_invocation_id"` + // DeliveryMode is the mode the server actually accepted. It is empty when + // the requested mode was not granted. + DeliveryMode string `json:"delivery_mode,omitempty"` +} + +// WriteMessage sends a user turn and returns once the server has durably +// queued it (HTTP 202). Replies are not in the ack; read them from Subscribe. +// +// API: POST /safari/session/message. +func (s *SessionsService) WriteMessage(ctx context.Context, req *SessionMessageRequest) (*SessionMessageAck, *Response, error) { + if req == nil { + return nil, nil, fmt.Errorf("flashduty: SessionMessageRequest is required") + } + out := new(SessionMessageAck) + resp, err := s.client.do(ctx, "/safari/session/message", req, out) + if err != nil { + return nil, resp, err + } + return out, resp, nil +} + +// SessionAttachmentUploadRequest carries one file to store under a session. +type SessionAttachmentUploadRequest struct { + // SessionID is the existing session the file belongs to. Required. + SessionID string + // File is the file content. Required. Max 50 MB. + File io.Reader + // Filename is the multipart file name. Required. + Filename string +} + +// SessionAttachment describes a stored attachment. +type SessionAttachment struct { + ObjectKey string `json:"object_key"` + Mime string `json:"mime"` + Filename string `json:"filename"` + Size int64 `json:"size"` + URL string `json:"url"` + ExpiresIn int `json:"expires_in"` + AttachmentID string `json:"attachment_id"` +} + +// FilePart converts the stored attachment into a message part for +// SessionMessageRequest.Parts. +func (a *SessionAttachment) FilePart() SessionMessagePart { + return SessionMessagePart{ + Type: "file", + File: &SessionFilePart{Mime: a.Mime, Filename: a.Filename, ObjectKey: a.ObjectKey, Size: a.Size}, + } +} + +// UploadAttachment stores a file under a session as a multipart upload. +// +// API: POST /safari/attachment/upload. +func (s *SessionsService) UploadAttachment(ctx context.Context, req *SessionAttachmentUploadRequest) (*SessionAttachment, *Response, error) { + if req == nil || req.File == nil || req.SessionID == "" || req.Filename == "" { + return nil, nil, fmt.Errorf("flashduty: SessionAttachmentUploadRequest requires SessionID, Filename and a non-nil File") + } + out := new(SessionAttachment) + fields := map[string]string{"session_id": req.SessionID} + resp, err := s.client.uploadFile(ctx, "/safari/attachment/upload", nil, fields, req.Filename, req.File, out) + if err != nil { + return nil, resp, err + } + return out, resp, nil +} diff --git a/sessions_message_test.go b/sessions_message_test.go new file mode 100644 index 0000000..7ba0840 --- /dev/null +++ b/sessions_message_test.go @@ -0,0 +1,138 @@ +package flashduty + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "testing" +) + +func TestSessionsWriteMessageSendsRequestAndDecodesAck(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/safari/session/message" { + t.Errorf("path = %s", r.URL.Path) + } + if got := r.URL.Query().Get("app_key"); got != "KEY" { + t.Errorf("app_key = %q", got) + } + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if body["surface"] != "cli" || body["delivery_mode"] != "reply_only" || body["query"] != "hi" { + t.Errorf("body = %v", body) + } + parts, _ := body["parts"].([]any) + if len(parts) != 1 { + t.Fatalf("parts = %v", body["parts"]) + } + file := parts[0].(map[string]any)["file"].(map[string]any) + if file["object_key"] != "attachments/a/b.txt" || file["mime"] != "text/plain" { + t.Errorf("file part = %v", file) + } + if _, ok := body["team_id"]; ok { + t.Errorf("zero team_id must be omitted: %v", body) + } + w.WriteHeader(http.StatusAccepted) + _, _ = io.WriteString(w, `{"request_id":"r1","data":{"session_id":"s1","invocation_id":"i1","root_invocation_id":"i1","delivery_mode":"reply_only"}}`) + }) + + ack, _, err := c.Sessions.WriteMessage(context.Background(), &SessionMessageRequest{ + AppName: "ai-sre", + EntryKind: "api", + Query: "hi", + Surface: "cli", + DeliveryMode: "reply_only", + Parts: []SessionMessagePart{{ + Type: "file", + File: &SessionFilePart{Mime: "text/plain", Filename: "b.txt", ObjectKey: "attachments/a/b.txt", Size: 3}, + }}, + }) + if err != nil { + t.Fatalf("WriteMessage: %v", err) + } + if ack.SessionID != "s1" || ack.InvocationID != "i1" || ack.RootInvocationID != "i1" || ack.DeliveryMode != "reply_only" { + t.Errorf("ack = %+v", ack) + } +} + +func TestSessionsWriteMessageSurfacesAPIError(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = io.WriteString(w, `{"request_id":"r1","error":{"code":"InvalidParameter","message":"bad team"}}`) + }) + _, _, err := c.Sessions.WriteMessage(context.Background(), &SessionMessageRequest{AppName: "ai-sre"}) + if err == nil || !strings.Contains(err.Error(), "bad team") { + t.Fatalf("err = %v", err) + } +} + +func TestSessionsWriteCreate(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/safari/session/create" { + t.Errorf("path = %s", r.URL.Path) + } + var body SessionCreateRequest + _ = json.NewDecoder(r.Body).Decode(&body) + if body.AppName != "ai-sre" || body.EntryKind != "api" || body.TeamID != 7 { + t.Errorf("body = %+v", body) + } + _, _ = io.WriteString(w, `{"request_id":"r1","data":{"session_id":"s9","app_name":"ai-sre","entry_kind":"api"}}`) + }) + out, _, err := c.Sessions.WriteCreate(context.Background(), &SessionCreateRequest{AppName: "ai-sre", EntryKind: "api", TeamID: 7}) + if err != nil || out.SessionID != "s9" { + t.Fatalf("out = %+v, err = %v", out, err) + } +} + +func TestSessionsUploadAttachmentMultipart(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/safari/attachment/upload" { + t.Errorf("path = %s", r.URL.Path) + } + r.Body = http.MaxBytesReader(w, r.Body, 1<<20) + if err := r.ParseMultipartForm(1 << 20); err != nil { //nolint:gosec // test server; body capped by MaxBytesReader above + t.Fatalf("ParseMultipartForm: %v", err) + } + if got := r.FormValue("session_id"); got != "s1" { + t.Errorf("session_id = %q", got) + } + f, hdr, err := r.FormFile("file") + if err != nil { + t.Fatalf("FormFile: %v", err) + } + defer func() { _ = f.Close() }() + data, _ := io.ReadAll(f) + if hdr.Filename != "notes.txt" || string(data) != "hello" { + t.Errorf("file = %q %q", hdr.Filename, data) + } + _, _ = io.WriteString(w, `{"request_id":"r1","data":{"object_key":"attachments/att_1/notes.txt","mime":"text/plain","filename":"notes.txt","size":5,"url":"https://x/y","expires_in":3600,"attachment_id":"att_1"}}`) + }) + + att, _, err := c.Sessions.UploadAttachment(context.Background(), &SessionAttachmentUploadRequest{ + SessionID: "s1", Filename: "notes.txt", File: strings.NewReader("hello"), + }) + if err != nil { + t.Fatalf("UploadAttachment: %v", err) + } + part := att.FilePart() + if part.Type != "file" || part.File.ObjectKey != "attachments/att_1/notes.txt" || part.File.Mime != "text/plain" || part.File.Size != 5 || part.File.Filename != "notes.txt" { + t.Errorf("part = %+v", part.File) + } +} + +func TestSessionsUploadAttachmentValidatesInput(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { t.Error("no request expected") }) + for _, req := range []*SessionAttachmentUploadRequest{ + nil, + {Filename: "a", File: strings.NewReader("x")}, + {SessionID: "s", File: strings.NewReader("x")}, + {SessionID: "s", Filename: "a"}, + } { + if _, _, err := c.Sessions.UploadAttachment(context.Background(), req); err == nil { + t.Errorf("expected error for %+v", req) + } + } +} diff --git a/sessions_subscribe.go b/sessions_subscribe.go new file mode 100644 index 0000000..15de917 --- /dev/null +++ b/sessions_subscribe.go @@ -0,0 +1,134 @@ +package flashduty + +import ( + "bufio" + "context" + "encoding/json" + "io" + "strings" +) + +// SessionSubscribeRequest opens a session event stream. The stream first +// replays persisted events strictly newer than the (LastEventAt, LastEventID) +// cursor, then tails live frames. +type SessionSubscribeRequest struct { + SessionID string `json:"session_id"` + // LastEventAt is the resume cursor: a Unix-millisecond timestamp. 0 replays + // from the start of the session. + LastEventAt int64 `json:"last_event_at,omitempty"` + // LastEventID breaks ties between events sharing LastEventAt. + LastEventID string `json:"last_event_id,omitempty"` +} + +// SessionEventFrame is one frame of a session event stream. Persisted events +// carry EventID and CreatedAt, with Type set to the event author ("user", +// "assistant", ...); ephemeral frames (turn_complete, assistant_token, ...) +// carry no EventID and Type is the frame type. Payload is the frame-specific +// JSON. +type SessionEventFrame struct { + EventID string `json:"event_id,omitempty"` + CreatedAt int64 `json:"created_at"` + Type string `json:"type"` + AgentPath string `json:"agent_path,omitempty"` + Payload json.RawMessage `json:"payload"` + InvocationID string `json:"invocation_id,omitempty"` + RootInvocationID string `json:"root_invocation_id,omitempty"` + DeliveryMode string `json:"delivery_mode,omitempty"` + IsError bool `json:"is_error,omitempty"` + ErrorCode string `json:"error_code,omitempty"` +} + +// Subscribe opens POST /safari/session/subscribe and returns the live SSE +// response body. The caller owns it and MUST Close it; read frames with +// NewSessionFrameReader. The stream is unbounded in time (the server closes it +// after an idle period), so it is served by a client with no overall timeout: +// bound its lifetime with ctx. +// +// On any non-2xx status the body is the usual JSON error envelope: Subscribe +// reads and closes it and returns a typed error with a nil ReadCloser. +// +// API: POST /safari/session/subscribe. +func (s *SessionsService) Subscribe(ctx context.Context, req *SessionSubscribeRequest) (io.ReadCloser, *Response, error) { + // The shared client's Timeout also bounds reading the body, which would cut + // a long-lived stream. Copy it without the overall timeout; the transport + // and redirect policy are shared. + streamClient := *s.client.client + streamClient.Timeout = 0 + return s.client.openStream(ctx, &streamClient, "/safari/session/subscribe", "text/event-stream", req) +} + +// SessionStreamError is a server-reported error carried in-band on the stream +// as an "event: error" record. +type SessionStreamError struct { + Message string +} + +func (e *SessionStreamError) Error() string { + return "flashduty: session stream error: " + e.Message +} + +// SessionFrameReader decodes the SSE stream returned by Subscribe into frames. +type SessionFrameReader struct { + sc *bufio.Scanner +} + +// NewSessionFrameReader wraps a Subscribe stream. Heartbeat comments and +// undecodable records are skipped. +// +// rc, _, err := client.Sessions.Subscribe(ctx, req) +// if err != nil { return err } +// defer rc.Close() +// fr := flashduty.NewSessionFrameReader(rc) +// for { +// frame, err := fr.Next() +// if err == io.EOF { break } // server finished the stream +// if err != nil { return err } +// // ... handle frame ... +// } +func NewSessionFrameReader(r io.Reader) *SessionFrameReader { + sc := bufio.NewScanner(r) + // One persisted frame can carry a large tool output. + sc.Buffer(make([]byte, 0, 64*1024), maxResponseBodySize) + return &SessionFrameReader{sc: sc} +} + +// Next returns the next frame. It returns io.EOF when the server ends the +// stream (the "[DONE]" sentinel or a clean close), a *SessionStreamError for an +// in-band error record, and the transport error (including context +// cancellation) otherwise. +func (r *SessionFrameReader) Next() (*SessionEventFrame, error) { + eventName := "" + for r.sc.Scan() { + line := strings.TrimRight(r.sc.Text(), "\r") + switch { + case line == "": + eventName = "" + case strings.HasPrefix(line, ":"): + // heartbeat comment + case strings.HasPrefix(line, "event:"): + eventName = strings.TrimSpace(strings.TrimPrefix(line, "event:")) + case strings.HasPrefix(line, "data:"): + data := strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " ") + if eventName == "error" { + msg := data + var s string + if json.Unmarshal([]byte(data), &s) == nil { + msg = s + } + return nil, &SessionStreamError{Message: msg} + } + if data == "[DONE]" { + return nil, io.EOF + } + var frame SessionEventFrame + if err := json.Unmarshal([]byte(data), &frame); err != nil || frame.Type == "" { + continue + } + return &frame, nil + } + } + if err := r.sc.Err(); err != nil { + return nil, err + } + return nil, io.EOF +} diff --git a/sessions_subscribe_test.go b/sessions_subscribe_test.go new file mode 100644 index 0000000..fe8f046 --- /dev/null +++ b/sessions_subscribe_test.go @@ -0,0 +1,149 @@ +package flashduty + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "testing" + "time" +) + +const sseBody = ": ping\n\n" + + `data: {"event_id":"e1","created_at":1000,"type":"assistant","invocation_id":"i1","root_invocation_id":"r1","payload":{"parts":[]}}` + "\n\n" + + ": ping\n\n" + + "data: not-json\n\n" + + `data: {"created_at":1001,"type":"turn_complete","payload":{}}` + "\n\n" + + "data: [DONE]\n\n" + +func TestSessionFrameReaderDecodesFramesAndStopsAtDone(t *testing.T) { + fr := NewSessionFrameReader(strings.NewReader(sseBody)) + + f1, err := fr.Next() + if err != nil || f1.EventID != "e1" || f1.Type != "assistant" || f1.RootInvocationID != "r1" || f1.CreatedAt != 1000 { + t.Fatalf("frame 1 = %+v, err = %v", f1, err) + } + f2, err := fr.Next() + if err != nil || f2.Type != "turn_complete" || f2.EventID != "" { + t.Fatalf("frame 2 = %+v, err = %v", f2, err) + } + if _, err := fr.Next(); err != io.EOF { + t.Fatalf("after [DONE] err = %v, want io.EOF", err) + } +} + +func TestSessionFrameReaderCleanCloseIsEOF(t *testing.T) { + fr := NewSessionFrameReader(strings.NewReader(`data: {"type":"assistant","payload":{}}` + "\n\n")) + if _, err := fr.Next(); err != nil { + t.Fatal(err) + } + if _, err := fr.Next(); err != io.EOF { + t.Fatalf("err = %v, want io.EOF", err) + } +} + +func TestSessionFrameReaderErrorEvent(t *testing.T) { + fr := NewSessionFrameReader(strings.NewReader("event: error\ndata: \"stream broke\"\n\n")) + _, err := fr.Next() + var se *SessionStreamError + if !errors.As(err, &se) || se.Message != "stream broke" { + t.Fatalf("err = %v", err) + } +} + +func TestSessionsSubscribeStreamsFrames(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/safari/session/subscribe" { + t.Errorf("path = %s", r.URL.Path) + } + if acc := r.Header.Get("Accept"); acc != "text/event-stream" { + t.Errorf("Accept = %q", acc) + } + body, _ := io.ReadAll(r.Body) + if !strings.Contains(string(body), `"session_id":"s1"`) || !strings.Contains(string(body), `"last_event_at":42`) { + t.Errorf("body = %s", body) + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, sseBody) + }) + + rc, _, err := c.Sessions.Subscribe(context.Background(), &SessionSubscribeRequest{SessionID: "s1", LastEventAt: 42}) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer func() { _ = rc.Close() }() + fr := NewSessionFrameReader(rc) + n := 0 + for { + _, err := fr.Next() + if err == io.EOF { + break + } + if err != nil { + t.Fatal(err) + } + n++ + } + if n != 2 { + t.Errorf("frames = %d, want 2", n) + } +} + +func TestSessionsSubscribeHTTPErrorIsTyped(t *testing.T) { + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusForbidden) + _, _ = io.WriteString(w, `{"request_id":"r1","error":{"code":"AccessDenied","message":"not yours"}}`) + }) + rc, _, err := c.Sessions.Subscribe(context.Background(), &SessionSubscribeRequest{SessionID: "s1"}) + if rc != nil { + t.Error("ReadCloser must be nil on error") + } + var apiErr *ErrorResponse + if !errors.As(err, &apiErr) || !strings.Contains(apiErr.Message, "not yours") { + t.Fatalf("err = %v", err) + } +} + +func TestSessionsSubscribeOutlivesClientTimeoutAndHonorsCancel(t *testing.T) { + release := make(chan struct{}) + c := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, ": ping\n\n") + w.(http.Flusher).Flush() + select { + case <-r.Context().Done(): + case <-release: + } + }) + t.Cleanup(func() { close(release) }) + // A short overall timeout would cut the stream if Subscribe reused it. + c.client.Timeout = 50 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + rc, _, err := c.Sessions.Subscribe(ctx, &SessionSubscribeRequest{SessionID: "s1"}) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer func() { _ = rc.Close() }() + + errCh := make(chan error, 1) + go func() { + _, err := NewSessionFrameReader(rc).Next() + errCh <- err + }() + select { + case err := <-errCh: + t.Fatalf("stream ended after the client timeout: %v", err) + case <-time.After(200 * time.Millisecond): + } + cancel() + select { + case err := <-errCh: + if err == nil || err == io.EOF { + t.Fatalf("err after cancel = %v, want a cancellation error", err) + } + case <-time.After(2 * time.Second): + t.Fatal("reader did not return after context cancel") + } +}