Skip to content

feat(constraints): add per-specifier provenance tracking #1189

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
smoparth wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(constraints): add per-specifier provenance tracking #1189

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 74 additions & 13 deletions src/fromager/constraints.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,23 @@ def _is_blocked_specifier(specifier: SpecifierSet) -> bool:
)


def _format_provenance(provenance: dict[str, list[str]]) -> str:
"""Format a per-package provenance dict as a human-readable string.

Args:
provenance: Mapping of source file to original constraint lines.

Returns:
Formatted string, e.g.
``"/path/to/base.txt (>=2.0), /path/to/override.txt (!=2.1.1)"``.
"""
parts: list[str] = []
for source, lines in provenance.items():
specifiers = ", ".join(str(Requirement(line).specifier) for line in lines)
parts.append(f"{source} ({specifiers})")
return ", ".join(parts)


class InvalidConstraintError(ValueError):
pass

Expand All @@ -36,6 +53,8 @@ def __init__(self) -> None:
# mapping of canonical names to requirements
# NOTE: Requirement.name is not normalized
self._data: dict[NormalizedName, Requirement] = {}
# per-package provenance: {canonical_name: {source_file: [original_lines]}}
self._provenance: dict[NormalizedName, dict[str, list[str]]] = {}

def __iter__(self) -> Generator[NormalizedName, None, None]:
yield from self._data
Expand All @@ -46,13 +65,21 @@ def __bool__(self) -> bool:
def __len__(self) -> int:
return len(self._data)

def add_constraint(self, unparsed: str) -> None:
"""Add new constraint, must not conflict with any existing constraints
def add_constraint(self, unparsed: str, *, source: str) -> None:
"""Add new constraint, must not conflict with any existing constraints.

Args:
unparsed: Raw constraint string, e.g. ``"foo>=2.0"``.
source: Path or URL of the file that contains this constraint.
Required for provenance tracking.

.. versionchanged: 0.83.0
.. versionchanged:: 0.83.0
Non-conflicting constraints are now combined. Constraints with
conflicts now raise :exc:`InvalidConstraintError`. Inputs without a
version specifier or with extras or url are also refused.

.. versionchanged:: 0.84.0
Added *source* parameter for provenance tracking.
"""
req = Requirement(unparsed)
canon_name = canonicalize_name(req.name)
Expand Down Expand Up @@ -81,43 +108,76 @@ def add_constraint(self, unparsed: str) -> None:
if previous is not None:
prev_blocked = _is_blocked_specifier(previous.specifier)
if blocked != prev_blocked:
prev_prov = _format_provenance(self._provenance.get(canon_name, {}))
raise InvalidConstraintError(
f"Cannot combine blocked and non-blocked constraints "
f"(existing: {previous}, new: {req})"
f"(existing: {previous} from {prev_prov}, "
f"new: {req} from {source})"
)
if not blocked:
logger.debug("combining constraints %s and %s", previous, req)
new_specifier = req.specifier & previous.specifier
if new_specifier.is_unsatisfiable():
prev_prov = _format_provenance(self._provenance.get(canon_name, {}))
raise InvalidConstraintError(
f"Combined specifier '{new_specifier}' is not satisfiable "
f"(existing: {previous}, new: {req})"
f"(existing: {previous} from {prev_prov}, "
f"new: {req} from {source})"
)
req.specifier = new_specifier
else:
logger.debug(f"adding constraint {req}")

self._data[canon_name] = req
pkg_prov = self._provenance.setdefault(canon_name, {})
pkg_prov.setdefault(source, []).append(unparsed)

def load_constraints_file(self, constraints_file: str | pathlib.Path) -> None:
"""Load constraints from a constraints file or URL"""
"""Load constraints from a constraints file or URL."""
logger.info("loading constraints from %s", constraints_file)
source = str(constraints_file)
content = requirements_file.parse_requirements_file(constraints_file)
for line in content:
self.add_constraint(line)
self.add_constraint(line, source=source)

def dump_constraints(self, output: typing.TextIO) -> None:
"""Dump combined constraints to a text stream"""
# sort by normalized name
for _, req in sorted(self._data.items()):
# write requirement without markers. They have been evaluated
# in add_constraint()
output.write(f"{req.name}{req.specifier}\n")
"""Dump combined constraints to a text stream.

Each line includes an inline comment showing which source file(s)
contributed each specifier.

Args:
output: Writable text stream.

.. versionchanged:: 0.84.0
Output now includes per-line provenance comments.
"""
# sort by normalized name, write requirement without markers.
# They have been evaluated in add_constraint()
for name, req in sorted(self._data.items()):
line = f"{req.name}{req.specifier}"
prov = self._provenance.get(name, {})
if prov:
line = f"{line} # {_format_provenance(prov)}"
output.write(f"{line}\n")

def get_constraint(self, name: str) -> Requirement | None:
"""Return the merged constraint for *name*, or ``None``."""
return self._data.get(canonicalize_name(name))

def get_provenance(self, name: str) -> dict[str, list[str]]:
"""Return provenance info for *name*.

