diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,24 @@
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to the [PVP](https://pvp.haskell.org/).
 
+## [Unreleased]
+
+## [0.3.0.0] - 2026-08-06
+
+### Added
+
+- `Atelier.Effects.Process`: `getExecutablePath` returns the absolute path of
+  the currently running executable, for re-invoking the program as a
+  subprocess.
+- `Atelier.Effects.Publishing.Pub:` `map` and `mapM` allows mapping over
+  published effects, producing a new `Pub` effect with the transformed values.
+
+### Changed
+
+- Moved `Pub` and `Sub` out of `Atelier.Effects.Publishing` to separate modules,
+  `Atelier.Effects.Publishing.Pub` and `Atelier.Effects.Publishing.Sub`
+  respectively.
+
 ## [0.2.0.0] - 2026-06-26
 
 ### Added
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -23,10 +23,10 @@
 
 ## Part of atelier
 
-- [`atelier-prelude`](https://github.com/atelier-hub/tricorder/tree/main/atelier-prelude) — relude-based prelude with Effectful conventions
-- [`atelier-core`](https://github.com/atelier-hub/tricorder/tree/main/atelier-core) — this package
-- [`atelier-db`](https://github.com/atelier-hub/tricorder/tree/main/atelier-db) — relational database effect (Hasql/Rel8)
-- [`atelier-testing`](https://github.com/atelier-hub/tricorder/tree/main/atelier-testing) — database-backed test utilities
+- [`atelier-prelude`](https://github.com/tweag/tricorder/tree/main/atelier-prelude) — relude-based prelude with Effectful conventions
+- [`atelier-core`](https://github.com/tweag/tricorder/tree/main/atelier-core) — this package
+- [`atelier-db`](https://github.com/tweag/tricorder/tree/main/atelier-db) — relational database effect (Hasql/Rel8)
+- [`atelier-testing`](https://github.com/tweag/tricorder/tree/main/atelier-testing) — database-backed test utilities
 
 ## License
 
diff --git a/atelier-core.cabal b/atelier-core.cabal
--- a/atelier-core.cabal
+++ b/atelier-core.cabal
@@ -5,12 +5,12 @@
 -- see: https://github.com/sol/hpack
 
 name:           atelier-core
-version:        0.2.0.0
+version:        0.3.0.0
 synopsis:       Foundational Effectful-based effects and utilities
 description:    Core effects and utilities for effect-based applications, built on Effectful — part of the atelier toolkit.
 category:       Control
-homepage:       https://github.com/atelier-hub/tricorder#readme
-bug-reports:    https://github.com/atelier-hub/tricorder/issues
+homepage:       https://github.com/tweag/tricorder#readme
+bug-reports:    https://github.com/tweag/tricorder/issues
 author:         Christian Georgii
 maintainer:     christian.georgii@tweag.io
 license:        MIT
@@ -27,7 +27,7 @@
 
 source-repository head
   type: git
-  location: https://github.com/atelier-hub/tricorder
+  location: https://github.com/tweag/tricorder
 
 library
   exposed-modules:
@@ -64,6 +64,8 @@
       Atelier.Effects.Process
       Atelier.Effects.Process.Internal
       Atelier.Effects.Publishing
+      Atelier.Effects.Publishing.Pub
+      Atelier.Effects.Publishing.Sub
       Atelier.Effects.Tally
       Atelier.Effects.Timeout
       Atelier.Effects.UUID
@@ -165,6 +167,7 @@
       Unit.Atelier.Effects.FileWatcherSpec
       Unit.Atelier.Effects.IteratorSpec
       Unit.Atelier.Effects.LogSpec
+      Unit.Atelier.Effects.Publishing.PubSpec
       Unit.Atelier.Effects.PublishingSpec
       Unit.Atelier.Effects.TallySpec
       Unit.Atelier.Effects.YieldSpec
@@ -199,7 +202,7 @@
   build-depends:
       aeson ==2.2.*
     , async ==2.2.*
-    , atelier-core ==0.2.*
+    , atelier-core ==0.3.*
     , atelier-prelude ==0.1.*
     , base >=4.18 && <4.23
     , bytestring >=0.11 && <0.13
diff --git a/src/Atelier/Effects/Conc.hs b/src/Atelier/Effects/Conc.hs
--- a/src/Atelier/Effects/Conc.hs
+++ b/src/Atelier/Effects/Conc.hs
@@ -19,6 +19,7 @@
       -- * Scope
     , Scope (..)
     , scoped
+    , restartableFork
     , restartableForkWith
     , restartableForkLoop
 
@@ -87,11 +88,33 @@
 makeEffect ''Conc
 
 
+-- | Forks an action in a loop. Each time @signal@ returns, the current fork is
+-- cancelled, and forked again.
+restartableFork
+    :: (Conc :> es)
+    => Eff es x
+    -- ^ Signal action
+    -> Eff es a
+    -- ^ Cancellable action
+    -> Eff es Void
+restartableFork signal action = forever $ scoped do
+    _ <- fork action
+    signal
+
+
 -- | Forks an action in a loop, with a setup step that runs in the scope before
 -- each fork. Each time @signal@ returns, the current fork is cancelled, and
 -- setup and fork are run again. The setup result is passed to the forked
 -- action, structurally guaranteeing it completes before the fork starts.
-restartableForkWith :: (Conc :> es) => Eff es () -> Eff es r -> (r -> Eff es a) -> Eff es Void
+restartableForkWith
+    :: (Conc :> es)
+    => Eff es x
+    -- ^ Signalling action to block on.
+    -> Eff es r
+    -- ^ Setup action that is run for each iteration
+    -> (r -> Eff es a)
+    -- ^ Action to run with the result of the setup
+    -> Eff es Void
 restartableForkWith signal setup action = forever $ scoped do
     r <- setup
     _ <- fork (action r)
diff --git a/src/Atelier/Effects/Input.hs b/src/Atelier/Effects/Input.hs
--- a/src/Atelier/Effects/Input.hs
+++ b/src/Atelier/Effects/Input.hs
@@ -8,13 +8,19 @@
     , input
     , runInputEff
     , runInputConst
+    , toReader
+    , fromState
     ) where
 
 import Effectful (Effect)
 import Effectful.Dispatch.Dynamic (interpret_)
+import Effectful.Reader.Static (Reader, runReader)
+import Effectful.State.Static.Shared (State)
 import Effectful.TH (makeEffect)
 
+import Effectful.State.Static.Shared qualified as State
 
+
 -- | Request a value from the effect system.
 --
 -- This effect does not carry any
@@ -45,3 +51,14 @@
 -- 'Effectful.Reader.Local', but is available here for testing.
 runInputConst :: i -> Eff (Input i : es) a -> Eff es a
 runInputConst = runInputEff . pure
+
+
+-- | Retrieves the value once, and exposes it as a 'Reader' effect.
+toReader :: (Input i :> es) => Eff (Reader i : es) a -> Eff es a
+toReader act = do
+    x <- input
+    runReader x act
+
+
+fromState :: (State i :> es) => Eff (Input i : es) a -> Eff es a
+fromState = interpret_ \Input -> State.get
diff --git a/src/Atelier/Effects/Internal/Coroutine.hs b/src/Atelier/Effects/Internal/Coroutine.hs
--- a/src/Atelier/Effects/Internal/Coroutine.hs
+++ b/src/Atelier/Effects/Internal/Coroutine.hs
@@ -52,11 +52,11 @@
 
 import Atelier.Effects.Chan (Chan)
 import Atelier.Effects.Conc (Conc)
-import Atelier.Effects.Publishing (Sub)
+import Atelier.Effects.Publishing.Sub (Sub)
 
 import Atelier.Effects.Chan qualified as Chan
 import Atelier.Effects.Conc qualified as Conc
-import Atelier.Effects.Publishing qualified as Sub
+import Atelier.Effects.Publishing.Sub qualified as Sub
 
 
 -- * Yield
diff --git a/src/Atelier/Effects/Iterator.hs b/src/Atelier/Effects/Iterator.hs
--- a/src/Atelier/Effects/Iterator.hs
+++ b/src/Atelier/Effects/Iterator.hs
@@ -17,11 +17,11 @@
 
 import Atelier.Effects.Chan (Chan)
 import Atelier.Effects.Conc (Conc)
-import Atelier.Effects.Publishing (Sub)
+import Atelier.Effects.Publishing.Sub (Sub)
 
 import Atelier.Effects.Chan qualified as Chan
 import Atelier.Effects.Conc qualified as Conc
-import Atelier.Effects.Publishing qualified as Sub
+import Atelier.Effects.Publishing.Sub qualified as Sub
 
 
 -- | A pull-based iterator of (potentially infinite) values.
diff --git a/src/Atelier/Effects/Process.hs b/src/Atelier/Effects/Process.hs
--- a/src/Atelier/Effects/Process.hs
+++ b/src/Atelier/Effects/Process.hs
@@ -28,11 +28,14 @@
 
       -- * Operations
     , readProcessStdout
+    , readProcess
+    , runProcess
     , readProcessSafe
     , withProcessGroup
     , terminateProcessGroup
     , interruptProcessGroup
     , waitExitCode
+    , getExecutablePath
 
       -- * Interpreters
     , runProcessIO
@@ -58,6 +61,7 @@
     , shell
     )
 
+import System.Environment qualified as Env
 import System.Process.Typed qualified as TP
 
 import Atelier.Effects.Process.Internal (RunningProcess (..))
@@ -81,6 +85,15 @@
 data Process :: Effect where
     -- | Run a process to completion, returning its exit code and captured stdout.
     ReadProcessStdout :: ProcessConfig i o e -> Process m (ExitCode, LByteString)
+    -- | Run a process to completion, capturing both stdout and stderr along with
+    -- its exit code.
+    ReadProcess :: ProcessConfig i o e -> Process m (ExitCode, LByteString, LByteString)
+    -- | Run a process to completion, returning only its exit code. Unlike
+    -- 'ReadProcessStdout' this does not capture stdout.
+    RunProcess :: ProcessConfig i o e -> Process m ExitCode
+    -- | The absolute path of the currently running executable, for re-invoking
+    -- this program as a subprocess.
+    GetExecutablePath :: Process m FilePath
     -- | Spawn a process and return its handle. Internal; callers use 'withProcessGroup'.
     StartProcess :: ProcessConfig i o e -> Process m (RunningProcess i o e)
     -- | Terminate the leader and close its streams. Internal; does not reach the
@@ -148,6 +161,9 @@
 runProcessIO :: (IOE :> es) => Eff (Process : es) a -> Eff es a
 runProcessIO = interpret_ \case
     ReadProcessStdout cfg -> liftIO $ TP.readProcessStdout cfg
+    ReadProcess cfg -> liftIO $ TP.readProcess cfg
+    RunProcess cfg -> liftIO $ TP.runProcess cfg
+    GetExecutablePath -> liftIO Env.getExecutablePath
     StartProcess cfg -> liftIO $ RunningProcess <$> TP.startProcess cfg
     StopProcess (RunningProcess p) -> liftIO $ TP.stopProcess p
     WaitExitCode (RunningProcess p) -> liftIO $ TP.waitExitCode p
diff --git a/src/Atelier/Effects/Publishing.hs b/src/Atelier/Effects/Publishing.hs
--- a/src/Atelier/Effects/Publishing.hs
+++ b/src/Atelier/Effects/Publishing.hs
@@ -1,144 +1,23 @@
--- | A typed publish/subscribe effect pair.
---
--- 'Pub' publishes events of a given type; 'Sub' subscribes and delivers each
--- published event to a listener. 'runPubSub' wires the two together over an
--- internal broadcast channel, propagating tracing context from publisher to
--- listener; 'runPubWriter' instead records published events to a 'Writer', for
--- tests.
---
--- Subscriptions are established asynchronously, so a listener you intend to
--- publish to should be started with 'forkListener' or 'forkListener_', which
--- block until the subscription is live and therefore cannot miss early events.
 module Atelier.Effects.Publishing
-    ( Pub
-    , Sub
-    , listen
-    , listen_
-    , listenWith
-    , listenWith_
-    , listenOnce
-    , listenOnce_
-    , forkListener
-    , forkListener_
-    , publish
-    , runPubSub
-    , runPubWriter
+    ( runPubSub
+    , runPubSub_
     )
 where
 
 import Data.Time (UTCTime)
-import Effectful (Effect)
-import Effectful.Concurrent.STM
-    ( Concurrent
-    , atomically
-    , newEmptyTMVar
-    , putTMVar
-    , takeTMVar
-    )
-import Effectful.Dispatch.Dynamic (interpretWith, interpretWith_, interpret_, localSeqUnlift)
-import Effectful.Error.Static (runErrorNoCallStack, throwError)
-import Effectful.TH (makeEffect)
-import Effectful.Writer.Static.Shared (Writer, tell)
-
-import Text.Show qualified as S
+import Effectful.Dispatch.Dynamic (interpret, interpretWith, interpretWith_, interpret_, localSeqUnlift)
 
 import Atelier.Effects.Chan (Chan)
 import Atelier.Effects.Clock (Clock)
-import Atelier.Effects.Conc (Conc, fork_)
 import Atelier.Effects.Monitoring.Tracing (SpanContext, Tracing)
+import Atelier.Effects.Publishing.Pub (Pub (..))
+import Atelier.Effects.Publishing.Sub (Sub (..))
 
 import Atelier.Effects.Chan qualified as Chan
 import Atelier.Effects.Clock qualified as Clock
 import Atelier.Effects.Monitoring.Tracing qualified as Tracing
 
 
--- | Effect for publishing events of type @event@.
-data Pub (event :: Type) :: Effect where
-    -- | Publish an event to all current subscribers.
-    Publish :: event -> Pub event m ()
-
-
--- | Effect for subscribing to events of type @event@.
-data Sub (event :: Type) :: Effect where
-    -- | Subscribe, then run @onSubscribed@ once the subscription is established
-    -- — after the internal channel has been duplicated and before any event is
-    -- delivered — and thereafter deliver every published event to the listener,
-    -- forever. The @onSubscribed@ hook lets a caller synchronize on "subscribed"
-    -- so a concurrently-started publisher cannot race ahead of the subscription
-    -- and have its events missed. Most callers want 'listen' (no hook); a caller
-    -- that forks the listener and then publishes must wait on this hook first.
-    ListenWith :: m () -> (UTCTime -> event -> m ()) -> Sub event m Void
-
-
-makeEffect ''Pub
-makeEffect ''Sub
-
-
--- | Subscribe and deliver every published event to the listener, forever.
--- Defined in terms of 'listenWith' with a no-op subscribed hook.
-listen :: (Sub event :> es) => (UTCTime -> event -> Eff es ()) -> Eff es Void
-listen = listenWith (pure ())
-
-
--- | Like 'listen', but the listener ignores the event timestamp.
-listen_ :: (Sub event :> es) => (event -> Eff es ()) -> Eff es Void
-listen_ listener = listen $ \_timestamp event -> listener event
-
-
--- | Like 'listen_', but runs @onSubscribed@ once the subscription is
--- established and before any event is delivered. See 'ListenWith'.
-listenWith_ :: (Sub event :> es) => Eff es () -> (event -> Eff es ()) -> Eff es Void
-listenWith_ onSubscribed listener = listenWith onSubscribed $ \_timestamp event -> listener event
-
-
--- | Fork a background listener and block until it has actually subscribed,
--- then return. The listener runs until the enclosing 'Conc' scope closes.
---
--- This is the safe way to start a listener you intend to publish to: a plain
--- @'fork_' . 'listen'@ followed by a 'publish' races the subscription (which
--- happens asynchronously in the forked thread) and, under scheduler pressure,
--- can drop early events and wedge the listener forever. 'forkListener' closes
--- that window by waiting on the subscribed hook before returning.
-forkListener
-    :: forall event es
-     . (Conc :> es, Concurrent :> es, Sub event :> es)
-    => (UTCTime -> event -> Eff es ())
-    -> Eff es ()
-forkListener listener = do
-    subscribed <- atomically newEmptyTMVar
-    fork_ $ listenWith (atomically (putTMVar subscribed ())) listener
-    atomically (takeTMVar subscribed)
-
-
--- | Like 'forkListener', but the listener ignores the timestamp.
-forkListener_
-    :: forall event es
-     . (Conc :> es, Concurrent :> es, Sub event :> es)
-    => (event -> Eff es ())
-    -> Eff es ()
-forkListener_ listener = forkListener @event (\_timestamp event -> listener event)
-
-
--- | Wait for a single event and then return said event.
-listenOnce :: forall event es. (Sub event :> es) => Eff es (UTCTime, event)
-listenOnce = do
-    res <- runErrorNoCallStack
-        $ listen
-        $ \timestamp event -> throwError $ OnceEx (timestamp, event)
-    case res of
-        Left (OnceEx x) -> pure x
-        Right v -> absurd v
-
-
--- | Same as 'listenOnce', but discards the timestamp.
-listenOnce_ :: (Sub event :> es) => Eff es event
-listenOnce_ = snd <$> listenOnce
-
-
-data OnceEx ev = OnceEx ev
-instance Show (OnceEx ev) where show _ = "OnceEx"
-
-
 -- | Internal wrapper for events with trace context
 data TracedEvent event = TracedEvent
     { event :: event
@@ -147,8 +26,9 @@
     }
 
 
--- | Runs Pub and Sub effects with an internal channel for a specific event type.
--- Automatically captures span context from the publisher and creates linked spans in listeners.
+-- | Runs 'Pub' and 'Sub' effects with an internal channel for a specific event
+-- type. Automatically captures span context from the publisher and creates
+-- linked spans in listeners.
 runPubSub
     :: forall event es a
      . ( Chan :> es
@@ -177,9 +57,25 @@
     handleSub . handlePub $ action
 
 
--- | Handler that uses a provided Writer effect instead of actually publishing.
--- Useful for testing and inspecting what events were published.
-runPubWriter :: forall event es a. (Writer [event] :> es) => Eff (Pub event : es) a -> Eff es a
-runPubWriter =
-    interpret_ \case
-        Publish event -> tell [event]
+-- | Runs 'Pub' and 'Sub' effects with an internal channel for a specific event
+-- type.
+runPubSub_
+    :: forall event es a
+     . (Chan :> es, Clock :> es)
+    => Eff (Pub event : Sub event : es) a -> Eff es a
+runPubSub_ action = do
+    (inChan, _) <- Chan.newChan @(UTCTime, event)
+    let handlePub = interpret_ \case
+            Publish event -> do
+                timestamp <- Clock.currentTime
+                Chan.writeChan inChan (timestamp, event)
+
+        handleSub = interpret \env -> \case
+            ListenWith onSubscribed listener -> localSeqUnlift env \unlift -> do
+                chan <- Chan.dupChan inChan
+                unlift onSubscribed
+                forever do
+                    (timestamp, event) <- Chan.readChan chan
+                    unlift $ listener timestamp event
+
+    handleSub . handlePub $ action
diff --git a/src/Atelier/Effects/Publishing/Pub.hs b/src/Atelier/Effects/Publishing/Pub.hs
new file mode 100644
--- /dev/null
+++ b/src/Atelier/Effects/Publishing/Pub.hs
@@ -0,0 +1,57 @@
+module Atelier.Effects.Publishing.Pub
+    ( -- * Effect
+      Pub (..)
+    , publish
+
+      -- * Interpeters
+    , runNoOp
+    , toWriter
+    , map
+    , mapM
+    , consume
+    ) where
+
+import Effectful (Effect)
+import Effectful.Dispatch.Dynamic (interpret_)
+import Effectful.TH (makeEffect)
+import Effectful.Writer.Static.Shared (Writer, tell)
+import Prelude hiding (map, mapM)
+
+
+-- | Effect for publishing events of type @event@.
+data Pub (event :: Type) :: Effect where
+    -- | Publish an event to all current subscribers.
+    Publish :: event -> Pub event m ()
+
+
+makeEffect ''Pub
+
+
+runNoOp :: forall event es a. Eff (Pub event : es) a -> Eff es a
+runNoOp = interpret_ \(Publish _) -> pure ()
+
+
+-- | Handler that uses a provided Writer effect instead of actually publishing.
+-- Useful for testing and inspecting what events were published.
+toWriter :: forall event es a. (Writer [event] :> es) => Eff (Pub event : es) a -> Eff es a
+toWriter =
+    interpret_ \case
+        Publish event -> tell [event]
+
+
+-- | Convert published events of one type into another, utilizing an existing
+-- effect in the effect stack.
+map :: forall e1 e2 es a. (Pub e2 :> es) => (e1 -> e2) -> Eff (Pub e1 : es) a -> Eff es a
+map f = interpret_ \(Publish event) -> publish $ f event
+
+
+-- | Convert published events of one type into another with an effectful
+-- transformation, utilizing an existing effect int he effect stack.
+mapM :: forall e1 e2 es a. (Pub e2 :> es) => (e1 -> Eff es e2) -> Eff (Pub e1 : es) a -> Eff es a
+mapM f = interpret_ \(Publish event) -> f event >>= publish
+
+
+-- | Perform an effectful action for each published event.
+consume :: (event -> Eff es ()) -> Eff (Pub event : es) a -> Eff es a
+consume handler = interpret_ \case
+    Publish event -> handler event
diff --git a/src/Atelier/Effects/Publishing/Sub.hs b/src/Atelier/Effects/Publishing/Sub.hs
new file mode 100644
--- /dev/null
+++ b/src/Atelier/Effects/Publishing/Sub.hs
@@ -0,0 +1,152 @@
+module Atelier.Effects.Publishing.Sub
+    ( Sub (..)
+    , listen
+    , listen_
+    , listenWith
+    , listenWith_
+    , listenOnce
+    , listenOnce_
+    , listenUntil
+    , listenUntil_
+    , listenUntilM
+    , listenUntilM_
+    , forkListener
+    , forkListener_
+    ) where
+
+import Data.Time (UTCTime)
+import Effectful (Effect, inject)
+import Effectful.Concurrent.STM
+    ( Concurrent
+    , atomically
+    , newEmptyTMVar
+    , putTMVar
+    , takeTMVar
+    )
+import Effectful.Error.Static (runErrorNoCallStack, throwError)
+import Effectful.TH (makeEffect)
+
+import Text.Show qualified as S
+
+import Atelier.Effects.Conc (Conc, fork_)
+
+
+-- | Effect for subscribing to events of type @event@.
+data Sub (event :: Type) :: Effect where
+    -- | Subscribe, then run @onSubscribed@ once the subscription is established
+    -- — after the internal channel has been duplicated and before any event is
+    -- delivered — and thereafter deliver every published event to the listener,
+    -- forever. The @onSubscribed@ hook lets a caller synchronize on "subscribed"
+    -- so a concurrently-started publisher cannot race ahead of the subscription
+    -- and have its events missed. Most callers want 'listen' (no hook); a caller
+    -- that forks the listener and then publishes must wait on this hook first.
+    ListenWith
+        :: m ()
+        -- ^ @onSubscribed@ hook to better synchronize on "subscribed".
+        -> (UTCTime -> event -> m ())
+        -- ^ Listener function to react to events.
+        -> Sub event m Void
+
+
+makeEffect ''Sub
+
+
+-- | Subscribe and deliver every published event to the listener, forever.
+-- Defined in terms of 'listenWith' with a no-op subscribed hook.
+listen :: (Sub event :> es) => (UTCTime -> event -> Eff es ()) -> Eff es Void
+listen = listenWith (pure ())
+
+
+-- | Like 'listen', but the listener ignores the event timestamp.
+listen_ :: (Sub event :> es) => (event -> Eff es ()) -> Eff es Void
+listen_ listener = listen $ \_timestamp event -> listener event
+
+
+-- | Like 'listen_', but runs @onSubscribed@ once the subscription is
+-- established and before any event is delivered. See 'ListenWith'.
+listenWith_ :: (Sub event :> es) => Eff es () -> (event -> Eff es ()) -> Eff es Void
+listenWith_ onSubscribed listener = listenWith onSubscribed $ \_timestamp event -> listener event
+
+
+-- | Listens until the passed function returns @Just a@, returning said @a@
+-- with a timestamp.
+listenUntil :: (Sub event :> es) => (UTCTime -> event -> Maybe a) -> Eff es (UTCTime, a)
+listenUntil f = do
+    res <- runErrorNoCallStack
+        $ listen
+        $ \timestamp event -> whenJust (f timestamp event) \a -> do
+            throwError $ OnceEx (timestamp, a)
+    case res of
+        Left (OnceEx x) -> pure x
+        Right v -> absurd v
+
+
+listenUntil_ :: (Sub event :> es) => (event -> Maybe a) -> Eff es a
+listenUntil_ f = snd <$> listenUntil (\_ -> f)
+
+
+-- | Listens until the passed event handler returns @Just a@, returning said
+-- @a@ with a timestamp.
+listenUntilM :: (Sub event :> es) => (UTCTime -> event -> Eff es (Maybe a)) -> Eff es (UTCTime, a)
+listenUntilM f = do
+    res <- runErrorNoCallStack
+        $ listen
+        $ \timestamp event -> whenJustM (inject $ f timestamp event) \a -> do
+            throwError $ OnceEx (timestamp, a)
+    case res of
+        Left (OnceEx x) -> pure x
+        Right v -> absurd v
+
+
+-- | Listens until the passed event handler returns @Just a@, returning said
+-- @a@.
+listenUntilM_ :: (Sub event :> es) => (event -> Eff es (Maybe a)) -> Eff es a
+listenUntilM_ f = snd <$> listenUntilM (\_ -> f)
+
+
+-- | Fork a background listener and block until it has actually subscribed,
+-- then return. The listener runs until the enclosing 'Conc' scope closes.
+--
+-- This is the safe way to start a listener you intend to publish to: a plain
+-- @'fork_' . 'listen'@ followed by a 'publish' races the subscription (which
+-- happens asynchronously in the forked thread) and, under scheduler pressure,
+-- can drop early events and wedge the listener forever. 'forkListener' closes
+-- that window by waiting on the subscribed hook before returning.
+forkListener
+    :: forall event es
+     . (Conc :> es, Concurrent :> es, Sub event :> es)
+    => (UTCTime -> event -> Eff es ())
+    -> Eff es ()
+forkListener listener = do
+    subscribed <- atomically newEmptyTMVar
+    fork_ $ listenWith (atomically (putTMVar subscribed ())) listener
+    atomically (takeTMVar subscribed)
+
+
+-- | Like 'forkListener', but the listener ignores the timestamp.
+forkListener_
+    :: forall event es
+     . (Conc :> es, Concurrent :> es, Sub event :> es)
+    => (event -> Eff es ())
+    -> Eff es ()
+forkListener_ listener = forkListener @event (\_timestamp event -> listener event)
+
+
+-- | Wait for a single event and then return said event.
+listenOnce :: forall event es. (Sub event :> es) => Eff es (UTCTime, event)
+listenOnce = do
+    res <- runErrorNoCallStack
+        $ listen
+        $ \timestamp event -> throwError $ OnceEx (timestamp, event)
+    case res of
+        Left (OnceEx x) -> pure x
+        Right v -> absurd v
+
+
+-- | Same as 'listenOnce', but discards the timestamp.
+listenOnce_ :: (Sub event :> es) => Eff es event
+listenOnce_ = snd <$> listenOnce
+
+
+data OnceEx ev = OnceEx ev
+instance Show (OnceEx ev) where show _ = "OnceEx"
diff --git a/test/Unit/Atelier/Effects/Conc/TeardownStressSpec.hs b/test/Unit/Atelier/Effects/Conc/TeardownStressSpec.hs
--- a/test/Unit/Atelier/Effects/Conc/TeardownStressSpec.hs
+++ b/test/Unit/Atelier/Effects/Conc/TeardownStressSpec.hs
@@ -28,9 +28,12 @@
 import Atelier.Effects.Clock (Clock, runClock)
 import Atelier.Effects.Conc (Conc, fork, fork_, runConc, scoped)
 import Atelier.Effects.Monitoring.Tracing (Tracing, runTracingNoOp)
-import Atelier.Effects.Publishing (Pub, Sub, publish, runPubSub)
+import Atelier.Effects.Publishing (runPubSub)
+import Atelier.Effects.Publishing.Pub (Pub)
+import Atelier.Effects.Publishing.Sub (Sub)
 
 import Atelier.Effects.Iterator qualified as Iter
+import Atelier.Effects.Publishing.Pub qualified as Pub
 
 
 spec_ConcTeardownStress :: Spec
@@ -78,7 +81,7 @@
                     $ Iter.fromEvents @Int \iter -> do
                         _ <- fork do
                             liftIO (threadDelay publishDelayUs)
-                            traverse_ publish [1, 2, 3 :: Int]
+                            traverse_ Pub.publish [1, 2, 3 :: Int]
                         _ <- replicateM 3 (Iter.next iter)
                         pure ()
             completed `shouldBe` Just ()
diff --git a/test/Unit/Atelier/Effects/IteratorSpec.hs b/test/Unit/Atelier/Effects/IteratorSpec.hs
--- a/test/Unit/Atelier/Effects/IteratorSpec.hs
+++ b/test/Unit/Atelier/Effects/IteratorSpec.hs
@@ -9,9 +9,12 @@
 import Atelier.Effects.Clock (Clock, runClock)
 import Atelier.Effects.Conc (Conc, fork, runConc)
 import Atelier.Effects.Monitoring.Tracing (Tracing, runTracingNoOp)
-import Atelier.Effects.Publishing (Pub, Sub, publish, runPubSub)
+import Atelier.Effects.Publishing (runPubSub)
+import Atelier.Effects.Publishing.Pub (Pub)
+import Atelier.Effects.Publishing.Sub (Sub)
 
 import Atelier.Effects.Iterator qualified as Iter
+import Atelier.Effects.Publishing.Pub qualified as Pub
 
 
 spec_Iterator :: Spec
@@ -28,7 +31,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    publish (42 :: Int)
+                    Pub.publish (42 :: Int)
                 Iter.next iter
         result `shouldBe` 42
 
@@ -37,7 +40,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    traverse_ publish [1, 2, 3]
+                    traverse_ Pub.publish [1, 2, 3]
                 replicateM 3 (Iter.next iter)
         result `shouldBe` [1, 2, 3]
 
@@ -46,7 +49,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    traverse_ publish [1, 2, 3]
+                    traverse_ Pub.publish [1, 2, 3]
                 liftIO $ threadDelay 5_000
                 replicateM 3 (Iter.next iter)
         result `shouldBe` [1, 2, 3]
@@ -59,7 +62,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    traverse_ publish [1 .. 4]
+                    traverse_ Pub.publish [1 .. 4]
                 Iter.next (Iter.filter even iter)
         result `shouldBe` 2
 
@@ -68,7 +71,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    traverse_ publish [1 .. 6]
+                    traverse_ Pub.publish [1 .. 6]
                 replicateM 3 (Iter.next (Iter.filter even iter))
         result `shouldBe` [2, 4, 6]
 
@@ -80,7 +83,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    traverse_ publish [0, 0, 1]
+                    traverse_ Pub.publish [0, 0, 1]
                 Iter.next (Iter.changes 0 iter)
         result `shouldBe` 1
 
@@ -89,7 +92,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    traverse_ publish [1, 2, 3]
+                    traverse_ Pub.publish [1, 2, 3]
                 replicateM 3 (Iter.next (Iter.changes 0 iter))
         result `shouldBe` [1, 2, 3]
 
@@ -98,7 +101,7 @@
             Iter.fromEvents @Int \iter -> do
                 _ <- fork do
                     liftIO $ threadDelay 10
-                    traverse_ publish [0, 1, 0, 2, 0, 3]
+                    traverse_ Pub.publish [0, 1, 0, 2, 0, 3]
                 replicateM 3 (Iter.next (Iter.changes 0 iter))
         result `shouldBe` [1, 2, 3]
 
diff --git a/test/Unit/Atelier/Effects/Publishing/PubSpec.hs b/test/Unit/Atelier/Effects/Publishing/PubSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Unit/Atelier/Effects/Publishing/PubSpec.hs
@@ -0,0 +1,53 @@
+module Unit.Atelier.Effects.Publishing.PubSpec (spec_Pub) where
+
+import Effectful (runPureEff)
+import Effectful.Writer.Static.Shared (execWriter)
+import Test.Hspec (Spec, context, describe, it, shouldMatchList)
+
+import Atelier.Effects.Publishing.Pub qualified as Pub
+
+
+data TestEvent = TestEvent Text
+    deriving stock (Eq, Show)
+
+
+spec_Pub :: Spec
+spec_Pub = do
+    describe "toWriter" do
+        context "no events published" do
+            it "doesn't record events" do
+                let events =
+                        runPureEff . execWriter . Pub.toWriter @TestEvent
+                            $ pure ()
+
+                events `shouldMatchList` []
+
+        context "events published" do
+            it "records events" do
+                let events =
+                        runPureEff . execWriter . Pub.toWriter @TestEvent $ do
+                            Pub.publish $ TestEvent "payload"
+                            pure ()
+
+                events `shouldMatchList` [TestEvent "payload"]
+
+    describe "map" do
+        it "maps over one event" do
+            let events =
+                    runPureEff
+                        . execWriter
+                        . Pub.toWriter @Text
+                        . Pub.map show
+                        $ Pub.publish @Int 1
+
+            events `shouldMatchList` ["1"]
+
+        it "maps over many events" do
+            let events =
+                    runPureEff
+                        . execWriter
+                        . Pub.toWriter @Text
+                        . Pub.map show
+                        $ traverse Pub.publish [1 .. 10 :: Int]
+
+            events `shouldMatchList` ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]
diff --git a/test/Unit/Atelier/Effects/PublishingSpec.hs b/test/Unit/Atelier/Effects/PublishingSpec.hs
--- a/test/Unit/Atelier/Effects/PublishingSpec.hs
+++ b/test/Unit/Atelier/Effects/PublishingSpec.hs
@@ -1,53 +1,36 @@
-module Unit.Atelier.Effects.PublishingSpec (spec_Pub) where
+module Unit.Atelier.Effects.PublishingSpec (spec_Publishing) where
 
 import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
 import Data.Time (UTCTime, getCurrentTime)
-import Effectful (IOE, runEff, runPureEff)
+import Effectful (IOE, runEff)
 import Effectful.Concurrent (Concurrent, runConcurrent)
-import Effectful.Writer.Static.Shared (runWriter)
-import Test.Hspec (Spec, context, describe, expectationFailure, it, shouldBe)
+import Test.Hspec (Spec, describe, it, shouldBe)
 
 import Atelier.Effects.Chan (Chan, runChan)
 import Atelier.Effects.Clock (Clock, runClock, runClockConst)
 import Atelier.Effects.Conc (Conc, runConc)
 import Atelier.Effects.Monitoring.Tracing (Tracing, runTracingNoOp)
-import Atelier.Effects.Publishing (Pub, Sub, forkListener, forkListener_, publish, runPubSub, runPubWriter)
+import Atelier.Effects.Publishing (runPubSub)
+import Atelier.Effects.Publishing.Pub (Pub)
+import Atelier.Effects.Publishing.Sub (Sub)
 
+import Atelier.Effects.Publishing.Pub qualified as Pub
+import Atelier.Effects.Publishing.Sub qualified as Sub
 
+
 data TestEvent = TestEvent Text
     deriving stock (Eq, Show)
 
 
-spec_Pub :: Spec
-spec_Pub = do
-    describe "Writer implementation" $ do
-        context "no events published" do
-            it "doesn't record events" $ do
-                let ((), events) = runPureEff . runWriter . runPubWriter @TestEvent $ do
-                        pure ()
-
-                length events `shouldBe` 0
-
-        context "events published" do
-            it "records events" $ do
-                let ((), events) = runPureEff . runWriter . runPubWriter @TestEvent $ do
-                        publish $ TestEvent "payload"
-                        pure ()
-
-                length events `shouldBe` 1
-                case events of
-                    [TestEvent payload] ->
-                        payload `shouldBe` "payload"
-                    xs ->
-                        expectationFailure $ "Expected 1 TestEvent event, got: " <> show (length xs)
-
-    describe "PubSub implementation" do
+spec_Publishing :: Spec
+spec_Publishing = do
+    describe "runPubSub" do
         it "listener receives a published event" do
             result <- runPubSubTest $ do
                 received <- liftIO newEmptyMVar
-                forkListener_ @TestEvent \event ->
+                Sub.forkListener_ @TestEvent \event ->
                     liftIO $ putMVar received event
-                publish (TestEvent "hello")
+                Pub.publish (TestEvent "hello")
                 liftIO $ takeMVar received
             result `shouldBe` TestEvent "hello"
 
@@ -55,9 +38,9 @@
             t0 <- getCurrentTime
             result <- runPubSubTestWithClock t0 $ do
                 received <- liftIO newEmptyMVar
-                forkListener @TestEvent \ts _event ->
+                Sub.forkListener @TestEvent \ts _event ->
                     liftIO $ putMVar received ts
-                publish (TestEvent "hello")
+                Pub.publish (TestEvent "hello")
                 liftIO $ takeMVar received
             result `shouldBe` t0
 
@@ -65,9 +48,9 @@
             result <- runPubSubTest $ do
                 recv1 <- liftIO newEmptyMVar
                 recv2 <- liftIO newEmptyMVar
-                forkListener_ @TestEvent \event -> liftIO $ putMVar recv1 event
-                forkListener_ @TestEvent \event -> liftIO $ putMVar recv2 event
-                publish (TestEvent "hello")
+                Sub.forkListener_ @TestEvent \event -> liftIO $ putMVar recv1 event
+                Sub.forkListener_ @TestEvent \event -> liftIO $ putMVar recv2 event
+                Pub.publish (TestEvent "hello")
                 e1 <- liftIO $ takeMVar recv1
                 e2 <- liftIO $ takeMVar recv2
                 pure (e1, e2)