@@ -309,9 +309,9 @@ def set_context_option(key: str, value: Any) -> None:
     """
     Set a capture option for every event captured within the current context.
 
-    Context options fill options an event leaves unset, after ``before_send``
-    runs. They override ``super_options``. An event's own ``options`` and
-    ``before_send`` changes override them.
+    Context options fill options an event leaves unset, before ``before_send``
+    runs, so the hook sees them. They override ``super_options``. An event's
+    own ``options`` override them, and ``before_send`` can change them.
 
     Args:
         key: The option name, such as ``"process_person_profile"``
@@ -385,12 +385,14 @@ def get_context_options() -> Dict[str, Any]:
         requests after network, transport, or timeout failures. Defaults to 1.
         Set to 0 to disable retries.
     super_properties: Properties for every captured event. They fill only
-        keys the event leaves unset, after ``before_send`` runs. An event's own
-        properties, ``before_send`` changes and context tags override them.
+        keys the event leaves unset, before ``before_send`` runs. ``$set``,
+        ``$set_once``, ``$groups`` and ``$group_set`` fill one level deep. An
+        event's own properties and context tags override them, and
+        ``before_send`` can change or remove them.
     super_options: Capture options for every captured event, such as
         ``{"cookieless_mode": True}``. They fill only options the event leaves
-        unset, after ``before_send`` runs. An event's own ``options``,
-        ``before_send`` changes and context options override them. They also
+        unset, before ``before_send`` runs. An event's own ``options`` and
+        context options override them, and ``before_send`` can change them. They also
         win over an event's legacy property for the same key, such as
         ``$cookieless_mode``, so pass per-event overrides of that key as
         ``options``.
@@ -417,9 +419,9 @@ def get_context_options() -> Dict[str, Any]:
     project_root: Root path used to determine in-app exception stack frames.
     privacy_mode: Capture AI usage metadata without prompt inputs or outputs.
     before_send: Optional callback that can modify or drop events before upload.
-        Return ``None`` to drop an event. It does not see context tags, context
-        options, ``super_properties`` or ``super_options``. They fill in after
-        it runs.
+        Return ``None`` to drop an event. Context tags, context options,
+        ``super_properties`` and ``super_options`` fill in before it runs, so
+        it can change or remove them.
     enable_local_evaluation: Whether to poll feature flag definitions for local
         evaluation when a personal API key is configured.
     flag_definition_cache_provider: Optional external cache provider for sharing
 
@@ -453,10 +453,8 @@ def _prepare_event(
     async def _process_event(
         self, msg: dict[str, Any], defaults: Optional[_EventDefaults]
     ) -> Optional[dict[str, Any]]:
-        processed = await self._run_before_send(msg)
-        if processed is not None:
-            _fill_event_defaults(processed, defaults)
-        return processed
+        _fill_event_defaults(msg, defaults)
+        return await self._run_before_send(msg)
 
     async def _run_before_send(self, msg: dict[str, Any]) -> Optional[dict[str, Any]]:
         if self.before_send is None:
 
@@ -40,6 +40,9 @@
 # Top-level legacy keys relocated into properties (v1 has no top-level form).
 _RELOCATE_TO_PROPERTIES = ("$set", "$set_once")
 
+# Properties that defaults fill one level deep when both values are dicts.
+_NESTED_FILL_PROPERTIES = frozenset({"$set", "$set_once", "$groups", "$group_set"})
+
 # Properties dropped from v1 events (server injects them from PostHog-Sdk-Info).
 _STRIP_FROM_PROPERTIES = ("$lib", "$lib_version")
 
@@ -92,8 +95,8 @@ def _event_options(value: Any) -> dict[str, Any]:
 class _EventDefaults:
     """Context, global and SDK-derived values for one event, highest layer first.
 
-    They fill in after ``before_send``, so the hook never sees them, and the
-    event's own values and the hook's changes always win.
+    They fill in before ``before_send``, so the hook sees them and can change
+    or remove them. The event's own values win over every default.
     """
 
     property_layers: tuple[Mapping[str, Any], ...] = ()
@@ -132,7 +135,9 @@ def _fill_event_defaults(
     """Fill the keys an event left unset from ``defaults``, layer by layer.
 
     A property is unset only when its key is missing. An option is unset when
