diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 3a1c7b75..0ba07b59 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -85,6 +85,13 @@ jobs:
          ./slua tests/conformance/assert.luau
          ./slua-analyze tests/conformance/assert.luau
          ./slua-compile tests/conformance/assert.luau
+    # ServerLua: scoped to SLExecutor since TSan is slow and the threading lives there
+    - name: run SLExecutor tests under TSan
+      if: matrix.os.name == 'ubuntu'
+      run: |
+        make -j4 config=tsan slua-tests
+        ./build/tsan/slua-tests -ts=SLExecutor
+        ./build/tsan/slua-tests -ts=SLExecutor --fflags=true
 
   windows:
     runs-on: windows-2022
diff --git a/.github/workflows/build_release.yml b/.github/workflows/build_release.yml
index a1325101..ee935917 100644
--- a/.github/workflows/build_release.yml
+++ b/.github/workflows/build_release.yml
@@ -19,7 +19,7 @@ jobs:
     - name: configure
       run: cmake . -DCMAKE_BUILD_TYPE=Release
     - name: build
-      run: cmake --build . --target Luau.Repl.CLI Luau.Analyze.CLI Luau.Compile.CLI Luau.Ast.CLI --config Release -j 2
+      run: cmake --build . --target Luau.Repl.CLI Luau.Analyze.CLI Luau.Compile.CLI Luau.Ast.CLI Luau.Harness.CLI --config Release -j 2
     - name: pack
       if: matrix.os.name != 'windows'
       run: zip slua-${{github.event.release.tag_name}}-${{matrix.os.name}}.zip slua*
diff --git a/CLI/src/Harness.cpp b/CLI/src/Harness.cpp
index 0f085e89..b8da56b0 100644
--- a/CLI/src/Harness.cpp
+++ b/CLI/src/Harness.cpp
@@ -91,6 +91,194 @@ static void log_to_stderr(LogLevel level, const char* source, const char* messag
     fprintf(stderr, "[%s] %s: %s\n", level_names[(int)level], source, message);
 }
 
+// Drives the script through run windows until it completes, faults, or
+// refuses. `lsl_state` is left at whatever state it ended in so a later
+// dispatch lands in the right handler table.
+static RunResult run_to_completion(Script& script, double quanta, bool is_lsl, double& accum_sleep, size_t& slices, int& lsl_state)
+{
+    // state_entry is implicit in Lua, but LSL needs it specifically dispatched.
+    bool dispatch_state_entry = is_lsl;
+    RunResult result;
+    for (;;)
+    {
+        // Fake the sleep, just collect it so we know the accumulated
+        // sleep across the entire script run.
+        if (script.getSleep() > 0.0f)
+        {
+            accum_sleep += script.getSleep();
+            script.setSleep(0.0f);
+        }
+
+        {
+            RunWindow window(script, quanta);
+            if (dispatch_state_entry)
+            {
+                result = script.callEventHandler(lsl_state, "state_entry", nullptr);
+                dispatch_state_entry = false;
+            }
+            else
+            {
+                result = script.resumeEventHandler();
+            }
+        }
+        ++slices;
+
+        if (result.status == HandlerRunStatus::Preempted)
+            continue;
+        if (result.status == HandlerRunStatus::StateChange)
+        {
+            // TODO: Do state_exit too... meh.
+            lsl_state = result.newState;
+            dispatch_state_entry = true;
+            continue;
+        }
+
+        // Anything else means we're done.
+        break;
+    }
+
+    // Bank sleep the final slice left behind
+    if (script.getSleep() > 0.0f)
+        accum_sleep += script.getSleep();
+
+    return result;
+}
+
+// Prints why a run stopped early, for the statuses that mean the script is
+// done for good. Returns the process exit code.
+static int report_failure(Script& script, const RunResult& result)
+{
+    if (result.status == HandlerRunStatus::Fault)
+    {
+        auto& fault_str = script.getExtendedFaultString().empty() ? script.getFaultString() : script.getExtendedFaultString();
+        fprintf(stderr, "Fault: %s\n", fault_str.c_str());
+        return 1;
+    }
+    if (result.status == HandlerRunStatus::Refused)
+    {
+        fprintf(stderr, "Error: engine refused to run the handler\n");
+        return 1;
+    }
+    return 0;
+}
+
+static void print_watchdog_stats(const IProvisioner& provisioner)
+{
+    // Resident never fires or wakes, so this only shows for the deadline installers
+    WatchdogStats wd_stats = provisioner.getWatchdogStats();
+    if (wd_stats.fires == 0 && wd_stats.wakes == 0)
+        return;
+
+    double avg = wd_stats.fires > 0 ? wd_stats.latenessSum / (double)wd_stats.fires : 0.0;
+    fprintf(
+        stderr,
+        "Watchdog lead: %.1f usecs, wakes: %llu, fires: %llu, late fires: %llu, lateness usecs min/avg/max: %.1f/%.1f/%.1f\n",
+        wd_stats.fireLead * 1e6,
+        (unsigned long long)wd_stats.wakes,
+        (unsigned long long)wd_stats.fires,
+        (unsigned long long)wd_stats.lateFires,
+        wd_stats.latenessMin * 1e6,
+        avg * 1e6,
+        wd_stats.latenessMax * 1e6
+    );
+}
+
+struct WindowTiming
+{
+    // Windows actually run, short of the request only when the body bailed
+    size_t completed = 0;
+    double total = 0.0;
+    double min = 0.0;
+    double max = 0.0;
+};
+
+// Runs `body` `count` times, timing each. The body returns false to stop.
+template<typename Body>
+static WindowTiming time_windows(size_t count, Body&& body)
+{
+    WindowTiming timing;
+    double phase_start = lua_clock();
+    for (size_t i = 0; i < count; ++i)
+    {
+        double window_start = lua_clock();
+        bool keep_going = body();
+        double window_time = lua_clock() - window_start;
+
+        ++timing.completed;
+        if (i == 0 || window_time < timing.min)
+            timing.min = window_time;
+        if (window_time > timing.max)
+            timing.max = window_time;
+        if (!keep_going)
+            break;
+    }
+    timing.total = lua_clock() - phase_start;
+    return timing;
+}
+
+static void print_window_timing(const char* label, const WindowTiming& timing)
+{
+    double avg = timing.completed > 0 ? timing.total / (double)timing.completed : 0.0;
+    fprintf(
+        stderr,
+        "%s windows: %zu, total %.3fs, per window usecs avg/min/max: %.3f/%.1f/%.1f\n",
+        label,
+        timing.completed,
+        timing.total,
+        avg * 1e6,
+        timing.min * 1e6,
+        timing.max * 1e6
+    );
+}
+
+// Opens and closes `count` windows twice over: once empty, so the installer
+// and GC bookkeeping are all that's measured, then once dispatching a
+// handler each time. The difference is the Lua dispatch cost.
+static int run_window_bench(Script& script, double quanta, size_t count, int lsl_state)
+{
+    WindowTiming empty = time_windows(count, [&]() {
+        RunWindow window(script, quanta);
+        return true;
+    });
+    print_window_timing("Empty", empty);
+
+    // A deadline install can land at the first safepoint when the fire lead
+    // exceeds the quanta, so a preempted handler is resumed in the next window
+    // rather than treated as a failure.
+    bool resuming = false;
+    RunResult result;
+    WindowTiming handler = time_windows(count, [&]() {
+        RunWindow window(script, quanta);
+        if (resuming)
+            result = script.resumeEventHandler();
+        else
+            script.callEventHandler(lsl_state, "touch_start", [](lua_State* L, void *ctx)
+            {
+                lua_pushnumber(L, 0);
+            });
+        resuming = result.status == HandlerRunStatus::Preempted;
+        return resuming || result.status == HandlerRunStatus::Ok;
+    });
+    print_window_timing("Handler", handler);
+
+    // The script can't be torn down with a handler still staged
+    while (resuming)
+    {
+        RunWindow window(script, quanta);
+        result = script.resumeEventHandler();
+        resuming = result.status == HandlerRunStatus::Preempted;
+    }
+
+    // A missing state_entry is fine for a normal run, but here it means the
+    // user's script has nothing to dispatch.
+    if (result.status == HandlerRunStatus::NotRun)
+    {
+        fprintf(stderr, "Error: script has no touch_start handler to benchmark\n");
+        return 1;
+    }
+    return report_failure(script, result);
+}
+
 static void displayHelp(const char* argv0)
 {
     printf("Usage: %s [options] script\n", argv0);
@@ -100,15 +288,28 @@ static void displayHelp(const char* argv0)
     printf("\n");
     printf("Options:\n");
     printf("  --quanta=<usecs>: time slice per run window (default 200)\n");
+    printf("  --window-bench=<n>: after the script completes, open and close n run\n"
+           "                 windows, empty and then dispatching its touch_start handler,\n"
+           "                 to measure the per-window overhead\n");
+    printf("  --fire-lead=<usecs>: how early the threaded or signal installer puts the\n"
+           "                 interrupt handler in ahead of the deadline (default: per policy)\n");
     printf("  -O<n>: compile with optimization level n (default 1)\n");
     printf("  --fflags=<list>: comma-separated fast flag settings (name=true/false),\n");
+    printf("  --use-lua-clock: Use lua_clock() instead of a specialized quanta timer\n");
+    printf("  --interrupt=<resident|threaded|signal>: keep the interrupt handler resident,\n"
+           "                 have the watchdog thread install it at the deadline, or have a\n"
+           "                 POSIX timer signal install it (Linux only)\n");
 }
 
 int main(int argc, char** argv)
 {
     const char* script_path = nullptr;
     double quanta_usec = 200.0;
+    double fire_lead_usec = 0.0;
+    size_t window_bench = 0;
+    bool use_lua_clock = false;
     int optimization_level = 1;
+    InterruptInstallPolicy interrupt_policy = InterruptInstallPolicy::Default;
 
     for (int i = 1; i < argc; ++i)
     {
@@ -126,10 +327,54 @@ int main(int argc, char** argv)
                 return 1;
             }
         }
+        else if (strncmp(argv[i], "--fire-lead=", 12) == 0)
+        {
+            fire_lead_usec = atof(argv[i] + 12);
+            if (fire_lead_usec <= 0.0)
+            {
+                fprintf(stderr, "Error: --fire-lead must be a positive number of usecs.\n");
+                return 1;
+            }
+        }
+        else if (strncmp(argv[i], "--window-bench=", 15) == 0)
+        {
+            long long count = atoll(argv[i] + 15);
+            if (count <= 0)
+            {
+                fprintf(stderr, "Error: --window-bench must be a positive number of windows.\n");
+                return 1;
+            }
+            window_bench = (size_t)count;
+        }
         else if (strncmp(argv[i], "--fflags=", 9) == 0)
         {
             setLuauFlags(argv[i] + 9);
         }
