fix: values typed during a node change survive its answer; preset/import releases the run wait; a group move outlives a queued save; corrupt captures don't block a new Generate

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/ui/shell/pipeline_state.js

@@ -218,6 +218,12 @@ export function createPipelineState(rawApi, { answerMs = ANSWER_MS } = {}) {
       if (slot && kv && typeof kv === "object" && !Array.isArray(kv)) slot.inputs = { ...(slot.inputs || {}), ...kv };
     });
   }
+  // An answer replaces the slots with the server's copy of what was SENT: whatever was typed while it was on its way
+  // (node values still queued, module values) goes back on, or the screen and the next run lose it.
+  function keepTyped() {
+    placeInputs(pendingBody && pendingBody.inputs);
+    if (Object.keys(pendingValues).length) mirrorValues(currentValues());
+  }
   async function save(body) {
     if (adoptGate) await adoptGate;                     // an edit made while a project's pipeline goes in lands on THAT pipeline
     if (blocked()) return;
@@ -237,8 +243,7 @@ export function createPipelineState(rawApi, { answerMs = ANSWER_MS } = {}) {
         // this answer is about the old one and must not overwrite the new.
         if (mine !== epoch) continue;
         if (res && res.slots) slots = res.slots;
-        placeInputs(pendingBody && pendingBody.inputs);   // typed while this was in flight: still what is on screen
-        if (Object.keys(pendingValues).length) mirrorValues(currentValues());    // an edit queued behind this one stays in the slots a run reads
+        keepTyped();    // an edit queued behind this one stays in the slots a run reads
         applyGroups();
         incomplete = (res && res.incomplete) || [];
         refused = (res && res.refused) || [];
@@ -443,7 +448,7 @@ export function createPipelineState(rawApi, { answerMs = ANSWER_MS } = {}) {
       const res = await API.editPipeline({ slots, ...body });
       if (mine !== epoch) return { refused: ["The project was changed while this was being sent."] };
       refusedNow = (res && res.refused) || [];
-      if (res && res.slots && !refusedNow.length) { slots = res.slots; applyGroups(); }
+      if (res && res.slots && !refusedNow.length) { slots = res.slots; keepTyped(); applyGroups(); }
       incomplete = (res && res.incomplete) || [];
       refused = refusedNow;
       queueable = !!(res && res.queueable);
@@ -506,6 +511,7 @@ export function createPipelineState(rawApi, { answerMs = ANSWER_MS } = {}) {
       refusedNow = (res && res.refused) || [];
       if (res && res.slots && !refusedNow.length) {
         slots = res.slots;
+        keepTyped();
         // A default slot the new pipeline lacks is removed, whatever the caller said: otherwise the next
         // open lays the default back over it (a preset, an import or the wizard would regain the default's loaders).
         const kept = new Set(res.slots.map((s) => s.id));
@@ -522,7 +528,7 @@ export function createPipelineState(rawApi, { answerMs = ANSWER_MS } = {}) {
     } catch (e) {
       refusedNow = [e && e.message ? e.message : String(e)];
     } finally {
-      saving = false;
+      saving = false; shaping = false;
     }
     if (refill) await save({});
     return { refused: refusedNow };
@@ -543,7 +549,8 @@ export function createPipelineState(rawApi, { answerMs = ANSWER_MS } = {}) {
     groupEdits[slotId] = String(group || "").trim();
     applyGroups();
     const mine = groupEdits[slotId];
-    return save({}).then(() => { if (groupEdits[slotId] === mine) delete groupEdits[slotId]; });
+    // save() returns once queued behind a save in flight: let go of the move only when the queue has landed
+    return save({}).then(() => settled()).then(() => { if (groupEdits[slotId] === mine) delete groupEdits[slotId]; });
   }
 
   // Whoever keeps the pipeline with the project hears every landed edit.

app/ui/shell/tests/pipeline_state.test.js

@@ -11,6 +11,7 @@ function load(posts, opts = {}) {
   const API = {
       pipeline: async () => ({ slots: JSON.parse(JSON.stringify(defaults)), incomplete: [], refused: [], queueable: true }),
       editPipeline: async (body) => {
+        body = JSON.parse(JSON.stringify(body));     // sent over HTTP: the server never shares the page's objects
         posts.push(body);
         if (opts.delay) await new Promise((r) => setTimeout(r, opts.delay));
         const bad = (body.slots || []).findIndex((s) => typeof s.node !== "string");
@@ -662,3 +663,26 @@ test("a run does not wait on a value save (the value is already live), but does
   assert.ok(Date.now() - t1 > 100, "a node change is waited for");
   await shape;
 });
+
+test("a value typed while a node change is on its way survives the change's answer, and the run sees it", async () => {
+  const PS = load([], { delay: 60 });
+  await PS.ensureLoaded();
+  const shape = PS.edit({ remove: [] });
+  await new Promise((r) => setTimeout(r, 10));
+  PS.save({ inputs: { gen: { steps: 33 } } });                       // typed during it
+  await PS.readyToRun();
+  assert.strictEqual(PS.slots().find((s) => s.id === "gen").inputs.steps, 33);
+  await shape;
+});
+
+test("after a preset or import goes in, a run is ready at once (the node-change wait is released)", async () => {
+  const PS = load([], { delay: 5 });
+  await PS.ensureLoaded();
+  const next = JSON.parse(JSON.stringify(PS.slots()));
+  next.find((s) => s.id === "gen").inputs.steps = 40;               // a different pipeline: restore really sends it
+  await PS.restore({ slots: next });
+  assert.strictEqual(PS.slots().find((s) => s.id === "gen").inputs.steps, 40);
+  const t0 = Date.now();
+  assert.strictEqual(await PS.readyToRun(), true);
+  assert.ok(Date.now() - t0 < 100);
+});

modules/system/taste/store.py

@@ -171,6 +171,17 @@ def capture(key, kind, rows, prompt_id=None, keep=MAX_ROWS, only=None, mixed=Fal
         _latest_path().write_text(json.dumps({"key": key, "prompt_id": prompt_id, "runs": runs, "forgot": state.get("forgot", [])}))
 
 
+def _held_by(path, prompt_id):
+    """Whether a waiting capture is `prompt_id`'s. An earlier run's carries its id in the name; only the latest
+    (`<kind>.pending.pt`) is opened, and one that cannot be read is nobody's (it must not stop the rest going)."""
+    if path.name.count(".") > 2:
+        return path.name.split(".")[1] == prompt_id
+    try:
+        return (_read(path, None) or {}).get("prompt_id") == prompt_id
+    except Exception:  # noqa: BLE001
+        return False
+
+
 def new_generation(keep=None):
     """A Generate is starting: every capture still waiting for a rating is forgotten, but `keep`'s (the run this
     Generate already queued, which may have captured before this call). -> how many were dropped."""
@@ -178,7 +189,7 @@ def new_generation(keep=None):
         dropped = 0
         for key in keys():
             for path in _dir(key).glob("*.pending.pt"):
-                if keep is not None and (_read(path, None) or {}).get("prompt_id") == keep:
+                if keep is not None and _held_by(path, keep):
                     continue
                 path.unlink(missing_ok=True)
                 dropped += 1

modules/system/taste/tests/test_taste.py

@@ -269,3 +269,10 @@ def test_a_new_generate_spares_the_run_it_already_queued():
     store.new_generation(keep="mine")
     assert store.rate("mine", "liked")["recorded"] == ["vel"]
     assert store.rate("old", "liked")["reason"] == "forgotten"
+
+
+def test_an_unreadable_waiting_capture_does_not_stop_a_new_generate_forgetting_the_rest():
+    store.capture("k", "vel", {"a": torch.zeros(2)}, prompt_id="old")
+    store._pending_path("k", "vel").write_bytes(b"not a tensor file")
+    assert store.new_generation(keep="mine") >= 1
+    assert not store._pending_path("k", "vel").exists()