diff --git a/app/ui/shell/pipeline_state.js b/app/ui/shell/pipeline_state.js index f5a55abe..112d6ae7 100644 --- a/app/ui/shell/pipeline_state.js +++ b/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. diff --git a/app/ui/shell/tests/pipeline_state.test.js b/app/ui/shell/tests/pipeline_state.test.js index 3f5786fb..7489effb 100644 --- a/app/ui/shell/tests/pipeline_state.test.js +++ b/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); +}); diff --git a/modules/system/taste/store.py b/modules/system/taste/store.py index f688fc90..e20f2a71 100644 --- a/modules/system/taste/store.py +++ b/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 + (`.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 diff --git a/modules/system/taste/tests/test_taste.py b/modules/system/taste/tests/test_taste.py index 12322cd7..de2d41ee 100644 --- a/modules/system/taste/tests/test_taste.py +++ b/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()