Repository: zefhub/zef (language: python)

Issue:
Token caching



Discussion / hints:

Fix patch:
diff --git a/.github/actions/build/action.yml b/.github/actions/build/action.yml
index 952b77c4..bf1b4f47 100644
--- a/.github/actions/build/action.yml
+++ b/.github/actions/build/action.yml
@@ -197,6 +197,6 @@ runs:
         set "PYTHONPATH=%GITHUB_WORKSPACE%/python;%PYTHONPATH%"
         copy python\pyzef\build\Release\pyzef.* python\zef
         python3 python/tests/quick_start.py
-        # Because of some strange weirdness, going to try unittest here instead of pytest
-        #python3 -mpytest python/tests
+        REM Because of some strange weirdness, going to try unittest here instead of pytest
+        REM python3 -mpytest python/tests
         python3 -munittest discover -s python/tests
\ No newline at end of file
diff --git a/.github/scripts/check-license.sh b/.github/scripts/check-license.sh
old mode 100644
new mode 100755
diff --git a/.github/workflows/pyzef-pypi-alpha.yml b/.github/workflows/pyzef-pypi-alpha.yml
index 444e6beb..8bd1b8fd 100644
--- a/.github/workflows/pyzef-pypi-alpha.yml
+++ b/.github/workflows/pyzef-pypi-alpha.yml
@@ -62,6 +62,7 @@ jobs:
   upload-to-pypi:
     name: Upload wheels to PyPI
     runs-on: ubuntu-20.04
+    environment: Release
     needs: [build-wheels]
     # We need this if in case build-wheels doesn't run, because for some reason
     # a skipped status is treated as a reason for the following jobs to also be
diff --git a/.github/workflows/pyzef-pypi.yml b/.github/workflows/pyzef-pypi.yml
index 7f05a09b..ed09c1bc 100644
--- a/.github/workflows/pyzef-pypi.yml
+++ b/.github/workflows/pyzef-pypi.yml
@@ -66,6 +66,7 @@ jobs:
   upload-to-pypi:
     name: Upload wheels to PyPI
     runs-on: ubuntu-20.04
+    environment: Release
     needs: [build-wheels]
     # We need this if in case build-wheels doesn't run, because for some reason
     # a skipped status is treated as a reason for the following jobs to also be
diff --git a/.gitignore b/.gitignore
index b3aa4f2a..0c618c4a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -127,3 +127,4 @@ tests/Testing/Temporary/
 /scripts/early.tokens
 python/zef/pyzef.pyd
 core/CMakeSettings.json
+/python/zef/auth.html
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2a23aa3b..03f2fea6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,8 @@ Breaking changes:
 
 - SimpleGQL: no longer takes the schema_file and graph as regular arguments, but
   as `--schema-file` and `--data-tag` instead.
+- Rename `ZEFDB_CONFIG_PATH` to `ZEFDB_SESSION_PATH`
+- `schema` zefop renamed to `blueprint`
   
 New features:
 
@@ -14,7 +16,16 @@ New features:
     Commandline args take priority.
     - `--init-hook` option to run a hook on every startup
     - `@dynamic(hook: "...")` option for field resolvers
+    
+- Tokens are cached in the session directory
 
+- Dictionary syntax for graph wishes: `{ET.Something: {RT.A: 3, RT.B: 4}}`
+
+- `set_field` operation for atomically updating "fields" 
+
+Fixes:
+
+- `login` auth flow was broken for the new default of statically built libzef.
 
 pyzef-0.15.6
 ============
diff --git a/compile_for_local_dev.sh b/compile_for_local_dev.sh
index be4564f5..ec1f25f9 100755
--- a/compile_for_local_dev.sh
+++ b/compile_for_local_dev.sh
@@ -82,3 +82,4 @@ python3 -m pip install -qr python/requirements.txt || exit 1
 ) || exit 1
 
 ln -fs $(realpath python/pyzef/build/pyzef.* --relative-to=python/zef) python/zef/
