diff --git a/toolgate/api/workflows.py b/toolgate/api/workflows.py
index 986cae7..4244bd3 100644
--- a/toolgate/api/workflows.py
+++ b/toolgate/api/workflows.py
@@ -40,8 +40,11 @@ def _workflow_value(value, state: dict) -> Any:
         return _lookup_path(state["args"], value[6:])
     if isinstance(value, str) and value.startswith("$vars."):
         return _lookup_path(state["vars"], value[6:])
-    if isinstance(value, str) and value.startswith("$last"):
-        return _lookup_path(state.get("last"), value[6:] if value.startswith("$last.") else "")
+    # Exactly "$last" or "$last.<path>": text such as "$lastname" or "$last week" stays text.
+    if value == "$last":
+        return state.get("last")
+    if isinstance(value, str) and value.startswith("$last."):
+        return _lookup_path(state.get("last"), value[6:])
     if isinstance(value, list):
         return [_workflow_value(item, state) for item in value]
     if isinstance(value, dict):
diff --git a/toolgate/core/control_plane.py b/toolgate/core/control_plane.py
index f43aef8..8505cfa 100644
--- a/toolgate/core/control_plane.py
+++ b/toolgate/core/control_plane.py
@@ -398,7 +398,8 @@ def validate_inputs(schema: list[dict], args: dict) -> list[str]:
             errors.append(f"'{key}' is not an allowed argument")
     for field in schema:
         name, value = field.get("name"), args.get(field.get("name"))
-        if field.get("required") and name not in args:
+        # A required field sent as null is as missing as one left out.
+        if field.get("required") and value is None:
             errors.append(f"'{name}' is required")
             continue
         if value is None:
diff --git a/toolgate/tests/test_workflow_values.py b/toolgate/tests/test_workflow_values.py
index 9e365f2..9970f43 100644
--- a/toolgate/tests/test_workflow_values.py
+++ b/toolgate/tests/test_workflow_values.py
@@ -1,95 +1,31 @@
-"""Runtime value contracts used by the visual workflow editor."""
-import time
-import unittest
+"""Workflow references and required inputs (hunt toolgate-2 F1, F6)."""
 
-from fastapi import HTTPException
+import pytest
 
-from toolgate.api import definitions, workflows
+from toolgate.api.workflows import _workflow_value
+from toolgate.core import control_plane
 
+STATE = {"args": {"who": "Ada"}, "vars": {"n": 2}, "last": {"name": "previous", "items": [5]}}
 
-class WorkflowValueTests(unittest.TestCase):
-    def state(self, variables=None):
-        return {"automation_id": "value-test", "args": {}, "vars": variables or {},
-                "last": None, "results": [], "count": 0, "max_steps": 50,
-                "started_at": time.monotonic(), "runtime_ceiling": 5}
 
-    def run_steps(self, steps, state):
-        self.assertEqual([], definitions.workflow_definition_errors(steps))
-        return workflows._run_workflow_steps(steps, state, "test", False)
+@pytest.mark.parametrize("text", ["$lastname", "$last week", "$latest", "$last_seen"])
+def test_text_that_only_starts_like_last_stays_text(text):
+    assert _workflow_value(text, STATE) == text
 
-    def test_membership_selects_the_matching_branch(self):
-        for selected, expected in (("ready", "yes"), ("missing", "no")):
-            with self.subTest(selected=selected):
-                result = self.run_steps([{
-                    "type": "condition", "left": selected, "operator": "in",
-                    "right": ["ready", "waiting"],
-                    "then": [{"type": "return", "value": "yes"}],
-                    "else": [{"type": "return", "value": "no"}],
-                }], self.state())
-                self.assertEqual(expected, result["result"])
 
-    def test_loop_restores_absent_null_and_existing_variables(self):
-        for original in ({}, {"item": None}, {"item": "outer"}):
-            for body in ([], [{"type": "return", "value": "$vars.item"}]):
-                with self.subTest(original=original, body=body):
-                    state = self.state(dict(original))
-                    self.run_steps([{"type": "loop", "items": ["inner"],
-                                     "max_iterations": 2, "steps": body}], state)
-                    self.assertEqual(original, state["vars"])
+def test_last_and_its_paths_still_resolve():
+    assert _workflow_value("$last", STATE) == STATE["last"]
+    assert _workflow_value("$last.name", STATE) == "previous"
+    assert _workflow_value({"x": ["$last.items.0", "$args.who"]}, STATE) == {"x": [5, "Ada"]}
 
-    def test_loop_restores_variables_when_body_fails(self):
-        state = self.state({"item": "outer"})
-        with self.assertRaises(HTTPException):
-            self.run_steps([{"type": "loop", "items": [1], "max_iterations": 2,
-                             "steps": [{"type": "calculation", "operation": "divide",
-                                        "values": [1, 0]}]}], state)
-        self.assertEqual({"item": "outer"}, state["vars"])
 
-    def test_calculation_rejects_nonfinite_input_and_output(self):
-        for values in ([float("nan"), 1], [float("inf"), 1], [1e308, 1e308]):
-            with self.subTest(values=values):
-                state = self.state()
-                with self.assertRaises(HTTPException) as error:
-                    self.run_steps([{"type": "calculation", "operation": "multiply",
-                                     "values": values, "save_as": "answer"}], state)
-                self.assertEqual(422, error.exception.status_code)
-                self.assertNotIn("answer", state["vars"])
-                self.assertEqual([], state["results"])
+SCHEMA = [{"name": "title", "type": "string", "required": True, "min_length": 3},
+          {"name": "note", "type": "string"}]
 
-    def test_a_branch_that_returns_keeps_its_own_receipt(self):
-        branch = [{"type": "set", "name": "seen", "value": 1},
-                  {"type": "return", "value": "done"}]
-        expected_codes = {"condition": "CONDITION_MET", "switch": "SWITCH_SELECTED",
-                          "loop": "LOOP_RETURNED"}
-        for kind, step in (
-            ("condition", {"type": "condition", "left": "ready", "operator": "equals",
-                           "right": "ready", "then": branch}),
-            ("switch", {"type": "switch", "value": "ready", "cases": {"ready": branch}}),
-            ("loop", {"type": "loop", "items": ["a", "b"], "max_iterations": 2,
-                      "steps": branch}),
-        ):
-            with self.subTest(kind=kind):
-                state = self.state()
-                result = self.run_steps([step], state)
-                self.assertEqual("done", result["result"])
-                self.assertEqual(["set", kind], [r["type"] for r in state["results"]])
-                receipt = state["results"][-1]["result"]
-                self.assertEqual(expected_codes[kind], receipt["code"])
-                if kind == "loop":
-                    self.assertEqual(1, receipt["iterations"])
 
-    def test_nested_loop_restores_outer_item_for_following_steps(self):
-        state = self.state()
-        result = self.run_steps([
-            {"type": "loop", "items": ["outer"], "max_iterations": 2, "steps": [
-                {"type": "loop", "items": ["inner"], "max_iterations": 2, "steps": []},
-                {"type": "set", "name": "observed", "value": "$vars.item"},
-            ]},
-            {"type": "return", "value": "$vars.observed"},
-        ], state)
-        self.assertEqual("outer", result["result"])
-        self.assertNotIn("item", state["vars"])
+def test_a_required_input_sent_as_null_is_missing():
+    assert control_plane.validate_inputs(SCHEMA, {"title": None}) == ["'title' is required"]
 
 
-if __name__ == "__main__":
-    unittest.main()
+def test_an_optional_input_may_still_be_null():
+    assert control_plane.validate_inputs(SCHEMA, {"title": "Hello", "note": None}) == []