diff --git a/core/graph.py b/core/graph.py
index be403d0c..39dea3fa 100644
--- a/core/graph.py
+++ b/core/graph.py
@@ -55,6 +55,11 @@ class Schemas:
         found = self.of(class_type)
         return dict(found["inputs"]) if found else {}
 
+    def required(self, class_type: str) -> List[str]:
+        """Inputs that must be filled. Absent from a schema means none are."""
+        found = self.of(class_type)
+        return list(found.get("required", [])) if found else []
+
 
 def from_comfyui() -> Schemas:
     """Schemas read from whatever ComfyUI has registered, FunPack's and everyone
@@ -65,14 +70,20 @@ def from_comfyui() -> Schemas:
         if node is None:
             return None
         spec = node.INPUT_TYPES()
-        inputs = {}
+        inputs, required = {}, []
         for section in ("required", "optional"):
             for name, declared in (spec.get(section) or {}).items():
                 kind = declared[0] if isinstance(declared, (list, tuple)) else declared
                 # A list of choices IS the type for a combo; what matters here is
                 # only that it is not a socket, so it cannot be wired.
                 inputs[name] = kind if isinstance(kind, str) else "COMBO"
-        return {"inputs": inputs, "outputs": list(node.RETURN_TYPES)}
+                if section == "required":
+                    # Kept, because flattening the two sections loses the only
+                    # thing that says a slot is incomplete -- and a slot missing
+                    # a required input built clean and failed once queued.
+                    required.append(name)
+        return {"inputs": inputs, "outputs": list(node.RETURN_TYPES),
+                "required": required}
     return Schemas(lookup)
 
 
@@ -130,6 +141,11 @@ def build(slots: Sequence[dict], schemas: Optional[Schemas] = None) -> Tuple[dic
                 inputs[name] = [source, index]
             else:
                 inputs[name] = value
+        for name in schemas.required(class_type):
+            if name not in inputs:
+                problems.append(f"{slot_id}: {class_type} needs {name!r} and nothing "
+                                f"fills it")
+
         prompt[slot_id] = {"class_type": class_type, "inputs": inputs}
 
     return prompt, problems
diff --git a/core/routes.py b/core/routes.py
index 0855e33b..0d2a2819 100644
--- a/core/routes.py
+++ b/core/routes.py
@@ -5,7 +5,7 @@ so the real routes can be mounted on a throwaway app in a test. The handlers are
 thin adapters over pure functions in `serve`.
 """
 
-from . import config, log, registry as registry_mod, serve as static
+from . import config, graph as graph_mod, log, registry as registry_mod, serve as static
 from .contract import CONTRACT_VERSION
 from .relations import order
 from .traits import split
@@ -92,6 +92,63 @@ def register(routes, prefix=None):
         traits = [t for t in raw.split(",") if t] if raw is not None else None
         return web.json_response(manifest(traits))
 