+        else if (strncmp(argv[i], "--interrupt=", 12) == 0)
+        {
+            const char* value = argv[i] + 12;
+            if (strcmp(value, "resident") == 0)
+            {
+                interrupt_policy = InterruptInstallPolicy::Resident;
+            }
+            else if (strcmp(value, "threaded") == 0)
+            {
+                interrupt_policy = InterruptInstallPolicy::Threaded;
+            }
+            else if (strcmp(value, "signal") == 0)
+            {
+                interrupt_policy = InterruptInstallPolicy::Signal;
+            }
+            else
+            {
+                fprintf(stderr, "Error: invalid --interrupt value.\n");
+                return 1;
+            }
+        }
+        else if (strcmp(argv[i], "--use-lua-clock") == 0)
+        {
+            use_lua_clock = true;
+        }
         else if (strncmp(argv[i], "-O", 2) == 0)
         {
             int level = atoi(argv[i] + 2);
@@ -199,7 +444,13 @@ int main(int argc, char** argv)
     HostCallbacks callbacks;
     callbacks.clockProvider = script_clock;
     callbacks.populateEnvironment = populate_environment;
-    // quantaClockProvider stays null so we exercise the engine's default
+    callbacks.interruptInstallPolicy = interrupt_policy;
+    callbacks.interruptFireLead = fire_lead_usec * 1e-6;
+    if (use_lua_clock)
+    {
+        // Leaving this null uses a platform-optimized quanta clock provider
+        callbacks.quantaClockProvider = lua_clock;
+    }
 
     Provisioner<> provisioner(callbacks);
 
@@ -239,67 +490,16 @@ int main(int argc, char** argv)
     double quanta = quanta_usec * 1e-6;
     double accum_sleep = 0.0;
     size_t slices = 0;
-    double start = lua_clock();
-
-    // state_entry is implicit in Lua, but LSL needs it specifically dispatched.
-    bool dispatch_state_entry = is_lsl;
     int lsl_state = 0;
-    RunResult result;
-    for (;;)
-    {
-        // Fake the sleep, just collect it so we know the accumulated
-        // sleep across the entire script run.
-        if (script->getSleep() > 0.0f)
-        {
-            accum_sleep += script->getSleep();
-            script->setSleep(0.0f);
-        }
-
-        script->beginRunWindow(quanta);
-        if (dispatch_state_entry)
-        {
-            result = script->callEventHandler(lsl_state, "state_entry", nullptr);
-            dispatch_state_entry = false;
-        }
-        else
-        {
-            result = script->resumeEventHandler();
-        }
-        script->endRunWindow();
-        ++slices;
-
-        if (result.status == HandlerRunStatus::Preempted)
-            continue;
-        if (result.status == HandlerRunStatus::StateChange)
-        {
-            // TODO: Do state_exit too... meh.
-            lsl_state = result.newState;
-            dispatch_state_entry = true;
-            continue;
-        }
-
-        // Anything else means we're done.
-        break;
-    }
-
-    // Bank sleep the final slice left behind
-    if (script->getSleep() > 0.0f)
-        accum_sleep += script->getSleep();
-
+    double start = lua_clock();
+    RunResult result = run_to_completion(*script, quanta, is_lsl, accum_sleep, slices, lsl_state);
     double runtime = lua_clock() - start;
     fprintf(stderr, "Runtime: %f, Accum. Sleep: %f, Time Slices: %zu\n", runtime, accum_sleep, slices);
 
-    if (result.status == HandlerRunStatus::Fault)
-    {
-        auto &fault_str = script->getExtendedFaultString().empty() ? script->getFaultString() : script->getExtendedFaultString();
-        fprintf(stderr, "Fault: %s\n", fault_str.c_str());
-        return 1;
-    }
-    if (result.status == HandlerRunStatus::Refused)
-    {
-        fprintf(stderr, "Error: engine refused to run the handler\n");
-        return 1;
-    }
+    int exit_code = report_failure(*script, result);
+    if (exit_code == 0 && window_bench > 0)
+        exit_code = run_window_bench(*script, quanta, window_bench, lsl_state);
 
-    return 0;
+    print_watchdog_stats(provisioner);
+    return exit_code;
 }
diff --git a/CMakeLists.txt b/CMakeLists.txt
index ce344864..fc5e21a8 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -14,6 +14,7 @@ option(LUAU_STATIC_CRT "Link with the static CRT (/MT)" OFF)
 option(LUAU_EXTERN_C "Use extern C for all APIs" OFF)
 option(LUAU_USE_TAILSLIDE "Link against tailslide for LSL support" OFF)
 option(LUAU_BUILD_SHARED "Build as a shared library" OFF)
+option(LUAU_TSAN "Build with ThreadSanitizer" OFF)
 
 if(LUAU_BUILD_SHARED AND NOT LUAU_EXTERN_C)
     message(FATAL_ERROR "LUAU_BUILD_SHARED requires LUAU_EXTERN_C to be ON")
@@ -40,6 +41,15 @@ endif()
 
 project(Luau LANGUAGES CXX C)
 
+if(LUAU_TSAN)
+    if(MSVC)
+        message(FATAL_ERROR "LUAU_TSAN requires a Clang or GCC toolchain")
+    endif()
+    add_compile_options(-fsanitize=thread)
+    set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fsanitize=thread")
+    set(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} -fsanitize=thread")
+endif()
+
 # Embed builtins.txt into a header file at configure time
 file(READ "${CMAKE_SOURCE_DIR}/builtins.txt" BUILTINS_HEX HEX)
 string(REGEX REPLACE "([0-9a-f][0-9a-f])" "0x\\1, " BUILTINS_HEX_ARRAY "${BUILTINS_HEX}")
@@ -311,6 +321,13 @@ if(CMAKE_SYSTEM_NAME MATCHES "Linux|Darwin|iOS")
     target_link_libraries(osthreads INTERFACE "-lpthread")
 endif ()
 
+# ServerLua: We use threads internally, library consumers need pthreads if applciable.
+target_link_libraries(Luau.Executor PUBLIC osthreads)
+# ServerLua: POSIX timers live in librt on glibc < 2.34, a stub elsewhere
+if(CMAKE_SYSTEM_NAME MATCHES "Linux")
+    target_link_libraries(Luau.Executor PUBLIC "-lrt")
+endif ()
+
 if(LUAU_BUILD_CLI)
     target_compile_options(Luau.Repl.CLI PRIVATE ${LUAU_OPTIONS})
     target_compile_options(Luau.Reduce.CLI PRIVATE ${LUAU_OPTIONS})
diff --git a/Common/include/Luau/Common.h b/Common/include/Luau/Common.h
index 4f086619..3f9082b5 100644
--- a/Common/include/Luau/Common.h
+++ b/Common/include/Luau/Common.h
@@ -3,6 +3,11 @@
 
 #include <string.h>
 
+// ServerLua: for the std::atomic_thread_fence fallback below
+#if defined(_MSC_VER) && !defined(__clang__)
+#include <atomic>
+#endif
+
 // Compiler codegen control macros
 #ifdef _MSC_VER
 #define LUAU_NORETURN __declspec(noreturn)
@@ -56,6 +61,31 @@
 namespace Luau
 {
 
+// ServerLua: TSan-visible accessors for fields the VM reads unsynchronized by
+// contract, like lua_Callbacks::interrupt. Same codegen as a plain access on
+// x86/ARM64 (the release store is stlr on ARM64).
+// clang-cl defines both _MSC_VER and __clang__, so guard on __clang__ too.
+template<typename T>
+inline T opaque_load(T* p)
+{
+#if defined(_MSC_VER) && !defined(__clang__)
+    return *static_cast<T volatile*>(p);
+#else
+    return __atomic_load_n(p, __ATOMIC_RELAXED);
+#endif
+}
+
+template<typename T>
+inline void release_store(T* p, T v)
+{
+#if defined(_MSC_VER) && !defined(__clang__)
+    std::atomic_thread_fence(std::memory_order_release);
+    *static_cast<T volatile*>(p) = v;
+#else
+    __atomic_store_n(p, v, __ATOMIC_RELEASE);
+#endif
+}
+
 using AssertHandler = int (*)(const char* expression, const char* file, int line, const char* function);
 
 inline AssertHandler& assertHandler()
@@ -88,6 +118,10 @@ LUAU_NOINLINE inline int assertCallHandler(const char* expression, const char* f
 #define LUAU_ASSERT(expr) (void)sizeof(!!(expr))
 #endif
 
+// ServerLua: for invariants about the host process rather than our own code,
+// which only release builds will ever get to see
+#define LUAU_ASSERT_ALWAYS(expr) ((void)(!!(expr) || (Luau::assertCallHandler(#expr, __FILE__, __LINE__, __FUNCTION__) && (LUAU_DEBUGBREAK(), 0))))
+
 namespace Luau
 {
 
diff --git a/Executor/include/Luau/Executor.h b/Executor/include/Luau/Executor.h
index a4ecea6d..243fd6cd 100644
--- a/Executor/include/Luau/Executor.h
+++ b/Executor/include/Luau/Executor.h
@@ -1,6 +1,7 @@
 // ServerLua: per-script execution engine shared with the script host.
 #pragma once
 
+#include <atomic>
 #include <memory>
 #include <string>
 
@@ -32,6 +33,7 @@ enum class LogLevel : uint8_t
     Debug = 0,
     Info,
     Warn,
+    Error,
 };
 
 // Parameters that define the sealed image consumed by buildImage()
@@ -87,8 +89,95 @@ void logDebug(const char* source, const char* fmt, ...) LUA_PRINTF_ATTR(2, 3);
 void logInfo(const char* source, const char* fmt, ...) LUA_PRINTF_ATTR(2, 3);
 void logWarn(const char* source, const char* fmt, ...) LUA_PRINTF_ATTR(2, 3);
 
-// Figure out which clock source to use
-lua_clockProvider resolveDefaultQuantaClock();
+// Monotonic seconds for the per-safepoint elapsed check, which is the one
+// place clock cost shows up in throughput. Nothing else reads it.
+using QuantaClock = double (*)();
+
+QuantaClock resolveDefaultQuantaClock();
+
+using InterruptCallback = void (*)(lua_State* L, int gc);
+
+// Lateness is how far past the intended fire instant the handler actually got
+// installed.
+struct WatchdogStats
+{
+    uint64_t fires = 0;
+    // Fires that landed after the deadline itself, not just after the fire lead
+    uint64_t lateFires = 0;
+    // Watchdog thread wakeups, whether or not they led to a fire. Always zero
+    // for installers without a thread.
+    uint64_t wakes = 0;
+    double latenessSum = 0.0;
+    double latenessMin = 0.0;
+    double latenessMax = 0.0;
+    // The lead in effect, so a stats dump is self-describing
+    double fireLead = 0.0;
+};
+
+// Owns `cb.interrupt` for a run window and decides when it's installed.
+// The handler checks the clock itself, so the contract is a latest install
+// time, not an exact one. One per provisioner, one open window at a time.
+class InterruptInstaller
+{
+public:
+    explicit InterruptInstaller(InterruptCallback handler)
+        : mHandler(handler)
+    {
+    }
+
+    virtual ~InterruptInstaller();
+
+    // Install `cb.interrupt` no later than `seconds` from now. Installing it
+    // earlier is allowed.
+    virtual void installWithin(lua_Callbacks* target, double seconds) = 0;
+
+    // For things that need the script to yield immediately (sleep, force-yield).
+    // Harmless if it's already installed.
+    virtual void installNow() = 0;
+
+    // Forget the deadline. Nothing touches `target` after this returns.
+    virtual void cancel() = 0;
+
+    // From the handler when it had nothing to do: uninstall it if a deadline
+    // install is still pending to bring it back, otherwise leave it resident.
+    // Called at every safepoint while the handler is in early, so the common
+    // "nothing pending" answer stays inline.
+    void uninstallIfPending()
+    {
+        if (mPending.load(std::memory_order_relaxed))
+            uninstallPending();
+    }
+
+    // How far past the deadline the current window's handler went in, in
+    // seconds. Zero until it does, and zero again after the next installWithin().
+    virtual double getInstallOverrun() { return 0.0; }
+
+    virtual WatchdogStats getStats() { return {}; }
+
+protected:
+    virtual void uninstallPending() = 0;
+
+    InterruptCallback mHandler = nullptr;
+    // An installWithin() whose install hasn't landed yet
+    std::atomic<bool> mPending{false};
+};
+
+enum class InterruptInstallPolicy
+{
+    // Resident, or Threaded when the SLuaThreadedQuantaWatchdog fflag is on
+    Default,
+    // Handler stays resident and checks the clock at every safepoint
+    Resident,
+    // A watchdog thread installs the handler just ahead of the deadline
+    Threaded,
+    // A POSIX timer signals the script thread, which installs the handler in
+    // the signal handler. No second thread and no scheduling priority needed.
+    // Linux only, falls back to Threaded elsewhere.
+    Signal,
+};
+
+// Throws std::system_error if the threaded policy can't create its thread.
+std::unique_ptr<InterruptInstaller> createInterruptInstaller(InterruptInstallPolicy policy, InterruptCallback handler, double fireLead);
 
 // Give the embedder a chance to plop their own things into the environment before it's
 // fully set up. This is called before GC fixing / ares perms registration.
@@ -102,13 +191,24 @@ struct HostCallbacks
     lua_randomProvider randomProvider = nullptr;
     lua_setTimerEventCallback setTimerEventCb = nullptr;
     lua_eventHandlerRegistrationCallback eventHandlerRegistrationCb = nullptr;
-    lua_clockProvider quantaClockProvider = nullptr;
+    QuantaClock quantaClockProvider = nullptr;
+    // Resident is required when quantaClockProvider doesn't track real time
+    // (test fake clocks), since the deadline installers schedule on lua_clock()
+    InterruptInstallPolicy interruptInstallPolicy = InterruptInstallPolicy::Default;
+    // How early, in seconds, the Threaded and Signal policies put the handler in
+    // ahead of the deadline to cover delivery latency. Zero takes the policy's
+    // default. Size it from the harness's lateness stats on the target hardware.
+    double interruptFireLead = 0.0;
     PopulateEnvironmentCallback populateEnvironment = nullptr;
 };
 
 // An environment is... basically just a Lua VM with some particular settings. It's
 // intended for multiple of these to be able to be living at any given moment, one
 // for LSL, one for a particular version of the Lua API, etc.
+//
+// It is expected that any use of a given environment is _exclusive_ to a particular
+// thread at any given time, with no interleaved access. In almost all cases, it is
+// in fact exclusive to a particular thread for its lifetime.
 class IEnvironment
 {
 public:
@@ -340,6 +440,10 @@ class IProvisioner
 
     virtual const HostCallbacks& getCallbacks() const = 0;
 
+    // Engine-internal, for Script. Null until the first environment exists.
+    virtual InterruptInstaller* getInterruptInstaller() const = 0;
+    virtual WatchdogStats getWatchdogStats() const = 0;
+
     virtual std::shared_ptr<IEnvironment> createEnvironment(bool is_lsl, uint32_t api_version) = 0;
     virtual std::shared_ptr<IImage> buildImage(std::shared_ptr<IEnvironment> environment, const ImageConfig& config) = 0;
     virtual std::shared_ptr<Script> instantiateScript(const std::shared_ptr<IImage>& image, const ScriptConfig& config) = 0;
@@ -365,10 +469,20 @@ class Provisioner : public IProvisioner
 
     const HostCallbacks& getCallbacks() const override { return mCallbacks; }
 
+    InterruptInstaller* getInterruptInstaller() const override { return mInterruptInstaller.get(); }
+    WatchdogStats getWatchdogStats() const override { return mInterruptInstaller ? mInterruptInstaller->getStats() : WatchdogStats{}; }
+
     std::shared_ptr<IEnvironment> createEnvironment(bool is_lsl, uint32_t api_version) override
     {
         std::shared_ptr<Environment> environment = makeEnvironment(is_lsl, api_version);
         environment->build<S>();
+
+        if (mInterruptInstaller == nullptr)
+        {
+            InterruptCallback handler = lua_callbacks(environment->getBaseState())->interrupt;
+            mInterruptInstaller = createInterruptInstaller(mCallbacks.interruptInstallPolicy, handler, mCallbacks.interruptFireLead);
+        }
+
         return environment;
     }
 
@@ -433,6 +547,7 @@ class Provisioner : public IProvisioner
 
 private:
     HostCallbacks mCallbacks;
+    std::unique_ptr<InterruptInstaller> mInterruptInstaller;
 };
 
 } // namespace Executor