-    it is missing or ``None``, the same rule hoisting uses.
+    it is missing or ``None``, the same rule hoisting uses. ``$set``,
+    ``$set_once``, ``$groups`` and ``$group_set`` fill one level deep when the
+    event's value and the default are both dicts.
     """
     if defaults is None:
         return
@@ -143,9 +148,18 @@ def _fill_event_defaults(
         msg["properties"] = properties
     for property_layer in defaults.property_layers:
         for key, value in property_layer.items():
-            if key in properties or (allowlist is not None and key not in allowlist):
+            if allowlist is not None and key not in allowlist:
+                continue
+            if key not in properties:
+                properties[key] = _clean(value)
                 continue
-            properties[key] = _clean(value)
+            existing = properties[key]
+            if (
+                key in _NESTED_FILL_PROPERTIES
+                and isinstance(existing, dict)
+                and isinstance(value, Mapping)
+            ):
+                properties[key] = {**_clean(dict(value)), **existing}
     options = _event_options(msg.get("options"))
     for option_layer in defaults.option_layers:
         for key, value in option_layer.items():
@@ -177,16 +191,16 @@ def _to_v1_event(msg: dict) -> dict:
     properties = dict(msg.get("properties") or {})
 
     # Relocate top-level $set/$set_once into properties; v1 has no top-level
-    # form. On the unusual collision where properties already carries the key,
-    # the properties value wins.
+    # form. The top-level value comes from the set() or set_once() call, so it
+    # wins key by key over a $set in properties, as ingestion merges them.
     for key in _RELOCATE_TO_PROPERTIES:
         top_val = msg.get(key)
         if top_val is None:
             continue
         existing = properties.get(key)
         if isinstance(top_val, dict) and isinstance(existing, dict):
-            properties[key] = {**top_val, **existing}
-        elif key not in properties:
+            properties[key] = {**existing, **top_val}
+        else:
             properties[key] = top_val
 
     for key in _STRIP_FROM_PROPERTIES:
 
@@ -823,14 +823,15 @@ def __init__(
                 requests after network, transport, or timeout failures. Defaults
                 to 1. Set to 0 to disable retries.
             super_properties: Properties for every captured event. They fill
-                only keys the event leaves unset, after ``before_send`` runs.
-                An event's own properties, ``before_send`` changes and context
-                tags override them.
+                only keys the event leaves unset, before ``before_send`` runs.
+                ``$set``, ``$set_once``, ``$groups`` and ``$group_set`` fill
+                one level deep. An event's own properties and context tags
+                override them, and ``before_send`` can change or remove them.
             super_options: Capture options for every captured event, such as
                 ``{"cookieless_mode": True}``. They fill only options the event
-                leaves unset, after ``before_send`` runs. An event's own
-                ``options``, ``before_send`` changes and context options
-                override them. They also win over an event's legacy property
+                leaves unset, before ``before_send`` runs. An event's own
+                ``options`` and context options override them, and
+                ``before_send`` can change them. They also win over an event's legacy property
                 for the same key, such as ``$cookieless_mode``, so pass
                 per-event overrides of that key as ``options``.
             enable_exception_autocapture: Automatically capture uncaught
@@ -847,9 +848,9 @@ def __init__(
                 through unredacted. ``privacy_mode`` always wins. Defaults to
                 False.
             before_send: Optional callback that can modify or drop events before
-                upload. Return ``None`` to drop an event. It does not see
-                context tags, context options, ``super_properties`` or
-                ``super_options``. They fill in after it runs.
+                upload. Return ``None`` to drop an event. Context tags,
+                context options, ``super_properties`` and ``super_options``
+                fill in before it runs, so it can change or remove them.
             flag_fallback_cache_url: Optional feature flag fallback cache URL,
                 such as ``memory://local/?ttl=300&size=10000`` or a Redis URL.
             enable_local_evaluation: Whether to poll feature flag definitions for
@@ -1357,9 +1358,10 @@ def set_context_option(self, key: str, value: Any) -> None:
         """
         Set a capture option for every event captured within the current context.
 
