visible_block(messages: object, block: str, selected_lines: tuple[str, ...]) -> bool
True only for one verbatim system occurrence; refuse partial/duplicate.
The absence of an entire block is an observed compiler or assembly drop.
A line surfacing outside the block could have come from user text or a
transformed contribution, so it cannot be attributed to hydration.
Source code in src/symfonic/capabilities/memory/capture_visibility.py
| def visible_block(
messages: object, block: str, selected_lines: tuple[str, ...]
) -> bool:
"""True only for one verbatim system occurrence; refuse partial/duplicate.
The absence of an entire block is an observed compiler or assembly drop.
A line surfacing outside the block could have come from user text or a
transformed contribution, so it cannot be attributed to hydration.
"""
if not isinstance(messages, Sequence) or isinstance(messages, (str, bytes)):
raise UnobservableWire("provider call has no message sequence")
texts = []
systems = []
for message in messages:
value = _text(message)
texts.append(value)
if getattr(message, "type", None) == "system":
systems.append(value)
if not block:
return False
occurrences = sum(value.count(block) for value in texts)
system_occurrences = sum(value.count(block) for value in systems)
if occurrences == system_occurrences == 1:
for value in texts:
# Split away only the declared system block. An echoed selected
# line in any remaining text cannot be attributed to hydration.
remaining = value.split(block, 1) if block in value else (value,)
if any(line in part for line in selected_lines for part in remaining):
raise ValueError("hydration is duplicated, moved, or transformed on provider wire")
return True
if occurrences > 0 or any(
line in value for line in selected_lines for value in texts
):
raise ValueError("hydration is duplicated, moved, or transformed on provider wire")
return False
|