diff --git a/Executor/include/Luau/Script.h b/Executor/include/Luau/Script.h
index be5bfa73..103912d5 100644
--- a/Executor/include/Luau/Script.h
+++ b/Executor/include/Luau/Script.h
@@ -101,22 +101,25 @@ class Script : protected lua_SLRuntimeState
     bool loadDefaultState();
 
     // Drops any live instance and returns the script to the image's default
-    // state. Unlike loadDefaultState(), this is valid at any time.
+    // state. Unlike loadDefaultState(), this is valid at any time, though
+    // they're functionally the same.
     bool reset();
 
-    // Start a run window for `quanta` seconds. _must_ be followed by symmetric `endRunWindow()`
-    // TODO: RAII helper thingy maybe?
+    // Start a run window for `quanta` seconds. Pair with endRunWindow().
     void beginRunWindow(double quanta);
-    // And finish it.
+    // isYieldDue(), getExcludedTime() and getSleep() stay readable until the
+    // next begin. Clears any force-yield.
     void endRunWindow();
     // True once the engine has decided the current run window is over, sticky
     // for the rest of the window.
     bool isYieldDue() const { return mYieldDue; }
 
-    // Level-triggered forced preemption, port of the host's
-    // `getReset() || !mIsEnabled` check. The host both sets and clears it.
-    void setForceYield(bool force) { mForceYield = force; }
+    // Make the current window yield at its next safepoint. Port of the host's
+    // `getReset() || !mIsEnabled` check. Cleared by endRunWindow().
+    void setForceYield(bool force);
 
+    // Only has meaning for embedders, in `indra` this allows us to get the wrapping
+    // `LLScriptExecute` ptr.
     void *getHostContext() const { return mHostContext; }
 
     // Call an event handler associated with the given state, works for both
@@ -157,7 +160,7 @@ class Script : protected lua_SLRuntimeState
     // How long we've been told to sleep. Never decremented, only zeroed out when the scheduler
     // decides we're done the sleep.
     float getSleep() const { return mSleep; }
-    void setSleep(float sleep) { mSleep = sleep; }
+    void setSleep(float sleep);
 
     // Wall time excluded from punishment accounting in the current window
     double getExcludedTime() const { return mExcludedTime; }
@@ -218,16 +221,25 @@ class Script : protected lua_SLRuntimeState
 
     // Monotonic stopwatch used for quanta-elapsed measurement, seeded from the
     // provisioner's.
-    lua_clockProvider mQuantaClockProvider = nullptr;
+    QuantaClock mQuantaClockProvider = nullptr;
+    // Puts the interrupt handler on mCallbacks by each run window's deadline
+    InterruptInstaller* mInterruptInstaller = nullptr;
+    // The environment VM's, which outlives us
+    lua_Callbacks* mCallbacks = nullptr;
 
     // When did we start running
     double mWindowStart = 0.0;
     // How long are we supposed to run?
     double mQuanta = 0.0;
-    // Wall time this window spent inside GC steps and reachability walks.
+    // Wall time this window spent inside reachability walks.
     double mExcludedTime = 0.0;
-    // Quanta clock reading at the opening bracket of the GC step in flight.
-    double mGCStepStart = 0.0;
+    // When we're supposed to invoke GC next. Held here since we disable GC during a run.
+    // TODO: there is probably some threshold of churn after which we want a script to take
+    //  a nap. There needs to be some quantifiable cost for abusive garbage creation during
+    //  a quanta, but we don't really have a "proper" scheduler that would let us deprioritize.
+    //  All we have is the blunt instrument of "sleep"... When there is one, churn should
+    //  accrue in memoryLimitCallback(), where each allocation is attributable to a script.
+    size_t mSavedGCThreshold = 0;
     // We've planned a script kill for this time, the script will be killed if
     // it doesn't finish before the deadline.
     double mMandatoryDeadline = 0.0;
@@ -237,12 +249,10 @@ class Script : protected lua_SLRuntimeState
     bool mYieldDue = false;
     // We already threw a catchable error trying to force a yield.
     bool mMandatoryYieldRaised = false;
-    // Between the paired pre and post interrupts of one GC step.
-    bool mGCStepInFlight = false;
     bool mInExecution = false;
     bool mMainFunctionComplete = false;
     // The host has indicated that a yield should happen at the next interrupt.
-    // TODO: Hmm. Seems whe might be able to merge this with `mYieldDue` if we're smart.
+    // TODO: Hmm. Seems we might be able to merge this with `mYieldDue` if we're smart.
     bool mForceYield = false;
 
     FaultKind mFaultKind = FaultKind::None;
@@ -250,5 +260,25 @@ class Script : protected lua_SLRuntimeState
     std::string mExtendedFaultString;
 };
 
+/// An execution span for a single script that may contain multiple handler invocations.
+/// The script must outlive the guard.
+class RunWindow
+{
+public:
+    RunWindow(Script& script, double quanta);
+    ~RunWindow();
+
+    RunWindow(RunWindow&& other) noexcept;
+    RunWindow(const RunWindow&) = delete;
+    RunWindow& operator=(const RunWindow&) = delete;
+    RunWindow& operator=(RunWindow&&) = delete;
+
+    // End the window early. No-op if it's already closed.
+    void close();
+
+private:
+    Script* mScript;
+};
+
 } // namespace Executor
 } // namespace Luau
diff --git a/Executor/src/Scheduler.cpp b/Executor/src/Scheduler.cpp
index 40c8abe6..aaf17bfc 100644
--- a/Executor/src/Scheduler.cpp
+++ b/Executor/src/Scheduler.cpp
@@ -1,33 +1,70 @@
-// low-overhead timestamp-counter clock for quanta preemption checks.
-//  This is largely a formalization of the `LLRDTSCTimer` pattern in
-//  viewer and server.
+// Installers to plop in the interrupt handler as-needed, and
+// the low-overhead timestamp-counter clock it reads (largely a formalization
+// of the `LLRDTSCTimer` pattern in viewer and server).
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <thread>
+
 #include "Luau/Executor.h"
 
 #include "Luau/Common.h"
 
 #include "lua.h"
 
-#if defined(__x86_64__) || defined(_M_X64)
+#if defined(__x86_64__) || defined(_M_X64) || defined(__i386__) || defined(_M_IX86)
 #if defined(_MSC_VER)
 #include <intrin.h>
 #else
 #include <x86intrin.h>
 #endif
-#define SLUA_QUANTA_TSC_X64 1
+#define SLUA_QUANTA_TSC_X86 1
 #elif defined(__aarch64__) && !defined(_MSC_VER)
 #define SLUA_QUANTA_TSC_ARM64 1
 #endif
 
-#if defined(SLUA_QUANTA_TSC_X64) || defined(SLUA_QUANTA_TSC_ARM64)
+#if defined(SLUA_QUANTA_TSC_X86) || defined(SLUA_QUANTA_TSC_ARM64)
 #define SLUA_QUANTA_TSC 1
 #endif
 
