per-atom tok_pos/text_span and per-root Hyperedge.tokens source-position metadata; continuity-aware sub-edge text derivation with verbatim character-offset slicing

Author: telmomenezes

CHANGELOG.md

@@ -4,16 +4,25 @@
 
 ### Added
 
-- Hyperedge.transform(): pattern-based rewrites.
+- `Hyperedge.transform()`: pattern-based rewrites.
+- per-atom `tok_pos`/`text_span` and per-root `Hyperedge.tokens` source-position metadata.
+- continuity-aware sub-edge text derivation with verbatim character-offset slicing.
+- `transforms.tok_pos_tree(edge)` rebuilds the parallel `tok_pos` tree from in-memory atoms.
 - readers provide source information.
 - `hyperbase.parsers.badness` module for parser-agnostic combined structural + token-matching validation.
 - built-in REPL setting `check_badness` to render a badness panel after each parse, available regardless of the active parser plugin.
 - more REPL commands: /load, /search, /count, /types and /transform.
 
 ### Changed
 
+- transforms propagate per-atom `tok_pos`/`text_span` through rewrites.
+- REPL `/load` now uses `ParseResult.from_dict` (was dropping `tokens`/`tok_pos`).
 - REPL shows multiple results if available.
 
+### Fixed
+
+- pattern matcher no longer returns partial bindings on structural mismatch in `{...}` argrole patterns.
+
 ### Removed
 
 ## [0.10.0] - 11-04-2026

src/hyperbase/builders.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from collections.abc import Iterable
+import itertools
+from collections.abc import Iterable, Iterator
 from typing import Any, cast
 
 from hyperbase.constants import ATOM_ENCODE_TABLE
@@ -130,30 +131,123 @@ def _collect_positions(tok_pos: Hyperedge) -> list[int]:
         return positions
 
 