-        Context options fill options an event leaves unset, after
-        ``before_send`` runs. They override ``super_options``. An event's own
-        ``options`` and ``before_send`` changes override them.
+        Context options fill options an event leaves unset, before
+        ``before_send`` runs, so the hook sees them. They override
+        ``super_options``. An event's own ``options`` override them, and
+        ``before_send`` can change them.
 
         Args:
             key: The option name, such as ``"process_person_profile"``.
@@ -1703,7 +1705,7 @@ def capture(
             options: Capture options for this event, such as
                 ``{"process_person_profile": False}``. Sent as given, for
                 PostHog to validate. They override context options and
-                ``super_options``, which fill in after ``before_send`` runs.
+                ``super_options``, which fill in before ``before_send`` runs.
                 An option set at any layer wins over its
                 legacy ``$`` property, such as ``$process_person_profile``,
                 set at any layer. A ``None`` option counts as unset.
@@ -2540,10 +2542,22 @@ def _enqueue(
         if self.is_server:
             msg["properties"]["$is_server"] = True
 
-        # Applied after every enrichment step (system context, $lib/$lib_version)
-        # and again to the defaults filled in after before_send, so the final
-        # event shape is exactly the allowlist regardless of where a property
-        # came from.
+        _fill_event_defaults(
+            msg,
+            _build_event_defaults(
+                super_properties=self.super_properties,
+                super_options=self.super_options,
+                release_id=self._release_id,
+                context_properties=context_properties,
+                context_options=context_options,
+                derived_options=derived_options,
+                property_allowlist=property_allowlist,
+            ),
+        )
+
+        # Applied after every enrichment step (system context, $lib/$lib_version,
+        # filled defaults), so the final event shape is exactly the allowlist
+        # regardless of where a property came from.
         if property_allowlist is not None:
             msg["properties"] = {
                 k: v for k, v in msg["properties"].items() if k in property_allowlist
@@ -2566,19 +2580,6 @@ def _enqueue(
                 self.log.exception(f"Error in before_send callback: {e}")
                 return None
 
-        _fill_event_defaults(
-            msg,
-            _build_event_defaults(
-                super_properties=self.super_properties,
-                super_options=self.super_options,
-                release_id=self._release_id,
-                context_properties=context_properties,
-                context_options=context_options,
-                derived_options=derived_options,
-                property_allowlist=property_allowlist,
-            ),
-        )
-
         # Re-normalized after before_send, which may have replaced or removed
         # msg["uuid"], so the returned uuid always matches the wire event.
         self._normalize_event_uuid(msg)
 
@@ -303,10 +303,10 @@ def set_context_option(key: str, value: Any) -> None:
     """
     Set a capture option for every event captured within the current context.
 
-    Context options fill options an event leaves unset, after ``before_send``
-    runs. They override the client's ``super_options``. An event's own
-    ``options`` and ``before_send`` changes override them. Child contexts
-    inherit them unless they are fresh.
+    Context options fill options an event leaves unset, before ``before_send``
+    runs, so the hook sees them. They override the client's ``super_options``.
+    An event's own ``options`` override them, and ``before_send`` can change
+    them. Child contexts inherit them unless they are fresh.
 
     Args:
         key: The option name, such as ``"process_person_profile"``
 
@@ -204,15 +204,14 @@ def test_top_level_set_relocated_into_properties(self, _name, key) -> None:
         self.assertNotIn(key, event)  # not a top-level v1 field
 
     def test_top_level_set_merges_with_existing_properties_set(self) -> None:
-        # properties wins on key collision.
         msg = _legacy_msg(
             properties={"$set": {"a": "from_props", "b": "props_only"}},
             **{"$set": {"a": "from_top", "c": "top_only"}},
         )
         event = _to_v1_event(msg)
         self.assertEqual(
             event["properties"]["$set"],
-            {"a": "from_props", "b": "props_only", "c": "top_only"},
+            {"a": "from_top", "b": "props_only", "c": "top_only"},
         )
 
     def test_groups_left_in_properties(self) -> None:
 
@@ -252,7 +252,10 @@ async def test_async_event_and_context_properties_beat_super_properties():
 def _recording_hook(seen):
     def hook(msg):
         seen.append((dict(msg["properties"]), dict(msg["options"])))
-        properties = {**msg["properties"], "shared": "hook"}
+        properties = {
+            **{k: v for k, v in msg["properties"].items() if k != "only_super"},
+            "shared": "hook",
+        }
         return {**msg, "properties": properties, "options": {"cookieless_mode": False}}
 
     return hook
@@ -280,40 +283,71 @@ async def immediate():
     return call
 
 
-def _assert_defaults_fill_after_hook(seen, events):
+def _assert_hook_sees_defaults_and_has_final_say(seen, events):
     hook_properties, hook_options = seen[0]
-    assert not {"shared", "only_super", "from_context"} & hook_properties.keys()
-    assert hook_options == {}
-    properties = events[0]["properties"]
     assert (
-        properties["shared"],
-        properties["only_super"],
-        properties["from_context"],
-    ) == ("hook", 1, "context")
-    assert events[0]["options"] == {
-        "cookieless_mode": False,
-        "product_tour_id": "context-tour",
-    }
+        hook_properties["shared"],
+        hook_properties["only_super"],
+        hook_properties["from_context"],
+    ) == ("super", 1, "context")
+    assert hook_options == {"cookieless_mode": True, "product_tour_id": "context-tour"}
+    properties = events[0]["properties"]
+    assert (properties["shared"], properties["from_context"]) == ("hook", "context")
+    assert "only_super" not in properties
+    assert events[0]["options"] == {"cookieless_mode": False}
 
 
-def test_sync_defaults_fill_after_before_send():
+def test_sync_defaults_fill_before_before_send():
     seen: list = []
     events = _sync_wire_events(
         _context_capture("capture"),
         before_send=_recording_hook(seen),
         **DEFAULTS_CONFIG,
     )
-    _assert_defaults_fill_after_hook(seen, events)
+    _assert_hook_sees_defaults_and_has_final_say(seen, events)
 
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("method", ["capture", "capture_immediate"])
-async def test_async_defaults_fill_after_before_send(method):
+async def test_async_defaults_fill_before_before_send(method):
     seen: list = []
     events = await _async_wire_events(
         _context_capture(method), before_send=_recording_hook(seen), **DEFAULTS_CONFIG
     )
-    _assert_defaults_fill_after_hook(seen, events)
+    _assert_hook_sees_defaults_and_has_final_say(seen, events)
+
+
+NESTED_SUPER_PROPERTIES = {
+    "$set": {"plan": "free", "source": "super"},
+    "$groups": {"company": "acme", "team": "core"},
+    "$unset": ["stale"],
+}
+
+
+def test_nested_properties_fill_one_level_deep():
+    events = _sync_wire_events(
+        lambda c: c.capture(
+            "e",
+            distinct_id="u",
+            properties={"$set": {"plan": "pro"}, "$unset": ["other"]},
+            groups={"company": "posthog"},
+        ),
+        super_properties=NESTED_SUPER_PROPERTIES,
+    )
+    properties = events[0]["properties"]
+    assert properties["$set"] == {"plan": "pro", "source": "super"}
+    assert properties["$groups"] == {"company": "posthog", "team": "core"}
+    assert properties["$unset"] == ["other"]
+
+
+@pytest.mark.parametrize("method", ["set", "set_once"])
+def test_set_call_wins_over_super_person_properties(method):
+    key = f"${method}"
+    events = _sync_wire_events(
+        lambda c: getattr(c, method)(distinct_id="u", properties={"plan": "pro"}),
+        super_properties={key: {"plan": "free", "source": "super"}},
+    )
+    assert events[0]["properties"][key] == {"plan": "pro", "source": "super"}
 
 
 def test_late_options_replace_and_remove_legacy_properties():