+#if defined(_WIN32)
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif
+#ifndef NOMINMAX
+#define NOMINMAX
+#endif
+#include <windows.h>
+#else
+#include <pthread.h>
+#include <sched.h>
+#if defined(__linux__)
+#include <cerrno>
+#include <csignal>
+#include <ctime>
+#include <sys/prctl.h>
+#include <sys/resource.h>
+#include <sys/syscall.h>
+#include <unistd.h>
+// Only newer glibc names the SIGEV_THREAD_ID target, older ones just have the union member
+#if defined(__GLIBC__) && !defined(sigev_notify_thread_id)
+#define sigev_notify_thread_id _sigev_un._tid
+#endif
+#endif
+#endif
+
+LUAU_FASTFLAGVARIABLE(SLuaThreadedQuantaWatchdog)
+LUAU_FASTFLAGVARIABLE(SLuaElevateWatchdogPriority)
+
 namespace Luau
 {
 namespace Executor
 {
 
-#if defined(SLUA_QUANTA_TSC_X64)
+#if defined(SLUA_QUANTA_TSC_X86)
 static inline uint64_t read_tsc()
 {
     // Shouldn't need a fence.
@@ -70,6 +107,7 @@ static double measure_seconds_per_tick()
     // The architecture publishes the exact frequency, no calibration needed.
     // Apple Silicon reports 24MHz -> ~41.7ns per tick, ~3600 ticks per 150us
     // quanta.
+    // TODO: is this the case for _all_ modern ARM64?
     uint64_t freq;
     asm volatile("mrs %0, cntfrq_el0" : "=r"(freq));
     LUAU_ASSERT(freq != 0);
@@ -78,30 +116,522 @@ static double measure_seconds_per_tick()
 #endif
 
 #if defined(SLUA_QUANTA_TSC)
-static double tsc_quanta_clock([[maybe_unused]] lua_State* L)
+static double tsc_quanta_clock()
 {
+    // `static` so it's only computed on first call.
     static const double seconds_per_tick = measure_seconds_per_tick();
     return (double)read_tsc() * seconds_per_tick;
 }
 #endif
 
-#if !defined(SLUA_QUANTA_TSC)
-static double fallback_quanta_clock([[maybe_unused]] lua_State* L)
-{
-    return lua_clock();
-}
-#endif
-
-lua_clockProvider resolveDefaultQuantaClock()
+QuantaClock resolveDefaultQuantaClock()
 {
 #if defined(SLUA_QUANTA_TSC)
     // Warm up now so any calibration cost lands at
     // provisioner setup rather than inside the first script's run window
-    tsc_quanta_clock(nullptr);
+    tsc_quanta_clock();
     return tsc_quanta_clock;
 #else
-    return fallback_quanta_clock;
+    // If we don't have anything better, just use whatever `lua_clock()` uses for the platform.
+    return lua_clock;
+#endif
+}
+
+static InterruptInstallPolicy resolveDefaultInterruptInstallPolicy()
+{
+    return FFlag::SLuaThreadedQuantaWatchdog ? InterruptInstallPolicy::Threaded : InterruptInstallPolicy::Resident;
+}
+
+InterruptInstaller::~InterruptInstaller() = default;
+
+namespace
+{
+
+/// Stays resident on `cb->interrupt`, triggering the interrupt callback
+/// wherever it is possible.
+///
+/// Expensive on platforms where constantly reading the current cycle count
+/// is expensive, but preferred everywhere else (like modern ARM64)
+class ResidentInterruptInstaller final : public InterruptInstaller
+{
+public:
+    using InterruptInstaller::InterruptInstaller;
+
+    void installWithin(lua_Callbacks* target, double /* seconds */) override
+    {
+        mTarget = target;
+        installNow();
+    }
+
+    void installNow() override
+    {
+        LUAU_ASSERT(mTarget != nullptr);
+        mTarget->interrupt = mHandler;
+    }
+
+    void cancel() override {}
+
+private:
+    // Never pending, the install is immediate
+    void uninstallPending() override
+    {
+        LUAU_ASSERT(false);
+    }
+
+    lua_Callbacks* mTarget = nullptr;
+};
+
+// How early the watchdog installs the handler when the host doesn't say, to
+// cover wakeup jitter
+constexpr double kDefaultThreadedFireLead = 50e-6;
+
+// Elevating the priority helps a lot if we want to stick to a strict schedule.
+// If we don't do this, the watchdog may fire much later than we intend because
+// the kernel is allowed to take its time waking the thread up.
+void elevate_watchdog_thread_priority()
+{
+#if defined(_WIN32)
+    // wait_for granularity is ~1ms at best on Windows, fires will be late regardless
+    if (SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_TIME_CRITICAL))
+    {
+        logInfo("InterruptInstaller", "Watchdog thread priority raised to TIME_CRITICAL");
+        return;
+    }
+    logWarn("InterruptInstaller", "Watchdog priority elevation failed: SetThreadPriority error %lu", GetLastError());
+#else
+#if defined(__linux__)
+    // Default timer slack is 50us, the whole fire lead
+    if (prctl(PR_SET_TIMERSLACK, 1) != 0)
+        logWarn("InterruptInstaller", "Couldn't reduce watchdog timer slack: errno %d", errno);
+#endif
+    // The lowest RT priority is enough, we only need to preempt SCHED_OTHER threads
+    sched_param param = {};
+    param.sched_priority = sched_get_priority_min(SCHED_RR);
+    int rr_err = pthread_setschedparam(pthread_self(), SCHED_RR, &param);
+    if (rr_err == 0)
+    {
+        logInfo("InterruptInstaller", "Watchdog thread scheduling class raised to SCHED_RR");
+        return;
+    }
+#if defined(__linux__)
+    // who == 0 is the calling thread under NPTL
+    if (setpriority(PRIO_PROCESS, 0, -10) == 0)
+    {
+        logInfo("InterruptInstaller", "Watchdog thread niced to -10 (SCHED_RR unavailable, errno %d)", rr_err);
+        return;
+    }
+    logWarn("InterruptInstaller", "Watchdog priority elevation failed: SCHED_RR errno %d, setpriority errno %d", rr_err, errno);
+#else
+    logWarn("InterruptInstaller", "Watchdog priority elevation failed: SCHED_RR errno %d", rr_err);
+#endif
+#endif
+}
+
+/// Uses a background watchdog thread to only set `cb->interrupt` when quanta is up.
+///
+/// A little more convoluted than the above "resident" installer, but yields a 30% performance
+/// boost in math-heavy code on x86-64 and x86-32.
+class ThreadedInterruptInstaller final : public InterruptInstaller
+{
+public:
+    ThreadedInterruptInstaller(InterruptCallback handler, double fire_lead)
+        : InterruptInstaller(handler)
+        , mFireLead(fire_lead)
+    {
+        mThread = std::thread([this] { loopAndWatch(); });
+    }
+
+    ~ThreadedInterruptInstaller() override
+    {
+        {
+            std::lock_guard<std::mutex> guard(mMutex);
+            mExit = true;
+        }
+        mCondition.notify_one();
+        mThread.join();
+    }
+
+    ThreadedInterruptInstaller(const ThreadedInterruptInstaller&) = delete;
+    ThreadedInterruptInstaller& operator=(const ThreadedInterruptInstaller&) = delete;
+
+    void installWithin(lua_Callbacks* target, double seconds) override
+    {
+        // Clear before publishing the deadline, otherwise a fire landing in between
+        // would get wiped and the window would never be preempted.
+        target->interrupt = nullptr;
+
+        const double deadline = lua_clock() + seconds;
+        bool wake;
+        {
+            std::lock_guard<std::mutex> guard(mMutex);
+            LUAU_ASSERT(!mPending.load(std::memory_order_relaxed));
+            mTarget = target;
+            mDeadline = deadline;
+            mOverrun = 0.0;
+            mPending.store(true, std::memory_order_relaxed);
+            // Waking is a syscall and a context switch, so only do it if the
+            // thread isn't already due up in time. Otherwise it wakes at its
+            // old target, sees this deadline, and sleeps again toward it.
+            wake = mIdle || deadline - mFireLead < mSleepUntil;
+        }
+        if (wake)
+            mCondition.notify_one();
+    }
+
+    void installNow() override
+    {
+        std::lock_guard<std::mutex> guard(mMutex);
+        // But only if it was pending, don't touch it if we already installed.
+        if (mPending.load(std::memory_order_relaxed))
+            mTarget->interrupt = mHandler;
+    }
+
+    void cancel() override
+    {
+        std::lock_guard<std::mutex> guard(mMutex);
+        mPending.store(false, std::memory_order_relaxed);
+    }
+
+    double getInstallOverrun() override
+    {
+        std::lock_guard<std::mutex> guard(mMutex);
+        return mOverrun;
+    }
+
+    WatchdogStats getStats() override
+    {
+        std::lock_guard<std::mutex> guard(mMutex);
+        WatchdogStats stats = mStats;
+        stats.fireLead = mFireLead;
+        return stats;
+    }
+
+private:
+    void uninstallPending() override
+    {
+        // The inline check that got us here was lockless. Only installWithin()
+        // goes false -> true, on this thread, so a false reading there is
+        // stable, but a true one needs the lock to make sure the fire didn't
+        // land in between.
+        std::lock_guard<std::mutex> guard(mMutex);
+        if (mPending.load(std::memory_order_relaxed))
+            mTarget->interrupt = nullptr;
+    }
+
+    void loopAndWatch()
+    {
+        if (FFlag::SLuaElevateWatchdogPriority)
+            elevate_watchdog_thread_priority();
+
+        std::unique_lock<std::mutex> lock(mMutex);
+        while (!mExit)
+        {
+            if (!mPending.load(std::memory_order_relaxed))
+            {
+                // cancel() doesn't notify, a stale deadline lands us here too.
+                // Just sit until we're notified.
+                mIdle = true;
+                mCondition.wait(lock);
+                mIdle = false;
+                mStats.wakes++;
+                continue;
+            }
+
+            const double remaining = mDeadline - lua_clock();
+            if (remaining <= mFireLead)
+            {
+                release_store(&mTarget->interrupt, mHandler);
+                mPending.store(false, std::memory_order_relaxed);
+                mOverrun = std::max(0.0, -remaining);
+
+                double lateness = mFireLead - remaining;
+                mStats.latenessSum += lateness;
+                mStats.latenessMin = mStats.fires == 0 ? lateness : std::min(mStats.latenessMin, lateness);
+                mStats.latenessMax = std::max(mStats.latenessMax, lateness);
+                mStats.fires++;
+                if (mOverrun > 0.0)
+                    mStats.lateFires++;
+                continue;
+            }
+            mSleepUntil = mDeadline - mFireLead;
+            mCondition.wait_for(lock, std::chrono::duration<double>(remaining - mFireLead));
+            mStats.wakes++;
+        }
+    }
+
+    double mFireLead = 0.0;
+
+    std::mutex mMutex;
+    std::condition_variable mCondition;
+    lua_Callbacks* mTarget = nullptr;
+    double mDeadline = 0.0;
+    // Seconds past mDeadline that the last fire landed
+    double mOverrun = 0.0;
+    WatchdogStats mStats;
+    // Moment we expect the thread to wake back up
+    double mSleepUntil = 0.0;
+    // Are we currently sitting idle waiting for another script to work on
+    bool mIdle = true;
+    // Should we start tearing down the watchdog
+    bool mExit = false;
+
+    std::thread mThread;
+};
+
+#if defined(__linux__)
+// How early the signal is asked for when the host doesn't say. Sized to cover
+// the bulk of hrtimer expiry plus delivery to a running thread on a
+// virtualized simhost, where the average is ~11us with a long tail.
+constexpr double kDefaultSignalFireLead = 25e-6;
+
+// signal number that shouldn't be taken by indra, value not important
+// this is only a function because `SIGRTMIN` does a function call :(
+static int get_watchdog_signal()
+{
+    return SIGRTMIN + 4;
+}
+
+// One sigaction per process, however many installers get made
+static std::once_flag sSignalHandlerRegistered;
+
+/// A POSIX timer bound to the script thread delivers a signal at the deadline,
+/// and the signal handler installs `cb->interrupt` from the script thread itself.
+/// Nothing has to be woken up, so nothing has to win a scheduling decision to
+/// land on time. That makes it the pick where SCHED_RR isn't on the table.
+///
+/// Everything here runs on the script thread. The only concurrency is the
+/// signal handler landing between two of our own instructions, so the atomics
+/// and fences are there for the compiler, not for another core.
+class SignalInterruptInstaller final : public InterruptInstaller
+{
+public:
+    SignalInterruptInstaller(InterruptCallback handler, double fire_lead)
+        : InterruptInstaller(handler)
+        , mFireLead(fire_lead)
+    {
+        std::call_once(sSignalHandlerRegistered, [] {
+            const int signum = get_watchdog_signal();
+
+            // Mono and friends pick their RT signals by searching for SIG_DFL, so
+            // anything else here is someone we'd be clobbering.
+            struct sigaction current = {};
+            int query_rc = sigaction(signum, nullptr, &current);
+            LUAU_ASSERT_ALWAYS(query_rc == 0);
+            LUAU_ASSERT_ALWAYS(current.sa_handler == SIG_DFL);
+
+            struct sigaction action = {};
+            action.sa_sigaction = onSignal;
+            // SA_RESTART so no syscall in the host sees an EINTR from us
+            action.sa_flags = SA_SIGINFO | SA_RESTART;
+            sigemptyset(&action.sa_mask);
+            // Without our handler the first timer expiry would terminate the process
+            int install_rc = sigaction(signum, &action, nullptr);
+            LUAU_ASSERT_ALWAYS(install_rc == 0);
+        });
+    }
+
+    ~SignalInterruptInstaller() override
+    {
+        // Any signal still queued for it is dropped along with the timer
+        if (mHaveTimer)
+            timer_delete(mTimer);
+    }
+
+    SignalInterruptInstaller(const SignalInterruptInstaller&) = delete;
+    SignalInterruptInstaller& operator=(const SignalInterruptInstaller&) = delete;
+
+    void installWithin(lua_Callbacks* target, double seconds) override
+    {
+        target->interrupt = nullptr;
+        LUAU_ASSERT(!mPending.load(std::memory_order_relaxed));
+        ensureTimer();
+
+        // Clock read before the syscall, so its entry latency doesn't count
+        // against the fire.
+        mDeadline = lua_clock() + seconds;
+        mTarget = target;
+        mOverrun = 0.0;
+
+        // Publish the window before the signal handler can see pending
+        std::atomic_signal_fence(std::memory_order_release);
+        mPending.store(true, std::memory_order_relaxed);
+        // If the timer can't deliver, whether it's too late already or we never
+        // got one, the install happens here instead.
+        if (seconds <= mFireLead || !requestSignalAt(mDeadline - mFireLead))
+            deadlineInstall();
+    }
+
+    void installNow() override
+    {
+        // But only if it was pending, don't touch it if we already installed.
+        if (mPending.load(std::memory_order_relaxed))
+            mTarget->interrupt = mHandler;
+    }
+
+    void cancel() override
+    {
+        // Once the install has happened there's no signal left to withdraw, so
+        // a preempted window only pays for the request. The exchange can't be
+        // split by the signal handler.
+        if (mPending.exchange(false, std::memory_order_relaxed))
+        {
+            if (!mHaveTimer)
+                return;
+            itimerspec spec = {};
+            timer_settime(mTimer, 0, &spec, nullptr);
+        }
+    }
+
+    double getInstallOverrun() override
+    {
+        return mOverrun;
+    }
+
+    WatchdogStats getStats() override
+    {
+        WatchdogStats stats = mStats;
+        stats.fireLead = mFireLead;
+        return stats;
+    }
+
+private:
+    void uninstallPending() override
+    {
+        mTarget->interrupt = nullptr;
+        // The signal can land between the inline pending check and that store,
+        // in which case we just wiped a real install. Recheck and put it back.
+        std::atomic_signal_fence(std::memory_order_seq_cst);
+        if (!mPending.load(std::memory_order_relaxed))
+            mTarget->interrupt = mHandler;
+    }
+
+    static void onSignal(int, siginfo_t* info, void*)
+    {
+        // If we're not being woken up for a timer, this isn't for us.
+        if (info->si_code != SI_TIMER)
+            return;
+        // Get a reference to the specific interrupt installer
+        auto* self = static_cast<SignalInterruptInstaller*>(info->si_value.sival_ptr);
+        // A signal that was already queued when its window was cancelled still
+        // gets delivered
+        if (!self->mPending.load(std::memory_order_relaxed))
+            return;
+        std::atomic_signal_fence(std::memory_order_acquire);
+        self->deadlineInstall();
+    }
+
+    // The install installWithin() promised, plus the timing for it. Usually from
+    // the signal handler, so only atomics, plain stores and the clock: it has
+    // to stay async-signal-safe.
+    void deadlineInstall()
+    {
+        mTarget->interrupt = mHandler;
+        mPending.store(false, std::memory_order_relaxed);
+
+        const double remaining = mDeadline - lua_clock();
+        mOverrun = std::max(0.0, -remaining);
+
+        double lateness = mFireLead - remaining;
+        mStats.latenessSum += lateness;
+        mStats.latenessMin = mStats.fires == 0 ? lateness : std::min(mStats.latenessMin, lateness);
+        mStats.latenessMax = std::max(mStats.latenessMax, lateness);
+        mStats.fires++;
+        if (mOverrun > 0.0)
+            mStats.lateFires++;
+    }
+
+    // The timer is bound to a thread, so it follows whichever one opens windows.
+    void ensureTimer()
+    {
+        pid_t tid = (pid_t)syscall(SYS_gettid);
+        // We have a timer and it's for this thread
+        if (mHaveTimer && mTimerThread == tid)
+            return;
+
+        if (mHaveTimer)
+        {
+            // This should never be used cross-thread!
+            LUAU_ASSERT(mTimerThread == tid);
+            timer_delete(mTimer);
+            mHaveTimer = false;
+        }
+
+        mTimerThread = tid;
+
+        struct sigevent event = {};
+        event.sigev_notify = SIGEV_THREAD_ID;
+        event.sigev_signo = get_watchdog_signal();
+        event.sigev_value.sival_ptr = this;
+        event.sigev_notify_thread_id = tid;
+        mHaveTimer = timer_create(CLOCK_MONOTONIC, &event, &mTimer) == 0;
+        if (!mHaveTimer)
+        {
+            logWarn("InterruptInstaller", "Couldn't create the watchdog timer (errno %d), installing at window start instead", errno);
+            return;
+        }
+
+        // Slack is a property of the thread that sets the timer and defaults to
+        // 50us, which would swallow the lead several times over. This tightens
+        // every timer on the script thread, not just ours.
+        // I'm a little squeamish about setting this since it's in the main thread and process-global,
+        // so let's see if it's a problem in practice...
+        // if (prctl(PR_SET_TIMERSLACK, 1) != 0)
+        //     logWarn("InterruptInstaller", "Couldn't reduce script thread timer slack: errno %d", errno);
+    }
+
+    // Ask for one signal at `when`, in lua_clock() seconds, which on Linux is
+    // CLOCK_MONOTONIC, the timer's own clock. False when there is no timer or
+    // the kernel refused.
+    bool requestSignalAt(double when)
+    {
+        if (!mHaveTimer)
+            return false;
+
+        itimerspec spec = {};
+        spec.it_value.tv_sec = (time_t)when;
+        spec.it_value.tv_nsec = (long)((when - (double)spec.it_value.tv_sec) * 1e9);
+        return timer_settime(mTimer, TIMER_ABSTIME, &spec, nullptr) == 0;
+    }
+
+    double mFireLead = 0.0;
+
+    lua_Callbacks* mTarget = nullptr;
+    double mDeadline = 0.0;
+    // Seconds past mDeadline that the last fire landed
+    double mOverrun = 0.0;
+    WatchdogStats mStats;
+
+    timer_t mTimer{};
+    bool mHaveTimer = false;
+    // Thread the timer signals, 0 until the first window
+    pid_t mTimerThread = 0;
+};
+#endif
+
+} // namespace
+
+std::unique_ptr<InterruptInstaller> createInterruptInstaller(InterruptInstallPolicy policy, InterruptCallback handler, double fire_lead)
+{
+    if (policy == InterruptInstallPolicy::Default)
+        policy = resolveDefaultInterruptInstallPolicy();
+    if (policy == InterruptInstallPolicy::Signal)
+    {
+#if defined(__linux__)
+        if (fire_lead <= 0.0)
+            fire_lead = kDefaultSignalFireLead;
+        return std::make_unique<SignalInterruptInstaller>(handler, fire_lead);
+#else
+        logWarn("InterruptInstaller", "Signal install policy is Linux only, using the watchdog thread instead");
+        policy = InterruptInstallPolicy::Threaded;
 #endif
+    }
+    if (policy == InterruptInstallPolicy::Threaded)
+    {
+        if (fire_lead <= 0.0)
+            fire_lead = kDefaultThreadedFireLead;
+        return std::make_unique<ThreadedInterruptInstaller>(handler, fire_lead);
+    }
+    return std::make_unique<ResidentInterruptInstaller>(handler);
 }
 
 } // namespace Executor
diff --git a/Executor/src/Script.cpp b/Executor/src/Script.cpp
index c046c472..ec635c60 100644
--- a/Executor/src/Script.cpp
+++ b/Executor/src/Script.cpp
@@ -8,6 +8,7 @@
 #include "lualib.h"
 #include "llsl.h"
 
+#include "lgc.h"
 #include "lstate.h"
 
 namespace Luau
@@ -129,13 +130,42 @@ Script::Script(const std::shared_ptr<IImage>& image, const ScriptConfig& config)
     setTimerEventCb = callbacks.setTimerEventCb;
     eventHandlerRegistrationCb = callbacks.eventHandlerRegistrationCb;
     mQuantaClockProvider = callbacks.quantaClockProvider;
+
+    mInterruptInstaller = image->getProvisioner().getInterruptInstaller();
+    LUAU_ASSERT(mInterruptInstaller != nullptr);
+    mCallbacks = lua_callbacks(image->getEnvironment().getBaseState());
 }
 
 Script::~Script()
 {
-    LUAU_ASSERT(!mHandlerState && !mInExecution);
-    // Nothing to do: mInstance releases its anchor before mImage lets go of
-    // the VM, by member declaration order.
+    LUAU_ASSERT(!mHandlerState && !mInExecution && !mRunWindowOpen);
+    // mInstance releases its anchor before mImage lets go of the VM, by
+    // member declaration order.
+}
+
+RunWindow::RunWindow(Script& script, double quanta)
+    : mScript(&script)
+{
+    script.beginRunWindow(quanta);
+}
+
+RunWindow::~RunWindow()
+{
+    close();
+}
+
+RunWindow::RunWindow(RunWindow&& other) noexcept
+    : mScript(other.mScript)
+{
+    other.mScript = nullptr;
+}
+
+void RunWindow::close()
+{
+    if (mScript == nullptr)
+        return;
+    mScript->endRunWindow();
+    mScript = nullptr;
 }
 
 // static
