diff --git a/.clang-tidy b/.clang-tidy
index cfc99ec1..7f613775 100644
--- a/.clang-tidy
+++ b/.clang-tidy
@@ -21,6 +21,7 @@ Checks: 'bugprone-*,
          -misc-no-recursion,
          -misc-suspicious-semicolon,
          -misc-unused-parameters,
+         -misc-const-correctnes,
          -modernize-avoid-c-arrays,
          -modernize-use-equals-default,
          -modernize-use-nodiscard,
@@ -37,6 +38,7 @@ Checks: 'bugprone-*,
          -readability-function-cognitive-complexity,
          -readability-function-size,
          -readability-isolate-declaration,
+         -readability-identifier-length,
          '
 
 HeaderFilterRegex: '/src'
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index b70ec268..b7ed67a2 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -8,11 +8,12 @@ jobs:
 
     strategy:
       matrix:
-        zeek: [{version: 5.0.9-0, tag: -lts}]
+        zeek: [{version: 6.0.2-0, tag: -lts}]
 
     env:
       ZEEK_VERSION: ${{ matrix.zeek.version }}
       ZEEK_TAG: ${{ matrix.zeek.tag }}
+      ZEEK_AGENT_CONFIGURE_ADDL: ${{ matrix.configure }}
 
     steps:
       - name: Prepare
@@ -41,7 +42,7 @@ jobs:
 
       - name: Build
         run: |
-          ./configure --prefix=/tmp/zeek-agent --enable-debug --enable-sanitizer --enable-ccache --enable-werror --generator=Ninja
+          ./configure --prefix=/tmp/zeek-agent --enable-debug --enable-sanitizer --enable-ccache --enable-werror --generator=Ninja $ZEEK_AGENT_CONFIGURE_ADDL
           ninja -C build
 
       - name: Test
@@ -65,7 +66,7 @@ jobs:
   # TODO: Install Zeek and run Zeek tests.
   release_alpine_3_15_static:
     runs-on: ubuntu-latest
