feat: add stop_tags() to prevent tag propagation from sub-configs

Introduces ConfigWrapper as a unified value wrapper with properties
(tagged, stop_tags) replacing the old TaggedValue class. The
stop_tags() function prevents tags from a sub-config and its
descendants from propagating to the parent's tag collection.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: bpiwowar

src/experimaestro/__init__.py

@@ -7,6 +7,7 @@
     tag,
     tags,
     tagspath,
+    stop_tags,
     STDOUT,
     STDERR,
     deprecate,
@@ -94,6 +95,7 @@
     "tag",
     "tags",
     "tagspath",
+    "stop_tags",
     "STDOUT",
     "STDERR",
     "deprecate",

src/experimaestro/annotations.py

@@ -108,7 +108,29 @@ def tag(value):
     :param value: The value to tag (str, int, float, or bool)
     :return: A tagged value wrapper that preserves the original value
     """
-    return objects.TaggedValue(value)
+    wrapper = objects.ConfigWrapper.ensure(value)
+    wrapper.tagged = True
+    return wrapper
+
+
+def stop_tags(value):
+    """Prevent tags from a sub-configuration from propagating to the parent.
+
+    Example::
+
+        config = MyConfig.C(x=stop_tags(SubConfig.C(lr=tag(0.001))))
+        tags(config)  # Will NOT include "lr"
+
+    Can be combined with ``tag``::
+
+        config = MyConfig.C(x=stop_tags(tag(SubConfig.C(lr=tag(0.001)))))
+
+    :param value: The value to wrap
+    :return: A wrapped value that stops tag propagation
+    """
+    wrapper = objects.ConfigWrapper.ensure(value)
+    wrapper.stop_tags = True
+    return wrapper
 
 
 class TagDict(SortedDict):

src/experimaestro/core/objects/__init__.py

@@ -22,6 +22,7 @@
     add_to_path,
     ObjectStore,
     SealedError,
+    ConfigWrapper,
     TaggedValue,
 )
 
@@ -41,6 +42,7 @@
     "WatchedOutput",
     "SealedError",
     "DependentMarker",
+    "ConfigWrapper",
     "TaggedValue",
     "getqualattr",
     "copyconfig",

src/experimaestro/core/objects/config.py

@@ -47,7 +47,7 @@
 from .config_utils import (
     getqualattr,
     add_to_path,
-    TaggedValue,
+    ConfigWrapper,
     ObjectStore,
     classproperty,
 )
@@ -201,6 +201,10 @@ def __init__(self, pyobject: "ConfigMixin"):
         self._tags: dict[str, tuple[Any, str]] = {}
         self._initinfo = ""
 
+        # Per-argument properties set by ConfigWrapper (e.g., "stop_tags")
+        # Maps argument name -> set of property strings
+        self._args_properties: dict[str, set[str]] = {}
+
         self._taskoutput = None
         """Task output (caches the value of a submit)"""
 
@@ -402,6 +406,10 @@ def __init__(self):
                 # Store {name: (value, source)} for conflict detection
                 self.tags_with_source: dict[str, tuple[Any, str]] = {}
 
+            def should_recurse_arg(self, config, arg_name: str) -> bool:
+                props = config.__xpm__._args_properties.get(arg_name)
+                return props is None or "stop_tags" not in props
+
             def postprocess(self, stub, config: Config, values):
                 for name, (value, source) in config.__xpm__._tags.items():
                     if name in self.tags_with_source:
@@ -1606,14 +1614,12 @@ def __setattr__(self, name: str, value):
         # Check if this is an XPM argument
         xpmtype = self.__xpmtype__
         if name in xpmtype.arguments:
-            # Handle TaggedValue: extract value and add tag
-            if isinstance(value, TaggedValue):
-                actual_value = value.value
+            # Handle ConfigWrapper (tag, stop_tags, etc.)
+            if isinstance(value, ConfigWrapper):
                 source = get_caller_location(skip_frames=1)
-                xpm.addtag(name, actual_value, source=source)
-                xpm.set(name, actual_value)
-            else:
-                xpm.set(name, value)
+                value.apply(xpm, name, source=source)
+                value = value.value
+            xpm.set(name, value)
             return
 
         # Check for deprecated replacement warning
@@ -1671,11 +1677,10 @@ def __init__(self, **kwargs):
                     continue
                 raise ValueError("%s is not an argument for %s" % (name, xpmtype))
 
-            # Special case of a tagged value
-            if isinstance(value, TaggedValue):
+            # Handle ConfigWrapper (tag, stop_tags, etc.)
+            if isinstance(value, ConfigWrapper):
+                value.apply(xpm, name, source=xpm._initinfo)
                 value = value.value
-                # Use _initinfo as source since tag is set at config creation
-                self.__xpm__.addtag(name, value, source=xpm._initinfo)
 
             # Really set the value
             xpm.set(name, value)

src/experimaestro/core/objects/config_utils.py

@@ -48,9 +48,40 @@ class SealedError(Exception):
     pass
 
 
-class TaggedValue:
-    def __init__(self, value):
+class ConfigWrapper:
+    """Wraps a config value with properties that modify how the parent processes it.
+
+    Properties:
+        tagged: If True, the value is tagged (appears in experiment tags)
+        stop_tags: If True, tags from this sub-config don't propagate to parent
+
+    Wrappers can be nested — properties are merged automatically::
+
+        stop_tags(tag(value))  # both tagged=True and stop_tags=True
+    """
+
+    def __init__(self, value, *, tagged: bool = False, stop_tags: bool = False):
         self.value = value
+        self.tagged = tagged
+        self.stop_tags = stop_tags
+
+    @staticmethod
+    def ensure(value) -> "ConfigWrapper":
+        """Return value as a ConfigWrapper, reusing it if already one."""
+        if isinstance(value, ConfigWrapper):
+            return value
+        return ConfigWrapper(value)
+
+    def apply(self, config_info: "ConfigInformation", arg_name: str, *, source: str):  # noqa: F821
+        """Apply all wrapper properties to the parent ConfigInformation"""
+        if self.tagged:
+            config_info.addtag(arg_name, self.value, source=source)
+        if self.stop_tags:
+            config_info._args_properties.setdefault(arg_name, set()).add("stop_tags")
+
+
+# Backwards-compatible alias
+TaggedValue = ConfigWrapper
 
 
 class classproperty(property):

src/experimaestro/core/objects/config_walk.py

@@ -94,6 +94,10 @@ def map(self, k: str):
         """Provides a path context when processing a tree"""
         return self.context.push(k)
 
+    def should_recurse_arg(self, config, arg_name: str) -> bool:
+        """Whether to recurse into this argument. Override to skip."""
+        return True
+
     def stub(self, config):
         return config
 
@@ -123,7 +127,7 @@ def __call__(self, x):
             # Process all the arguments
             result = {}
             for arg, v in info.xpmvalues():
-                if v is not None:
+                if v is not None and self.should_recurse_arg(x, arg.name):
                     with self.map(arg.name):
                         result[arg.name] = self(v)
                 else:

src/experimaestro/tests/test_tags.py

@@ -6,6 +6,7 @@
 
 from experimaestro import (
     tag,
+    stop_tags,
     LightweightTask,
     Config,
     Task,
@@ -260,3 +261,109 @@ class Outer(Config):
     assert any("value" in record.message for record in caplog.records)
     assert tags["value"] == 99  # Last value wins
     assert tags["x"] == 2
+
+
+# --- stop_tags tests ---
+
+
+def test_stop_tags_prevents_propagation():
+    """stop_tags prevents nested tags from propagating to parent"""
+
+    class Inner(Config):
+        lr: Param[float]
+
+    class Outer(Config):
+        x: Param[Inner]
+
+    inner = Inner.C(lr=tag(0.1))
+    outer = Outer.C(x=stop_tags(inner))
+
+    assert inner.tags() == {"lr": 0.1}
+    assert outer.tags() == {}
+
+
+def test_stop_tags_parent_tags_still_work():
+    """Tags on the parent itself still work with stop_tags on a child"""
+
+    class Inner(Config):
+        lr: Param[float]
+
+    class Outer(Config):
+        x: Param[Inner]
+        y: Param[int]
+
+    inner = Inner.C(lr=tag(0.1))
+    outer = Outer.C(x=stop_tags(inner), y=tag(42))
+
+    assert outer.tags() == {"y": 42}
+
+
+def test_stop_tags_via_setattr():
+    """stop_tags works when set via attribute assignment"""
+
+    class Inner(Config):
+        lr: Param[float]
+
+    class Outer(Config):
+        x: Param[Inner]
+
+    inner = Inner.C(lr=tag(0.1))
+    outer = Outer.C(x=Inner.C(lr=0.5))
+    outer.x = stop_tags(inner)
+
+    assert outer.tags() == {}
+
+
+def test_stop_tags_deeply_nested():
+    """stop_tags blocks tags from all descendants, not just direct child"""
+
+    class Deep(Config):
+        v: Param[int]
+
+    class Mid(Config):
+        d: Param[Deep]
+
+    class Top(Config):
+        m: Param[Mid]
+        label: Param[str]
+
+    deep = Deep.C(v=tag(1))
+    mid = Mid.C(d=deep).tag("mid_tag", "hello")
+    top = Top.C(m=stop_tags(mid), label=tag("test"))
+
+    assert top.tags() == {"label": "test"}
+
+
+def test_stop_tags_combined_with_tag():
+    """stop_tags and tag can be combined via nesting"""
+
+    class Inner(Config):
+        lr: Param[float]
+
+    class Outer(Config):
+        x: Param[Inner]
+
+    inner = Inner.C(lr=tag(0.1))
+    # tag the argument itself, but stop inner tags from propagating
+    outer = Outer.C(x=stop_tags(tag(inner)))
+
+    # "x" is tagged (from tag()), but "lr" is blocked (from stop_tags())
+    assert outer.tags() == {"x": inner}
+
+
+def test_stop_tags_does_not_affect_siblings():
+    """stop_tags on one argument doesn't affect other arguments"""
+
+    class Inner(Config):
+        v: Param[int]
+
+    class Outer(Config):
+        a: Param[Inner]
+        b: Param[Inner]
+
+    inner_a = Inner.C(v=tag(1))
+    inner_b = Inner.C(v=tag(2))
+    outer = Outer.C(a=stop_tags(inner_a), b=inner_b)
+
+    # Only inner_b's tag propagates
+    assert outer.tags() == {"v": 2}