Stream negotiated canonical model reads with snapshot framing

internal/platform/canvas_model_http.go

@@ -172,6 +172,11 @@ func (s *service) getCanvasModel(w http.ResponseWriter, r *http.Request, project
 	if err = tx.Rollback(); err != nil {
 		return err
 	}
+	if r.Header.Get("Accept") == modelStreamType {
+		writeCanvasModelStream(r.Context(), w, model)
+		return nil
+	}
+	w.Header().Add("Vary", "Accept")
 	reply(w, 200, response{"model": model.view(nil)})
 	return nil
 }

internal/platform/canvas_model_integration_test.go

@@ -214,6 +214,16 @@ func TestCanonicalModelMigrationAndIncrementalLifecycle(t *testing.T) {
 	if call("GET", base+"/model", nil, "", 200)["model"].(map[string]any)["complete"] != true {
 		t.Fatal("initial read has no explicit coverage")
 	}
+	// Negotiate the same authenticated canonical read as a streamed transfer.
+	streamRequest := httptest.NewRequest("GET", base+"/model", nil)
+	streamRequest.Header.Set("Authorization", "Bearer "+token)
+	streamRequest.Header.Set("Accept", modelStreamType)
+	streamWriter := httptest.NewRecorder()
+	mux.ServeHTTP(streamWriter, streamRequest)
+	streamed := streamRecords(t, streamWriter)
+	if streamWriter.Code != 200 || streamWriter.Header().Get("Content-Type") != modelStreamType || len(streamed) < 2 || streamed[0]["projectId"] != root || streamed[len(streamed)-1]["kind"] != "complete" || !modelEqual(streamed[0]["versions"], read().Versions) {
+		t.Fatal("authenticated stream lost canonical snapshot coverage")
+	}
 	// An object created and removed inside one transaction still has a lifecycle fact.
 	ghostID := modelTestID(4000)
 	ghost := modelTestNode(ghostID, root)

internal/platform/canvas_model_stream.go

@@ -0,0 +1,91 @@
+package platform
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+)
+
+const modelStreamType = "application/vnd.foundry.canvas-model+ndjson;version=1"
+
+// The caller captures a coherent model while holding the workspace lock, then
+// releases its transaction before writing. A slow reader must not hold a writer.
+// This streams transfer/encoding; it does not stream database row acquisition.
+func writeCanvasModelStream(ctx context.Context, w http.ResponseWriter, model canvasModel) {
+	token := newModelID()
+	if token == "" {
+		reject(w, 500, "model_stream_unavailable")
+		return
+	}
+	view := model.view(nil)
+	header := response{"streamVersion": 1, "kind": "header", "sequence": 0, "snapshotToken": token,
+		"documentFormatVersion": canvasModelFormat, "projectId": model.Root, "canvases": view.Canvases,
+		"versions": view.Versions, "tombstones": view.Tombstones, "removedCanvasIds": view.Removed, "elementCount": len(view.Elements)}
+	const maxLine = 8 * 1024 * 1024
+	const maxBytes = 64 * 1024 * 1024
+	sequence, total := 0, 0
+	started := false
+	write := func(value response) bool {
+		if ctx.Err() != nil {
+			return false
+		}
+		raw, err := json.Marshal(value)
+		if err != nil || len(raw) > maxLine || total+len(raw)+1 > maxBytes || sequence >= 4096 {
+			if !started {
+				reject(w, 413, "model_stream_budget")
+			}
+			return false
+		}
+		if !started {
+			w.Header().Set("Content-Type", modelStreamType)
+			w.Header().Set("Cache-Control", "no-store")
+			w.Header().Add("Vary", "Accept")
+			w.Header().Set("X-Content-Type-Options", "nosniff")
+			w.WriteHeader(http.StatusOK)
+			started = true
+		}
+		total += len(raw) + 1
+		if n, writeErr := w.Write(append(raw, '\n')); writeErr != nil || n != len(raw)+1 {
+			return false
+		}
+		sequence++
+		// A missing terminal record is an explicit incomplete stream at the client.
+		// Never append an unrelated JSON error after streaming has begun.
+		return http.NewResponseController(w).Flush() == nil
+	}
+	if !write(header) {
+		return
+	}
+	batch := []map[string]any{}
+	batchBytes := 0
+	flush := func() bool {
+		if len(batch) == 0 {
+			return true
+		}
+		if !write(response{"streamVersion": 1, "kind": "elements", "sequence": sequence, "snapshotToken": token, "elements": batch}) {
+			return false
+		}
+		batch, batchBytes = []map[string]any{}, 0
+		return true
+	}
+	for _, element := range view.Elements {
+		if ctx.Err() != nil {
+			return
+		}
+		raw, err := json.Marshal(element)
+		if err != nil {
+			return
+		}
+		if len(batch) > 0 && (len(batch) == 64 || batchBytes+len(raw) > 512*1024) {
+			if !flush() {
+				return
+			}
+		}
+		batch = append(batch, element)
+		batchBytes += len(raw)
+	}
+	if !flush() {
+		return
+	}
+	write(response{"streamVersion": 1, "kind": "complete", "sequence": sequence, "snapshotToken": token, "elementCount": len(view.Elements)})
+}