-def _rebuild_with_text(
+def _compute_token_offsets(
+    tokens: list[str], text: str
+) -> list[tuple[int, int] | None]:
+    """Cursor-scan ``text`` for each token in order, returning per-token
+    (start, end) byte offsets. Tokens that can't be located after the running
+    cursor get ``None`` and the cursor is left unchanged so a single failed
+    token does not poison the rest of the sentence.
+    """
+    spans: list[tuple[int, int] | None] = []
+    cursor = 0
+    for tok in tokens:
+        idx = text.find(tok, cursor)
+        if idx < 0:
+            spans.append(None)
+        else:
+            spans.append((idx, idx + len(tok)))
+            cursor = idx + len(tok)
+    return spans
+
+
+def _rebuild_with_metadata(
     edge: Hyperedge,
     tok_pos: Hyperedge,
     tokens: list[str],
+    text: str,
+    offsets: list[tuple[int, int] | None],
+    claimed: frozenset[int],
 ) -> Hyperedge:
-    """Recursively rebuild an edge, assigning text from tokens and tok_pos."""
+    """Recursively rebuild an edge, populating per-atom tok_pos/text_span and
+    per-non-atom continuity-aware ``text``. ``claimed`` is the global set of
+    token positions actually mapped to atoms anywhere in the loaded root.
+    """
     if edge.atom:
         atom = cast(Atom, edge)
         pos = int(str(tok_pos))
-        text = tokens[pos] if pos >= 0 else None
-        return Atom(str(atom), atom.parens, text=text)
+        if pos < 0:
+            return Atom(str(atom), atom.parens)
+        span = offsets[pos] if 0 <= pos < len(offsets) else None
+        atom_text = (
+            text[span[0] : span[1]]
+            if span is not None
+            else (tokens[pos] if 0 <= pos < len(tokens) else None)
+        )
+        return Atom(str(atom), atom.parens, text=atom_text, tok_pos=pos, text_span=span)
     else:
         new_children = tuple(
-            _rebuild_with_text(sub_edge, sub_tok_pos, tokens)
+            _rebuild_with_metadata(
+                sub_edge, sub_tok_pos, tokens, text, offsets, claimed
+            )
             for sub_edge, sub_tok_pos in zip(edge, tok_pos, strict=False)
         )
-        positions = _collect_positions(tok_pos)
-        if positions:
-            min_pos = min(positions)
-            max_pos = max(positions)
-            text = " ".join(tokens[min_pos : max_pos + 1])
+        sub_text = _derive_subedge_text(new_children, text, tokens, offsets, claimed)
+        return Hyperedge(new_children, text=sub_text)
+
+
+def _derive_subedge_text(
+    children: tuple[Hyperedge, ...],
+    root_text: str,
+    root_tokens: list[str],
+    offsets: list[tuple[int, int] | None],
+    claimed: frozenset[int],
+) -> str | None:
+    """Continuity-aware text derivation for a non-atomic edge.
+
+    Walks the descendant atoms, uses their ``tok_pos`` to identify which
+    source positions the sub-edge references (``used``), and slices the
+    root's ``text`` per contiguous run. A run is broken only by a position
+    in ``claimed`` (a real atom anywhere in the root) that is not in
+    ``used`` -- synthetic atoms and unclaimed tokens (e.g. punctuation that
+    no atom maps to) keep the run continuous.
+    """
+    used_positions: set[int] = set()
+    for a in _walk_atoms(children):
+        pos = cast(Atom, a).tok_pos
+        if pos is not None:
+            used_positions.add(pos)
+    used: list[int] = sorted(used_positions)
+    if not used:
+        return None
+
+    span_by_pos: dict[int, tuple[int, int] | None] = {}
+    for a in _walk_atoms(children):
+        atom = cast(Atom, a)
+        if atom.tok_pos is not None and atom.tok_pos not in span_by_pos:
+            span_by_pos[atom.tok_pos] = atom.text_span
+
+    # Group used positions into runs, splitting whenever a claimed-but-unused
+    # position falls between two consecutive used positions.
+    used_set = set(used)
+    runs: list[list[int]] = [[used[0]]]
+    for prev, curr in itertools.pairwise(used):
+        broken = any(prev < p < curr and p not in used_set for p in claimed)
+        if broken:
+            runs.append([curr])
+        else:
+            runs[-1].append(curr)
+
+    slices: list[str] = []
+    for run in runs:
+        first_span = span_by_pos.get(run[0])
+        last_span = span_by_pos.get(run[-1])
+        if first_span is not None and last_span is not None:
+            slices.append(root_text[first_span[0] : last_span[1]])
         else:
-            text = None
-        return Hyperedge(new_children, text=text)
+            # Local fallback: token-join for this run.
+            slices.append(
+                " ".join(root_tokens[p] for p in run if 0 <= p < len(root_tokens))
+            )
+    return " ".join(slices)
+
+
+def _walk_atoms(edges: tuple[Hyperedge, ...]) -> Iterator[Hyperedge]:
+    for e in edges:
+        if e.atom:
+            yield e
+        else:
+            yield from _walk_atoms(tuple(e))
 
 
 def hedge(
@@ -162,8 +256,23 @@ def hedge(
     """Create a hyperedge."""
     if isinstance(source, ParseResult):
         _source = source
-        edge = _rebuild_with_text(_source.edge, _source.tok_pos, _source.tokens)
-        object.__setattr__(edge, "text", _source.text)
+        offsets = _compute_token_offsets(_source.tokens, _source.text)
+        # Compute claimed = all token positions referenced by atoms in the
+        # original edge tree. Used to detect real (non-synthetic, non-padding)
+        # discontinuities in sub-edge text derivation.
+        claimed = frozenset(_collect_positions(_source.tok_pos))
+        edge = _rebuild_with_metadata(
+            _source.edge,
+            _source.tok_pos,
+            _source.tokens,
+            _source.text,
+            offsets,
+            claimed,
+        )
+        # Override the root's text with the verbatim original sentence and
+        # store tokens on the root for any later derivation.
+        object.__setattr__(edge, "_text", _source.text)
+        object.__setattr__(edge, "tokens", tuple(_source.tokens))
         return edge
     if type(source) in {tuple, list}:
         _source = cast(Iterable, source)

src/hyperbase/cli/repl.py

@@ -1115,6 +1115,8 @@ def _render_count_row(self, n: int, key: object, count: int) -> None:
         self.console.print()
 
     def _load_edges_from_jsonl(self, path: Path) -> tuple[list[Hyperedge], int]:
+        from hyperbase.parsers.parse_result import ParseResult
+
         edges: list[Hyperedge] = []
         skipped = 0
         with open(path) as f:
@@ -1128,12 +1130,23 @@ def _load_edges_from_jsonl(self, path: Path) -> tuple[list[Hyperedge], int]:
                     if not isinstance(edge_str, str):
                         skipped += 1
                         continue
-                    edge = hedge(edge_str)
+                    # Prefer the full parse-result path so loaded edges carry
+                    # tokens, text, and per-atom tok_pos/text_span metadata.
+                    # Fall back to a bare hedge() parse if the row only has
+                    # the edge string.
+                    if (
+                        "tokens" in d
+                        and "tok_pos" in d
+                        and isinstance(d.get("text"), str)
+                    ):
+                        edge = hedge(ParseResult.from_dict(d))
+                    else:
+                        edge = hedge(edge_str)
                     if edge is None:
                         skipped += 1
                         continue
                     edges.append(edge)
-                except (json.JSONDecodeError, ValueError, TypeError):
+                except (json.JSONDecodeError, ValueError, TypeError, KeyError):
                     skipped += 1
         return edges, skipped
 

src/hyperbase/hyperedge.py

@@ -12,14 +12,21 @@ class Hyperedge:
     """Non-atomic hyperedge."""
 
     _edges: tuple[Hyperedge, ...]
-    text: str | None
+    _text: str | None
+    tokens: tuple[str, ...] | None = field(default=None, compare=False, hash=False)
     _cache: dict[str, Any] = field(
         default_factory=dict, repr=False, compare=False, hash=False
     )
 
-    def __init__(self, edges: Iterable[Hyperedge], text: str | None = None) -> None:
+    def __init__(
+        self,
+        edges: Iterable[Hyperedge],
+        text: str | None = None,
+        tokens: tuple[str, ...] | None = None,
+    ) -> None:
         object.__setattr__(self, "_edges", tuple(edges))
-        object.__setattr__(self, "text", text)
+        object.__setattr__(self, "_text", text)
+        object.__setattr__(self, "tokens", tokens)
         object.__setattr__(self, "_cache", {})
 
     def __iter__(self) -> Iterator[Hyperedge]:
@@ -49,6 +56,20 @@ def __eq__(self, other: object) -> bool:
     def __bool__(self) -> bool:
         return True
 
+    @property
+    def text(self) -> str | None:
+        """Original-source text for this (sub-)edge.
+
+        Populated eagerly by the parse-result builder: an atom carries the
+        substring it was located at (or ``None`` for synthetic atoms); a
+        non-atom carries the continuity-aware derivation against its loaded
+        root (one or more verbatim slices of ``root.text``, joined). Returns
+        ``None`` for edges built outside of a parse-result load (e.g. fresh
+        transform output) -- those preserve per-atom ``tok_pos``/``text_span``
+        but not the cached non-atom slice.
+        """
+        return self._text
+
     @property
     def atom(self) -> bool:
         """True if edge is an atom."""
@@ -451,16 +472,23 @@ class Atom(Hyperedge):
 
     atom_str: str
     parens: bool
+    tok_pos: int | None = field(default=None, compare=False, hash=False)
+    text_span: tuple[int, int] | None = field(default=None, compare=False, hash=False)
 
     def __init__(
         self,
         atom_str: str,
         parens: bool = False,
         text: str | None = None,
+        tok_pos: int | None = None,
+        text_span: tuple[int, int] | None = None,
     ) -> None:
         object.__setattr__(self, "atom_str", atom_str)
         object.__setattr__(self, "parens", parens)
-        object.__setattr__(self, "text", text)
+        object.__setattr__(self, "_text", text)
+        object.__setattr__(self, "tok_pos", tok_pos)
+        object.__setattr__(self, "text_span", text_span)
+        object.__setattr__(self, "tokens", None)
         object.__setattr__(self, "_edges", ())
         object.__setattr__(self, "_cache", {})
 
@@ -492,11 +520,19 @@ def root(self) -> str:
         return self.parts()[0]
 
     def replace_atom_part(self, part_pos: int, part: str) -> Atom:
-        """Build a new atom by replacing an atom part in a given atom."""
+        """Build a new atom by replacing an atom part in a given atom.
+
+        Source-position metadata (``tok_pos`` / ``text_span``) is preserved.
+        """
         parts = self.parts()
         parts[part_pos] = part
         atom_str = "/".join([part for part in parts if part])
-        return Atom(atom_str)
+        return Atom(
+            atom_str,
+            text=self._text,
+            tok_pos=self.tok_pos,
+            text_span=self.text_span,
+        )
 
     def label(self) -> str:
         """Generate human-readable label from entity."""

src/hyperbase/patterns/matcher.py

@@ -339,12 +339,12 @@ def _match_by_argroles(
         for pitem in pitems
     ]
 
-    # early exit if any pattern position has zero candidates
+    # early exit if any pattern position has zero candidates: the role is
+    # present but no edge item can fill it, which is a hard mismatch -- never
+    # a partial. (This differs from the len(eitems) < n case above, which
+    # corresponds to a missing-but-optional role under the {req,opt} syntax.)
     if any(len(c) == 0 for c in candidates):
-        if len(curvars) >= min_vars:
-            return [curvars]
-        else:
-            return []
+        return []
 
     result: list[dict[str, Hyperedge]] = []