fix(channels): DeltaChannel subclass safety and order-independent Overwrite

copy() and from_checkpoint() hardcoded DeltaChannel instead of
self.__class__, breaking subclasses. Now mirrors the BinaryOperatorAggregate
pattern: self.__class__(self.operator) with explicit typ/key assignment.

update() applied non-overwrite values that arrived before an Overwrite in
the sequence, then discarded them when the Overwrite fired — order-dependent
behaviour in a method whose contract says order is arbitrary. Now pre-scans
for an Overwrite and applies only it (or folds all values normally if none).

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sydney Runkle
2026-04-30 14:49:05 -04:00
co-authored by Claude Sonnet 4.6
parent b59daa1f7c
commit b0a9ca7daa
+10 -8
View File
@@ -145,19 +145,21 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
def update(self, values: Sequence[Any]) -> bool:
if not values:
return False
seen_overwrite = False
for value in values:
is_overwrite, _ = _get_overwrite(value)
if is_overwrite:
if seen_overwrite:
overwrite_idx: int | None = None
for i, v in enumerate(values):
is_ow, _ = _get_overwrite(v)
if is_ow:
if overwrite_idx is not None:
msg = create_error_message(
message="Can receive only one Overwrite value per super-step.",
error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
)
raise InvalidUpdateError(msg)
seen_overwrite = True
elif seen_overwrite:
continue
overwrite_idx = i
if overwrite_idx is not None:
self.value = self._apply_write(self.value, values[overwrite_idx])
return True
for value in values:
self.value = self._apply_write(self.value, value)
return True