+ln -fs $(realpath core/auth.html --relative-to=python/zef) python/zef/
diff --git a/core/CMakeLists.txt b/core/CMakeLists.txt
index 992adc3e..cfb6301e 100644
--- a/core/CMakeLists.txt
+++ b/core/CMakeLists.txt
@@ -73,9 +73,9 @@ set(SOURCES
     src/uids.cpp
     src/from_json.cpp
     src/zef_config.cpp
+    src/globals.cpp
 
     src/butler/butler.cpp
-    src/butler/butler_zefhub.cpp
     src/butler/communication.cpp
     src/butler/threadsafe_map.cpp
     src/butler/auth.cpp
@@ -108,7 +108,6 @@ set(SOURCES
 #     include/zwitch.h
 
 #     include/butler/butler.h
-#     include/butler/butler_zefhub.h
 #     include/butler/msgqueue.h
 #     include/butler/messages.h
 #     include/butler/communication.h
@@ -174,6 +173,8 @@ if(ZEFDB_TEST_NO_MMAP_CHECKS)
 endif()
 
 # Make auth.html accessible for uninstalled builds
+# Note: this is really only for dynamic builds, but it will be a double-bounced
+# link for static builds
 if(NOT MSVC)
   add_custom_command(TARGET zef POST_BUILD
                     COMMAND ${CMAKE_COMMAND} -E create_symlink
@@ -189,7 +190,7 @@ if(MSVC)
 endif()
 
 if(LIBZEF_PYZEF_BUNDLED)
-  install(TARGETS zef LIBRARY DESTINATION .)
+  # install(TARGETS zef LIBRARY DESTINATION .)
   install(FILES auth.html DESTINATION .)
   install(FILES ${LICENSE_FILES} DESTINATION .)
 else()
diff --git a/core/include/butler/butler.h b/core/include/butler/butler.h
index b05e1c9d..63e45dba 100644
--- a/core/include/butler/butler.h
+++ b/core/include/butler/butler.h
@@ -34,8 +34,8 @@ namespace zefDB {
 
     namespace Butler {
 
-		const QuantityFloat zefhub_generic_timeout(10, EN.Unit.seconds);
-		const QuantityFloat butler_generic_timeout(15, EN.Unit.seconds);
+		const double zefhub_generic_timeout(10);
+		const double butler_generic_timeout(15);
 
         using namespace zefDB::Messages;
 
@@ -83,8 +83,6 @@ namespace zefDB {
         }
             
 
-        std::filesystem::path zefdb_config_path();
-
         // If this is set, the butler no longer needs to commmunicate with
         // upstream to ask permission or take primary etc...
         extern bool butler_is_master;
@@ -106,7 +104,7 @@ namespace zefDB {
                 bool is_online;
                 // Time last_activity;
                 std::atomic<double> last_activity;
-                QuantityFloat timeout;
+                double timeout;
                 // The nothing constructor is for receiving a looked-up task
                 // only.
                 Task() {}
@@ -122,7 +120,7 @@ namespace zefDB {
                 // Handle when making a Task from a RequestMesssage, which
                 // is forwarding a client request straight to zefhub, letting
                 // the client wait on the response rather than the butler.
-                Task(Time time, bool is_online, QuantityFloat timeout, std::promise<Response> & promise, bool acquire_future)
+                Task(Time time, bool is_online, double timeout, std::promise<Response> & promise, bool acquire_future)
                     : started_time(time),
                       is_online(is_online),
                       last_activity(time.seconds_since_1970),
@@ -155,8 +153,8 @@ namespace zefDB {
             // Note: we can't have add_task return a reference here, as it will
             // reference a location in the vector which may later change from
             // multi-threaded behaviour.
-            task_ptr add_task(bool is_online, QuantityFloat timeout);
-            task_ptr add_task(bool is_online, QuantityFloat timeout, std::promise<Response> && promise, bool acquire_future=false);
+            task_ptr add_task(bool is_online, double timeout);
+            task_ptr add_task(bool is_online, double timeout, std::promise<Response> && promise, bool acquire_future=false);
             task_promise_ptr find_task(std::string task_uid, bool forget=false);
             bool forget_task(std::string task_uid);
             void cancel_online_tasks();
@@ -173,18 +171,18 @@ namespace zefDB {
                 timeout_exception() : std::runtime_error("Timeout exception") {}
             };
 
-            Response wait_on_zefhub_message_any(json & j, const std::vector<std::string> & rest = {}, QuantityFloat timeout = zefhub_generic_timeout, bool throw_on_failure = false, bool chunked = false);
+            Response wait_on_zefhub_message_any(json & j, const std::vector<std::string> & rest = {}, double timeout = zefhub_generic_timeout, bool throw_on_failure = false, bool chunked = false);
 
             template<class T=GenericZefHubResponse>
-            T wait_on_zefhub_message(json & j, const std::vector<std::string> & rest = {}, QuantityFloat timeout = zefhub_generic_timeout, bool throw_on_failure = false, bool chunked = false);
+            T wait_on_zefhub_message(json & j, const std::vector<std::string> & rest = {}, double timeout = zefhub_generic_timeout, bool throw_on_failure = false, bool chunked = false);
 
             // This is to allow brace initialisation, while still preferring pass-by-reference.
-            Response wait_on_zefhub_message_any(json && j, const std::vector<std::string> & rest = {}, QuantityFloat timeout = zefhub_generic_timeout, bool throw_on_failure = false, bool chunked = false) {
+            Response wait_on_zefhub_message_any(json && j, const std::vector<std::string> & rest = {}, double timeout = zefhub_generic_timeout, bool throw_on_failure = false, bool chunked = false) {
                 return wait_on_zefhub_message_any(j, rest, timeout, throw_on_failure);
             }
 
             template<class T=GenericZefHubResponse>
-            T wait_on_zefhub_message(json && j, const std::vector<std::string> & rest = {}, QuantityFloat timeout = zefhub_generic_timeout, bool throw_on_failure = false) {
+            T wait_on_zefhub_message(json && j, const std::vector<std::string> & rest = {}, double timeout = zefhub_generic_timeout, bool throw_on_failure = false) {
                 return wait_on_zefhub_message(j, rest, timeout, throw_on_failure);
             }
 
@@ -220,6 +218,7 @@ namespace zefDB {
 
             int chunked_transfer_size_user = -1;
             int chunked_transfer_size = 10*1024;
+            int chunked_transfer_size_min = 1024;
             int chunked_transfer_queued = 10;
             double chunked_safety_factor = 5;
             bool chunked_transfer_auto_adjust = true;
@@ -314,7 +313,7 @@ namespace zefDB {
             // only time this could be useful is if there is nobody blocking on
             // a ZH query, e.g. a zearch request.
             template <class T>
-            T msg_push_timeout(Request && content, QuantityFloat timeout = butler_generic_timeout, bool ignore_closed=false);
+            T msg_push_timeout(Request && content, double timeout = butler_generic_timeout, bool ignore_closed=false);
 
             ////////////////////////////////////////////////////
             // * Graph manager functions
@@ -377,6 +376,7 @@ namespace zefDB {
 
 
             std::string upstream_name();
+            std::filesystem::path credentials_path();
 
             std::optional<Graph> local_process_graph;
             Graph get_local_process_graph();
@@ -405,12 +405,6 @@ namespace zefDB {
 
         LIBZEF_DLL_EXPORTED void ensure_or_get_range(void * ptr, size_t size);
 
-        // These are just for butler_zefhub.cpp
-        template<typename T1, typename T2>
-        void update_bidirection_name_map(thread_safe_bidirectional_map<T1,T2>& map_to_update, const T1 & indx, const T2 & name);
-        void update_zef_enum_bidirectional_map(thread_safe_zef_enum_bidirectional_map& map_to_update, const enum_indx& indx, const std::string & name);
-
-
         GenericResponse generic_from_json(json j);
 
         ////////////////////////////////////////
diff --git a/core/include/butler/butler.hpp b/core/include/butler/butler.hpp
index 851e1fba..ce340e22 100644
--- a/core/include/butler/butler.hpp
+++ b/core/include/butler/butler.hpp
@@ -20,7 +20,7 @@ namespace zefDB {
 
 
         template<class T>
-        T Butler::wait_on_zefhub_message(json & j, const std::vector<std::string> & rest, QuantityFloat timeout, bool throw_on_failure, bool chunked) {
+        T Butler::wait_on_zefhub_message(json & j, const std::vector<std::string> & rest, double timeout, bool throw_on_failure, bool chunked) {
             auto response = wait_on_zefhub_message_any(j, rest, timeout, throw_on_failure, chunked);
 
             return std::visit(overloaded {
@@ -54,11 +54,10 @@ namespace zefDB {
         }
 
         template <class T>
-        T Butler::msg_push_timeout(Request && content, QuantityFloat timeout, bool ignore_closed) {
+        T Butler::msg_push_timeout(Request && content, double timeout, bool ignore_closed) {
             std::string msg_type = std::visit([](auto & content) {return msgqueue_to_str(content);}, content);
             auto future = msg_push_internal(std::move(content), ignore_closed);
             std::future_status status;
-            assert(timeout.unit == EN.Unit.seconds);
             // This needs to be changed to timeout on the butler side instead of here.
             // if(timeout.value == 0)
             //     status = std::future_status::ready;
diff --git a/core/include/butler/butler_zefhub.h b/core/include/butler/butler_zefhub.h
deleted file mode 100644
index bbb0bc0f..00000000
--- a/core/include/butler/butler_zefhub.h
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Copyright 2022 Synchronous Technologies Pte Ltd
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#pragma once
-
-#include "butler/butler.h"
- 
-// These are low-level functions necessary for the functionality of zefhub.
-
-namespace zefDB {
-
-    namespace Butler {
-        LIBZEF_DLL_EXPORTED void add_entity_type(const token_value_t & indx, const std::string & name);
-        LIBZEF_DLL_EXPORTED void add_relation_type(const token_value_t & indx, const std::string & name);
-        LIBZEF_DLL_EXPORTED void add_enum_type(const enum_indx & indx, const std::string& name);
-        LIBZEF_DLL_EXPORTED void add_keyword(const token_value_t & indx, const std::string & name);
-    }
-}
diff --git a/core/include/butler/threadsafe_map.h b/core/include/butler/threadsafe_map.h
index b735ca18..8c17f3e3 100644
--- a/core/include/butler/threadsafe_map.h
+++ b/core/include/butler/threadsafe_map.h
@@ -270,6 +270,10 @@ namespace zefDB {
             //     if (found_val_it != map2.end()) return found_val_it->second;
             //     else return insert(val);			
             // }
+
+            size_t size() {
+                return map1.size();
+            }
         
             bool contains(const T1& val) {
                 return map1.find(val) != map1.end();
@@ -338,6 +342,7 @@ namespace zefDB {
 
         READ_FUNC(at);
         READ_FUNC(contains);
+        READ_FUNC(size);
         READ_FUNC(all_entries_as_list);
         READ_FUNC(all_keys);
         READ_FUNC(all_indices);
@@ -416,6 +421,7 @@ namespace zefDB {
             // enum_indx get_or_insert_and_get_enum_number(const string_pair& name_pair) ;
             bool contains(enum_indx indx_key) ;
             bool contains(const string_pair& name_pair) ;
+            size_t size();
         };
     }
 
@@ -460,6 +466,7 @@ namespace zefDB {
 
         READ_FUNC(at);
         READ_FUNC(contains);
+        READ_FUNC(size);
         READ_FUNC(all_entries_as_list);
         READ_FUNC(all_enum_types);
         READ_FUNC(all_enum_values);
diff --git a/core/include/mmap_windows.h b/core/include/mmap_windows.h
index 811e8073..4e0ac6d1 100644
--- a/core/include/mmap_windows.h
+++ b/core/include/mmap_windows.h
@@ -93,22 +93,6 @@
 namespace zefDB {
     namespace MMap {
 
-        constexpr size_t WIN_FILE_LOCK_BYTES = 1024;
-        inline std::string WindowsErrorMsg() {
-            LPVOID lpMsgBuf;
-            FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
-            NULL,
-            GetLastError(),
-            MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
-            (LPTSTR)&lpMsgBuf,
-            0, NULL);
-
-            std::string ret((LPCTSTR)lpMsgBuf);
-
-            LocalFree(lpMsgBuf);
-            return ret;
-        }
-
         inline void *WindowsMMap(int fd, size_t offset, size_t size, void * location) {
             if(fd == 0) {
                 // Anonymous map
@@ -135,22 +119,6 @@ namespace zefDB {
             }
         }
 
-        inline bool WindowsLockFile(int fd) {
-            HANDLE h = (HANDLE)_get_osfhandle(fd);
-            if (!LockFile(h, 0, 0, WIN_FILE_LOCK_BYTES, 0))
-                return false;
-            return true;
-        }
-        inline void WindowsUnlockFile(int fd, bool flush=false) {
-            HANDLE h = (HANDLE)_get_osfhandle(fd);
-            if(flush) {
-                if (!FlushFileBuffers(h))
-                    throw std::runtime_error("Problem flushing file to disk: " + WindowsErrorMsg());
-            }
-            if (!UnlockFile(h, 0, 0, WIN_FILE_LOCK_BYTES, 0))
-                    throw std::runtime_error("Problem unlocking file: " + WindowsErrorMsg());
-        }
-
         constexpr unsigned int PAGE_BITMAP_BITS = ZEF_UID_SHIFT / ZEF_PAGE_SIZE;
 
         // * Utils
diff --git a/core/include/scalars.h b/core/include/scalars.h
index ffc321cc..19fef94b 100644
--- a/core/include/scalars.h
+++ b/core/include/scalars.h
@@ -107,6 +107,10 @@ namespace zefDB {
 
     namespace internals {
         inline ZefEnumValue assert_that_is_unit_val(ZefEnumValue x) {
+            // This is a quick exit to avoid calling anything at initialisation
+            // of library, i.e. before the token store might be initialised.
+            if(x == EN.Unit._undefined)
+                return x;
             if (x.enum_type() != "Unit" || x.enum_value() == "")
                 throw std::runtime_error("ZefEnumValue passed is not a Unit value (e.g. 'kilograms')");
             return x;
@@ -269,25 +273,15 @@ namespace zefDB {
 
 
 
-    const QuantityFloat seconds{ 1, EN.Unit.seconds };
-    const QuantityFloat minutes{ 60, EN.Unit.seconds };
-    const QuantityFloat hours{ 60 * 60, EN.Unit.seconds };
-    const QuantityFloat days{ 60 * 60 * 24, EN.Unit.seconds };
-    const QuantityFloat weeks{ 60 * 60 * 24 * 7, EN.Unit.seconds };
-    const QuantityFloat months{ 60 * 60 * 24 * 30, EN.Unit.seconds };
-    const QuantityFloat years{ 60 * 60 * 24 * 365, EN.Unit.seconds };
+    extern const QuantityFloat seconds;
+    extern const QuantityFloat minutes;
+    extern const QuantityFloat hours;
+    extern const QuantityFloat days;
+    extern const QuantityFloat weeks;
+    extern const QuantityFloat months;
+    extern const QuantityFloat years;
 
 
     LIBZEF_DLL_EXPORTED std::ostream& operator<< (std::ostream& os, Time t);
 
-
-
-
-
-
-
-
-
-
-
 }
diff --git a/core/include/tokens.h b/core/include/tokens.h
index b374de59..e337efb0 100644
--- a/core/include/tokens.h
+++ b/core/include/tokens.h
@@ -398,6 +398,9 @@ namespace zefDB {
         void init_KWs();
         void init_ENs();
 
+        void load_cached_tokens();
+        void save_cached_tokens();
+
         std::optional<EntityType> ET_from_string(const std::string & s);
         EntityType ET_from_string_failhard(const std::string & s);
         std::optional<std::string> string_from_ET(const EntityType & et);
@@ -427,6 +430,12 @@ namespace zefDB {
 
         TokenStore(MMap::FileGraph & fg);
         TokenStore();
+
+        // This is internal-use only
+        void force_add_entity_type(const token_value_t & indx, const std::string & name);
+        void force_add_relation_type(const token_value_t & indx, const std::string & name);
+        void force_add_enum_type(const enum_indx & indx, const std::string& name);
+        void force_add_keyword(const token_value_t & indx, const std::string & name);
     };
 
 	// on the first run this is created and lives forever. All graphs created and destroyed register and unregister here
diff --git a/core/include/verification.h b/core/include/verification.h
index a765b329..9ab431d7 100644
--- a/core/include/verification.h
+++ b/core/include/verification.h
@@ -45,8 +45,13 @@ namespace zefDB {
         LIBZEF_DLL_EXPORTED void break_graph(Graph&g, blob_index index, int style);
 
         inline bool verify_graph(Graph&g) {
-            return (verify_graph_double_linking(g)
-                    && verify_chronological_instantiation_order(g));
+            // This is to handle mixed kinds of return false/true or the exception throwing
+            try {
+                return (verify_graph_double_linking(g)
+                        && verify_chronological_instantiation_order(g));
+            } catch(...) {
+                return false;
+            }
         }
 	}
 }
diff --git a/core/include/zefDB.h b/core/include/zefDB.h
index 903bc0d1..9904084c 100644
--- a/core/include/zefDB.h
+++ b/core/include/zefDB.h
@@ -39,7 +39,6 @@
 #include "tools.h"
 
 #include "butler/butler.h"
-#include "butler/butler_zefhub.h"
 
 #include "from_json.h"
 #include "zef_config.h"
\ No newline at end of file
diff --git a/core/include/zefDB_utils.h b/core/include/zefDB_utils.h
index a51d9b63..fafca02b 100644
--- a/core/include/zefDB_utils.h
+++ b/core/include/zefDB_utils.h
@@ -333,4 +333,116 @@ namespace zefDB {
         exiter.callbacks.emplace_front(name, std::move(func));
     }
 
+#ifdef ZEF_WIN32
+}
+#include <windows.h>
+#include <sysinfoapi.h>
+#include <io.h>
+#include <fcntl.h>
+
+namespace zefDB {
+        inline std::string WindowsErrorMsg() {
+            LPVOID lpMsgBuf;
+            FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
+            NULL,
+            GetLastError(),
+            MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
+            (LPTSTR)&lpMsgBuf,
+            0, NULL);
+
+            std::string ret((LPCTSTR)lpMsgBuf);
+
+            LocalFree(lpMsgBuf);
+            return ret;
+        }
+
+        constexpr size_t WIN_FILE_LOCK_BYTES = 1024;
+inline bool WindowsLockFile(int fd, bool exclusive=true, bool block=false) {
+            HANDLE h = (HANDLE)_get_osfhandle(fd);
+            OVERLAPPED overlapped;
+            overlapped.Offset = 0;
+            overlapped.OffsetHigh = 0;
+            overlapped.hEvent = 0;
+            int flags = 0;
+            if(exclusive)
+                flags |= LOCKFILE_EXCLUSIVE_LOCK;
+            if(!block)
+                flags |= LOCKFILE_FAIL_IMMEDIATELY;
+            if (!LockFileEx(h, flags, NULL, WIN_FILE_LOCK_BYTES, 0, &overlapped))
+                return false;
+            return true;
+        }
+        inline void WindowsUnlockFile(int fd, bool flush=false) {
+            HANDLE h = (HANDLE)_get_osfhandle(fd);
+            if(flush) {
+                if (!FlushFileBuffers(h))
+                    throw std::runtime_error("Problem flushing file to disk: " + WindowsErrorMsg());
+            }
+            if (!UnlockFile(h, 0, 0, WIN_FILE_LOCK_BYTES, 0))
+                    throw std::runtime_error("Problem unlocking file: " + WindowsErrorMsg());
+        }
+
+    struct FileLock {
+        std::filesystem::path path;
+        bool locked = false;
+        int fd = -1;
+        FileLock(std::filesystem::path path, bool exclusive=true)
+            : path(path) {
+            if(std::filesystem::is_directory(path))
+                path /= ".lock";
+            auto path_s = path.string();
+            fd = open(path_s.c_str(), O_RDWR | O_CREAT, 0600);
+            if(fd == -1)
+                throw std::runtime_error("Opening lockfile '" + path.string() + "' failed.");
+                
+            locked = WindowsLockFile(fd, exclusive, true);
+            if(!locked)
+                throw std::runtime_error("Locking file failed.");
+        }
+        ~FileLock() {
+            if(fd != -1) {
+                if(locked)
+                    WindowsUnlockFile(fd);
+                close(fd);
+            }
+        }
+    };
+#else
+}
+
+#include <sys/file.h>
+#include <unistd.h>
+
+namespace zefDB {
+    // Acquires a lock on a given named file in the given directory. This file
+    // should not itself contain any data. Uses flock or similar.
+    struct FileLock {
+        std::filesystem::path path;
+        int flock_ret = -1;
+        int fd = -1;
+        FileLock(std::filesystem::path path, bool exclusive=true)
+            : path(path) {
+            if(std::filesystem::is_directory(path))
+                path /= ".lock";
+            fd = open(path.c_str(), O_RDWR | O_CREAT, 0600);
+            if(fd == -1)
+                throw std::runtime_error("Opening lockfile '" + path.string() + "' failed.");
+                
+            if(exclusive)
+                flock_ret = flock(fd, LOCK_EX);
+            else
+                flock_ret = flock(fd, LOCK_SH);
+            if(flock_ret == -1)
+                throw std::runtime_error("Locking file failed. Errno: " + to_str(flock_ret));
+        }
+        ~FileLock() {
+            if(fd != -1) {
+                if(flock_ret != -1)
+                    flock(fd, LOCK_UN);
+                close(fd);
+            }
+        }
+    };
+#endif
+
 }
diff --git a/core/include/zef_config.h b/core/include/zef_config.h
index a3d0b23c..c142e9a5 100644
--- a/core/include/zef_config.h
+++ b/core/include/zef_config.h
@@ -23,6 +23,7 @@
 #include <vector>
 #include <array>
 #include <algorithm>
+#include "include_fs.h"
 
 namespace zefDB {
 
@@ -66,5 +67,7 @@ namespace zefDB {
     LIBZEF_DLL_EXPORTED config_var_t get_config_var(std::string key);
     LIBZEF_DLL_EXPORTED void set_config_var(std::string key, config_var_t val);
     LIBZEF_DLL_EXPORTED std::vector<std::pair<std::string,config_var_t>> list_config(std::string filter="");
+
+    LIBZEF_DLL_EXPORTED std::filesystem::path zefdb_config_path();
 }
 
diff --git a/core/src/blobs.cpp b/core/src/blobs.cpp
index 50460406..81b4f370 100644
--- a/core/src/blobs.cpp
+++ b/core/src/blobs.cpp
@@ -284,7 +284,7 @@ namespace zefDB {
 
     AllEdgeIndexes::Iterator AllEdgeIndexes::begin() const {
         blob_index* ptr_to_first_el_in_array = internals::edge_indexes(uzr_with_edges);
-        GraphData & gd = Graph(uzr_with_edges).my_graph_data();
+        GraphData & gd = *graph_data(uzr_with_edges);
         // Need to ensure here for the entire edge list, as normally
         // EZefRef{...} just does the minimum blob amount.
         Butler::ensure_or_get_range(uzr_with_edges.blob_ptr, size_of_blob(uzr_with_edges));
diff --git a/core/src/butler/auth.cpp b/core/src/butler/auth.cpp
index d5cc856d..e304722a 100644
--- a/core/src/butler/auth.cpp
+++ b/core/src/butler/auth.cpp
@@ -127,8 +127,35 @@ namespace zefDB {
     std::string last_token = "";
     time token_expire_time = clock::now();
 
+    bool using_private_key() {
+        char * env = getenv("ZEFDB_PRIVATE_KEY");
+        return (env != NULL && env[0] != '\0');
+    }
+
+    std::string get_private_token(std::string auth_id) {
+        assert(using_private_key());
+
+        char * env = getenv("ZEFDB_PRIVATE_KEY");
+        std::string private_key(env);
+
+        std::string token = jwt::create<jwt::traits::nlohmann_json>()
+            .set_issuer("local")
+            .set_type("JWT")
+            .set_audience("zefhub-io")
+            .set_payload_claim("sub", auth_id)
+            .set_issued_at(jwt::date::clock::now())
+            .set_expires_at(jwt::date::clock::now() + std::chrono::minutes(60))
+            .sign(jwt::algorithm::hs256{private_key});
+
+        return token;
+    }
 
     std::string get_firebase_refresh_token_email(std::string key_string) {
+        if(using_private_key()) {
+            if(zwitch.developer_output())
+                std::cerr << "Going to generate JWT from private key ourselves." << std::endl;
+            return get_private_token(key_string);
+        }
         int colon = key_string.find(':');
         std::string auth_username = key_string.substr(0,colon);
         std::string key = key_string.substr(colon+1);
@@ -185,6 +212,8 @@ namespace zefDB {
     }
 
     std::string get_firebase_token_refresh_token(std::string refresh_token) {
+        if(using_private_key())
+            return refresh_token;
 
         time start = clock::now();
 
diff --git a/core/src/butler/auth_server.cpp b/core/src/butler/auth_server.cpp
index 91ee6e26..e0ce270e 100644
--- a/core/src/butler/auth_server.cpp
+++ b/core/src/butler/auth_server.cpp
@@ -26,20 +26,24 @@ using json = nlohmann::json;
 #include "zefDB_utils.h"
 #include "zwitch.h"
 
+#ifdef LIBZEF_STATIC
+std::filesystem::path find_auth_html_dir() {
+    char* path = std::getenv("LIBZEF_AUTH_HTML_PATH");
+    if (zefDB::zwitch.developer_output())
+        std::cerr << "getenv returned (" << path << ") for LIBZEF_AUTH_HTML_PATH" << std::endl;
+    if (path == NULL)
+        throw std::runtime_error("Path for auth html was not set by controlling process!");
+    return std::filesystem::path(path);
+}
 
-#if defined(ZEF_WIN32)
-    std::filesystem::path find_libzef_path() {
-        char* path = std::getenv("LIBZEF_AUTH_HTML_PATH");
-        if (zefDB::zwitch.developer_output())
-            std::cerr << "getenv returned (" << path << ") for LIBZEF_AUTH_HTML_PATH" << std::endl;
-        if (path == NULL)
-            throw std::runtime_error("Path was not set by controlling process!");
-        // As the caller expects a library, we should give them something even if it doesn't exist.
-        return std::filesystem::path(path);
+#elif defined(ZEF_WIN32)
+#error "Windows code doesn't know how to find auth html dir if not built as static code."
+    std::filesystem::path find_auth_html_dir() {
+        throw std::runtime_error("Windows code doesn't know how to find the auth.html directory yet.")
     }
 #elif __APPLE__
 #include <dlfcn.h>
-    std::filesystem::path find_libzef_path() {
+    std::filesystem::path find_auth_html_dir() {
         void * handle = NULL;
         handle = dlopen("libzef.dylib", RTLD_NOW);
         if(handle == NULL)
@@ -54,7 +58,7 @@ using json = nlohmann::json;
 #elif __linux__
 #include <dlfcn.h>
 #include <link.h>
-    std::filesystem::path find_libzef_path() {
+    std::filesystem::path find_auth_html_dir() {
         void * handle = NULL;
         handle = dlopen("libzef.so", RTLD_NOW);
         if(handle == NULL)
@@ -65,8 +69,8 @@ using json = nlohmann::json;
 
         return std::filesystem::path(filename).parent_path();
     }
-#elif __unix__ // all unices not caught above
 #else
+#error "Unknown platform"
 #endif
 
 namespace zefDB {
@@ -118,7 +122,17 @@ namespace zefDB {
 
         if(zwitch.developer_output())
             std::cerr << "About to start io_server thread" << std::endl;
-        thread = std::make_shared<std::thread>([this]() {io_service_.run();});
+        thread = std::make_shared<std::thread>([this]() {
+            try {
+                asio::error_code ec;
+                io_service_.run(ec);
+                if(ec)
+                    std::cerr << "Auth server crashed with exception: " << ec << std::endl;
+            } catch(const std::exception & exc) {
+                std::cerr << "Auth server caught an unhandled exception: " << exc.what() << std::endl;
+            }
+            update(locker, should_stop, true);
+        });
         // TODO: Trigger opening of browser
         std::string url("http://localhost:" + std::to_string(cur_port) + "/auth");
         url += "?redirectUrl=callback";
@@ -281,14 +295,9 @@ namespace zefDB {
     std::string AuthServer::auth_reply() {
         if (zwitch.developer_output())
             std::cerr << "Going to reply with auth.html" << std::endl;
-        auto p = find_libzef_path();
+        auto p = find_auth_html_dir();
         if (zwitch.developer_output())
-            std::cerr << "Libzef path is: " << p.string() << std::endl;
-        // Note: this changed, it was that find_libzef_path would give the path to the library, but now it gives the path to the directory containing the library.
-        // p = p.parent_path();
-        // p /= "..";
-        // p /= "share";
-        // p /= "zefDB";
+            std::cerr << "auth.html dir is: " << p.string() << std::endl;
         p /= "auth.html";
         if (zwitch.developer_output())
             std::cerr << "auth.html path is: " << p.string() << std::endl;
diff --git a/core/src/butler/butler.cpp b/core/src/butler/butler.cpp
index 7ff77d8a..92a231dd 100644
--- a/core/src/butler/butler.cpp
+++ b/core/src/butler/butler.cpp
@@ -39,6 +39,10 @@ namespace zefDB {
 #include "butler_handlers_main.cpp"
 #include "butler_handlers_ws.cpp"
 
+#include "butler_connection.cpp"
+#include "butler_tasks.cpp"
+#include "butler_update_payloads.cpp"
+
         //////////////////////////////////////////////////////////
         // * Butler lifecycle management
 
@@ -193,7 +197,6 @@ namespace zefDB {
             }
         }
         void stop_butler() {
-            std::cerr << "Start of stop_butler" << std::endl;
             // This is to prevent multiple threads attacking this function at
             // the same time.
             static std::atomic<bool> _running = false;
@@ -206,7 +209,8 @@ namespace zefDB {
             }
 
             if (!butler) {
-                std::cerr << "Butler wasn't running." << std::endl;
+                if(zwitch.developer_output())
+                    std::cerr << "Butler wasn't running." << std::endl;
                 _running = false;
                 return;
             }
@@ -275,6 +279,9 @@ namespace zefDB {
             if(zwitch.developer_output())
                 std::cerr << "Clear waiting tasks" << std::endl;
             butler->waiting_tasks.clear();
+            if(zwitch.developer_output())
+                std::cerr << "Save tokens to cache" << std::endl;
+            global_token_store().save_cached_tokens();
 
             if(zwitch.developer_output())
                 std::cerr << "Joining main butler thread" << std::endl;
@@ -347,385 +354,9 @@ namespace zefDB {
         }
 
 
-        ////////////////////////////////
-        // * Task handling and WS msgs
-
-        Butler::task_ptr Butler::add_task(bool is_online, QuantityFloat timeout) {
-            return add_task(is_online, timeout, std::promise<Response>(), true);
-        };
-        Butler::task_ptr Butler::add_task(bool is_online, QuantityFloat timeout, std::promise<Response> && promise, bool acquire_future) {
-            std::lock_guard lock(waiting_tasks_mutex);
-#ifdef ZEF_DEBUG
-            if(waiting_tasks.size() > 100)
-                std::cerr << "Warning, there are a lot of tasks building up in the task_list! These should probably be removed at some point." << std::endl;
-#endif
-            task_ptr task = std::make_shared<Task>(now(), is_online, timeout, promise, acquire_future);
-            waiting_tasks.emplace_back(std::make_shared<TaskPromise>(task, std::move(promise)));
-            return task;
-        };
-        // Search and pop a task from the waiting tasks. Return true/false if found.
-        // Note the iteration style is due to the forward_list type.
-        Butler::task_promise_ptr Butler::find_task(std::string task_uid, bool forget) {
-            std::lock_guard lock(waiting_tasks_mutex);
-            for(auto it = waiting_tasks.begin() ; it != waiting_tasks.end() ; it++) {
-                if ((*it)->task->task_uid == task_uid) {
-                    if (forget) {
-                        Butler::task_promise_ptr ret = *it;
-                        waiting_tasks.erase(it);
-                        return ret;
-                    } else {
-                        return *it;
-                    }
-                }
-            }
-            return {};
-        }
-        bool Butler::forget_task(std::string task_uid) {
-            return (bool)find_task(task_uid, true);
-        }
-        void Butler::cancel_online_tasks() {
-            std::lock_guard lock(waiting_tasks_mutex);
-
-            waiting_tasks.erase(
-                std::remove_if(waiting_tasks.begin(),
-                               waiting_tasks.end(),
-                               [](task_promise_ptr & task_promise) {
-                                   if(task_promise->task->is_online) {
-                                       task_promise->promise.set_exception(std::make_exception_ptr(Communication::disconnected_exception()));
-                                       return true;
-                                   }
-                                   return false;
-                               }),
-                waiting_tasks.end()
-            );
-        }
-
-        Response wait_future(Butler::task_ptr task) {
-            if(task->timeout.value > 0) {
-                while(true) {
-                    assert(task->timeout.unit == EN.Unit.seconds);
-                    // TODO: We might have to make last_activity an atomic to make sure updates come through.
-                    auto wait_time = Time(task->last_activity.load()) + task->timeout - now();
-                    auto status = task->future.wait_for(std::chrono::duration<double>(wait_time.value));
-                    if(status == std::future_status::ready) {
-                        if(zwitch.developer_output())
-                            std::cerr << "Zefhub message took " << (now() - task->started_time) << std::endl;
-                        break;
-                    }
-
-                    if(now() > Time(task->last_activity.load()) + task->timeout) {
-                        std::cerr << "Throwing timeout exception because last_activity was " << task->last_activity << " and now is " << now() << std::endl;
-                        throw Butler::timeout_exception();
-                    }
-                }
-            }
-            return task->future.get();
-        }
-        Response Butler::wait_on_zefhub_message_any(json & j, const std::vector<std::string> & rest, QuantityFloat timeout, bool throw_on_failure, bool chunked) {
-            std::string msg_type = j["msg_type"].get<std::string>();
-            task_ptr task = add_task(true, timeout);
-            j["task_uid"] = task->task_uid;
-
-            if(chunked) {
-                send_chunked_ZH_message(task->task_uid, j, rest);
-            }
-            else
-                // Note: send returns before the data is actually sent.
-                send_ZH_message(j, rest);
-
-            task->last_activity = now().seconds_since_1970;
-
-            // Make sure we clean up the task from the task_list in all cases
-            RAII_CallAtEnd call_at_end([this,&task]() {
-                forget_task(task->task_uid);
-            });
-            
-            // The timeout is an inactivity timeout
-            auto result = wait_future(task);
-
-            if(throw_on_failure) {
-                GenericResponse generic = std::visit(overloaded {
-                        [](auto & v)->GenericResponse { return v.generic; },
-                        [](GenericResponse & v)->GenericResponse { return v; },
-                        [](std::monostate & v)->GenericResponse { throw std::runtime_error("Invalid response state."); }
-                    }, result);
-                if(!generic.success)
-                    throw std::runtime_error("ZefHub replied with failure: " + generic.reason);
-            }
-
-            return result;
-        }
-
-        void Butler::fill_out_ZH_message(json & j) { //, bool add_hints) {
-            j["protocol_type"] = "ZEFDB";
-            j["protocol_version"] = zefdb_protocol_version.load();
-            j["who"] = get_firebase_token_refresh_token(refresh_token);
-            // if(add_hints) {
-                if(estimated_transfer_speed_accum_time > 0) {
-                    double speed = estimated_transfer_speed_accum_bytes / estimated_transfer_speed_accum_time;
-                    double goal_time = chunk_timeout / 5;
-                    double est_chunk_size = goal_time * speed;
-                    j["hint_chunk_size"] = est_chunk_size;
-                }
-            // }
-        }
-
-        void Butler::send_ZH_message(json & j, const std::vector<std::string> & rest) {
-            if(butler_is_master)
-                throw std::runtime_error("Should not be trying to send messages to ZefHub when we are running in offline mode.");
-            if(!want_upstream_connection()) {
-                throw Communication::disconnected_exception();
-            }
-
-            // We do our own wait here, so that this is a proper timeout.
-            // network.wait_for_connected(constants::zefhub_reconnect_timeout);
-            // wait_for_auth(constants::zefhub_reconnect_timeout);
-            wait_for_auth();
-            if(!network.connected)
-                throw Communication::disconnected_exception();
-
-            // Note that "fill out" has to happen after the auth wait, as we
-            // won't know the protocol versions etc before then.
-            fill_out_ZH_message(j);
-
-            if(zwitch.debug_zefhub_json_output())
-                std::cerr << "About to send out: " << j << std::endl;
-
-            auto zh_msg = Communication::prepare_ZH_message(j, rest);
-
-            network.send(zh_msg);
-        }
-
-        void Butler::send_chunked_ZH_message(std::string main_task_uid, json & j, const std::vector<std::string> & rest) {
-            if(zefdb_protocol_version <= 4) {
-                send_ZH_message(j, rest);
-                return;
-            }
-
-            Butler::task_promise_ptr main_task = find_task(main_task_uid);
-
-            // Currently hardcoded but we need a way to extract this from config.
-            int chunk_size = this->chunked_transfer_size;
-            // This uid links all chunks in the transfer together.
-            std::string chunk_uid = generate_random_task_uid();
-
-            std::vector<int> rest_sizes(rest.size());
-            for(int i = 0; i < rest.size(); i++)
-                rest_sizes[i] = rest[i].size();
-
-            fill_out_ZH_message(j);
-            task_ptr task = add_task(true, chunk_timeout*seconds);
-            send_ZH_message(json{
-                    {"msg_type", "chunked"},
-                    {"task_uid", task->task_uid},
-                    {"chunk_type", "new"},
-                    {"chunk_uid", chunk_uid},
-                    {"rest_sizes", rest_sizes},
-                    {"msg", j}
-                });
-            std::deque<std::tuple<task_ptr,Time,int>> futures;
-            futures.emplace_back(task, now(), 1);
-
-            auto do_wait = [&](task_ptr & task) -> double {
-                try {
-                    auto start_time = now();
-                    auto result = wait_future(task);
-                    double this_waited_time = (now() - start_time).value;
-                    
-                    auto resp = std::visit(overloaded {
-                            [](const GenericZefHubResponse & response) { return response; },
-                            [](const auto & response) -> GenericZefHubResponse {
-                                throw std::runtime_error("Response from ZefHub is not of the right type in send_chunked_ZH_message.");
-                                }
-                            }, result);
-                    if(!resp.generic.success)
-                        throw std::runtime_error("Failure in chunked send: " + resp.generic.reason);
-                    main_task->task->last_activity = now().seconds_since_1970;
-                    // std::cerr << "Got back one response to a chunk after waiting: " << this_waited_time << std::endl;
-                    return this_waited_time;
-                } catch(const timeout_exception & exc) {
-                    // Reduce the chunk size. Unfortunately we can't restart the
-                    // transfer (without making a new chunk uid) as it could be
-                    // that the packets are still en route. This needs to be
-                    // improved.
-                    //
-                    // What could happen is a query back to ask "where are you
-                    // up to?" and then continuing from that point. This would
-                    // require removing the for loop over rest_index below, and
-                    // reacting to variables.
-                    if(chunked_transfer_auto_adjust)
-                        // Big adjustment down, smaller adjustments up.
-                        chunked_transfer_size /= 100;
-                    std::rethrow_exception(std::current_exception());
-                }
-            };
-
-            auto start_time = now();
-            int transfer_total_size = 0;
-
-            // Split rest into chunks
-            for(int rest_index = 0;rest_index < rest.size();rest_index++) {
-                // The chunk size can change over the message, so we do this chunking in a kind of weird way.
-                int rest_size = rest[rest_index].size();
-                transfer_total_size += rest_size;
-                int cur_start = 0;
-                while(cur_start < rest_size) {
-                    int this_chunk_size = std::min(chunk_size, rest_size - cur_start);
-                    std::string this_bytes = rest[rest_index].substr(cur_start, this_chunk_size);
-                    // std::cerr << "Sending a chunk, size: " << this_bytes.size() << ", start: " << cur_start << "/" << rest_size << ", rest_index: " << rest_index << std::endl;
-                    task = add_task(true, chunk_timeout*seconds);
-                    send_ZH_message(json{
-                            {"task_uid", task->task_uid},
-                            {"msg_type", "chunked"},
-                            {"chunk_type", "payload"},
-                            {"chunk_uid", chunk_uid},
-                            {"rest_index", rest_index},
-                            {"bytes_start", cur_start}
-                        }, {this_bytes});
-                    futures.emplace_back(task, now(), this_chunk_size);
-
-                    if(futures.size() >= this->chunked_transfer_queued) {
-                        // Wait on first future
-                        auto & it = futures.front();
-                        // Wait time = time spent in waiting here
-                        // Delta time = time since first starting the task
-                        // Size = is used to put the time in context - a super
-                        //        fast send of 10 bytes should not up later sends of
-                        //        1MB without caution.
-                        //
-                        // Note that sizes are a bit misleading too - zstd is
-                        // compressing the chunks so the calculated speed is not
-                        // exactly the same. This is open to huge problems so
-                        // need to revisit this I think.
-
-                        double first_wait_time = do_wait(std::get<0>(it));
-                        double first_delta_time = (now() - std::get<1>(it)).value;
-                        int first_size = std::get<2>(it);
-                        futures.pop_front();
-
-                        if(chunked_transfer_auto_adjust) {
-                            // Now wait on more until the queue is half-empty, this
-                            // gives us a way to check the response times. If they
-                            // are ping-dominated then the spacing between these
-                            // times will be heavily bimodal (i.e. one slow and the
-                            // rest nearly instant afterwards).
-                            double max_wait_time = first_wait_time;
-                            double second_wait_time = 0;
-                            double min_delta_time = first_delta_time;
-                            int min_size = first_size;
-                            while(futures.size() > this->chunked_transfer_queued / 2) {
-                                auto & it = futures.front();
-                                double this_wait_time = do_wait(std::get<0>(it));
-                                if(this_wait_time > max_wait_time) {
-                                    second_wait_time = max_wait_time;
-                                    max_wait_time = this_wait_time;
-                                } else if (this_wait_time > second_wait_time) {
-                                    second_wait_time = this_wait_time;
-                                }
-
-                                double this_delta_time = (now() - std::get<1>(it)).value;
-                                min_delta_time = std::min(this_delta_time, min_delta_time);
-
-                                int this_size = std::get<2>(it);
-                                min_size = std::min(min_size, this_size);
-                                futures.pop_front();
-
-                                if(this_delta_time < chunk_timeout/ chunked_safety_factor) {
-                                    // Because we are so much smaller, we can grow a lot in here.
-                                    int new_chunk_size = (int)(min_size * chunked_safety_factor/2);
-                                    if(new_chunk_size > chunk_size) {
-                                        chunk_size = new_chunk_size;
-                                        if(zwitch.developer_output())
-                                            std::cerr << "Increased chunk size (" << chunk_size << ") as the time to return (" << min_delta_time << ") is very small from a set of packets with min size (" << min_size << ")" << std::endl;
-                                    }
-                                }
-                            }
-
-                            if(min_delta_time < chunk_timeout / chunked_safety_factor && second_wait_time < max_wait_time/chunked_safety_factor) {
-                                // Although this is not much, it grows fast.
-                                int new_chunk_size = (int)(min_size * 1.5);
-                                if(new_chunk_size > chunk_size) {
-                                    chunk_size = new_chunk_size;
-                                    if(zwitch.developer_output())
-                                        std::cerr << "Increasing chunk size as the wait times are very noisy" << std::endl;
-                                }
-                            } else {
-                                // std::cerr << "Leaving chunk size as is because the ratio is " << second_wait_time / max_wait_time << std::endl;
-                            }
-                        }
-                    }
-
-                    cur_start += this_chunk_size;
-                }
-            }
-
-            // Wait on remaining futures
-            for(auto & it : futures) {
-                do_wait(std::get<0>(it));
-            }
-
-            // Note: the do_wait sets last_activity on the main task, so there's
-            // nothing special to do at the end.
-
-            // Update estimated transfer speed
-            auto transfer_duration = now() - start_time;
-            estimated_transfer_speed_accum_bytes += transfer_total_size;
-            estimated_transfer_speed_accum_time += transfer_duration.value;
-            double ratio = estimated_transfer_speed_accum_time / limit_estimated_transfer_speed_accum_time;
-            if(ratio > 1) {
-                estimated_transfer_speed_accum_time /= ratio;
-                estimated_transfer_speed_accum_bytes /= ratio;
-            }
-
-            if(chunked_transfer_auto_adjust) {
-                double speed = estimated_transfer_speed_accum_bytes / estimated_transfer_speed_accum_time;
-                double goal_time = chunk_timeout / 5;
-                int est_chunk_size = (int)(goal_time * speed);
-                // if(est_chunk_size > chunk_size)
-                if(zwitch.developer_output()) {
-                    std::cerr << "est_chunk_size: " << est_chunk_size << std::endl;
-                    std::cerr << "est_speed: " << speed << std::endl;
-                    std::cerr << "previous chunk_size: " << chunked_transfer_size << std::endl;
-                }
-                if(est_chunk_size < chunked_transfer_size)
-                    chunked_transfer_size = est_chunk_size;
-                else
-                    chunked_transfer_size = (int)((chunked_transfer_size + est_chunk_size) / chunked_safety_factor / chunked_transfer_queued);
-                if(zwitch.developer_output())
-                    std::cerr << "Adjusting chunk size to " << chunked_transfer_size << std::endl;
-            }
-        }
-
-
-        GenericResponse generic_from_json(json j) {
-            GenericResponse generic;
-            if(j.contains("success")) {
-                if(j["success"].is_boolean())
-                    generic.success = j["success"].get<bool>();
-                else {
-                    std::string temp = j["success"].get<std::string>();
-                    if(temp == "0")
-                        generic.success = false;
-                    else if(temp == "1")
-                        generic.success = true;
-                    else
-                        throw std::runtime_error("Unknown value for success in GenericZefHubResponse");
-                }
-                if(j.contains("reason"))
-                    generic.reason = json_string_default(j, "reason");
-                else if(j.contains("response"))
-                    generic.reason = json_string_default(j, "response");
-            } else {
-                generic.success = true;
-                generic.reason = "";
-            }
-            return generic;
-        }
-
         ////////////////////////////////////////////////////////////////
         // * Butler main msgqueue handling
 
-
         void Butler::msgqueue_listener(void) {
             try {
                 Butler::msg_ptr msg;
@@ -843,409 +474,13 @@ namespace zefDB {
             if(std::filesystem::exists(dir))
                 throw std::runtime_error("Can't create local graph at '" + dir.string() + "' as it is already a file.");
 
-            std::filesystem::create_directory(dir);
+            std::filesystem::create_directories(dir.parent_path());
 
             auto data = spawn_graph_manager(make_random_uid());
             data->queue.push(std::make_shared<RequestWrapper>(std::move(msg->promise), LocalGraph{dir, true}));
         }
 
 
-        ////////////////////////////////////////////////////////
-        // * Updates to upstream
-
-        bool is_up_to_date(const UpdateHeads & update_heads) {
-            // Note we use < instead of != here to ease checks on zefhub where clients can be ahead.
-            if(update_heads.blobs.from < update_heads.blobs.to)
-                return false;
-
-            for(auto & cache : update_heads.caches) {
-                if(cache.from < cache.to)
-                    return false;
-            }
-
-            return true;
-        }
-
-        UpdatePayload create_update_payload(const GraphData & gd, const UpdateHeads & update_heads) {
-            if(update_heads.blobs.from > update_heads.blobs.to)
-                throw std::runtime_error("Somehow upstream is ahead of us and we're primary!");
-
-            blob_index last_tx = gd.latest_complete_tx;
-
-            // This here is a dodgy internal hack to grab the whole graph so we can send the blobs/uids out:
-            char * blobs_ptr = (char*)(&gd) + update_heads.blobs.from * constants::blob_indx_step_in_bytes;
-            size_t len = (update_heads.blobs.to - update_heads.blobs.from) * constants::blob_indx_step_in_bytes;
-            ensure_or_get_range(blobs_ptr, len);
-            // TODO: Turn these back into string_views once everything is safe
-            std::string blobs(blobs_ptr, len);
-
-            UpdatePayload p;
-            p.j = json{
-                {"blob_index_lo", update_heads.blobs.from},
-                {"blob_index_hi", update_heads.blobs.to},
-                {"graph_uid", str(internals::get_graph_uid(gd))},
-                {"index_of_latest_complete_tx_node", last_tx},
-                {"hash_full_graph", gd.hash(constants::ROOT_NODE_blob_index, update_heads.blobs.to)},
-                {"data_layout_version", internals::get_data_layout_version_info(gd)}
-            };
-            p.rest = {blobs};
-            std::vector<json> caches;
-
-            for(auto & cache : update_heads.caches) {
-#define GEN_CACHE(x, y) else if(cache.name == x) { \
-                    if(cache.from != cache.to) { \
-                        caches.push_back({ \
-                                {"name", x}, \
-                                {"index_lo", cache.from}, \
-                                {"index_hi", cache.to}, \
-                                {"revision", cache.revision}, \
-                            }); \
-                        auto ptr = gd.y->get(); \
-                        p.rest.push_back(ptr->create_diff(cache.from, cache.to)); \
-                    } \
-                }
-
-                if(false) {}
-                GEN_CACHE("_ETs_used", ETs_used)
-                GEN_CACHE("_RTs_used", RTs_used)
-                GEN_CACHE("_ENs_used", ENs_used)
-                GEN_CACHE("_uid_lookup", uid_lookup)
-                GEN_CACHE("_euid_lookup", euid_lookup)
-                GEN_CACHE("_tag_lookup", tag_lookup)
-                else {
-                    throw std::runtime_error("Unknown cache");
-                }
-#undef GEN_CACHE
-
-            }
-
-            p.j["caches"] = caches;
-
-            return p;
-        }
-
-        json create_heads_json_from_sync_head(const GraphData & gd, const UpdateHeads & update_heads) {
-            json j{
-                {"blobs_head", update_heads.blobs.from},
-            };
-            json caches;
-
-            for(auto & cache : update_heads.caches) {
-#define GEN_CACHE(x, y) else if(cache.name == x) { \
-                    caches[x] = {                       \
-                        {"name", x},                    \
-                        {"head", cache.from},   \
-                        {"revision", cache.revision},   \
-                    };                                  \
-                }                                       \
-
-                if(false) {}
-                GEN_CACHE("_ETs_used", ETs_used)
-                GEN_CACHE("_RTs_used", RTs_used)
-                GEN_CACHE("_ENs_used", ENs_used)
-                GEN_CACHE("_uid_lookup", uid_lookup)
-                GEN_CACHE("_euid_lookup", euid_lookup)
-                GEN_CACHE("_tag_lookup", tag_lookup)
-                else {
-                    throw std::runtime_error("Unknown cache");
-                }
-#undef GEN_CACHE
-
-            }
-
-            j["cache_heads"] = caches;
-
-            return j;
-        }
-
-        UpdateHeads client_create_update_heads(const GraphData & gd) {
-            if(gd.open_tx_thread != std::this_thread::get_id())
-                throw std::runtime_error("Need write lock to carefully read update heads.");
-
-            blob_index update_from = gd.sync_head;
-            if(update_from == 0)
-                update_from = constants::ROOT_NODE_blob_index;
-            UpdateHeads update_heads{ {update_from, gd.read_head} };
-
-#define GEN_CACHE(x, y) { \
-                auto ptr = gd.y->get(); \
-                update_heads.caches.push_back({x, ptr->upstream_size(), ptr->size(), ptr->revision()}); \
-            }
-            GEN_CACHE("_ETs_used", ETs_used)
-            GEN_CACHE("_RTs_used", RTs_used)
-            GEN_CACHE("_ENs_used", ENs_used)
-            GEN_CACHE("_uid_lookup", uid_lookup)
-            GEN_CACHE("_euid_lookup", euid_lookup)
-            GEN_CACHE("_tag_lookup", tag_lookup)
-#undef GEN_CACHE
-            return update_heads;
-        }
-
-        void parse_filegraph_update_heads(MMap::FileGraph & fg, json & j) {
-            j["blobs_head"] = fg.get_latest_blob_index();
-
-            json cache_heads;
-            auto prefix = fg.get_prefix();
-            {
-                decltype(GraphData::ETs_used)::element_type mapping(fg, prefix->tokens_ET);
-                cache_heads["_ETs_used"] = json{
-                    {"head", mapping.get()->size()},
-                    {"revision", mapping.get()->revision()},
-                };
-            }
-            {
-                decltype(GraphData::RTs_used)::element_type mapping(fg, prefix->tokens_RT);
-                cache_heads["_RTs_used"] = json{
-                    {"head", mapping.get()->size()},
-                    {"revision", mapping.get()->revision()},
-                };
-            }
-            {
-                decltype(GraphData::ENs_used)::element_type mapping(fg, prefix->tokens_EN);
-                cache_heads["_ENs_used"] = json{
-                    {"head", mapping.get()->size()},
-                    {"revision", mapping.get()->revision()},
-                };
-            }
-            {
-                decltype(GraphData::uid_lookup)::element_type mapping(fg, prefix->uid_lookup);
-                cache_heads["_uid_lookup"] = json{
-                    {"head", mapping.get()->size()},
-                    {"revision", mapping.get()->revision()},
-                };
-            }
-            {
-                decltype(GraphData::euid_lookup)::element_type mapping(fg, prefix->euid_lookup);
-                cache_heads["_euid_lookup"] = json{
-                    {"head", mapping.get()->size()},
-                    {"revision", mapping.get()->revision()},
-                };
-            }
-            {
-                decltype(GraphData::tag_lookup)::element_type mapping(fg, prefix->tag_lookup);
-                cache_heads["_tag_lookup"] = json{
-                    {"head", mapping.get()->size()},
-                    {"revision", mapping.get()->revision()},
-                };
-            }
-            j["cache_heads"] = cache_heads;
-        }
-
-        void apply_sync_heads(GraphData & gd, const UpdateHeads & update_heads) {
-            update(gd.heads_locker, [&]() {
-                gd.sync_head = update_heads.blobs.to;
-                gd.currently_subscribed = true;
-
-                for(auto & cache : update_heads.caches) {
-#define GEN_CACHE(x, y) else if(cache.name == x) { \
-                        auto ptr = gd.y->get_writer(); \
-                        ptr->upstream_size() = cache.to; \
-                        ptr->revision() = cache.revision; \
-                    }
-
-                    if(false) {}
-                    GEN_CACHE("_ETs_used", ETs_used)
-                    GEN_CACHE("_RTs_used", RTs_used)
-                    GEN_CACHE("_ENs_used", ENs_used)
-                    GEN_CACHE("_uid_lookup", uid_lookup)
-                    GEN_CACHE("_euid_lookup", euid_lookup)
-                    GEN_CACHE("_tag_lookup", tag_lookup)
-                    else
-                        throw std::runtime_error("Don't understand cache: " + cache.name);
-#undef GEN_CACHE
-                }
-            });
-        }
-
-        UpdateHeads parse_message_update_heads(const json & j) {
-            UpdateHeads update_heads{
-                {0, j["blobs_head"].get<blob_index>()}
-            };
-
-            for(auto & it : j["cache_heads"].items()) {
-                std::string name = it.key();
-                update_heads.caches.push_back({
-                        name,
-                        0,
-                        it.value()["head"].get<size_t>(),
-                        it.value()["revision"].get<size_t>()
-                    });
-            }
-
-            return update_heads;
-        }
-
-        UpdateHeads parse_payload_update_heads(const UpdatePayload & payload) {
-            UpdateHeads heads{
-                {payload.j["blob_index_lo"].get<blob_index>(), payload.j["blob_index_hi"].get<blob_index>()}
-            };
-
-            for(auto & cache : payload.j["caches"]) {
-                std::string name = cache["name"].get<std::string>();
-                size_t from = cache["index_lo"].get<size_t>();
-                size_t to = cache["index_hi"].get<size_t>();
-
-                heads.caches.push_back({name, from, to});
-            }
-
-            return heads;
-        }
-
-        bool heads_apply(const UpdateHeads & heads, const GraphData & gd) {
-            if(gd.write_head != heads.blobs.from) {
-                std::cerr << "Blobs update (" + to_str(heads.blobs.from) + ") doesn't fit onto write head (" + to_str(gd.write_head.load()) + ")" << std::endl;
-                return false;
-            }
-
-            // Check cache heads align
-            for(auto & cache : heads.caches) {
-#define GEN_CACHE(x, y) else if(cache.name == x) { \
-                    auto ptr = gd.y->get(); \
-                    if(ptr->revision() != cache.revision || ptr->size() != cache.from) {      \
-                        std::cerr << "Update for cache '" + cache.name + "' doesn't fit. Current: (" << ptr->revision() << ", " << ptr->size() << "), update: (" << cache.revision << ", " << cache.from << ")" << std::endl; \
-                        return false; \
-                    } \
-                }
-                if(false) {}
-                GEN_CACHE("_ETs_used", ETs_used)
-                GEN_CACHE("_RTs_used", RTs_used)
-                GEN_CACHE("_ENs_used", ENs_used)
-                GEN_CACHE("_uid_lookup", uid_lookup)
-                GEN_CACHE("_euid_lookup", euid_lookup)
-                GEN_CACHE("_tag_lookup", tag_lookup)
-                else
-                    throw std::runtime_error("Unknown cache");
-            }
-
-            return true;
-        }
-
-
-        void Butler::send_update(Butler::GraphTrackingData & me) {
-            if(me.gd->error_state != GraphData::ErrorState::OK)
-                throw std::runtime_error("Can't send_update with a graph in an invalid state");
-            if(!me.gd->is_primary_instance)
-                throw std::runtime_error("We don't have primary role, how did we send an update??");
-
-            // If we aren't connected, then we'll worry about sending out updates later.
-            if(!network.connected) {
-                // Make sure we know we aren't subscribed for later.
-                me.gd->currently_subscribed = false;
-                return;
-            }
-
-            // If we have yet to resubscribe to a graph, then wait for that
-            // before trying to send out an update. The wait can happen in the caller (most likely the sync thread)
-            if(me.gd->sync_head > 0 && !me.gd->currently_subscribed)
-                return;
-
-            UpdateHeads update_heads;
-            {
-                // Let's stop transactions from happening while we do this.
-                LockGraphData lock{me.gd};
-
-                update_heads = client_create_update_heads(*me.gd);
-            }
-
-            if(is_up_to_date(update_heads))
-                return;
-
-            UpdatePayload payload = create_update_payload(*me.gd, update_heads);
-
-            if(me.gd->sync_head == 0)
-                payload.j["msg_type"] = "full_graph";
-            else
-                payload.j["msg_type"] = "graph_update";
-
-            payload.j["msg_version"] = 1;
-
-            // Handling errors:
-            // ================
-            // Regardless of what happened, we need to queue a new shot at this.
-            // A "queue" implies a timeout though and that's ugly. Instead, we have a choice of options:
-            // a) try again immediately, this could lead to spam, maybe have a maximum try with a sleep in between (this blocks anything else happening with the graph though).
-            // b) check if the websocket went down. If so, wait until the trigger message for when that comes back up.
-            // c) wait for the next transaction. We could lose data that way though.
-            // A mix between a) and b) seems good.
-
-            const int max_attempts = 3;
-            int attempts_left = max_attempts;
-            for(; attempts_left > 0; attempts_left--) {
-                // TODO: Check if websocket is down, and quit in that case.
-                if(!network.connected) {
-                    std::cerr << "Gave up updating as websocket is disconnected" << std::endl;
-                    return;
-                }
-
-                // A little pause for each retry except the first.
-                if(attempts_left != max_attempts)
-                    std::this_thread::sleep_for(std::chrono::seconds(1));
-                    
-
-                GenericZefHubResponse response;
-                try {
-                    if(zwitch.developer_output())
-                        std::cerr << "Trying to send update for graph " << me.uid << " of range " << update_heads.blobs.from << " to " << update_heads.blobs.to << std::endl;
-                    response = wait_on_zefhub_message(payload.j, payload.rest, zefhub_generic_timeout, true, true);
-
-                    if(!response.generic.success) {
-                        // There's generally only one reason for a failure. But
-                        // in any case return and leave this to try again in the
-                        // sync loop. It's better to just keep retrying rather
-                        // than crashing.
-
-                        if(response.j.contains("upstream_head")) {
-                            UpdateHeads parsed_heads = parse_message_update_heads(response.j);
-                            apply_sync_heads(*me.gd, parsed_heads);
-                        } else {
-                            // Unknown error - wait a bit
-                            std::cerr << "Unknown error received from ZH from our graph update ('" << response.generic.reason << "'). Setting graph to invalid state." << std::endl;
-                            me.gd->error_state = GraphData::ErrorState::UNSPECIFIED_ERROR;
-                            // std::this_thread::sleep_for(std::chrono::seconds(1));
-                        }
-
-                        return;
-                    }
-                } catch(const timeout_exception & e) {
-                    std::cerr << "Got timeout during send_update. Going to retry" << std::endl;
-                    continue;
-                } catch(const Communication::disconnected_exception & e) {
-                    std::cerr << "Disconnected during send_update. Going to retry" << std::endl;
-                    continue;
-                } catch(const std::exception & e) {
-                    std::cerr << "An exception occurred during a send_update ('" << e.what() << "'). Setting graph to invalid state." << std::endl;
-                    me.gd->error_state = GraphData::ErrorState::UNSPECIFIED_ERROR;
-                    wake(me.gd->heads_locker);
-                    return;
-                }
-
-                // if(response.j["blob_index_hi"].get<blob_index>() != me.gd->read_head)
-                //     throw std::runtime_error("Somehow zefhub replied with a different blob update than we sent!");
-
-                // Getting here should mean we succeeded.
-                me.gd->currently_subscribed = true;
-                apply_sync_heads(*me.gd, update_heads);
-                return;
-            }
-
-            // If we get here, we ran out of attempts
-            if(attempts_left != 0)
-                throw std::runtime_error("Huh?");
-
-            std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
-            std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
-            std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
-            std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
-            // We don't want to set this after all as it is better to enter a
-            // never-ending cycle of resend attempts, than potentially lose
-            // data.
-
-            // me.gd->error_state = GraphData::ErrorState::UNSPECIFIED_ERROR;
-            // throw std::runtime_error("Gave up after " + to_str(max_attempts) + " attempts of sending out graph update!");
-            // It's always better to just return and let this try again rather than crashing.
-            return;
-        }
-
 
         ////////////////////////////////////////////////////
         // * Graph manager management
@@ -1572,453 +807,6 @@ namespace zefDB {
         }
 
 
-        ////////////////////////////////////////////////
-        // * Connection management
-
-        bool Butler::want_upstream_connection() {
-            if(network.uri == "")
-                return false;
-
-            // If we are running, then we want to continue.
-            if(network.is_running())
-                return true;
-
-            std::string auto_connect = std::get<std::string>(get_config_var("login.autoConnect"));
-
-            if(auto_connect == "auto" && have_auth_credentials())
-                return true;
-            if(auto_connect == "always")
-                return true;
-
-            return false;
-        }
-
-        bool Butler::wait_for_auth(std::chrono::duration<double> timeout) {
-            if(butler_is_master)
-                throw std::runtime_error("Butler started as master - should never have got to this point of requesting auth.");
-            if(network.uri == "")
-                throw std::runtime_error("Not connecting to ZefHub without a URL");
-            if(!want_upstream_connection())
-                throw std::runtime_error("Not connecting to ZefHub until we have credentials. Either login using `login | run` to store your credentials or `login_as_guest | run` for a temporary guest login.");
-            if(!network.managing_thread) {
-                // throw std::runtime_error("Network is not trying to connect, can't wait for auth.");
-                debug_time_print("before start connection");
-                if(should_stop) {
-                    return false;
-                }
-                start_connection();
-            }
-            auto done_auth = [this]() { return should_stop || connection_authed || fatal_connection_error; };
-            if(timeout < std::chrono::duration<double>(0)) {
-                // wait_pred(auth_locker, done_auth);
-                // We have a special spam message in here to let users know what's going on. 
-                wait_pred(auth_locker, done_auth, std::chrono::seconds(3));
-                if(!done_auth()) {
-                    std::cerr << "Warning: waiting for connection with ZefHub is taking a long time.\n\nIf you would like to see more information enable debug messages through `zwitch.zefhub_communication_output(True)` or setting the environment variable `ZEFDB_VERBOSE=true`.\n\nIf you would like to run Zef in offline mode, then start a new python session with the environment variable `ZEFDB_OFFLINE_MODE=TRUE`." << std::endl;
-                    wait_pred(auth_locker, done_auth);
-                }
-            }
-            else {
-                if(!wait_pred(auth_locker, done_auth, timeout))
-                    return false;
-            }
-
-            if(no_credentials_warning)
-                throw std::runtime_error("No credentials present to allow auth to take place.");
-            if(fatal_connection_error)
-                throw std::runtime_error("Fatal connection error while trying to auth with ZefHub.\n\nIf you would like to run in offline mode, please restart your python session with the environment variable `ZEFDB_OFFLINE_MODE=TRUE`.");
-            debug_time_print("finish wait_for_auth");
-            return true;
-        }
-
-        void Butler::determine_refresh_token() {
-            std::optional<std::string> key_string = load_forced_zefhub_key();
-            if(key_string) {
-                if(*key_string == constants::zefhub_guest_key) {
-                    std::cerr << "Connecting as guest user" << std::endl;
-                    refresh_token = "";
-                    have_logged_in_as_guest = true;
-                } else 
-                    refresh_token = get_firebase_refresh_token_email(*key_string);
-                return;
-            } else if(!have_auth_credentials()) {
-                no_credentials_warning = true;
-                throw std::runtime_error("Have no existing credentials to determine auth token. You must login first.");
-            } else {
-                ensure_auth_credentials();
-                if(have_logged_in_as_guest) {
-                    refresh_token = "";
-                } else {
-                    auto credentials_file = zefdb_config_path() / "credentials";
-                    std::ifstream file(credentials_file);
-                    json j = json::parse(file);
-                    refresh_token = j["refresh_token"].get<std::string>();
-                }
-            }
-        }
-
-        std::string Butler::who_am_i() {
-            std::string name;
-
-            std::optional<std::string> key_string = load_forced_zefhub_key();
-            if(key_string) {
-                if(*key_string == constants::zefhub_guest_key) {
-                    name = "GUEST via forced key";
-                } else {
-                    // This is  duplicated with get_firebase_refresh_token_email which is annoying
-                    int colon = key_string->find(':');
-                    name = key_string->substr(0,colon) + " via forced key";
-                }
-            } else if(!have_auth_credentials()) {
-                name = "";
-            } else {
-                if(have_logged_in_as_guest) {
-                    name = "GUEST via login prompt";
-                } else {
-                    auto credentials_file = zefdb_config_path() / "credentials";
-                    if(!std::filesystem::exists(credentials_file)) {
-                        name = "";
-                    } else {
-                        std::ifstream file(credentials_file);
-                        json j = json::parse(file);
-                        refresh_token = j["refresh_token"].get<std::string>();
-                        std::string token = get_firebase_token_refresh_token(refresh_token);
-                        using traits = jwt::traits::nlohmann_json;
-                        auto decoded = jwt::decode<traits>(token);
-                        name = decoded.get_payload_claim("email").to_json();
-
-                        auto firebase_field = decoded.get_payload_claim("firebase").to_json();
-                        if(firebase_field.contains("sign_in_provider"))
-                            name += " through firebase provider: " + firebase_field["sign_in_provider"].get<std::string>();
-                    }
-                }
-            }
-
-            if(connection_authed)
-                name += " (CONNECTED)";
-            else if(fatal_connection_error)
-                name += " (CONNECITON ERROR)";
-            else
-                name += " (DISCONNECTED)";
-            return name;
-        }
-
-        Butler::header_list_t Butler::prepare_send_headers(void) {
-            determine_refresh_token();
-            decltype(network)::header_list_t headers;
-            std::string auth_token = get_firebase_token_refresh_token(refresh_token);
-            headers.emplace_back("X-Auth-Token", auth_token);
-            headers.emplace_back("X-Requested-Version", to_str(zefdb_protocol_version_max));
-            return headers;
-        }
-
-        void Butler::start_connection() {
-            if(should_stop) {
-                print_backtrace();
-                std::cerr << "What the hell is getting here all the time?" << std::endl;
-                return;
-            }
-            // This is to prevent multiple threads attacking this function at
-            // the same time.
-            static std::atomic<bool> _starting = false;
-            bool was_starting = _starting.exchange(true);
-            if(was_starting)
-                return;
-
-            if(network.is_running())
-                return;
-
-            std::string auto_connect = std::get<std::string>(get_config_var("login.autoConnect"));
-            if(have_auth_credentials() || auto_connect == "always")
-                // Note: in the case of already having credentials,
-                // ensure_auth_credentials, will make sure the credentials are
-                // up to date
-                ensure_auth_credentials();
-
-            // if(want_upstream_connection()) {
-                network.prepare_headers_func = std::bind(&Butler::prepare_send_headers, this);
-                network.start_running();
-            // }
-            _starting = false;
-        }
-
-        void Butler::stop_connection() {
-            if(!network.is_running())
-                return;
-
-            network.stop_running();
-            if(zwitch.zefhub_communication_output())
-                std::cerr << "Disconnecting from ZefHub" << std::endl;
-        }
-
-        void Butler::handle_successful_auth() {
-            if(zwitch.zefhub_communication_output())
-                std::cerr << "Authenticated with ZefHub" << std::endl;
-
-            if(have_logged_in_as_guest && !zwitch.extra_quiet()) {
-                std::cerr << std::endl;
-                std::cerr << "=================================" << std::endl;
-                std::cerr << "You are logged in as a guest user, which allows you to view public graphs but" << std::endl;
-                std::cerr << "does not allow for synchronising new graphs with ZefHub." << std::endl;
-                std::cerr << std::endl;
-                std::cerr << "Disclaimer: any ETs, RTs, ENs, and KWs that you query will be stored with ZefHub." << std::endl;
-                std::cerr << "=================================" << std::endl;
-                std::cerr << std::endl;
-            }
-            debug_time_print("successful auth");
-            update(auth_locker, connection_authed, true);
-
-            // Get hostname with C function. Have to make sure it is null terminated ourselves.
-            char hostname[80];
-            gethostname(hostname, 80);
-            hostname[79] = 0;
-            send_ZH_message({
-                    {"msg_type", "register_metadata"},
-                    {"hostname", hostname},
-                    {"client_version", zefdb_protocol_version.load()}
-                });
-
-            // Request our tokens
-            send_ZH_message({
-                    {"msg_type", "token"},
-                    {"msg_version", 1},
-                    {"action", "list"}
-                });
-
-            // To tell the managers about the reconnection, we should jump out
-            // of the network thread and into the butler thread, which requires
-            // a msg push here.
-            msg_push(Reconnected{}, false, true);
-            debug_time_print("finish successful auth");
-        }
-
-        std::future<Response> Butler::msg_push_internal(Request && content, bool ignore_closed) {
-            // TODO: with this style, we no longer need shared pointers... I
-            // don't think? I'm less confident about this now...
-            auto msg = std::make_shared<RequestWrapper>(std::move(content));
-            auto future = msg->promise.get_future();
-            msgqueue.push(std::move(msg), ignore_closed);
-            return future;
-        }
-
-        void Butler::msg_push(Request && content, bool wait, bool ignore_closed) {
-            auto future = msg_push_internal(std::move(content), ignore_closed);
-            if(wait)
-                future.get();
-        }
-
-        std::string Butler::upstream_name() {
-            if(network.uri == "")
-                return "LOCAL";
-
-            std::string name = network.uri;
-            // Remove any "wss://" at the front
-            int after_protocol = name.find("//");
-            if(after_protocol != std::string::npos)
-                name = name.substr(after_protocol+2);
-
-            // Remove anypart with a "/".
-            int maybe_slash = name.find("/");
-            if(maybe_slash != std::string::npos)
-                name = name.substr(0, maybe_slash);
-
-            return name;
-        }
-
-        //////////////////////////////
-        // * Credentials
-
-
-        std::optional<std::string> Butler::load_forced_zefhub_key() {
-            if(session_auth_key)
-                return session_auth_key;
-
-            char * env = std::getenv("ZEFHUB_AUTH_KEY");
-            if (env != nullptr && env[0] != '\0')
-                return std::string(env);
-
-            auto path = zefdb_config_path();
-            path /= "zefhub.key";
-            if (std::filesystem::exists(path)) {
-                std::ifstream file(path);
-                std::string output;
-                std::getline(file, output);
-                return output;
-            }
-
-            // Old location for fallback
-#ifdef ZEF_WIN32
-            env = std::getenv("LOCALAPPDATA");
-#else
-            env = std::getenv("HOME");
-#endif
-            std::filesystem::path path2(env);
-            path2 /= ".zefdb";
-            path2 /= "zefhub.key";
-            if (std::filesystem::exists(path2)) {
-                std::ifstream file(path2);
-                std::string output;
-                std::getline(file, output);
-                return output;
-            }
-
-            return {};
-        }
-
-        bool Butler::have_auth_credentials() {
-            // TODO: This should become more sophisticated in the future
-            if(load_forced_zefhub_key())
-                return true;
-
-            if(is_credentials_file_valid())
-                return true;
-
-            if(have_logged_in_as_guest)
-                return true;
-
-            return false;
-        }
-
-        bool Butler::is_credentials_file_valid() {
-            auto credentials_file = zefdb_config_path() / "credentials";
-            if(!std::filesystem::exists(credentials_file))
-                return false;
-
-            std::ifstream file(credentials_file);
-            if(!json::accept(file)) {
-                if(!zwitch.extra_quiet())
-                    std::cerr << "Credentials file is not in json format" << std::endl;
-                return false;
-            }
-
-            file.seekg(0, file.beg);
-            json j = json::parse(file);
-            if(!j.contains("refresh_token")) {
-                if(!zwitch.extra_quiet())
-                    std::cerr << "Credentials file does not have a refresh_token field" << std::endl;
-                return false;
-            }
-
-            // TODO: Future auth methods may have expiry times here.
-            return true;
-        }
-
-        void Butler::ensure_auth_credentials() {
-            // TODO: mutex here
-            
-            std::optional<std::string> forced_zefhub_key = load_forced_zefhub_key(); 
-            if(forced_zefhub_key) {
-                if(*forced_zefhub_key == constants::zefhub_guest_key)
-                    have_logged_in_as_guest = true;
-                else
-                    get_firebase_refresh_token_email(*forced_zefhub_key);
-            } else {
-                auto credentials_file = zefdb_config_path() / "credentials";
-                if(is_credentials_file_valid())
-                    return;
-                if(have_logged_in_as_guest)
-                    return;
-                // TODO: Get from config
-                port_t port_start = 7000;
-                port_t port_end = 9000;
-                auto auth_server = manage_local_auth_server(port_start, port_end);
-                if(!auth_server->wait_with_timeout()) {
-                    throw std::runtime_error("Unable to obtain credentials");
-                }
-
-                if (!auth_server->reply) {
-                    throw std::runtime_error("Someting went wrong with the auth server");
-                }
-                if(auth_server->reply == "GUEST") {
-                    have_logged_in_as_guest = true;
-                    if(zwitch.zefhub_communication_output())
-                        std::cerr << "Logging in as guest" << std::endl;
-                } else {
-                    have_logged_in_as_guest = false;
-                    std::ofstream file(credentials_file);
-                    file << *(auth_server->reply);
-                    if(zwitch.zefhub_communication_output())
-                        std::cerr << "Successful obtained credentials" << std::endl;
-                    if(!zwitch.extra_quiet()) {
-                        std::cerr << std::endl;
-                        std::cerr << "=================================" << std::endl;
-                        std::cerr << "You are now logged in to ZefHub. You can synchronize graphs which will enable them to be stored on" << std::endl;
-                        std::cerr << "ZefHub. Any ETs, RT, ENs and KWs you create will also be synchronized with ZefHub." << std::endl;
-                        std::cerr << std::endl;
-                        std::cerr << "Note: your credentials have been stored at '" + credentials_file.string() + "'." << std::endl;
-                        std::cerr << "By default these will be used to automatically connect to ZefHub on zef import." << std::endl;
-                        std::cerr << "If you would like to change this behavior, please see the `config` zefop for more information."  << std::endl;
-                        std::cerr << "=================================" << std::endl;
-                        std::cerr << std::endl;
-                    }
-                }
-            }
-        }
-
-        void Butler::user_login() {
-            // Always remove this. It can't hurt and can only be confusing if we leave it set.
-            butler->session_auth_key.reset();
-
-            if(load_forced_zefhub_key())
-                throw std::runtime_error("Can't login when an explicit key is given in ZEFHUB_AUTH_KEY or zefhub.key");
-            if(have_auth_credentials())
-                throw std::runtime_error("Can't login when credentials already present.");
-            ensure_auth_credentials();
-
-            // Immediately try and connect - but flag that we have just authed,
-            // so it could be that there is a delay for firebase to update
-            // zefhub about the new account.
-            network.allowed_silent_failures = 5;
-            start_connection();
-        }
-
-        void Butler::user_logout() {
-            // // Always remove any graphs that were being synced with ZefHub. Do
-            // // this first as the sync messages might require the right
-            // // credentials.
-            // std::vector<std::shared_ptr<Butler::GraphTrackingData>> saved_list;
-            // {
-            //     std::lock_guard lock(butler->graph_manager_list_mutex);
-            //     saved_list = butler->graph_manager_list;
-            // }
-            // for(auto data : butler->graph_manager_list) {
-            //     if(!data->gd->should_sync)
-            //         continue;
-
-            //     std::cerr << "Going to stop graph manager: " << uid(*(data->gd)) << std::endl;
-
-            //     try {
-            //         data->queue.set_closed(false);
-            //         long_wait_or_kill(*data->managing_thread, data->return_value, data->uid);
-            //     } catch(const std::exception & e) {
-            //         std::cerr << "Exception while trying to close queue for graph manager thread: " << e.what() << std::endl;
-            //     }
-            // }
-            
-
-            butler->session_auth_key.reset();
-
-            // Now remove credentials
-            if(load_forced_zefhub_key())
-                throw std::runtime_error("Can't logout when an explicit key is given in ZEFHUB_AUTH_KEY or zefhub.key");
-            if(have_auth_credentials()) {
-                auto credentials_file = zefdb_config_path() / "credentials";
-                if(std::filesystem::exists(credentials_file))
-                    std::filesystem::remove(credentials_file);
-            } else if(have_logged_in_as_guest) {
-            } else {
-                std::cerr << "Warning: no credentials, so logout did not remove any." << std::endl;
-            }
-
-            // Always just in case
-            have_logged_in_as_guest = false;
-
-            // Always disconnect
-            stop_connection();
-
-        }
-
-
         ////////////////////////////////////////
         // * Memory management
 
@@ -2092,26 +880,6 @@ namespace zefDB {
             return *local_process_graph;
         }
 
-        std::filesystem::path zefdb_config_path() {
-            char * env = std::getenv("ZEFDB_CONFIG_PATH");
-            if (env != nullptr)
-                return std::string(env);
-
-#ifdef ZEF_WIN32
-            env = std::getenv("LOCALAPPDATA");
-#else
-            env = std::getenv("HOME");
-#endif
-            if (env == nullptr)
-                throw std::runtime_error("No HOME env set!");
-
-            std::filesystem::path path(env);
-            path /= ".zef";
-            if(!std::filesystem::exists(path))
-                std::filesystem::create_directories(path);
-            return path;
-        }
-
         ////////////////////////////////////////////////////////
         // * External handlers
 
diff --git a/core/src/butler/butler_connection.cpp b/core/src/butler/butler_connection.cpp
new file mode 100644
index 00000000..6535118d
--- /dev/null
+++ b/core/src/butler/butler_connection.cpp
@@ -0,0 +1,474 @@
+// Copyright 2022 Synchronous Technologies Pte Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+////////////////////////////////////////////////
+// * Connection management
+
+bool Butler::want_upstream_connection() {
+    if(check_env_bool("ZEFDB_OFFLINE_MODE"))
+        return false;
+
+    if(network.uri == "")
+        return false;
+
+    // If we are running, then we want to continue.
+    if(network.is_running())
+        return true;
+
+    std::string auto_connect = std::get<std::string>(get_config_var("login.autoConnect"));
+
+    if(auto_connect == "auto" && have_auth_credentials())
+        return true;
+    if(auto_connect == "always")
+        return true;
+
+    return false;
+}
+
+bool Butler::wait_for_auth(std::chrono::duration<double> timeout) {
+    if(butler_is_master)
+        throw std::runtime_error("Butler started as master - should never have got to this point of requesting auth.");
+    if(network.uri == "")
+        throw std::runtime_error("Not connecting to ZefHub without a URL");
+    if(!want_upstream_connection())
+        throw std::runtime_error("Not connecting to ZefHub until we have credentials. Either login using `login | run` to store your credentials or `login_as_guest | run` for a temporary guest login.");
+    if(!network.managing_thread) {
+        // throw std::runtime_error("Network is not trying to connect, can't wait for auth.");
+        debug_time_print("before start connection");
+        if(should_stop) {
+            return false;
+        }
+        start_connection();
+    }
+    auto done_auth = [this]() { return should_stop || connection_authed || fatal_connection_error; };
+    if(timeout < std::chrono::duration<double>(0)) {
+        // wait_pred(auth_locker, done_auth);
+        // We have a special spam message in here to let users know what's going on. 
+        wait_pred(auth_locker, done_auth, std::chrono::seconds(3));
+        if(!done_auth()) {
+            std::cerr << "Warning: waiting for connection with ZefHub is taking a long time.\n\nIf you would like to see more information enable debug messages through `zwitch.zefhub_communication_output(True)` or setting the environment variable `ZEFDB_VERBOSE=true`.\n\nIf you would like to run Zef in offline mode, then start a new python session with the environment variable `ZEFDB_OFFLINE_MODE=TRUE`." << std::endl;
+            wait_pred(auth_locker, done_auth);
+        }
+    }
+    else {
+        if(!wait_pred(auth_locker, done_auth, timeout))
+            return false;
+    }
+
+    if(no_credentials_warning)
+        throw std::runtime_error("No credentials present to allow auth to take place.");
+    if(fatal_connection_error)
+        throw std::runtime_error("Fatal connection error while trying to auth with ZefHub.\n\nIf you would like to run in offline mode, please restart your python session with the environment variable `ZEFDB_OFFLINE_MODE=TRUE`.");
+    debug_time_print("finish wait_for_auth");
+    return true;
+}
+
+void Butler::determine_refresh_token() {
+    std::optional<std::string> key_string = load_forced_zefhub_key();
+    if(key_string) {
+        if(*key_string == constants::zefhub_guest_key) {
+            std::cerr << "Connecting as guest user" << std::endl;
+            refresh_token = "";
+            have_logged_in_as_guest = true;
+        } else 
+            refresh_token = get_firebase_refresh_token_email(*key_string);
+        return;
+    } else if(!have_auth_credentials()) {
+        no_credentials_warning = true;
+        throw std::runtime_error("Have no existing credentials to determine auth token. You must login first.");
+    } else {
+        ensure_auth_credentials();
+        if(have_logged_in_as_guest) {
+            refresh_token = "";
+        } else {
+            auto credentials_file = credentials_path();
+            std::ifstream file(credentials_file);
+            json j = json::parse(file);
+            refresh_token = j["refresh_token"].get<std::string>();
+        }
+    }
+}
+
+std::string Butler::who_am_i() {
+    std::string name;
+
+    std::optional<std::string> key_string = load_forced_zefhub_key();
+    if(key_string) {
+        if(*key_string == constants::zefhub_guest_key) {
+            name = "GUEST via forced key";
+        } else {
+            // This is  duplicated with get_firebase_refresh_token_email which is annoying
+            int colon = key_string->find(':');
+            name = key_string->substr(0,colon) + " via forced key";
+        }
+    } else if(!have_auth_credentials()) {
+        name = "";
+    } else {
+        if(have_logged_in_as_guest) {
+            name = "GUEST via login prompt";
+        } else {
+            auto credentials_file = credentials_path();
+            if(!std::filesystem::exists(credentials_file)) {
+                name = "";
+            } else {
+                std::ifstream file(credentials_file);
+                json j = json::parse(file);
+                refresh_token = j["refresh_token"].get<std::string>();
+                std::string token = get_firebase_token_refresh_token(refresh_token);
+                using traits = jwt::traits::nlohmann_json;
+                auto decoded = jwt::decode<traits>(token);
+                name = decoded.get_payload_claim("email").to_json();
+
+                auto firebase_field = decoded.get_payload_claim("firebase").to_json();
+                if(firebase_field.contains("sign_in_provider"))
+                    name += " through firebase provider: " + firebase_field["sign_in_provider"].get<std::string>();
+            }
+        }
+    }
+
+    if(connection_authed)
+        name += " (CONNECTED)";
+    else if(fatal_connection_error)
+        name += " (CONNECITON ERROR)";
+    else
+        name += " (DISCONNECTED)";
+    return name;
+}
+
+Butler::header_list_t Butler::prepare_send_headers(void) {
+    determine_refresh_token();
+    decltype(network)::header_list_t headers;
+    std::string auth_token = get_firebase_token_refresh_token(refresh_token);
+    headers.emplace_back("X-Auth-Token", auth_token);
+    headers.emplace_back("X-Requested-Version", to_str(zefdb_protocol_version_max));
+    return headers;
+}
+
+void Butler::start_connection() {
+    if(should_stop) {
+        print_backtrace();
+        std::cerr << "What the hell is getting here all the time?" << std::endl;
+        return;
+    }
+    // This is to prevent multiple threads attacking this function at
+    // the same time.
+    static std::atomic<bool> _starting = false;
+    bool was_starting = _starting.exchange(true);
+    if(was_starting)
+        return;
+
+    if(network.is_running())
+        return;
+
+    std::string auto_connect = std::get<std::string>(get_config_var("login.autoConnect"));
+    if(have_auth_credentials() || auto_connect == "always")
+        // Note: in the case of already having credentials,
+        // ensure_auth_credentials, will make sure the credentials are
+        // up to date
+        ensure_auth_credentials();
+
+    // if(want_upstream_connection()) {
+    network.prepare_headers_func = std::bind(&Butler::prepare_send_headers, this);
+    network.start_running();
+    // }
+    _starting = false;
+}
+
+void Butler::stop_connection() {
+    if(!network.is_running())
+        return;
+
+    network.stop_running();
+    if(zwitch.zefhub_communication_output())
+        std::cerr << "Disconnecting from ZefHub" << std::endl;
+}
+
+void Butler::handle_successful_auth() {
+    if(zwitch.zefhub_communication_output())
+        std::cerr << "Authenticated with ZefHub" << std::endl;
+
+    if(have_logged_in_as_guest && !zwitch.extra_quiet()) {
+        std::cerr << std::endl;
+        std::cerr << "=================================" << std::endl;
+        std::cerr << "You are logged in as a guest user, which allows you to view public graphs but" << std::endl;
+        std::cerr << "does not allow for synchronising new graphs with ZefHub." << std::endl;
+        std::cerr << std::endl;
+        std::cerr << "Disclaimer: any ETs, RTs, ENs, and KWs that you query will be stored with ZefHub." << std::endl;
+        std::cerr << "=================================" << std::endl;
+        std::cerr << std::endl;
+    }
+    debug_time_print("successful auth");
+    update(auth_locker, connection_authed, true);
+
+    // Get hostname with C function. Have to make sure it is null terminated ourselves.
+    char hostname[80];
+    gethostname(hostname, 80);
+    hostname[79] = 0;
+    send_ZH_message({
+            {"msg_type", "register_metadata"},
+            {"hostname", hostname},
+            {"client_version", zefdb_protocol_version.load()}
+        });
+
+    // Request our tokens
+    send_ZH_message({
+            {"msg_type", "token"},
+            {"msg_version", 1},
+            {"action", "list"}
+        });
+
+    // To tell the managers about the reconnection, we should jump out
+    // of the network thread and into the butler thread, which requires
+    // a msg push here.
+    msg_push(Reconnected{}, false, true);
+    debug_time_print("finish successful auth");
+}
+
+std::future<Response> Butler::msg_push_internal(Request && content, bool ignore_closed) {
+    // TODO: with this style, we no longer need shared pointers... I
+    // don't think? I'm less confident about this now...
+    auto msg = std::make_shared<RequestWrapper>(std::move(content));
+    auto future = msg->promise.get_future();
+    msgqueue.push(std::move(msg), ignore_closed);
+    return future;
+}
+
+void Butler::msg_push(Request && content, bool wait, bool ignore_closed) {
+    auto future = msg_push_internal(std::move(content), ignore_closed);
+    if(wait)
+        future.get();
+}
+
+std::string Butler::upstream_name() {
+    if(network.uri == "")
+        return "LOCAL";
+
+    std::string name = network.uri;
+    // Remove any "wss://" at the front
+    int after_protocol = name.find("//");
+    if(after_protocol != std::string::npos)
+        name = name.substr(after_protocol+2);
+
+    // Remove anypart with a "/".
+    int maybe_slash = name.find("/");
+    if(maybe_slash != std::string::npos)
+        name = name.substr(0, maybe_slash);
+
+    return name;
+}
+
+std::filesystem::path Butler::credentials_path() {
+    return zefdb_config_path() / upstream_name() / "credentials";
+}
+
+//////////////////////////////
+// * Credentials
+
+
+std::optional<std::string> Butler::load_forced_zefhub_key() {
+    if(session_auth_key)
+        return session_auth_key;
+
+    char * env = std::getenv("ZEFHUB_AUTH_KEY");
+    if (env != nullptr && env[0] != '\0')
+        return std::string(env);
+
+    auto path = zefdb_config_path();
+    path /= "zefhub.key";
+    if (std::filesystem::exists(path)) {
+        std::ifstream file(path);
+        std::string output;
+        std::getline(file, output);
+        return output;
+    }
+
+    // Old location for fallback
+#ifdef ZEF_WIN32
+    env = std::getenv("LOCALAPPDATA");
+#else
+    env = std::getenv("HOME");
+#endif
+    std::filesystem::path path2(env);
+    path2 /= ".zefdb";
+    path2 /= "zefhub.key";
+    if (std::filesystem::exists(path2)) {
+        std::ifstream file(path2);
+        std::string output;
+        std::getline(file, output);
+        return output;
+    }
+
+    return {};
+}
+
+bool Butler::have_auth_credentials() {
+    // TODO: This should become more sophisticated in the future
+    if(load_forced_zefhub_key())
+        return true;
+
+    if(is_credentials_file_valid())
+        return true;
+
+    if(have_logged_in_as_guest)
+        return true;
+
+    return false;
+}
+
+bool Butler::is_credentials_file_valid() {
+    auto credentials_file = credentials_path();
+    if(!std::filesystem::exists(credentials_file))
+        return false;
+
+    std::ifstream file(credentials_file);
+    if(!json::accept(file)) {
+        if(!zwitch.extra_quiet())
+            std::cerr << "Credentials file is not in json format" << std::endl;
+        return false;
+    }
+
+    file.seekg(0, file.beg);
+    json j = json::parse(file);
+    if(!j.contains("refresh_token")) {
+        if(!zwitch.extra_quiet())
+            std::cerr << "Credentials file does not have a refresh_token field" << std::endl;
+        return false;
+    }
+
+    // TODO: Future auth methods may have expiry times here.
+    return true;
+}
+
+void Butler::ensure_auth_credentials() {
+    // TODO: mutex here
+            
+    std::optional<std::string> forced_zefhub_key = load_forced_zefhub_key(); 
+    if(forced_zefhub_key) {
+        if(*forced_zefhub_key == constants::zefhub_guest_key)
+            have_logged_in_as_guest = true;
+        else
+            get_firebase_refresh_token_email(*forced_zefhub_key);
+    } else {
+        auto credentials_file = credentials_path();
+        if(is_credentials_file_valid())
+            return;
+        if(have_logged_in_as_guest)
+            return;
+        // TODO: Get from config
+        port_t port_start = 7000;
+        port_t port_end = 9000;
+        auto auth_server = manage_local_auth_server(port_start, port_end);
+        if(!auth_server->wait_with_timeout()) {
+            throw std::runtime_error("Unable to obtain credentials");
+        }
+
+        if (!auth_server->reply) {
+            throw std::runtime_error("Someting went wrong with the auth server");
+        }
+        if(auth_server->reply == "GUEST") {
+            have_logged_in_as_guest = true;
+            if(zwitch.zefhub_communication_output())
+                std::cerr << "Logging in as guest" << std::endl;
+        } else {
+            have_logged_in_as_guest = false;
+            {
+                std::filesystem::create_directories(credentials_file.parent_path());
+                std::ofstream file(credentials_file);
+                file << *(auth_server->reply);
+                if(file.fail())
+                    throw std::runtime_error("Problem saving credentials to file!");
+            }
+            if(zwitch.zefhub_communication_output())
+                std::cerr << "Successful obtained credentials" << std::endl;
+            if(!zwitch.extra_quiet()) {
+                std::cerr << std::endl;
+                std::cerr << "=================================" << std::endl;
+                std::cerr << "You are now logged in to ZefHub. You can synchronize graphs which will enable them to be stored on" << std::endl;
+                std::cerr << "ZefHub. Any ETs, RT, ENs and KWs you create will also be synchronized with ZefHub." << std::endl;
+                std::cerr << std::endl;
+                std::cerr << "Note: your credentials have been stored at '" + credentials_file.string() + "'." << std::endl;
+                std::cerr << "By default these will be used to automatically connect to ZefHub on zef import." << std::endl;
+                std::cerr << "If you would like to change this behavior, please see the `config` zefop for more information."  << std::endl;
+                std::cerr << "=================================" << std::endl;
+                std::cerr << std::endl;
+            }
+        }
+    }
+}
+
+void Butler::user_login() {
+    // Always remove this. It can't hurt and can only be confusing if we leave it set.
+    auto butler = get_butler();
+    butler->session_auth_key.reset();
+
+    if(load_forced_zefhub_key())
+        throw std::runtime_error("Can't login when an explicit key is given in ZEFHUB_AUTH_KEY or zefhub.key");
+    if(have_auth_credentials())
+        throw std::runtime_error("Can't login when credentials already present.");
+    ensure_auth_credentials();
+
+    // Immediately try and connect - but flag that we have just authed,
+    // so it could be that there is a delay for firebase to update
+    // zefhub about the new account.
+    network.allowed_silent_failures = 5;
+    start_connection();
+}
+
+void Butler::user_logout() {
+    // // Always remove any graphs that were being synced with ZefHub. Do
+    // // this first as the sync messages might require the right
+    // // credentials.
+    // std::vector<std::shared_ptr<Butler::GraphTrackingData>> saved_list;
+    // {
+    //     std::lock_guard lock(butler->graph_manager_list_mutex);
+    //     saved_list = butler->graph_manager_list;
+    // }
+    // for(auto data : butler->graph_manager_list) {
+    //     if(!data->gd->should_sync)
+    //         continue;
+
+    //     std::cerr << "Going to stop graph manager: " << uid(*(data->gd)) << std::endl;
+
+    //     try {
+    //         data->queue.set_closed(false);
+    //         long_wait_or_kill(*data->managing_thread, data->return_value, data->uid);
+    //     } catch(const std::exception & e) {
+    //         std::cerr << "Exception while trying to close queue for graph manager thread: " << e.what() << std::endl;
+    //     }
+    // }
+            
+
+    auto butler = get_butler();
+    butler->session_auth_key.reset();
+
+    // Now remove credentials
+    if(load_forced_zefhub_key())
+        throw std::runtime_error("Can't logout when an explicit key is given in ZEFHUB_AUTH_KEY or zefhub.key");
+    if(have_auth_credentials()) {
+        auto credentials_file = credentials_path();
+        if(std::filesystem::exists(credentials_file))
+            std::filesystem::remove(credentials_file);
+    } else if(have_logged_in_as_guest) {
+    } else {
+        std::cerr << "Warning: no credentials, so logout did not remove any." << std::endl;
+    }
+
+    // Always just in case
+    have_logged_in_as_guest = false;
+
+    // Always disconnect
+    stop_connection();
+
+}
+
diff --git a/core/src/butler/butler_handlers_graph_manager.cpp b/core/src/butler/butler_handlers_graph_manager.cpp
index 5f2e1820..6d36bdbd 100644
--- a/core/src/butler/butler_handlers_graph_manager.cpp
+++ b/core/src/butler/butler_handlers_graph_manager.cpp
@@ -308,13 +308,6 @@ void Butler::graph_worker_handle_message(Butler::GraphTrackingData & me, NewGrap
             apply_update_with_caches(*me.gd, *content.payload, false, false);
             me.gd->manager_tx_head = me.gd->latest_complete_tx.load();
             // Note: don't set sync_head here, it should remain at 0.
-        } else {
-            // create a first transaction on a new graph. This layout allows us to assume that there always is a tx,
-            // which helps in various high level functions.
-            //
-            // Note: this is done here instead of in the GraphData constructor
-            // as creating a transaction will trigger the sync thread.
-            auto my_tx = Transaction(*me.gd, false, false);
         }
 
         if(!content.internal_use_only) {
@@ -389,11 +382,12 @@ void Butler::graph_worker_handle_message(Butler::GraphTrackingData & me, LoadGra
             // this out for a call to do_reconnect, but there is some custom
             // logic to handle in here, with regards to safety checks on the
             // graph.
-            me.gd->read_head = fg->get_latest_blob_index();
-            me.gd->write_head = fg->get_latest_blob_index();
-            me.gd->latest_complete_tx = index(internals::get_latest_complete_tx_node(*me.gd));
             me.gd->manager_tx_head = me.gd->latest_complete_tx.load();
 
+            if(zwitch.developer_output()) {
+                std::cerr << "Loaded GUID: " << me.uid << " has heads of read:tx = " << me.gd->read_head.load() << ":" << me.gd->latest_complete_tx.load() << std::endl;
+            }
+
             json j{
                     {"msg_type", "subscribe_to_graph"},
                     {"msg_version", 3},
@@ -1031,7 +1025,7 @@ void Butler::graph_worker_handle_message(Butler::GraphTrackingData & me, TagGrap
         msg->promise.set_value(GenericResponse("Can't tag graph when not being synchronised."));
         return;
     }
-    if(!wait_diff(me.gd->heads_locker, me.gd->sync_head, 0, std::chrono::duration<double>(butler_generic_timeout.value))) {
+    if(!wait_diff(me.gd->heads_locker, me.gd->sync_head, 0, std::chrono::duration<double>(butler_generic_timeout))) {
         msg->promise.set_value(GenericResponse("Timed out waiting for graph to be synced."));
         return;
     }
diff --git a/core/src/butler/butler_handlers_ws.cpp b/core/src/butler/butler_handlers_ws.cpp
index 497a5178..9223360b 100644
--- a/core/src/butler/butler_handlers_ws.cpp
+++ b/core/src/butler/butler_handlers_ws.cpp
@@ -254,48 +254,6 @@ void Butler::handle_incoming_message(json & j, std::vector<std::string> & rest)
 
 
 
-// TODO: Replace these with the individual type updates, which will occur in new messages from zefhub.
-template<typename T1, typename T2>
-void update_bidirection_name_map(thread_safe_bidirectional_map<T1,T2>& map_to_update, const T1 & indx, const T2 & name) {			
-    if(map_to_update.contains(indx)) {
-        auto it = map_to_update.at(indx);
-        if(it != name) {
-            std::cout << it << " doesn't agree with " << name << " (" << indx << ")" << std::endl;
-            throw std::runtime_error("existing index assigned to a ET/RT does not agree with newly received ET/RT type list");
-        }
-    } else if(map_to_update.contains(name)) {
-        auto it = map_to_update.at(name);
-        if(it != indx) {
-            std::cout << it << " doesn't agree with " << name << " (" << indx << ")" << std::endl;
-            throw std::runtime_error("existing name assigned to a ET/RT does not agree with newly received ET/RT type list");
-        }
-    } else
-        map_to_update.insert(indx, name);
-}
-
-
-void update_zef_enum_bidirectional_map(thread_safe_zef_enum_bidirectional_map& map_to_update, const enum_indx & indx, const std::string& name) {
-    auto [enum_type,enum_value] = internals::split_enum_string(name);
-    auto pair = std::make_pair(enum_type, enum_value);
-    if(map_to_update.contains(indx)) {
-        auto it = map_to_update.at(indx);
-        if(it.first != enum_type || it.second != enum_value) {
-                std::cout << it.first << "." << it.second << " doesn't agree with "
-                          << enum_type << "." << enum_value << " for indx " << it.first << std::endl;
-                throw std::runtime_error("existing index assigned to a enum does not agree with newly received enum type and name list");
-        }
-    } else if(map_to_update.contains(pair)) {
-        auto it = map_to_update.at(pair);
-        if(it != indx) {
-                std::cout << it << " doesn't agree with "
-                          << enum_type << "." << enum_value << " for indx " << indx << std::endl;
-                throw std::runtime_error("existing name assigned to a enum does not agree with newly received enum type and name list");
-        }
-    } else
-        map_to_update.insert(indx, pair);
-}
-
-
 void handle_token_response(Butler & butler, json & j) {
     auto& tokens = global_token_store();
 
@@ -304,34 +262,54 @@ void handle_token_response(Butler & butler, json & j) {
         std::cerr << "Got a butler-handlable token response with reason: " << reason << std::endl;
     if(reason == "list") {
         auto vec1 = j["groups"]["ET"].get<std::vector<std::pair<token_value_t, std::string>>>();
+        if(zwitch.developer_output())
+            std::cerr << "Got " << vec1.size() << " ETs in listing from upstream" << std::endl;
+        int num_before = tokens.ETs.size();
         for(auto & it : vec1) {
-            update_bidirection_name_map(tokens.ETs, std::get<0>(it), std::get<1>(it));
-            if(zwitch.developer_output())
-                std::cerr << "ET: " << std::get<1>(it) << std::endl;
+            tokens.force_add_entity_type(std::get<0>(it), std::get<1>(it));
+            // if(zwitch.developer_output())
+            //     std::cerr << "ET: " << std::get<1>(it) << std::endl;
         }
+        if(zwitch.developer_output())
+            std::cerr << "Of these, " << tokens.ETs.size() - num_before << " were new to us." << std::endl;
 
         auto vec2 = j["groups"]["RT"].get<std::vector<std::tuple<token_value_t, std::string>>>();
+        if(zwitch.developer_output())
+            std::cerr << "Got " << vec2.size() << " RTs in listing from upstream" << std::endl;
+        num_before = tokens.RTs.size();
         for(auto & it : vec2) {
-            update_bidirection_name_map(tokens.RTs, std::get<0>(it), std::get<1>(it));
-            if(zwitch.developer_output())
-                std::cerr << "RT: " << std::get<1>(it) << std::endl;
+            tokens.force_add_relation_type(std::get<0>(it), std::get<1>(it));
+            // if(zwitch.developer_output())
+            //     std::cerr << "RT: " << std::get<1>(it) << std::endl;
         }
+        if(zwitch.developer_output())
+            std::cerr << "Of these, " << tokens.RTs.size() - num_before << " were new to us." << std::endl;
 
         if(j["groups"].contains("KW")) {
             auto vec4 = j["groups"]["KW"].get<std::vector<std::tuple<token_value_t, std::string>>>();
+            if(zwitch.developer_output())
+                std::cerr << "Got " << vec4.size() << " KWs in listing from upstream" << std::endl;
+            num_before = tokens.KWs.size();
             for(auto & it : vec4) {
-                update_bidirection_name_map(tokens.KWs, std::get<0>(it), std::get<1>(it));
-                if(zwitch.developer_output())
-                    std::cerr << "KW: " << std::get<1>(it) << std::endl;
+                tokens.force_add_keyword(std::get<0>(it), std::get<1>(it));
+                // if(zwitch.developer_output())
+                //     std::cerr << "KW: " << std::get<1>(it) << std::endl;
             }
+            if(zwitch.developer_output())
+                std::cerr << "Of these, " << tokens.KWs.size() - num_before << " were new to us." << std::endl;
         }
 
         auto vec3 = j["groups"]["EN"].get<std::vector<std::tuple<enum_indx, std::string>>>();
+        if(zwitch.developer_output())
+            std::cerr << "Got " << vec3.size() << " ETs in listing from upstream" << std::endl;
+        num_before = tokens.ENs.size();
         for(auto & it : vec3) {
-            update_zef_enum_bidirectional_map(tokens.ENs, std::get<0>(it), std::get<1>(it));
-            if(zwitch.developer_output())
-                std::cerr << "EN: " << std::get<1>(it) << std::endl;
+            tokens.force_add_enum_type(std::get<0>(it), std::get<1>(it));
+            // if(zwitch.developer_output())
+            //     std::cerr << "EN: " << std::get<1>(it) << std::endl;
         }
+        if(zwitch.developer_output())
+            std::cerr << "Of these, " << tokens.ENs.size() - num_before << " were new to us." << std::endl;
     } else {
         throw std::runtime_error("Shouldn't get here - unknown reason for token response: " + reason);
     }
@@ -351,13 +329,13 @@ void handle_token_response(Butler & butler, json & j, Butler::task_promise_ptr &
 
             for(auto & [name,indx] : response.pairs) {
                 if (group == "ET")
-                    update_bidirection_name_map(tokens.ETs, indx, name);
+                    tokens.force_add_entity_type(indx, name);
                 else if(group == "RT")
-                    update_bidirection_name_map(tokens.RTs, indx, name);
+                    tokens.force_add_relation_type(indx, name);
                 else if(group == "KW")
-                    update_bidirection_name_map(tokens.KWs, indx, name);
+                    tokens.force_add_keyword(indx, name);
                 else if(group == "EN")
-                    update_zef_enum_bidirectional_map(tokens.ENs, indx, name);
+                    tokens.force_add_enum_type(indx, name);
             }
         } else {
             std::cerr << "WARNING: unexpected reason (" << response.generic.reason << ") during handle_token_response" << std::endl;
diff --git a/core/src/butler/butler_tasks.cpp b/core/src/butler/butler_tasks.cpp
new file mode 100644
index 00000000..a6419ec9
--- /dev/null
+++ b/core/src/butler/butler_tasks.cpp
@@ -0,0 +1,398 @@
+// Copyright 2022 Synchronous Technologies Pte Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+
+////////////////////////////////
+// * Task handling and WS msgs
+
+Butler::task_ptr Butler::add_task(bool is_online, double timeout) {
+    return add_task(is_online, timeout, std::promise<Response>(), true);
+};
+Butler::task_ptr Butler::add_task(bool is_online, double timeout, std::promise<Response> && promise, bool acquire_future) {
+    std::lock_guard lock(waiting_tasks_mutex);
+#ifdef ZEF_DEBUG
+    if(waiting_tasks.size() > 100)
+        std::cerr << "Warning, there are a lot of tasks building up in the task_list! These should probably be removed at some point." << std::endl;
+#endif
+    task_ptr task = std::make_shared<Task>(now(), is_online, timeout, promise, acquire_future);
+    waiting_tasks.emplace_back(std::make_shared<TaskPromise>(task, std::move(promise)));
+    return task;
+};
+// Search and pop a task from the waiting tasks. Return true/false if found.
+// Note the iteration style is due to the forward_list type.
+Butler::task_promise_ptr Butler::find_task(std::string task_uid, bool forget) {
+    std::lock_guard lock(waiting_tasks_mutex);
+    for(auto it = waiting_tasks.begin() ; it != waiting_tasks.end() ; it++) {
+        if ((*it)->task->task_uid == task_uid) {
+            if (forget) {
+                Butler::task_promise_ptr ret = *it;
+                waiting_tasks.erase(it);
+                return ret;
+            } else {
+                return *it;
+            }
+        }
+    }
+    return {};
+}
+bool Butler::forget_task(std::string task_uid) {
+    return (bool)find_task(task_uid, true);
+}
+void Butler::cancel_online_tasks() {
+    std::lock_guard lock(waiting_tasks_mutex);
+
+    waiting_tasks.erase(
+                        std::remove_if(waiting_tasks.begin(),
+                                       waiting_tasks.end(),
+                                       [](task_promise_ptr & task_promise) {
+                                           if(task_promise->task->is_online) {
+                                               task_promise->promise.set_exception(std::make_exception_ptr(Communication::disconnected_exception()));
+                                               return true;
+                                           }
+                                           return false;
+                                       }),
+                        waiting_tasks.end()
+                        );
+}
+
+Response wait_future(Butler::task_ptr task) {
+    if(task->timeout > 0) {
+        while(true) {
+            // TODO: We might have to make last_activity an atomic to make sure updates come through.
+            auto wait_time = Time(task->last_activity.load()) + task->timeout*seconds - now();
+            auto status = task->future.wait_for(std::chrono::duration<double>(wait_time.value));
+            if(status == std::future_status::ready) {
+                if(zwitch.developer_output())
+                    std::cerr << "Zefhub message took " << (now() - task->started_time) << std::endl;
+                break;
+            }
+
+            if(now() > Time(task->last_activity.load()) + task->timeout*seconds) {
+                std::cerr << "Throwing timeout exception because last_activity was " << task->last_activity << " and now is " << now() << std::endl;
+                throw Butler::timeout_exception();
+            }
+        }
+    }
+    return task->future.get();
+}
+Response Butler::wait_on_zefhub_message_any(json & j, const std::vector<std::string> & rest, double timeout, bool throw_on_failure, bool chunked) {
+    std::string msg_type = j["msg_type"].get<std::string>();
+    task_ptr task = add_task(true, timeout);
+    j["task_uid"] = task->task_uid;
+
+    if(chunked) {
+        send_chunked_ZH_message(task->task_uid, j, rest);
+    }
+    else
+        // Note: send returns before the data is actually sent.
+        send_ZH_message(j, rest);
+
+    task->last_activity = now().seconds_since_1970;
+
+    // Make sure we clean up the task from the task_list in all cases
+    RAII_CallAtEnd call_at_end([this,&task]() {
+        forget_task(task->task_uid);
+    });
+            
+    // The timeout is an inactivity timeout
+    auto result = wait_future(task);
+
+    if(throw_on_failure) {
+        GenericResponse generic = std::visit(overloaded {
+                [](auto & v)->GenericResponse { return v.generic; },
+                [](GenericResponse & v)->GenericResponse { return v; },
+                [](std::monostate & v)->GenericResponse { throw std::runtime_error("Invalid response state."); }
+            }, result);
+        if(!generic.success)
+            throw std::runtime_error("ZefHub replied with failure: " + generic.reason);
+    }
+
+    return result;
+}
+
+void Butler::fill_out_ZH_message(json & j) { //, bool add_hints) {
+    j["protocol_type"] = "ZEFDB";
+    j["protocol_version"] = zefdb_protocol_version.load();
+    j["who"] = get_firebase_token_refresh_token(refresh_token);
+    // if(add_hints) {
+    if(estimated_transfer_speed_accum_time > 0) {
+        double speed = estimated_transfer_speed_accum_bytes / estimated_transfer_speed_accum_time;
+        double goal_time = chunk_timeout / 5;
+        double est_chunk_size = goal_time * speed;
+        j["hint_chunk_size"] = est_chunk_size;
+    }
+    // }
+}
+
+void Butler::send_ZH_message(json & j, const std::vector<std::string> & rest) {
+    if(butler_is_master)
+        throw std::runtime_error("Should not be trying to send messages to ZefHub when we are running in offline mode.");
+    if(!want_upstream_connection()) {
+        throw Communication::disconnected_exception();
+    }
+
+    // We do our own wait here, so that this is a proper timeout.
+    // network.wait_for_connected(constants::zefhub_reconnect_timeout);
+    // wait_for_auth(constants::zefhub_reconnect_timeout);
+    wait_for_auth();
+    if(!network.connected)
+        throw Communication::disconnected_exception();
+
+    // Note that "fill out" has to happen after the auth wait, as we
+    // won't know the protocol versions etc before then.
+    fill_out_ZH_message(j);
+
+    if(zwitch.debug_zefhub_json_output()) {
+        std::cerr << "About to send out: " << j << std::endl;
+        std::cerr << "REST sizes: [";
+        for(auto & it : rest) {
+            std::cerr << it.size() << ",";
+        }
+        std::cerr << "]" << std::endl;
+    }
+
+    auto zh_msg = Communication::prepare_ZH_message(j, rest);
+
+    network.send(zh_msg);
+}
+
+void Butler::send_chunked_ZH_message(std::string main_task_uid, json & j, const std::vector<std::string> & rest) {
+    if(zefdb_protocol_version <= 4) {
+        send_ZH_message(j, rest);
+        return;
+    }
+
+    Butler::task_promise_ptr main_task = find_task(main_task_uid);
+
+    // Currently hardcoded but we need a way to extract this from config.
+    int chunk_size = std::max(this->chunked_transfer_size, this->chunked_transfer_size_min);
+    // This uid links all chunks in the transfer together.
+    std::string chunk_uid = generate_random_task_uid();
+
+    std::vector<int> rest_sizes(rest.size());
+    for(int i = 0; i < rest.size(); i++)
+        rest_sizes[i] = rest[i].size();
+
+    fill_out_ZH_message(j);
+    task_ptr task = add_task(true, chunk_timeout);
+    send_ZH_message(json{
+            {"msg_type", "chunked"},
+            {"task_uid", task->task_uid},
+            {"chunk_type", "new"},
+            {"chunk_uid", chunk_uid},
+            {"rest_sizes", rest_sizes},
+            {"msg", j}
+        });
+    std::deque<std::tuple<task_ptr,Time,int>> futures;
+    futures.emplace_back(task, now(), 1);
+
+    auto do_wait = [&](task_ptr & task) -> double {
+        try {
+            auto start_time = now();
+            auto result = wait_future(task);
+            double this_waited_time = (now() - start_time).value;
+                    
+            auto resp = std::visit(overloaded {
+                    [](const GenericZefHubResponse & response) { return response; },
+                    [](const auto & response) -> GenericZefHubResponse {
+                        throw std::runtime_error("Response from ZefHub is not of the right type in send_chunked_ZH_message.");
+                    }
+                }, result);
+            if(!resp.generic.success)
+                throw std::runtime_error("Failure in chunked send: " + resp.generic.reason);
+            main_task->task->last_activity = now().seconds_since_1970;
+            // std::cerr << "Got back one response to a chunk after waiting: " << this_waited_time << std::endl;
+            return this_waited_time;
+        } catch(const timeout_exception & exc) {
+            // Reduce the chunk size. Unfortunately we can't restart the
+            // transfer (without making a new chunk uid) as it could be
+            // that the packets are still en route. This needs to be
+            // improved.
+            //
+            // What could happen is a query back to ask "where are you
+            // up to?" and then continuing from that point. This would
+            // require removing the for loop over rest_index below, and
+            // reacting to variables.
+            if(chunked_transfer_auto_adjust)
+                // Big adjustment down, smaller adjustments up.
+                chunked_transfer_size = std::max(chunked_transfer_size/100, chunked_transfer_size_min);
+            std::rethrow_exception(std::current_exception());
+        }
+    };
+
+    auto start_time = now();
+    int transfer_total_size = 0;
+
+    // Split rest into chunks
+    for(int rest_index = 0;rest_index < rest.size();rest_index++) {
+        // The chunk size can change over the message, so we do this chunking in a kind of weird way.
+        int rest_size = rest[rest_index].size();
+        transfer_total_size += rest_size;
+        int cur_start = 0;
+        while(cur_start < rest_size) {
+            int this_chunk_size = std::min(chunk_size, rest_size - cur_start);
+            std::string this_bytes = rest[rest_index].substr(cur_start, this_chunk_size);
+            if(this_bytes.size() == 0) {
+                std::cerr << "SIZE is ZERO: ";
+                std::cerr << this_chunk_size << ", " << chunk_size << this->chunked_transfer_size << std::endl;
+            }
+            // std::cerr << "Sending a chunk, size: " << this_bytes.size() << ", start: " << cur_start << "/" << rest_size << ", rest_index: " << rest_index << std::endl;
+            task = add_task(true, chunk_timeout);
+            send_ZH_message(json{
+                    {"task_uid", task->task_uid},
+                    {"msg_type", "chunked"},
+                    {"chunk_type", "payload"},
+                    {"chunk_uid", chunk_uid},
+                    {"rest_index", rest_index},
+                    {"bytes_start", cur_start}
+                }, {this_bytes});
+            futures.emplace_back(task, now(), this_chunk_size);
+
+            if(futures.size() >= this->chunked_transfer_queued) {
+                // Wait on first future
+                auto & it = futures.front();
+                // Wait time = time spent in waiting here
+                // Delta time = time since first starting the task
+                // Size = is used to put the time in context - a super
+                //        fast send of 10 bytes should not up later sends of
+                //        1MB without caution.
+                //
+                // Note that sizes are a bit misleading too - zstd is
+                // compressing the chunks so the calculated speed is not
+                // exactly the same. This is open to huge problems so
+                // need to revisit this I think.
+
+                double first_wait_time = do_wait(std::get<0>(it));
+                double first_delta_time = (now() - std::get<1>(it)).value;
+                int first_size = std::get<2>(it);
+                futures.pop_front();
+
+                if(chunked_transfer_auto_adjust) {
+                    // Now wait on more until the queue is half-empty, this
+                    // gives us a way to check the response times. If they
+                    // are ping-dominated then the spacing between these
+                    // times will be heavily bimodal (i.e. one slow and the
+                    // rest nearly instant afterwards).
+                    double max_wait_time = first_wait_time;
+                    double second_wait_time = 0;
+                    double min_delta_time = first_delta_time;
+                    int min_size = first_size;
+                    while(futures.size() > this->chunked_transfer_queued / 2) {
+                        auto & it = futures.front();
+                        double this_wait_time = do_wait(std::get<0>(it));
+                        if(this_wait_time > max_wait_time) {
+                            second_wait_time = max_wait_time;
+                            max_wait_time = this_wait_time;
+                        } else if (this_wait_time > second_wait_time) {
+                            second_wait_time = this_wait_time;
+                        }
+
+                        double this_delta_time = (now() - std::get<1>(it)).value;
+                        min_delta_time = std::min(this_delta_time, min_delta_time);
+
+                        int this_size = std::get<2>(it);
+                        min_size = std::min(min_size, this_size);
+                        futures.pop_front();
+
+                        if(this_delta_time < chunk_timeout/ chunked_safety_factor) {
+                            // Because we are so much smaller, we can grow a lot in here.
+                            int new_chunk_size = (int)(min_size * chunked_safety_factor/2);
+                            if(new_chunk_size > chunk_size) {
+                                chunk_size = new_chunk_size;
+                                if(zwitch.developer_output())
+                                    std::cerr << "Increased chunk size (" << chunk_size << ") as the time to return (" << min_delta_time << ") is very small from a set of packets with min size (" << min_size << ")" << std::endl;
+                            }
+                        }
+                    }
+
+                    if(min_delta_time < chunk_timeout / chunked_safety_factor && second_wait_time < max_wait_time/chunked_safety_factor) {
+                        // Although this is not much, it grows fast.
+                        int new_chunk_size = (int)(min_size * 1.5);
+                        if(new_chunk_size > chunk_size) {
+                            chunk_size = new_chunk_size;
+                            if(zwitch.developer_output())
+                                std::cerr << "Increasing chunk size as the wait times are very noisy" << std::endl;
+                        }
+                    } else {
+                        // std::cerr << "Leaving chunk size as is because the ratio is " << second_wait_time / max_wait_time << std::endl;
+                    }
+                }
+            }
+
+            cur_start += this_chunk_size;
+        }
+    }
+
+    // Wait on remaining futures
+    for(auto & it : futures) {
+        do_wait(std::get<0>(it));
+    }
+
+    // Note: the do_wait sets last_activity on the main task, so there's
+    // nothing special to do at the end.
+
+    // Update estimated transfer speed
+    auto transfer_duration = now() - start_time;
+    estimated_transfer_speed_accum_bytes += transfer_total_size;
+    estimated_transfer_speed_accum_time += transfer_duration.value;
+    double ratio = estimated_transfer_speed_accum_time / limit_estimated_transfer_speed_accum_time;
+    if(ratio > 1) {
+        estimated_transfer_speed_accum_time /= ratio;
+        estimated_transfer_speed_accum_bytes /= ratio;
+    }
+
+    if(chunked_transfer_auto_adjust) {
+        double speed = estimated_transfer_speed_accum_bytes / estimated_transfer_speed_accum_time;
+        double goal_time = chunk_timeout / 5;
+        int est_chunk_size = (int)(goal_time * speed);
+        // if(est_chunk_size > chunk_size)
+        if(zwitch.developer_output()) {
+            std::cerr << "est_chunk_size: " << est_chunk_size << std::endl;
+            std::cerr << "est_speed: " << speed << std::endl;
+            std::cerr << "previous chunk_size: " << chunked_transfer_size << std::endl;
+        }
+        if(est_chunk_size < chunked_transfer_size)
+            chunked_transfer_size = est_chunk_size;
+        else
+            chunked_transfer_size = (int)((chunked_transfer_size + est_chunk_size) / chunked_safety_factor / chunked_transfer_queued);
+        if(zwitch.developer_output())
+            std::cerr << "Adjusting chunk size to " << chunked_transfer_size << std::endl;
+    }
+}
+
+
+GenericResponse generic_from_json(json j) {
+    GenericResponse generic;
+    if(j.contains("success")) {
+        if(j["success"].is_boolean())
+            generic.success = j["success"].get<bool>();
+        else {
+            std::string temp = j["success"].get<std::string>();
+            if(temp == "0")
+                generic.success = false;
+            else if(temp == "1")
+                generic.success = true;
+            else
+                throw std::runtime_error("Unknown value for success in GenericZefHubResponse");
+        }
+        if(j.contains("reason"))
+            generic.reason = json_string_default(j, "reason");
+        else if(j.contains("response"))
+            generic.reason = json_string_default(j, "response");
+    } else {
+        generic.success = true;
+        generic.reason = "";
+    }
+    return generic;
+}
diff --git a/core/src/butler/butler_update_payloads.cpp b/core/src/butler/butler_update_payloads.cpp
new file mode 100644
index 00000000..515cd182
--- /dev/null
+++ b/core/src/butler/butler_update_payloads.cpp
@@ -0,0 +1,410 @@
+// Copyright 2022 Synchronous Technologies Pte Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+
+////////////////////////////////////////////////////////
+// * Updates to upstream
+
+bool is_up_to_date(const UpdateHeads & update_heads) {
+    // Note we use < instead of != here to ease checks on zefhub where clients can be ahead.
+    if(update_heads.blobs.from < update_heads.blobs.to)
+        return false;
+
+    for(auto & cache : update_heads.caches) {
+        if(cache.from < cache.to)
+            return false;
+    }
+
+    return true;
+}
+
+UpdatePayload create_update_payload(const GraphData & gd, const UpdateHeads & update_heads) {
+    if(update_heads.blobs.from > update_heads.blobs.to)
+        throw std::runtime_error("Somehow upstream is ahead of us and we're primary!");
+
+    blob_index last_tx = gd.latest_complete_tx;
+
+    // This here is a dodgy internal hack to grab the whole graph so we can send the blobs/uids out:
+    char * blobs_ptr = (char*)(&gd) + update_heads.blobs.from * constants::blob_indx_step_in_bytes;
+    size_t len = (update_heads.blobs.to - update_heads.blobs.from) * constants::blob_indx_step_in_bytes;
+    ensure_or_get_range(blobs_ptr, len);
+    // TODO: Turn these back into string_views once everything is safe
+    std::string blobs(blobs_ptr, len);
+
+    UpdatePayload p;
+    p.j = json{
+        {"blob_index_lo", update_heads.blobs.from},
+        {"blob_index_hi", update_heads.blobs.to},
+        {"graph_uid", str(internals::get_graph_uid(gd))},
+        {"index_of_latest_complete_tx_node", last_tx},
+        {"hash_full_graph", gd.hash(constants::ROOT_NODE_blob_index, update_heads.blobs.to)},
+        {"data_layout_version", internals::get_data_layout_version_info(gd)}
+    };
+    p.rest = {blobs};
+    std::vector<json> caches;
+
+    for(auto & cache : update_heads.caches) {
+#define GEN_CACHE(x, y) else if(cache.name == x) {                      \
+            if(cache.from != cache.to) {                                \
+                caches.push_back({                                      \
+                        {"name", x},                                    \
+                        {"index_lo", cache.from},                       \
+                        {"index_hi", cache.to},                         \
+                        {"revision", cache.revision},                   \
+                    });                                                 \
+                auto ptr = gd.y->get();                                 \
+                p.rest.push_back(ptr->create_diff(cache.from, cache.to)); \
+            }                                                           \
+        }
+
+        if(false) {}
+        GEN_CACHE("_ETs_used", ETs_used)
+            GEN_CACHE("_RTs_used", RTs_used)
+            GEN_CACHE("_ENs_used", ENs_used)
+            GEN_CACHE("_uid_lookup", uid_lookup)
+            GEN_CACHE("_euid_lookup", euid_lookup)
+            GEN_CACHE("_tag_lookup", tag_lookup)
+        else {
+            throw std::runtime_error("Unknown cache");
+        }
+#undef GEN_CACHE
+
+    }
+
+    p.j["caches"] = caches;
+
+    return p;
+}
+
+json create_heads_json_from_sync_head(const GraphData & gd, const UpdateHeads & update_heads) {
+    json j{
+        {"blobs_head", update_heads.blobs.from},
+    };
+    json caches;
+
+    for(auto & cache : update_heads.caches) {
+#define GEN_CACHE(x, y) else if(cache.name == x) {  \
+            caches[x] = {                           \
+                {"name", x},                        \
+                {"head", cache.from},               \
+                {"revision", cache.revision},       \
+            };                                      \
+        }                                           \
+
+        if(false) {}
+        GEN_CACHE("_ETs_used", ETs_used)
+            GEN_CACHE("_RTs_used", RTs_used)
+            GEN_CACHE("_ENs_used", ENs_used)
+            GEN_CACHE("_uid_lookup", uid_lookup)
+            GEN_CACHE("_euid_lookup", euid_lookup)
+            GEN_CACHE("_tag_lookup", tag_lookup)
+        else {
+            throw std::runtime_error("Unknown cache");
+        }
+#undef GEN_CACHE
+
+    }
+
+    j["cache_heads"] = caches;
+
+    return j;
+}
+
+UpdateHeads client_create_update_heads(const GraphData & gd) {
+    if(gd.open_tx_thread != std::this_thread::get_id())
+        throw std::runtime_error("Need write lock to carefully read update heads.");
+
+    blob_index update_from = gd.sync_head;
+    if(update_from == 0)
+        update_from = constants::ROOT_NODE_blob_index;
+    UpdateHeads update_heads{ {update_from, gd.read_head} };
+
+#define GEN_CACHE(x, y) {                                               \
+        auto ptr = gd.y->get();                                         \
+        update_heads.caches.push_back({x, ptr->upstream_size(), ptr->size(), ptr->revision()}); \
+    }
+    GEN_CACHE("_ETs_used", ETs_used)
+        GEN_CACHE("_RTs_used", RTs_used)
+        GEN_CACHE("_ENs_used", ENs_used)
+        GEN_CACHE("_uid_lookup", uid_lookup)
+        GEN_CACHE("_euid_lookup", euid_lookup)
+        GEN_CACHE("_tag_lookup", tag_lookup)
+#undef GEN_CACHE
+        return update_heads;
+}
+
+void parse_filegraph_update_heads(MMap::FileGraph & fg, json & j) {
+    j["blobs_head"] = fg.get_latest_blob_index();
+
+    json cache_heads;
+    auto prefix = fg.get_prefix();
+    {
+        decltype(GraphData::ETs_used)::element_type mapping(fg, prefix->tokens_ET);
+        cache_heads["_ETs_used"] = json{
+            {"head", mapping.get()->size()},
+            {"revision", mapping.get()->revision()},
+        };
+    }
+    {
+        decltype(GraphData::RTs_used)::element_type mapping(fg, prefix->tokens_RT);
+        cache_heads["_RTs_used"] = json{
+            {"head", mapping.get()->size()},
+            {"revision", mapping.get()->revision()},
+        };
+    }
+    {
+        decltype(GraphData::ENs_used)::element_type mapping(fg, prefix->tokens_EN);
+        cache_heads["_ENs_used"] = json{
+            {"head", mapping.get()->size()},
+            {"revision", mapping.get()->revision()},
+        };
+    }
+    {
+        decltype(GraphData::uid_lookup)::element_type mapping(fg, prefix->uid_lookup);
+        cache_heads["_uid_lookup"] = json{
+            {"head", mapping.get()->size()},
+            {"revision", mapping.get()->revision()},
+        };
+    }
+    {
+        decltype(GraphData::euid_lookup)::element_type mapping(fg, prefix->euid_lookup);
+        cache_heads["_euid_lookup"] = json{
+            {"head", mapping.get()->size()},
+            {"revision", mapping.get()->revision()},
+        };
+    }
+    {
+        decltype(GraphData::tag_lookup)::element_type mapping(fg, prefix->tag_lookup);
+        cache_heads["_tag_lookup"] = json{
+            {"head", mapping.get()->size()},
+            {"revision", mapping.get()->revision()},
+        };
+    }
+    j["cache_heads"] = cache_heads;
+}
+
+void apply_sync_heads(GraphData & gd, const UpdateHeads & update_heads) {
+    update(gd.heads_locker, [&]() {
+        gd.sync_head = update_heads.blobs.to;
+        gd.currently_subscribed = true;
+
+        for(auto & cache : update_heads.caches) {
+#define GEN_CACHE(x, y) else if(cache.name == x) {  \
+                auto ptr = gd.y->get_writer();      \
+                ptr->upstream_size() = cache.to;    \
+                ptr->revision() = cache.revision;   \
+            }
+
+            if(false) {}
+            GEN_CACHE("_ETs_used", ETs_used)
+                GEN_CACHE("_RTs_used", RTs_used)
+                GEN_CACHE("_ENs_used", ENs_used)
+                GEN_CACHE("_uid_lookup", uid_lookup)
+                GEN_CACHE("_euid_lookup", euid_lookup)
+                GEN_CACHE("_tag_lookup", tag_lookup)
+            else
+                throw std::runtime_error("Don't understand cache: " + cache.name);
+#undef GEN_CACHE
+        }
+    });
+}
+
+UpdateHeads parse_message_update_heads(const json & j) {
+    UpdateHeads update_heads{
+        {0, j["blobs_head"].get<blob_index>()}
+    };
+
+    for(auto & it : j["cache_heads"].items()) {
+        std::string name = it.key();
+        update_heads.caches.push_back({
+                name,
+                0,
+                it.value()["head"].get<size_t>(),
+                it.value()["revision"].get<size_t>()
+            });
+    }
+
+    return update_heads;
+}
+
+UpdateHeads parse_payload_update_heads(const UpdatePayload & payload) {
+    UpdateHeads heads{
+        {payload.j["blob_index_lo"].get<blob_index>(), payload.j["blob_index_hi"].get<blob_index>()}
+    };
+
+    for(auto & cache : payload.j["caches"]) {
+        std::string name = cache["name"].get<std::string>();
+        size_t from = cache["index_lo"].get<size_t>();
+        size_t to = cache["index_hi"].get<size_t>();
+
+        heads.caches.push_back({name, from, to});
+    }
+
+    return heads;
+}
+
+bool heads_apply(const UpdateHeads & heads, const GraphData & gd) {
+    if(gd.write_head != heads.blobs.from) {
+        std::cerr << "Blobs update (" + to_str(heads.blobs.from) + ") doesn't fit onto write head (" + to_str(gd.write_head.load()) + ")" << std::endl;
+        return false;
+    }
+
+    // Check cache heads align
+    for(auto & cache : heads.caches) {
+#define GEN_CACHE(x, y) else if(cache.name == x) {                      \
+            auto ptr = gd.y->get();                                     \
+            if(ptr->revision() != cache.revision || ptr->size() != cache.from) { \
+                std::cerr << "Update for cache '" + cache.name + "' doesn't fit. Current: (" << ptr->revision() << ", " << ptr->size() << "), update: (" << cache.revision << ", " << cache.from << ")" << std::endl; \
+                return false;                                           \
+            }                                                           \
+        }
+        if(false) {}
+        GEN_CACHE("_ETs_used", ETs_used)
+            GEN_CACHE("_RTs_used", RTs_used)
+            GEN_CACHE("_ENs_used", ENs_used)
+            GEN_CACHE("_uid_lookup", uid_lookup)
+            GEN_CACHE("_euid_lookup", euid_lookup)
+            GEN_CACHE("_tag_lookup", tag_lookup)
+        else
+            throw std::runtime_error("Unknown cache");
+    }
+
+    return true;
+}
+
+
+void Butler::send_update(Butler::GraphTrackingData & me) {
+    if(me.gd->error_state != GraphData::ErrorState::OK)
+        throw std::runtime_error("Can't send_update with a graph in an invalid state");
+    if(!me.gd->is_primary_instance)
+        throw std::runtime_error("We don't have primary role, how did we send an update??");
+
+    // If we aren't connected, then we'll worry about sending out updates later.
+    if(!network.connected) {
+        // Make sure we know we aren't subscribed for later.
+        me.gd->currently_subscribed = false;
+        return;
+    }
+
+    // If we have yet to resubscribe to a graph, then wait for that
+    // before trying to send out an update. The wait can happen in the caller (most likely the sync thread)
+    if(me.gd->sync_head > 0 && !me.gd->currently_subscribed)
+        return;
+
+    UpdateHeads update_heads;
+    {
+        // Let's stop transactions from happening while we do this.
+        LockGraphData lock{me.gd};
+
+        update_heads = client_create_update_heads(*me.gd);
+    }
+
+    if(is_up_to_date(update_heads))
+        return;
+
+    UpdatePayload payload = create_update_payload(*me.gd, update_heads);
+
+    if(me.gd->sync_head == 0)
+        payload.j["msg_type"] = "full_graph";
+    else
+        payload.j["msg_type"] = "graph_update";
+
+    payload.j["msg_version"] = 1;
+
+    // Handling errors:
+    // ================
+    // Regardless of what happened, we need to queue a new shot at this.
+    // A "queue" implies a timeout though and that's ugly. Instead, we have a choice of options:
+    // a) try again immediately, this could lead to spam, maybe have a maximum try with a sleep in between (this blocks anything else happening with the graph though).
+    // b) check if the websocket went down. If so, wait until the trigger message for when that comes back up.
+    // c) wait for the next transaction. We could lose data that way though.
+    // A mix between a) and b) seems good.
+
+    const int max_attempts = 3;
+    int attempts_left = max_attempts;
+    for(; attempts_left > 0; attempts_left--) {
+        // TODO: Check if websocket is down, and quit in that case.
+        if(!network.connected) {
+            std::cerr << "Gave up updating as websocket is disconnected" << std::endl;
+            return;
+        }
+
+        // A little pause for each retry except the first.
+        if(attempts_left != max_attempts)
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+                    
+
+        GenericZefHubResponse response;
+        try {
+            if(zwitch.developer_output())
+                std::cerr << "Trying to send update for graph " << me.uid << " of range " << update_heads.blobs.from << " to " << update_heads.blobs.to << std::endl;
+            response = wait_on_zefhub_message(payload.j, payload.rest, zefhub_generic_timeout, true, true);
+
+            if(!response.generic.success) {
+                // There's generally only one reason for a failure. But
+                // in any case return and leave this to try again in the
+                // sync loop. It's better to just keep retrying rather
+                // than crashing.
+
+                if(response.j.contains("upstream_head")) {
+                    UpdateHeads parsed_heads = parse_message_update_heads(response.j);
+                    apply_sync_heads(*me.gd, parsed_heads);
+                } else {
+                    // Unknown error - wait a bit
+                    std::cerr << "Unknown error received from ZH from our graph update ('" << response.generic.reason << "'). Setting graph to invalid state." << std::endl;
+                    me.gd->error_state = GraphData::ErrorState::UNSPECIFIED_ERROR;
+                    // std::this_thread::sleep_for(std::chrono::seconds(1));
+                }
+
+                return;
+            }
+        } catch(const timeout_exception & e) {
+            std::cerr << "Got timeout during send_update. Going to retry" << std::endl;
+            continue;
+        } catch(const Communication::disconnected_exception & e) {
+            std::cerr << "Disconnected during send_update. Going to retry" << std::endl;
+            continue;
+        } catch(const std::exception & e) {
+            std::cerr << "An exception occurred during a send_update ('" << e.what() << "'). Setting graph to invalid state." << std::endl;
+            me.gd->error_state = GraphData::ErrorState::UNSPECIFIED_ERROR;
+            wake(me.gd->heads_locker);
+            return;
+        }
+
+        // if(response.j["blob_index_hi"].get<blob_index>() != me.gd->read_head)
+        //     throw std::runtime_error("Somehow zefhub replied with a different blob update than we sent!");
+
+        // Getting here should mean we succeeded.
+        me.gd->currently_subscribed = true;
+        apply_sync_heads(*me.gd, update_heads);
+        return;
+    }
+
+    // If we get here, we ran out of attempts
+    if(attempts_left != 0)
+        throw std::runtime_error("Huh?");
+
+    std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
+    std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
+    std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
+    std::cerr << "Failed to send out graph update! Graph " << me.uid << " is likely out of sync." << std::endl;
+    // We don't want to set this after all as it is better to enter a
+    // never-ending cycle of resend attempts, than potentially lose
+    // data.
+
+    // me.gd->error_state = GraphData::ErrorState::UNSPECIFIED_ERROR;
+    // throw std::runtime_error("Gave up after " + to_str(max_attempts) + " attempts of sending out graph update!");
+    // It's always better to just return and let this try again rather than crashing.
+    return;
+}
diff --git a/core/src/butler/butler_zefhub.cpp b/core/src/butler/butler_zefhub.cpp
deleted file mode 100644
index 76046270..00000000
--- a/core/src/butler/butler_zefhub.cpp
+++ /dev/null
@@ -1,48 +0,0 @@
-// Copyright 2022 Synchronous Technologies Pte Ltd
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#include "butler/butler_zefhub.h"
-
-namespace zefDB {
-    namespace Butler {
-        
-        void add_entity_type(const token_value_t & indx, const std::string & name) {
-            if (!butler_is_master)
-                throw std::runtime_error("Shouldn't be calling this function in normal execution.");
-
-            update_bidirection_name_map(global_token_store().ETs, indx, name);
-        }
-
-        void add_relation_type(const token_value_t & indx, const std::string & name) {
-            if (!butler_is_master)
-                throw std::runtime_error("Shouldn't be calling this function in normal execution.");
-
-            update_bidirection_name_map(global_token_store().RTs, indx, name);
-        }
-
-        void add_enum_type(const enum_indx & indx, const std::string& name) {
-            if (!butler_is_master)
-                throw std::runtime_error("Shouldn't be calling this function in normal execution.");
-
-            update_zef_enum_bidirectional_map(global_token_store().ENs, indx, name);
-        }
-
-        void add_keyword(const token_value_t & indx, const std::string & name) {
-            if (!butler_is_master)
-                throw std::runtime_error("Shouldn't be calling this function in normal execution.");
-
-            update_bidirection_name_map(global_token_store().KWs, indx, name);
-        }
-    }
-}
diff --git a/core/src/butler/threadsafe_map.cpp b/core/src/butler/threadsafe_map.cpp
index c69f2af7..09611583 100644
--- a/core/src/butler/threadsafe_map.cpp
+++ b/core/src/butler/threadsafe_map.cpp
@@ -82,6 +82,10 @@ namespace zefDB {
         bool zef_enum_bidirectional_map::contains(const string_pair& name_pair) {
             return zefDB::contains(enum_name_to_indx, name_pair.first + "." + name_pair.second);
         }
+
+        size_t zef_enum_bidirectional_map::size() {
+            return indx_to_string_pair.size();
+        }
     
 
         std::vector<std::tuple<enum_indx, std::string, std::string>> zef_enum_bidirectional_map::all_entries_as_list() {
diff --git a/core/src/globals.cpp b/core/src/globals.cpp
new file mode 100644
index 00000000..14aec41b
--- /dev/null
+++ b/core/src/globals.cpp
@@ -0,0 +1,30 @@
+// Copyright 2022 Synchronous Technologies Pte Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// This file exists to order the initialisation of any globals that need
+// ordering.
+#include "zwitch.h"
+#include "scalars.h"
+
+namespace zefDB {
+    Zwitch zwitch;
+
+    const QuantityFloat seconds{ 1, EN.Unit.seconds };
+    const QuantityFloat minutes{ 60, EN.Unit.seconds };
+    const QuantityFloat hours{ 60 * 60, EN.Unit.seconds };
+    const QuantityFloat days{ 60 * 60 * 24, EN.Unit.seconds };
+    const QuantityFloat weeks{ 60 * 60 * 24 * 7, EN.Unit.seconds };
+    const QuantityFloat months{ 60 * 60 * 24 * 30, EN.Unit.seconds };
+    const QuantityFloat years{ 60 * 60 * 24 * 365, EN.Unit.seconds };
+}
diff --git a/core/src/graph.cpp b/core/src/graph.cpp
index 6c8e6442..2cee7507 100644
--- a/core/src/graph.cpp
+++ b/core/src/graph.cpp
@@ -232,6 +232,9 @@ namespace zefDB {
             auto to_del_edge_uzr = internals::instantiate(root_uzr, BT.TO_DELEGATE_EDGE, delegate_tx_uzr, *this);
             auto del_inst_uzr = internals::instantiate(root_uzr, BT.DELEGATE_INSTANTIATION_EDGE, to_del_edge_uzr, *this);
 
+            // Every graph starts with an empty transaction
+            auto my_tx = Transaction(*this, false, false);
+
             auto & info = MMap::info_from_blobs(this);
             MMap::flush_mmap(info, write_head);
         } else if(preexisting_fg) {
@@ -241,10 +244,10 @@ namespace zefDB {
                 throw std::runtime_error("File graph doesn't have the same uid as passed in!");
             }
             
-            latest_complete_tx = index(internals::get_latest_complete_tx_node(*this, 0));
-
             write_head = fg->get_latest_blob_index();
             read_head = fg->get_latest_blob_index();
+            latest_complete_tx = index(internals::get_latest_complete_tx_node(*this, 0));
+
         } else {
             // Nothing to do here - just waiting for caller to fill in the graph
         }
@@ -299,16 +302,9 @@ namespace zefDB {
         if(!response.generic.success)
             throw std::runtime_error("Unable to create new graph: " + response.generic.reason);
         *this = std::move(*response.g);
-        if(internal_use_only) {
-            std::cerr << "Creating graph for internal use only - this shouldn't be happening anymore!" << std::endl;
-            std::cerr << "Creating graph for internal use only - this shouldn't be happening anymore!" << std::endl;
-            std::cerr << "Creating graph for internal use only - this shouldn't be happening anymore!" << std::endl;
-            std::cerr << "Creating graph for internal use only - this shouldn't be happening anymore!" << std::endl;
-            return;
-        }
         // std::cerr << "New graph Graph(), ref_count = " << my_graph_data().reference_count.load() << std::endl;
         
-        // After here, we should always destruct
+        // After here, we should always destruct if anything fails
         try {
             if(sync)
                 this->sync();
diff --git a/core/src/tokens.cpp b/core/src/tokens.cpp
index 0a44d8d3..7db50fda 100644
--- a/core/src/tokens.cpp
+++ b/core/src/tokens.cpp
@@ -17,6 +17,7 @@
 #include "zwitch.h"
 #include "butler/butler.h"
 #include "low_level_api.h"
+#include "zef_config.h"
 
 
 namespace zefDB {
@@ -255,6 +256,8 @@ namespace zefDB {
 
     TokenStore::TokenStore() {
         init_defaults();
+
+        load_cached_tokens();
     }
     TokenStore::TokenStore(MMap::FileGraph & fg) {
         init_defaults();
@@ -460,6 +463,125 @@ namespace zefDB {
     }
 
 
+    std::filesystem::path get_cached_tokens_path() {
+        std::string path = std::get<std::string>(get_config_var("tokens.cachePath"));
+        if(starts_with(path, "$CONFIG/"))
+            return zefdb_config_path() / path.substr(strlen("$CONFIG/"));
+        return path;
+    }
+
+    void TokenStore::save_cached_tokens() {
+        try {
+            std::filesystem::path cache_path = get_cached_tokens_path();
+            if(cache_path == "")
+                return;
+            auto lock_path = cache_path;
+            lock_path += ".lock";
+            FileLock file_lock(lock_path);
+
+            json j{
+                {"ETs", ETs.all_entries_as_list()},
+                {"RTs", RTs.all_entries_as_list()},
+                {"ENs", ENs.all_entries_as_list()},
+                {"KWs", KWs.all_entries_as_list()},
+            };
+
+            if(zwitch.developer_output())
+                std::cerr << "Going to save tokens to: " << cache_path << std::endl;
+            std::ofstream file(cache_path);
+            file << j;
+            if(!file.good())
+                throw std::runtime_error("Error in writing file.");
+        } catch(const std::exception & e) {
+            std::cerr << "Error while trying to save cached tokens: " << e.what() << std::endl;
+        }
+    }
+
+    void TokenStore::load_cached_tokens() {
+        try {
+            std::filesystem::path cache_path = get_cached_tokens_path();
+            if(cache_path == "")
+                return;
+            auto lock_path = cache_path;
+            lock_path += ".lock";
+            FileLock file_lock(lock_path);
+
+            if(zwitch.developer_output())
+                std::cerr << "Going to attempt loading cached tokens from: " << cache_path << std::endl;
+            
+            if(!std::filesystem::exists(cache_path))
+                return;
+
+            std::ifstream file(cache_path);
+            json j;
+            file >> j;
+
+            auto j_ETs = j["ETs"].get<std::vector<std::pair<token_value_t,std::string>>>();
+            auto j_RTs = j["RTs"].get<std::vector<std::pair<token_value_t,std::string>>>();
+            auto j_KWs = j["KWs"].get<std::vector<std::pair<token_value_t,std::string>>>();
+            auto j_ENs = j["ENs"].get<std::vector<std::tuple<token_value_t,std::string,std::string>>>();
+
+            for(auto & it : j_ETs)
+                force_add_entity_type(it.first, it.second);
+            for(auto & it : j_RTs)
+                force_add_relation_type(it.first, it.second);
+            for(auto & it : j_KWs)
+                force_add_keyword(it.first, it.second);
+            for(auto & it : j_ENs)
+                force_add_enum_type(std::get<0>(it), std::get<1>(it) + "." + std::get<2>(it));
+        } catch(const std::exception & e) {
+            std::cerr << "Error while trying to load cached tokens: " << e.what() << std::endl;
+        }
+    }
+
+    // TODO: Replace these with the individual type updates, which will occur in new messages from zefhub.
+    template<typename T1, typename T2>
+    void update_bidirection_name_map(thread_safe_bidirectional_map<T1,T2>& map_to_update, const T1 & indx, const T2 & name) {			
+        auto prior_name = map_to_update.maybe_at(indx);
+        auto prior_indx = map_to_update.maybe_at(name);
+        if(prior_name && *prior_name != name)
+            throw std::runtime_error("Force adding token disagrees with existing token: " + *prior_name);
+        else if(prior_indx && *prior_indx != indx)
+            throw std::runtime_error("Force adding token disagrees with existing token: " + to_str(*prior_indx));
+        else
+            map_to_update.insert(indx, name);
+    }
+
+
+    void update_zef_enum_bidirectional_map(thread_safe_zef_enum_bidirectional_map& map_to_update, const enum_indx & indx, const std::string& name) {
+        auto [enum_type,enum_value] = internals::split_enum_string(name);
+        auto pair = std::make_pair(enum_type, enum_value);
+        auto prior_pair = map_to_update.maybe_at(indx);
+        auto prior_indx = map_to_update.maybe_at(pair);
+        if(prior_pair && *prior_pair != pair)
+            throw std::runtime_error("Force adding EN disagrees with existing token: " + std::get<0>(*prior_pair) + "." + std::get<1>(*prior_pair));
+        else if(prior_indx && *prior_indx != indx)
+            throw std::runtime_error("Force adding EN disagrees with existing token: " + to_str(*prior_indx));
+        else
+            map_to_update.insert(indx, pair);
+    }
+
+
+
+
+    void TokenStore::force_add_entity_type(const token_value_t & indx, const std::string & name) {
+        update_bidirection_name_map(ETs, indx, name);
+    }
+
+    void TokenStore::force_add_relation_type(const token_value_t & indx, const std::string & name) {
+        update_bidirection_name_map(RTs, indx, name);
+    }
+
+    void TokenStore::force_add_keyword(const token_value_t & indx, const std::string & name) {
+        update_bidirection_name_map(KWs, indx, name);
+    }
+
+    void TokenStore::force_add_enum_type(const enum_indx & indx, const std::string& name) {
+        update_zef_enum_bidirectional_map(ENs, indx, name);
+    }
+
+
+
     //////////////////////////////////////////////////////
     // * Query upstream functions
 
diff --git a/core/src/zef_config.cpp b/core/src/zef_config.cpp
index dd2026f8..b5455235 100644
--- a/core/src/zef_config.cpp
+++ b/core/src/zef_config.cpp
@@ -19,24 +19,51 @@
 
 namespace zefDB {
 
-    const ConfigItem config_spec[] = {
-        {"login.autoConnect", "auto", {"no", "auto", "always"}, "ZEFDB_LOGIN_AUTOCONNECT",
-         "Whether zefdb should automatically connect to ZefHub on start of the butler."},
-        {"butler.autoStart", true, {}, "ZEFDB_BUTLER_AUTOSTART",
-         "Whether the butler will automatically be started on import of the zefdb module."},
-        {"login.zefhubURL", "wss://hub.zefhub.io", {}, "ZEFHUB_URL",
-         "Which URL to connect to ZefHub."},
-    };
-    const int num_config_spec = sizeof(config_spec) / sizeof(ConfigItem);
-
-    // Going to make this code trivially threadsafe by locking before each function call.
-    std::recursive_mutex config_mutex;
+    // We force config_spec to be loaded from within a function, rather than a
+    // global, as this allows others globals to make use of it when they
+    // themselves are being initialised.
+    //
+    // This used to be a const array but keeping it as a vector is fine.
+    const std::vector<ConfigItem>& all_config_spec() {
+        static const std::vector<ConfigItem> config_spec{
+            {"login.autoConnect", "auto", {"no", "auto", "always"}, "ZEFDB_LOGIN_AUTOCONNECT",
+            "Whether zefdb should automatically connect to ZefHub on start of the butler."},
+            {"butler.autoStart", true, {}, "ZEFDB_BUTLER_AUTOSTART",
+            "Whether the butler will automatically be started on import of the zefdb module."},
+            {"login.zefhubURL", "wss://hub.zefhub.io", {}, "ZEFHUB_URL",
+            "Which URL to connect to ZefHub."},
+            {"tokens.cachePath", "$CONFIG/tokens_cache.json", {}, "ZEFDB_TOKENS_CACHE_PATH",
+            "Where are the tokens cached between sessions"},
+            {"tokens.pullOnConnect", true, {}, "ZEFDB_TOKENS_PULL_ON_CONNECT",
+            "When connecting, get latest set for user."},
+        };
+        return config_spec;
+    }
 
-    
     std::unordered_map<std::string, config_var_t> session_overrides;
 
+    std::filesystem::path zefdb_config_path() {
+        char * env = std::getenv("ZEFDB_SESSION_PATH");
+        if (env != nullptr)
+            return std::string(env);
+
+#ifdef ZEF_WIN32
+        env = std::getenv("LOCALAPPDATA");
+#else
+        env = std::getenv("HOME");
+#endif
+        if (env == nullptr)
+            throw std::runtime_error("No HOME env set!");
+
+        std::filesystem::path path(env);
+        path /= ".zef";
+        if(!std::filesystem::exists(path))
+            std::filesystem::create_directories(path);
+        return path;
+    }
+
     std::filesystem::path config_file_path() {
-        return Butler::zefdb_config_path() / "config.yaml";
+        return zefdb_config_path() / "config.yaml";
     }
 
     bool config_file_exists() {
@@ -52,13 +79,9 @@ namespace zefDB {
     }
 
     const ConfigItem & get_spec(std::string key) {
-        // for(auto & item : config_spec) {
-        //     if(item.path == key)
-        //         return item;
-        // }
-        for(int i = 0; i < num_config_spec ; i++) {
-            if(config_spec[i].path == key)
-                return config_spec[i];
+        for(auto & item : all_config_spec()) {
+            if(item.path == key)
+                return item;
         }
         throw std::runtime_error("Don't recognise path '" + key + "' in config");
     }
@@ -79,7 +102,7 @@ namespace zefDB {
     }
 
     config_var_t get_config_var(std::string key) {
-        std::lock_guard lock(config_mutex);
+        FileLock lock(zefdb_config_path());
         auto search = session_overrides.find(key);
         if(search != session_overrides.cend()) {
             return search->second;
@@ -131,6 +154,7 @@ namespace zefDB {
     // TODO: Has to lookup every key for defaults/overrides and indicate this too in the return
     
     void set_config_var(std::string key, config_var_t val) {
+        FileLock lock(zefdb_config_path());
         ensure_config_file();
 
         auto & spec = get_spec(key);
@@ -185,11 +209,11 @@ namespace zefDB {
     std::vector<std::pair<std::string,config_var_t>> list_config(std::string filter) {
         std::vector<std::pair<std::string,config_var_t>> out;
 
-        for(int i = 0; i < num_config_spec ; i++) {
-            if(filter != "" && config_spec[i].path.find(filter) == std::string::npos)
+        for(auto & item : all_config_spec()) {
+            if(filter != "" && item.path.find(filter) == std::string::npos)
                 continue;
 
-            out.emplace_back(config_spec[i].path, get_config_var(config_spec[i].path));
+            out.emplace_back(item.path, get_config_var(item.path));
         }
 
         return out;
diff --git a/core/src/zefref.cpp b/core/src/zefref.cpp
index ad0da3ce..3f4ad33c 100644
--- a/core/src/zefref.cpp
+++ b/core/src/zefref.cpp
@@ -48,7 +48,7 @@ namespace zefDB {
     EZefRef::EZefRef(blob_index my_blob_index, const GraphData& graph_data_to_get_graph_from) : 
         blob_ptr(ptr_from_blob_index(my_blob_index, graph_data_to_get_graph_from))
 		{
-			if (my_blob_index < 0) { // TODO: add the following check, but gd not defined yet || my_blob_index > graph_data_to_get_graph_from.write_head)
+			if (my_blob_index < 0 || my_blob_index > graph_data_to_get_graph_from.write_head) {
 				std::cout << "EZefRef ctor called with index " << my_blob_index << std::endl;
                 print_backtrace();
 				throw std::runtime_error("EZefRef initialized with index outside of valid range for this graph");
diff --git a/core/src/zwitch.cpp b/core/src/zwitch.cpp
index 17a0a66b..3f0b496c 100644
--- a/core/src/zwitch.cpp
+++ b/core/src/zwitch.cpp
@@ -16,7 +16,6 @@
 #include "graph.h"
 
 namespace zefDB {
-    Zwitch zwitch{};
     std::chrono::steady_clock::time_point time_start_of_process = std::chrono::steady_clock::now();
 
     std::ostream& operator<< (std::ostream& o, Zwitch zw) {
diff --git a/python/pyproject.toml b/python/pyproject.toml
index a2c55be6..f4868065 100644
--- a/python/pyproject.toml
+++ b/python/pyproject.toml
@@ -1,7 +1,7 @@
 [tool.cibuildwheel]
 skip = ["pp*", "*musllinux*"]
 archs = ["auto64"]
-test-requires = "pytest networkx>=2.6.3"
+before-test = "python3 -mpip install -r {project}/python/tests/requirements_tests.txt"
 test-command = "bash {project}/scripts/assert_no_created_tokens.sh && ZEFHUB_AUTH_KEY=GUEST ZEFDB_QUIET=YES pytest {project}/python/tests && python {project}/python/tests/quick_start.py"
 
 [tool.cibuildwheel.linux]
diff --git a/python/pyzef/CMakeLists.txt b/python/pyzef/CMakeLists.txt
index dcc97f71..6617d950 100644
--- a/python/pyzef/CMakeLists.txt
+++ b/python/pyzef/CMakeLists.txt
@@ -107,41 +107,10 @@ if(CMAKE_BUILD_TYPE STREQUAL "Debug")
   endif()
 endif()
 
-
-# target_include_directories (pyzef PUBLIC ${RANGE_INCLUDE_DIRS})
-
-# if(UNIX AND NOT APPLE)
-#   # This should take us up the lib directory where both libzef and libpython are.
-#   set_target_properties(pyzef PROPERTIES INSTALL_RPATH "$ORIGIN")
-# elseif(APPLE)
-#   set_target_properties(pyzef PROPERTIES INSTALL_RPATH "@loader_path")
-# endif()
-# add_custom_command(TARGET pyzef
-#   POST_BUILD
-#   COMMAND ${CMAKE_COMMAND} -E create_symlink $<TARGET_FILE:pyzef> ${ZEFDB_DIR}/zef/`basename $<TARGET_FILE:pyzef>`
-#   COMMAND ${CMAKE_COMMAND} -E create_symlink $<TARGET_FILE:zef> ${ZEFDB_DIR}/zef/`basename $<TARGET_FILE:zef>`
-# )
-
 target_link_libraries(pyzef PUBLIC zef::zef)
-# target_include_directories(pyzef PRIVATE ${Zef_INCLUDE})
-
-# Hoping that hardcoding this will at least allow us to get somewhere.
-# target_link_libraries(pyzef PUBLIC openssl)
-# target_link_libraries(pyzef PUBLIC OpenSSL::SSL)
 
 target_link_libraries(pyzef PUBLIC pybind11_json)
 
-
-
-# Things needed for Visual Studio
-#if(MSVC)
-#	target_compile_options(pyzef PUBLIC "/permissive-")
-#    target_compile_options(pyzef PUBLIC "/bigobj")
-#    set_target_properties(pyzef PROPERTIES SUFFIX ".pyd")
-#    set_target_properties(pyzef PROPERTIES ENABLE_EXPORTS 1)
-#	set_target_properties(pyzef PROPERTIES WINDOWS_EXPORT_ALL_SYMBOLS 1)
-#endif()
-  
 # We are assuming this is an install into the python extension build directory.
 # Anything else should not be installing
 install(TARGETS pyzef LIBRARY DESTINATION .)
diff --git a/python/pyzef/internals.cpp b/python/pyzef/internals.cpp
index 61563b24..717f1624 100644
--- a/python/pyzef/internals.cpp
+++ b/python/pyzef/internals.cpp
@@ -715,12 +715,24 @@ void fill_internals_module(py::module_ & internals_submodule) {
 
     // TODO: In the future, these should be compiled into a version of pyzef
     // only available to zefhub/tokoloshes
-    internals_submodule.def("add_entity_type", &Butler::add_entity_type, "This should only be called by zefhub.");
-    internals_submodule.def("add_relation_type", &Butler::add_relation_type, "This should only be called by zefhub.");
-    internals_submodule.def("add_enum_type", &Butler::add_enum_type, "This should only be called by zefhub.");
-    internals_submodule.def("add_keyword", &Butler::add_keyword, "This should only be called by zefhub.");
-
-	internals_submodule.def("set_data_layout_version_info", &internals::set_data_layout_version_info, "new_string_value"_a, "g"_a);
+    internals_submodule.def("add_entity_type", [](const token_value_t indx, const std::string & name) {
+        auto & store = global_token_store();
+        store.force_add_entity_type(indx, name);
+    }, "This should only be called by zefhub.");
+    internals_submodule.def("add_relation_type", [](const token_value_t indx, const std::string & name) {
+        auto & store = global_token_store();
+        store.force_add_relation_type(indx, name);
+    }, "This should only be called by zefhub.");
+    internals_submodule.def("add_enum_type", [](const token_value_t indx, const std::string & name) {
+        auto & store = global_token_store();
+        store.force_add_enum_type(indx, name);
+    }, "This should only be called by zefhub.");
+    internals_submodule.def("add_keyword", [](const token_value_t indx, const std::string & name) {
+        auto & store = global_token_store();
+        store.force_add_keyword(indx, name);
+    }, "This should only be called by zefhub.");
+
+    internals_submodule.def("set_data_layout_version_info", &internals::set_data_layout_version_info, "new_string_value"_a, "g"_a);
 	internals_submodule.def("get_data_layout_version_info", &internals::get_data_layout_version_info, "g"_a);
 	internals_submodule.def("set_graph_revision_info", &internals::set_graph_revision_info, "new_string_value"_a, "g"_a);
 	internals_submodule.def("get_graph_revision_info", &internals::get_graph_revision_info, "g"_a);
diff --git a/python/zef/__init__.py b/python/zef/__init__.py
index 486498d4..91961502 100644
--- a/python/zef/__init__.py
+++ b/python/zef/__init__.py
@@ -29,11 +29,6 @@
 
 pyzef.internals.finished_loading_python_core()
 
-import os
-if os.environ.get("ZEFDB_DEVELOPER_CIRCULAR_IMPORTS", "FALSE") == "TRUE":
-    from .circular_imports import disable_check_circular_imports
-    disable_check_circular_imports()
-
 ############################################
 # * Starting the butler
 #------------------------------------------
diff --git a/python/zef/circular_imports.py b/python/zef/circular_imports.py
deleted file mode 100644
index 5d926222..00000000
--- a/python/zef/circular_imports.py
+++ /dev/null
@@ -1,126 +0,0 @@
-# Copyright 2022 Synchronous Technologies Pte Ltd
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# Heavily modified from https://stackoverflow.com/a/2406043
-
-import types
-# Getting really frustrated with circular imports
-beingimported = []
-originalimport = __import__
-def newimport(modulename, globs=None, locals=None, fromlist=(), level=0):
-    if isinstance(globs, dict):
-        parent_name = globs.get("__name__", None)
-        is_package = (parent_name == globs.get("__package__", "1"))
-    else:
-        parent_name = ""
-        is_package = False
-    if level == 0:
-        qual_name = modulename
-    else:
-        rel_level = level
-        if is_package:
-            rel_level -= 1
-        if rel_level == 0:
-            rel_name = parent_name
-        else:
-            rel_name = '.'.join(parent_name.split('.')[:-rel_level])
-        if modulename == "":
-            qual_name = rel_name
-        else:
-            qual_name = rel_name + '.' + modulename
-
-    # Trying method with all fromlist used instead
-    was_already_present = qual_name in beingimported
-    beingimported.append(qual_name)
-
-    sub_check = {}
-    if isinstance(fromlist, tuple):
-        for item in fromlist:
-            if item == "*":
-                continue
-            sub_qual_name = qual_name + "." + item
-            sub_check[item] = (sub_qual_name, sub_qual_name in beingimported)
-            beingimported.append(sub_qual_name)
-
-    try:
-        result = originalimport(modulename, globs, locals, fromlist, level)
-
-        try:
-            if "zef" in parent_name:
-                # only_importing_submodules = isinstance(fromlist, tuple) and ('*' not in fromlist) and all(isinstance(getattr(result, x), types.ModuleType) for x in fromlist)
-                # except Exception as exc:
-                #     print(f"Errored: {exc}")
-                #     print(dir(result))
-                #     only_importing_submodules = False
-                only_importing_submodules = (len(sub_check) > 0)
-                for sub_name, (sub_qual_name, sub_was_previous) in sub_check.items():
-                    try:
-                        sub_obj = getattr(result, sub_name)
-                    except:
-                        print(f"!!! After trying to import item {sub_qual_name=} it wasn't in the module!")
-                        raise
-                    if isinstance(sub_obj, types.ModuleType):
-                        if sub_was_previous:
-                            import traceback
-                            frame = traceback.extract_stack()[-2]
-                            print("=========")
-                            print(f"Importing in circles {sub_qual_name=} from import {qual_name=} {level=} {parent_name=} {is_package=} {fromlist=}")
-                            print("    Currently importing -> ", beingimported)
-                            print_all_imports_in_stack()
-                    else:
-                        only_importing_submodules = False
-                if was_already_present and not only_importing_submodules:
-                    # We don't worry about importing submodules from us when we're a package
-                    is_submodule_import = (is_package and level == 1)
-                    if not is_submodule_import:
-                        import traceback
-                        frame = traceback.extract_stack()[-2]
-                        print("=========")
-                        print(f"Importing in circles {modulename=} {qual_name=} {level=} {parent_name=} {is_package=} {fromlist=}")
-                        print("    Currently importing -> ", beingimported)
-                        print_all_imports_in_stack()
-        except Exception as exc:
-            print("Ignoring exc in detecting circular imports")
-    finally:
-        def remove_last_occurrence(l, x):
-            if x in l:
-                l.reverse()
-                l.remove(x)
-                l.reverse()
-        remove_last_occurrence(beingimported, qual_name)
-        for (sub_qual_name, sub_was_previous) in sub_check.values():
-            remove_last_occurrence(beingimported, sub_qual_name)
-
-    return result
-
-def check_circular_imports():
-    import builtins
-    builtins.__import__ = newimport
-
-def disable_check_circular_imports():
-    import builtins
-    builtins.__import__ = originalimport
-
-
-
-def print_all_imports_in_stack():
-    import traceback
-    stack = traceback.extract_stack()[:-2]
-    stack.reverse()
-
-    for frame in stack:
-        if "import" in frame.line and 'circular_imports' not in frame.filename:
-            print(f"   {frame.filename}: {frame.lineno} -- {frame.line}")
-            
-    
\ No newline at end of file
diff --git a/python/zef/core/VT/__init__.py b/python/zef/core/VT/__init__.py
index bc3ed84c..3d816f04 100644
--- a/python/zef/core/VT/__init__.py
+++ b/python/zef/core/VT/__init__.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from functools import partial
 from .value_type import *
 
 
@@ -42,6 +43,10 @@ def decimal_ctor(*args, **kwargs):
     return Decimal_(*args, **kwargs)
 
 
+def flatgraph_ctor(*args, **kwargs):
+    from ...core.flat_graph import FlatGraph_
+    return FlatGraph_(*args)
+
 
 def union_getitem(x):
     from ..op_structs import ZefOp
@@ -114,7 +119,13 @@ def rp_getitem(x):
     if not isinstance(x, tuple) or len(x)!=3:
         raise TypeError(f"`RP`[...]  must be initialized with a triple to match on. Got x={x}")
     return ValueType_(type_name='RP', absorbed=(x,))
-      
+
+
+def same_as_get_item(y):
+    from ..op_implementations.implementation_typing_functions import discard_frame_imp
+    return Is[lambda x: discard_frame_imp(x) == discard_frame_imp(y)]
+
+
 
 Nil        = ValueType_(type_name='Nil',        constructor_func=None)
 Any        = ValueType_(type_name='Any',        constructor_func=None)
@@ -131,7 +142,7 @@ def rp_getitem(x):
 ZefRef     = ValueType_(type_name='ZefRef',     constructor_func=None)     
 Graph      = ValueType_(type_name='Graph',      constructor_func=graph_ctor)     
 GraphSlice = ValueType_(type_name='GraphSlice', constructor_func=None)     
-FlatGraph  = ValueType_(type_name='FlatGraph',  constructor_func=None)     
+FlatGraph  = ValueType_(type_name='FlatGraph',  constructor_func=flatgraph_ctor)     
 ZefOp      = ValueType_(type_name='ZefOp',      constructor_func=None)     
 Stream     = ValueType_(type_name='Stream',     constructor_func=None)     
 TX         = ValueType_(type_name='TX',         constructor_func=None)     
@@ -168,11 +179,10 @@ def rp_getitem(x):
 RT         = ValueType_(type_name='RT',         constructor_func=None)
 BT         = ValueType_(type_name='BT',         constructor_func=None)
 Enum       = ValueType_(type_name='Enum',       constructor_func=None)
-Record     = ValueType_(type_name='Record',     constructor_func=None)       
+Tuple      = ValueType_(type_name='Tuple',     constructor_func=None)       
 Function   = ValueType_(type_name='Function',   constructor_func=None)
 GraphDelta = ValueType_(type_name='GraphDelta', constructor_func=None)     
 Query      = ValueType_(type_name='Query',      constructor_func=None) 
-DeltaQuery = ValueType_(type_name='DeltaQuery', constructor_func=None) 
 Effect     = ValueType_(type_name='Effect',     constructor_func=None) 
 DataFrame  = ValueType_(type_name='DataFrame',  constructor_func=None)     
 # EZefRefs   = ValueType_(type_name='EZefRefs', constructor_func=None)     
@@ -188,6 +198,10 @@ def rp_getitem(x):
 Complement     = ValueType_(type_name='Complement',          constructor_func=None,             get_item_func=complement_getitem)
 RP             = ValueType_(type_name='RP',                  constructor_func=None,             get_item_func=rp_getitem)
 HasValue       = ValueType_(type_name='HasValue',            constructor_func=None)
+SameAs         = ValueType_(type_name='SameAs',                                                 get_item_func = same_as_get_item)
+
+
+
 
 def operates_on_ctor(x):
     from .._ops import operates_on, contains
@@ -204,3 +218,22 @@ def used_for_ctor(x):
 OperatesOn     = ValueType_(type_name='OperatesOn',   constructor_func = operates_on_ctor,  get_item_func = operates_on_ctor)
 RelatedOps     = ValueType_(type_name='RelatedOps',   constructor_func = related_ops_ctor,  get_item_func = related_ops_ctor)
 UsedFor        = ValueType_(type_name='UsedFor',      constructor_func = used_for_ctor, get_item_func =used_for_ctor)
+
+
+#--------------------------ZefUI Types------------------
+def zef_ui_ctor(type_name, self, *args, **kwargs):
+    if len(args) != 0: 
+        raise ValueError(f'{type_name} constructor takes no positional arguments except self, got {args}')
+
+    ctor_dict = {"type_name":type_name, "constructor_func": partial(zef_ui_ctor, type_name), "pass_self": True}
+    if len(self.d['absorbed']) == 1:
+        return ValueType_( absorbed = ({**self.d['absorbed'][0], **kwargs},), **ctor_dict)
+
+    return ValueType_(absorbed = (kwargs,), **ctor_dict)
+
+Text  = ValueType_(type_name='Text', constructor_func = partial(zef_ui_ctor, 'Text'), pass_self = True)
+Code  = ValueType_(type_name='Code', constructor_func = partial(zef_ui_ctor, 'Code'), pass_self = True)
+Style = ValueType_(type_name='Style', constructor_func = partial(zef_ui_ctor, 'Style'), pass_self = True)
+Table = ValueType_(type_name='Table', constructor_func = partial(zef_ui_ctor, 'Table'), pass_self = True)
+Column = ValueType_(type_name='Column', constructor_func = partial(zef_ui_ctor, 'Column'), pass_self = True)
+Frame = ValueType_(type_name='Frame', constructor_func = partial(zef_ui_ctor, 'Frame'), pass_self = True)
\ No newline at end of file
diff --git a/python/zef/core/VT/value_type.py b/python/zef/core/VT/value_type.py
index d0f538ce..6a6fc5a2 100644
--- a/python/zef/core/VT/value_type.py
+++ b/python/zef/core/VT/value_type.py
@@ -78,7 +78,7 @@ class ValueType_:
     """ 
     Zef ValueTypes are Values themselves.
     """
-    def __init__(self, type_name: str, absorbed=(), constructor_func=None, get_item_func=None):
+    def __init__(self, type_name: str, absorbed=(), constructor_func=None, get_item_func=None, pass_self=False):
             self.d = {
                 'type_name': type_name,
                 'absorbed': absorbed,
@@ -88,8 +88,9 @@ def __init__(self, type_name: str, absorbed=(), constructor_func=None, get_item_
             if get_item_func is not None:
                 _value_type_get_item_funcs[type_name] = get_item_func
 
+            self.pass_self = pass_self
             self.allowed_types = (ValueType_, EntityTypeStruct, RelationTypeStruct, AtomicEntityTypeStruct, BlobTypeStruct, BlobType, AtomicEntityType, EntityType, RelationType)
-        
+
     def __repr__(self):
         return self.d['type_name'] + (
             '' if self.d['absorbed'] == () 
@@ -102,12 +103,14 @@ def __call__(self, *args, **kwargs):
             f = _value_type_constructor_funcs[self.d["type_name"]]
         except KeyError:
             return Error(f'{self.d["type_name"]}(...) was called, but no constructor function was registered for this type')
+        if self.pass_self: return f(self, *args, **kwargs)
         return f(*args, **kwargs)
 
 
     def __getitem__(self, x):
         try:
             f = _value_type_get_item_funcs[self.d["type_name"]]
+            if self.pass_self: return f(self, x)
             return f(x)
         except KeyError:
             return ValueType_(self.d["type_name"], absorbed=(*self.d['absorbed'], x))
@@ -153,6 +156,15 @@ def __rand__(self, other):
 
     def __invert__(self):
         return ValueType_(type_name='Complement', absorbed=(self,))
+
+
+    def __contains__(self, x):
+        """
+        Allows checking membership in form
+        >>> 4 in Int
+        """
+        from ..op_implementations.implementation_typing_functions import is_a_implementation
+        return is_a_implementation(x, self)
     
 
 def make_distinct(v):
diff --git a/python/zef/core/__init__.py b/python/zef/core/__init__.py
index bba34e76..98f1c4c2 100644
--- a/python/zef/core/__init__.py
+++ b/python/zef/core/__init__.py
@@ -83,7 +83,7 @@
 from .graph_slice import GraphSlice
 
 
-from .flat_graph import FlatGraph, FlatRef, FlatRefs, Val
+from .flat_graph import FlatGraph_, FlatRef, FlatRefs, Val
 
 # TODO: import the other ValueTypes here and implement constructor by forwarding args
 from .VT import (
@@ -104,6 +104,7 @@
     Instantiated, 
     Terminated, 
     Assigned,
+    FlatGraph,
 
     Union,
     Intersection,
@@ -113,6 +114,7 @@
     RP,
     HasValue,
     Pattern,
+    SameAs,
 
     RelatedOps,
     UsedFor,
diff --git a/python/zef/core/_ops.py b/python/zef/core/_ops.py
index 6cd20589..2cd81c13 100644
--- a/python/zef/core/_ops.py
+++ b/python/zef/core/_ops.py
@@ -363,7 +363,7 @@ def make_zefop(rt):
 
 # TODO: implement
 on              = make_zefop(RT.On)         
-
+gather          = make_zefop(RT.Gather)
 
           
 # match
diff --git a/python/zef/core/flat_graph.py b/python/zef/core/flat_graph.py
index e2c3af54..8b859c25 100644
--- a/python/zef/core/flat_graph.py
+++ b/python/zef/core/flat_graph.py
@@ -22,13 +22,13 @@
 class Val:
     arg: VT.Any
 
-class FlatGraph:
+class FlatGraph_:
     def __init__(self, *args):
         if args == ():
             self.key_dict = {}
             self.blobs = ()
         elif len(args) == 1 and isinstance(args[0], list):
-            new_fg = FlatGraph()
+            new_fg = FlatGraph_()
             new_fg = new_fg | insert[args[0]] | collect
             self.key_dict = new_fg.key_dict
             self.blobs = new_fg.blobs
diff --git a/python/zef/core/fx/fx_types.py b/python/zef/core/fx/fx_types.py
index 52ff67a6..32b11abe 100644
--- a/python/zef/core/fx/fx_types.py
+++ b/python/zef/core/fx/fx_types.py
@@ -43,10 +43,10 @@ def __call__(self, **kwargs):
         can be run, execute this before returning. If it fails, return an
         error.
         """
-        return Effect({
+        return {
             **kwargs,
             'type': self,
-        })
+        }
 
 
 
@@ -190,8 +190,7 @@ def Effect(*args, **kwargs) -> dict:
     or by passing in a dictionary.
     
     ---- Examples ----
-    Effect(type=FX.Subprocess.Start)
-    Effect({'type': FX.Subprocess.Start})
+    FX.Subprocess.Start()       # the call operator acts like a constructor. The type is inserted automatically.
     """
     if args!=():
         assert len(kwargs) == 0
diff --git a/python/zef/core/fx/graphql.py b/python/zef/core/fx/graphql.py
index 1a29929a..04d75190 100644
--- a/python/zef/core/fx/graphql.py
+++ b/python/zef/core/fx/graphql.py
@@ -69,7 +69,7 @@ def resolve_gql_query(req, port=port, playground_path=playground_path, gql_path=
 
 
     logging = eff.get("logging", False)
-    http_r = Effect({
+    http_r = {
         **eff,     # Verify that this is ok after change: the dict was mutated previously (using pop). There may be more key-value pairs in there now.
         'type': FX.HTTP.StartServer,
         'port': port,
@@ -78,7 +78,7 @@ def resolve_gql_query(req, port=port, playground_path=playground_path, gql_path=
         # 'pipe_into': middleware[[permit_cors, resolve_gql_query, send_response]]
         'pipe_into': map[middleware_worker[permit_cors, resolve_gql_query, fallback_not_found, send_response]] | subscribe[run],
         'logging': logging,
-    }) | run
+    } | run
 
     log.info(f"Started GQL server at http://localhost:{port}{gql_path}")
     if playground_path is not None:
@@ -97,14 +97,14 @@ def graphql_stop_server_handler(eff: dict):
     
 def graphql_start_playground_handler(eff: dict):
     """Example
-    Effect({
+    {
         "type": FX.GraphQL.StartPlayground,
         "schema_root": z_schema,
         "port": 5001, (default=5000)
         "path": "/gql", (default="/gql")
         "playground_path": "/", (default="/")
         "open_browser": True (default=True)
-    }) | run
+    } | run
     """   
     schema = eff['schema_root']
     port = eff.get("port", 5000)
diff --git a/python/zef/core/fx/http.py b/python/zef/core/fx/http.py
index 01f953f8..8124d071 100644
--- a/python/zef/core/fx/http.py
+++ b/python/zef/core/fx/http.py
@@ -15,6 +15,7 @@
 
 #=============================================================================== HTTP Server handlers =========================================================================================================
 from .fx_types import Effect, FX
+from ..VT import Pattern, SetOf
 from uuid import uuid4
 from .._ops import *
 from ..error import Error
@@ -79,8 +80,11 @@ def handle_generic(self, method):
                 # TODO: Cleanup from open_requests here
                 pass
 
+            # TODO: This needs improving to allow sending content with non-200.
+            # The only choice is when to put msg in content or in header-line
+            # response.
             status,headers,msg = result
-            if status != 200:
+            if status != 200 and msg is not None:
                 msg = msg.decode('utf-8')
                 self.send_response(status, msg)
             else:
@@ -88,7 +92,7 @@ def handle_generic(self, method):
             for key,val in headers.items():
                 self.send_header(key, val)
             self.end_headers()
-            if status == 200:
+            if status == 200 and msg is not None:
                 self.wfile.write(msg)
         except BrokenPipeError:
             log.error("Connection aborted unexpectedly")
@@ -129,7 +133,7 @@ def http_start_server_handler(eff: dict):
     thread = None
     try:
         open_requests = {}
-        pushable_stream = Effect({'type': FX.Stream.CreatePushableStream}) | run
+        pushable_stream = {'type': FX.Stream.CreatePushableStream} | run
         server_uuid = str(uid(pushable_stream.stream_ezefref))
         if eff.get('pipe_into', None) is not None:
             pushable_stream | eff['pipe_into']
@@ -206,9 +210,9 @@ def http_send_response_handler(eff: Effect):
     d = _effects_processes[eff["server_uuid"]]
     future = d["open_requests"][eff["request_id"]]
 
-    if "response" not in eff:
-        print(f"Warning: FX.HTTP.SendResponse wish did not contain 'response' field. This is probably an error. Received wish: {eff}")
-    msg = eff.get("response", "")
+    # if "response" not in eff:
+    #     print(f"Warning: FX.HTTP.SendResponse wish did not contain 'response' field. This is probably an error. Received wish: {eff}")
+    msg = eff.get("response", None)
     assert msg is None or isinstance(msg, str) or isinstance(msg, bytes)
     if isinstance(msg, str):
         msg = msg.encode("utf-8")
@@ -288,7 +292,7 @@ def send_response(req):
     if "response_headers" in req:
         eff_d["headers"] = req["response_headers"]
 
-    return Effect(eff_d)
+    return eff_d
 
 @func
 def fallback_not_found(req):
@@ -315,11 +319,12 @@ def middleware_worker(req, mw):
 
         if cur is None:
             return None
+        # Not sure if this is the right way to do things
+        SendResponseEffect = Pattern[{"type": SetOf(FX.HTTP.SendResponse)}]
+        if is_a(cur, SendResponseEffect):
+            return cur
         if isinstance(cur, dict):
             continue
-        # if isinstance(cur, Effect):
-        if isinstance(cur, dict):
-            return cur
 
         raise Exception(f"Unusual type in middleware_runner: {type(func)}")
         
diff --git a/python/zef/core/fx/local_file.py b/python/zef/core/fx/local_file.py
index 2a808d36..bfaec1f9 100644
--- a/python/zef/core/fx/local_file.py
+++ b/python/zef/core/fx/local_file.py
@@ -99,12 +99,12 @@ def save_localfile_handler(eff: Effect):
     """
     Example Effect
 
-    Effect({
+    {
         'type':     FX.LocalFile.Save,
         'filename': fname,
         'content':  content,
         'settings': settings,
-    })
+    }
     """
     try:
         content   = eff["content"]
@@ -143,11 +143,11 @@ def write_localfile_handler(eff: Effect):
     """
     Example Effect
 
-    Effect({
+    {
         'type':     FX.LocalFile.Write,
         'filename': fname,
         'content':  content,
-    })
+    }
     """
     content   = eff["content"]
     filename  = eff["filename"]
@@ -164,10 +164,10 @@ def system_open_with_handler(eff: Effect):
     Example
     =======
 
-    Effect({
+    {
             'type': FX.LocalFile.SystemOpenWith,
             'filepath': fname,
-    })
+    }
     """
     try:
         import subprocess, os, platform
diff --git a/python/zef/core/graph_delta.py b/python/zef/core/graph_delta.py
index 46fa6382..f44ad41a 100644
--- a/python/zef/core/graph_delta.py
+++ b/python/zef/core/graph_delta.py
@@ -50,7 +50,62 @@ class ListOrTuple(ABC):
 # - expressions: high-level things which match to user expectations. There are "complex" expressions and "primitive" expressions.
 #
 
-# TODO: REDO THE DESCRIPTION OF THIS FILE
+# 1. Entrypoints:
+# There are two entrypoints to the code in this file. Either:
+#
+# receipt = [... list of expressions ...] | transact[g] | run
+# or
+# unpacked_template = template | g | run
+#
+# The first is a "GraphDelta" whereas the second is the shorthand syntax.
+#
+# The shorthand syntax goes through `encode` which will first assign IDs to any
+# parts necessary for unpacking the result into the same template structure. It
+# then passes the result to `construct_commands`
+#
+# The `transact[g]` route goes directly to `construct_commands`
+#
+# 2. `construct_commands`
+# Each expressions given by the user is passed to the cmds_for_* functions to
+# convert it to either a) an equivalent command, and/or b) additional expressions.
+#
+# To make this work more easily, `realise_single_node` can take any expression
+# which should result in a "single node" (e.g. an ET.Something, or a
+# {ET.Something: ...} syntax) and returns a simplified set of expressions to
+# realise this, along with the ID to refer to that "single node"
+#
+# Pretty much the only object that cannot be passed to `realise_single_node` is
+# a relation triple. This is handled in `cmds_for_tuple`.
+#
+# Many expressions passed in are of a LazyValue type. These are consider the
+# same as values, and special forms such as assign, set_field or tag are just
+# LazyValues that evaluate to themselves.
+#
+# 3. Flow
+#
+# (maybe) `encode`
+# `construct_commands`:
+#     (pre-construct verification)
+#     - `obtain_ids` 
+#     - `verify_input_el`
+#     - `verify_relation_source_target`
+#     (construction)
+#     - `iteration_step` -> `dispatch_cmds_for` -> `cmds_for_*` -> `realise_single_node`
+#     - `verify_and_compact_commands`
+# `perform_transaction_commands`
+#
+# 4. IDs
+#
+# The IDs must be consistently unique throughout each GraphDelta. A `gen_id`
+# object is passed around the functions and produces ids beginning with
+# "tmp_id". This is exploited to filter out automatically-generated IDs in the
+# receipt.
+#
+# IDs are how the individual commands are linked to one another. In the end, any
+# command that references an ID must come after the ID which creates that item.
+# IDs refer to "internal IDs" which do not exist on the graph after the
+# transaction finishes. However they are returned in the receipt, allowing user
+# code to identify these objects later.
 
 
 
@@ -89,12 +144,12 @@ def insert_id_maybe(cmd: dict):
             return cmd
 
         commands_with_ids = [insert_id_maybe(c) for  c in commands]
-        return Effect({
+        return {
                 "type": FX.Graph.Transact,
                 "target_graph": g,
                 "commands": commands_with_ids,
                 "unpacking_template": unpacking_template,
-            })
+            }
     raise NotImplementedError(f"'x | g' for x of type {type(x)}")
 
 
@@ -569,7 +624,7 @@ def cmds_for_lv_set_field(x, gen_id):
 
     # assert is_a(source, ZefOp) and is_a(source, Z), f"fill_or_attach reached cmd creation with an incorrect input type: {source}"
     assert is_a(op, set_field)
-    rt,assignment = LazyValue(op) | absorbed | collect
+    rt,assignment,incoming = LazyValue(op) | absorbed | collect
     assert type(rt) == RelationType
 
     iid,exprs = realise_single_node(source, gen_id)
@@ -578,6 +633,7 @@ def cmds_for_lv_set_field(x, gen_id):
         'cmd': 'set_field', 
         'source_id': iid,
         'rt': LazyValue(rt) | without_absorbed | collect,
+        'incoming': incoming,
     }
 
     if type(assignment) in scalar_types:
@@ -1182,14 +1238,20 @@ def perform_transaction_commands(commands: list, g: Graph):
                         if z_target is None:
                             raise KeyError("set_field called with entity that is not known {cmd['target_id']}")
 
-                    opts = z_source | out_rels[cmd['rt']] | collect
+                    if cmd['incoming']:
+                        opts = z_source | in_rels[cmd['rt']] | collect
+                    else:
+                        opts = z_source | out_rels[cmd['rt']] | collect
                     if len(opts) == 2:
                         raise Exception(f"Can't set_field to {z_source} because it has two or more relations of kind {rt}")
                     elif len(opts) == 1:
                         zz = single(opts)
                         if 'value' in cmd:
                             # AE path - overwrite value
-                            ae = target(zz)
+                            if cmd['incoming']:
+                                ae = source(zz)
+                            else:
+                                ae = target(zz)
                             if aet != rae_type(ae):
                                 raise Exception(f"Can't fill or attach {ae} because it is the wrong type for value {cmd['value']}")
                             if value(ae) != cmd['value']:
@@ -1198,16 +1260,25 @@ def perform_transaction_commands(commands: list, g: Graph):
                             # Entity path - replace the relation
                             from ..pyzef import zefops as pyzefops
                             pyzefops.terminate(zz)
-                            zz = instantiate(z_source, cmd['rt'], z_target, g)
+                            if cmd['incoming']:
+                                zz = instantiate(z_target, cmd['rt'], z_source, g)
+                            else:
+                                zz = instantiate(z_source, cmd['rt'], z_target, g)
                     else:
                         if 'value' in cmd:
                             # AE path
                             ae = instantiate(aet, g)
                             internals.assign_value_imp(ae, cmd['value'])
-                            zz = instantiate(z_source, cmd['rt'], ae, g)
+                            if cmd['incoming']:
+                                zz = instantiate(ae, cmd['rt'], z_source, g)
+                            else:
+                                zz = instantiate(z_source, cmd['rt'], ae, g)
                         else:
                             # Entity path
-                            zz = instantiate(z_source, cmd['rt'], z_target, g)
+                            if cmd['incoming']:
+                                zz = instantiate(z_target, cmd['rt'], z_source, g)
+                            else:
+                                zz = instantiate(z_source, cmd['rt'], z_target, g)
                     # This zz is the relation that connects the source/target
                     zz = now(zz)
                     
@@ -1429,21 +1500,6 @@ def filter_temporary_ids(receipt):
             | func[dict]
             | collect)
 
-
-
-@func
-def single_or(itr, default):
-    itr = iter(itr)
-    try:
-        ret = next(itr)
-        try:
-            next(itr)
-            raise Exception("single_or detected more than one item in iterator")
-        except StopIteration:
-            return ret
-    except StopIteration:
-        return default
-
 @func
 def get_absorbed_id(obj):
     # THIS SHOULDN"T BE NEEDED! FIX!
@@ -1481,76 +1537,3 @@ def merge_no_overwrite(a,b):
             raise Exception(f"The internal id '{k}' refers to multiple objects, including '{a[k]!r}' and '{v!r}'. This is ambiguous and not allowed.")
         a[k] = v
     return a
-
-
-
-################################################
-# * Old not used anymore?
-#----------------------------------------------
-
-
-
-def verify_assign_values(assign_v, id_definitions):
-    new_value = assign_v['new_value']
-
-    # This function is called from 2 different locations. First inside the AssignValue_ constructor where we are able to cast a ZefRef to its rae_type
-    # Secondly, inside the GraphDelta constructor where we check against uid inside the id_definitions dict.
-    if "zr_type" in assign_v:
-        type_of_assignedto = assign_v['zr_type']
-    else:
-        if assign_v['some_id'] in id_definitions: type_of_assignedto = id_definitions[assign_v['some_id']]
-        else: type_of_assignedto = None
-
-    # This comes first as _eq_ for ZefRef throws in the next if statement
-    if isinstance(new_value, ZefRef) or isinstance(new_value, EZefRef) or isinstance(new_value, ZefRefs) or isinstance(new_value, EZefRefs):
-        raise RuntimeError(f'The passed value can\'t be of type Zefref(s) or EZefRef(s)')
-
-    # If no value was passed
-    if new_value == None:
-        raise RuntimeError(f'{assign_v} is missing a value to be assigned.')
-
-    # If type of the value passed isn't part of this set of allowed types
-    if type(new_value) in scalar_types:
-        raise RuntimeError(f'The type of the value passed in {assign_v} isn\'t of the allowed types: str, int, float, bool, Time, QuantityFloat, QuantityInt, ZefEnumValue')
-
-    if type_of_assignedto: 
-        type_map = {
-            str:           AET.String,
-            ZefEnumValue:  AET.Enum,
-            QuantityFloat: AET.QuantityFloat, 
-            QuantityInt:   AET.QuantityInt, 
-            Time:          AET.Time,
-        }
-        # If type of the rae that is being assigned isn't of an AET type i.e assigning a str to an ET.Cat
-        if type_of_assignedto not in {AET.Int, AET.Float, AET.Bool, AET.String, AET.Enum, AET.QuantityFloat, AET.QuantityInt, AET.Time}:
-            raise RuntimeError(f'The type_of_assignedto={type_of_assignedto} must be of the allowed types: AET.Int, AET.Float, AET.Bool, AET.String, AET.Enum, AET.QuantityFloat, AET.QuantityInt, AET.Time')
-
-        # Verify cast-ability
-        elif type(new_value) == int:
-            if type_of_assignedto not in {AET.Int, AET.Float, AET.Bool}:
-                raise RuntimeError(f'Can only assign values of type int to only AET.Int, AET.Float, AET.Bool')
-            if type_of_assignedto == AET.Bool and new_value not in [0,1]:
-                raise RuntimeError(f'Can\'t assign int value that isn\'t 0 or 1 to AET.Bool')
-            
-        elif type(new_value) == float:
-            if type_of_assignedto not in {AET.Int, AET.Float}:
-                raise RuntimeError(f'Can only assign values of type float to only AET.Int, AET.Float')
-            import math
-            if type_of_assignedto == AET.Int and math.fabs(new_value - round(new_value)) > 1e-8:
-                raise RuntimeError(f'Can only assign values of type float to int if the double is numerically sufficiently close to make rounding safe.')
-
-        elif type(new_value) == bool:    
-            if type_of_assignedto not in {AET.Int, AET.Bool}:
-                raise RuntimeError(f'Can only assign values of type bool to only AET.Int, AET.Bool')
-            if type_of_assignedto == AET.Int and new_value not in [False, True]:
-                raise RuntimeError(f'Can\'t assign bool value that isn\'t True or False to AET.Int')
-
-        # These are only castable one-to-one
-        elif type_map[type(new_value)] != type_of_assignedto:
-            raise RuntimeError(f'Can\'t assign value of type {type(new_value)} to an AET of type {type_of_assignedto}')
-
-
-
-
-
-
diff --git a/python/zef/core/image.py b/python/zef/core/image.py
index caa70f94..7534bca0 100644
--- a/python/zef/core/image.py
+++ b/python/zef/core/image.py
@@ -56,7 +56,7 @@ def _view(self, format=None):
 
         from . import FX, Effect
         from ._ops import run
-        Effect({
+        {
             "type": FX.LocalFile.SystemOpenWith,
             "filepath": filename
-        }) | run
\ No newline at end of file
+        } | run
\ No newline at end of file
diff --git a/python/zef/core/op_implementations/dispatch_dictionary.py b/python/zef/core/op_implementations/dispatch_dictionary.py
index 37909ca7..e62fdd54 100644
--- a/python/zef/core/op_implementations/dispatch_dictionary.py
+++ b/python/zef/core/op_implementations/dispatch_dictionary.py
@@ -309,4 +309,6 @@
         RT.RelatedOps:         (related_ops_imp, None),
         RT.OperatesOn:         (operates_on_imp, None), 
         RT.UsedFor:            (used_for_imp, None),
+
+        RT.Gather:             (gather_imp, None),
 }
diff --git a/python/zef/core/op_implementations/graphviz.py b/python/zef/core/op_implementations/graphviz.py
index 37316242..fc52f601 100644
--- a/python/zef/core/op_implementations/graphviz.py
+++ b/python/zef/core/op_implementations/graphviz.py
@@ -47,10 +47,12 @@ def graphviz_imp(zz, *flags):
     List[ZefRef] -> Image
     (List[ZefRef], ZefOp) -> Image
     """
-    if isinstance(zz, FlatGraph): 
+    if is_a(zz, FlatGraph): 
         g = Graph()
         zz | transact[g] | run 
+        contains_del = any([is_a(b[1], Delegate) for b in zz.blobs])
         zz = g | now | all[RAE] | collect
+        if contains_del: zz = zz + (g | blueprint[True] | collect)
     from functools import lru_cache
     try:
         import graphviz
@@ -187,7 +189,7 @@ def nice_shape(z):
 
     def nice_style(z):
         if BT(z) in {BT.ENTITY_NODE, BT.ATOMIC_ENTITY_NODE}:
-            return '' if internals.is_delegate(z) else 'filled'
+            return 'dashed' if internals.is_delegate(z) else 'filled'
         else:
             return ''
     
@@ -206,8 +208,9 @@ def add_gv_edges(z):
         # ------- hacky / dirty way for now: just try going to source / target and swallow error if anything fails :| --------
         try:
             col = edge_colors[index(z)]   # this was determined above
-            G.edge(str(index(z | source | collect)), str(index(z)), color=col, dir="back", arrowtail="dot")
-            G.edge(str(index(z)), str(index(z | target | collect)), color=col, dir="forward", arrowhead="normal")
+            additional = {"style": "dashed"} if is_a(z, Delegate) else {}
+            G.edge(str(index(z | source | collect)), str(index(z)), color=col, dir="back", arrowtail="dot", **additional)
+            G.edge(str(index(z)), str(index(z | target | collect)), color=col, dir="forward", arrowhead="normal", **additional)
         except Exception as e:
             #print(f"add_gv_edges failed for {z} e={e}")
             pass
diff --git a/python/zef/core/op_implementations/implementation_typing_functions.py b/python/zef/core/op_implementations/implementation_typing_functions.py
index c52f9e74..38b0fa51 100644
--- a/python/zef/core/op_implementations/implementation_typing_functions.py
+++ b/python/zef/core/op_implementations/implementation_typing_functions.py
@@ -203,7 +203,7 @@ def on_implementation(g, op):
     from ...pyzef import zefops as internal
     from ..fx import FX, Effect
 
-    stream =  Effect({'type': FX.Stream.CreatePushableStream}) | run
+    stream =  FX.Stream.CreatePushableStream() | run
     sub_decl = internal.subscribe[internal.keep_alive[True]]
     
     if isinstance(g, Graph):
@@ -379,7 +379,7 @@ def peel_tp(op, curr_type):
     if curr_type == VT.Effect:
         return VT.Dict[VT.String, VT.Any]
     elif curr_type in {VT.ZefOp, VT.LazyValue}:
-        return VT.Record
+        return VT.Tuple
     else:
         return absorbed(curr_type)[0]
 
@@ -1292,7 +1292,7 @@ def insert_imp(d: dict, key, val=None):
     - related zefop: remove
     - related zefop: update
     """
-    if isinstance(d, FlatGraph):
+    if is_a(d, FlatGraph):
         fg = d
         new_el = key
         return fg_insert_imp(fg, new_el)
@@ -1419,7 +1419,7 @@ def remove_imp(d: dict, key_to_remove: tuple):
     - related zefop: insert
     - related zefop: get
     """
-    if isinstance(d, FlatGraph):
+    if is_a(d, FlatGraph):
         fg = d
         return fg_remove_imp(fg, key_to_remove)
     return {p[0]: p[1] for p in d.items() if p[0] != key_to_remove}
@@ -1458,7 +1458,7 @@ def get_imp(d, key, default=Error('Key not found in "get"')):
     - related zefop: select_in
     """
     from typing import Generator
-    if isinstance(d, FlatGraph):
+    if is_a(d, FlatGraph):
         return fg_get_imp(d, key)
     elif isinstance(d, dict):
         return d.get(key, default)
@@ -1778,7 +1778,7 @@ def all_imp(*args):
 
     import builtins
     from typing import Generator, Iterator   
-    if isinstance(args[0], FlatGraph):
+    if is_a(args[0], FlatGraph):
         return fg_all_imp(*args)
     if isinstance(args[0], ZefRef) and is_a[ET.ZEF_List](args[0]):
         z_list = args[0]
@@ -2086,11 +2086,11 @@ def push_imp(item: VT.Any, stream: VT.Stream):  # -> Union[Effect, Error]
     (T, Stream) -> Effect
     """
     if isinstance(stream, Awaitable) or isinstance(stream, ZefRef) or isinstance(stream, EZefRef):
-        return Effect({
+        return {
             'type': FX.Stream.Push,
             'stream': stream,
             'item': item,
-        })
+        }
     raise RuntimeError('push must be called with a Stream pushable stream represented by a ZefRef[ET[PushableStream]] curried in.')
     
     
@@ -4964,8 +4964,8 @@ def fill_or_attach_implementation(z, rt, val):
 def fill_or_attach_type_info(op, curr_type):
     return curr_type
 
-def set_field_implementation(z, rt, val):
-    return LazyValue(z) | set_field[rt][val]
+def set_field_implementation(z, rt, val, incoming=False):
+    return LazyValue(z) | set_field[rt][val][incoming]
 
 def set_field_type_info(op, curr_type):
     return curr_type
@@ -5022,14 +5022,15 @@ def hasin_implementation(zr, rt):
 # -------------------------------- apply_functions -------------------------------------------------
 def apply_functions_imp(x, fns):
     """ 
-    Apply different functions to different elements in a list.
-    The use case is similar to using Zef's map[f1,f2,f3] with multiple
-    functions, but fills in the gap when the input is not a list / stream.
-    In this sense, it is closer to func.
-    The input list and list of functions must be of the same length.
+    Given a list of values and an associated list of functions,
+    apply the nth function to the nth value list element.
+
+    It is strongly recommended not to use impure functions inside
+    this operator.
 
     ---- Examples ----
-    ['Hello', 'World'] | apply_functions[to_upper_case, to_lower_case]    # => ['HELLO', 'world']
+    >>> ['Hello', 'World'] | apply_functions[to_upper_case, to_lower_case]    # => ['HELLO', 'world']
+    >>> (1,2) | apply_functions[add[1], add[10]]      # => (2, 12)
 
     ---- Signature ----
     (Tuple[T1, ...,TN], Tuple[T1->TT1, ..., TN->TTN] ) -> Tuple[TT1, ...,TTN]
@@ -5038,18 +5039,21 @@ def apply_functions_imp(x, fns):
     - used for: control flow
     - operates on: ZefOp
     - operates on: Function
-    - related zefop: func
     - related zefop: reverse_args
     - related zefop: map
+    - related zefop: apply
     """
     import builtins
-    if not (isinstance(x, tuple) or isinstance(x, list)):
-        raise TypeError(f"The dataflow argument for apply_functions must be a list/tuple: Got a type(x)={type(x)} Value: x={x}")
-    if not len(fns) == len(x):
-        raise TypeError(f"the length of the dataflow argument for apply_functions must be the same length as the list of functions provided. Got x={x} and fns={fns}" )
-    return tuple(
-        (f(el) for f, el in builtins.zip(fns, x))
-    )
+    from typing import Generator
+    if not (isinstance(fns, list) or isinstance(fns, tuple)):
+        raise TypeError(f"apply_functions must be given a tuple of functions.")
+    if not (isinstance(x, list) or isinstance(x, tuple) or isinstance(x, Generator)):
+        raise TypeError(f"apply_functions must be given a tuple of input args.")
+    xx = tuple(x)
+    if not len(fns) == len(xx):
+        raise ValueError(f"len(fs) must be equal to len(x)")
+        
+    return tuple((f(el) for f, el in builtins.zip(fns, x)))
 
 
 
@@ -5076,6 +5080,7 @@ def map_implementation(v, f):
     - used for: control flow
     - used for: function application
     - related zefop: apply_functions
+    - related zefop: apply
     """
     import builtins
     input_type = parse_input_type(type_spec(v))
@@ -5540,10 +5545,7 @@ def Out_imp(z, rt=VT.Any, target_filter= None):
     #     return Error(f'Invalid type "{rt}" specified in Out[...]')
     # res = target(out_rel_imp(z, rt))
     # if target_filter and not is_a_implementation(res, target_filter):
-    res = Outs_imp(z, rt, target_filter)
-    if len(res) != 1:
-        raise Exception(f"Out didn't find a single target node that matched the [{rt}][{target_filter}] filter")
-    return single(res)
+    return target_implementation(out_rel_imp(z, rt, target_filter))
 
 
 
@@ -5569,11 +5571,7 @@ def Outs_imp(z, rt=None, target_filter = None):
     assert isinstance(z, (ZefRef, EZefRef, FlatRef))
     if isinstance(z, FlatRef): return traverse_flatref_imp(z, rt, "outout", "multi")
 
-    res = out_rels_imp(z, rt) | map[target] | collect
-    if target_filter: 
-        if isinstance(target_filter, ZefOp): target_filter = Is[target_filter]
-        return res | filter[is_a[target_filter]] | collect
-    return res
+    return out_rels_imp(z, rt, target_filter) | map[target] | collect
 
 
 #---------------------------------------- In -----------------------------------------------
@@ -5599,10 +5597,7 @@ def In_imp(z, rt=None, source_filter = None):
     assert isinstance(z, (ZefRef, EZefRef, FlatRef))
     if isinstance(z, FlatRef): return traverse_flatref_imp(z, rt, "inin", "single")
 
-    res = Ins_imp(z, rt, source_filter)
-    if len(res) != 1:
-        raise Exception(f"In didn't find a single source node that matched the [{rt}][{source_filter}] filter")
-    return single(res)
+    return source_implementation(in_rel_imp(z, rt, source_filter))
 
 
 #---------------------------------------- Ins -----------------------------------------------
@@ -5626,11 +5621,7 @@ def Ins_imp(z, rt=None, source_filter = None):
     assert isinstance(z, (ZefRef, EZefRef, FlatRef))
     if isinstance(z, FlatRef): return traverse_flatref_imp(z, rt, "inin", "multi")
 
-    res = in_rels_imp(z, rt) | map[source] | collect
-    if source_filter: 
-        if isinstance(source_filter, ZefOp): source_filter = Is[source_filter]
-        return res | filter[is_a[source_filter]] | collect
-    return res
+    return in_rels_imp(z, rt, source_filter) | map[source] | collect
 
 
 #---------------------------------------- out_rel -----------------------------------------------
@@ -5659,8 +5650,30 @@ def out_rel_imp(z, rt=None, target_filter = None):
 
     opts = out_rels_imp(z, rt, target_filter)
     if len(opts) != 1:
-        # TODO: hinting if problems occur goes here
-        raise Exception("out_rel did not find a single edge. TODO hints here.")
+        # We use a function here for ease of control flow
+        def help_hint():
+            hint = ""
+            if len(opts) > 0:
+                hint += f"out_rel found too many relations that satisfy RT⊆{rt!r}"
+                if target_filter is not None:
+                    hint += f" and target={target_filter!r}"
+                return hint 
+            else:
+                hint += f"out_rel did not find any relations that satisfy RT⊆{rt!r}"
+                if target_filter is not None:
+                    hint += f" and target={target_filter!r}"
+                    num = len(out_rels_imp(z, rt, None))
+                    if num > 0:
+                        hint += f"\nThere are {num} relations of kind {rt!r}, maybe you did not mean to include the target filter?"
+                        return hint 
+
+                if len(in_rels_imp(z, rt, target_filter)) > 0:
+                    hint += f"\nThere are incoming relations of this kind, maybe you meant to write in_rel or In instead."
+                    return hint
+
+                return hint
+
+        raise Exception(help_hint())
     return single(opts)
 
 #---------------------------------------- out_rels -----------------------------------------------
@@ -5721,8 +5734,30 @@ def in_rel_imp(z, rt=None, source_filter = None):
 
     opts = in_rels_imp(z, rt, source_filter)
     if len(opts) != 1:
-        # TODO: hinting if problems occur goes here
-        raise Exception("in_rel did not find a single edge. TODO hints here.")
+        # We use a function here for ease of control flow
+        def help_hint():
+            hint = ""
+            if len(opts) > 0:
+                hint += f"in_rel found too many relations that satisfy RT⊆{rt!r}"
+                if source_filter is not None:
+                    hint += f" and source={source_filter!r}"
+                return hint 
+            else:
+                hint += f"in_rel did not find any relations that satisfy RT⊆{rt!r}"
+                if source_filter is not None:
+                    hint += f" and source={source_filter!r}"
+                    num = len(in_rels_imp(z, rt, None))
+                    if num > 0:
+                        hint += f"\nThere are {num} relations of kind {rt!r}, maybe you did not mean to include the source filter?"
+                        return hint 
+
+                if len(out_rels_imp(z, rt, source_filter)) > 0:
+                    hint += f"\nThere are outgoing relations of this kind, maybe you meant to write out_rel or Out instead."
+                    return hint
+
+                return hint
+
+        raise Exception(help_hint())
     return single(opts)
 
 
@@ -6097,6 +6132,54 @@ class Sentinel: pass
         
         raise NotImplementedError(f"Pattern ValueType isn't implemented for {x}")
 
+
+    def list_matching(x, tp):
+        import sys
+        from typing import Generator
+        if not (isinstance(x, list) or isinstance(x, tuple) or isinstance(x, Generator)):
+            return False
+        ab = tp.d['absorbed']
+        if ab != ():
+            if isinstance(x, Generator):
+                raise NotImplementedError()
+            if len(ab)!=1:    # List takes only one Type argument
+                print('Something went wrong in `is_a[List[...]]`: multiple args curried into ', file=sys.stderr)
+            return x | map[is_a[ab[0]]] | all | collect
+        else:
+            return True
+
+
+    def set_matching(x, tp):
+        import sys
+        if not isinstance(x, set):
+            return False
+        ab = tp.d['absorbed']
+        if ab != ():
+            if len(ab)!=1:    # List takes only one Type argument
+                print(f'Something went wrong in `is_a[Set[T1]]`: Set takes exactly one subtype, but got {x}', file=sys.stderr)
+            return x | map[is_a[ab[0]]] | all | collect
+        else:
+            return True
+
+
+    def dict_matching(x, tp):
+        import sys
+        if not (isinstance(x, dict)):
+            return False
+        ab = tp.d['absorbed']
+        if ab != ():
+            if (len(ab)==1 and len(ab[0])==2):    # Dict must contain a type in one [] and that must be a pair
+                T1, T2 = ab[0]
+                return x | items | map[lambda p: is_a(p[0], T1) and is_a(p[1], T2)] | all | collect
+            else:
+                print(f'Something went wrong in `is_a[Dict[T1, T2]]`: exactly two subtypes must be given. Got {ab}', file=sys.stderr)
+                return Error('Error matching Dict[...]')
+        else:
+            return True
+
+
+
+
     def valuetype_matching(el, vt):
         if vt == VT.Any: return True
 
@@ -6124,6 +6207,7 @@ def valuetype_matching(el, vt):
         if vt.d['type_name'] not in vt_name_to_python_type: return Error.NotImplementedError(f"ValueType_ matching not implemented for {vt}")
         python_type = vt_name_to_python_type[vt.d['type_name']]
         return isinstance(el, python_type)
+    
 
     if isinstance(typ, ValueType_):
         if typ.d['type_name'] == "Union":
@@ -6147,6 +6231,16 @@ def valuetype_matching(el, vt):
         if typ.d['type_name'] == "Complement":
             return not is_a(x, typ.d['absorbed'][0])
 
+        if typ.d['type_name'] == "List":
+            return list_matching(x, typ)
+
+        if typ.d['type_name'] == "Set":
+            return set_matching(x, typ)
+
+        if typ.d['type_name'] == "Dict":
+            return dict_matching(x, typ)
+
+
         if typ.d['type_name'] in  {"Instantiated", "Assigned", "Terminated"}:
             map_ = {"Instantiated": instantiated, "Assigned": assigned, "Terminated": terminated}
             def compare_absorbed(x, typ):
@@ -6161,6 +6255,10 @@ def compare_absorbed(x, typ):
         if typ.d['type_name'] == "Pattern":
             return pattern_vt_matching(x, typ)
 
+        if typ.d['type_name'] == "FlatGraph":
+            from zef.core.flat_graph import FlatGraph_
+            return isinstance(x, FlatGraph_)
+
         return valuetype_matching(x, typ)
                     
     # To handle user passing by int instead of Int by mistake
@@ -6266,6 +6364,15 @@ def compare_absorbed(x, typ):
     except TypeError:
         return False
 
+
+
+
+
+
+
+
+
+
 def _is_a_instance_delegate_generic(x, typ):
     # This function is for internal use only and does the comparisons against
     # ET, ET.x, RT, (Z, RT.x, ET.y), etc... ignoring whether the reference is an
@@ -6424,7 +6531,7 @@ def reduce_type_info(op, curr_type):
         return VT.List[VT.Any]
 
 def group_by_type_info(op, curr_type):
-    return VT.List[VT.Record[VT.Any]]
+    return VT.List[VT.Tuple[VT.Any]]
 
 def identity_type_info(op, curr_type):
     return curr_type
@@ -6626,13 +6733,13 @@ def tag_imp(x, tag_s: str, *args):
             force = args[0]
         else:
             force = False
-        return Effect({
+        return {
             'type': FX.Graph.Tag,
             'graph': x,
             'tag': tag_s,
             'force': force,
             'adding': True,
-        })
+        }
     if isinstance(x, ZefRef) or isinstance(x, EZefRef) or is_a(x, Z):
         assert len(args) == 0
         return LazyValue(x) | tag[tag_s]
@@ -6656,13 +6763,13 @@ def untag_imp(x, tag: str):
     (ZefRef, str) -> Effect
     """
     if isinstance(x, Graph):
-        return Effect({
+        return {
             'type': FX.Graph.Tag,
             'graph': x,
             'tag': tag,
             'adding': False,
             'force': False,
-        })
+        }
     if isinstance(x, ZefRef) or isinstance(x, EZefRef) or is_a(x, Z):
         raise Exception("Untagging a RAE is not supported at the moment.")
 
@@ -6687,11 +6794,11 @@ def sync_imp(x: VT.Graph, sync_state: bool = True):
     Graph -> Effect
     (Graph, bool) -> Effect
     """
-    return Effect({
+    return {
         'type': FX.Graph.Sync,
         'graph': x,
         'sync_state': sync_state,
-    })
+    }
     
 def sync_tp(op, curr_type):
     return VT.Effect
@@ -6722,7 +6829,7 @@ def merge_imp(a, second=None, *args):
     - related zefop: merge_with
     """
     from typing import Generator
-    if isinstance(a, FlatGraph) and isinstance(second, FlatGraph):
+    if is_a(a, FlatGraph) and is_a(second, FlatGraph):
         return fg_merge_imp(a, second)
     elif second is None:
         assert isinstance(a, tuple) or isinstance(a, list) or isinstance(a, Generator)
@@ -6819,10 +6926,10 @@ def to_clipboard_imp(x):
         return to_clipboard(str(x))
 
     assert type(x) in {str, int, float, bool}
-    return Effect({
+    return {
         'type': FX.Clipboard.CopyTo,
         'value': x
-    })
+    }
 
 
 def to_clipboard_tp(x):
@@ -6836,9 +6943,9 @@ def from_clipboard_imp():
     the content of a clipboard to be copied
     """
     # assert type(x) in {str, int, float, bool}
-    return Effect({
+    return {
         'type': FX.Clipboard.CopyFrom,
-    })
+    }
 
 _call_0_args_translation[RT.FromClipboard] = from_clipboard_imp
 
@@ -6969,10 +7076,10 @@ def read_file_imp(fname):
     - used for: file io
 
     """
-    return Effect({
+    return {
         'type': FX.LocalFile.Read,
         'filename' : fname
-    })
+    }
 
 def read_file_tp(op, curr_type):
     return VT.Effect
@@ -6995,11 +7102,11 @@ def load_file_imp(fname, format = None):
     - used for: file io
 
     """
-    return Effect({
+    return {
         'type'     : FX.LocalFile.Load,
         'filename' : fname,
         'format'   : format
-    })
+    }
 
 def load_file_tp(op, curr_type):
     return VT.Effect
@@ -7027,12 +7134,12 @@ def save_file_imp(content, fname, settings = {}):
     (VT.Any, VT.String) -> VT.Effect
 
     """
-    return Effect({
+    return {
         'type': FX.LocalFile.Save,
         'filename' : fname,
         'content': content,
         'settings': settings,
-    })
+    }
 
 def save_file_tp(op, curr_type):
     return VT.Effect
@@ -7053,11 +7160,11 @@ def write_file_imp(content, fname):
     (VT.Any, VT.String) -> VT.Effect
 
     """
-    return Effect({
+    return {
         'type': FX.LocalFile.Write,
         'filename' : fname,
         'content': content,
-    })
+    }
 
 def write_file_tp(op, curr_type):
     return VT.Effect
@@ -7977,13 +8084,13 @@ def to_screaming_snake_case_tp(op, curr_type):
     return VT.String
 
 def make_request_imp(url, method='GET', params={}, data={}):
-    return Effect({
+    return {
             "type":     FX.HTTP.Request,
             "url":      url,
             "method":   method,
             "params":   params,
             "data":     data,
-    })
+    }
 
 def make_request_tp(op, curr_type):
     return VT.Effect
@@ -8084,7 +8191,7 @@ def fg_insert_imp(fg, new_el):
     from ..graph_delta import map_scalar_to_aet_type
     from ...pyzef.internals import DelegateRelationTriple
 
-    assert isinstance(fg, FlatGraph)
+    assert is_a(fg, FlatGraph)
     new_fg = FlatGraph()
     new_blobs, new_key_dict = [*fg.blobs], {**fg.key_dict}
 
@@ -8472,7 +8579,7 @@ def fr_ins_and_outs_imp(fr):
     return FlatRefs(fr.fg, blob[2])
 
 def fg_all_imp(fg, selector=None):
-    assert isinstance(fg, FlatGraph)
+    assert is_a(fg, FlatGraph)
     if selector:
         selected_blobs = fg.blobs | filter[lambda b: is_a(b[1], selector)] | collect
     else:
@@ -8484,7 +8591,7 @@ def fg_all_imp(fg, selector=None):
 def transact_imp(data, g, **kwargs):
     from typing import Generator
     from ..graph_delta import construct_commands
-    if isinstance(data, FlatGraph):
+    if is_a(data, FlatGraph):
         commands = flatgraph_to_commands(data)
     elif type(data) in {list, tuple}:
         commands = construct_commands(data)
@@ -8493,11 +8600,11 @@ def transact_imp(data, g, **kwargs):
     else:
         raise ValueError(f"Expected FlatGraph or [] or () for transact, but got {data} instead.")
 
-    return Effect({
+    return {
             "type": FX.Graph.Transact,
             "target_graph": g,
             "commands":commands
-    })    
+    }
 
 def transact_tp(op, curr_type):
     return VT.Effect
@@ -8984,31 +9091,26 @@ def blueprint_imp(x, include_edges=False):
 
     if isinstance(x, Graph):
         g = x
-        if include_edges:
-            frontier_of = map_cat[out_rels[BT.TO_DELEGATE_EDGE]] | map_cat[apply[identity, target]]
-        else:
-            frontier_of = map_cat[Outs[BT.TO_DELEGATE_EDGE]]
-
-        frontier = [g | root | collect]
 
-        def step(found, last_frontier):
-            new_frontier = last_frontier | frontier_of | without[found] | collect
-            return found+new_frontier, new_frontier
+        rules = [
+            (Z[Any], BT.TO_DELEGATE_EDGE, Any)
+        ]
 
-        all_items = ((frontier,frontier)
-                     | iterate[unpack[step]]
-                     | take_until[second
-                                  | length
-                                  | equals[0]]
-                     | last
-                     | first
-                     | collect)
+        all_items = g | root | gather[rules] | collect
 
         all_equal = sliding[2] | map[unpack[equals]] | all
         is_delegate_rel_group = BT.RELATION_EDGE & Is[apply[identity, source, target] | all_equal]
+
         trim_items = [g | root | collect] + (all_items | filter[is_delegate_rel_group] | collect)
+        trim_items += trim_items | map_cat[out_rels[BT.TO_DELEGATE_EDGE]] | collect
+
+        all_items = all_items | without[trim_items] | collect
+
+        if not include_edges:
+            all_items = all_items | filter[Not[is_a[BT.TO_DELEGATE_EDGE]]] | collect
+
+        return all_items
 
-        return all_items | without[trim_items | apply[identity, map_cat[out_rels[BT.TO_DELEGATE_EDGE]]] | concat | collect] | collect
     elif isinstance(x, GraphSlice):
         rae_satisfies = match[
             (Delegate, exists_at[x]),
@@ -9148,6 +9250,7 @@ def apply_imp(x, f):
     
     ---- Tags ----
     - related zefop: map
+    - related zefop: apply_functions
     - related zefop: func
     - related zefop: call
     - used for: control flow
@@ -9203,7 +9306,7 @@ def wrapper():      # keep split_on_next a function
 
 
 # ----------------------------- ops acting on op doctstring -----------------------------
-def examples_imp(op: VT.ZefOp) -> VT.List[VT.Record]:
+def examples_imp(op: VT.ZefOp) -> VT.List[VT.Tuple]:
     """
     Returns the examples portion of a docstring as a list of tuples mapping
     example string to a possible output string.
@@ -9219,7 +9322,7 @@ def examples_imp(op: VT.ZefOp) -> VT.List[VT.Record]:
     '"external_data_with_unusualcharacters"')]
 
     ---- Signature ----
-    (ZefOp) -> List[Record[String | Nil]]
+    (ZefOp) -> List[Tuple[String | Nil]]
 
     ---- Tags ----
     - related zefop: tags
@@ -9509,3 +9612,112 @@ def indexes_of_imp(v, ElType):
     """
     return v | enumerate | filter[second | is_a[ElType]] | map[first] | collect
 
+
+
+
+
+
+
+
+def gather_imp(z_initial, rules=None):
+    """ 
+    An operator that given a launch point on a graph, gathers up 
+    a subgraph by traversing based on a rules pattern that is specified.
+    There are two main modes of usage: 
+
+    ---------------- A: Explicit Rules -----------------
+    Giving an explicit rule set:
+    >>> rules = [
+    >>>     (Z[ET.A], RT.R1, ET.B),        # from source
+    >>>     (Z[Any], RT.R2, AET.String),    
+    >>>     (ET.Bar, RT.R2, Z[ET.Foo]),    # from target moving in the opposite direction of the relation
+    >>> ]
+    then 
+    >>> z1 | gather[rules]
+    will iteratively traverse the graph until there is nothing more to be collected.
+    What do the rules specify in this case? For example, the first rule (ET.A, RT.R1, ET.B)
+    being part of 'from_source' mean that if one were to stand on a node of type "ET.A" as a source,
+    with an outgoing relation of type "RT.R1" which leads to a "ET.B", then take that step
+    and include both the edge and the "ET.B" in the gathered set.
+
+    ---------------- B: Implicit Rules -----------------
+    For certain entity types, predefined rule sets are associated.
+    B1) For ET.ZEF_Function, a rule set to gather all things related to that function is predefined.
+    Future: Also for ET.ZEF_EffectModule, ET.ZEF_Op
+    
+    """
+
+
+    # ------------------------------ catch all cases for implicit rules ----------------------------------
+    if rules is None:
+        if z_initial | is_a[ET.ZEF_Function] | collect:
+            zef_fct_rules =  [
+                        (Z[ET.ZEF_Function], RT.PythonSourceCode, AET.String),
+                        (Z[ET.ZEF_Function], RT.OriginalName,     AET.String),
+                        (Z[ET.ZEF_Function], RT.Binding,          ET.ZEF_Function),
+                        (Z[RT.Binding],      RT.Name,             AET.String),
+                        (Z[RT.Binding],      RT.UseTimeSlice,     AET.String),
+                    ]
+            return gather_imp(z_initial, zef_fct_rules)           
+        else:
+            raise TypeError(f"Implicit rules not defined in 'gather' operator for {rae_type(z_initial)}")
+
+
+    # --------------------------------- verify the rules data structure-------------------------------
+    # TODO: once type checking is in place, hoist this out of the body into the function signature
+    ValidRule = (
+        List &
+        Is[length | equals[3]] & (
+            Is[first | without_absorbed | equals[Z]] |
+            Is[last | without_absorbed | equals[Z]] 
+        ))
+    def EveryElement(T):
+        return List & Is[ map[is_a[T]] | all  ]
+
+    ValidRuleList = EveryElement(ValidRule)
+
+    if not rules | is_a[ValidRuleList] | collect:
+        return Error(f'`gather` called with an invalid set of rules: {rules}')
+
+
+    # --------------------------------- explicit rules -------------------------------
+    def traverse_rae_step(z: ZefRef) -> List[ZefRef]:
+
+        def from_single_rule_from_source(triple) -> List[ZefRef]:
+            src_tp, rel_tp, trg_tp = triple
+            if not is_a[src_tp](z):
+                return []
+            return z | out_rels[rel_tp][trg_tp] | map[lambda rel: (rel, target(rel))] | concat | collect
+
+        def from_single_rule_from_target(triple) -> List[ZefRef]:
+            src_tp, rel_tp, trg_tp = triple
+            if not is_a[trg_tp](z):
+                return []
+            return z | in_rels[rel_tp][src_tp] | map[lambda rel: (rel, source(rel))] | concat | collect
+ 
+        from_source_rules = rules | filter[first | without_absorbed | equals[Z]] | map[lambda x: (absorbed(x[0])[0], x[1], x[2])] | collect
+        from_target_rules = rules | filter[last  | without_absorbed | equals[Z]] | map[lambda x: (x[0], x[1], absorbed(x[2])[0])] | collect
+
+        return (
+            from_source_rules | map[from_single_rule_from_source] | concat | collect,
+            from_target_rules | map[from_single_rule_from_target] | concat | collect,
+            ) | concat | collect
+
+    def step(d: dict) -> dict:
+        new_gathered = {*d['gathered'], *d['frontier']}
+        new_frontier = d['frontier'] | map[traverse_rae_step] | concat | filter[Not[contained_in[new_gathered]]] | collect
+        return {
+            'gathered': new_gathered,
+            'frontier': new_frontier,
+        }
+    
+    frontier_is_empty = get['frontier'] | length | equals[0]    
+    return {
+        'gathered': set(),
+        'frontier': {z_initial, },
+    } | iterate[step] | take_until[frontier_is_empty] | last | get['gathered'] | collect
+
+
+
+
+
diff --git a/python/zef/core/op_structs.py b/python/zef/core/op_structs.py
index 60b79bf2..20c7f042 100644
--- a/python/zef/core/op_structs.py
+++ b/python/zef/core/op_structs.py
@@ -302,9 +302,9 @@ def __ror__(self, other):
             from .op_implementations.implementation_typing_functions import on_implementation
             return on_implementation(other, *self.el_ops[0][1])
         elif is_supported_stream(other):
-            from .fx import FX, Effect
+            from .fx import FX
             from ._ops import run
-            stream =  Effect({'type': FX.Stream.CreatePushableStream}) | run
+            stream =  {'type': FX.Stream.CreatePushableStream} | run
             return stream | self
         elif is_supported_value(other) or is_supported_zef_value(other):
             return LazyValue(other) | self
@@ -1082,7 +1082,7 @@ def type_spec_dict(obj):
     return res
 
 def type_spec_tuple(obj):
-    new_tup = VT.Record
+    new_tup = VT.Tuple
     if len(obj) > 0:
         return new_tup[type_spec(obj[0])]
     else:
diff --git a/python/zef/core/serialization.py b/python/zef/core/serialization.py
index b06ec8e5..fc082789 100644
--- a/python/zef/core/serialization.py
+++ b/python/zef/core/serialization.py
@@ -25,7 +25,7 @@
 from .error import _ErrorType, Error
 from .image import Image
 from .fx.fx_types import FXElement, Effect
-from .flat_graph import FlatGraph, FlatRef, FlatRefs
+from .flat_graph import FlatGraph_, FlatRef, FlatRefs
 from ..pyzef import internals as pyinternals
 from .VT import ValueType_
 
@@ -125,7 +125,7 @@ def deserialize_internal(v):
 # * Implementations
 #----------------------------------
 def serialize_flatgraph_or_flatref(fg_or_fr) -> dict:
-    if isinstance(fg_or_fr, FlatGraph):
+    if isinstance(fg_or_fr, FlatGraph_):
         return {
             "_zeftype": "FlatGraph",
             "key_dict": serialize_dict(fg_or_fr.key_dict),
@@ -391,13 +391,13 @@ def deserialize_zeftypes(z) -> dict:
         return Error.__getattribute__(z['type'])(*deserialize_list(z['args']))
 
     elif z['_zeftype'] == "Effect":
-        return Effect(deserialize_dict(z['internal_dict']))
+        return deserialize_dict(z['internal_dict'])
 
     elif z['_zeftype'] == "FXElement":
         return FXElement(tuple(z['elements']))
 
     elif z['_zeftype'] == "FlatGraph":
-        new_fg = FlatGraph()
+        new_fg = FlatGraph_()
         new_fg.key_dict = deserialize_dict(z['key_dict'])
         new_fg.blobs = deserialize_tuple(z['blobs'])
         return new_fg
@@ -481,7 +481,7 @@ def deserialize_valuetype(d):
 serialization_mapping[Image] = serialize_zeftypes
 serialization_mapping[FXElement] = serialize_zeftypes
 serialization_mapping[Delegate] = serialize_delegate
-serialization_mapping[FlatGraph] = serialize_flatgraph_or_flatref
+serialization_mapping[FlatGraph_] = serialize_flatgraph_or_flatref
 serialization_mapping[FlatRef] = serialize_flatgraph_or_flatref
 serialization_mapping[FlatRefs] = serialize_flatgraph_or_flatref
 serialization_mapping[pyinternals.DelegateTX] = serialize_delegate
diff --git a/python/zef/core/zef_rich.py b/python/zef/core/zef_rich.py
new file mode 100644
index 00000000..34aae2e3
--- /dev/null
+++ b/python/zef/core/zef_rich.py
@@ -0,0 +1,265 @@
+# Copyright 2022 Synchronous Technologies Pte Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ._core import *
+from ._ops import *
+from .VT import *
+from .zef_functions import func
+
+@func
+def is_a_component(component, vt):
+    return isinstance(component, ValueType_) and without_absorbed(component) == vt
+
+#--------------------------TEXT--------------------------------------
+def dispatch_rich_text(component):
+    import rich.text as rt
+
+    def resolve_styles(d):
+        # TODO filter out non-rich styles if found
+        from rich.style import Style
+        if "background_color" in d:
+            d['bgcolor'] = d['background_color']
+            d.pop("background_color")
+        return Style(**d)
+
+    def resolve_nonstyle_attributes(d):
+        new_d = {}
+        if "justify" in d: new_d['justify'] = d.pop("justify")
+        if "overflow" in d: new_d['overflow'] = d.pop("overflow")
+        if "no_wrap" in d: new_d['no_wrap'] = d.pop("no_wrap")
+        if "tab_size" in d: new_d['tab_size'] = d.pop("tab_size")
+
+        return new_d
+    
+    def resolve_text_and_nested_data(t):
+        pairs_or_text = []
+        internals = t | absorbed | collect
+        assert isinstance(internals[0], dict), "First absorbed argument for ZefUI Text should be of type dict!"
+        text = internals[0].get("text", "")
+        attributes = {**internals[0]}
+        attributes.pop("text", None)
+        nonstyle_attributes = resolve_nonstyle_attributes(attributes)
+
+        if "data" in attributes:
+            data = attributes["data"]
+            def dispatch_on_data_type(data):
+                # Just a str
+                if isinstance(data, str):
+                    pairs_or_text.append((data, ""))
+
+                # Nested Text component
+                elif is_a_component(data, Text) : 
+                    pairs_or_text.extend(dispatch_rich_text(data))
+                
+                # List of the above 2
+                elif isinstance(data, list):
+                    [dispatch_on_data_type(x) for x in data]
+
+                else:
+                    raise ValueError("Text's data field could contain only a str, Text component, or a list composed of the two.")
+
+            dispatch_on_data_type(data)
+            attributes.pop("data")
+        
+        if "style" in attributes:
+            style = dispatch_rich_style(attributes["style"])
+        else:
+            style = resolve_styles(attributes)
+
+        return [(text, style), *pairs_or_text], nonstyle_attributes
+
+    pairs_or_text, nonstyle_attributes = resolve_text_and_nested_data(component)
+    return rt.Text.assemble(*pairs_or_text, **nonstyle_attributes)
+
+
+
+#--------------------------Code--------------------------------------
+def dispatch_rich_syntax(component):
+    import rich.syntax as rs
+
+    def filter_attributes(d):
+        # TODO filter out non-rich styles if found
+        lexer = d.get("language", "python3")
+        d.pop("language", None)
+        d.pop("code", None)
+        return {**d, "lexer": lexer}
+    
+    internals = component | absorbed | collect
+    assert isinstance(internals[0], dict), "First absorbed argument for ZefUI Code should be of type dict!"
+    code = internals[0].get("code", "")
+    attributes = {**internals[0]}
+    attributes = filter_attributes(attributes)
+    return rs.Syntax(code, **attributes)
+
+
+#--------------------------Style--------------------------------------
+def dispatch_rich_style(component):
+    def resolve_styles(new_d):
+        from rich.style import Style
+        if "background_color" in new_d:
+            new_d['bgcolor'] = new_d['background_color']
+            new_d.pop("background_color")
+        return Style(**new_d)
+    
+    internals = component | absorbed | collect
+    assert isinstance(internals[0], dict), "First absorbed argument for ZefUI Style should be of type dict!"
+    attributes = {**internals[0]}
+    return resolve_styles(attributes)
+
+
+
+#--------------------------Table--------------------------------------
+def dispatch_rich_table(component):
+    import rich.table as rt
+    import itertools as itr
+
+    def handle_nested_components(maybe_component):
+        if isinstance(maybe_component, str):
+            return maybe_component
+        elif is_a_component(maybe_component, Text):
+            return dispatch_rich_text(maybe_component)
+        elif is_a_component(maybe_component, Column):
+            return dispatch_rich_column(maybe_component)
+        elif is_a_component(maybe_component, Style):
+            return dispatch_rich_style(maybe_component)
+        elif isinstance(maybe_component, tuple):
+            return tuple([handle_nested_components(el) for el in maybe_component])
+        else:
+            raise NotImplementedError(f"{maybe_component}:{type(maybe_component)}")
+
+    def resolve_attributes(d):
+        # row_styles: List of styles or strings
+        if "row_styles" in d:
+            d["row_styles"] = [handle_nested_components(x) for x in d["row_styles"]]
+        
+        # title: String or Text
+        if "title" in d:
+            d["title"] = handle_nested_components(d["title"])
+
+        # cols: string or text
+        if "cols" in d:
+            d["cols"] = [handle_nested_components(x) for x in d["cols"]]
+        
+        # rows: string or text
+        if "rows" in d:
+            d["rows"] = [handle_nested_components(x) for x in d["rows"]]
+
+        if "box" in d:
+            d["box"] = box_constants_mapping(d["box"])
+
+        return d
+    
+    internals = component | absorbed | collect
+    assert isinstance(internals[0], dict), "First absorbed argument for ZefUI Table should be of type dict!"
+    attributes = resolve_attributes({**internals[0]})
+    cols = attributes.pop('cols', [])
+    rows = attributes.pop('rows', [])
+    rich_table = rt.Table(*cols, **attributes)
+    # [rich_table.add_column(col, style = style) for (col, style) in itr.zip_longest(cols, col_styles)]
+    [rich_table.add_row(*row) for row in rows]
+    return rich_table
+
+#--------------------------Column--------------------------------------
+def dispatch_rich_column(component):
+    import rich.table as rt
+
+    def resolve_attributes(d):
+        allowed_keys = ["header_style", "footer_style", "style", "justify", "vertical", "width", "min_width", "max_width", "ratio", "no_wrap"]
+        attributes = select_keys(d, *allowed_keys)
+        # Resolve the non-string styles if found
+        for special_key in ["header_style", "footer_style", "style"]:
+            if special_key in attributes and is_a_component(attributes[special_key], Style):
+                attributes[special_key] = dispatch_rich_style(attributes[special_key])
+        
+        return attributes
+    
+    internals = component | absorbed | collect
+    assert isinstance(internals[0], dict), "First absorbed argument for ZefUI Column should be of type dict!"
+    
+    header = internals[0].get("text", "")
+    if is_a_component(header, Text): header = dispatch_rich_text(header)
+    attributes = resolve_attributes(internals[0])
+
+    return rt.Column(header, **attributes)
+
+
+#--------------------------Frame--------------------------------------
+def dispatch_rich_panel(component):
+    import rich.panel as rp
+
+    def filter_and_resolve_attributes(d):
+        d.pop("displayable", None)
+
+        if "title" in d and is_a_component(d["title"], Text): 
+            d["title"] = dispatch_rich_text(d["title"])
+        
+        if "subtitle" in d and is_a_component(d["title"], Text): 
+            d["subtitle"] = dispatch_rich_text(d["subtitle"])
+
+        if "box" in d:
+            d["box"] = box_constants_mapping(d["box"])
+
+        return d
+    
+    internals = component | absorbed | collect
+    assert isinstance(internals[0], dict), "First absorbed argument for ZefUI Code should be of type dict!"
+    displayable = internals[0].get("displayable")
+    displayable = match_and_dispatch(displayable)
+    if not displayable: raise ValueError("Frame's displayable field wasn't passed a component!")
+    
+    attributes = {**internals[0]}
+    attributes.pop("displayable", None)
+    attributes = filter_and_resolve_attributes(attributes)
+    return rp.Panel(displayable, **attributes)
+
+
+#-------------------Dispatch-------------------------------------
+def box_constants_mapping(box_style: str):
+    from rich import box
+    str_to_constant = {
+        'ascii':                   box.ASCII,
+        'square':                  box.SQUARE,                
+        'minimal':                 box.MINIMAL,  
+        'minimal_heavy_head':      box.MINIMAL_HEAVY_HEAD,     
+        'minimal_double_head':     box.MINIMAL_DOUBLE_HEAD,           
+        'simple':                  box.SIMPLE,    
+        'heavy':                   box.HEAVY,                 
+        'heavy_edge':              box.HEAVY_EDGE,             
+        'heavy_head':              box.HEAVY_HEAD,
+        'double':                  box.DOUBLE,               
+        'double_edge':             box.DOUBLE_EDGE,
+        'simple_heavy':            box.SIMPLE_HEAVY,            
+        'horizontals':             box.HORIZONTALS,              
+        'rounded':                 box.ROUNDED,
+    }
+    return str_to_constant.get(box_style, box.ROUNDED)
+
+def match_and_dispatch(component):
+    return component | match[
+        (Is[is_a_component[Text]], dispatch_rich_text),
+        (Is[is_a_component[Code]], dispatch_rich_syntax),
+        (Is[is_a_component[Style]], dispatch_rich_style),
+        (Is[is_a_component[Table]], dispatch_rich_table),
+        (Is[is_a_component[Column]], dispatch_rich_column),
+        (Is[is_a_component[Frame]], dispatch_rich_panel),
+        (Any, lambda x: print(f"Show not defined for {x}"))
+    ] | collect
+
+def print_rich(displayable):
+    import rich
+    console = rich.console.Console()
+    displayable = match_and_dispatch(displayable)
+    console.print(displayable)
+
+show = run[print_rich]
\ No newline at end of file
diff --git a/python/zef/gql/simplegql/__main__.py b/python/zef/gql/simplegql/__main__.py
index e2c85e11..10c66e00 100644
--- a/python/zef/gql/simplegql/__main__.py
+++ b/python/zef/gql/simplegql/__main__.py
@@ -80,6 +80,9 @@
 
 if args.init_hook is not None:
     g_schema = Graph(root)
+    if args.init_hook not in now(g_schema):
+        log.error("Could not find init hook on schema graph", init_hook=args.init_hook)
+        raise SystemExit(4)
     hook_func = g_schema | now | get[args.init_hook] | collect
     hook_func(g_data)
     log.info(f"Ran init hook: {args.init_hook}")
@@ -94,7 +97,7 @@
 except KeyboardInterrupt:
     pass
 finally:
-    Effect({
+    {
         "type": FX.HTTP.StopServer,
         "server_uuid": server_uuid
-    }) | run
\ No newline at end of file
+    } | run
\ No newline at end of file
diff --git a/python/zef/gql/simplegql/generate_api.py b/python/zef/gql/simplegql/generate_api.py
index 7fccbbe2..ac297b49 100644
--- a/python/zef/gql/simplegql/generate_api.py
+++ b/python/zef/gql/simplegql/generate_api.py
@@ -279,7 +279,7 @@ def generate_resolvers_fcts(schema_root):
 
     # Always generate the Int scalar type 
     int_type = schema_root | Outs[RT.GQL_CoreScalarType][AET.Int] | single | collect
-    if int_type not in [rae_type(x) for x in extra_filters.keys()]:
+    if rae_type(int_type) not in [rae_type(x) for x in extra_filters.keys()]:
         extra_filters[int_type] = schema_generate_scalar_filter(int_type)
 
     query_fields = '\n\t'.join(query_fields)
@@ -312,11 +312,17 @@ def generate_resolvers_fcts(schema_root):
 # * Schema specific parts
 #----------------------------------------------
 
-def schema_generate_list_params(z_type, extra_filters):
-    name = z_type | F.Name | collect
+def schema_generate_type_dispatch(z_type, extra_filters):
     if z_type not in extra_filters:
         extra_filters[z_type] = None
-        extra_filters[z_type] = schema_generate_type_filter(z_type, extra_filters)
+        if op_is_scalar(z_type):
+            extra_filters[z_type] = schema_generate_scalar_filter(z_type)
+        else:
+            extra_filters[z_type] = schema_generate_type_filter(z_type, extra_filters)
+
+def schema_generate_list_params(z_type, extra_filters):
+    name = z_type | F.Name | collect
+    schema_generate_type_dispatch(z_type, extra_filters)
 
     query_params = [
         f"filter: {name}Filter",
@@ -359,12 +365,7 @@ def schema_generate_type_filter(z_type, extra_filters):
             filter_name = Boolean
         else:
             filter_name = f"{field_type_name}Filter"
-            if field_type not in extra_filters:
-                extra_filters[field_type] = None
-                if op_is_scalar(field_type):
-                    extra_filters[field_type] = schema_generate_scalar_filter(field_type)
-                else:
-                    extra_filters[field_type] = schema_generate_type_filter(field_type, extra_filters)
+            schema_generate_type_dispatch(field_type, extra_filters)
 
         if op_is_list(field):
             fields += [f"{field_name}: {filter_name}List"]
@@ -444,7 +445,8 @@ def schema_generate_scalar_filter(z_node):
 \tmax: {type_name}!
 }}"""
 
-    return {"schema": schema}
+    return {"schema": schema,
+            "orderable": op_is_orderable(z_node)}
 
 ####################################
 # * Query resolvers
@@ -854,7 +856,7 @@ def maybe_paginate_result(opts, first=None, offset=None):
 # ** Resolution
 
 @func
-def internal_resolve_field(z, info, z_field):
+def internal_resolve_field(z, info, z_field, auth_required=True):
     # This returns a LazyValue so we can deal with whatever comes out without
     # instantiating the whole list.
 
@@ -894,8 +896,8 @@ def internal_resolve_field(z, info, z_field):
     else:
         raise Exception(f"Don't know how to resolve this field: {z_field}")
 
-    # Auth filtering todo
-    opts = opts | filter[pass_query_auth[target(z_field)][info]]
+    if auth_required:
+        opts = opts | filter[pass_query_auth[target(z_field)][info]]
 
     # We must convert final objects from AEs to python types.
     # if z_field | target | is_core_scalar | collect:
@@ -1014,7 +1016,7 @@ def add_new_entity(info, type_node, params, name_gen):
                 l = [val]
 
             for item in l:
-                obj,obj_actions,obj_post_checks = find_or_add_entity(item, info, target(z_field), params, name_gen)
+                obj,obj_actions,obj_post_checks = find_or_add_entity(item, info, target(z_field), name_gen)
                 actions += obj_actions
                 post_checks += obj_post_checks
                 if z_field | op_is_incoming | collect:
@@ -1024,9 +1026,14 @@ def add_new_entity(info, type_node, params, name_gen):
 
     return this, actions, post_checks
 
-def find_or_add_entity(val, info, target_node, params, name_gen):
+def find_or_add_entity(val, info, target_node, name_gen):
     if isinstance(val, dict) and val.get("id", None) is not None:
+        # There should be no other fields given for this entity, otherwise the meaning is unclear.
+        if set(val.keys()) != {"id"}:
+            raise ExternalError("An entity should be designated with either an 'id' or a set of new fields to create a new entity.")
         obj = find_existing_entity_by_id(info, target_node, val["id"])
+        if obj is None:
+            raise ExternalError(f"Unable to find entity of kind '{target_node | F.Name | collect}' with id '{val['id']}'.")
         return obj,[],[]
     else:
         obj_name,actions,post_checks = add_new_entity(info, target_node, val, name_gen)
@@ -1066,27 +1073,19 @@ def update_entity(z, info, type_node, set_d, remove_d, name_gen):
             raise Exception(f"Updating list things is a todo (for z_field={z_field})")
         else:
             if op_is_unique(z_field):
-                et = ET(type_node | Out[RT.GQL_Delegate] | collect)
-                others = info.context["g"] | now | all[et] | filter[fvalue[rt][None] | equals[val]] | collect
-                others = set(others) - {z}
-                if len(others) > 0:
-                    log.error("Trying to modify entity with unique field that conflicts with others", z=z, et=et, field=field_name, others=others)
-                    raise ExternalError(f"Unique field '{field_name}' conflicts with existing items.")
+                # This used to be checked directly, here but just in case
+                # there's a mutation that renames two things at the same time,
+                # we change this up a bit.
+                post_checks += [("unique", z_field, type_node)]
 
-            if op_is_incoming(z_field):
-                maybe_prior_rel = z | in_rels[rt] | optional | collect
-            else:
-                maybe_prior_rel = z | out_rels[rt] | optional | collect
             if z_field | target | op_is_scalar | collect:
-                if maybe_prior_rel is None:
-                    if op_is_incoming(z_field):
-                        actions += [(val, rt, z)]
-                    else:
-                        actions += [(z, rt, val)]
-                else:
-                    actions += [(maybe_prior_rel | target | assign[val] | collect)]
+                actions += [z | set_field[rt][val][op_is_incoming(z_field)]]
             else:
-                raise Exception("Updating non-scalars is TODO")
+                found_z,new_actions,new_post_checks = find_or_add_entity(val, info, target(z_field), name_gen)
+                actions += new_actions
+                post_checks += new_post_checks
+
+                actions += [z | set_field[rt][found_z][op_is_incoming(z_field)]]
     
     for key,val in remove_d.items():
         if val is not None:
@@ -1103,6 +1102,11 @@ def update_entity(z, info, type_node, set_d, remove_d, name_gen):
             rels = z | out_rels[rt]
         actions += [terminate(rel) for rel in rels]
 
+        # TODO: Add a post-check here that this would not remove a relation on a
+        # type that has a required relation. Note that a new relation could be
+        # created in the same transaction, so it needs to be post, not directly
+        # here.
+
         # Also delete scalars
         if z_field | target | op_is_scalar | collect:
             actions += [terminate(target(rel)) for rel in rels]
@@ -1215,11 +1219,15 @@ def commit_with_post_checks(actions, post_checks, info):
             r = actions | transact[g] | run
             # Test all post checks
             for (kind,obj,type_node) in post_checks:
-                if type(obj) == str:
-                    obj = r[obj]
-                assert type(obj) == ZefRef
-                obj = obj | in_frame[g | now | collect][allow_tombstone] | collect
-                type_name = {type_node | fvalue[RT.Name][''] | collect}
+                type_name = type_node | fvalue[RT.Name][''] | collect
+
+                if kind in ["unique"]:
+                    pass
+                else:
+                    if type(obj) == str:
+                        obj = r[obj]
+                    assert type(obj) == ZefRef
+                    obj = obj | in_frame[g | now | collect][allow_tombstone] | collect
 
                 if kind == "add":
                     for z_func in type_node | Outs[RT.OnCreate]:
@@ -1243,6 +1251,22 @@ def commit_with_post_checks(actions, post_checks, info):
                             func[z_func](obj)
                         except:
                             raise ExternalError(f"OnRemove hook for type_node of '{type_name}' threw an exception")
+                elif kind == "unique":
+                    print("Doing a unqiue check for", obj, type_name)
+                    # In this case, obj is the field which must be unique
+                    z_field = obj
+                    assert op_is_unique(z_field)
+                    # Get all values - note this is not filtered by the user's
+                    # viewpoint, so we need to be a little careful.
+                    ents = g | now | all[ET(type_node | Out[RT.GQL_Delegate] | collect)]
+                    vals = ents | map[internal_resolve_field[info][z_field][False]] | concat | collect
+
+                    dis = distinct(vals)
+                    print(vals)
+                    print(dis)
+                    if len(dis) != len(vals):
+                        log.error("Non-unique values", vals=set(vals) - set(dis))
+                        raise ExternalError(f"Non-unique values found for field '{z_field | F.Name | collect}' of type_node '{type_name}'")
 
 
         except Exception as exc:
diff --git a/python/zef/gql/simplegql/schema_file_parser.py b/python/zef/gql/simplegql/schema_file_parser.py
index 421e1d52..32cdd050 100644
--- a/python/zef/gql/simplegql/schema_file_parser.py
+++ b/python/zef/gql/simplegql/schema_file_parser.py
@@ -280,7 +280,9 @@ def name_to_raet(name):
                     actions += [(Z[type_name], RT(field_name[1:]), field)]
                 elif field_name in ["_OnCreate", "_OnRemove", "_OnUpdaate"]:
                     # Find the zef function that this corresponds to
-                    z_func = g[field]
+                    if field not in now(g):
+                        raise Exception(f"Hook named {field} not found on schema graph")
+                    z_func = g | now | get[field] | collect
                     actions += [(Z[type_name], RT(field_name[1:]), z_func)]
 
                 elif field_name == "_Upfetch":
@@ -305,7 +307,9 @@ def name_to_raet(name):
 
                 if "dynamic" in field:
                     assert "relation" not in field, f"Can't give a relation with a dynamically resolved field: {field}"
-                    z_func = g[field["dynamic"]]
+                    if field["dynamic"] not in now(g):
+                        raise Exception(f"Hook named {field['dynamic']} not found on schema graph")
+                    z_func = g | now | get[field["dynamic"]] | collect
                     actions += [(Z[qual_name], RT.GQL_FunctionResolver, z_func)]
                     del field["dynamic"]
                 else:
diff --git a/python/zef/gql/simplegql/server.py b/python/zef/gql/simplegql/server.py
index 529de753..ec01a3a0 100644
--- a/python/zef/gql/simplegql/server.py
+++ b/python/zef/gql/simplegql/server.py
@@ -62,7 +62,10 @@ def resolve_token(context, request):
     else:
         raise Exception("Shouldn't get here")
 
-    if auth_result["aud"] != aud:
+    if isinstance(auth_result["aud"], list):
+        if aud not in auth_result["aud"]:
+            raise Exception("Invalid token for wrong audience")
+    elif auth_result["aud"] != aud:
         raise Exception("Invalid token for wrong audience")
 
     if namespace is None:
@@ -137,7 +140,7 @@ def start_server(z_gql_root,
         context["jwk_client"] = PyJWKClient(url)
 
 
-    http_r = Effect({
+    http_r = {
         'type': FX.HTTP.StartServer,
         'port': port,
         'pipe_into': (map[middleware_worker[permit_cors,
@@ -147,7 +150,7 @@ def start_server(z_gql_root,
                       | subscribe[run]),
         'logging': logging,
         'bind_address': bind_address,
-    }) | run
+    } | run
     if is_a(http_r, Error):
         raise Exception("Error in creating server") from http_r.args[0]
 
diff --git a/python/zef/ops/privileges.py b/python/zef/ops/privileges.py
index 1c41943a..c9ea6746 100644
--- a/python/zef/ops/privileges.py
+++ b/python/zef/ops/privileges.py
@@ -47,11 +47,11 @@ def privileges_implementation(user, action, privilege, target):
     if typ is None:
         raise Exception(f"Unknown (action,privilege) combination {(action,privilege)}")
 
-    return Effect({
+    return {
         "type": typ,
         "user": user,
         "target": target
-    })
+    }
 
 def privileges_type_info(op, curr_type):
     return VT.Effect
@@ -63,8 +63,8 @@ def privileges_type_info(op, curr_type):
 grant = privileges[KW.grant]
 revoke = privileges[KW.revoke]
 
-login = Effect({"type": FX.ZefHub.Login})
-login_as_guest = Effect({"type": FX.ZefHub.Login, "auth_key": "GUEST"})
-logout = Effect({"type": FX.ZefHub.Logout})
+login = {"type": FX.ZefHub.Login}
+login_as_guest = {"type": FX.ZefHub.Login, "auth_key": "GUEST"}
+logout = {"type": FX.ZefHub.Logout}
 
 from ..pyzef.internals import who_am_i
\ No newline at end of file
diff --git a/python/zef/ops/transactor_role.py b/python/zef/ops/transactor_role.py
index dd708fd7..ebddeedd 100644
--- a/python/zef/ops/transactor_role.py
+++ b/python/zef/ops/transactor_role.py
@@ -24,13 +24,14 @@
 from ..core.op_implementations.dispatch_dictionary import _op_to_functions
 
 def take_transactor_role_implementation(g):
-    return Effect({"type": FX.Graph.TakeTransactorRole,
-                   "graph": g})
+    return {"type": FX.Graph.TakeTransactorRole, "graph": g}
+    
 def take_transactor_role_tp(x):
     return VT.Any
+
 def release_transactor_role_implementation(g):
-    return Effect({"type": FX.Graph.ReleaseTransactorRole,
-                   "graph": g})
+    return {"type": FX.Graph.ReleaseTransactorRole, "graph": g}
+
 def release_transactor_role_tp(x):
     return VT.Any
 
diff --git a/scripts/create_tokens_files.py b/scripts/create_tokens_files.py
index 3ab74904..d3140c64 100644
--- a/scripts/create_tokens_files.py
+++ b/scripts/create_tokens_files.py
@@ -15,6 +15,7 @@
 import os
 os.environ["ZEFDB_DEVELOPER_EARLY_TOKENS"] = "1"
 os.environ["ZEFDB_OFFLINE_MODE"] = "TRUE"
+os.environ["ZEFDB_TOKENS_CACHE_PATH"] = ""
 
 from zef import *
 from zef.ops import *


Test patch:
diff --git a/python/tests/test_attach.py b/python/tests/test_attach.py
index aa5a6ef3..a8ae3554 100644
--- a/python/tests/test_attach.py
+++ b/python/tests/test_attach.py
@@ -72,6 +72,28 @@ def test_set_field(self):
         self.assertEqual(r["z"] | now | Out[RT.Supervisor] | collect, r3["new"])
 
 
+    def test_set_field_reversed(self):
+        g = Graph()
+
+        z = ET.Machine | g | run
+        z | set_field[RT.Something][42][True] | g | run
+
+        self.assertEqual(z | now | in_rels[RT.Something] | length | collect, 1)
+        self.assertEqual(z | now | In[RT.Something] | value | collect, 42)
+
+        z | set_field[RT.Something][43][True] | g | run
+
+        self.assertEqual(z | now | In[RT.Something] | value | collect, 43)
+
+        z2 = ET.Machine | g | run
+        z | set_field[RT.Entity][z2][True] | g | run
+
+        self.assertEqual(z | now | In[RT.Entity] | collect, now(z2))
+
+        z3 = ET.Machine | g | run
+        z | set_field[RT.Entity][z3][True] | g | run
+        self.assertEqual(z | now | In[RT.Entity] | collect, now(z3))
+
 
     def test_shorthand(self):
         g = Graph()
diff --git a/python/tests/test_http.py b/python/tests/test_http.py
index 0777f22f..b15f13ed 100644
--- a/python/tests/test_http.py
+++ b/python/tests/test_http.py
@@ -26,6 +26,10 @@ def test_http_server(self):
             # We hope that this is not taken by anything else for this test...
             'port': port,
             'pipe_into': (map[middleware_worker[permit_cors,
+                                                route["/fixed-url"][lambda query: {"type": FX.HTTP.SendResponse,
+                                                                                   "server_uuid": query["server_uuid"],
+                                                                                   "request_id": query["request_id"],
+                                                                                   "response": "Fixed"}],
                                                 route["/test-url"][lambda query: {**query, "response_body": payload}],
                                                 fallback_not_found,
                                                 send_response]]
@@ -40,6 +44,10 @@ def test_http_server(self):
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.content, payload)
 
+        r = requests.get(f"http://localhost:{port}/fixed-url")
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.content, b"Fixed")
+
         r = requests.get(f"http://localhost:{port}")
         self.assertEqual(r.status_code, 404)
 
diff --git a/python/tests/test_simplegql.py b/python/tests/test_simplegql.py
index 7bfb087a..abf8099e 100644
--- a/python/tests/test_simplegql.py
+++ b/python/tests/test_simplegql.py
@@ -51,6 +51,19 @@ def setUp(self):
   user: User @relation(rt: "User") @search
   amount: Int @search
   date: DateTime @search
+  category: Category
+}
+
+type Category {
+  transactions: [Transaction] @relation(rt: "Category") @incoming
+  name: String @unique
+
+  testListScalar: [String]
+  testListEntity: [Simple]
+}
+
+type Simple {
+  name: String
 }
 
 enum TestEnum {
@@ -79,10 +92,10 @@ def dynamicHook(z, info):
         self.server_uuid = start_server(root, g_data, self.port, "localhost", logging=False)
 
     def tearDown(self):
-        Effect({
+        {
             "type": FX.HTTP.StopServer,
             "server_uuid": self.server_uuid,
-        }) | run
+        } | run
 
     def test_simplegql(self):
         import jwt
@@ -161,10 +174,8 @@ def assert_no_error(r):
         assert_no_error(r)
         user2_id = r.json()["data"]["addUser"]["user"][0]["id"]
 
-        if False:
-            # This needs to be implemented!
-            r = do_query(jwt_user1, 'mutation { updateTransaction(input: {filter: {id: "' + trans_id + '"}, set: {user: {id: "' + user2_id + '"}}}) { transaction { id } } }')
-            assert_error_with(r, "TODO")
+        r = do_query(jwt_user1, 'mutation { updateTransaction(input: {filter: {id: "' + trans_id + '"}, set: {user: {id: "' + user2_id + '"}}}) { transaction { id } } }')
+        assert_error_with(r, "Unable to find entity")
 
         # Test updating and removing things
         r = do_query(jwt_user1, '''mutation {
@@ -203,6 +214,46 @@ def assert_no_error(r):
         self.assertEqual(len(r.json()["data"]["queryUser"]), 1)
         self.assertEqual(len(r.json()["data"]["queryUser"][0]["transactions"]), 0)
 
+        # Adding Category explicitly and updating transaction
+        r = do_query(jwt_user1, 'mutation { addCategory(input: {name: "explicit create"}) { category { id } } }')
+        assert_no_error(r)
+        cat_id = r.json()["data"]["addCategory"]["category"][0]["id"]
+
+        r = do_query(jwt_user1, 'mutation { updateTransaction(input: {filter: {id: "' + trans_id + '"}, set: {category: {id: "' + cat_id + '"} } }) { count } }')
+        assert_no_error(r)
+
+        # Adding category during update to transaction
+        r = do_query(jwt_user1, 'mutation { updateTransaction(input: {filter: {id: "' + trans_id + '"}, set: {category: {name: "on the fly"} } }) { transaction { category { id name } } } }')
+        assert_no_error(r)
+        cat_data = r.json()["data"]["updateTransaction"]["transaction"][0]["category"]
+        self.assertNotEqual(cat_data["id"], cat_id)
+        self.assertEqual(cat_data["name"], "on the fly")
+
+        # Unique checks
+        r = do_query(jwt_user1, 'mutation { addCategory(input: {name: "soon to be duplicated"}) { category { id } } }')
+        assert_no_error(r)
+
+        r = do_query(jwt_user1, 'mutation { updateCategory(input: {filter: {id: "' + cat_id + '"}, set: {name: "soon to be duplicated"} }) { category { id name } } }')
+        assert_error_with(r, "Non-unique values")
+
+
+        # List creation
+        r = do_query(jwt_user1, 'mutation { addCategory(input: {testListScalar: ["a", "b", "c"], testListEntity: [{name: "a"}, {name: "b"}]}) { category { id testListScalar testListEntity { name } } } }')
+        assert_no_error(r)
+        cat_data = r.json()["data"]["addCategory"]["category"][0]
+        cat_list_id = cat_data["id"]
+        self.assertSetEqual(set(cat_data["testListScalar"]), {"a", "b", "c"})
+        expected = [{"name": "a"}, {"name": "b"}]
+        for x in cat_data["testListEntity"]:
+            self.assertIn(x, expected)
+        for x in expected:
+            self.assertIn(x, cat_data["testListEntity"])
+
+        # TODO: List set
+        # TODO: List manipulation
+        # TODO: List remove
+
+
 
 if __name__ == '__main__':
     unittest.main()