feat!: export gateway.Project, and return the whole StreamChange from applyStreamEvent

docs/saving.md

@@ -23,12 +23,47 @@ func (s *server) saveConfigMap(w http.ResponseWriter, r *http.Request) {
     }
 
     // Your app performs this ordinary Kubernetes PATCH as the signed-in user.
-    result, err := s.dynamicFor(user).Resource(configMaps).Namespace(scope.Namespace).
+    _, err := s.dynamicFor(user).Resource(configMaps).Namespace(scope.Namespace).
         Patch(r.Context(), scope.Name, types.MergePatchType, patch, metav1.PatchOptions{})
-    writeResult(w, result, err)
+    if err != nil {
+        http.Error(w, "save failed", http.StatusBadGateway)
+        return
+    }
+
+    // 204, and NOT the object Kubernetes just handed back. See below.
+    w.WriteHeader(http.StatusNoContent)
 }
 ```
 
+## Answer 204 and let the watch echo it
+
+This is the recommended shape, and it is the one to reach for unless you have a specific reason not
+to.
+
+The object returned by a Kubernetes write is a *raw* object: `managedFields`, the last-applied
+annotation, `status`, and the Secret values your projection withholds. Writing it to the response
+hands the browser, through your save endpoint, precisely what the stream spent its whole design
+refusing to send. The save endpoint is not covered by the projection unless you cover it.
+
+You do not need to. The write goes to the API server, the watch sees it, and it arrives back down the
+stream as an ordinary `modified` event — projected, redacted, three-way merged into the draft the user
+is still holding. The store converges on its own. Dirty state is derived from `draft` versus `server`,
+so there is nothing to clear and nothing to adopt: the echo settles it.
+
+## If you must answer with the object
+
+`store.adoptSaved(object)` exists for a host that already holds a **projected** object — a host doing
+its own optimistic update, or one that cannot wait a round-trip for the echo. Project it first:
+
+```go
+projected, redacted := gateway.Project(gateway.ProjectionFull, result)
+_ = redacted // the paths withheld; the client keeps the redactions it already has
+writeJSON(w, projected)
+```
+
+`gateway.Project` applies the same projection the stream applies. Never hand `adoptSaved` an object
+straight from the Kubernetes client.
+
 The guard rejects:
 
 - a value declared in `redacted`, including deletion of a parent map such as `data: null`;

examples/vanilla-browser/index.html

@@ -92,8 +92,10 @@ <h2>patch — RFC 7386, editable changes only</h2>
       let pending = new Set();
 
       connectWithEventSource(`/resource-stream/v1?fixture=${fixture}&pace=${pace}`, store, {
-        onChange: (flashed) => {
-          for (const p of flashed) pending.add(key(p));
+        // `change` names the resource (change.uid) as well as the paths, which is what a UI showing
+        // more than one object per stream needs — this demo shows one, so it only reads flashed.
+        onChange: (change) => {
+          for (const p of change.flashed) pending.add(key(p));
           render();
         },
         onSynced: () => { statusLine.textContent = "synced — live"; },

gateway/patch_test.go

@@ -53,3 +53,55 @@ func TestValidateMergePatchHonorsProjectionAndPatchShape(t *testing.T) {
 		}
 	}
 }
+
+// Project is exported because the SAVE path needs it, and until it was, a host answering a save with
+// the written object had no supported way to make that object safe. The obvious thing — hand back
+// what the Kubernetes client returned — leaks exactly what the stream withholds, through the one
+// endpoint the stream does not guard.
+//
+// So this asserts the property a host is relying on: what Project returns is safe to send to a
+// browser, and it names what it withheld.
+func TestProjectMakesASavedObjectSafeToReturn(t *testing.T) {
+	// What a Kubernetes write actually hands back.
+	raw := KRMObject{
+		"apiVersion": "v1",
+		"kind":       "Secret",
+		"metadata": map[string]any{
+			"uid":  "u-1",
+			"name": "creds",
+			"managedFields": []any{
+				map[string]any{"manager": "kubectl"},
+			},
+			"annotations": map[string]any{
+				lastAppliedAnnotation: `{"apiVersion":"v1"}`,
+			},
+		},
+		"data": map[string]any{"token": "s3cret"},
+	}
+
+	projected, redacted := Project(ProjectionFull, raw)
+
+	meta, _ := projected["metadata"].(map[string]any)
+	if _, found := meta["managedFields"]; found {
+		t.Error("managedFields survived Project — a save response would put it in the browser")
+	}
+	if ann, found := meta["annotations"].(map[string]any); found {
+		if _, found := ann[lastAppliedAnnotation]; found {
+			t.Error("last-applied-configuration survived Project")
+		}
+	}
+
+	data, _ := projected["data"].(map[string]any)
+	if got, found := data["token"]; found {
+		t.Errorf("the Secret VALUE survived Project: %v — this is the leak the projection exists to prevent", got)
+	}
+	if len(redacted) != 1 || redacted[0] != "/data/token" {
+		t.Errorf("redacted = %v, want [/data/token] — a consumer must still learn the key exists", redacted)
+	}
+
+	// And the input is untouched: the caller's object may be an informer's, shared with every other
+	// stream on the same scope.
+	if raw["data"].(map[string]any)["token"] != "s3cret" {
+		t.Error("Project mutated its input")
+	}
+}

gateway/project.go

@@ -47,6 +47,36 @@ type redactedValue struct {
 	value any
 }
 
+// Project applies a projection to an object, returning the object as it would go on the wire and the
+// RFC 6901 pointers whose values were withheld from it.
+//
+// This exists because the SAVE path needs it. A host that answers a save with the written object must
+// answer with a PROJECTED one — hand a browser what Kubernetes returned and you have just sent it
+// managedFields and the Secret value that the whole projection exists to withhold, through the one
+// endpoint the stream does not guard. Before this was exported there was no supported way to do that,
+// which made `store.adoptSaved(object)` a trap: the obvious thing to reach for, and no safe way to
+// feed it.
+//
+// The recommended save shape is still to answer 204 and let the watch echo the write back (see
+// docs/saving.md): the store converges on its own, dirty state is derived, and nothing needs
+// projecting. Use this when a host has a reason to hand the object back directly.
+//
+// # Why []string and not []Redaction
+//
+// Redaction carries a Rev, and Rev is not a property of an object — it is a counter the STREAM keeps
+// per uid, incremented when a withheld value changes underneath a consumer that cannot see it. There
+// is no honest Rev to give you here, and inventing a zero would be a field that looks like an answer.
+// The paths are the part that is true of the object alone. `adoptSaved` needs no Rev either: it keeps
+// the redactions the store already has.
+func Project(p Projection, in KRMObject) (KRMObject, []string) {
+	out, values := project(p, in)
+	paths := make([]string, 0, len(values))
+	for _, v := range values {
+		paths = append(paths, v.path)
+	}
+	return out, paths
+}
+
 func project(p Projection, in KRMObject) (KRMObject, []redactedValue) {
 	out := deepCopyObject(in)
 	redacted := []redactedValue{}

packages/krm-stream/e2e/wire.ts

@@ -89,8 +89,8 @@ for (const f of fixtures) {
     const connections = (f.watch ?? []).filter((op) => (op as { op: string }).op === "disconnect").length + 1;
     for (let conn = 0; conn < connections; conn++) {
       const handle = connectResourceStream(url(f.id, conn), store, {
-        onChange: (paths) => {
-          flashed.push(...paths);
+        onChange: (change) => {
+          flashed.push(...change.flashed);
           onEvent();
         },
         onError: () => onEvent(), // an `error` event occupies an index too (see resync-midstream)

packages/krm-stream/src/index.ts

@@ -21,7 +21,7 @@ export { get, has, isPrefix, parsePointer, pathKey } from "./path.ts";
 export { DEFAULT_EDITABLE_REGIONS, defaultPolicy, readOnlyPolicy, regionPolicy } from "./policy.ts";
 export type { KubernetesStructuralSchema } from "./schema.ts";
 export { withOpenAPIKeyedLists } from "./schema.ts";
-export type { StreamHandle, StreamOptions } from "./sse.ts";
+export type { StreamChange, StreamHandle, StreamOptions } from "./sse.ts";
 export { applyStreamEvent, connectResourceStream, connectWithEventSource, SSEDecoder, StreamSequence } from "./sse.ts";
 export type { ApplyOptions, ApplyResult } from "./store.ts";
 export { LiveResourceStore } from "./store.ts";

packages/krm-stream/src/sse.ts

@@ -14,7 +14,7 @@
 // be allowed to see — forever, from every open tab.
 
 import type { LiveResourceStore } from "./store.ts";
-import type { ErrorCode, Path, StreamEvent } from "./types.ts";
+import type { ErrorCode, EventType, Path, StreamEvent } from "./types.ts";
 
 /** Incremental SSE parser. Bytes arrive in whatever chunks the network feels like — a frame can be
  * split down the middle, and it WILL be, under exactly the load where you least want to debug it —
@@ -92,37 +92,66 @@ function parseFrame(frame: string): StreamEvent | null {
  * it is exported because it IS the protocol — a host feeding a store from its own transport should
  * not have to reimplement the switch and get `synced` subtly wrong.
  *
- * Returns the paths that flashed, so a UI can highlight them. */
-export function applyStreamEvent(store: LiveResourceStore, ev: StreamEvent): Path[] {
+ * Returns the complete StreamChange — which resource, and what happened to it. */
+export function applyStreamEvent(store: LiveResourceStore, ev: StreamEvent): StreamChange {
   switch (ev.type) {
     case "reset":
       store.beginSnapshot();
-      return [];
+      return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] };
     case "added":
-    case "modified":
-      if (!ev.object) return [];
-      return store.applyServerEvent(ev.object, { redacted: ev.redacted }).flashed;
-    case "deleted":
-      if (ev.identity?.uid) store.removeResource(ev.identity.uid);
-      return [];
+    case "modified": {
+      if (!ev.object) return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] };
+      const result = store.applyServerEvent(ev.object, { redacted: ev.redacted });
+      return { type: ev.type, uid: ev.object.metadata.uid, ...result };
+    }
+    case "deleted": {
+      const uid = ev.identity?.uid;
+      if (uid) store.removeResource(uid);
+      // A removal IS structural: a row left the collection, and a UI that only re-reads values would
+      // keep rendering it.
+      return { type: ev.type, uid, added: false, structural: true, flashed: [], conflicts: [] };
+    }
     case "synced":
       store.endSnapshot();
-      return [];
+      return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] };
     default:
       // An unknown event type MUST be ignored, not treated as an error (spec §0). That is what lets
       // the gateway add an optional event type later without breaking a browser nobody can update.
-      return [];
+      return { type: ev.type, added: false, structural: false, flashed: [], conflicts: [] };
   }
 }
 
+/** What one stream event did to the store.
+ *
+ * This is the whole ApplyResult and the uid it belongs to, because anything less makes a host
+ * reimplement the stream loop to get the rest back. A UI rendering more than ONE resource per stream
+ * — which is most of them — cannot use a bare list of paths: it knows what moved and not what moved.
+ *
+ * Each field answers a question a renderer actually has:
+ *
+ *   uid         which resource. Absent only on reset/synced, which are about the stream, not an object.
+ *   added       an arrival, not a change. Animate it in; do not flash it as if a value moved.
+ *   structural  keys or rows appeared or disappeared. REBUILD the list; re-reading values is not enough.
+ *   flashed     the paths the server moved. Highlight these.
+ *   conflicts   the paths now conflicted, complete — not just the new ones.
+ */
+export interface StreamChange {
+  type: EventType;
+  uid?: string;
+  added: boolean;
+  structural: boolean;
+  flashed: Path[];
+  conflicts: Path[];
+}
+
 export interface StreamOptions {
   /** Called for every `error` event. A terminal one has already closed the connection by the time
    * this returns — there is nothing to retry, and retrying is the bug. */
   onError?: (code: ErrorCode, message: string, terminal: boolean) => void;
   /** Called at the end of every snapshot cycle. The store is now consistent: a good moment to paint. */
   onSynced?: () => void;
-  /** Called after any change, with the paths that flashed. */
-  onChange?: (flashed: Path[]) => void;
+  /** Called after any change, with what the event did and which resource it did it to. */
+  onChange?: (change: StreamChange) => void;
   /** A missing or duplicated event was observed. The connection is closed; reconnect for a snapshot. */
   onGap?: (expected: number, received: number) => void;
   /** Abort the stream from outside. */
@@ -238,8 +267,8 @@ function feed(store: LiveResourceStore, sequence: StreamSequence, ev: StreamEven
     opts.onError?.(ev.code ?? "INTERNAL", ev.message ?? "", ev.terminal ?? false);
     return ev.terminal === true;
   }
-  const flashed = applyStreamEvent(store, ev);
+  const change = applyStreamEvent(store, ev);
   if (ev.type === "synced") opts.onSynced?.();
-  opts.onChange?.(flashed);
+  opts.onChange?.(change);
   return false;
 }

packages/krm-stream/src/store.ts

@@ -139,6 +139,17 @@ export class LiveResourceStore {
   /** The save succeeded and this is the object it produced. The watch will echo it too — and that
    * echo is a harmless no-op (I-IDEMPOTENT) — but a UI should not have to wait for it to stop
    * showing the field as dirty. */
+  /** Adopt the object a save returned.
+   *
+   * `object` MUST be projected — the same projection the stream uses. An object straight from a
+   * Kubernetes client carries managedFields, status and the Secret values the projection withholds,
+   * and handing it here puts all of them in the browser through the one endpoint the stream does not
+   * guard. Project it on the server with `gateway.Project` first.
+   *
+   * You probably do not need this. The recommended save is 204: the write reaches the API server, the
+   * watch echoes it back down the stream already projected, and the store converges. Dirty state is
+   * derived from draft-versus-server, so there is nothing to adopt. Use this only when a host cannot
+   * wait for the echo. See docs/saving.md. */
   adoptSaved(object: KRMObject): void {
     const existing = this.#resources.get(object.metadata.uid);
     if (!existing) {

packages/krm-stream/test/store.test.ts

@@ -27,7 +27,7 @@ for (const f of clientFixtures()) {
       // applyStreamEvent is library code, not test code: the switch from event to store call IS the
       // protocol (spec §4), and a host feeding a store from its own transport must not have to
       // reimplement it and get `synced` subtly wrong.
-      flashed.push(...applyStreamEvent(store, resolve(f, fe)));
+      flashed.push(...applyStreamEvent(store, resolve(f, fe)).flashed);
 
       for (const edit of f.client?.edits ?? []) {
         if (edit.after === i) applyEdit(store, edit);
@@ -45,3 +45,66 @@ for (const f of clientFixtures()) {
 test("the client suite actually ran the corpus", () => {
   assert.ok(clientFixtures().length >= 10, "the client half of the corpus is missing fixtures");
 });
+
+// The gap that sent the first host to adopt this back to writing its own EventSource loop.
+//
+// applyStreamEvent used to return Path[] — the flashed paths and nothing else — which is unusable the
+// moment a stream carries more than one resource: a UI learns that something moved, and not what. So
+// a host that wanted to highlight per-resource had to abandon connectWithEventSource, drive its own
+// EventSource, and reimplement the event switch. That is precisely the work this library exists to do
+// once, correctly, on everyone's behalf.
+//
+// It returns the whole StreamChange now. This pins each field, because each one was being computed in
+// the store and thrown away at the seam.
+test("a change says WHICH resource changed, and what kind of change it was", () => {
+  const store = new LiveResourceStore();
+  const object = {
+    apiVersion: "v1",
+    kind: "ConfigMap",
+    metadata: { uid: "u-1", name: "app", namespace: "default", resourceVersion: "1" },
+    data: { greeting: "hello" },
+  };
+
+  store.beginSnapshot();
+  const arrival = applyStreamEvent(store, { seq: 1, type: "added", object });
+  assert.equal(arrival.uid, "u-1", "an added event must name the resource it added");
+  assert.equal(arrival.added, true, "an arrival is not a change to an existing object");
+
+  // A value moving is NOT structural: the keys are the same, so a UI re-reads rather than rebuilds.
+  const moved = applyStreamEvent(store, {
+    seq: 2,
+    type: "modified",
+    object: { ...object, metadata: { ...object.metadata, resourceVersion: "2" }, data: { greeting: "hi" } },
+  });
+  assert.equal(moved.uid, "u-1");
+  assert.equal(moved.added, false, "a modification of a known uid is not an arrival");
+  assert.equal(moved.structural, false, "no key appeared or disappeared");
+  // metadata.resourceVersion flashes too — it is a read-only field and it genuinely moved. Assert on
+  // the one this test is about rather than pinning the whole set, which would be a test of the
+  // flashing rules and those have their own fixtures.
+  assert.ok(
+    moved.flashed.some((p) => p.join("/") === "data/greeting"),
+    `the changed value must flash: ${JSON.stringify(moved.flashed)}`,
+  );
+
+  // A key APPEARING is structural: a renderer that only re-reads known values never shows it.
+  const grew = applyStreamEvent(store, {
+    seq: 3,
+    type: "modified",
+    object: {
+      ...object,
+      metadata: { ...object.metadata, resourceVersion: "3" },
+      data: { greeting: "hi", farewell: "bye" },
+    },
+  });
+  assert.equal(grew.structural, true, "a new key must be reported as structural, or the row is never drawn");
+
+  // And a delete names the uid too — a host cannot remove a row it cannot identify.
+  const gone = applyStreamEvent(store, {
+    seq: 4,
+    type: "deleted",
+    identity: { uid: "u-1", apiVersion: "v1", kind: "ConfigMap", name: "app", namespace: "default" },
+  });
+  assert.equal(gone.uid, "u-1", "a deleted event must name the resource it removed");
+  assert.equal(gone.structural, true, "a row left the collection");
+});

packages/krm-stream/test/wire.test.ts

@@ -49,7 +49,7 @@ for (const f of wireFixtures()) {
     const store = new LiveResourceStore();
     const flashed: Path[] = [];
     for (const [i, ev] of events.entries()) {
-      flashed.push(...applyStreamEvent(store, ev));
+      flashed.push(...applyStreamEvent(store, ev).flashed);
       for (const edit of f.client?.edits ?? []) {
         if (edit.after === i) applyEdit(store, edit);
       }