Returns:
Mapping of ``{source_file: [original_constraint_lines]}``,
or an empty dict if the package has no constraints.

.. versionadded:: 0.84.0
"""
return dict(self._provenance.get(canonicalize_name(name), {}))
Comment on lines +168 to +177
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot Jun 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Return a deep copy from get_provenance() to prevent internal state mutation.

The current return value is a shallow copy; callers can mutate nested lists and silently alter self._provenance, which then changes dump_constraints() output and provenance shown in exceptions/logs.

Suggested fix 
     def get_provenance(self, name: str) -> dict[str, list[str]]:
@@
-        return dict(self._provenance.get(canonicalize_name(name), {}))
+        prov = self._provenance.get(canonicalize_name(name), {})
+        return {source: list(lines) for source, lines in prov.items()}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def get_provenance(self, name: str) -> dict[str, list[str]]:
"""Return provenance info for *name*.
Returns:
Mapping of ``{source_file: [original_constraint_lines]}``,
or an empty dict if the package has no constraints.
.. versionadded:: 0.84.0
"""
return dict(self._provenance.get(canonicalize_name(name), {}))
def get_provenance(self, name: str) -> dict[str, list[str]]:
"""Return provenance info for *name*.
Returns:
Mapping of ``{source_file: [original_constraint_lines]}``,
or an empty dict if the package has no constraints.
.. versionadded:: 0.84.0
"""
prov = self._provenance.get(canonicalize_name(name), {})
return {source: list(lines) for source, lines in prov.items()}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/fromager/constraints.py` around lines 168 - 177, get_provenance currently
returns a shallow copy of self._provenance so callers can mutate nested lists
and change internal state; update get_provenance to return a deep copy of the
provenance mapping (e.g., use copy.deepcopy on
self._provenance[canonicalize_name(name)] or construct a new dict copying each
list) so callers cannot modify internal data structures accessed via
get_provenance; ensure you still canonicalize the name with
canonicalize_name(name) and return an empty dict when missing.


def allow_prerelease(self, pkg_name: str) -> bool:
"""Return ``True`` if the constraint for *pkg_name* allows prereleases."""
constraint = self.get_constraint(pkg_name)
if constraint:
return bool(constraint.specifier.prereleases)
Expand All @@ -131,6 +191,7 @@ def is_blocked(self, pkg_name: str) -> bool:
return False

def is_satisfied_by(self, pkg_name: str, version: Version) -> bool:
"""Return ``True`` if *version* satisfies the constraint for *pkg_name*."""
constraint = self.get_constraint(pkg_name)
if constraint:
return constraint.specifier.contains(version, prereleases=True)
Expand Down
16 changes: 13 additions & 3 deletions src/fromager/resolver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@

from . import overrides
from .candidate import Candidate, Cooldown
from .constraints import Constraints
from .constraints import Constraints, _format_provenance
from .extras_provider import ExtrasProvider
from .http_retry import RETRYABLE_EXCEPTIONS, retry_on_exception
from .request_session import session
Expand Down Expand Up @@ -287,10 +287,15 @@ def find_all_matching_from_provider(
)
except resolvelib.resolvers.ResolverException as err:
constraint = provider.constraints.get_constraint(req.name)
provenance = provider.constraints.get_provenance(req.name)
provider_desc = provider.get_provider_description()
original_msg = str(err)
prov_msg = ""
if provenance:
prov_msg = f" (from {_format_provenance(provenance)})"
raise resolvelib.resolvers.ResolverException(
f"Unable to resolve requirement specifier {req} with constraint {constraint} using {provider_desc}: {original_msg}"
f"Unable to resolve requirement specifier {req} with constraint "
f"{constraint}{prov_msg} using {provider_desc}: {original_msg}"
) from err

# Materialize candidates so we can iterate more than once if filtering
Expand Down Expand Up @@ -689,8 +694,13 @@ def is_satisfied_by(self, requirement: Requirement, candidate: Candidate) -> boo
if not self.constraints.is_satisfied_by(requirement.name, candidate.version):
if DEBUG_RESOLVER:
c = self.constraints.get_constraint(requirement.name)
provenance = self.constraints.get_provenance(requirement.name)
prov_msg = ""
if provenance:
prov_msg = f" from {_format_provenance(provenance)}"
logger.debug(
f"{requirement.name}: skipping {candidate.version} due to constraint {c}"
f"{requirement.name}: skipping {candidate.version} "
f"due to constraint {c}{prov_msg}"
)
return False

Expand Down
Loading
Loading