+    def _pipeline():
+        """Whatever module offers a default pipeline, or nothing.
+
+        Core does not hold one: a default pipeline is feature content, and the
+        point of it being data is that it can be replaced by other data.
+        """
+        for _spec, make in modules().providers("default_pipeline"):
+            return make()
+        return []
+
+    @routes.get(P + "/api/pipeline")
+    async def _pipeline_get(_req):
+        slots = _pipeline()
+        prompt, incomplete = graph_mod.build(slots)
+        return web.json_response({"slots": slots, "refused": [],
+                                  "incomplete": incomplete,
+                                  "queueable": not incomplete})
+
+    @routes.post(P + "/api/pipeline")
+    async def _pipeline_edit(req):
+        """Replace or remove a slot, and say what that did to the graph.
+
+        This is what makes "a built-in node can be swapped or removed" a thing
+        the running server can do rather than a property of a library nothing
+        calls. Refusals come back as data, so the app can show the reason.
+        """
+        try:
+            body = await req.json()
+        except Exception:  # noqa: BLE001
+            return web.json_response({"problems": ["that is not JSON"]}, status=400)
+
+        slots = body.get("slots") or _pipeline()
+        action, slot_id = body.get("action"), body.get("slot")
+
+        if action == "replace":
+            slots, problems = graph_mod.replace(slots, slot_id, body.get("node"))
+        elif action == "remove":
+            slots, problems = graph_mod.remove(slots, slot_id)
+        elif action in (None, "check"):
+            problems = []
+        else:
+            return web.json_response({"problems": [f"unknown action {action!r}"]}, status=400)
+
+        # Two different things, kept apart. "refused" means the edit did not
+        # happen. "incomplete" means it did and the pipeline still is not ready
+        # -- an unset file picker on a fresh install is the normal case, not a
+        # failed edit, and an app that showed them together would say the wrong
+        # thing about both.
+        prompt, incomplete = graph_mod.build(slots)
+        return web.json_response({
+            "slots": slots,
+            "refused": problems,
+            "incomplete": incomplete,
+            "queueable": not (problems or incomplete),
+            "prompt": prompt if not (problems or incomplete) else None,
+        })
+
     @routes.get(P + "/api/log")
     async def _log(req):
         level = req.query.get("level") or None
diff --git a/core/schema.py b/core/schema.py
index a367198f..8fcd80f8 100644
--- a/core/schema.py
+++ b/core/schema.py
@@ -194,6 +194,28 @@ def validate_traits(value: Any, module_id: str):
     return value
 
 
+def refuse_or_warn(problems: List[str], where: str, guards_off: bool) -> None:
+    """Raise, or say it and carry on.
+
+    Full control means the user asked for the raw behaviour -- a one-step
+    schedule, a width the model cannot use -- so a refusal becomes a warning.
+    It is never silent: a run made without the guards has to be identifiable as
+    one afterwards, because the consequences land later than the choice.
+    """
+    if not problems:
+        return
+    if not guards_off:
+        # RuntimeError, not SchemaError: SchemaError means a DECLARATION is
+        # wrong and the module does not load. This is a value supplied at run
+        # time, and the run is what stops.
+        raise RuntimeError(f"{where}:\n  " + "\n  ".join(problems))
+
+    from . import log
+    for problem in problems:
+        log.warning(where, f"{problem} -- allowed because full control is on, "
+                           f"and the consequences are yours")
+
+
 def validate_provides(value: Any, module_id: str) -> dict:
     """Named capabilities: {name: callable}."""
     if value is None:
@@ -211,12 +233,20 @@ def validate_provides(value: Any, module_id: str) -> dict:
     return dict(value)
 
 
-def check_values(spec, values: Any) -> Tuple[Dict[str, Any], List[str]]:
+def check_values(spec, values: Any, keep_bad: bool = False) -> Tuple[Dict[str, Any], List[str]]:
     """One module's live values, checked against its own declaration.
 
     Returns (values, problems). Anything missing falls back to the declared
     default, so a partial payload is fine; anything WRONG is reported rather than
-    coerced. v4 carried these as JSON in a string widget that nothing validated,
+    coerced.
+
+    `keep_bad` is full control: the problem is still reported, but the value is
+    passed through instead of being replaced by the default. Reporting a
+    complaint and then quietly using the default would be the worst of both --
+    the user asked for 999 and would get 0.4 with a warning about 999.
+
+    A key NO module declares is never kept, whatever the setting: nothing reads
+    it, so keeping it cannot honour the request, only disguise the typo. v4 carried these as JSON in a string widget that nothing validated,
     which is why it needed a hand-maintained mirror of every default and a list of
     keys that no longer meant anything.
     """
@@ -237,7 +267,8 @@ def check_values(spec, values: Any) -> Tuple[Dict[str, Any], List[str]]:
         problem = _value_problem(declared, key, value, spec.id)
         if problem:
             problems.append(problem)
-            continue
+            if not keep_bad:
+                continue
         clean[key] = value
     return clean, problems
 
