feat(sessions): add message, attachment upload and event subscribe

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

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
+}

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)
+		}
+	}
+}