@@ -190,7 +220,6 @@ bool Script::loadDefaultState()
     mMandatoryYieldRaised = false;
     mMandatoryDeadline = 0.0;
     mExcludedTime = 0.0;
-    mGCStepInFlight = false;
     mSleep = 0.0f;
     // in LSL we treat the main function as complete, since it runs
     // while loading the image.
@@ -266,20 +295,59 @@ void Script::beginRunWindow(double quanta)
 {
     LUAU_ASSERT(mInstance);
     LUAU_ASSERT(!mRunWindowOpen);
+    LUAU_ASSERT(mSleep == 0.0f);
+    LUAU_ASSERT(!mForceYield);
+
     mRunWindowOpen = true;
     mYieldDue = false;
     mMandatoryYieldRaised = false;
     mMandatoryDeadline = 0.0;
     mExcludedTime = 0.0;
-    mGCStepInFlight = false;
-    mWindowStart = mQuantaClockProvider(mInstance.thread());
+    mWindowStart = mQuantaClockProvider();
     mQuanta = quanta;
+
+    // Park the GC for the window, endRunWindow() pays down the debt.
+    // The time it took to GC something arbitrary during the window doesn't
+    // necessarily have any relation to anything that happened during the window,
+    // so we just do it at the end to discount it for now.
+    global_State* global = mImage->getEnvironment().getBaseState()->global;
+    mSavedGCThreshold = global->GCthreshold;
+    global->GCthreshold = SIZE_MAX;
+
+    mInterruptInstaller->installWithin(mCallbacks, quanta);
 }
 
 void Script::endRunWindow()
 {
     LUAU_ASSERT(mRunWindowOpen);
     mRunWindowOpen = false;
+    mForceYield = false;
+
+    mInterruptInstaller->cancel();
+    // A fired handler would otherwise linger through the catch-up below
+    mCallbacks->interrupt = nullptr;
+
+    lua_State* base = mImage->getEnvironment().getBaseState();
+    global_State* global = base->global;
+    LUAU_ASSERT(global->GCthreshold == SIZE_MAX);
+    // Restore before stepping, the pacer computes its debt from the threshold.
+    global->GCthreshold = mSavedGCThreshold;
+    while (luaC_needsGC(base))
+        luaC_step(base, false);
+}
+
+void Script::setSleep(float sleep)
+{
+    mSleep = sleep;
+    if (sleep > 0.0f && mRunWindowOpen)
+        mInterruptInstaller->installNow();
+}
+
+void Script::setForceYield(bool force)
+{
+    mForceYield = force;
+    if (force && mRunWindowOpen)
+        mInterruptInstaller->installNow();
 }
 
 RunResult Script::callEventHandler(int lsl_state, const char* event_name, PushArgsFn pushArgs, void* push_args_ctx)
@@ -715,21 +783,9 @@ void Script::interruptHandler(lua_State* L, int gc)
     if (!execute || !execute->mInExecution || execute->mHandlerState == nullptr)
         return;
 
-    // Keep track of GC phases so we can discount their cost
+    // GC is parked for the whole window, this can't be ours
     if (gc >= 0)
-    {
-        if (!execute->mGCStepInFlight)
-        {
-            execute->mGCStepInFlight = true;
-            execute->mGCStepStart = execute->mQuantaClockProvider(L);
-        }
-        else
-        {
-            execute->mGCStepInFlight = false;
-            execute->mExcludedTime += execute->mQuantaClockProvider(L) - execute->mGCStepStart;
-        }
         return;
-    }
 
     // This particular handler had better be _active_, or we're somehow triggering
     // interrupts through pushing handler arguments or something like that.
@@ -741,11 +797,17 @@ void Script::interruptHandler(lua_State* L, int gc)
         return;
     }
 
-    double elapsed = execute->mQuantaClockProvider(L) - execute->mWindowStart;
+    double elapsed = execute->mQuantaClockProvider() - execute->mWindowStart;
     if (elapsed > execute->mQuanta)
     {
         execute->tryYield(L, elapsed);
     }
+    else
+    {
+        // Installed early (fire lead, or a withdrawn force-yield). Uninstalls
+        // only if the deadline will bring us back.
+        execute->mInterruptInstaller->uninstallIfPending();
+    }
 }
 
 // static
@@ -801,13 +863,15 @@ int Script::memoryLimitCallback(lua_State* L, size_t osize, size_t nsize)
         // Discount only the baseline measurement's walk: its timing depends on engine
         // bookkeeping state, while walks forced by allocating near the limit are the
         // script's own work and stay charged.