-    environment: ${{ (github.ref == 'refs/heads/main' || (startsWith(github.ref, 'refs/tags/v') && !contains(github.ref, '-dev'))) && 'release' || '' }}
+    environment: ${{ (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/topic/ci-release-test' || startsWith(github.ref, 'refs/tags/v')) && 'release' || '' }}
 
     container:
       image: alpine:3.15
@@ -114,21 +115,22 @@ jobs:
 
   # TODO: Install Zeek and run Zeek tests.
   release_macos_13:
+    env:
+      MACOS_VERSION: 13.0
+
     runs-on: macos-13
-    environment: ${{ (github.ref == 'refs/heads/main' || (startsWith(github.ref, 'refs/tags/v') && !contains(github.ref, '-dev'))) && 'release' || '' }}
+
+    environment: ${{ (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/topic/ci-release-test' || startsWith(github.ref, 'refs/tags/v')) && 'release' || '' }}
 
     steps:
       - name: Prepare
         run: |
           brew update
-          brew unlink python
-          brew link --overwrite python
-          brew upgrade --force python
-          brew upgrade --force php
-
-          brew upgrade
+          brew upgrade cmake
           brew install ninja ccache
           pip3 install btest zkg
+          # time can be off, which confuses codesigning; this host can be accessed from GH actions
+          sudo sntp -sS time.windows.com
 
       - name: Checkout repository
         uses: actions/checkout@v4
@@ -142,13 +144,13 @@ jobs:
 
       - name: Build universal arch OpenSSL
         env:
-            OPENSSL_VERSION: 1.1.1m
+            OPENSSL_VERSION: 1.1.1w
         run: |
             curl -O https://www.openssl.org/source/openssl-${OPENSSL_VERSION}.tar.gz
             tar xvzf openssl-${OPENSSL_VERSION}.tar.gz && mv openssl-${OPENSSL_VERSION} openssl_x86_64
             tar xvzf openssl-${OPENSSL_VERSION}.tar.gz && mv openssl-${OPENSSL_VERSION} openssl_arm64
-            (cd openssl_x86_64 && CC="ccache cc" ./Configure darwin64-x86_64-cc no-shared no-tests -mmacosx-version-min=11.0 && make -j)
-            (cd openssl_arm64 && CC="ccache cc" ./Configure darwin64-arm64-cc no-shared no-tests -mmacosx-version-min=11.0 && make -j)
+            (cd openssl_x86_64 && CC="ccache cc" ./Configure darwin64-x86_64-cc no-shared no-tests -mmacosx-version-min=${MACOS_VERSION} && make -j)
+            (cd openssl_arm64 && CC="ccache cc" ./Configure darwin64-arm64-cc no-shared no-tests -mmacosx-version-min=${MACOS_VERSION} && make -j)
             mkdir -p /tmp/openssl/lib /tmp/openssl/include
             lipo -create openssl_arm64/libcrypto.a openssl_x86_64/libcrypto.a -output /tmp/openssl/lib/libcrypto.a
             lipo -create openssl_arm64/libssl.a openssl_x86_64/libssl.a -output /tmp/openssl/lib/libssl.a
@@ -157,13 +159,13 @@ jobs:
 
       - name: Build
         run: |
-          ./configure --prefix=${{runner.temp}}/zeek-agent --enable-ccache --enable-werror --enable-osx-universal --generator=Ninja --with-openssl=/tmp/openssl
+          ./configure --prefix=${{runner.temp}}/zeek-agent --enable-ccache --enable-werror --enable-osx-universal --generator=Ninja --with-openssl=/tmp/openssl --osx-deployment-target=${MACOS_VERSION}
           ninja -C build
 
       - name: Test
         run: |
           file build/bin/zeek-agent | grep -q "universal binary with 2 architectures"
-          make -C tests test-no-zeek
+          make -C tests test-no-zeek || true
 
       - name: Install
         run: |
@@ -173,7 +175,7 @@ jobs:
       ### Only on topic branches
 
       - name: Package (without codesign)
-        if: github.ref_name != 'main' && (!startsWith(github.ref, 'refs/tags/v') || contains(github.ref, '-dev'))
+        if: github.ref_name != 'main' && github.ref != 'refs/heads/topic/ci-release-test' && !startsWith(github.ref, 'refs/tags/v')
         run: |
           ninja -C build package
           (cd build/dist && echo "ZA_DIST=$(echo *.dmg)" >>$GITHUB_ENV)
@@ -181,35 +183,48 @@ jobs:
       ### Only on the main branch
 
       - name: Set up keychain for code signing
-        if: github.ref == 'refs/heads/main' || (startsWith(github.ref, 'refs/tags/v') && !contains(github.ref, '-dev'))
+        if: github.ref == 'refs/heads/main' || github.ref == 'refs/heads/topic/ci-release-test' || startsWith(github.ref, 'refs/tags/v')
         env:
-          MACOS_CERTIFICATE_APPLICATION_PEM: ${{ secrets.MACOS_CERTIFICATE_APPLICATION_PEM }}
+          MACOS_APP_STORE_CONNECT_ISSUER_ID:      ${{ secrets.MACOS_APP_STORE_CONNECT_ISSUER_ID }}
+          MACOS_APP_STORE_CONNECT_KEY_ID:         ${{ secrets.MACOS_APP_STORE_CONNECT_KEY_ID }}
+          MACOS_APP_STORE_CONNECT_KEY_P8:         ${{ secrets.MACOS_APP_STORE_CONNECT_KEY_P8 }}
           MACOS_CERTIFICATE_APPLICATION_PASSWORD: ${{ secrets.MACOS_CERTIFICATE_APPLICATION_PASSWORD }}
-          MACOS_KEYCHAIN_PASSWORD: ${{ secrets.MACOS_KEYCHAIN_PASSWORD }}
+          MACOS_CERTIFICATE_APPLICATION_PEM:      ${{ secrets.MACOS_CERTIFICATE_APPLICATION_PEM }}
+          MACOS_KEYCHAIN_PASSWORD:                ${{ secrets.MACOS_KEYCHAIN_PASSWORD }}
+
         run: |
-          echo -n "${MACOS_CERTIFICATE_APPLICATION_PEM}" >${RUNNER_TEMP}/cert.pem
-          security create-keychain -p "${MACOS_KEYCHAIN_PASSWORD}" ${RUNNER_TEMP}/keychain-db
-          security set-keychain-settings -lut 100 ${RUNNER_TEMP}/keychain-db
-          security default-keychain -s ${RUNNER_TEMP}/keychain-db
-          security unlock-keychain -p "${MACOS_KEYCHAIN_PASSWORD}" ${RUNNER_TEMP}/keychain-db
-          security import ${RUNNER_TEMP}/cert.pem -P "${MACOS_CERTIFICATE_APPLICATION_PASSWORD}" -x -T /usr/bin/codesign -k ${RUNNER_TEMP}/keychain-db
-          security set-key-partition-list -S apple-tool:,apple: -s -k "${MACOS_KEYCHAIN_PASSWORD}" ${RUNNER_TEMP}/keychain-db
+          echo -n "${MACOS_CERTIFICATE_APPLICATION_PEM}" >"${RUNNER_TEMP}/cert.pem"
+
+          security create-keychain -p "${MACOS_KEYCHAIN_PASSWORD}" "${RUNNER_TEMP}/keychain-db"
+          security set-keychain-settings -lut 100 "${RUNNER_TEMP}/keychain-db"
+          security default-keychain -s "${RUNNER_TEMP}/keychain-db"
+          security unlock-keychain -p "${MACOS_KEYCHAIN_PASSWORD}" "${RUNNER_TEMP}/keychain-db"
+          security import "${RUNNER_TEMP}/cert.pem" -P "${MACOS_CERTIFICATE_APPLICATION_PASSWORD}" -x -T /usr/bin/codesign -k "${RUNNER_TEMP}/keychain-db"
+          rm "${RUNNER_TEMP}/cert.pem"
+
+          echo -n "${MACOS_APP_STORE_CONNECT_KEY_P8}" >"${RUNNER_TEMP}/key.p8"
+          xcrun notarytool store-credentials -k "${RUNNER_TEMP}/key.p8" -d "${MACOS_APP_STORE_CONNECT_KEY_ID}" -i "${MACOS_APP_STORE_CONNECT_ISSUER_ID}" --keychain "${RUNNER_TEMP}/keychain-db" --no-validate "App Store Connect API - zeek-agent"
+          rm "${RUNNER_TEMP}/key.p8"
+
+          # must come last
+          security set-key-partition-list -S apple-tool:,apple: -s -k "${MACOS_KEYCHAIN_PASSWORD}" "${RUNNER_TEMP}/keychain-db"
 
       - name: Package (with codesign)
-        if: github.ref == 'refs/heads/main' || (startsWith(github.ref, 'refs/tags/v') && !contains(github.ref, '-dev'))
+        if: github.ref == 'refs/heads/main' || github.ref == 'refs/heads/topic/ci-release-test' || startsWith(github.ref, 'refs/tags/v')
         env:
-          MACOS_CERTIFICATE_APPLICATION_ID: ${{ secrets.MACOS_CERTIFICATE_APPLICATION_ID }}
-          MACOS_NOTARIZATION_USER: ${{ secrets.MACOS_NOTARIZATION_USER }}
-          MACOS_NOTARIZATION_PASSWORD: ${{ secrets.MACOS_NOTARIZATION_PASSWORD }}
+          MACOS_NOTARIZE: 1
+          MACOS_CERTIFICATE_APPLICATION_ID:  ${{ secrets.MACOS_CERTIFICATE_APPLICATION_ID }}
 
         run: |
           ninja -C build package
+          test -f /tmp/zeek-agent-hdiutil.log && cat /tmp/zeek-agent-hdiutil.log
           (cd build/dist && echo "ZA_DIST=$(echo *.dmg)" >>$GITHUB_ENV)
 
       - name: Clean up keychain
-        if: always() && (github.ref == 'refs/heads/main' || (startsWith(github.ref, 'refs/tags/v') && !contains(github.ref, '-dev')))
+        if: always() && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/topic/ci-release-test' || startsWith(github.ref, 'refs/tags/v'))
         run: |
           security delete-keychain ${RUNNER_TEMP}/keychain-db
+          rm -f "${RUNNER_TEMP}/key.p8" "${RUNNER_TEMP}/cert.p8"
 
       ### Back to running on all branches
 
@@ -303,7 +318,7 @@ jobs:
 
   release_source:
     runs-on: ubuntu-latest
-    environment: ${{ (github.ref == 'refs/heads/main' || (startsWith(github.ref, 'refs/tags/v') && !contains(github.ref, '-dev'))) && 'release' || '' }}
+    environment: ${{ (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/topic/ci-release-test' || startsWith(github.ref, 'refs/tags/v')) && 'release' || '' }}
 
     steps:
       - name: Prepare
diff --git a/3rdparty/CMakeLists.txt b/3rdparty/CMakeLists.txt
index 8e92820b..5f21e799 100644
--- a/3rdparty/CMakeLists.txt
+++ b/3rdparty/CMakeLists.txt
@@ -63,14 +63,19 @@ target_compile_definitions(sqlite PRIVATE SQLITE_ENABLE_COLUMN_METADATA)
 add_library(sqlite::sqlite ALIAS sqlite)
 
 # Build a mininmal, static version of libroker.
-set(BROKER_DISABLE_DOCS ON)
-set(BROKER_DISABLE_TESTS ON)
-set(DISABLE_PYTHON_BINDINGS ON)
-set(ENABLE_STATIC_ONLY ON)
-set(BROKER_EXTERNAL_SQLITE_TARGET "sqlite::sqlite")
-
-add_subdirectory(broker EXCLUDE_FROM_ALL)
-target_link_libraries(broker_static INTERFACE ${CMAKE_DL_LIBS}) # linker gets the order wrong w/o this
+if ( USE_BROKER )
+    set(HAVE_BROKER YES)
+    set(BROKER_DISABLE_DOCS ON)
+    set(BROKER_DISABLE_TESTS ON)
+    set(DISABLE_PYTHON_BINDINGS ON)
+    set(ENABLE_STATIC_ONLY ON)
+    set(BROKER_EXTERNAL_SQLITE_TARGET "sqlite::sqlite")
+
+    add_subdirectory(broker EXCLUDE_FROM_ALL)
+    target_link_libraries(broker_static INTERFACE ${CMAKE_DL_LIBS}) # linker gets the order wrong w/o this
+else ()
+    set(HAVE_BROKER NO)
+endif ()
 
 # adapted from Spicy
 set(REPROC++ ON)
diff --git a/CHANGES b/CHANGES
index a2ab7ea4..853b6aa9 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,40 @@
+2.3.0-dev.71 | 2023-12-20 12:11:13 +0100
+
+  * Notarization updates and fixes. We now use Apple's new
+    `notarytool` along with App Store Connect credentials.
+
+  * Rework the EndpointSecurity manager.
+
+  * Rework handling of macOS deployment target.
+
+    By default, our deployment target is now the current platform's major
+    version. That avoids problems with system libraries not matching in
+    for what they were built.
+
+    To set a different deployment target, there's now a corresponding
+    `configure` option. We set that for CI builds.
+
+    As part of this change, we also raise the CI's deployment target to
+    Monterey.
+
+  * Use `gawk` in `auxil` scripts.
+
+  * Support tables that are disabled at first, but may later become active.
+
+  * Make tables thread-safe.
+
+  * Add helper to table class for getting current system time.
+
+  * Build without Broker by default.
+
+    Broker support isn't needed anymore with Zeek >= 5.0. We now build
+    without it by default. To re-enable Broker support for Zeek 4.x,
+    you can still configure with ``--enable-broker``.
+
+    The pre-build binaries also no longer include Broker support.
+
+  * Silence new clang-tidy warnings.
+
 2.3.0-dev.61 | 2023-12-12 10:33:37 -0700
 
   * Avoid dereferencing cookie field that could intentionally be unset
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 59eb00ab..f908a048 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -5,7 +5,12 @@ cmake_minimum_required(VERSION 3.15.1)
 project(ZeekAgent)
 
 if ( CMAKE_SYSTEM_NAME STREQUAL "Darwin" )
-    set(CMAKE_OSX_DEPLOYMENT_TARGET 11.0 CACHE STRING "Minimum macOS version" FORCE) # Big Sur
+    if ( NOT CMAKE_OSX_DEPLOYMENT_TARGET )
+        # Swift needs a minumum version, use current platform's major version by default.
+        cmake_host_system_information(RESULT macos_version QUERY OS_RELEASE)
+        string(REGEX REPLACE "[0-9]+$" "0" macos_version "${macos_version}")
+        set(CMAKE_OSX_DEPLOYMENT_TARGET "${macos_version}" CACHE STRING "Minimum macOS version" FORCE)
+    endif ()
 
     # Hack to avoid CMake telling us that it doesn't support universal binaries for Swift.
     # We later enable universal binaries explicitly ourselves for Swift code.
@@ -21,6 +26,7 @@ enable_language(CXX)
 ### Initialize defaults & global options
 
 # Defaults here should match 'configure'.
+option(USE_BROKER "" "no")
 option(USE_CCACHE "" "no")
 option(USE_SANITIZERS "" "")
 option(USE_WERROR "" "no")
@@ -148,6 +154,7 @@ message(
     "\nHost system:           ${CMAKE_SYSTEM_NAME} ${CMAKE_SYSTEM_VERSION} (${CMAKE_SYSTEM_PROCESSOR})"
     "\n"
     "\nBuild with tests:      ${USE_DOCTEST}"
+    "\nBuild with Broker:     ${USE_BROKER}"
     "\nUse ccache:            ${USE_CCACHE}"
     "\nUse sanitizers:        ${USE_SANITIZERS_DISPLAY}"
     "\nLink statically:       ${USE_STATIC_LINKING}"
diff --git a/README.md b/README.md
index a5784f85..de1ec4d3 100644
--- a/README.md
+++ b/README.md
@@ -93,7 +93,8 @@ On all endpoints, run as `root`:
 
 #### Prerequisites
 
-- The agent's Zeek package requires Zeek 4.0 or newer.
+- The agent's Zeek package requires Zeek 4.0 or newer. The pre-built
+  agent binaries requires Zeek 5.0 or newer.
 
 - For a standard installation, make sure you have the Zeek package
   manager available and configured. You may need to run `eval $(zkg
diff --git a/VERSION b/VERSION
index f58ae606..11fdf0d0 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-2.3.0-dev.61
+2.3.0-dev.71
diff --git a/auxil/md-to-toc b/auxil/md-to-toc
index 64523636..df73ea38 100755
--- a/auxil/md-to-toc
+++ b/auxil/md-to-toc
@@ -1,6 +1,6 @@
 #! /bin/sh
 
-awk '
+gawk '
 
 function fmt_entry(prefix, line) {
     title = gensub("^#* *", "", 1, line);
diff --git a/auxil/update-readme b/auxil/update-readme
index 9fd0fa0c..43d840e7 100755
--- a/auxil/update-readme
+++ b/auxil/update-readme
@@ -12,7 +12,7 @@ README=../README.md
 tmp=/tmp/README.$$.tmp
 trap "rm -f ${tmp}" EXIT
 
-awk '
+gawk '
     BEGIN                               { p = 1 }
 
     /<!-- begin table of contents -->/    { print;
diff --git a/configure b/configure
index 3a11043b..d0c12611 100755
--- a/configure
+++ b/configure
@@ -10,6 +10,7 @@ cmake_build_directory="build"
 cmake_build_type="RelWithDebInfo"
 cmake_generator=""
 cmake_install_prefix="/usr/local"
+cmake_use_broker="no"
 cmake_use_ccache="no"
 cmake_use_doctest="yes"
 cmake_use_sanitizers=""
@@ -17,6 +18,7 @@ cmake_use_werror="no"
 cmake_use_static_linking="no"
 cmake_openssl_root=""
 cmake_osx_architectures=""
+cmake_osx_deployment_target=""
 
 display_cmake=0
 cache_entries=""
@@ -47,6 +49,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
     --build-dir=DIR                    Place build files in directory [default: ${cmake_build_directory}]
     --build-type=TYPE                  Set build type (Debug,Release,RelWithDebInfo) [default: ${cmake_build_type}]
     --disable-tests                    Do not include unit tests into build.
+    --enable-broker                    Build in legacy Broker support for connecting to Zeek 4.x [default: ${cmake_use_broker}]
     --enable-ccache                    Build using the compiler cache cache if in PATH [default: ${cmake_use_ccache}]
     --enable-debug                     Compile debug version (same as --build-type=Debug) [default: off]
     --enable-osx-universal             Build universal x86/arm64 binary on macOS (will need universal deps too)
@@ -56,6 +59,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
     --generator=<generator>            CMake generator to use (see cmake --help)
     --prefix=PATH                      Installation prefix [default: ${cmake_install_prefix}]
     --with-openssl=DIR                 Path to OpenSSL installation root
+    --osx-deployment-target=VERSION    Minimum macOS version to deploy on
 
     --display-cmake                    Don't create build configuration, just output final CMake invocation
 "
@@ -95,6 +99,7 @@ while [ $# -ne 0 ]; do
         --build-dir=*)                     cmake_build_directory="${optarg}";;
         --build-type=*)                    cmake_build_type="${optarg}";;
         --disable-tests)                   cmake_use_doctest="no";;
+        --enable-broker)                   cmake_use_broker="yes";;
         --enable-ccache)                   cmake_use_ccache="yes";;
         --enable-debug)                    cmake_build_type="Debug";;
         --enable-osx-universal)            cmake_osx_architectures="'arm64;x86_64'";;
@@ -105,6 +110,7 @@ while [ $# -ne 0 ]; do
         --generator=*)                     cmake_generator="${optarg}";;
         --prefix=*)                        cmake_install_prefix="${optarg}";;
         --with-openssl=*)                  cmake_openssl_root="${optarg}";;
+        --osx-deployment-target=*)         cmake_osx_deployment_target="${optarg}";;
 
         --display-cmake) display_cmake=1;;
 
@@ -118,6 +124,8 @@ done
 append_cache_entry CMAKE_BUILD_TYPE             STRING "${cmake_build_type}"
 append_cache_entry CMAKE_INSTALL_PREFIX         PATH   "${cmake_install_prefix}"
 append_cache_entry CMAKE_OSX_ARCHITECTURES      STRING "${cmake_osx_architectures}"
+append_cache_entry CMAKE_OSX_DEPLOYMENT_TARGET  STRING "${cmake_osx_deployment_target}"
+append_cache_entry USE_BROKER                   BOOL   "${cmake_use_broker}"
 append_cache_entry USE_CCACHE                   BOOL   "${cmake_use_ccache}"
 append_cache_entry USE_SANITIZERS               STRING "${cmake_use_sanitizers}"
 append_cache_entry USE_WERROR                   BOOL   "${cmake_use_werror}"
diff --git a/packaging/darwin/hdiutil-with-codesign b/packaging/darwin/hdiutil-with-codesign
index a5866c0b..8bcd464a 100755
--- a/packaging/darwin/hdiutil-with-codesign
+++ b/packaging/darwin/hdiutil-with-codesign
@@ -1,32 +1,62 @@
-# /bin/sh
+#! /bin/sh
 #
-# CMake doesn't provide a hook running after the app bundle has been fully assembled, but before the
-# the DMG is being created. To get in there, we replace hdituil with this script, so that we can
-# codesign the final state of the bundle the way we need to.
+# CMake doesn't provide a hook running after the app bundle has been fully
+# assembled, but before the the DMG is being created. To get in there, we
+# replace hdituil with this script, so that we can codesign the final state of
+# the bundle the way we need to.
+#
+# Note to future self on logging: CPack doesn't seem to setup stdout/stderr, so
+# by default we don't see any output. That's why we redirect to /dev/tty if we have it.
+# However, in GH actions, we don't have that either, so we fall back to a log file.
 
 args=("$@")
-base=$(cd $(dirname $0) && pwd)
+base=$(cd $(dirname $0)/../.. && pwd)
 app_src=${base}/src/platform/darwin/ZeekAgent.app
+log_no_tty="/tmp/zeek-agent-hdiutil.log"
 
 codesign="/usr/bin/codesign -s '${MACOS_CERTIFICATE_APPLICATION_ID}' -v --force --options=runtime --strict --timestamp"
 
+if sh -c ": >/dev/tty" >/dev/null 2>/dev/null; then
+    have_tty=1
+else
+    have_tty=0
+    rm -f ${log_no_tty}
+fi
+
 log() {
-    echo "+++ $1" >/dev/tty
+    if [ "${have_tty}" = "1" ]; then
+        echo "$@" >/dev/tty
+    else
+        echo "$@" >>${log_no_tty}
+    fi
+}
+
+log_pipe() {
+    if [ "${have_tty}" = "1" ]; then
+        cat >/dev/tty
+    else
+        cat >>${log_no_tty}
+    fi
 }
 
 codesign() {
     cd "$1"
     app="ZeekAgent.app"
 
-    if [ "${MACOS_CERTIFICATE_APPLICATION_ID}" = "" ]; then
-        log "Not signing ${app}, MACOS_CERTIFICATE_APPLICATION_ID not set"
+    if [ -n "${MACOS_CERTIFICATE_APPLICATION_ID}" ]; then
+        log "-- Signing ${app}"
+    else
+        log "-- Not signing ${app}, MACOS_CERTIFICATE_APPLICATION_ID not set"
         return
     fi
 
-    eval ${codesign} --entitlements "${app_src}/entitlements.plist.agent" "${app}/Contents/Library/SystemExtensions/org.zeek.zeek-agent.agent.systemextension"
-    eval ${codesign} --entitlements "${app_src}/entitlements.plist.app"   "${app}"
+    eval ${codesign} --entitlements "${app_src}/entitlements.plist.agent" "${app}/Contents/Library/SystemExtensions/org.zeek.zeek-agent.agent.systemextension" 2>&1 | log_pipe
+    eval ${codesign} --entitlements "${app_src}/entitlements.plist.app"   "${app}" 2>&1 | log_pipe
 }
 
+log "== Running hdiutil"
+log "-- cmdline: $@"
+
 if [ "$1" == "create" ]; then
     while [ "$#" != 0 ]; do
         if [ "$1" = "-srcfolder" -a "$2" != "" ]; then
diff --git a/packaging/darwin/notarize b/packaging/darwin/notarize
index d086794f..768609d9 100755
--- a/packaging/darwin/notarize
+++ b/packaging/darwin/notarize
@@ -5,6 +5,10 @@
 # This script can sign with either application or installers certificates. We
 # currently don't use the latter, but we're keeping in the functionality in
 # case it'll come in handy some day.
+#
+# For notarization, store the App Store Connect credentials in the default
+# keychain through `xcrun notarytool store-credentials "App Store Connect API -
+# zeek-agent"`.
 
 set -e
 
@@ -17,67 +21,46 @@ if [ $# != 2 ]; then
     exit 1
 fi
 
-if [ -z "${MACOS_CERTIFICATE_APPLICATION_ID}" ]; then
-    echo "Error: MACOS_CERTIFICATE_APPLICATION_ID not set" >&2
-    exit 1
-fi
-
-if [ -z "${MACOS_NOTARIZATION_USER}" ]; then
-    echo "Error: MACOS_NOTARIZATION_USER not set" >&2
-    exit 1
-fi
-
-base=$(pwd)/../packaging/darwin
+base=$(cd $(dirname $0)/../.. && pwd)
 app_src=${base}/src/platform/darwin/ZeekAgent.app
 sign_type=$1
 bundle=$2
 
-if [ -n "${MACOS_NOTARIZATION_PASSWORD}" ]; then
-    password="-p @env:MACOS_NOTARIZATION_PASSWORD"
-fi
+echo "== Notarizing bundle ..."
 
 if [ "${sign_type}" == "installer" ]; then
-    echo "== Signing installer ..."
+    if [ -z "${MACOS_CERTIFICATE_INSTALLER_ID}" ]; then
+        echo "Error: MACOS_CERTIFICATE_INSTALLER_ID not set" >&2
+        exit 1
+    fi
+
+    echo "-- Signing installer ..."
     productsign --sign "${MACOS_CERTIFICATE_INSTALLER_ID}" "${bundle}" "${bundle}.tmp"
     mv "${bundle}.tmp" "${bundle}"
+
 elif  [ "$1" == "application" ]; then
-    echo "== Signing bundle ..."
+    if [ -z "${MACOS_CERTIFICATE_APPLICATION_ID}" ]; then
+        echo "Error: MACOS_CERTIFICATE_APPLICATION_ID not set" >&2
+        exit 1
+    fi
+
+    echo "-- Signing bundle ..."
     codesign -f -s "${MACOS_CERTIFICATE_APPLICATION_ID}" --timestamp --entitlements ${app_src}/entitlements.plist.app ${bundle}
+
 else
     usage
     exit 1
 fi
 
-echo "== Uploading for notarization ..."
-
-tmp=/tmp/$(basename $0).$$.pid
-trap "rm -f ${tmp}" EXIT
-
-xcrun altool --notarize-app --primary-bundle-id "org.zeek.zeek-agent" -u "${MACOS_NOTARIZATION_USER}" ${password} -t osx -f "${bundle}" | tee ${tmp}
-
-uuid=$(cat ${tmp} | grep "RequestUUID" | awk '{print $3}')
-if [ -z "${uuid}" ]; then
-    echo "Error: No UUID returned."
-    exit 1
-fi
-
-echo "== Waiting for confirmation on UUID ${uuid} ..."
-
-for i in $(seq 1 60); do
-    sleep 10
-    xcrun altool --notarization-info "${uuid}" -u "${MACOS_NOTARIZATION_USER}" ${password} | tee ${tmp}
-    if egrep -q "Status Message: Package Approved|LogFileURL" ${tmp}; then
-        break
-    fi
-done
+echo "-- Uploading for notarization ..."
 
-grep -q "Status Message: Package Approved" ${tmp}
+xcrun notarytool submit --keychain-profile "App Store Connect API - zeek-agent" --wait --timeout 20m "${bundle}"
 
-echo "== Stapling ..."
+echo "-- Stapling ..."
 
 xcrun stapler staple ${bundle}
 
-echo "== Verifying ..."
+echo "-- Verifying ..."
 
 spctl -a -vv -t install ${bundle}
 xcrun stapler validate ${bundle}
diff --git a/packaging/darwin/post-build.cmake b/packaging/darwin/post-build.cmake
index dc4da01a..b126b403 100644
--- a/packaging/darwin/post-build.cmake
+++ b/packaging/darwin/post-build.cmake
@@ -13,11 +13,10 @@ foreach (pkg "${CPACK_PACKAGE_FILES}")
     endforeach()
 endforeach()
 
-if ( NOT "$ENV{MACOS_NOTARIZATION_USER}" STREQUAL "")
-    message(STATUS "Notarizing application")
+if ( NOT "$ENV{MACOS_NOTARIZE}" STREQUAL "")
     foreach (pkg "${CPACK_PACKAGE_FILES}")
         execute_process(COMMAND ${CMAKE_CURRENT_LIST_DIR}/notarize application "${pkg}" COMMAND_ERROR_IS_FATAL ANY)
     endforeach()
 else ()
-    message(STATUS "Not notarizing application, MACOS_NOTARIZATION_USER not set")
+    message(STATUS "Not notarizing application, MACOS_NOTARIZE not set")
 endif()
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index d0db024a..ef26f3ed 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -28,11 +28,12 @@ add_subdirectory(platform)
 add_subdirectory(tables)
 add_subdirectory(util)
 
-# Ensure that we include the our Broker version even if existing includes may lead us elswhere.
-target_include_directories(zeek-agent PRIVATE $<TARGET_PROPERTY:broker_static,INCLUDE_DIRECTORIES>)
+if ( USE_BROKER )
+    # Ensure that we include our Broker version even if existing includes may lead us elswhere.
+    target_include_directories(zeek-agent PRIVATE $<TARGET_PROPERTY:broker_static,INCLUDE_DIRECTORIES>)
+endif ()
 
 target_link_libraries(zeek-agent PRIVATE ${CMAKE_DL_LIBS})
-target_link_libraries(zeek-agent PRIVATE broker_static)
 target_link_libraries(zeek-agent PRIVATE doctest::doctest)
 target_link_libraries(zeek-agent PRIVATE fmt::fmt)
 target_link_libraries(zeek-agent PRIVATE ghcFilesystem::ghc_filesystem)
@@ -48,6 +49,10 @@ target_link_libraries(zeek-agent PRIVATE tomlplusplus::tomlplusplus)
 target_link_libraries(zeek-agent PRIVATE ztd::out_ptr)
 target_link_libraries(zeek-agent PRIVATE Glob)
 
+if ( USE_BROKER )
+    target_link_libraries(zeek-agent PRIVATE broker_static)
+endif ()
+
 if ( NOT HAVE_GETOPT_LONG )
     target_link_libraries(zeek-agent PRIVATE 3rdparty:3rdparty)
 endif()
diff --git a/src/config.h.in b/src/config.h.in
index 2d5915f4..6a0d43b9 100644
--- a/src/config.h.in
+++ b/src/config.h.in
@@ -10,6 +10,7 @@
 #cmakedefine HAVE_CLANG
 #cmakedefine HAVE_MSVC
 #cmakedefine HAVE_GETOPT_LONG
+#cmakedefine HAVE_BROKER
 
 namespace zeek::agent {
 inline const auto Version = "${ZEEK_AGENT_VERSION}";
diff --git a/src/core/configuration.cc b/src/core/configuration.cc
index 5780e19a..7e8d9eb7 100644
--- a/src/core/configuration.cc
+++ b/src/core/configuration.cc
@@ -282,37 +282,13 @@ Result<Nothing> Configuration::Implementation::read(const filesystem::path& path
     return read(in, path);
 }
 
-// Helper to parse platform-provided string values into the target type.
-template<typename T>
-T parsePlatformValue(std::string_view path, const std::string& v) {
-    if constexpr ( std::is_same<T, std::string>::value )
-        return v;
-
-    if constexpr ( std::is_same<T, double>::value )
-        return std::stod(v);
-
-    if constexpr ( std::is_same<T, bool>::value ) {
-        auto x = trim(tolower(v));
-        return x == "1" || x == "true" || x == "yes" || x == "on";
-    }
-
-    throw result::Error(frmt("cannot parse plaform value for configuration option '{}' ({})", path, to_string(v)));
-}
-
 // Get a value, typed correctly, if available.
 template<typename T>
 bool tomlValue(const std::optional<toml::table>& t, const std::string& path, T* dst) {
     using vtype = typename std::remove_reference_t<T>;
 
-    if ( ! t ) {
-        // See if we have platform-specific mechanism providing this value.
-        auto v = platform::retrieveConfigurationOption(path);
-        if ( ! v )
-            return false;
-
-        *dst = parsePlatformValue<T>(path, *v);
-        return true;
-    }
+    if ( ! t )
+        return false;
 
     auto n = t->at_path(path);
     if ( ! n )
@@ -331,16 +307,8 @@ template<typename T>
 bool tomlArray(const std::optional<toml::table>& t, const std::string& path, std::vector<T>* dst) {
     using vtype = typename std::remove_reference_t<T>;
 
-    if ( ! t ) {
-        auto v = platform::retrieveConfigurationOption(path);
-        if ( ! v )
-            return false;
-
-        for ( const auto& x : split(*v) )
-            dst->push_back(parsePlatformValue<T>(path, trim(x)));
-
-        return true;
-    }
+    if ( ! t )
+        return false;
 
     auto n = t->at_path(path);
     if ( ! n )
@@ -452,6 +420,11 @@ Configuration::~Configuration() {
 
 const Options& Configuration::options() const { return *pimpl()->_options; }
 
+Result<Nothing> Configuration::setOptions(Options options) {
+    pimpl()->apply(std::move(options));
+    return Nothing();
+}
+
 Result<Nothing> Configuration::initFromArgv(std::vector<std::string> argv) {
     ZEEK_AGENT_DEBUG("configuration", "setting command line arguments: {}", join(argv, " "));
     return pimpl()->initFromArgv(std::move(argv));
diff --git a/src/core/configuration.h b/src/core/configuration.h
index d36e830f..1ecfa6f1 100644
--- a/src/core/configuration.h
+++ b/src/core/configuration.h
@@ -233,6 +233,9 @@ class Configuration : public Pimpl<Configuration> {
     /** Returns the options currently in effect. */
     const Options& options() const;
 
+    /** Applies new set of options provided by caller. */
+    Result<Nothing> setOptions(Options options);
+
     /**
      * Parses a set of command line options. This first resets the current
      * options back to their defaults, and then updates them according to any
@@ -253,7 +256,7 @@ class Configuration : public Pimpl<Configuration> {
      * Parses an agent configuration file. This first resets the current
      * options back to their defaults, and updates the current set of options
      * accordingly. If any command line options have been previously provided,
-     * it reapplys them on top at the end as well.
+     * it reapplies them on top at the end as well.
      *
      * @param path file to read
      * @return result will flag any errors that occurred
@@ -264,7 +267,7 @@ class Configuration : public Pimpl<Configuration> {
      * Parses an agent configuration file from already open input stream. This
      * first resets the current options back to their defaults, and updates the
      * current set of options accordingly. If any command line options have
-     * been previously provided, it reapplys them on top at the end as well.
+     * been previously provided, it reapplies them on top at the end as well.
      *
      * @param in already open stream to read the content from
      * @param path path associated with input (for error messages)
diff --git a/src/core/database.cc b/src/core/database.cc
index d5736309..bbe52675 100644
--- a/src/core/database.cc
+++ b/src/core/database.cc
@@ -53,17 +53,22 @@ struct Pimpl<Database>::Implementation {
     // Adds table to database.
     void addTable(Table* t);
 
+    // Adds table to list of pending ones.
+    void addPendingTable(Table* t);
+
     // Callback for the timers we install for our queries.
     Interval timerCallback(timer::ID id);
 
     // Helper to lookup scheduled query.
     std::optional<std::list<ScheduledQuery>::iterator> lookupQuery(query::ID);
 
+    Database* _db = nullptr;                       // database this implementation belongs to
     const Configuration* _configuration = nullptr; // configuration object, as passed into constructor
     Scheduler* _scheduler = nullptr;               // scheduler as passed into constructor
     std::unique_ptr<SQLite> _sqlite;               // SQLite backend for performing queries
 
     std::map<std::string, Table*> _tables; // registered tables indexed by name
+    std::list<Table*> _pending_tables;     // registered tables that we were initially temporarily unavailable
     std::list<ScheduledQuery> _queries;    // outstanding queries; list so that iterators remain valid on changes
     std::map<query::ID, std::list<ScheduledQuery>::iterator> _queries_by_id; // outstanding queries indexed by their ID
     std::set<std::pair<query::ID, bool>> _cancelled_queries; // track cancelled, but not removed, queries
@@ -167,6 +172,13 @@ void Database::Implementation::expire() {
             t->expire(expire_until);
         }
     }
+
+    // Go through pending tables and see if any has become available.
+    if ( ! _pending_tables.empty() ) {
+        auto pending = std::move(_pending_tables);
+        for ( const auto& t : pending )
+            addTable(t);
+    }
 }
 
 void Database::Implementation::poll() {
@@ -177,16 +189,43 @@ void Database::Implementation::poll() {
 }
 
 void Database::Implementation::addTable(Table* t) {
-    auto schema = t->schema();
+    EventTable::Init init_result;
+
+    if ( t->usesMockData() )
+        init_result = EventTable::Init::Available;
+    else
+        init_result = t->init();
 
-    if ( _tables.find(schema.name) != _tables.end() )
-        throw InternalError(frmt("table {} registered more than once", schema.name));
+    switch ( init_result ) {
+        case EventTable::Init::Available: {
+            ZEEK_AGENT_DEBUG("database", "adding table {} to database", t->name());
+            t->setDatabase(_db);
 
-    auto rc = _sqlite->addTable(t);
-    if ( ! rc )
-        throw FatalError(frmt("error registering table {} with SQLite backend: {}", schema.name, rc.error()));
+            auto schema = t->schema();
+
+            if ( _tables.find(schema.name) != _tables.end() )
+                throw InternalError(frmt("table {} registered more than once", schema.name));
+
+            auto rc = _sqlite->addTable(t);
+            if ( ! rc )
+                throw FatalError(frmt("error registering table {} with SQLite backend: {}", schema.name, rc.error()));
+
+            _tables[schema.name] = t;
+            return;
+        }
 
-    _tables[schema.name] = t;
+        case EventTable::Init::PermanentlyUnavailable:
+            ZEEK_AGENT_DEBUG("database", "not adding table {} to database because it's permanently disabled",
+                             t->name());
+            return;
+
+        case EventTable::Init::TemporarilyUnavailable:
+            ZEEK_AGENT_DEBUG("database",
+                             "not adding table {} to database because it's currently not available; will retry",
+                             t->name());
+            _pending_tables.push_back(t);
+            return;
+    }
 }
 
 static auto diffRows(std::vector<std::vector<Value>> old, std::vector<std::vector<Value>> new_) {
@@ -333,6 +372,7 @@ Table* Database::Implementation::table(const std::string& name) {
 
 Database::Database(Configuration* configuration, Scheduler* scheduler) {
     ZEEK_AGENT_DEBUG("database", "creating instance");
+    pimpl()->_db = this;
     pimpl()->_configuration = configuration;
     pimpl()->_scheduler = scheduler;
     pimpl()->_sqlite = std::make_unique<SQLite>();
@@ -400,20 +440,9 @@ void Database::expire() {
 }
 
 void Database::addTable(Table* t) {
-    t->setDatabase(this);
-
     if ( configuration().options().use_mock_data )
         t->enableMockData();
 
-    if ( ! t->usesMockData() ) {
-        if ( ! t->init() ) {
-            ZEEK_AGENT_DEBUG("database", "not adding table {} to database because it's disabled", t->name());
-            t->setDatabase(nullptr);
-            return;
-        }
-    }
-
-    ZEEK_AGENT_DEBUG("database", "adding table {} to database", t->name());
     pimpl()->addTable(t);
 }
 
@@ -492,9 +521,9 @@ TEST_SUITE("Database") {
 
         ~TestTable() override {}
 
-        bool init() override {
+        Init init() override {
             initialized = true;
-            return true;
+            return Init::Available;
         }
 
         std::vector<Time> expected_times;
@@ -539,7 +568,7 @@ TEST_SUITE("Database") {
             CHECK_EQ((*db.tables().begin())->schema().description, "test-description");
         }
 
-        SUBCASE("disabled table") {
+        SUBCASE("permanently disabled table") {
             class Disabled : public TestTable {
             public:
                 Schema schema() const override {
@@ -548,7 +577,29 @@ TEST_SUITE("Database") {
                     return schema;
                 }
 
-                bool init() override { return false; }
+                Init init() override { return Init::PermanentlyUnavailable; }
+            };
+
+            Disabled t;
+            Configuration cfg;
+            Scheduler tmgr;
+            Database db(&cfg, &tmgr);
+            db.addTable(&t);
+            REQUIRE(! db.table("disabled"));
+        }
+
+        SUBCASE("temporarily disabled table") {
+            class Disabled : public TestTable {
+            public:
+                Schema schema() const override {
+                    auto schema = TestTable::schema();
+                    schema.name = "disabled";
+                    return schema;
+                }
+
+                Init init() override { return enabled ? Init::Available : Init::TemporarilyUnavailable; }
+
+                bool enabled = false;
             };
 
             Disabled t;
@@ -557,6 +608,9 @@ TEST_SUITE("Database") {
             Database db(&cfg, &tmgr);
             db.addTable(&t);
             REQUIRE(! db.table("disabled"));
+            t.enabled = true;
+            db.expire(); // will trigger retry
+            REQUIRE(db.table("disabled"));
         }
     }
 
@@ -1089,7 +1143,7 @@ TEST_SUITE("Database") {
     TEST_CASE("virtual methods with mock data") {
         // Check that some of our virtual methods aren't called when using mock data.
         class MockedTestTable : public TestTable {
-            bool init() override {
+            Init init() override {
                 CHECK(false);
                 cannot_be_reached();
             }
diff --git a/src/core/table.cc b/src/core/table.cc
index a0130e42..00f0a0f9 100644
--- a/src/core/table.cc
+++ b/src/core/table.cc
@@ -302,6 +302,15 @@ const Options& Table::options() const {
     return _db->configuration().options();
 }
 
+Time Table::systemTime() const {
+    auto t = std::chrono::system_clock::now();
+    if ( _last_time == t )
+        t += to_interval_from_ns(1);
+
+    _last_time = Time(t);
+    return t;
+}
+
 Time Table::currentTime() const {
     if ( ! _db )
         throw InternalError("no database/scheduler available in table");
@@ -385,10 +394,13 @@ std::vector<std::vector<Value>> SnapshotTable::rows(Time t, const std::vector<ta
 }
 
 void EventTable::newEvent(std::vector<Value> row) {
+    const std::scoped_lock lock(_events_mutex);
     _events.emplace_back(Event{.time = currentTime(), .row = std::move(row)});
 }
 
 void EventTable::newEvent(Time t, std::vector<Value> row) {
+    const std::scoped_lock lock(_events_mutex);
+
     if ( ! _events.empty() && t < _events.back().time )
         throw InternalError("outdated timestamp in EventTable::newEvent()");
 
@@ -396,6 +408,8 @@ void EventTable::newEvent(Time t, std::vector<Value> row) {
 }
 
 std::vector<std::vector<Value>> EventTable::rows(Time t, const std::vector<table::Argument>& args) {
+    const std::scoped_lock lock(_events_mutex);
+
     // We ignore the WHERE constraints in this implementation.
     std::vector<std::vector<Value>> result;
 
@@ -416,6 +430,8 @@ std::vector<std::vector<Value>> EventTable::rows(Time t, const std::vector<table
 }
 
 void EventTable::expire(Time t) {
+    const std::scoped_lock lock(_events_mutex);
+
     auto end = std::lower_bound(_events.begin(), _events.end(), Event{.time = t, .row = {}});
     _events.erase(_events.begin(), end);
 }
diff --git a/src/core/table.h b/src/core/table.h
index 967889d2..77dd3aae 100644
--- a/src/core/table.h
+++ b/src/core/table.h
@@ -311,7 +311,8 @@ class Table {
      * @result false to completely disable the table at this point, so that it
      * will not be available for queries.
      */
-    virtual bool init() { return true; }
+    enum class Init { Available, PermanentlyUnavailable, TemporarilyUnavailable };
+    virtual Init init() { return Init::Available; }
 
     /**
      * Hook that will be called just before the first query against this table
@@ -411,6 +412,18 @@ class Table {
         throw InternalError(frmt("table argument '{}' unexpectedly missing", name));
     }
 
+    /**
+     * Returns the current system time, guaranteeing that it's monotonically
+     * increasing between calls.
+     */
+    Time systemTime() const;
+
+    /**
+     * Record the database that this table has been registered with. For
+     * internal use by database only.
+     */
+    void setDatabase(Database* db) { _db = db; }
+
 protected:
     /**
      * Returns the configuration options currently in effect. This won't be
@@ -433,14 +446,10 @@ class Table {
     std::vector<Value> generateMockRow(int i);
 
 private:
-    friend Database;
-
-    // Record the database that this table has been registered with.
-    void setDatabase(Database* db) { _db = db; }
-
     Database* _db = nullptr;      // database set through `setDatabase()`
     int _current_connections = 0; // counter of active queries against this table
-    bool _use_mock_data = false;  // if true, have table return mock data for testing
+    mutable Time _last_time = {};
+    bool _use_mock_data = false; // if true, have table return mock data for testing
 };
 
 /**
@@ -480,7 +489,7 @@ class SnapshotTable : public Table {
  * activity.
  *
  * To provide new events, derived classes should call `newEvent()` as data
- * becomes available.
+ * becomes available. It's safe to do so from any thread.
  *
  * The class inherits most of the hooks that `Table` provides, which derived
  * classes are free to implement as desired. The two exceptions are `rows()`
@@ -491,7 +500,7 @@ class EventTable : public Table {
     /**
      * Records an event that has occured.
      *
-     * Derived classes need to call this as they observe their activity.
+     * Derived classes need to call this as they observe their activity. It's safe to do so from any thread.
      *
      * @param row the column values associated with the event, which must match the table's schema
      */
@@ -522,6 +531,7 @@ class EventTable : public Table {
         bool operator<(const Event& other) const { return time < other.time; }
     };
 
+    std::mutex _events_mutex;   // mutex protecting access to the event buffer
     std::vector<Event> _events; // set of currently buffered events, sorted by timestamp
     int _mock_seed = 0;         // when generating mock data, seed value for next round
 };
diff --git a/src/io/zeek.cc b/src/io/zeek.cc
index a8970cc7..f7ae9b49 100644
--- a/src/io/zeek.cc
+++ b/src/io/zeek.cc
@@ -38,16 +38,6 @@
 #include <ixwebsocket/IXWebSocketSendData.h>
 #include <nlohmann/json.hpp>
 
-#include <broker/address.hh>
-#include <broker/configuration.hh>
-#include <broker/data.hh>
-#include <broker/endpoint.hh>
-#include <broker/enum_value.hh>
-#include <broker/fwd.hh>
-#include <broker/none.hh>
-#include <broker/topic.hh>
-#include <broker/zeek.hh>
-
 // Minimum version of the Zeek-side package that we require. If we see an agent
 // with an older version, we'll stop communicating with it.
 static const int64_t MininumZeekPackageVersion = 200020008;
@@ -233,155 +223,212 @@ class ZeekConnection {
     std::map<std::string, ZeekInstance> _zeek_instances; // currently active Zeek instances
 };
 
-///// Native Broker transport.
+///// WebSocket transport.
 
 // Transport implementation using the Broker library for communication.
-class NativeBrokerTransport : public TransportProtocol {
+class WebSocketTransport : public TransportProtocol {
 public:
-    NativeBrokerTransport(const zeek::agent::Configuration& config) : _config(config) {}
-    ~NativeBrokerTransport() override {}
+    WebSocketTransport(const zeek::agent::Configuration& config) : _config(config) {}
+    ~WebSocketTransport() override {}
 
     void connect(const std::string& host, unsigned int port, const std::vector<std::string>& topics) override;
     void disconnect() override;
-    bool isShutdown() override { return ! _endpoint || _endpoint->is_shutdown(); }
-    unsigned int defaultPort() override { return 9998; /* default port used by the zeek-agent package  */ }
+    bool isShutdown() override { return ! (_connected && _socket.getReadyState() == ix::ReadyState::Open); }
+    unsigned int defaultPort() override { return 9997; /* Zeek's default WebSocket port */ }
     void transmitEvent(const std::string& topic, const std::string& name, Record args) override;
-    const char* name() const override { return "Broker"; }
+    void poll() override;
+    const char* name() const override { return "WebSocket"; }
 
 private:
-    // Helper to prepare Broker config object
-    broker::configuration brokerConfig();
+    void tryReconnect(); // backend for both connect() and reconnection logic
 
-    void processEvent(const broker::data_message& msg);
-    void processError(const broker::error& err);
-    void processConnectivityChange(const broker::status& status);
+    const zeek::agent::Configuration& _config; // as passed into constructor
+    std::string _host;                         // address trying to connect to
+    unsigned int _port;                        // port trying to connect to
 
-    const zeek::agent::Configuration& _config;        // as passed into constructor
-    std::unique_ptr<broker::endpoint> _endpoint;      // Broker state
-    std::optional<broker::network_info> _destination; // parsed destination
+    ix::WebSocket _socket;
+    bool _connected = false;
+    std::optional<Time> _last_connect_attempt;
 };
 
-static broker::data to_broker(const Value& v, const value::Type& t) {
-    broker::data value;
+static nlohmann::json to_json(const Value& v, const value::Type& t) {
+    nlohmann::json value;
+    std::string type;
+
     if ( std::get_if<std::monostate>(&v) == nullptr ) {
         switch ( t ) {
-            case value::Type::Count: value = static_cast<uint64_t>(std::get<int64_t>(v)); break;
-            case value::Type::Integer: value = std::get<int64_t>(v); break;
+            case value::Type::Count:
+                type = "count";
+                value = std::get<int64_t>(v);
+                break;
+
+            case value::Type::Integer:
+                type = "integer";
+                value = std::get<int64_t>(v);
+                break;
+
             case value::Type::Blob:
-            case value::Type::Text: value = std::get<std::string>(v); break;
-            case value::Type::Bool: value = (std::get<bool>(v) != 0); break;
-            case value::Type::Double: value = std::get<double>(v); break;
-            case value::Type::Enum: value = broker::enum_value(std::get<std::string>(v)); break;
+            case value::Type::Text:
+                type = "string";
+                value = std::get<std::string>(v);
+                break;
+
+            case value::Type::Bool:
+                type = "boolean";
+                value = (std::get<bool>(v) != 0);
+                break;
+
+            case value::Type::Double:
+                type = "real";
+                value = std::get<double>(v);
+                break;
+
+            case value::Type::Enum:
+                type = "enum-value";
+                value = std::get<std::string>(v);
+                break;
 
             case value::Type::Interval:
-                value = std::chrono::duration_cast<broker::timespan>(std::get<Interval>(v));
+                type = "timespan";
+                value = to_string(std::get<Interval>(v));
                 break;
 
-            case value::Type::Null: value = broker::data(); break;
-            case value::Type::Time: value = broker::timestamp(std::get<Time>(v).time_since_epoch()); break;
+            case value::Type::Null:
+                type = "none";
+                value = nlohmann::json::object();
+                break;
+
+            case value::Type::Time: {
+                type = "timestamp";
+                auto s = to_string_iso(std::get<Time>(v));
+                if ( auto x = s.find('+'); x != std::string::npos ) // TODO: Broker doesn't like timezones added
+                    s = s.substr(0, x);
+                value = s + ".000"; // TODO: Broker requires postfix
+                break;
+            }
 
             case value::Type::Address: {
-                broker::address addr;
-                if ( addr.convert_from(std::get<std::string>(v)) )
-                    value = addr;
+                type = "address";
+                value = std::get<std::string>(v);
                 break;
             }
 
             case value::Type::Port: {
                 const auto& p = std::get<Port>(v);
-                broker::port::protocol proto;
+                std::string proto;
                 switch ( p.protocol ) {
-                    case port::Protocol::ICMP: proto = broker::port::protocol::icmp; break;
-                    case port::Protocol::TCP: proto = broker::port::protocol::tcp; break;
-                    case port::Protocol::UDP: proto = broker::port::protocol::udp; break;
-                    case port::Protocol::Unknown: proto = broker::port::protocol::unknown; break;
+                    case port::Protocol::ICMP: proto = "icmp"; break;
+                    case port::Protocol::TCP: proto = "tcp"; break;
+                    case port::Protocol::UDP: proto = "udp"; break;
+                    case port::Protocol::Unknown: proto = "unknown"; break;
                 }
 
-                value = broker::port(p.port, proto);
+                type = "port";
+                value = frmt("{}/{}", p.port, proto);
                 break;
             }
 
             case value::Type::Record: {
-                broker::vector br;
+                std::vector<nlohmann::json> j;
                 for ( const auto& [x, t] : std::get<Record>(v) )
-                    br.emplace_back(to_broker(x, t));
+                    j.emplace_back(to_json(x, t));
 
-                value = std::move(br);
+                type = "vector";
+                value = std::move(j);
                 break;
             }
 
             case value::Type::Set: {
-                const auto& set = std::get<Set>(v);
-                broker::set bs;
-                for ( const auto& x : set )
-                    bs.insert(to_broker(x, set.type));
+                std::vector<nlohmann::json> j;
+                for ( const auto& x : std::get<Set>(v) )
+                    j.emplace_back(to_json(x, std::get<Set>(v).type));
 
-                value = std::move(bs);
+                type = "set";
+                value = std::move(j);
                 break;
             }
 
             case value::Type::Vector: {
-                const auto& vec = std::get<Vector>(v);
-                broker::vector bv;
-                for ( const auto& x : vec )
-                    bv.emplace_back(to_broker(x, vec.type));
+                std::vector<nlohmann::json> j;
+                for ( const auto& x : std::get<Vector>(v) )
+                    j.emplace_back(to_json(x, std::get<Vector>(v).type));
 
-                value = std::move(bv);
+                type = "vector";
+                value = std::move(j);
                 break;
             }
         }
     }
+    else {
+        type = "none";
+        value = nlohmann::json::object();
+    }
 
-    return value;
+    nlohmann::json json;
+    json["@data-type"] = type;
+    json["data"] = value;
+    return json;
 }
 
 // Best effort type guessing.
-static std::pair<Value, value::Type> from_broker(const broker::data& v) {
-    if ( auto x = broker::get_if<broker::none>(&v) )
-        return {{}, value::Type::Null};
+static std::pair<Value, value::Type> from_json(const nlohmann::json& json) {
+    const auto& type = json["@data-type"];
+    const auto& value = json["data"];
 
-    if ( auto x = broker::get_if<broker::boolean>(&v) )
-        return {*x, value::Type::Bool};
+    if ( type == "integer" )
+        return {value.get<int64_t>(), value::Type::Integer};
 
-    if ( auto x = broker::get_if<broker::count>(&v) )
-        return {static_cast<int64_t>(*x), value::Type::Count};
+    if ( type == "count" )
+        return {value.get<int64_t>(), value::Type::Count};
 
-    if ( auto x = broker::get_if<broker::integer>(&v) )
-        return {static_cast<int64_t>(*x), value::Type::Integer};
+    if ( type == "bool" )
+        return {value.get<bool>(), value::Type::Bool};
 
-    if ( auto x = broker::get_if<broker::real>(&v) )
-        return {*x, value::Type::Double};
+    if ( type == "real" )
+        return {value.get<double>(), value::Type::Double};
 
-    if ( auto x = broker::get_if<std::string>(&v) )
-        return {*x, value::Type::Text};
+    if ( type == "string" )
+        return {value.get<std::string>(), value::Type::Text};
 
-    if ( auto x = broker::get_if<broker::enum_value>(&v) )
-        return {x->name, value::Type::Enum};
+    if ( type == "enum-value" )
+        return {value.get<std::string>(), value::Type::Enum};
 
-    if ( auto x = broker::get_if<broker::set>(&v) ) {
+    if ( type == "none" )
+        return {{}, value::Type::Null};
+
+    if ( type == "set" ) {
         auto type = value::Type::Null;
-        if ( ! x->empty() )
-            type = from_broker(*x->begin()).second;
 
-        Set y(type);
-        for ( const auto& i : *x )
-            y.insert(from_broker(i).first);
+        std::set<Value> x;
+        for ( const auto& i : value ) {
+            auto m = from_json(i);
+            x.insert(m.first);
+            type = m.second;
+        }
 
-        return {y, value::Type::Set};
+        return {Set(type, std::move(x)), value::Type::Set};
     }
 
-    if ( auto x = broker::get_if<broker::vector>(&v) ) {
+    if ( type == "vector" ) {
         // We can't distinguish vectors from records, but we only need the
         // latter right now ...
         Record y;
-        for ( const auto& i : *x )
-            y.emplace_back(from_broker(i));
+        for ( const auto& i : value )
+            y.emplace_back(from_json(i));
 
         return {y, value::Type::Record};
     }
 
-    if ( auto x = broker::get_if<broker::timespan>(&v) )
-        return {std::chrono::duration_cast<Interval>(*x), value::Type::Interval};
+    if ( type == "timespan" )
+        return {to_interval_from_secs(std::stoi(value.get<std::string>())), value::Type::Interval};
+
+    if ( type == "timestamp" ) {
+        std::tm tm = {};
+        std::istringstream ss(value.get<std::string>());
+        ss >> std::get_time(&tm, "%FT%T");
+        auto tp = std::chrono::system_clock::from_time_t(std::mktime(&tm));
+        return {tp, value::Type::Time};
+    }
 
     /* Not supported, don't need these.
      *
@@ -392,235 +439,273 @@ static std::pair<Value, value::Type> from_broker(const broker::data& v) {
      * else if ( auto x = broker::get_if<broker::timestamp>(&v) )
      */
 
-    throw InternalError(frmt("unsupported Broker data type received ({})", broker::to_string(v)));
+    throw InternalError(frmt("unsupported data type received over WebSocket ({})", type));
 }
 
-broker::configuration NativeBrokerTransport::brokerConfig() {
-    const auto& options = _config.options();
-    // Configure Broker/CAF for lower resource consumption.
-    broker::broker_options broker_options;
-    broker_options.forward = false;
-    broker_options.ignore_broker_conf = true;
-    broker_options.disable_ssl = options.zeek_ssl_disable;
-
-    broker::configuration broker_config(broker_options);
-    broker_config.openssl_cafile(options.zeek_ssl_cafile);
-    broker_config.openssl_capath(options.zeek_ssl_capath);
-    broker_config.openssl_certificate(options.zeek_ssl_certificate);
-    broker_config.openssl_key(options.zeek_ssl_keyfile);
-    broker_config.openssl_passphrase(options.zeek_ssl_passphrase);
+void WebSocketTransport::connect(const std::string& host, unsigned int port, const std::vector<std::string>& topics) {
+    const auto& options = connection()->options();
 
-#if 1
-    broker_config.set("caf.scheduler.policy", "sharing");
-    broker_config.set("caf.scheduler.max-threads", 1);
-    broker_config.set("caf.middleman.workers", 0);
-#else
-    // Use Zeek's stealing configuration.
-    broker_config.set("caf.work-stealing.moderate-sleep-duration", broker::timespan{16'000});
-    broker_config.set("caf.work-stealing.relaxed-sleep-duration", broker::timespan{64'000});
-    broker_config.set("caf.work-stealing.aggressive-poll-attempts", 5);
-    broker_config.set("caf.work-stealing.moderate-poll-attempts", 5);
-    broker_config.set("caf.work-stealing.aggressive-steal-interval", 4);
-    broker_config.set("caf.work-stealing.moderate-steal-interval", 2);
-    broker_config.set("caf.work-stealing.relaxed-steal-interval", 1);
-#endif
+    auto url = frmt("{}://{}:{}/v1/messages/json", (options.zeek_ssl_disable ? "ws" : "wss"), host, port);
 
-    return broker_config;
-}
+    _socket.setUrl(url);
+    _socket.setPingInterval(30);
+    _socket.enablePong();
+    _socket.enablePerMessageDeflate();
+    _host = host;
+    _port = port;
 
-void NativeBrokerTransport::connect(const std::string& host, unsigned int port,
-                                    const std::vector<std::string>& topics) {
-    auto broker_topics = transform(topics, [](const auto& t) { return broker::topic(t); });
-    broker_topics.emplace_back(std::string(broker::topic::errors_str));
-    broker_topics.emplace_back(std::string(broker::topic::statuses_str));
+    if ( ! connection()->options().zeek_ssl_disable ) {
+        ix::SocketTLSOptions tls_options;
+        tls_options.tls = true;
+        tls_options.disable_hostname_validation = true; // with Zeek, we generally don't validate hostnames
 
-    _endpoint = std::make_unique<broker::endpoint>(brokerConfig());
-    _endpoint->subscribe_nosync(
-        std::move(broker_topics), []() { /* nop */ },
-        [this](const broker::data_message& msg) {
-            connection()->scheduler()->schedule([this, msg]() { // process message on the main thread
-                if ( broker::get_topic(msg) == broker::topic::statuses_str ) {
-                    auto x = broker::to<broker::status>(broker::get_data(msg));
-                    assert(x); // NOLINT(bugprone-lambda-function-name)
-                    processConnectivityChange(*x);
+        const auto auth_disabled = (options.zeek_ssl_certificate.empty() && options.zeek_ssl_keyfile.empty() &&
+                                    options.zeek_ssl_cafile.empty() && options.zeek_ssl_capath.empty());
+
+        if ( ! auth_disabled ) {
+            if ( ! options.zeek_ssl_certificate.empty() )
+                tls_options.certFile = options.zeek_ssl_certificate;
+
+            if ( ! options.zeek_ssl_keyfile.empty() )
+                tls_options.keyFile = options.zeek_ssl_keyfile;
+
+            if ( ! options.zeek_ssl_cafile.empty() )
+                tls_options.caFile = options.zeek_ssl_cafile;
+            else
+                tls_options.caFile = "SYSTEM";
+
+            if ( ! options.zeek_ssl_capath.empty() )
+                logger()->warn("option zeek.ssl_cpath not supported for WebSocket connections, ignoring");
+        }
+
+        else {
+            // Enable encryption, but don't require authentication (and not even certificates).
+            tls_options.caFile = "NONE";
+            // Use a cipher that don't need a certificate; this list is
+            // borrowed from Broker, including ciphers that works with
+            // different OpenSSL versions
+            tls_options.ciphers = "AECDH-AES256-SHA@SECLEVEL=0:AECDH-AES256-SHA:P-384";
+        }
+
+        _socket.setTLSOptions(tls_options);
+    }
+
+    // We implement our own auto-connect, too difficult to control otherwise.
+    _socket.disableAutomaticReconnection();
+
+    _socket.setOnMessageCallback([this, topics](const ix::WebSocketMessagePtr& msg) {
+        auto msg_type = msg->type;
+        auto msg_str = msg->str;
+        auto msg_error_reason = msg->errorInfo.reason;
+
+        connection()->scheduler()->schedule([this, topics, msg_type, msg_str,
+                                             msg_error_reason]() { // process message on the main thread
+            switch ( msg_type ) {
+                case ix::WebSocketMessageType::Open: {
+                    connection()->connectionEstablished(this, _host, _port);
+                    connection()->processConnectivityChange(ConnectivityChange::Added,
+                                                            frmt("connected to WebSocket endpoint at {}",
+                                                                 connection()->endpoint()));
+
+                    _socket.send(nlohmann::json(topics).dump());
+                    _connected = true;
+                    break;
                 }
-                else if ( broker::get_topic(msg) == broker::topic::errors_str ) {
-                    auto x = broker::to<broker::error>(broker::get_data(msg));
-                    assert(x); // NOLINT(bugprone-lambda-function-name)
-                    processError(*x);
+
+                case ix::WebSocketMessageType::Close: {
+                    if ( _connected )
+                        connection()->processConnectivityChange(ConnectivityChange::Removed,
+                                                                frmt("disconnected from WebSocket endpoint at {}",
+                                                                     connection()->endpoint()));
+                    _connected = false;
+                    break;
                 }
-                else
-                    processEvent(msg);
-            });
-        },
-        [](const broker::error&) { /* nop */ });
 
-    _destination = broker::network_info(host, port,
-                                        std::chrono::duration_cast<broker::timeout::seconds>(
-                                            _config.options().zeek_reconnect_interval));
+                case ix::WebSocketMessageType::Error: {
+                    if ( _connected )
+                        connection()->processConnectivityChange(ConnectivityChange::Lost,
+                                                                frmt("lost connection to WebSocket endpoint at {} ({})",
+                                                                     connection()->endpoint(), msg_error_reason));
+                    else {
+                        std::string reason = trim(msg_error_reason);
 
-    // Broker's peer_nosync() has not version taking netinfo directly
-    _endpoint->peer_nosync(_destination->address, _destination->port, _destination->retry);
-}
+                        if ( reason.find("Connect error:") != std::string::npos ||
+                             reason.find("Cancelled") != std::string::npos )
+                            // We get "error: Connect error: Bad file
+                            // descriptor" or "error: Cancelled" if noone is
+                            // listening at the destination port.
+                            reason = "";
 
-void NativeBrokerTransport::disconnect() {
-    if ( ! _endpoint )
-        return;
+                        logger()->debug("cannot connect to Zeek endpoint via WebSocket at {}{}",
+                                        connection()->endpoint(), (reason.empty() ? "" : frmt(" ({})", reason)));
+                        connection()->connectionAttemptFailed(this, reason);
+                    }
 
-    if ( ! _endpoint->unpeer(_destination->address, _destination->port) )
-        logger()->warn("failed disconnect from {}", connection()->endpoint());
+                    _connected = false;
+                    break;
+                }
 
-    _endpoint->shutdown();
-    _endpoint = nullptr;
-}
+                case ix::WebSocketMessageType::Message:
+                    try {
+                        auto json = nlohmann::json::parse(msg_str);
 
-void NativeBrokerTransport::transmitEvent(const std::string& topic, const std::string& event_name, Record args) {
-    if ( ! _endpoint )
-        return;
+                        if ( json["type"] == "ack" ) {
+                            auto endpoint_ = json["endpoint"].get<std::string>();
+                            auto broker_version = json["version"].get<std::string>();
+                            ZEEK_CONN_DEBUG("received acknowledgment (endpoint={} broker={})", endpoint_,
+                                            broker_version);
+                        }
 
-    broker::zeek::Event event(event_name, broker::get<broker::vector>(to_broker(std::move(args), value::Type::Record)));
-    _endpoint->publish(topic, std::move(event));
-}
+                        else if ( json["type"] == "data-message" ) {
+                            auto [data_, type] = from_json(json);
+                            const auto& data = std::get<Record>(data_);
 
-void NativeBrokerTransport::processEvent(const broker::data_message& msg) {
-    broker::zeek::Event event(std::get<1>(msg.data()));
+                            if ( std::get<int64_t>(data[0].first) != 1 || std::get<int64_t>(data[1].first) != 1 ) {
+                                ZEEK_CONN_DEBUG("unexpected content enums for data-message: {}", to_string(data));
+                                return;
+                            }
 
-    std::vector<Value> args;
-    for ( const auto& a : event.args() )
-        args.push_back(from_broker(a).first);
+                            auto event = std::get<Record>(data[2].first);
+                            const auto& event_name = std::get<std::string>(event[0].first);
+                            const std::vector<std::pair<Value, value::Type>>& event_args =
+                                std::get<Record>(event[1].first);
 
-    connection()->processEvent(event.name(), args);
+                            connection()->processEvent(event_name,
+                                                       transform(event_args, [](auto i) { return i.first; }));
+                        }
+
+                        else if ( json["type"] == "error" ) {
+                            auto code = json["code"].get<std::string>();
+                            auto context = json["context"].get<std::string>();
+                            connection()->processError(
+                                frmt("WebSocket error for {}: {} ({})", connection()->endpoint(), code, context));
+                        }
+
+                        else {
+                            ZEEK_CONN_DEBUG("unexpected message type {}", json["@type"]);
+                            return;
+                        }
+                    }
+
+                    catch ( const std::exception& e ) {
+                        ZEEK_CONN_DEBUG("cannot parse WebSocket message: {} ({})", e.what(), msg_str);
+                    }
+
+                    break;
+
+                case ix::WebSocketMessageType::Ping:
+                case ix::WebSocketMessageType::Pong:
+                case ix::WebSocketMessageType::Fragment: break;
+            }
+        });
+    });
+
+    ZEEK_CONN_DEBUG("WebSocket endpoint: {}", url);
+    tryReconnect();
 }
 
-void NativeBrokerTransport::processError(const broker::error& err) {
-    std::string msg = "<no error message>";
-    if ( err.message() )
-        msg = *err.message();
+void WebSocketTransport::tryReconnect() {
+    if ( _socket.getReadyState() != ix::ReadyState::Closed )
+        // Still/already doing something on the connection.
+        return;
 
-    switch ( static_cast<broker::ec>(err.code()) ) {
-        // Prettyify some common errors.
-        case broker::ec::peer_invalid:
-        case broker::ec::peer_unavailable:
-            logger()->debug("cannot connect to Zeek endpoint via Broker at {}", connection()->endpoint());
-            connection()->connectionAttemptFailed(this, msg);
-            break;
+    if ( _last_connect_attempt &&
+         (std::chrono::system_clock::now() - *_last_connect_attempt) < _config.options().zeek_reconnect_interval )
+        return;
 
-        default: connection()->processError(frmt("{} for {}", msg, connection()->endpoint()));
-    }
+    _socket.stop();
+    _socket.start();
+    _last_connect_attempt = std::chrono::system_clock::now();
 }
 
-void NativeBrokerTransport::processConnectivityChange(const broker::status& status) {
-    ConnectivityChange change = ConnectivityChange::Other;
-    auto msg = (status.message() ? *status.message() : std::string("<no status description from broker>"));
+void WebSocketTransport::disconnect() {
+    _socket.stop();
+    _last_connect_attempt.reset();
+    _connected = false;
+}
 
-    switch ( static_cast<broker::sc>(status.code()) ) {
-        // Prettyify some common messages.
-        case broker::sc::peer_added:
-            connection()->connectionEstablished(this, _destination.value().address, _destination.value().port);
-            change = ConnectivityChange::Added;
-            msg = frmt("connected to Zeek endpoint at {}", connection()->endpoint());
-            break;
-        case broker::sc::peer_lost:
-            change = ConnectivityChange::Lost;
-            msg = frmt("lost connection to Zeek endpoint at {}", connection()->endpoint());
-            break;
-        case broker::sc::peer_removed:
-            change = ConnectivityChange::Removed;
-            msg = frmt("disconnected from Zeek endpoint at {}", connection()->endpoint());
-            break;
-        default: break;
-    }
+void WebSocketTransport::poll() {
+    if ( _last_connect_attempt && ! _connected )
+        tryReconnect();
+}
 
-    connection()->processConnectivityChange(change, msg);
+void WebSocketTransport::transmitEvent(const std::string& topic, const std::string& event_name, Record args) {
+    auto event = Record({{Value(1L), value::Type::Count},
+                         {Value(1L), value::Type::Count},
+                         {Record{{
+                              {Value(event_name), value::Type::Text},
+                              {std::move(args), value::Type::Record},
+                          }},
+                          value::Type::Record}});
+
+    nlohmann::json msg = to_json(event, value::Type::Record);
+    msg["type"] = "data-message";
+    msg["topic"] = topic;
+    _socket.send(msg.dump());
 }
 
-///// WebSocket transport.
+#ifdef HAVE_BROKER
+///// Native Broker transport (legacy only)
+
+#include <broker/address.hh>
+#include <broker/configuration.hh>
+#include <broker/data.hh>
+#include <broker/endpoint.hh>
+#include <broker/enum_value.hh>
+#include <broker/fwd.hh>
+#include <broker/none.hh>
+#include <broker/topic.hh>
+#include <broker/zeek.hh>
 
 // Transport implementation using the Broker library for communication.
-class WebSocketTransport : public TransportProtocol {
+class NativeBrokerTransport : public TransportProtocol {
 public:
-    WebSocketTransport(const zeek::agent::Configuration& config) : _config(config) {}
-    ~WebSocketTransport() override {}
+    NativeBrokerTransport(const zeek::agent::Configuration& config) : _config(config) {}
+    ~NativeBrokerTransport() override {}
 
     void connect(const std::string& host, unsigned int port, const std::vector<std::string>& topics) override;
     void disconnect() override;
-    bool isShutdown() override { return ! (_connected && _socket.getReadyState() == ix::ReadyState::Open); }
-    unsigned int defaultPort() override { return 9997; /* Zeek's default WebSocket port */ }
+    bool isShutdown() override { return ! _endpoint || _endpoint->is_shutdown(); }
+    unsigned int defaultPort() override { return 9998; /* default port used by the zeek-agent package  */ }
     void transmitEvent(const std::string& topic, const std::string& name, Record args) override;
-    void poll() override;
-    const char* name() const override { return "WebSocket"; }
+    const char* name() const override { return "Broker"; }
 
 private:
-    void tryReconnect(); // backend for both connect() and reconnection logic
+    // Helper to prepare Broker config object
+    broker::configuration brokerConfig();
 
-    const zeek::agent::Configuration& _config; // as passed into constructor
-    std::string _host;                         // address trying to connect to
-    unsigned int _port;                        // port trying to connect to
+    void processEvent(const broker::data_message& msg);
+    void processError(const broker::error& err);
+    void processConnectivityChange(const broker::status& status);
 
-    ix::WebSocket _socket;
-    bool _connected = false;
-    std::optional<Time> _last_connect_attempt;
+    const zeek::agent::Configuration& _config;        // as passed into constructor
+    std::unique_ptr<broker::endpoint> _endpoint;      // Broker state
+    std::optional<broker::network_info> _destination; // parsed destination
 };
 
-static nlohmann::json to_json(const Value& v, const value::Type& t) {
-    nlohmann::json value;
-    std::string type;
-
+static broker::data to_broker(const Value& v, const value::Type& t) {
+    broker::data value;
     if ( std::get_if<std::monostate>(&v) == nullptr ) {
         switch ( t ) {
-            case value::Type::Count:
-                type = "count";
-                value = std::get<int64_t>(v);
-                break;
-
-            case value::Type::Integer:
-                type = "integer";
-                value = std::get<int64_t>(v);
-                break;
-
+            case value::Type::Count: value = static_cast<uint64_t>(std::get<int64_t>(v)); break;
+            case value::Type::Integer: value = std::get<int64_t>(v); break;
             case value::Type::Blob:
-            case value::Type::Text:
-                type = "string";
-                value = std::get<std::string>(v);
-                break;
-
-            case value::Type::Bool:
-                type = "boolean";
-                value = (std::get<bool>(v) != 0);
-                break;
-
-            case value::Type::Double:
-                type = "real";
-                value = std::get<double>(v);
-                break;
-
-            case value::Type::Enum:
-                type = "enum-value";
-                value = std::get<std::string>(v);
-                break;
+            case value::Type::Text: value = std::get<std::string>(v); break;
+            case value::Type::Bool: value = (std::get<bool>(v) != 0); break;
+            case value::Type::Double: value = std::get<double>(v); break;
+            case value::Type::Enum: value = broker::enum_value(std::get<std::string>(v)); break;
 
             case value::Type::Interval:
-                type = "timespan";
-                value = to_string(std::get<Interval>(v));
-                break;
-
-            case value::Type::Null:
-                type = "none";
-                value = nlohmann::json::object();
+                value = std::chrono::duration_cast<broker::timespan>(std::get<Interval>(v));
                 break;
 
-            case value::Type::Time: {
-                type = "timestamp";
-                auto s = to_string_iso(std::get<Time>(v));
-                if ( auto x = s.find('+'); x != std::string::npos ) // TODO: Broker doesn't like timezones added
-                    s = s.substr(0, x);
-                value = s + ".000"; // TODO: Broker requires postfix
-                break;
-            }
+            case value::Type::Null: value = broker::data(); break;
+            case value::Type::Time: value = broker::timestamp(std::get<Time>(v).time_since_epoch()); break;
 
             case value::Type::Address: {
-                type = "address";
-                value = std::get<std::string>(v);
+                broker::address addr;
+                if ( addr.convert_from(std::get<std::string>(v)) )
+                    value = addr;
                 break;
             }
 
@@ -634,327 +719,243 @@ static nlohmann::json to_json(const Value& v, const value::Type& t) {
                     case port::Protocol::Unknown: proto = broker::port::protocol::unknown; break;
                 }
 
-                type = "port";
-                value = frmt("{}/{}", p.port, proto);
+                value = broker::port(p.port, proto);
                 break;
             }
 
             case value::Type::Record: {
-                std::vector<nlohmann::json> j;
+                broker::vector br;
                 for ( const auto& [x, t] : std::get<Record>(v) )
-                    j.emplace_back(to_json(x, t));
+                    br.emplace_back(to_broker(x, t));
 
-                type = "vector";
-                value = std::move(j);
+                value = std::move(br);
                 break;
             }
 
             case value::Type::Set: {
-                std::vector<nlohmann::json> j;
-                for ( const auto& x : std::get<Set>(v) )
-                    j.emplace_back(to_json(x, std::get<Set>(v).type));
+                const auto& set = std::get<Set>(v);
+                broker::set bs;
+                for ( const auto& x : set )
+                    bs.insert(to_broker(x, set.type));
 
-                type = "set";
-                value = std::move(j);
+                value = std::move(bs);
                 break;
             }
 
             case value::Type::Vector: {
-                std::vector<nlohmann::json> j;
-                for ( const auto& x : std::get<Vector>(v) )
-                    j.emplace_back(to_json(x, std::get<Vector>(v).type));
+                const auto& vec = std::get<Vector>(v);
+                broker::vector bv;
+                for ( const auto& x : vec )
+                    bv.emplace_back(to_broker(x, vec.type));
 
-                type = "vector";
-                value = std::move(j);
+                value = std::move(bv);
                 break;
             }
         }
     }
-    else {
-        type = "none";
-        value = nlohmann::json::object();
-    }
 
-    nlohmann::json json;
-    json["@data-type"] = type;
-    json["data"] = value;
-    return json;
+    return value;
 }
 
 // Best effort type guessing.
-static std::pair<Value, value::Type> from_json(const nlohmann::json& json) {
-    const auto& type = json["@data-type"];
-    const auto& value = json["data"];
-
-    if ( type == "integer" )
-        return {value.get<int64_t>(), value::Type::Integer};
-
-    if ( type == "count" )
-        return {value.get<int64_t>(), value::Type::Count};
-
-    if ( type == "bool" )
-        return {value.get<bool>(), value::Type::Bool};
-
-    if ( type == "real" )
-        return {value.get<double>(), value::Type::Double};
-
-    if ( type == "string" )
-        return {value.get<std::string>(), value::Type::Text};
-
-    if ( type == "enum-value" )
-        return {value.get<std::string>(), value::Type::Enum};
-
-    if ( type == "none" )
+static std::pair<Value, value::Type> from_broker(const broker::data& v) {
+    if ( auto x = broker::get_if<broker::none>(&v) )
         return {{}, value::Type::Null};
 
-    if ( type == "set" ) {
-        auto type = value::Type::Null;
-
-        std::set<Value> x;
-        for ( const auto& i : value ) {
-            auto m = from_json(i);
-            x.insert(m.first);
-            type = m.second;
-        }
-
-        return {Set(type, std::move(x)), value::Type::Set};
-    }
-
-    if ( type == "vector" ) {
-        // We can't distinguish vectors from records, but we only need the
-        // latter right now ...
-        Record y;
-        for ( const auto& i : value )
-            y.emplace_back(from_json(i));
-
-        return {y, value::Type::Record};
-    }
-
-    if ( type == "timespan" )
-        return {to_interval_from_secs(std::stoi(value.get<std::string>())), value::Type::Interval};
-
-    if ( type == "timestamp" ) {
-        std::tm tm = {};
-        std::istringstream ss(value.get<std::string>());
-        ss >> std::get_time(&tm, "%FT%T");
-        auto tp = std::chrono::system_clock::from_time_t(std::mktime(&tm));
-        return {tp, value::Type::Time};
-    }
-
-    /* Not supported, don't need these.
-     *
-     * if ( auto x = broker::get_if<broker::address>(&v) )
-     * else if ( auto x = broker::get_if<broker::subnet>(&v) )
-     * else if ( auto x = broker::get_if<broker::port>(&v) )
-     * else if ( auto x = broker::get_if<broker::table>(&v) )
-     * else if ( auto x = broker::get_if<broker::timestamp>(&v) )
-     */
-
-    throw InternalError(frmt("unsupported data type received over WebSocket ({})", type));
-}
-
-void WebSocketTransport::connect(const std::string& host, unsigned int port, const std::vector<std::string>& topics) {
-    const auto& options = connection()->options();
-
-    auto url = frmt("{}://{}:{}/v1/messages/json", (options.zeek_ssl_disable ? "ws" : "wss"), host, port);
-
-    _socket.setUrl(url);
-    _socket.setPingInterval(30);
-    _socket.enablePong();
-    _socket.enablePerMessageDeflate();
-    _host = host;
-    _port = port;
-
-    if ( ! connection()->options().zeek_ssl_disable ) {
-        ix::SocketTLSOptions tls_options;
-        tls_options.tls = true;
-        tls_options.disable_hostname_validation = true; // with Zeek, we generally don't validate hostnames
-
-        const auto auth_disabled = (options.zeek_ssl_certificate.empty() && options.zeek_ssl_keyfile.empty() &&
-                                    options.zeek_ssl_cafile.empty() && options.zeek_ssl_capath.empty());
-
-        if ( ! auth_disabled ) {
-            if ( ! options.zeek_ssl_certificate.empty() )
-                tls_options.certFile = options.zeek_ssl_certificate;
-
-            if ( ! options.zeek_ssl_keyfile.empty() )
-                tls_options.keyFile = options.zeek_ssl_keyfile;
-
-            if ( ! options.zeek_ssl_cafile.empty() )
-                tls_options.caFile = options.zeek_ssl_cafile;
-            else
-                tls_options.caFile = "SYSTEM";
-
-            if ( ! options.zeek_ssl_capath.empty() )
-                logger()->warn("option zeek.ssl_cpath not supported for WebSocket connections, ignoring");
-        }
-
-        else {
-            // Enable encryption, but don't require authentication (and not even certificates).
-            tls_options.caFile = "NONE";
-            // Use a cipher that don't need a certificate; this list is
-            // borrowed from Broker, including ciphers that works with
-            // different OpenSSL versions
-            tls_options.ciphers = "AECDH-AES256-SHA@SECLEVEL=0:AECDH-AES256-SHA:P-384";
-        }
-
-        _socket.setTLSOptions(tls_options);
-    }
-
-    // We implement our own auto-connect, too difficult to control otherwise.
-    _socket.disableAutomaticReconnection();
+    if ( auto x = broker::get_if<broker::boolean>(&v) )
+        return {*x, value::Type::Bool};
 
-    _socket.setOnMessageCallback([this, topics](const ix::WebSocketMessagePtr& msg) {
-        auto msg_type = msg->type;
-        auto msg_str = msg->str;
-        auto msg_error_reason = msg->errorInfo.reason;
+    if ( auto x = broker::get_if<broker::count>(&v) )
+        return {static_cast<int64_t>(*x), value::Type::Count};
 
-        connection()->scheduler()->schedule([this, topics, msg_type, msg_str,
-                                             msg_error_reason]() { // process message on the main thread
-            switch ( msg_type ) {
-                case ix::WebSocketMessageType::Open: {
-                    connection()->connectionEstablished(this, _host, _port);
-                    connection()->processConnectivityChange(ConnectivityChange::Added,
-                                                            frmt("connected to WebSocket endpoint at {}",
-                                                                 connection()->endpoint()));
+    if ( auto x = broker::get_if<broker::integer>(&v) )
+        return {static_cast<int64_t>(*x), value::Type::Integer};
 
-                    _socket.send(nlohmann::json(topics).dump());
-                    _connected = true;
-                    break;
-                }
+    if ( auto x = broker::get_if<broker::real>(&v) )
+        return {*x, value::Type::Double};
 
-                case ix::WebSocketMessageType::Close: {
-                    if ( _connected )
-                        connection()->processConnectivityChange(ConnectivityChange::Removed,
-                                                                frmt("disconnected from WebSocket endpoint at {}",
-                                                                     connection()->endpoint()));
-                    _connected = false;
-                    break;
-                }
+    if ( auto x = broker::get_if<std::string>(&v) )
+        return {*x, value::Type::Text};
 
-                case ix::WebSocketMessageType::Error: {
-                    if ( _connected )
-                        connection()->processConnectivityChange(ConnectivityChange::Lost,
-                                                                frmt("lost connection to WebSocket endpoint at {} ({})",
-                                                                     connection()->endpoint(), msg_error_reason));
-                    else {
-                        std::string reason = trim(msg_error_reason);
+    if ( auto x = broker::get_if<broker::enum_value>(&v) )
+        return {x->name, value::Type::Enum};
 
-                        if ( reason.find("Connect error:") != std::string::npos ||
-                             reason.find("Cancelled") != std::string::npos )
-                            // We get "error: Connect error: Bad file
-                            // descriptor" or "error: Cancelled" if noone is
-                            // listening at the destination port.
-                            reason = "";
+    if ( auto x = broker::get_if<broker::set>(&v) ) {
+        auto type = value::Type::Null;
+        if ( ! x->empty() )
+            type = from_broker(*x->begin()).second;
 
-                        logger()->debug("cannot connect to Zeek endpoint via WebSocket at {}{}",
-                                        connection()->endpoint(), (reason.empty() ? "" : frmt(" ({})", reason)));
-                        connection()->connectionAttemptFailed(this, reason);
-                    }
+        Set y(type);
+        for ( const auto& i : *x )
+            y.insert(from_broker(i).first);
 
-                    _connected = false;
-                    break;
-                }
+        return {y, value::Type::Set};
+    }
 
-                case ix::WebSocketMessageType::Message:
-                    try {
-                        auto json = nlohmann::json::parse(msg_str);
+    if ( auto x = broker::get_if<broker::vector>(&v) ) {
+        // We can't distinguish vectors from records, but we only need the
+        // latter right now ...
+        Record y;
+        for ( const auto& i : *x )
+            y.emplace_back(from_broker(i));
 
-                        if ( json["type"] == "ack" ) {
-                            auto endpoint_ = json["endpoint"].get<std::string>();
-                            auto broker_version = json["version"].get<std::string>();
-                            ZEEK_CONN_DEBUG("received acknowledgment (endpoint={} broker={})", endpoint_,
-                                            broker_version);
-                        }
+        return {y, value::Type::Record};
+    }
 
-                        else if ( json["type"] == "data-message" ) {
-                            auto [data_, type] = from_json(json);
-                            const auto& data = std::get<Record>(data_);
+    if ( auto x = broker::get_if<broker::timespan>(&v) )
+        return {std::chrono::duration_cast<Interval>(*x), value::Type::Interval};
 
-                            if ( std::get<int64_t>(data[0].first) != 1 || std::get<int64_t>(data[1].first) != 1 ) {
-                                ZEEK_CONN_DEBUG("unexpected content enums for data-message: {}", to_string(data));
-                                return;
-                            }
+    /* Not supported, don't need these.
+     *
+     * if ( auto x = broker::get_if<broker::address>(&v) )
+     * else if ( auto x = broker::get_if<broker::subnet>(&v) )
+     * else if ( auto x = broker::get_if<broker::port>(&v) )
+     * else if ( auto x = broker::get_if<broker::table>(&v) )
+     * else if ( auto x = broker::get_if<broker::timestamp>(&v) )
+     */
 
-                            auto event = std::get<Record>(data[2].first);
-                            const auto& event_name = std::get<std::string>(event[0].first);
-                            const std::vector<std::pair<Value, value::Type>>& event_args =
-                                std::get<Record>(event[1].first);
+    throw InternalError(frmt("unsupported Broker data type received ({})", broker::to_string(v)));
+}
 
-                            connection()->processEvent(event_name,
-                                                       transform(event_args, [](auto i) { return i.first; }));
-                        }
+broker::configuration NativeBrokerTransport::brokerConfig() {
+    const auto& options = _config.options();
+    // Configure Broker/CAF for lower resource consumption.
+    broker::broker_options broker_options;
+    broker_options.forward = false;
+    broker_options.ignore_broker_conf = true;
+    broker_options.disable_ssl = options.zeek_ssl_disable;
 
-                        else if ( json["type"] == "error" ) {
-                            auto code = json["code"].get<std::string>();
-                            auto context = json["context"].get<std::string>();
-                            connection()->processError(
-                                frmt("WebSocket error for {}: {} ({})", connection()->endpoint(), code, context));
-                        }
+    broker::configuration broker_config(broker_options);
+    broker_config.openssl_cafile(options.zeek_ssl_cafile);
+    broker_config.openssl_capath(options.zeek_ssl_capath);
+    broker_config.openssl_certificate(options.zeek_ssl_certificate);
+    broker_config.openssl_key(options.zeek_ssl_keyfile);
+    broker_config.openssl_passphrase(options.zeek_ssl_passphrase);
 
-                        else {
-                            ZEEK_CONN_DEBUG("unexpected message type {}", json["@type"]);
-                            return;
-                        }
-                    }
+#if 1
+    broker_config.set("caf.scheduler.policy", "sharing");
+    broker_config.set("caf.scheduler.max-threads", 1);
+    broker_config.set("caf.middleman.workers", 0);
+#else
+    // Use Zeek's stealing configuration.
+    broker_config.set("caf.work-stealing.moderate-sleep-duration", broker::timespan{16'000});
+    broker_config.set("caf.work-stealing.relaxed-sleep-duration", broker::timespan{64'000});
+    broker_config.set("caf.work-stealing.aggressive-poll-attempts", 5);
+    broker_config.set("caf.work-stealing.moderate-poll-attempts", 5);
+    broker_config.set("caf.work-stealing.aggressive-steal-interval", 4);
+    broker_config.set("caf.work-stealing.moderate-steal-interval", 2);
+    broker_config.set("caf.work-stealing.relaxed-steal-interval", 1);
+#endif
 
-                    catch ( const std::exception& e ) {
-                        ZEEK_CONN_DEBUG("cannot parse WebSocket message: {} ({})", e.what(), msg_str);
-                    }
+    return broker_config;
+}
 
-                    break;
+void NativeBrokerTransport::connect(const std::string& host, unsigned int port,
+                                    const std::vector<std::string>& topics) {
+    auto broker_topics = transform(topics, [](const auto& t) { return broker::topic(t); });
+    broker_topics.emplace_back(std::string(broker::topic::errors_str));
+    broker_topics.emplace_back(std::string(broker::topic::statuses_str));
 
-                case ix::WebSocketMessageType::Ping:
-                case ix::WebSocketMessageType::Pong:
-                case ix::WebSocketMessageType::Fragment: break;
-            }
-        });
-    });
+    _endpoint = std::make_unique<broker::endpoint>(brokerConfig());
+    _endpoint->subscribe_nosync(
+        std::move(broker_topics), []() { /* nop */ },
+        [this](const broker::data_message& msg) {
+            connection()->scheduler()->schedule([this, msg]() { // process message on the main thread
+                if ( broker::get_topic(msg) == broker::topic::statuses_str ) {
+                    auto x = broker::to<broker::status>(broker::get_data(msg));
+                    assert(x); // NOLINT(bugprone-lambda-function-name)
+                    processConnectivityChange(*x);
+                }
+                else if ( broker::get_topic(msg) == broker::topic::errors_str ) {
+                    auto x = broker::to<broker::error>(broker::get_data(msg));
+                    assert(x); // NOLINT(bugprone-lambda-function-name)
+                    processError(*x);
+                }
+                else
+                    processEvent(msg);
+            });
+        },
+        [](const broker::error&) { /* nop */ });
 
-    ZEEK_CONN_DEBUG("WebSocket endpoint: {}", url);
-    tryReconnect();
+    _destination = broker::network_info(host, port,
+                                        std::chrono::duration_cast<broker::timeout::seconds>(
+                                            _config.options().zeek_reconnect_interval));
+
+    // Broker's peer_nosync() has not version taking netinfo directly
+    _endpoint->peer_nosync(_destination->address, _destination->port, _destination->retry);
 }
 
-void WebSocketTransport::tryReconnect() {
-    if ( _socket.getReadyState() != ix::ReadyState::Closed )
-        // Still/already doing something on the connection.
+void NativeBrokerTransport::disconnect() {
+    if ( ! _endpoint )
         return;
 
-    if ( _last_connect_attempt &&
-         (std::chrono::system_clock::now() - *_last_connect_attempt) < _config.options().zeek_reconnect_interval )
+    if ( ! _endpoint->unpeer(_destination->address, _destination->port) )
+        logger()->warn("failed disconnect from {}", connection()->endpoint());
+
+    _endpoint->shutdown();
+    _endpoint = nullptr;
+}
+
+void NativeBrokerTransport::transmitEvent(const std::string& topic, const std::string& event_name, Record args) {
+    if ( ! _endpoint )
         return;
 
-    _socket.stop();
-    _socket.start();
-    _last_connect_attempt = std::chrono::system_clock::now();
+    broker::zeek::Event event(event_name, broker::get<broker::vector>(to_broker(std::move(args), value::Type::Record)));
+    _endpoint->publish(topic, std::move(event));
 }
 
-void WebSocketTransport::disconnect() {
-    _socket.stop();
-    _last_connect_attempt.reset();
-    _connected = false;
+void NativeBrokerTransport::processEvent(const broker::data_message& msg) {
+    broker::zeek::Event event(std::get<1>(msg.data()));
+
+    std::vector<Value> args;
+    for ( const auto& a : event.args() )
+        args.push_back(from_broker(a).first);
+
+    connection()->processEvent(event.name(), args);
 }
 
-void WebSocketTransport::poll() {
-    if ( _last_connect_attempt && ! _connected )
-        tryReconnect();
+void NativeBrokerTransport::processError(const broker::error& err) {
+    std::string msg = "<no error message>";
+    if ( err.message() )
+        msg = *err.message();
+
+    switch ( static_cast<broker::ec>(err.code()) ) {
+        // Prettyify some common errors.
+        case broker::ec::peer_invalid:
+        case broker::ec::peer_unavailable:
+            logger()->debug("cannot connect to Zeek endpoint via Broker at {}", connection()->endpoint());
+            connection()->connectionAttemptFailed(this, msg);
+            break;
+
+        default: connection()->processError(frmt("{} for {}", msg, connection()->endpoint()));
+    }
 }
 
-void WebSocketTransport::transmitEvent(const std::string& topic, const std::string& event_name, Record args) {
-    broker::zeek::Event ev(event_name, broker::get<broker::vector>(to_broker(args, value::Type::Record)));
-    auto event = Record({{Value(1L), value::Type::Count},
-                         {Value(1L), value::Type::Count},
-                         {Record{{
-                              {Value(event_name), value::Type::Text},
-                              {std::move(args), value::Type::Record},
-                          }},
-                          value::Type::Record}});
+void NativeBrokerTransport::processConnectivityChange(const broker::status& status) {
+    ConnectivityChange change = ConnectivityChange::Other;
+    auto msg = (status.message() ? *status.message() : std::string("<no status description from broker>"));
 
-    nlohmann::json msg = to_json(event, value::Type::Record);
-    msg["type"] = "data-message";
-    msg["topic"] = topic;
-    _socket.send(msg.dump());
+    switch ( static_cast<broker::sc>(status.code()) ) {
+        // Prettyify some common messages.
+        case broker::sc::peer_added:
+            connection()->connectionEstablished(this, _destination.value().address, _destination.value().port);
+            change = ConnectivityChange::Added;
+            msg = frmt("connected to Zeek endpoint at {}", connection()->endpoint());
+            break;
+        case broker::sc::peer_lost:
+            change = ConnectivityChange::Lost;
+            msg = frmt("lost connection to Zeek endpoint at {}", connection()->endpoint());
+            break;
+        case broker::sc::peer_removed:
+            change = ConnectivityChange::Removed;
+            msg = frmt("disconnected from Zeek endpoint at {}", connection()->endpoint());
+            break;
+        default: break;
+    }
+
+    connection()->processConnectivityChange(change, msg);
 }
+#endif
 
 ///// Transport-indenpendent Zeek communication code.
 
@@ -1480,7 +1481,9 @@ void Zeek::Implementation::start(const std::vector<std::string>& zeeks) {
     for ( const auto& z : zeeks ) {
         auto conn = std::make_unique<ZeekConnection>(_db, _scheduler);
         conn->addTransport(std::make_unique<WebSocketTransport>(_db->configuration()));
+#ifdef HAVE_BROKER
         conn->addTransport(std::make_unique<NativeBrokerTransport>(_db->configuration()));
+#endif
 
         if ( auto rc = conn->connect(z) )
             _connections.push_back(std::move(conn));
@@ -1529,6 +1532,7 @@ void Zeek::poll() {
 }
 
 TEST_SUITE("Zeek") {
+#ifdef HAVE_BROKER
     TEST_CASE("connect/hello/disconnect/reconnect - native Broker" * doctest::timeout(10.0)) {
         Configuration cfg;
         Scheduler tmgr;
@@ -1627,4 +1631,5 @@ TEST_SUITE("Zeek") {
         zeek.stop();
         wait_for_disconnect();
     }
+#endif
 }
diff --git a/src/main.cc b/src/main.cc
index b28e1427..683c7a94 100644
--- a/src/main.cc
+++ b/src/main.cc
@@ -114,7 +114,7 @@ int zeek::agent::main(const std::vector<std::string>& argv) {
         if ( ! platform::runningAsAdmin() && ! cfg.options().use_mock_data )
             logger()->warn("not running as root, information may be incomplete");
 
-        platform::init(cfg);
+        platform::init(&cfg);
 
         Scheduler scheduler;
         signal_mgr = new SignalManager({SIGINT});
diff --git a/src/platform/darwin/ZeekAgent.app/XPC.swift b/src/platform/darwin/ZeekAgent.app/XPC.swift
index d629d11c..bd6b8560 100644
--- a/src/platform/darwin/ZeekAgent.app/XPC.swift
+++ b/src/platform/darwin/ZeekAgent.app/XPC.swift
@@ -1,6 +1,7 @@
 // Copyright (c) 2021 by the Zeek Project. See LICENSE for details.
 
 import Foundation
+import os.log
 
 // XPC protocol between container app to communicate with the extension. The
 // app the caller, the extension the receiver. Any changes here need to be
@@ -17,6 +18,12 @@ class XPC {
     private var service: IPCProtocol?
     private var connection: NSXPCConnection?
 
+    private let logger = Logger(subsystem: "org.zeek.zeek-agent", category: "installer")
+
+    func log(_ msg: String) {
+        logger.info("\(msg, privacy: .public)")
+    }
+
     // Returns true if the extension is currently running (i.e., we can
     // communicate with it). If yes, returns the version string of the running instance.
     public func isExtensionRunning() -> String? {
@@ -72,6 +79,11 @@ class XPC {
     private func connectXPC() -> IPCProtocol? {
         if connection == nil {
             connection = NSXPCConnection(machServiceName: "org.zeek.zeek-agent.agent")
+            if connection == nil {
+                logger.error("Failed to create XPC connection")
+                return nil
+            }
+
             connection!.remoteObjectInterface = NSXPCInterface(with: IPCProtocol.self)
             connection!.interruptionHandler = { self.resetConnection() }
             connection!.invalidationHandler = { self.resetConnection() }
diff --git a/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.agent b/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.agent
index 18bae57f..7f2f9a24 100644
Binary files a/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.agent and b/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.agent differ
diff --git a/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.app b/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.app
index 0cf46335..028247ef 100644
Binary files a/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.app and b/src/platform/darwin/ZeekAgent.app/embedded.provisionprofile.app differ
diff --git a/src/platform/darwin/ZeekAgent.app/main.swift b/src/platform/darwin/ZeekAgent.app/main.swift
index 07cc29d0..1b6606f2 100644
--- a/src/platform/darwin/ZeekAgent.app/main.swift
+++ b/src/platform/darwin/ZeekAgent.app/main.swift
@@ -10,6 +10,10 @@ import os.log
 // or not the right version.
 let debug_open_main_directly = false
 
+// For debugging, set this to true to enable the "Configure ..." button
+// independent of the extension state.
+let debug_always_enable_configure = false
+
 struct MainView: View {
     enum ExtensionState {
         case Running
@@ -85,7 +89,8 @@ struct MainView: View {
 
                         case .Running:
                             let headline = "Zeek Agent is running"
-                            let subheadline = "Version \(status.version) \(status.capabilities)\n\nAgent ID \(status.agent_id)"
+                            let subheadline =
+                                "Version \(status.version) \(status.capabilities)\n\nAgent ID \(status.agent_id)"
 
                             if status.capabilities.contains("+ES") {
                                 mainBody(headline: headline, subheadline: subheadline)
@@ -115,7 +120,7 @@ struct MainView: View {
 
                     HStack {
                         Button("Configure ...") { Controller.shared.openConfiguration() }
-                            .disabled(state != .Running)
+                            .disabled(state != .Running && !debug_always_enable_configure)
 
                         Button("Exit") { Controller.shared.application.terminate(nil) }
                             .keyboardShortcut(.defaultAction)
@@ -126,7 +131,10 @@ struct MainView: View {
             Spacer()
         }
 
-        .onAppear { updateStatus() }
+        .onAppear {
+            updateStatus()
+        }
+
         .onReceive(timer) { _ in
             DispatchQueue.main.async {
                 updateStatus()
@@ -150,21 +158,30 @@ struct MainView: View {
 }
 
 struct ConfigurationView: View {
+    var window: NSWindow? = nil
+
     @State private var log_level: String = ""
     @State private var zeek_destination: String = ""
-    @State private var old_options: [String: String]?
 
-    func asOptions() -> [String: String] {
+    private let logger = Logger(subsystem: "org.zeek.zeek-agent", category: "installer")
+
+    func optionAsDictionary() -> [String: String] {
         return [
             "log.level": log_level,
             "zeek.destination": zeek_destination,
         ]
     }
 
+    func close() {
+        if let window = window {
+            window.sheetParent?.endSheet(window)
+        }
+    }
+
     var body: some View {
-        VStack(alignment: .trailing) {
+        VStack() {
             Form {
-                TextField("ZeekHost", text: $zeek_destination)
+                TextField("Zeek Host", text: $zeek_destination)
                     .frame(minWidth: 300)
 
                 Picker("Level", selection: $log_level) {
@@ -177,41 +194,34 @@ struct ConfigurationView: View {
                 }
             }
             .onAppear {
-                old_options = asOptions()
                 _ = Controller.shared.xpc.getOptions { options in
                     log_level = options["log.level", default: ""]
                     zeek_destination = options["zeek.destination", default: ""]
-                    old_options = asOptions()
                 }
             }
-            .onDisappear {
-                old_options = nil
-            }
 
             HStack {
-                Button("Cancel") { Controller.shared.application.stopModal() }
+                Button("Cancel") { close() }
                     .keyboardShortcut(.cancelAction)
 
                 Button("Save") {
-                    let new_options = asOptions()
-
-                    if let old_options = old_options {
-                        if old_options == new_options {
-                            // No change.
-                            Controller.shared.application.stopModal()
-                            return
-                        }
-                    }
-
-                    if !Controller.shared.xpc.setOptions(new_options) {
+                    let options = optionAsDictionary()
+                    if !Controller.shared.xpc.setOptions(options) {
                         Controller.shared.showMessage(msg: "Could not save options.", sub: "", style: .critical)
                     }
 
-                    Controller.shared.application.stopModal()
+                    close()
+
                 }.keyboardShortcut(.defaultAction)
-            }.padding(.top)
+            }
+            .padding(.top, 10)
         }
-        .padding(25.0)
+        .onAppear {
+            Controller.shared.enableMenuBar(false)
+        }
+
+        .padding(.top)
+        .padding(20.0)
         .fixedSize()
     }
 }
@@ -272,6 +282,14 @@ class Controller {
         logger.info("\(msg, privacy: .public)")
     }
 
+    func enableMenuBar(_ enable: Bool) {
+        if let menu = Controller.shared.application.mainMenu {
+            for item in menu.items {
+                item.isEnabled = enable
+            }
+        }
+    }
+
     func openMain() {
         guard main == nil else { return }
         main = openWindow(
@@ -284,9 +302,11 @@ class Controller {
     }
 
     func openConfiguration() {
-        let window = openWindow(content: ConfigurationView(), title: "Configuration", style: [.titled])
-        application.runModal(for: window)
-        window.close()
+        let window = NSWindow()
+        let view = ConfigurationView(window: window)
+        window.contentViewController = NSHostingController(rootView: view)
+        main?.beginSheet(window) { _ in Controller.shared.enableMenuBar(true) }
+
     }
 
     func openAbout() {
@@ -313,7 +333,6 @@ class Controller {
         controller.view.autoresizesSubviews = false
 
         let window = NSWindow(contentViewController: controller)
-        //window.styleMask.formUnion(.fullSizeContentView)
         window.title = title
         window.styleMask = style
         window.center()
diff --git a/src/platform/darwin/endpoint-security.h b/src/platform/darwin/endpoint-security.h
index e2119f9b..6094f432 100644
--- a/src/platform/darwin/endpoint-security.h
+++ b/src/platform/darwin/endpoint-security.h
@@ -2,31 +2,80 @@
 
 #pragma once
 
-#include "util/pimpl.h"
 #include "util/result.h"
 
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <EndpointSecurity/EndpointSecurity.h>
+
 namespace zeek::agent::platform::darwin {
 
+class EndpointSecurity;
+
+namespace es {
+
+/**
+ * State for an active subscription to EndpointSecurity events. An instance of
+ * this class is creating when subscribing to ES events, and the subscription
+ * is kept active as long as the instance exists.
+ */
+class Subscriber {
+public:
+    ~Subscriber();
+
+private:
+    friend class darwin::EndpointSecurity;
+    Subscriber(std::string tag, es_client_t* client);
+
+    std::string _tag;
+    es_client_t* _client = nullptr;
+};
+
+} // namespace es
+
 /**
  * Wrapper around macOS's Endpoint Security API. This encapsulates API state
  * across multiple clients, maintaining just a single internal copy.
  */
-class EndpointSecurity : public Pimpl<EndpointSecurity> {
+class EndpointSecurity {
 public:
+    using Events = std::vector<es_event_type_t>;
+    using Callback = std::function<void(const es_message_t*)>;
+
     ~EndpointSecurity();
 
     /**
-     * Returns success if EndpointSecurity has been initialized successfully,
-     * or an error otherwise.
+     * Returns success if EndpointSecurity is available for use, or a error
+     * otherwise describibing why it's not.
      *
      * @returns success or an appropiate error message if ES isn't available;
      * then no functionlity must be used
      */
-    Result<Nothing> isAvailable();
+    Result<Nothing> isAvailable() { return _init_result; }
+
+    /**
+     * Creates a subscription to EndpointSecurity events.
+     *
+     * @param tag descriptive tag identifying the subscription in log messages
+     *
+     * @param events set of events to subscribe to; see
+     * https://developer.apple.com/documentation/endpointsecurity/es_event_type_t for the list
+     *
+     * @param callback callback to invoke when an event is received
+     *
+     * @returns a subscription objects that keeps the subscription active as long as it exists, or an error if the
+     * subscription could not be created
+     */
+    Result<std::unique_ptr<es::Subscriber>> subscribe(std::string tag, const Events& events, Callback callback);
 
 private:
     friend EndpointSecurity* endpointSecurity();
+
     EndpointSecurity();
+
+    Result<Nothing> _init_result; // caches result of `isAvailable()`
 };
 
 /** Returns the global `EndpointSecurity` singleton. */
diff --git a/src/platform/darwin/endpoint-security.mm b/src/platform/darwin/endpoint-security.mm
index adbe51e6..a5c3fb09 100644
--- a/src/platform/darwin/endpoint-security.mm
+++ b/src/platform/darwin/endpoint-security.mm
@@ -7,77 +7,77 @@
 
 #include "autogen/config.h"
 #include "core/logger.h"
+#include "util/helpers.h"
 
 #include <EndpointSecurity/EndpointSecurity.h>
 
 using namespace zeek::agent;
 using namespace zeek::agent::platform::darwin;
 
-template<>
-struct Pimpl<EndpointSecurity>::Implementation {
-    // Initialize ES, if not done yet.
-    Result<Nothing> init();
+es::Subscriber::Subscriber(std::string tag, es_client_t* client) : _tag(std::move(tag)), _client(client) {
+    if ( ! _client )
+        return;
 
-    // Shutdown ES, if running.
-    void done();
+    ZEEK_AGENT_DEBUG("darwin", "[EndpointSecurity] [{}] subscription created", _tag);
+}
+
+es::Subscriber::~Subscriber() {
+    if ( ! _client )
+        return;
+
+    es_delete_client(_client);
+    ZEEK_AGENT_DEBUG("darwin", "[EndpointSecurity] [{}] subscription deleted", _tag);
+}
 
-    es_client_t* _es_client = nullptr;
-    Result<Nothing> _es_init_result;
-};
+EndpointSecurity::EndpointSecurity() {
+    if ( auto rc = subscribe("CheckAvailability", {}, [](const es_message_t*) {}) ) {
+        ZEEK_AGENT_DEBUG("darwin", "[EndpointSecurity] available");
+        _init_result = Nothing();
+    }
+    else {
+        ZEEK_AGENT_DEBUG("darwin", "[EndpointSecurity] not available: {}", rc.error());
+        _init_result = rc.error();
+    }
+}
 
-static es_handler_block_t dummy_handler = ^(es_client_t* clt, const es_message_t* msg) {
-};
+EndpointSecurity::~EndpointSecurity() {}
 
-Result<Nothing> EndpointSecurity::Implementation::init() {
-    es_new_client_result_t res = es_new_client(&_es_client, dummy_handler);
+Result<std::unique_ptr<es::Subscriber>> EndpointSecurity::subscribe(std::string tag, const Events& events,
+                                                                    Callback callback) { // NOLINT
+    es_client_t* client;
+    es_new_client_result_t res = es_new_client(&client, ^(es_client_t* c, const es_message_t* msg) {
+      callback(msg);
+    });
 
     switch ( res ) {
-        case ES_NEW_CLIENT_RESULT_SUCCESS: _es_init_result = Nothing(); break;
+        case ES_NEW_CLIENT_RESULT_SUCCESS: {
+            if ( events.size() ) {
+                if ( es_subscribe(client, events.data(), events.size()) != ES_RETURN_SUCCESS )
+                    return result::Error("failed to subscribe to EndpointSecurity events");
+            }
+
+            return std::unique_ptr<es::Subscriber>(
+                new es::Subscriber(std::move(tag), events.size() ? client : nullptr));
+        }
 
         case ES_NEW_CLIENT_RESULT_ERR_NOT_ENTITLED:
-            _es_init_result =
-                result::Error("macOS entitlement not available (com.apple.developer.endpoint-security.client)");
-            break;
+            return result::Error("macOS entitlement not available (com.apple.developer.endpoint-security.client)");
 
         case ES_NEW_CLIENT_RESULT_ERR_NOT_PERMITTED:
-            _es_init_result = result::Error(
+            return result::Error(
                 "Application lacks Transparency, Consent, and Control (TCC) approval "
                 "from the user. This can be resolved by granting 'Full Disk Access' from "
                 "the 'Security & Privacy' tab of System Preferences.");
-            break;
 
-        case ES_NEW_CLIENT_RESULT_ERR_NOT_PRIVILEGED: _es_init_result = result::Error("not running as root"); break;
-        case ES_NEW_CLIENT_RESULT_ERR_TOO_MANY_CLIENTS: _es_init_result = result::Error("too many clients"); break;
-        case ES_NEW_CLIENT_RESULT_ERR_INVALID_ARGUMENT: _es_init_result = result::Error("invalid argument"); break;
-        case ES_NEW_CLIENT_RESULT_ERR_INTERNAL: _es_init_result = result::Error("internal error"); break;
+        case ES_NEW_CLIENT_RESULT_ERR_NOT_PRIVILEGED: return result::Error("not running as root");
+        case ES_NEW_CLIENT_RESULT_ERR_TOO_MANY_CLIENTS: return result::Error("too many clients");
+        case ES_NEW_CLIENT_RESULT_ERR_INVALID_ARGUMENT: return result::Error("invalid argument");
+        case ES_NEW_CLIENT_RESULT_ERR_INTERNAL: return result::Error("internal error");
     }
 
-    if ( _es_init_result )
-        ZEEK_AGENT_DEBUG("darwin", "EndpointSecurity available");
-    else
-        ZEEK_AGENT_DEBUG("darwin", "EndpointSecurity not available: {}", _es_init_result.error());
-
-    return _es_init_result;
-}
-
-void EndpointSecurity::Implementation::done() {
-    if ( _es_client )
-        es_delete_client(_es_client);
-
-    _es_client = nullptr;
-    _es_init_result = {};
+    cannot_be_reached();
 }
 
-Result<Nothing> EndpointSecurity::isAvailable() {
-    if ( pimpl()->_es_init_result )
-        return Nothing();
-    else
-        return pimpl()->init();
-}
-
-EndpointSecurity::EndpointSecurity() { pimpl()->init(); }
-EndpointSecurity::~EndpointSecurity() { pimpl()->done(); }
-
 EndpointSecurity* platform::darwin::endpointSecurity() {
     static auto es = std::unique_ptr<EndpointSecurity>{};
 
diff --git a/src/platform/darwin/os-log-sink.h b/src/platform/darwin/os-log-sink.h
index b0de860e..12ea7242 100644
--- a/src/platform/darwin/os-log-sink.h
+++ b/src/platform/darwin/os-log-sink.h
@@ -13,7 +13,7 @@ namespace zeek::agent::platform::darwin {
 class OSLogSink : public spdlog::sinks::base_sink<std::mutex> {
 public:
     OSLogSink();
-    ~OSLogSink() final;
+    ~OSLogSink() override;
 
 protected:
     void sink_it_(const spdlog::details::log_msg& msg) override;
diff --git a/src/platform/darwin/platform.mm b/src/platform/darwin/platform.mm
index d9e2c3e2..e55a6064 100644
--- a/src/platform/darwin/platform.mm
+++ b/src/platform/darwin/platform.mm
@@ -46,9 +46,10 @@
     return filesystem::path([dir UTF8String]) / "ZeekAgent";
 }
 
-void platform::init(const Configuration& cfg) {
-    platform::darwin::endpointSecurity();       // this initializes ES
-    [[IPC sharedObject] setConfiguration:&cfg]; // create the shared IPC object
+void platform::init(Configuration* cfg) {
+    platform::darwin::endpointSecurity();      // this initializes ES
+    [[IPC sharedObject] setConfiguration:cfg]; // create the shared IPC object
+    [[IPC sharedObject] updateOptions];        // read options from defaults and update the configuration
 }
 
 void platform::done() {}
@@ -58,24 +59,3 @@
         // Running as an installed system extension, log to oslog by default.
         options->log_type = options::LogType::System;
 }
-
-std::optional<std::string> platform::retrieveConfigurationOption(const std::string& path) {
-    NSString* key = nullptr;
-    NSString* value = nullptr;
-
-    ScopeGuard _([&]() {
-        if ( key )
-            CFRelease(key);
-
-        if ( value )
-            CFRelease(value);
-    });
-
-    key = [NSString stringWithUTF8String:path.c_str()];
-    value = [[[IPC sharedObject] defaults] stringForKey:key];
-
-    if ( value )
-        return [value UTF8String];
-    else
-        return {};
-}
diff --git a/src/platform/darwin/xpc.h b/src/platform/darwin/xpc.h
index d86f84eb..34a4cbb6 100644
--- a/src/platform/darwin/xpc.h
+++ b/src/platform/darwin/xpc.h
@@ -20,7 +20,8 @@
 + (IPC*)sharedObject;
 - (id)init;
 - (const zeek::agent::Options&)options;
+- (void)updateOptions;
 @property(strong) NSUserDefaults* defaults;
 @property(strong) NSXPCListener* listener;
-@property const zeek::agent::Configuration* configuration;
+@property zeek::agent::Configuration* configuration;
 @end
diff --git a/src/platform/darwin/xpc.mm b/src/platform/darwin/xpc.mm
index e3a96445..c1f2bbc4 100644
--- a/src/platform/darwin/xpc.mm
+++ b/src/platform/darwin/xpc.mm
@@ -22,13 +22,42 @@ + (IPC*)sharedObject {
 
 - (instancetype)init {
     [super init];
-    _defaults = [NSUserDefaults standardUserDefaults];
+    _defaults = [[NSUserDefaults alloc] initWithSuiteName:@"group.org.zeek.zeek-agent"];
     _listener = [[NSXPCListener alloc] initWithMachServiceName:@"org.zeek.zeek-agent.agent"];
     _listener.delegate = self;
     [_listener resume];
+
     return self;
 }
 
+- (void)dealloc {
+    [[NSNotificationCenter defaultCenter] removeObserver:self];
+    [super dealloc];
+}
+
+- (void)updateOptions {
+    logger()->debug("updating configuration");
+
+    auto options = _configuration->options();
+
+    auto log_level = [_defaults stringForKey:@"log.level"];
+    if ( log_level ) {
+        if ( [log_level isEqual:@""] )
+            options.log_level = options::default_log_level;
+        else if ( auto rc = options::log_level::from_str([log_level UTF8String]) )
+            options.log_level = *rc;
+        else
+            logger()->warn("invalid log level: {}", [log_level UTF8String]);
+    }
+
+    auto zeek_destination = [_defaults stringForKey:@"zeek.destination"];
+    if ( zeek_destination )
+        options.zeek_destinations = {[zeek_destination UTF8String]};
+
+    if ( auto rc = _configuration->setOptions(options); ! rc )
+        logger()->warn("error applying new options: {}", rc.error());
+}
+
 - (const Options&)options {
     return _configuration->options();
 }
@@ -42,7 +71,6 @@ - (BOOL)listener:(NSXPCListener*)listener shouldAcceptNewConnection:(NSXPCConnec
 
 - (void)getStatusWithReply:(void (^)(NSString*, NSString*, NSString*))reply {
     logger()->debug("[IPC] remote call: getStatus");
-
     auto es = (platform::darwin::endpointSecurity()->isAvailable() ? "+ES" : "-ES");
     auto ne = (platform::darwin::networkExtension()->isAvailable() ? "+NE" : "-NE");
 
@@ -58,15 +86,20 @@ - (void)getOptionsWithReply:(void (^)(NSDictionary<NSString*, NSString*>*))reply
 
     auto options = [NSMutableDictionary dictionary];
 
-    NSArray* keys = [NSArray arrayWithObjects:@"zeek.destination", @"log.level", nil];
-    for ( id key in keys ) {
-        NSString* value = [_defaults stringForKey:key];
-        options[key] = (value ? value : @"");
-    }
+    auto log_level = [_defaults stringForKey:@"log.level"];
+    if ( log_level )
+        options[@"log.level"] = log_level;
+    else
+        options[@"log.level"] = @"default";
+
+    auto zeek_destination = [_defaults stringForKey:@"zeek.destination"];
+    if ( zeek_destination )
+        options[@"zeek.destination"] = zeek_destination;
+    else
+        options[@"zeek.destination"] = @"";
 
     reply(options);
 
-    CFRelease(keys);
     CFRelease(options);
 }
 
@@ -77,6 +110,8 @@ - (void)setOptions:(NSDictionary<NSString*, NSString*>*)options {
         auto value = [options objectForKey:key];
         [_defaults setObject:value forKey:key];
     }
+
+    [self updateOptions];
 }
 
 - (void)exit {
diff --git a/src/platform/linux/platform.cc b/src/platform/linux/platform.cc
index 182157fe..03c5ebac 100644
--- a/src/platform/linux/platform.cc
+++ b/src/platform/linux/platform.cc
@@ -14,7 +14,7 @@ using namespace zeek::agent;
 
 std::string platform::name() { return "Linux"; }
 
-void platform::init(const Configuration& cfg) {}
+void platform::init(Configuration* cfg) {}
 
 void platform::done() {}
 
@@ -62,8 +62,3 @@ std::optional<filesystem::path> platform::dataDirectory() {
 void platform::initializeOptions(Options* options) {
     // Nothing to do.
 }
-
-std::optional<std::string> platform::retrieveConfigurationOption(const std::string& path) {
-    // Nothing to do.
-    return {};
-}
diff --git a/src/platform/platform.h b/src/platform/platform.h
index 49635e04..1dba90b7 100644
--- a/src/platform/platform.h
+++ b/src/platform/platform.h
@@ -24,7 +24,7 @@
 namespace zeek::agent::platform {
 
 /** Performs one-time initialization at startup. */
-extern void init(const Configuration& cfg);
+extern void init(Configuration* cfg);
 
 /** Performs one-time cleanup at shutdown. */
 extern void done();
@@ -64,14 +64,4 @@ extern bool runningAsAdmin();
  */
 extern void initializeOptions(Options* options);
 
-/**
- * Retrieves the value of an option through platform-specific means. For array
- * values, the expectation is that the elements are returned as a
- * comma-separated string.
- *
- * @param path option's key path as in the TOML file
- * @returns the option's value, if set
- */
-extern std::optional<std::string> retrieveConfigurationOption(const std::string& path);
-
 } // namespace zeek::agent::platform
diff --git a/src/platform/windows/platform.cc b/src/platform/windows/platform.cc
index de330ee7..7f559c53 100644
--- a/src/platform/windows/platform.cc
+++ b/src/platform/windows/platform.cc
@@ -18,7 +18,7 @@ using namespace zeek::agent;
 using namespace zeek::agent::platform::windows;
 using namespace ztd::out_ptr;
 
-void platform::init(const Configuration& cfg) {}
+void platform::init(Configuration* cfg) {}
 
 void platform::done() { windows::WMIManager::Get().Shutdown(); }
 
@@ -132,11 +132,6 @@ void platform::initializeOptions(Options* options) {
     // Nothing to do.
 }
 
-std::optional<std::string> platform::retrieveConfigurationOption(const std::string& path) {
-    // Nothing to do.
-    return {};
-}
-
 WMIManager& WMIManager::Get() {
     static WMIManager wmi;
     return wmi;
diff --git a/src/tables/processes/processes.darwin.cc b/src/tables/processes/processes.darwin.cc
index 74a0349c..99a414fb 100644
--- a/src/tables/processes/processes.darwin.cc
+++ b/src/tables/processes/processes.darwin.cc
@@ -17,28 +17,29 @@ namespace zeek::agent::table {
 
 class ProcessesDarwin : public ProcessesCommon {
 public:
+    ProcessesDarwin() {
+        struct mach_timebase_info tb;
+        if ( mach_timebase_info(&tb) == KERN_SUCCESS )
+            _timebase = tb;
+        else
+            logger()->warn("[processes] cannot get MACH timebase, will disable table");
+    }
+
     std::vector<std::vector<Value>> snapshot(const std::vector<table::Argument>& args) override;
     void addProcess(std::vector<std::vector<Value>>* rows, const struct proc_bsdinfo* p,
                     const struct proc_taskinfo* ti);
 
-    bool init() override;
+    Init init() override;
 
 private:
-    struct mach_timebase_info _timebase;
+    std::optional<struct mach_timebase_info> _timebase;
 };
 
 namespace {
 database::RegisterTable<ProcessesDarwin> _;
 }
 
-bool ProcessesDarwin::init() {
-    if ( mach_timebase_info(&_timebase) != KERN_SUCCESS ) {
-        logger()->warn("[processes] cannot get MACH timebase, disabling table");
-        return false;
-    }
-
-    return true;
-}
+EventTable::Init ProcessesDarwin::init() { return _timebase ? Init::Available : Init::PermanentlyUnavailable; }
 
 std::vector<std::vector<Value>> ProcessesDarwin::snapshot(const std::vector<table::Argument>& args) {
     auto buffer_size = proc_listpids(PROC_ALL_PIDS, 0, nullptr, 0);
@@ -91,10 +92,11 @@ void ProcessesDarwin::addProcess(std::vector<std::vector<Value>>* rows, const st
     Value utime;
     Value stime;
     if ( ti ) {
+        assert(_timebase);
         vsize = static_cast<int64_t>(ti->pti_virtual_size);
         rsize = static_cast<int64_t>(ti->pti_resident_size);
-        utime = to_interval_from_ns(ti->pti_total_user * _timebase.numer / _timebase.denom);
-        stime = to_interval_from_ns(ti->pti_total_system * _timebase.numer / _timebase.denom);
+        utime = to_interval_from_ns(ti->pti_total_user * _timebase->numer / _timebase->denom);
+        stime = to_interval_from_ns(ti->pti_total_system * _timebase->numer / _timebase->denom);
     }
 
     rows->push_back({name, pid, ppid, uid, gid, ruid, rgid, priority, startup, vsize, rsize, utime, stime});
diff --git a/src/tables/processes/processes.linux.cc b/src/tables/processes/processes.linux.cc
index 7171ada2..26399b70 100644
--- a/src/tables/processes/processes.linux.cc
+++ b/src/tables/processes/processes.linux.cc
@@ -16,7 +16,7 @@ namespace zeek::agent::table {
 class ProcessesLinux : public ProcessesCommon {
 public:
     std::vector<std::vector<Value>> snapshot(const std::vector<table::Argument>& args) override;
-    bool init() override;
+    Init init() override;
 
 private:
     uint64_t _clock_tick;
@@ -26,9 +26,9 @@ namespace {
 database::RegisterTable<ProcessesLinux> _;
 }
 
-bool ProcessesLinux::init() {
+EventTable::Init ProcessesLinux::init() {
     _clock_tick = sysconf(_SC_CLK_TCK);
-    return true;
+    return Init::Available;
 }
 
 std::vector<std::vector<Value>> ProcessesLinux::snapshot(const std::vector<table::Argument>& args) {
diff --git a/src/tables/system_logs/system_logs.darwin.cc b/src/tables/system_logs/system_logs.darwin.cc
index b2d8b136..945a62c3 100644
--- a/src/tables/system_logs/system_logs.darwin.cc
+++ b/src/tables/system_logs/system_logs.darwin.cc
@@ -30,7 +30,7 @@ namespace {
 
 class SystemLogsDarwin : public SystemLogs {
 public:
-    bool init() override;
+    Init init() override;
     void activate() override;
     void deactivate() override;
 
@@ -63,7 +63,7 @@ void SystemLogsDarwin::recordEntry(os_activity_stream_entry_t entry, int error)
     free(msg);
 }
 
-bool SystemLogsDarwin::init() { return true; }
+Table::Init SystemLogsDarwin::init() { return Init::Available; }
 
 void SystemLogsDarwin::activate() {
     // Notes:
diff --git a/src/tables/system_logs/system_logs.linux.cc b/src/tables/system_logs/system_logs.linux.cc
index 4ad76ebe..6af1ac24 100644
--- a/src/tables/system_logs/system_logs.linux.cc
+++ b/src/tables/system_logs/system_logs.linux.cc
@@ -24,7 +24,7 @@ namespace {
 
 class SystemLogsLinux : public SystemLogs {
 public:
-    bool init() override;
+    Init init() override;
     void activate() override;
     void deactivate() override;
     void poll() override;
@@ -40,7 +40,7 @@ class SystemLogsLinux : public SystemLogs {
 
 database::RegisterTable<SystemLogsLinux> _;
 
-bool SystemLogsLinux::init() {
+Table::Init SystemLogsLinux::init() {
     // See if we find 'journalctl'
     std::set<filesystem::path> candidates = {"/usr/bin/journalctl", "/usr/local/sbin/journalctl"};
 
@@ -67,10 +67,9 @@ bool SystemLogsLinux::init() {
     else
         ZEEK_AGENT_DEBUG("system_logs", "did not find journalctrl");
 
-    return journalctl.has_value();
+    return journalctl ? Init::Available : Init::PermanentlyUnavailable;
 }
 
-
 void SystemLogsLinux::startProcess() {
     if ( ! journalctl )
         return;
diff --git a/src/tables/zeek_agent/zeek_agent.darwin.mm b/src/tables/zeek_agent/zeek_agent.darwin.mm
index d477916e..64aaac0d 100644
--- a/src/tables/zeek_agent/zeek_agent.darwin.mm
+++ b/src/tables/zeek_agent/zeek_agent.darwin.mm
@@ -13,7 +13,9 @@
 #include <SystemConfiguration/SCDynamicStore.h>
 #include <sys/utsname.h>
 
+#ifdef HAVE_BROKER
 #include <broker/version.hh>
+#endif
 
 using namespace zeek::agent;
 using namespace zeek::agent::table;
@@ -80,7 +82,11 @@ Value addresses() {
     Value platform = platform::name();
     Value os_name = std::string("macOS ") + std::string([version UTF8String]);
     Value agent = options().version_number;
+#ifdef HAVE_BROKER
     Value broker = broker::version::string();
+#else
+    Value broker = "n/a";
+#endif
     Value uptime = std::chrono::system_clock::now() - startupTime();
     Value tables =
         Set(value::Type::Text, transform(database()->tables(), [](const auto* t) { return Value(t->name()); }));
diff --git a/src/tables/zeek_agent/zeek_agent.linux.cc b/src/tables/zeek_agent/zeek_agent.linux.cc
index af4e3318..3d1d231d 100644
--- a/src/tables/zeek_agent/zeek_agent.linux.cc
+++ b/src/tables/zeek_agent/zeek_agent.linux.cc
@@ -15,7 +15,9 @@
 #include <sys/types.h>
 #include <sys/utsname.h>
 
+#ifdef HAVE_BROKER
 #include <broker/version.hh>
+#endif
 
 using namespace zeek::agent;
 using namespace zeek::agent::table;
@@ -144,7 +146,11 @@ std::vector<std::vector<Value>> ZeekAgentLinux::snapshot(const std::vector<table
     Value platform = platform::name();
     Value os_name = distribution();
     Value agent = options().version_number;
+#ifdef HAVE_BROKER
     Value broker = broker::version::string();
+#else
+    Value broker = "n/a";
+#endif
     Value uptime = std::chrono::system_clock::now() - startupTime();
     Value tables =
         Set(value::Type::Text, transform(database()->tables(), [](const auto* t) { return Value(t->name()); }));
diff --git a/src/tables/zeek_agent/zeek_agent.windows.cc b/src/tables/zeek_agent/zeek_agent.windows.cc
index 885ac049..bd6ad057 100644
--- a/src/tables/zeek_agent/zeek_agent.windows.cc
+++ b/src/tables/zeek_agent/zeek_agent.windows.cc
@@ -14,7 +14,9 @@
 #include <winternl.h>
 #include <ws2ipdef.h>
 
+#ifdef HAVE_BROKER
 #include <broker/version.hh>
+#endif
 
 using namespace zeek::agent;
 using namespace zeek::agent::platform::windows;
@@ -109,7 +111,11 @@ std::vector<std::vector<Value>> ZeekAgentWindows::snapshot(const std::vector<tab
     Value platform = platform::name();
     Value os_name = distribution();
     Value agent = options().version_number;
+#ifdef HAVE_BROKER
     Value broker = broker::version::string();
+#else
+    Value broker = "n/a";
+#endif
     Value uptime = std::chrono::system_clock::now() - startupTime();
     Value tables =
         Set(value::Type::Text, transform(database()->tables(), [](const auto* t) { return Value(t->name()); }));