internal/platform/canvas_model_stream_test.go

@@ -0,0 +1,92 @@
+package platform
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http/httptest"
+	"testing"
+)
+
+func streamRecords(t *testing.T, writer *httptest.ResponseRecorder) []response {
+	t.Helper()
+	var records []response
+	scanner := bufio.NewScanner(writer.Body)
+	for scanner.Scan() {
+		var record response
+		if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
+			t.Fatal(err)
+		}
+		records = append(records, record)
+	}
+	if err := scanner.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return records
+}
+func TestCanonicalModelStreamFramesAndSnapshot(t *testing.T) {
+	model := modelTestDocument()
+	for i := 0; i < 129; i++ {
+		id := fmt.Sprintf("extra-%03d", i)
+		model.Elements[id] = modelTestNode(id, modelRootID)
+	}
+	model.Tombstones["deleted"] = 2
+	model.TombstoneOwners["deleted"] = modelRootID
+	writer := httptest.NewRecorder()
+	writeCanvasModelStream(context.Background(), writer, model)
+	if writer.Code != 200 || writer.Header().Get("Content-Type") != modelStreamType || !writer.Flushed {
+		t.Fatal("stream was not negotiated/flushed", writer.Code, writer.Header())
+	}
+	records := streamRecords(t, writer)
+	if len(records) != 5 || records[0]["kind"] != "header" || records[4]["kind"] != "complete" {
+		t.Fatal("missing stream framing", len(records))
+	}
+	token := modelString(records[0]["snapshotToken"])
+	if !uuidSlugPattern.MatchString(token) || records[0]["projectId"] != modelRootID || !modelEqual(records[0]["versions"], model.Versions) || !modelEqual(records[0]["tombstones"], model.Tombstones) {
+		t.Fatal("stream header lost snapshot identity")
+	}
+	seen := map[string]bool{}
+	for i, record := range records {
+		if record["sequence"] != float64(i) || record["snapshotToken"] != token || record["streamVersion"] != float64(1) {
+			t.Fatal("mixed or unordered snapshot")
+		}
+		if record["kind"] == "elements" {
+			items := record["elements"].([]any)
+			if len(items) == 0 || len(items) > 64 {
+				t.Fatal("unbounded record batch")
+			}
+			for _, item := range items {
+				element := item.(map[string]any)
+				id := modelString(element["id"])
+				if seen[id] || !modelEqual(element, model.Elements[id]) {
+					t.Fatal("duplicated or corrupted stream element")
+				}
+				seen[id] = true
+			}
+		}
+	}
+	if len(seen) != len(model.Elements) || records[4]["elementCount"] != float64(len(seen)) {
+		t.Fatal("terminal coverage mismatch")
+	}
+}
+
+type cancelStreamWriter struct {
+	*httptest.ResponseRecorder
+	cancel context.CancelFunc
+}
+
+func (writer cancelStreamWriter) Flush() {
+	writer.ResponseRecorder.Flush()
+	writer.cancel()
+}
+func TestCanonicalModelStreamCancellationNeverClaimsCompletion(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	writer := httptest.NewRecorder()
+	writeCanvasModelStream(ctx, cancelStreamWriter{writer, cancel}, modelTestDocument())
+	records := streamRecords(t, writer)
+	if len(records) != 1 || records[0]["kind"] != "header" {
+		t.Fatal("cancelled stream wrote a tail", len(records))
+	}
+}