diff --git a/core/tests/test_routes_pipeline.py b/core/tests/test_routes_pipeline.py
new file mode 100644
index 00000000..8bc2e770
--- /dev/null
+++ b/core/tests/test_routes_pipeline.py
@@ -0,0 +1,149 @@
+"""The pipeline, over HTTP.
+
+core/graph.py was well tested and called from nowhere -- proven correct in a
+vacuum while the claim it backs ("a built-in node can be replaced or removed")
+was not something the running server could do. These tests are about
+reachability as much as behaviour.
+"""
+
+import json
+import socket
+import threading
+
+import pytest
+
+pytest.importorskip("aiohttp")
+
+from aiohttp import web  # noqa: E402
+
+from core import routes  # noqa: E402
+
+
+@pytest.fixture(scope="module")
+def registered(comfyui):
+    """FunPack's nodes registered the way ComfyUI registers them.
+
+    Without this the pipeline's slots all look like missing nodes, because the
+    schema lookup reads ComfyUI's registry and nothing had put anything in it.
+    """
+    import asyncio
+    import nodes as comfy_nodes
+    from pathlib import Path
+
+    async def load():
+        await comfy_nodes.init_extra_nodes(init_custom_nodes=False)
+        await comfy_nodes.load_custom_node(
+            str(Path(__file__).resolve().parents[2]), module_parent="custom_nodes")
+
+    asyncio.run(load())
+    return comfy_nodes
+
+
+@pytest.fixture(scope="module")
+def server(comfyui, registered):
+    """The real route table on a throwaway app, on a real socket."""
+    from aiohttp import web as aioweb
+
+    app = aioweb.Application()
+    table = aioweb.RouteTableDef()
+    routes.register(table, prefix="/funpack")
+    app.add_routes(table)
+
+    holder = {}
+
+    def run():
+        import asyncio
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        runner = aioweb.AppRunner(app)
+        loop.run_until_complete(runner.setup())
+        site = aioweb.TCPSite(runner, "127.0.0.1", 0)
+        loop.run_until_complete(site.start())
+        holder["port"] = site._server.sockets[0].getsockname()[1]
+        holder["loop"] = loop
+        holder["ready"].set()
+        loop.run_forever()
+
+    holder["ready"] = threading.Event()
+    thread = threading.Thread(target=run, daemon=True)
+    thread.start()
+    assert holder["ready"].wait(20), "the test server did not start"
+    yield holder["port"]
+    holder["loop"].call_soon_threadsafe(holder["loop"].stop)
+
+
+def _request(port, method, path, body=None):
+    payload = b"" if body is None else json.dumps(body).encode()
+    head = (f"{method} {path} HTTP/1.1\r\nHost: localhost\r\n"
+            f"Content-Type: application/json\r\n"
+            f"Content-Length: {len(payload)}\r\nConnection: close\r\n\r\n").encode()
+    with socket.create_connection(("127.0.0.1", port), timeout=30) as sock:
+        sock.sendall(head + payload)
+        chunks = []
+        while True:
+            got = sock.recv(65536)
+            if not got:
+                break
+            chunks.append(got)
+    raw = b"".join(chunks)
+    header, _, rest = raw.partition(b"\r\n\r\n")
+    status = int(header.split()[1])
+    try:
+        return status, json.loads(rest.decode() or "{}")
+    except ValueError:
+        return status, {"raw": rest.decode(errors="replace")}
+
+
+def test_the_default_pipeline_is_served(server):
+    status, body = _request(server, "GET", "/funpack/api/pipeline")
+    assert status == 200
+    assert [slot["id"] for slot in body["slots"]], "no slots came back"
+    assert "queueable" in body
+    # A fresh install has no model picked, so the default pipeline is
+    # incomplete rather than wrong -- and the two are reported separately.
+    assert body["refused"] == []
+    assert any("nothing fills it" in p for p in body["incomplete"])
+
+
+def test_a_built_in_slot_can_be_removed_over_http(server):
+    """The claim, exercised the way the app will: the modifiers node taken out,
+    and what it fed rewired to what fed it."""
+    status, body = _request(server, "POST", "/funpack/api/pipeline",
+                            {"action": "remove", "slot": "modifiers"})
+    assert status == 200, body
+    assert body["refused"] == [], body["refused"]
+    ids = [slot["id"] for slot in body["slots"]]
+    assert "modifiers" not in ids
+
+    sampler = next(s for s in body["slots"] if s["id"] == "sampler")
+    assert sampler["inputs"]["model"] == ["model", 0], "the sampler was not rewired"
+
+
+def test_a_removal_that_cannot_work_comes_back_as_a_reason(server):
+    status, body = _request(server, "POST", "/funpack/api/pipeline",
+                            {"action": "remove", "slot": "latent"})
+    assert status == 200
+    assert any("LATENT" in p for p in body["refused"])
+    assert body["queueable"] is False
+    assert body["prompt"] is None
+
+
+def test_a_built_in_slot_can_be_replaced_over_http(server):
+    status, body = _request(server, "POST", "/funpack/api/pipeline",
+                            {"action": "replace", "slot": "vae", "node": "VAELoader"})
+    assert status == 200, body
+    vae = next(s for s in body["slots"] if s["id"] == "vae")
+    assert vae["node"] == "VAELoader"
+
+
+def test_a_replacement_that_gives_the_wrong_thing_is_refused(server):
+    status, body = _request(server, "POST", "/funpack/api/pipeline",
+                            {"action": "replace", "slot": "model", "node": "SaveImage"})
+    assert status == 200
+    assert body["refused"], "a nonsense replacement was accepted"
+
+
+def test_an_unknown_action_is_refused(server):
+    status, body = _request(server, "POST", "/funpack/api/pipeline",
+                            {"action": "explode", "slot": "model"})
+    assert status == 400 and body["problems"]  # a malformed request, not an edit
diff --git a/modules/sampling/modifiers/nodes.py b/modules/sampling/modifiers/nodes.py
index 34bbce41..b3048559 100644
--- a/modules/sampling/modifiers/nodes.py
+++ b/modules/sampling/modifiers/nodes.py
@@ -87,7 +87,18 @@ class FunPackModifierSettings(io.ComfyNode):
         the only place it can be understood before the queue accepts it.
         """
         problems = cls._problems(settings)
-        return True if not problems else "FunPack Modifier Settings: " + "; ".join(problems)
+        if not problems:
+            return True
+        # With the guards off a bad value is the user's business, so the queue
+        # accepts it and the warning is said at execution instead.
+        import json as _json
+        try:
+            raw = _json.loads((settings or "").strip() or "{}")
+        except ValueError:
+            raw = {}
+        if cls._guards_off(raw):
+            return True
+        return "FunPack Modifier Settings: " + "; ".join(problems)
 
     @classmethod
     def _problems(cls, settings: str):
@@ -112,6 +123,14 @@ class FunPackModifierSettings(io.ComfyNode):
             problems.extend(said)
         return problems
 
+    @classmethod
+    def _guards_off(cls, raw) -> bool:
+        """Asked by capability, so nothing here names the module that answers."""
+        if not isinstance(raw, dict):
+            return False
+        return any(answer(raw) for _spec, answer
+                   in registry_mod.current().providers(GUARDS_OFF))
+
     @classmethod
     def execute(cls, settings: str) -> io.NodeOutput:
         text = (settings or "").strip() or "{}"
@@ -125,6 +144,7 @@ class FunPackModifierSettings(io.ComfyNode):
                 f"got {type(raw).__name__}.")
 
         specs = registry_mod.current().specs
+        guards_off = cls._guards_off(raw)
         problems, checked = [], {}
         for module_id, values in raw.items():
             spec = specs.get(module_id)
@@ -133,12 +153,13 @@ class FunPackModifierSettings(io.ComfyNode):
                 # nothing and a list of dead keys to remember.
                 problems.append(f"no module named {module_id!r} is installed.")
                 continue
-            clean, said = schema_mod.check_values(spec, values)
+            clean, said = schema_mod.check_values(spec, values, keep_bad=guards_off)
             checked[module_id] = clean
             problems.extend(said)
 
-        if problems:
-            raise RuntimeError("FunPack Modifier Settings:\n  " + "\n  ".join(problems))
+        # Full control reaches this node too: it is in the payload being checked.
+        schema_mod.refuse_or_warn(problems, "FunPack Modifier Settings",
+                                  guards_off=cls._guards_off(raw))
 
         named = ", ".join(sorted(checked)) or "nothing set"
         return io.NodeOutput(checked, f"settings for {named}")
@@ -202,6 +223,8 @@ class FunPackLoadModifiers(io.ComfyNode):
         # as guarded, the safe direction.
         full_control = any(answer(settings)
                            for _spec, answer in registry.providers(GUARDS_OFF))
+        # Read before the values are checked, because it changes what checking
+        # them DOES: refuse, or let it through and say so.
         if full_control:
             notes.append("full control is ON: modifiers are not guarded, and a "
                          "failing one will end the run. Any consequences of "
@@ -219,11 +242,12 @@ class FunPackLoadModifiers(io.ComfyNode):
         problems = []
         chosen = {}
         for spec in ordered:
-            values, said = schema_mod.check_values(spec, (settings or {}).get(spec.id))
+            values, said = schema_mod.check_values(
+                spec, (settings or {}).get(spec.id), keep_bad=full_control)
             chosen[spec.id] = values
             problems.extend(said)
-        if problems:
-            raise RuntimeError("FunPack Load Modifiers:\n  " + "\n  ".join(problems))
+        schema_mod.refuse_or_warn(problems, "FunPack Load Modifiers",
+                                  guards_off=full_control)
 
         for spec in ordered:
             values = chosen[spec.id]
diff --git a/modules/sampling/sampler/nodes.py b/modules/sampling/sampler/nodes.py
index eb91744f..8fe9f5ce 100644
--- a/modules/sampling/sampler/nodes.py
+++ b/modules/sampling/sampler/nodes.py
@@ -189,13 +189,15 @@ class FunPackSampler(io.ComfyNode):
         compatible, incompatible = traits_mod.split(offering, available)
         ordered, rejected = relations_mod.order(compatible)
 
+        guards_off = any(answer(settings or {}) for _spec, answer
+                         in registry.providers("guards_off"))
         values, problems = {}, []
         for spec in ordered:
-            clean, said = schema_mod.check_values(spec, (settings or {}).get(spec.id))
+            clean, said = schema_mod.check_values(
+                spec, (settings or {}).get(spec.id), keep_bad=guards_off)
             values[spec.id] = clean
             problems.extend(said)
-        if problems:
-            raise RuntimeError("FunPack Sampler:\n  " + "\n  ".join(problems))
+        schema_mod.refuse_or_warn(problems, "FunPack Sampler", guards_off=guards_off)
 
         chain, notes = chain_mod.build(ordered, values, ACCEPTS, dropped)
         for spec in incompatible:
diff --git a/modules/sampling/tests/test_modifiers.py b/modules/sampling/tests/test_modifiers.py
index 98446f36..255173b5 100644
--- a/modules/sampling/tests/test_modifiers.py
+++ b/modules/sampling/tests/test_modifiers.py
@@ -286,3 +286,68 @@ def test_a_setting_no_module_declares_is_refused_at_this_node_too(registry, patc
     })
     with pytest.raises(RuntimeError, match="no setting named"):
         FunPackLoadModifiers.execute(patcher, settings={"alg": {"strenght": 4.0}})
+
+
+# --- full control turns off the refusals too --------------------------------
+
+def _full_control(registry):
+    """The real module, so the capability is the one the product uses."""
+    from core.contract import ModuleSpec
+    from modules.system import full_control
+    registry.specs["full_control"] = ModuleSpec(
+        id="full_control", title="Full control", mount="settings.general",
+        settings=full_control.SETTINGS, provides=full_control.PROVIDES)
+
+
+def test_full_control_lets_an_out_of_range_value_through(registry, patcher):
+    """The stated intent: a one-step schedule, a width the model cannot use --
+    let them, and say the consequences are theirs."""
+    from core import log
+    from modules.sampling.modifiers.nodes import FunPackLoadModifiers
+
+    got = {}
+    registry.specs["m"] = _spec("m", lambda t, v, key: got.update(v) or "on", settings={
+        "amount": {"type": "float", "default": 0.4, "min": -1.0, "max": 1.0, "label": "A"},
+    })
+    _full_control(registry)
+
+    log._reset()
+    FunPackLoadModifiers.execute(patcher, settings={
+        "m": {"amount": 999.0},
+        "full_control": {"enabled": True},
+    })
+
+    assert got["amount"] == 999.0, "the value was refused with full control on"
+    assert any("consequences are yours" in r["message"] for r in log.history())
+
+
+def test_without_full_control_the_same_value_is_still_refused(registry, patcher):
+    from modules.sampling.modifiers.nodes import FunPackLoadModifiers
+
+    registry.specs["m"] = _spec("m", lambda t, v, key: "on", settings={
+        "amount": {"type": "float", "default": 0.4, "min": -1.0, "max": 1.0, "label": "A"},
+    })
+    _full_control(registry)
+
+    with pytest.raises(RuntimeError, match="above its maximum"):
+        FunPackLoadModifiers.execute(patcher, settings={
+            "m": {"amount": 999.0}, "full_control": {"enabled": False}})
+
+
+def test_full_control_is_never_silent_about_what_it_allowed(registry, patcher):
+    """A run made without the guards has to be identifiable as one afterwards,
+    because the consequences land later than the choice."""
+    from core import log
+    from modules.sampling.modifiers.nodes import FunPackLoadModifiers
+
+    registry.specs["m"] = _spec("m", lambda t, v, key: "on", settings={
+        "amount": {"type": "float", "default": 0.4, "min": -1.0, "max": 1.0, "label": "A"},
+    })
+    _full_control(registry)
+
+    log._reset()
+    FunPackLoadModifiers.execute(patcher, settings={
+        "m": {"amount": 999.0}, "full_control": {"enabled": True}})
+
+    said = [r for r in log.history() if r["level"] == log.WARNING]
+    assert said, "nothing was said about a value that was let through"
diff --git a/tools/devserver.py b/tools/devserver.py
index 9e486149..9cfe1c3d 100644
--- a/tools/devserver.py
+++ b/tools/devserver.py
@@ -71,6 +71,19 @@ from core import config, routes, serve as static  # noqa: E402
 P = config.UI_PREFIX
 
 
+def _pipeline_payload():
+    """The same answer the aiohttp route gives, so the preview shows what
+    ComfyUI would."""
+    from core import graph as graph_mod
+    slots = []
+    for _spec, make in routes.modules().providers("default_pipeline"):
+        slots = make()
+        break
+    prompt, incomplete = graph_mod.build(slots)
+    return {"slots": slots, "refused": [], "incomplete": incomplete,
+            "queueable": not incomplete}
+
+
 class Handler(BaseHTTPRequestHandler):
     def do_GET(self):
         parsed = urlparse(self.path)
@@ -78,7 +91,10 @@ class Handler(BaseHTTPRequestHandler):
 
         # The same manifest the aiohttp route serves, so what you see in the
         # browser here is what ComfyUI would send.
-        if path == P + "/api/log":
+        if path == P + "/api/pipeline":
+            body = json.dumps(_pipeline_payload()).encode()
+            served = static.Served(200, body, "application/json")
+        elif path == P + "/api/log":
             query = parse_qs(parsed.query)
             level = (query.get("level") or [None])[0]
             body = json.dumps({"levels": list(routes.log.LEVELS),