+        // TODO: Maybe simpler to get rid of the time exclusion mechanism and just opportunistically
+        //  scan whenever we load a script?
         bool discount_walk = execute->mExactSize == 0 && execute->mHandlerState != nullptr;
-        double walk_start = discount_walk ? execute->mQuantaClockProvider(L) : 0.0;
+        double walk_start = discount_walk ? execute->mQuantaClockProvider() : 0.0;
 
         size_t current_size = lua_userthreadgc(execute->mInstance.thread(), &execute->mImage->getFreeObjects()) + execute->mChargedBytecodeSize;
 
         if (discount_walk)
-            execute->mExcludedTime += execute->mQuantaClockProvider(L) - walk_start;
+            execute->mExcludedTime += execute->mQuantaClockProvider() - walk_start;
 
         execute->mExactSize = execute->mMaxPossibleSize = (int)current_size;
         int new_size = execute->mExactSize + (int)net_gain;
@@ -833,8 +897,11 @@ void Script::tryYield(lua_State* L, double elapsed, bool mandatory)
     // Some quantas are extremely small, so don't punish if they were overrun due to blocking code.
     // Use a higher threshold for punishment in those cases.
     double punish_quanta = std::max(mQuanta, 0.001);
+    // The script ran unbounded through the overrun, but that's the watchdog's
+    // lateness, not the script's. Preemption still sees it in `elapsed`.
+    double overrun = mInterruptInstaller->getInstallOverrun();
     // Discount time spent on work outside the script's control
