From 8797cfce3122ff39991ab74f20ea76034445ca43 Mon Sep 17 00:00:00 2001
From: Tim Hatch <tim@timhatch.com>
Date: Wed, 16 Sep 2026 18:47:04 +0000
Subject: [PATCH] Add targeted source map lookup

---
 README.md                                |  23 ++
 parse_errors/context.py                  |   6 +-
 parse_errors/json_source_map/__init__.py | 227 ++++++++++++++++--
 parse_errors/source_map.py               |  83 ++++++-
 parse_errors/toml_source_map/__init__.py | 291 +++++++++++++++--------
 parse_errors/yaml_source_map/__init__.py | 101 +++++++-
 setup.cfg                                |   7 +-
 tests/test_parse_context_json.py         |  21 ++
 tests/test_source_map.py                 | 191 ++++++++++++++-
 9 files changed, 812 insertions(+), 138 deletions(-)

diff --git a/README.md b/README.md
index beb030d..ddff3b5 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,7 @@ positional information (`at line N, column M`) and re-raises them in the same
 
 ```python
 from parse_errors import ParseContext, ParseError
+from parse_errors.source_map import SourceMap, build_source_map, locate_pointer
 ```
 
 **`ParseContext(filename, *, data=None, format=None)`** — context manager.
@@ -68,6 +69,28 @@ from parse_errors import ParseContext, ParseError
 **`ParseError`** — the exception raised inside the context.  Has attributes
 `filename`, `line` (1-based), and `column` (1-based).
 
+**`locate_pointer(source, fmt, pointer)`** — returns the best source-map entry
+for one JSON Pointer without building a full map.  `fmt` is `"json"`,
+`"toml"`, or `"yaml"`; `source` is `str` or UTF-8 `bytes`; `pointer` uses RFC
+6901 escaping.  If the exact pointer is not present, the result matches
+`closest_entry(build_source_map(source, fmt), pointer)`: the nearest enclosing
+value when one exists, otherwise `None`.
+
+**`SourceMap(source, fmt)`** — caches the parsed document for repeated targeted
+lookups.  Use `SourceMap(...).locate(pointer)` when several errors in the same
+document need locations.  It still avoids constructing a full pointer-to-entry
+map.
+
+**`build_source_map(source, fmt)`** — builds the full pointer-to-location map.
+This is useful when callers need many arbitrary entries or need to inspect all
+locations.
+
+**Warning:** source-map helpers locate nodes in a document; they are not
+validating parsers.  JSON and TOML location support uses tree-sitter so it can
+return a location from a syntax tree even when a real decoder would reject the
+source.  Parse or validate the document with your normal parser first, then use
+these helpers only to map known error paths back to source locations.
+
 # Version Compat
 
 This library is compatile with Python 3.10+, but should be linted under the
diff --git a/parse_errors/context.py b/parse_errors/context.py
index 404f8cc..6f6b3b1 100644
--- a/parse_errors/context.py
+++ b/parse_errors/context.py
@@ -8,7 +8,7 @@
 from pathlib import Path
 from typing import Iterator
 
-from .source_map import detect_format, build_source_map, closest_entry, Location
+from .source_map import Location, detect_format, locate_pointer
 from ._jsonpath import extract_jsonpath, jsonpath_to_pointer
 
 POSITIONAL_RE = re.compile(r"at line (\d+), column (\d+)")