-    double charged = std::max(elapsed - mExcludedTime, 0.0);
+    double charged = std::max(elapsed - mExcludedTime - overrun, 0.0);
 
     if (interrupt_status == YieldableStatus::OK)
     {
@@ -865,7 +932,7 @@ void Script::tryYield(lua_State* L, double elapsed, bool mandatory)
         {
             // They're over the time limit. Set a deadline. Don't kill immediately in case
             // a temporary performance blip caused them to go over time. The deadline is
-            // in charged time so excluded work (GC steps, heap walks) can't eat the grace.
+            // in charged time so excluded work (heap walks) can't eat the grace.
             if (mMandatoryDeadline == 0.0)
             {
                 mMandatoryDeadline = charged + punish_quanta * 0.5;
diff --git a/Makefile b/Makefile
index f39c1e70..60fdbc09 100644
--- a/Makefile
+++ b/Makefile
@@ -150,6 +150,11 @@ ifeq ($(config),sanitize)
 	LDFLAGS+=-fsanitize=address,undefined
 endif
 
+ifeq ($(config),tsan)
+	CXXFLAGS+=-fsanitize=thread -O1
+	LDFLAGS+=-fsanitize=thread
+endif
+
 ifeq ($(config),analyze)
 	CXXFLAGS+=--analyze
 endif
@@ -208,11 +213,14 @@ $(TEST_LINK_VM_OBJECTS): CXXFLAGS+=-std=c++11 -ICommon/include -IVM/include
 $(TEST_LINK_CODEGEN_OBJECTS): CXXFLAGS+=-std=c++17 -ICommon/include -IVM/include -ICodeGen/include
 $(FUZZ_OBJECTS): CXXFLAGS+=-std=c++17 -ICommon/include -IAst/include -IBytecode/include -IInliner/include -ICompiler/include -IAnalysis/include -IVM/include -ICodeGen/include -IConfig/include -ILSLBuiltins/include -Istage/packages/include
 
-$(TESTS_TARGET): LDFLAGS+=-lpthread -Lstage/packages/lib/release -ltailslide
+# POSIX timers live in librt on glibc < 2.34, a stub elsewhere
+LIBRT=$(if $(filter Darwin,$(shell uname -s)),,-lrt)
+
+$(TESTS_TARGET): LDFLAGS+=-lpthread $(LIBRT) -Lstage/packages/lib/release -ltailslide
 $(REPL_CLI_TARGET): LDFLAGS+=-lpthread -Lstage/packages/lib/release -ltailslide
 $(ANALYZE_CLI_TARGET): LDFLAGS+=-lpthread
 $(COMPILE_CLI_TARGET): LDFLAGS+=-Lstage/packages/lib/release -ltailslide
-$(HARNESS_CLI_TARGET): LDFLAGS+=-lpthread -Lstage/packages/lib/release -ltailslide
+$(HARNESS_CLI_TARGET): LDFLAGS+=-lpthread $(LIBRT) -Lstage/packages/lib/release -ltailslide
 
 fuzz-proto fuzz-prototest: LDFLAGS+=$(LPROTOBUF)
 fuzz-lsl_script: LDFLAGS+=-Lstage/packages/lib/release -ltailslide
diff --git a/VM/include/lua.h b/VM/include/lua.h
index 5ef157e8..c1e8a583 100644
--- a/VM/include/lua.h
+++ b/VM/include/lua.h
@@ -906,6 +906,7 @@ struct lua_Callbacks
 
     void (*onallocate)(lua_State* L, size_t osize, size_t nsize); // gets called when memory is allocated
     // gets called before memory is allocated, return non-zero to fail the alloc. Only called for allocs in user memcats.
+    // ServerLua: called even while the GC is paused (GCthreshold == SIZE_MAX), so gate on your own state if that matters.
     int (*beforeallocate)(lua_State* L, size_t osize, size_t nsize);
 };
 typedef struct lua_Callbacks lua_Callbacks;
diff --git a/VM/include/lyieldablemacros.h b/VM/include/lyieldablemacros.h
index b9b1a014..73460ef8 100644
--- a/VM/include/lyieldablemacros.h
+++ b/VM/include/lyieldablemacros.h
@@ -98,7 +98,7 @@ using Luau::YieldGuard;
         (_yieldable_phase) = Phase::phase_name;                                     \
         if ((L)->status == LUA_OK)                                                  \
         {                                                                           \
-            void (*_yc_int)(lua_State*, int) = _yieldable_slots.callbacks->interrupt; \
+            void (*_yc_int)(lua_State*, int) = Luau::opaque_load(&_yieldable_slots.callbacks->interrupt); \
             if (LUAU_LIKELY(!!_yc_int))                                             \
                 _yc_int((L), (reason));                                             \
         }                                                                           \
diff --git a/VM/src/lgc.cpp b/VM/src/lgc.cpp
index 06d9a69e..4effaa83 100644
--- a/VM/src/lgc.cpp
+++ b/VM/src/lgc.cpp
@@ -130,7 +130,7 @@ LUAU_FASTFLAG(LuauManagedDebugNames)
 
 #define GC_INTERRUPT(state) \
     { \
-        void (*interrupt)(lua_State*, int) = g->cb.interrupt; \
+        void (*interrupt)(lua_State*, int) = Luau::opaque_load(&g->cb.interrupt); \
         if (LUAU_UNLIKELY(!!interrupt)) \
             interrupt(L, state); \
     }
diff --git a/VM/src/llsl.cpp b/VM/src/llsl.cpp
index 8936facb..52dffcdb 100644
--- a/VM/src/llsl.cpp
+++ b/VM/src/llsl.cpp
@@ -1182,7 +1182,7 @@ static int push_uuid_common(lua_State *L, const char *str, size_t len, bool comp
         {
             // See `lgctraverse.cpp` for reasoning behind this
             const size_t uuid_size = 4;
-            if (LUAU_LIKELY(g->GCthreshold != SIZE_MAX) && g->cb.beforeallocate(L, 0, uuid_size))
+            if (g->cb.beforeallocate(L, 0, uuid_size))
                 luaD_throw(L, LUA_ERRMEM);
         }
 
diff --git a/VM/src/lmem.cpp b/VM/src/lmem.cpp
index 0aeae0b8..7c44589c 100644
--- a/VM/src/lmem.cpp
+++ b/VM/src/lmem.cpp
@@ -509,8 +509,7 @@ void* luaM_new_(lua_State* L, size_t nsize, uint8_t memcat)
     // ServerLua: enforce memory allocation limits in user-owned memcats
     if (LUAU_LIKELY(!!g->cb.beforeallocate) && memcat >= LUA_FIRST_USER_MEMCAT)
     {
-        // But don't do invoke the hook if GC is disabled
-        if (LUAU_LIKELY(g->GCthreshold != SIZE_MAX) && g->cb.beforeallocate(L, 0, nsize))
+        if (g->cb.beforeallocate(L, 0, nsize))
             luaD_throw(L, LUA_ERRMEM);
     }
 
@@ -548,7 +547,7 @@ GCObject* luaM_newgco_(lua_State* L, size_t nsize, uint8_t memcat)
     // ServerLua: enforce memory allocation limits in user-owned memcats
     if (LUAU_LIKELY(!!g->cb.beforeallocate) && memcat >= LUA_FIRST_USER_MEMCAT)
     {
-        if (LUAU_LIKELY(g->GCthreshold != SIZE_MAX) && g->cb.beforeallocate(L, 0, nsize))
+        if (g->cb.beforeallocate(L, 0, nsize))
             luaD_throw(L, LUA_ERRMEM);
     }
 
@@ -696,7 +695,7 @@ void* luaM_realloc_(lua_State* L, void* block, size_t osize, size_t nsize, uint8
     // ServerLua: enforce memory allocation limits in user-owned memcats
     if (LUAU_LIKELY(!!g->cb.beforeallocate) && memcat >= LUA_FIRST_USER_MEMCAT)
     {
-        if (LUAU_LIKELY(g->GCthreshold != SIZE_MAX) && g->cb.beforeallocate(L, osize, nsize))
+        if (g->cb.beforeallocate(L, osize, nsize))
             luaD_throw(L, LUA_ERRMEM);
     }
 
diff --git a/VM/src/lstate.cpp b/VM/src/lstate.cpp
index ad5686a3..dff6ac4d 100644
--- a/VM/src/lstate.cpp
+++ b/VM/src/lstate.cpp
@@ -342,7 +342,7 @@ void luau_callinterrupthandler(lua_State* L, int code)
 {
     LUAU_ASSERT(code < 0); // Interrupt codes must be negative (>= 0 reserved for GC)
 
-    void (*interrupt)(lua_State*, int) = L->global->cb.interrupt;
+    void (*interrupt)(lua_State*, int) = Luau::opaque_load(&L->global->cb.interrupt); // ServerLua
 
     if (LUAU_UNLIKELY(!!interrupt))
     {
diff --git a/VM/src/lstring.cpp b/VM/src/lstring.cpp
index 04ed8b33..006e2d74 100644
--- a/VM/src/lstring.cpp
+++ b/VM/src/lstring.cpp
@@ -136,7 +136,7 @@ static TString* findstring(lua_State * L, const char *data, unsigned int len, un
             {
                 // See `lgctraverse.cpp` for reasoning behind this
                 const size_t str_size = 16 + el->len;
-                if (LUAU_LIKELY(g->GCthreshold != SIZE_MAX) && g->cb.beforeallocate(L, 0, str_size))
+                if (g->cb.beforeallocate(L, 0, str_size))
                     luaD_throw(L, LUA_ERRMEM);
             }
             return el;
diff --git a/VM/src/lstrlib.cpp b/VM/src/lstrlib.cpp
index 2f5407e2..f2cf8644 100644
--- a/VM/src/lstrlib.cpp
+++ b/VM/src/lstrlib.cpp
@@ -436,7 +436,7 @@ static const char* match(MatchState* ms, const char* s, const char* p)
         luaL_error(ms->L, "pattern too complex");
 
     lua_State* L = ms->L;
-    void (*interrupt)(lua_State*, int) = L->global->cb.interrupt;
+    void (*interrupt)(lua_State*, int) = Luau::opaque_load(&L->global->cb.interrupt); // ServerLua
 
     if (LUAU_UNLIKELY(!!interrupt))
     {
diff --git a/VM/src/lvmexecute.cpp b/VM/src/lvmexecute.cpp
index 4f20570d..5d708cf2 100644
--- a/VM/src/lvmexecute.cpp
+++ b/VM/src/lvmexecute.cpp
@@ -114,7 +114,7 @@ LUAU_FLAGVERSION(LuauBackedgeHeapCheck, 2)
 #define VM_INTERRUPT() VM_INTERRUPT_WITHCODE(LUA_INTERRUPT_NORMAL)
 #define VM_INTERRUPT_WITHCODE(code) \
     { \
-        void (*interrupt)(lua_State*, int) = L->global->cb.interrupt; \
+        void (*interrupt)(lua_State*, int) = Luau::opaque_load(&L->global->cb.interrupt); \
         if (LUAU_UNLIKELY(!!interrupt)) \
         { /* the interrupt hook is called right before we advance pc */ \
             static_assert((code) < 0, "code is negative");              \
diff --git a/autobuild.xml b/autobuild.xml
index b52f19e7..13c41298 100644
--- a/autobuild.xml
+++ b/autobuild.xml
@@ -180,6 +180,7 @@
           <key>manifest</key>
           <array>
             <string>lib</string>
+            <string>bin</string>
           </array>
           <key>name</key>
           <string>darwin</string>
diff --git a/build-cmd.sh b/build-cmd.sh
index 30472602..31779394 100755
--- a/build-cmd.sh
+++ b/build-cmd.sh
@@ -84,6 +84,7 @@ pushd "$top"
             cp -v "Release/Luau.Common.lib" "$stage/lib/release/"
 
             cp -v Release/slua.exe "$stage/bin/"
+            cp -v Release/slua-harness.exe "$stage/bin/"
         ;;
         darwin*|linux*)
             # Optimized builds use the inherited release flags as-is, any local
@@ -111,6 +112,7 @@ pushd "$top"
             cp -v "libLuau.Executor.a" "$stage/lib/release"
 
             cp -v "slua" "$stage/bin/"
+            cp -v "slua-harness" "$stage/bin/"
 
             # Run the conformance test for good measure
             "${stage}/build/Luau.Conformance"
diff --git a/tests/SLConformance.test.cpp b/tests/SLConformance.test.cpp
index 6a17befb..807868aa 100644
--- a/tests/SLConformance.test.cpp
+++ b/tests/SLConformance.test.cpp
@@ -172,6 +172,10 @@ static int memoryLimitCallback(lua_State *L, size_t osize, size_t nsize)
     L = lua_mainthread(L);
     RuntimeState* state = (RuntimeState*)L->userdata;
 
+    // Bail if GC is disabled (host bulk work like luau_load)
+    if (L->global->GCthreshold == SIZE_MAX)
+        return 0;
+
     // This is a net shrink in memory, not relevant for our limiting logic. We can't assume that
     // memory being freed has anything to do with us, given that the GC can work on things unrelated
     // to the currently executing task.
diff --git a/tests/SLExecutor.test.cpp b/tests/SLExecutor.test.cpp
index c961f26e..6c662d61 100644
--- a/tests/SLExecutor.test.cpp
+++ b/tests/SLExecutor.test.cpp
@@ -15,6 +15,7 @@
 
 // For the handler thread's stack and CallInfo capacities, which the API surface
 // has no reason to expose
+#include "../VM/src/lgc.h"
 #include "../VM/src/lstate.h"
 
 #ifdef LUAU_USE_TAILSLIDE
@@ -31,13 +32,47 @@ using namespace Luau::Executor;
 
 LUAU_FASTFLAG(SLuaEagerWeakClear)
 
-// Provisioner with deterministic fakes: a virtual clock that advances by
-// clock_step on every reading, plus capture of print output and dynamic
-// handler registrations.
-struct TestProvisioner : Provisioner<>
+// A virtual quanta clock that advances by clock_step on every reading. The
+// quanta clock takes no lua_State, so the fake finds its host through
+// `current`: last constructed wins, and no test steps the clock while two
+// hosts are alive.
+struct FakeQuantaClock
 {
     double clock = 0.0;
     double clock_step = 0.0;
+
+    FakeQuantaClock()
+        : previous(current)
+    {
+        current = this;
+    }
+
+    ~FakeQuantaClock()
+    {
+        current = previous;
+    }
+
+    FakeQuantaClock(const FakeQuantaClock&) = delete;
+    FakeQuantaClock& operator=(const FakeQuantaClock&) = delete;
+
+    static double read()
+    {
+        LUAU_ASSERT(current != nullptr);
+        current->clock += current->clock_step;
+        return current->clock;
+    }
+
+private:
+    FakeQuantaClock* previous;
+    static FakeQuantaClock* current;
+};
+
+FakeQuantaClock* FakeQuantaClock::current = nullptr;
+
+// Provisioner with deterministic fakes: the virtual quanta clock above, plus
+// capture of print output and dynamic handler registrations.
+struct TestProvisioner : Provisioner<>, FakeQuantaClock
+{
     // The script-visible stopwatch LLTimers schedules against. Deliberately
     // separate from the quanta clock above, which advances on every reading.
     double script_clock = 0.0;
@@ -58,7 +93,9 @@ struct TestProvisioner : Provisioner<>
         callbacks.performanceClockProvider = script_clock_provider;
         callbacks.setTimerEventCb = set_timer_event;
         callbacks.eventHandlerRegistrationCb = handler_registration;
-        callbacks.quantaClockProvider = quanta_clock;
+        callbacks.quantaClockProvider = FakeQuantaClock::read;
+        // The fake clock only moves when the test says, no use for a watchdog thread
+        callbacks.interruptInstallPolicy = InterruptInstallPolicy::Resident;
         callbacks.populateEnvironment = populate_environment;
         return callbacks;
     }
@@ -71,18 +108,11 @@ struct TestProvisioner : Provisioner<>
         lua_pushcfunction(L, lua_break, "preempt");
         lua_setglobal(L, "preempt");
 
-        lua_pushcfunction(L, simulate_gc_step, "simulate_gc_step");
-        lua_setglobal(L, "simulate_gc_step");
-
         lua_pushcfunction(L, jump_clock, "jump_clock");
         lua_setglobal(L, "jump_clock");
-    }
 
-    static double quanta_clock(lua_State* L)
-    {
-        TestProvisioner& host = of(L);
-        host.clock += host.clock_step;
-        return host.clock;
+        lua_pushcfunction(L, gc_count, "gc_count");
+        lua_setglobal(L, "gc_count");
     }
 
     static double script_clock_provider(lua_State* L)
@@ -106,19 +136,6 @@ struct TestProvisioner : Provisioner<>
         return static_cast<TestProvisioner&>(executor->getProvisioner());
     }
 
-    // Invoke the GC callbacks as if GC is happening, without actually triggering it.
-    static int simulate_gc_step(lua_State* L)
-    {
-        double cost = luaL_checknumber(L, 1);
-        int post_state = luaL_optinteger(L, 2, 0);
-        void (*interrupt)(lua_State*, int) = lua_callbacks(L)->interrupt;
-        REQUIRE((interrupt != nullptr));
-        interrupt(L, 0);
-        of(L).clock += cost;
-        interrupt(L, post_state);
-        return 0;
-    }
-
     // An engine pause the executor has no bracket for: time passes with no
     // exclusion banked, as if the host got descheduled mid-execution.
     static int jump_clock(lua_State* L)
@@ -133,6 +150,13 @@ struct TestProvisioner : Provisioner<>
         return 0;
     }
 
+    // Heap size in KB as the script sees it mid-window
+    static int gc_count(lua_State* L)
+    {
+        lua_pushnumber(L, lua_gc(L, LUA_GCCOUNT, 0));
+        return 1;
+    }
+
     static bool handler_registration(lua_State* L, const char* event_name, bool registered)
     {
         TestProvisioner::of(L).registrations.push_back(std::string(registered ? "+" : "-") + event_name);
@@ -160,21 +184,17 @@ static ScriptConfig makeScriptConfig()
     return config;
 }
 
-// One-line drivers for the begin/call/end run-window brackets the engine API is built around.
+// One-line drivers for a single call inside its own run window.
 static RunResult dispatchRaw(Script& exec, int lsl_state, const char* event_name, PushArgsFn push_args = nullptr, void* ctx = nullptr, double quanta = 1.0)
 {
-    exec.beginRunWindow(quanta);
-    RunResult result = exec.callEventHandler(lsl_state, event_name, push_args, ctx);
-    exec.endRunWindow();
-    return result;
+    RunWindow window(exec, quanta);
+    return exec.callEventHandler(lsl_state, event_name, push_args, ctx);
 }
 
 static RunResult resumeRaw(Script& exec, double quanta = 1.0)
 {
-    exec.beginRunWindow(quanta);
-    RunResult result = exec.resumeEventHandler();
-    exec.endRunWindow();
-    return result;
+    RunWindow window(exec, quanta);
+    return exec.resumeEventHandler();
 }
 
 static RunResult dispatch(Script& exec, int lsl_state, const char* event_name, HandlerRunStatus expect = HandlerRunStatus::Ok,
@@ -206,6 +226,8 @@ static RunResult resumeToCompletion(Script& exec, double quanta, HandlerRunStatu
     RunResult result{HandlerRunStatus::Preempted, 0};
     while (result.status == HandlerRunStatus::Preempted)
     {
+        // Bank-and-zero like a real host would
+        exec.setSleep(0.0f);
         result = resumeRaw(exec, quanta);
         if (preemptions && result.status == HandlerRunStatus::Preempted)
             ++(*preemptions);
@@ -867,9 +889,8 @@ TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor weak references die before OoM")
 
     // Keep the real GC from completing a cycle on its own so any weak
     // clearing is attributable to the memory limit callback's heap walk.
-    // Don't use LUA_GCSTOP, it disables the beforeallocate callback entirely.
-    lua_gc(script->getInstanceState(), LUA_GCSETGOAL, 100000);
     lua_gc(script->getInstanceState(), LUA_GCCOLLECT, 0);
+    lua_gc(script->getInstanceState(), LUA_GCSTOP, 0);
 
     resume(*script);
     CHECK(script->getFaultKind() == FaultKind::None);
@@ -1074,66 +1095,6 @@ TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor mandatory yield kill")
 
 TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor engine pauses are not charged")
 {
-    SUBCASE("unyieldable region survives a GC-inflated window")
-    {
-        // A long GC step lands while execution is inside a metamethod frame,
-        // where a yield can't be injected. GC isn't the executing code's fault,
-        // necessarily, so we discount it against their runtime.
-        TestScript ts(R"(
-            local mt = {}
-            mt.__index = function(t, k)
-                simulate_gc_step(10, 2)
-                local total = 0
-                for i = 1, 20 do
-                    total += i
-                end
-                return total
-            end
-            local obj = setmetatable({}, mt)
-            print(obj.missing)
-        )");
-        ts.loadDefaultState();
-        Script& exec = ts.exec;
-
-        // Charged time only accrues one clock step per interrupt reading, so
-        // it stays far below the punishment thresholds against a 5ms quanta.
-        ts.host.clock_step = 0.0001;
-        resumeToCompletion(exec, 0.005, HandlerRunStatus::Ok);
-        CHECK(exec.getFaultKind() == FaultKind::None);
-        CHECK(exec.getSleep() == 0.0f);
-        checkCapture(ts.host.printed, {"210"});
-    }
-
-    SUBCASE("GC-inflated time is not punished, preemption stays on wall time")
-    {
-        TestScript ts(R"(
-            simulate_gc_step(10)
-            local total = 0
-            for i = 1, 100 do
-                total += i
-            end
-            print("finished")
-        )");
-        ts.loadDefaultState();
-        Script& exec = ts.exec;
-
-        // Wall elapsed is ~10s against a 50ms quanta, so the first yieldable
-        // interrupt still preempts, but the charged handful of clock steps
-        // stays under the 3x threshold, so no sleep punishment lands.
-        ts.host.clock_step = 0.001;
-        resume(exec, 0.05, HandlerRunStatus::Preempted);
-        CHECK(exec.isYieldDue());
-        CHECK(exec.getSleep() == 0.0f);
-        // The banked exclusion is readable until the next window resets it
-        CHECK(exec.getExcludedTime() >= 10.0);
-
-        // Zero out the sleep so we don't deadlock the suite if the earlier
-        // getSleep() == 0.0f check didn't hold.
-        exec.setSleep(0.0f);
-        resumeToCompletion(exec, 0.05);
-        checkCapture(ts.host.printed, {"finished"});
-    }
-
     SUBCASE("baseline heap walk feeds the exclusion accumulator")
     {
         // Trigger a ton of allocations so the lua_userthreadgc() check kicks
@@ -1254,39 +1215,6 @@ TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor mandatory yield grace")
         CHECK(ts.host.printed.empty());
     }
 
-    SUBCASE("excluded time does not consume the grace window")
-    {
-        // The grace deadline must be measured in charged time: a GC step
-        // landing after the deadline is armed must not burn the window.
-        TestScript ts(R"(
-            local mt = {}
-            mt.__index = function(t, k)
-                jump_clock(10)
-                local total = 0
-                for i = 1, 5 do
-                    total += i
-                end
-                simulate_gc_step(10)
-                for i = 1, 20 do
-                    total += i
-                end
-                return total
-            end
-            local obj = setmetatable({}, mt)
-            print(obj.missing)
-        )");
-        ts.loadDefaultState();
-        Script& exec = ts.exec;
-
-        ts.host.clock_step = 0.000001;
-        resume(exec, 0.005, HandlerRunStatus::Preempted);
-        CHECK(exec.getFaultKind() == FaultKind::None);
-        CHECK(exec.getSleep() > 0.0f);
-
-        exec.setSleep(0.0f);
-        resumeToCompletion(exec, 0.005);
-        checkCapture(ts.host.printed, {"225"});
-    }
 }
 
 TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor pending sleep preempts")
@@ -1300,16 +1228,206 @@ TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor pending sleep preempts")
     )");
     ts.loadDefaultState();
 
-    ts.exec.setSleep(1.0f);
-    resume(ts.exec, 1.0, HandlerRunStatus::Preempted);
+    RunResult result;
+    {
+        RunWindow window(ts.exec, 1.0);
+        ts.exec.setSleep(1.0f);
+        result = ts.exec.resumeEventHandler();
+    }
+    REQUIRE(result.status == HandlerRunStatus::Preempted);
     CHECK(ts.exec.isYieldDue());
     // Sleep-driven preemption carries no punishment
     CHECK(ts.exec.getSleep() == 1.0f);
 
-    ts.exec.setSleep(0.0f);
     resumeToCompletion(ts.exec, 1.0);
 }
 
+TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor run window guard")
+{
+    TestScript ts(R"(
+        local total = 0
+        for i = 1, 100 do
+            total += i
+        end
+        print("finished")
+    )");
+    ts.loadDefaultState();
+
+    RunResult result;
+    {
+        RunWindow window(ts.exec, 1.0);
+        ts.exec.setForceYield(true);
+        result = ts.exec.resumeEventHandler();
+    }
+    REQUIRE(result.status == HandlerRunStatus::Preempted);
+    // Scope exit closed the window: GC unparked, force-yield cleared
+    CHECK(ts.exec.getInstanceState()->global->GCthreshold < SIZE_MAX);
+    resumeToCompletion(ts.exec, 1.0);
+    checkCapture(ts.host.printed, {"finished"});
+}
+
+TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor run window catch-up collects the window's garbage")
+{
+    TestScript ts(R"(
+        for i = 1, 20000 do
+            local t = {i, i, i}
+        end
+        peak_kb = gc_count()
+    )");
+    ts.loadDefaultState();
+    lua_State* instance = ts.exec.getInstanceState();
+
+    // Parked for the whole window, so the heap only grows until the end
+    resumeToCompletion(ts.exec, 1.0);
+    lua_getglobal(instance, "peak_kb");
+    int peak_kb = lua_tointeger(instance, -1);
+    lua_pop(instance, 1);
+    REQUIRE(peak_kb > 0);
+
+    // The catch-up loop's postcondition, and proof it actually collected
+    CHECK(!luaC_needsGC(instance));
+    CHECK(lua_gc(instance, LUA_GCCOUNT, 0) < peak_kb);
+}
+
+// Real quanta clock, handler put in at the deadline by a watchdog thread or a
+// timer signal. The policies below all get the same tests.
+static HostCallbacks deadlineCallbacks(InterruptInstallPolicy policy)
+{
+    HostCallbacks callbacks;
+    callbacks.interruptInstallPolicy = policy;
+    return callbacks;
+}
+
+static const std::vector<InterruptInstallPolicy> kDeadlinePolicies = {
+    InterruptInstallPolicy::Threaded,
+#if defined(__linux__)
+    InterruptInstallPolicy::Signal,
+#endif
+};
+
+static const char* kBusyLoop = R"(
+    local total = 0
+    for i = 1, 1000000000 do
+        total += i
+    end
+)";
+
+TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor watchdog preempts a busy script")
+{
+    for (InterruptInstallPolicy policy : kDeadlinePolicies)
+    {
+        INFO("policy ", (int)policy);
+        TestProvisioner host{deadlineCallbacks(policy)};
+        std::shared_ptr<Script> script = host.provisionScript(makeImageConfig(Luau::compile(kBusyLoop)), makeScriptConfig());
+        REQUIRE(script != nullptr);
+        REQUIRE(script->loadDefaultState());
+
+        // 1e9 iterations can't finish inside 2ms, so preemption is the only way out
+        RunResult result = resumeRaw(*script, 0.002);
+        REQUIRE(result.status == HandlerRunStatus::Preempted);
+        CHECK(script->isYieldDue());
+        CHECK(script->getInstanceState()->global->GCthreshold < SIZE_MAX);
+        // The fired handler doesn't outlive the window
+        CHECK((lua_callbacks(script->getInstanceState())->interrupt == nullptr));
+        CHECK(host.getWatchdogStats().fires == 1);
+    }
+}
+
+TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor watchdog honors a mid-window sleep")
+{
+    for (InterruptInstallPolicy policy : kDeadlinePolicies)
+    {
+        INFO("policy ", (int)policy);
+        TestProvisioner host{deadlineCallbacks(policy)};
+        std::shared_ptr<Script> script = host.provisionScript(makeImageConfig(Luau::compile(kBusyLoop)), makeScriptConfig());
+        REQUIRE(script != nullptr);
+        REQUIRE(script->loadDefaultState());
+
+        // Quanta generous enough that only the sleep can preempt us
+        RunResult result;
+        {
+            RunWindow window(*script, 10.0);
+            script->setSleep(0.5f);
+            result = script->resumeEventHandler();
+        }
+        REQUIRE(result.status == HandlerRunStatus::Preempted);
+        CHECK(script->isYieldDue());
+    }
+}
+
+TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor watchdog uninstalls a withdrawn force-yield")
+{
+    for (InterruptInstallPolicy policy : kDeadlinePolicies)
+    {
+        INFO("policy ", (int)policy);
+        TestProvisioner host{deadlineCallbacks(policy)};
+        std::shared_ptr<Script> script = host.provisionScript(makeImageConfig(Luau::compile(R"(
+            local total = 0
+            for i = 1, 1000 do
+                total += i
+            end
+        )")), makeScriptConfig());
+        REQUIRE(script != nullptr);
+        REQUIRE(script->loadDefaultState());
+
+        // The set installs the handler. The first safepoint finds nothing to do
+        // and the deadline far away, so it uninstalls itself.
+        RunResult result;
+        {
+            RunWindow window(*script, 10.0);
+            script->setForceYield(true);
+            script->setForceYield(false);
+            result = script->resumeEventHandler();
+        }
+        REQUIRE(result.status == HandlerRunStatus::Ok);
+        CHECK(!script->isYieldDue());
+        CHECK((lua_callbacks(script->getInstanceState())->interrupt == nullptr));
+    }
+}
+
+TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor deadline installers honor the host fire lead")
+{
+    for (InterruptInstallPolicy policy : kDeadlinePolicies)
+    {
+        INFO("policy ", (int)policy);
+        HostCallbacks callbacks = deadlineCallbacks(policy);
+        // A lead longer than the window means the deadline is already inside
+        // it when the window opens, so the handler goes in up front and the
+        // first safepoint yields.
+        callbacks.interruptFireLead = 1.0;
+        TestProvisioner host{callbacks};
+        std::shared_ptr<Script> script = host.provisionScript(makeImageConfig(Luau::compile(kBusyLoop)), makeScriptConfig());
+        REQUIRE(script != nullptr);
+        REQUIRE(script->loadDefaultState());
+
+        RunResult result = resumeRaw(*script, 0.002);
+        REQUIRE(result.status == HandlerRunStatus::Preempted);
+        CHECK(script->isYieldDue());
+        WatchdogStats stats = host.getWatchdogStats();
+        CHECK(stats.fires == 1);
+        CHECK(stats.fireLead == 1.0);
+    }
+}
+
+#if defined(__linux__)
+TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor signal installer handles a deadline that is already due")
+{
+    TestProvisioner host{deadlineCallbacks(InterruptInstallPolicy::Signal)};
+    std::shared_ptr<Script> script = host.provisionScript(makeImageConfig(Luau::compile(kBusyLoop)), makeScriptConfig());
+    REQUIRE(script != nullptr);
+    REQUIRE(script->loadDefaultState());
+
+    // A zero quanta puts the deadline inside the fire lead before the timer
+    // could be asked for anything, so the install has to happen inline and
+    // the first safepoint yields.
+    RunResult result = resumeRaw(*script, 0.0);
+    REQUIRE(result.status == HandlerRunStatus::Preempted);
+    CHECK(script->isYieldDue());
+    CHECK(host.getWatchdogStats().fires == 1);
+    CHECK((lua_callbacks(script->getInstanceState())->interrupt == nullptr));
+}
+#endif
+
 TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor memory limit")
 {
     std::string bytecode = Luau::compile(R"(
@@ -1402,20 +1520,23 @@ TEST_CASE_FIXTURE(SLuaFixture, "SLExecutor pushes args outside a resume")
 
     resume(ts.exec);
 
-    // A pending forced yield makes the first interrupt mandatory, so one reaching
-    // the marshalling would kill the script rather than pass unnoticed. The first
-    // one that does fire lands in `handleEvent`, which is yieldable.
-    ts.exec.setForceYield(true);
-
-    RunResult result = dispatchRaw(ts.exec, 0, "dataserver", [](lua_State* handler, void*) {
-        luaSL_pushuuidstring(handler, "8d4a1e26-3f2b-4c7d-9a15-6e0b3c8f2d41");
-        lua_pushstring(handler, "payload");
-    });
+    // A forced yield makes the first interrupt mandatory, so one reaching the
+    // marshalling would kill the script rather than pass unnoticed. The first
+    // one that does fire lands in `handleEvent`, which is yieldable. Driven
+    // by hand because force-yield can only be set inside a window.
+    RunResult result;
+    {
+        RunWindow window(ts.exec, 1.0);
+        ts.exec.setForceYield(true);
+        result = ts.exec.callEventHandler(0, "dataserver", [](lua_State* handler, void*) {
+            luaSL_pushuuidstring(handler, "8d4a1e26-3f2b-4c7d-9a15-6e0b3c8f2d41");
+            lua_pushstring(handler, "payload");
+        });
+    }
     REQUIRE(result.status == HandlerRunStatus::Preempted);
     CHECK(ts.exec.getFaultKind() == FaultKind::None);
     CHECK(ts.host.printed.empty());
 
-    ts.exec.setForceYield(false);
     resumeToCompletion(ts.exec, 1.0);
     checkCapture(ts.host.printed, {"uuid/payload"});
 }
@@ -1963,10 +2084,8 @@ class StatefulImage : public Image
 
 // Replaces every tier. TestProvisioner's callbacks recover it by downcasting
 // getProvisioner() to itself, so a provisioner of other types needs its own.
-struct StatefulProvisioner : Provisioner<StatefulScript>
+struct StatefulProvisioner : Provisioner<StatefulScript>, FakeQuantaClock
 {
-    double clock = 0.0;
-    double clock_step = 0.0;
     int environments_made = 0;
     int images_made = 0;
 
@@ -1978,20 +2097,11 @@ struct StatefulProvisioner : Provisioner<StatefulScript>
     static HostCallbacks makeCallbacks()
     {
         HostCallbacks callbacks;
-        callbacks.quantaClockProvider = quanta_clock;
+        callbacks.quantaClockProvider = FakeQuantaClock::read;
+        callbacks.interruptInstallPolicy = InterruptInstallPolicy::Resident;
         return callbacks;
     }
 
-    static double quanta_clock(lua_State* L)
-    {
-        Script* executor = Script::fromLuaState(L);
-        LUAU_ASSERT(executor != nullptr);
-
-        auto& host = static_cast<StatefulProvisioner&>(executor->getProvisioner());
-        host.clock += host.clock_step;
-        return host.clock;
-    }
-
     // The only place the downcast happens, since everything this provisioner
     // mints comes through makeScript()
     std::shared_ptr<StatefulScript> provision(const std::string& bytecode)