@@ -91,9 +91,7 @@ def ParseContext(
         assert fmt is not None
 
         source = data if data is not None else path.read_bytes()
-        source_map = build_source_map(source, fmt)
-
-        entry = closest_entry(source_map, pointer)
+        entry = locate_pointer(source, fmt, pointer)
         if entry is None:  # pragma: no cover
             raise exc
 
diff --git a/parse_errors/json_source_map/__init__.py b/parse_errors/json_source_map/__init__.py
index 043616e..1fd7eaa 100644
--- a/parse_errors/json_source_map/__init__.py
+++ b/parse_errors/json_source_map/__init__.py
@@ -1,25 +1,222 @@
-"""Thin wrapper around the json-source-map package that returns our own types."""
+"""Calculate JSON source locations using tree-sitter, targeted or full.
+
+tree-sitter does the parse in C instead of a hand-written pure-Python
+character scanner, and locate() additionally never visits a subtree that
+isn't on the path to the one pointer asked for -- indexing directly into
+tree-sitter's parsed node tree.
+
+calculate() (the full map, used by anything that still wants build_source_map
++ closest_entry) is kept for comparison/completeness.
+"""
 
 from __future__ import annotations
 
-import json_source_map as _ext
+import json
+
+import tree_sitter_json
+import tree_sitter as ts
 
 from ..source_map import Entry, Location, TSourceMap
+from .._jsonpath import _escape
+
+_LANGUAGE = ts.Language(tree_sitter_json.language())
+
+_Parsed = tuple[ts.Node, list[bytes]]
+
+
+def calculate(source: str | bytes) -> TSourceMap:
+    """Calculate the full source map for a JSON document."""
+    src_bytes = source.encode("utf-8") if isinstance(source, str) else source
+    root = ts.Parser(_LANGUAGE).parse(src_bytes).root_node
+    lines = src_bytes.splitlines(True) + [b""]
+    byte_to_char = [_byte_to_char_offsets(line) for line in lines]
+    line_start_chars = [0]
+    for offsets in byte_to_char[:-1]:
+        line_start_chars.append(line_start_chars[-1] + offsets[-1])
+
+    result: TSourceMap = {}
+    top = root.named_children[0] if root.named_children else root
+    _walk_all(top, [], result, byte_to_char, line_start_chars)
+    return result
+
+
+def _parse(source: str | bytes) -> _Parsed | None:
+    """Parse *source* into the (node, lines) pair :func:`_locate_in` walks.
+
+    Split out from :func:`locate` so :class:`~parse_errors.source_map.SourceMap`
+    can parse once and reuse it across several ``locate()`` calls on the same
+    document, instead of reparsing per pointer.
+    """
+    src_bytes = source.encode("utf-8") if isinstance(source, str) else source
+    root = ts.Parser(_LANGUAGE).parse(src_bytes).root_node
+    lines = src_bytes.splitlines(True) + [b""]
+    top = root.named_children[0] if root.named_children else root
+    return top, lines
+
+
+def _locate_in(parsed: _Parsed | None, pointer: str) -> Entry | None:
+    if parsed is None:
+        return None
+    top, lines = parsed
+    segments = _pointer_segments(pointer)
+    return _walk_targeted(top, segments, 0, lines)
+
+
+def locate(source: str | bytes, pointer: str) -> Entry | None:
+    """Find *pointer*'s location, visiting only nodes on the path to it."""
+    return _locate_in(_parse(source), pointer)
+
+
+# --- shared helpers ---
+
 
+def _pointer_segments(pointer: str) -> list[str]:
+    if not pointer:
+        return []
+    return [p.replace("~1", "/").replace("~0", "~") for p in pointer.split("/")[1:]]
 
-def calculate(source: str) -> TSourceMap:
-    """Calculate the source map for a JSON document."""
-    return {
-        pointer: Entry(
-            value_start=_loc(e.value_start),
-            value_end=_loc(e.value_end),
-            key_start=_loc(e.key_start) if e.key_start is not None else None,
-            key_end=_loc(e.key_end) if e.key_end is not None else None,
+
+def _to_pointer(segments: list[str]) -> str:
+    return "/" + "/".join(_escape(s) for s in segments) if segments else ""
+
+
+def _byte_to_char_offsets(line: bytes) -> list[int]:
+    offsets = [0] * (len(line) + 1)
+    char_count = 0
+    for i, b in enumerate(line):
+        if b < 0x80 or b >= 0xC0:
+            char_count += 1
+        offsets[i + 1] = char_count
+    return offsets
+
+
+# --- full map (calculate) ---
+
+
+def _walk_all(
+    node: ts.Node,
+    segments: list[str],
+    result: TSourceMap,
+    byte_to_char: list[list[int]],
+    line_start_chars: list[int],
+    key_node: ts.Node | None = None,
+) -> None:
+    pointer = _to_pointer(segments)
+    result[pointer] = Entry(
+        value_start=_loc_precomputed(node.start_point, byte_to_char, line_start_chars),
+        value_end=_loc_precomputed(node.end_point, byte_to_char, line_start_chars),
+        key_start=_loc_precomputed(key_node.start_point, byte_to_char, line_start_chars)
+        if key_node
+        else None,
+        key_end=_loc_precomputed(key_node.end_point, byte_to_char, line_start_chars)
+        if key_node
+        else None,
+    )
+    if node.type == "object":
+        for pair in node.named_children:
+            if pair.type != "pair":
+                continue  # pragma: no cover
+            k = pair.child_by_field_name("key")
+            v = pair.child_by_field_name("value")
+            assert k is not None and k.text is not None
+            assert v is not None
+            key_text = json.loads(k.text.decode("utf-8"))
+            _walk_all(
+                v,
+                segments + [key_text],
+                result,
+                byte_to_char,
+                line_start_chars,
+                key_node=k,
+            )
+    elif node.type == "array":
+        for i, child in enumerate(node.named_children):
+            _walk_all(
+                child, segments + [str(i)], result, byte_to_char, line_start_chars
+            )
+
+
+def _loc_precomputed(
+    point: ts.Point, byte_to_char: list[list[int]], line_start_chars: list[int]
+) -> Location:
+    char_column = byte_to_char[point.row][point.column]
+    return Location(
+        line=point.row,
+        column=char_column,
+        position=line_start_chars[point.row] + char_column,
+    )
+
+
+# --- targeted lookup ---
+
+
+def _walk_targeted(
+    node: ts.Node,
+    segments: list[str],
+    idx: int,
+    lines: list[bytes],
+    key_node: ts.Node | None = None,
+) -> Entry:
+    if idx == len(segments):
+        return Entry(
+            value_start=_loc(node.start_point, lines),
+            value_end=_loc(node.end_point, lines),
+            key_start=_loc(key_node.start_point, lines)
+            if key_node is not None
+            else None,
+            key_end=_loc(key_node.end_point, lines) if key_node is not None else None,
         )
-        for pointer, e in _ext.calculate(source).items()
-    }
+
+    if node.type == "object":
+        target_key = segments[idx]
+        for pair in node.named_children:
+            if pair.type != "pair":
+                continue  # pragma: no cover
+            k = pair.child_by_field_name("key")
+            assert k is not None and k.text is not None
+            key_text = json.loads(k.text.decode("utf-8"))
+            if key_text == target_key:
+                v = pair.child_by_field_name("value")
+                assert v is not None
+                return _walk_targeted(v, segments, idx + 1, lines, key_node=k)
+        # target_key not present -- this object is the closest ancestor
+        return _ancestor(node, lines, key_node)
+
+    if node.type == "array":
+        try:
+            target_index = int(segments[idx])
+        except ValueError:
+            target_index = -1
+        children = node.named_children
+        if 0 <= target_index < len(children):
+            return _walk_targeted(children[target_index], segments, idx + 1, lines)
+        return _ancestor(node, lines, key_node)
+
+    # A scalar, but the pointer wants to go deeper -- no such path; this
+    # value is the closest ancestor.
+    return _ancestor(node, lines, key_node)
+
+
+def _ancestor(node: ts.Node, lines: list[bytes], key_node: ts.Node | None) -> Entry:
+    # A fallback entry still carries key_start/key_end when this node is
+    # itself the value of an enclosing key -- matching the entry it would
+    # have gotten in the full map, at the pointer for that key.
+    return Entry(
+        value_start=_loc(node.start_point, lines),
+        value_end=_loc(node.end_point, lines),
+        key_start=_loc(key_node.start_point, lines) if key_node is not None else None,
+        key_end=_loc(key_node.end_point, lines) if key_node is not None else None,
+    )
 
 
-def _loc(ext: _ext.Location) -> Location:
-    """Translate to our internal structure."""
-    return Location(line=ext.line, column=ext.column, position=ext.position)
+def _loc(point: ts.Point, lines: list[bytes]) -> Location:
+    # Only called O(depth) times per locate() call (a handful), so decoding
+    # fresh each time -- rather than precomputing a whole-document table --
+    # is cheap; the table only paid for itself when amortized over every
+    # node in the document, which locate() deliberately never visits.
+    line = lines[point.row]
+    char_column = len(line[: point.column].decode("utf-8"))
+    position = (
+        sum(len(prior.decode("utf-8")) for prior in lines[: point.row]) + char_column
+    )
+    return Location(line=point.row, column=char_column, position=position)
diff --git a/parse_errors/source_map.py b/parse_errors/source_map.py
index 3207129..41dfcd9 100644
--- a/parse_errors/source_map.py
+++ b/parse_errors/source_map.py
@@ -4,7 +4,7 @@
 
 import dataclasses
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Any, Dict, Mapping, Optional, TypeVar
 
 
 # These are identical to the ones in json-source-map, but I feel icky exporting
@@ -27,6 +27,7 @@ class Entry:
 
 
 TSourceMap = Dict[str, Entry]
+TEntry = TypeVar("TEntry")
 
 
 def detect_format(path: Path) -> str | None:
@@ -62,7 +63,85 @@ def build_source_map(source: str | bytes, fmt: str) -> TSourceMap:
         raise ValueError(f"Unknown format: {fmt!r}")
 
 
-def closest_entry(source_map: TSourceMap, pointer: str) -> Entry | None:
+def _parse_for(fmt: str, source: str | bytes) -> Any:
+    # The three formats' parse trees (tree-sitter nodes for json/toml, a
+    # PyYAML Node for yaml) share no common type -- this is deliberately
+    # opaque, threaded straight to the matching format's own _locate_in().
+    if fmt == "json":
+        from . import json_source_map
+
+        return json_source_map._parse(source)
+    if fmt == "toml":
+        from . import toml_source_map
+
+        return toml_source_map._parse(source)
+    if fmt in ("yaml", "yml"):
+        from . import yaml_source_map
+
+        return yaml_source_map._parse(source)
+    raise ValueError(f"Unknown format: {fmt!r}")
+
+
+def _locate_in_for(fmt: str, parsed: Any, pointer: str) -> Entry | None:
+    if fmt == "json":
+        from . import json_source_map
+
+        return json_source_map._locate_in(parsed, pointer)
+    if fmt == "toml":
+        from . import toml_source_map
+
+        return toml_source_map._locate_in(parsed, pointer)
+    if fmt in ("yaml", "yml"):
+        from . import yaml_source_map
+
+        return yaml_source_map._locate_in(parsed, pointer)
+    raise ValueError(f"Unknown format: {fmt!r}")
+
+
+def locate_pointer(source: str | bytes, fmt: str, pointer: str) -> Entry | None:
+    """Find the location of one pointer directly, without building the whole map.
+
+    A targeted walk over each format's own parse tree -- tree-sitter for
+    json/toml, PyYAML's Node tree (via ``get_single_node()``, CSafeLoader when
+    available) for yaml -- visiting only nodes on the path to *pointer*,
+    unlike :func:`build_source_map` which computes an entry for every value
+    in the document regardless of which one is wanted.
+
+    Looking up several pointers in the same document this way reparses it
+    each time; use :class:`SourceMap` instead to parse once and reuse it.
+    """
+    return _locate_in_for(fmt, _parse_for(fmt, source), pointer)
+
+
+class SourceMap:
+    """Repeated pointer lookups against one document, parsed at most once.
+
+    :func:`locate_pointer` costs O(pointer depth) per call, but reparses the
+    document from scratch every time -- fine for one lookup, wasteful for
+    several against the same source, such as a merged config reporting one
+    error per field. ``SourceMap`` defers parsing until the first
+    :meth:`locate` call, caches that parse, and reuses it for every pointer
+    asked after -- still never building a whole-document map the way
+    :func:`build_source_map` does, since each lookup still only walks the
+    path to the one pointer it was asked for.
+    """
+
+    __slots__ = ("_source", "_fmt", "_parsed", "_has_parsed")
+
+    def __init__(self, source: str | bytes, fmt: str) -> None:
+        self._source = source
+        self._fmt = fmt
+        self._parsed: Any = None
+        self._has_parsed = False
+
+    def locate(self, pointer: str) -> Entry | None:
+        if not self._has_parsed:
+            self._parsed = _parse_for(self._fmt, self._source)
+            self._has_parsed = True
+        return _locate_in_for(self._fmt, self._parsed, pointer)
+
+
+def closest_entry(source_map: Mapping[str, TEntry], pointer: str) -> TEntry | None:
     """Return the source map entry for ``pointer``, falling back to the longest prefix."""
     if pointer in source_map:
         return source_map[pointer]
diff --git a/parse_errors/toml_source_map/__init__.py b/parse_errors/toml_source_map/__init__.py
index b3277de..89c2039 100644
--- a/parse_errors/toml_source_map/__init__.py
+++ b/parse_errors/toml_source_map/__init__.py
@@ -1,138 +1,226 @@
-"""Calculate the source map for a TOML document using tree-sitter."""
+"""Calculate TOML source locations using tree-sitter, targeted or full.
+
+locate() extends the existing tree-sitter-based calculate() with a targeted
+walk that only visits table/pair/array members on the path to the one
+pointer asked for -- a sibling pair under a non-matching key, or a whole
+non-matching [[array-of-tables]] element, is skipped without recursing into
+it or building any Entry for it. Array-of-tables members still have to be
+counted in document order to know which occurrence is index N (TOML doesn't
+number them explicitly), but skipped occurrences are only counted, not
+walked. calculate() and locate() share that counting through a single
+_header_candidates() generator rather than each keeping its own.
+"""
 
 from __future__ import annotations
 
+from typing import Iterator, Optional
+
 import tree_sitter_toml
 import tree_sitter as ts
 from ..source_map import Entry, Location, TSourceMap
 from .._jsonpath import _escape
 
+_LANGUAGE = ts.Language(tree_sitter_toml.language())
 
-def calculate(source: str | bytes) -> TSourceMap:
-    """Calculate the source map for a TOML document.
+_Parsed = tuple[ts.Node, bytes]
 
-    Args:
-        source: The TOML document as a string or bytes.
 
-    Returns:
-        A dict mapping JSON Pointer paths to Entry objects with location info.
-    """
+def calculate(source: str | bytes) -> TSourceMap:
+    """Calculate the full source map for a TOML document."""
     src_bytes = source if isinstance(source, bytes) else source.encode("utf-8")
-    parser = ts.Parser(ts.Language(tree_sitter_toml.language()))
-    root = parser.parse(src_bytes).root_node
+    root = ts.Parser(_LANGUAGE).parse(src_bytes).root_node
 
     result: TSourceMap = {}
-    aot_counts: dict[str, int] = {}
-
-    result[""] = Entry(
-        value_start=_loc(root.start_point, src_bytes),
-        value_end=_loc(root.end_point, src_bytes),
-    )
-
+    result[""] = _own_entry(root, src_bytes)
     for child in root.children:
         if child.type == "pair":
             _process_pair(child, [], result, src_bytes)
-        elif child.type == "table":
-            _process_table(child, result, aot_counts, src_bytes)
+    for key_segs, node, entry in _header_candidates(root.children, src_bytes):
+        result[_to_pointer(key_segs)] = entry
+        if node is not None:
+            for grandchild in node.children:
+                if grandchild.type == "pair":
+                    _process_pair(grandchild, key_segs, result, src_bytes)
+    return result
+
+
+def _parse(source: str | bytes) -> _Parsed | None:
+    """Parse *source* into the (root, src_bytes) pair :func:`_locate_in` walks.
+
+    Split out from :func:`locate` so :class:`~parse_errors.source_map.SourceMap`
+    can parse once and reuse it across several ``locate()`` calls on the same
+    document, instead of reparsing per pointer.
+    """
+    src_bytes = source if isinstance(source, bytes) else source.encode("utf-8")
+    root = ts.Parser(_LANGUAGE).parse(src_bytes).root_node
+    return root, src_bytes
+
+
+def _locate_in(parsed: _Parsed | None, pointer: str) -> Optional[Entry]:
+    if parsed is None:
+        return None
+    root, src_bytes = parsed
+    segments = _pointer_segments(pointer)
+    if not segments:
+        return _own_entry(root, src_bytes)
+    found = _find_in_container(root.children, segments, 0, src_bytes)
+    return found if found is not None else _own_entry(root, src_bytes)
+
+
+def locate(source: str | bytes, pointer: str) -> Optional[Entry]:
+    """Find *pointer*'s location, visiting only members on the path to it."""
+    return _locate_in(_parse(source), pointer)
+
+
+def _find_in_container(
+    children: list[ts.Node], target: list[str], idx: int, src: bytes
+) -> Optional[Entry]:
+    """*children* is a document's or table's direct pair/table/table_array_element
+    nodes. Returns the entry for target[idx:] if reachable through them, else None.
+
+    Matches calculate()'s existing scope exactly: a plain array is a leaf --
+    its elements are never individually addressable, only [[[array-of-
+    tables]]] elements are (by document-order occurrence, counted here as
+    they're seen; a skipped occurrence is counted, never walked).
+    """
+    remaining = target[idx:]
+
+    # A bare pair's key can't be extended by anything else -- pairs are
+    # always leaves (or an inline_table to descend into, see
+    # _find_in_value) -- so the first matching one is unambiguously the
+    # answer, unlike headers below.
+    for child in children:
+        if child.type != "pair":
+            continue
+        key_node, value_node = _pair_key_value(child)
+        key_segs = _key_segments(key_node)
+        if _is_prefix(key_segs, remaining):
+            new_idx = idx + len(key_segs)
+            if new_idx == len(target):
+                return _own_entry(value_node, src, key_node)
+            return _find_in_value(value_node, target, new_idx, src, key_node)
+
+    # Headers (tables and array-of-tables elements) form one flat,
+    # document-ordered namespace where a later header can continue an
+    # earlier one's dotted path -- e.g. `[items.detail]` implicitly nests
+    # under whichever `[[items]]` element most recently opened, even though
+    # tree-sitter parses it as a sibling, not a child, of that element. So
+    # unlike a pair, the first header whose key is a prefix of `remaining`
+    # isn't necessarily the answer: resolve every header's fully expanded
+    # path via _header_candidates() -- shared with calculate(), see its
+    # docstring -- then take the one whose path is the longest prefix of
+    # `remaining`, mirroring closest_entry()'s longest-prefix-wins rule over
+    # calculate()'s full map.
+    best: Optional[tuple[list[str], Optional[ts.Node], Entry]] = None
+    for key_segs, node, entry in _header_candidates(children, src):
+        if _is_prefix(key_segs, remaining) and (
+            best is None or len(key_segs) > len(best[0])
+        ):
+            best = (key_segs, node, entry)
+
+    if best is None:
+        return None
+    key_segs, node, entry = best
+    new_idx = idx + len(key_segs)
+    if new_idx == len(target) or node is None:
+        return entry
+    found = _find_in_container(node.children, target, new_idx, src)
+    return found if found is not None else entry
+
+
+def _header_candidates(
+    children: list[ts.Node], src: bytes
+) -> Iterator[tuple[list[str], Optional[ts.Node], Entry]]:
+    """Yield (expanded key segments, header node to recurse into or None, own
+    Entry) for each table/table_array_element among *children*, in document
+    order.
+
+    The one place that resolves a header's key path: a `[dotted.table]`
+    header implicitly nesting under the array element it continues
+    (`_expand_aot_segments`), and an array-of-tables element's occurrence
+    index among same-named siblings (TOML doesn't number them explicitly).
+    calculate() writes every candidate into its result dict and recurses
+    into all of them; `_find_in_container()` picks the longest-prefix match
+    against the pointer it wants and recurses into just that one. Both share
+    this single `aot_counts`.
+
+    An array's own group pointer (e.g. "/arr", not "/arr/0") is yielded once,
+    at its first occurrence, as a childless (node=None) candidate: calculate()
+    records this as a zero-width entry, and locate() needs it as a leaf, in
+    case a pointer stops exactly there or names an occurrence that never
+    happened.
+    """
+    aot_counts: dict[str, int] = {}
+    for child in children:
+        if child.type == "table":
+            key_node = _table_key(child)
+            key_segs = _expand_aot_segments(_key_segments(key_node), aot_counts)
+            yield key_segs, child, _own_entry(child, src)
         elif child.type == "table_array_element":
-            _process_aot(child, result, aot_counts, src_bytes)
+            key_node = _table_key(child)
+            raw_segs = _key_segments(key_node)
+            array_pointer = _to_pointer(raw_segs)
+            occurrence = aot_counts.get(array_pointer, 0)
+            if occurrence == 0:
+                loc0 = _loc(child.start_point, src)
+                yield raw_segs, None, Entry(value_start=loc0, value_end=loc0)
+            aot_counts[array_pointer] = occurrence + 1
+            yield raw_segs + [str(occurrence)], child, _own_entry(child, src)
 
-    return result
+
+def _find_in_value(
+    node: ts.Node, target: list[str], idx: int, src: bytes, key_node: ts.Node
+) -> Entry:
+    """*node* is a pair's value; target[idx:] wants to navigate past it, but
+    only an inline_table can be navigated into -- a plain array (like a
+    scalar) is always a leaf here, matching calculate()'s existing scope."""
+    if node.type == "inline_table":
+        pairs = [c for c in node.children if c.type == "pair"]
+        found = _find_in_container(pairs, target, idx, src)
+        if found is not None:
+            return found
+    return _own_entry(node, src, key_node)
+
+
+def _is_prefix(prefix: list[str], target: list[str]) -> bool:
+    return len(prefix) <= len(target) and target[: len(prefix)] == prefix
+
+
+def _pointer_segments(pointer: str) -> list[str]:
+    if not pointer:
+        return []
+    return [p.replace("~1", "/").replace("~0", "~") for p in pointer.split("/")[1:]]
+
+
+# --- shared by calculate() and locate() ---
 
 
 def _process_pair(
-    node: ts.Node,
-    prefix: list[str],
-    result: TSourceMap,
-    src: bytes,
+    node: ts.Node, prefix: list[str], result: TSourceMap, src: bytes
 ) -> None:
     key_node, value_node = _pair_key_value(node)
     segments = prefix + _key_segments(key_node)
-    pointer = _to_pointer(segments)
+    result[_to_pointer(segments)] = _own_entry(value_node, src, key_node)
 
     if value_node.type == "inline_table":
-        result[pointer] = Entry(
-            value_start=_loc(value_node.start_point, src),
-            value_end=_loc(value_node.end_point, src),
-            key_start=_loc(key_node.start_point, src),
-            key_end=_loc(key_node.end_point, src),
-        )
         for child in value_node.children:
             if child.type == "pair":
                 _process_pair(child, segments, result, src)
-    else:
-        result[pointer] = Entry(
-            value_start=_loc(value_node.start_point, src),
-            value_end=_loc(value_node.end_point, src),
-            key_start=_loc(key_node.start_point, src),
-            key_end=_loc(key_node.end_point, src),
-        )
-
-
-def _process_table(
-    node: ts.Node,
-    result: TSourceMap,
-    aot_counts: dict[str, int],
-    src: bytes,
-) -> None:
-    key_node = _table_key(node)
-    segments = _expand_aot_segments(_key_segments(key_node), aot_counts)
-    pointer = _to_pointer(segments)
-
-    result[pointer] = Entry(
-        value_start=_loc(node.start_point, src),
-        value_end=_loc(node.end_point, src),
-    )
-
-    for child in node.children:
-        if child.type == "pair":
-            _process_pair(child, segments, result, src)
-
-
-def _process_aot(
-    node: ts.Node,
-    result: TSourceMap,
-    aot_counts: dict[str, int],
-    src: bytes,
-) -> None:
-    key_node = _table_key(node)
-    segments = _key_segments(key_node)
-    array_pointer = _to_pointer(segments)
-
-    idx = aot_counts.get(array_pointer, 0)
-    aot_counts[array_pointer] = idx + 1
-
-    item_segments = segments + [str(idx)]
-    item_pointer = _to_pointer(item_segments)
-
-    entry_loc = _loc(node.start_point, src)
-    result.setdefault(array_pointer, Entry(value_start=entry_loc, value_end=entry_loc))
-    result[item_pointer] = Entry(
-        value_start=entry_loc,
-        value_end=_loc(node.end_point, src),
-    )
-
-    for child in node.children:
-        if child.type == "pair":
-            _process_pair(child, item_segments, result, src)
 
 
 def _pair_key_value(node: ts.Node) -> tuple[ts.Node, ts.Node]:
-    """Return (key_node, value_node) from a pair node."""
     key_node = value_node = None
     for child in node.children:
         if child.type in ("bare_key", "quoted_key", "dotted_key"):
             key_node = child
         elif child.type not in ("=", "comment"):
             value_node = child
-
     assert key_node is not None
     assert value_node is not None
     return key_node, value_node
 
 
 def _table_key(node: ts.Node) -> ts.Node:
-    """Return the key node from a table or table_array_element node."""
     for child in node.children:
         if child.type in ("bare_key", "quoted_key", "dotted_key"):
             return child
@@ -140,7 +228,6 @@ def _table_key(node: ts.Node) -> ts.Node:
 
 
 def _key_segments(node: ts.Node) -> list[str]:
-    """Extract key path segments from a key node."""
     if node.type == "bare_key":
         assert node.text is not None
         return [node.text.decode()]
@@ -153,23 +240,14 @@ def _key_segments(node: ts.Node) -> list[str]:
 
 
 def _unquote(s: str) -> str:
-    """Strip surrounding quotes from a TOML quoted key."""
     if s.startswith('"') and s.endswith('"'):
-        # Basic string: full TOML escape sequences via unicode_escape codec.
-        # Encode to latin-1 first to preserve non-ASCII literals, then decode escapes.
         return s[1:-1].encode("raw_unicode_escape").decode("unicode_escape")
     elif s.startswith("'") and s.endswith("'"):
-        # Literal string: no escaping
         return s[1:-1]
     return s  # pragma: no cover
 
 
 def _expand_aot_segments(segments: list[str], aot_counts: dict[str, int]) -> list[str]:
-    """If a prefix of segments matches a known AoT, splice in its current index.
-
-    e.g. segments=[fruits, details] with aot_counts={/fruits: 1}
-    → [fruits, 0, details]
-    """
     for i in range(1, len(segments)):
         prefix_pointer = _to_pointer(segments[:i])
         if prefix_pointer in aot_counts:
@@ -183,12 +261,19 @@ def _to_pointer(segments: list[str]) -> str:
 
 
 def _loc(point: ts.Point, src: bytes) -> Location:
-    # tree-sitter gives byte-based row/column; convert to char-based for consistency
-    # with json-source-map and yaml-source-map.
-    # Append b"" so end_point row (one past final newline) is always a valid index.
     lines = src.splitlines(True) + [b""]
     char_column = len(lines[point.row][: point.column].decode("utf-8"))
     position = (
         sum(len(line.decode("utf-8")) for line in lines[: point.row]) + char_column
     )
     return Location(line=point.row, column=char_column, position=position)
+
+
+def _own_entry(node: ts.Node, src: bytes, key_node: Optional[ts.Node] = None) -> Entry:
+    """*node*'s own span, with *key_node*'s span attached if it has one."""
+    return Entry(
+        value_start=_loc(node.start_point, src),
+        value_end=_loc(node.end_point, src),
+        key_start=_loc(key_node.start_point, src) if key_node is not None else None,
+        key_end=_loc(key_node.end_point, src) if key_node is not None else None,
+    )
diff --git a/parse_errors/yaml_source_map/__init__.py b/parse_errors/yaml_source_map/__init__.py
index dbd695b..8acaaf0 100644
--- a/parse_errors/yaml_source_map/__init__.py
+++ b/parse_errors/yaml_source_map/__init__.py
@@ -2,9 +2,21 @@
 
 from __future__ import annotations
 
+from typing import Optional
+
 import yaml
 from ..source_map import Entry, Location, TSourceMap
 
+# PyYAML's get_single_node() returns a Node tree (Mapping/Sequence/Scalar)
+# with start_mark/end_mark before constructing any Python values -- the same
+# role tree-sitter plays for json/toml. CSafeLoader (libyaml, C) gives the
+# identical Node/Mark interface as the pure-Python SafeLoader, so use it when
+# available; no third-party grammar package needed the way JSON needed
+# tree_sitter_json.
+_LoaderCls = getattr(yaml, "CSafeLoader", yaml.SafeLoader)
+
+_Parsed = tuple[yaml.Node, yaml.SafeLoader]
+
 
 def calculate(source: str) -> TSourceMap:
     """Calculate the source map for a YAML document.
@@ -15,8 +27,11 @@ def calculate(source: str) -> TSourceMap:
     Returns:
         A dict mapping JSON Pointer paths to Entry objects with location info.
     """
-    loader = yaml.SafeLoader(source)
-    node = loader.get_single_node()
+    loader = _LoaderCls(source)
+    try:
+        node = loader.get_single_node()
+    finally:
+        loader.dispose()
     if node is None:
         return {}
     result: TSourceMap = {}
@@ -24,6 +39,88 @@ def calculate(source: str) -> TSourceMap:
     return result
 
 
+def _parse(source: str | bytes) -> _Parsed | None:
+    """Parse *source* into the (node, loader) pair :func:`_locate_in` walks.
+
+    Split out from :func:`locate` so :class:`~parse_errors.source_map.SourceMap`
+    can parse once and reuse it across several ``locate()`` calls on the same
+    document, instead of reparsing per pointer. The loader is disposed (its
+    input buffer freed) before being returned -- kept only because node
+    construction (``construct_scalar``, for mapping keys) reads from the node
+    tree itself, not the buffer.
+    """
+    text = source.decode("utf-8") if isinstance(source, bytes) else source
+    loader = _LoaderCls(text)
+    try:
+        node = loader.get_single_node()
+    finally:
+        loader.dispose()
+    if node is None:
+        return None
+    return node, loader
+
+
+def _locate_in(parsed: _Parsed | None, pointer: str) -> Optional[Entry]:
+    if parsed is None:
+        return None
+    node, loader = parsed
+    segments = _pointer_segments(pointer)
+    return _walk_targeted(node, segments, 0, loader)
+
+
+def locate(source: str | bytes, pointer: str) -> Optional[Entry]:
+    """Find *pointer*'s location, visiting only nodes on the path to it."""
+    return _locate_in(_parse(source), pointer)
+
+
+def _pointer_segments(pointer: str) -> list[str]:
+    if not pointer:
+        return []
+    return [p.replace("~1", "/").replace("~0", "~") for p in pointer.split("/")[1:]]
+
+
+def _walk_targeted(
+    node: yaml.Node,
+    segments: list[str],
+    idx: int,
+    loader: yaml.SafeLoader,
+    key_node: Optional[yaml.Node] = None,
+) -> Entry:
+    if idx == len(segments):
+        return _entry(node, key_node)
+
+    if isinstance(node, yaml.MappingNode):
+        target_key = segments[idx]
+        for k, v in node.value:
+            if str(loader.construct_scalar(k)) == target_key:
+                return _walk_targeted(v, segments, idx + 1, loader, key_node=k)
+        # target_key not present -- this mapping is the closest ancestor
+        return _entry(node, key_node)
+
+    if isinstance(node, yaml.SequenceNode):
+        try:
+            target_index = int(segments[idx])
+        except ValueError:
+            target_index = -1
+        items = node.value
+        if 0 <= target_index < len(items):
+            return _walk_targeted(items[target_index], segments, idx + 1, loader)
+        return _entry(node, key_node)
+
+    # A scalar, but the pointer wants to go deeper -- no such path; this
+    # value is the closest ancestor.
+    return _entry(node, key_node)
+
+
+def _entry(node: yaml.Node, key_node: Optional[yaml.Node]) -> Entry:
+    return Entry(
+        value_start=_location(node.start_mark),
+        value_end=_location(node.end_mark),
+        key_start=_location(key_node.start_mark) if key_node is not None else None,
+        key_end=_location(key_node.end_mark) if key_node is not None else None,
+    )
+
+
 def _location(mark: yaml.Mark) -> Location:
     return Location(line=mark.line, column=mark.column, position=mark.index)
 
diff --git a/setup.cfg b/setup.cfg
index 01f55cb..667b268 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -16,23 +16,26 @@ setup_requires =
     setuptools >= 65
 include_package_data = true
 install_requires =
-    json-source-map
     pyyaml
-    tree-sitter
+    tree-sitter!=0.26.0
     tree-sitter-toml
+    tree-sitter-json
 
 [options.extras_require]
 dev =
     checkdeps == 0.9.0
     mypy == 1.19.1
     ruff == 0.15.6
+    setuptools >= 65
     tox == 4.50.0
     tox-uv == 1.33.4
+    ty == 0.0.81
     types-pyyaml
 test =
     coverage >= 6
     pytest >= 8
     msgspec
+    tomli
 
 [options.entry_points]
 # console_scripts =
diff --git a/tests/test_parse_context_json.py b/tests/test_parse_context_json.py
index 9bd57b3..65d1011 100644
--- a/tests/test_parse_context_json.py
+++ b/tests/test_parse_context_json.py
@@ -79,6 +79,27 @@ def test_json_original_exception_is_cause():
     assert isinstance(exc_info.value.__cause__, msgspec.ValidationError)
 
 
+def test_jsonpath_error_uses_targeted_location(monkeypatch):
+    from parse_errors import context
+    from parse_errors.source_map import Entry, Location
+
+    calls = []
+
+    def locate_pointer(source, fmt, pointer):
+        calls.append((source, fmt, pointer))
+        loc = Location(line=9, column=4, position=0)
+        return Entry(value_start=loc, value_end=loc)
+
+    monkeypatch.setattr(context, "locate_pointer", locate_pointer)
+
+    with pytest.raises(ParseError) as exc_info:
+        with ParseContext("config.json", data=JSON_SOURCE):
+            msgspec.json.decode(JSON_SOURCE.encode(), type=Config)
+
+    assert calls == [(JSON_SOURCE, "json", "/port")]
+    assert str(exc_info.value).startswith("config.json:10:5:")
+
+
 def test_incomplete_json():
     with pytest.raises(
         ParseError, match=r"config.json: DecodeError\('Input data was truncated'\)"
diff --git a/tests/test_source_map.py b/tests/test_source_map.py
index 8cf9ab4..ef2c10b 100644
--- a/tests/test_source_map.py
+++ b/tests/test_source_map.py
@@ -1,13 +1,25 @@
-from parse_errors.source_map import build_source_map, Location, Entry, closest_entry
+import pytest
+
+from parse_errors.source_map import (
+    build_source_map,
+    Location,
+    Entry,
+    closest_entry,
+    locate_pointer,
+    SourceMap,
+)
 
 
 def test_build_toml():
     # This isn't an exhaustive test of the toml source mapper, just as something
     # a minimal example that lets us exercise str/bytes
-    sm1 = build_source_map("""\
+    sm1 = build_source_map(
+        """\
 x=1
 b='foo'
-""", fmt="toml")
+""",
+        fmt="toml",
+    )
     assert sm1 == {
         "": Entry(
             value_start=Location(line=0, column=0, position=0),
@@ -40,10 +52,13 @@ def test_closest_entry():
 
 
 def test_build_toml_table():
-    sm = build_source_map("""\
+    sm = build_source_map(
+        """\
 [section]
 key = "val"
-""", fmt="toml")
+""",
+        fmt="toml",
+    )
     assert sm[""] == Entry(
         value_start=Location(line=0, column=0, position=0),
         value_end=Location(line=2, column=0, position=22),
@@ -61,12 +76,15 @@ def test_build_toml_table():
 
 
 def test_build_toml_aot():
-    sm = build_source_map("""\
+    sm = build_source_map(
+        """\
 [[items]]
 name = "a"
 [[items]]
 name = "b"
-""", fmt="toml")
+""",
+        fmt="toml",
+    )
     assert sm["/items"] == Entry(
         value_start=Location(line=0, column=0, position=0),
         value_end=Location(line=0, column=0, position=0),
@@ -120,10 +138,13 @@ def test_build_toml_dotted_key():
 
 
 def test_build_toml_quoted_keys():
-    sm = build_source_map('''\
+    sm = build_source_map(
+        """\
 "foo" = 1
 'bar' = 2
-''', fmt="toml")
+""",
+        fmt="toml",
+    )
     assert sm["/foo"] == Entry(
         value_start=Location(line=0, column=8, position=8),
         value_end=Location(line=0, column=9, position=9),
@@ -140,7 +161,7 @@ def test_build_toml_quoted_keys():
 
 def test_build_toml_aot_nested_table():
     sm = build_source_map(
-        "[[fruits]]\nname=\"apple\"\n[fruits.details]\ncolor=\"red\"\n", fmt="toml"
+        '[[fruits]]\nname="apple"\n[fruits.details]\ncolor="red"\n', fmt="toml"
     )
     assert "/fruits/0/details" in sm
     assert sm["/fruits/0/details/color"] == Entry(
@@ -154,3 +175,153 @@ def test_build_toml_aot_nested_table():
 def test_closest_entry_fallthrough():
     sm = {}
     assert closest_entry(sm, "/baz") is None
+
+
+_JSON_DOC = '{"a": {"b": [1, 2, {"c": 3}]}, "x": 1}'
+_YAML_DOC = "a:\n  b:\n    - 1\n    - 2\n    - c: 3\nx: 1\n"
+
+_JSON_YAML_POINTERS = [
+    "",
+    "/a",
+    "/a/b",
+    "/a/b/0",
+    "/a/b/2",
+    "/a/b/2/c",
+    "/x",
+    "/nonexistent",
+    "/a/b/99",
+]
+
+# A plain array is a leaf in TOML -- only [[array-of-tables]] elements are
+# individually addressable -- so it needs its own doc/pointer shape rather
+# than sharing the json/yaml ones above.
+_TOML_DOC = (
+    "top = 1\n"
+    '[[items]]\nname = "a"\n'
+    '[[items]]\nname = "b"\n'
+    # A [dotted.table] header right after an array-of-tables element nests
+    # under that most-recently-opened element, even though tree-sitter
+    # parses it as a sibling, not a child, of the [[items]] node.
+    '[items.detail]\ncolor = "red"\n'
+    "nested = { deep = { deeper = 5 } }\n"
+    "[[other.sub]]\nbar = 2\n"
+    "[[other.sub]]\nbar = 3\n"
+)
+
+_TOML_POINTERS = [
+    "",
+    "/top",
+    "/items",
+    "/items/0",
+    "/items/0/name",
+    "/items/1",
+    "/items/1/name",
+    "/items/1/detail",
+    "/items/1/detail/color",
+    "/items/1/detail/nested/deep/deeper",
+    "/items/1/detail/nonexistent",
+    "/items/99",
+    "/other/sub/0/bar",
+    "/other/sub/1/bar",
+    "/other/sub/2/bar",
+    "/nonexistent",
+    "/items/1/name/toofar",  # past a scalar pair value
+    "/items/1/detail/nested/deep/deeper/toofar",  # past a scalar inside an inline_table
+]
+
+# (fmt, doc, its own pointer list) -- the one place each format's example
+# document and matching pointers are paired up, so no test looks either up
+# by fmt at run time.
+_FMT_DOC_POINTERS = [
+    ("json", _JSON_DOC, _JSON_YAML_POINTERS),
+    ("yaml", _YAML_DOC, _JSON_YAML_POINTERS),
+    ("toml", _TOML_DOC, _TOML_POINTERS),
+]
+
+# Flattened to one (fmt, doc, pointer) row per pointer, for tests that check
+# one pointer per case rather than iterating a whole document's list.
+_FMT_DOC_POINTER = [
+    (fmt, doc, pointer)
+    for fmt, doc, pointers in _FMT_DOC_POINTERS
+    for pointer in pointers
+]
+
+# Grows over time: add a document here whenever a bug is found, so the
+# regression is covered by a full-map/locate_pointer comparison over every
+# pointer that document has, not just the specific pointer that was wrong.
+_EXAMPLE_DOCS = [(fmt, doc) for fmt, doc, _ in _FMT_DOC_POINTERS]
+
+
+@pytest.mark.parametrize("fmt,doc", _EXAMPLE_DOCS)
+def test_locate_pointer_matches_every_full_map_entry(fmt, doc):
+    """locate_pointer() must agree with build_source_map() + a direct lookup
+    (not just closest_entry()'s fallback) for every pointer the full map
+    actually has an entry for."""
+    full = build_source_map(doc, fmt=fmt)
+    for pointer, expected in full.items():
+        assert locate_pointer(doc, fmt, pointer) == expected, pointer
+
+
+@pytest.mark.parametrize("fmt,doc,pointer", _FMT_DOC_POINTER)
+def test_locate_pointer_matches_closest_entry(fmt, doc, pointer):
+    full = build_source_map(doc, fmt=fmt)
+    assert locate_pointer(doc, fmt, pointer) == closest_entry(full, pointer)
+
+
+@pytest.mark.parametrize("fmt", ["json", "toml", "yaml"])
+def test_locate_pointer_empty_document(fmt):
+    full = build_source_map("", fmt=fmt)
+    assert locate_pointer("", fmt, "/x") == closest_entry(full, "/x")
+
+
+def test_locate_pointer_unknown_format():
+    with pytest.raises(ValueError, match="Unknown format"):
+        locate_pointer("{}", "ini", "/x")
+
+
+@pytest.mark.parametrize("fmt,doc,pointers", _FMT_DOC_POINTERS)
+def test_source_map_matches_locate_pointer(fmt, doc, pointers):
+    sm = SourceMap(doc, fmt)
+    for pointer in pointers:
+        assert sm.locate(pointer) == locate_pointer(doc, fmt, pointer)
+
+
+def _submodule(fmt):
+    import importlib
+
+    return importlib.import_module(f"parse_errors.{fmt}_source_map")
+
+
+def _count_parse_calls(mod, monkeypatch):
+    real_parse = mod._parse
+    calls = []
+
+    def counting_parse(source):
+        calls.append(source)
+        return real_parse(source)
+
+    monkeypatch.setattr(mod, "_parse", counting_parse)
+    return calls
+
+
+@pytest.mark.parametrize("fmt,doc,pointers", _FMT_DOC_POINTERS)
+def test_source_map_parses_once(fmt, doc, pointers, monkeypatch):
+    calls = _count_parse_calls(_submodule(fmt), monkeypatch)
+
+    sm = SourceMap(doc, fmt)
+    for pointer in pointers:
+        sm.locate(pointer)
+
+    assert len(calls) == 1
+
+
+@pytest.mark.parametrize("fmt", ["json", "toml", "yaml"])
+def test_source_map_empty_document_parses_once(fmt, monkeypatch):
+    calls = _count_parse_calls(_submodule(fmt), monkeypatch)
+
+    full = build_source_map("", fmt=fmt)
+    sm = SourceMap("", fmt)
+    assert sm.locate("/x") == closest_entry(full, "/x")
+    assert sm.locate("/y") == closest_entry(full, "/y")
+
+    assert len(calls) == 1