feat: Add compiler warning framework with opt-in rich warning rendering

guppylang-internals/src/guppylang_internals/diagnostic.py

@@ -174,6 +174,14 @@ class Error(Diagnostic, Protocol):
     level: ClassVar[Literal[DiagnosticLevel.ERROR]] = DiagnosticLevel.ERROR
 
 
+@runtime_checkable
+@dataclass(frozen=True)
+class Warning(Diagnostic, Protocol):
+    """Compiler diagnostic for non-fatal warnings."""
+
+    level: ClassVar[Literal[DiagnosticLevel.WARNING]] = DiagnosticLevel.WARNING
+
+
 @runtime_checkable
 @dataclass(frozen=True)
 class Note(SubDiagnostic, Protocol):

guppylang-internals/src/guppylang_internals/error.py

@@ -106,7 +106,11 @@ def saved_exception_hook() -> Iterator[None]:
 
 
 def pretty_errors(f: FuncT) -> FuncT:
-    """Decorator to print custom error banners when a `GuppyError` occurs."""
+    """Decorator to print custom error banners when a `GuppyError` occurs.
+
+    This is also the standard boundary for warning collection on top-level engine
+    operations: wrapped calls participate in one `diagnostic_report()` session.
+    """
 
     def hook(
         excty: type[BaseException], err: BaseException, traceback: TracebackType | None
@@ -127,7 +131,9 @@ def hook(
 
     @functools.wraps(f)
     def pretty_errors_wrapped(*args: Any, **kwargs: Any) -> Any:
-        with exception_hook(hook):
+        from guppylang_internals.warning import diagnostic_report
+
+        with diagnostic_report(), exception_hook(hook):
             return f(*args, **kwargs)
 
     return cast("FuncT", pretty_errors_wrapped)

guppylang-internals/src/guppylang_internals/warning.py

@@ -0,0 +1,201 @@
+import sys
+import warnings
+from collections.abc import Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, NamedTuple
+
+from guppylang_internals.error import InternalGuppyError
+
+if TYPE_CHECKING:
+    from guppylang_internals.diagnostic import Diagnostic
+
+
+class GuppyWarning(UserWarning):
+    """Warning category for non-fatal compiler diagnostics."""
+
+
+class WarningKey(NamedTuple):
+    """Stable identity for deduplicating warnings within one operation."""
+
+    filename: str | None
+    lineno: int | None
+    column: int | None
+    message: str
+
+
+@dataclass(frozen=True)
+class PendingWarning:
+    """Buffered warning waiting to be emitted at the end of a top-level operation."""
+
+    diagnostic: "Diagnostic"
+    message: str
+    filename: str | None
+    lineno: int | None
+    key: WarningKey
+
+
+@dataclass
+class DiagnosticSession:
+    """Per-operation diagnostic state shared across nested compiler calls."""
+
+    rich_warnings: bool = False
+    pending_warnings: list[PendingWarning] = field(default_factory=list)
+    seen_warnings: set[WarningKey] = field(default_factory=set)
+
+
+_DIAGNOSTIC_SESSION: ContextVar[DiagnosticSession | None] = ContextVar(
+    "_DIAGNOSTIC_SESSION", default=None
+)
+_RICH_WARNINGS: ContextVar[bool] = ContextVar("_RICH_WARNINGS", default=False)
+
+
+@contextmanager
+def rich_warnings() -> Iterator[None]:
+    """Enable rich stderr rendering for compiler warnings within the current scope."""
+
+    token = _RICH_WARNINGS.set(True)
+    try:
+        yield
+    finally:
+        _RICH_WARNINGS.reset(token)
+
+
+@contextmanager
+def diagnostic_report() -> Iterator[None]:
+    """Collects compiler warnings and flushes them once per top-level operation."""
+
+    session = _DIAGNOSTIC_SESSION.get()
+    # Nested compiler entrypoints reuse the same session so one user operation only
+    # flushes once, at the outermost boundary.
+    outermost = session is None
+    token = None
+    if outermost:
+        session = DiagnosticSession(rich_warnings=_RICH_WARNINGS.get())
+        token = _DIAGNOSTIC_SESSION.set(session)
+    assert session is not None
+
+    try:
+        yield
+    except Exception:
+        if outermost:
+            # Failed operations should not emit queued warnings. Clear eagerly so the
+            # exception path behaves the same whether the warning producer ran before
+            # or after the eventual failure.
+            session.pending_warnings.clear()
+            session.seen_warnings.clear()
+        raise
+    else:
+        if outermost:
+            # Only the outermost context flushes to Python warnings. Inner contexts
+            # merely contribute to the shared session.
+            for pending_warning in session.pending_warnings:
+                _emit_pending_warning(pending_warning)
+    finally:
+        if outermost and token is not None:
+            # Restore the previous ContextVar value even if warning emission itself
+            # raises, so subsequent compiler operations start with a clean session.
+            _DIAGNOSTIC_SESSION.reset(token)
+
+
+def emit_warning(diag: "Diagnostic") -> None:
+    """Queue or emit a non-fatal compiler warning."""
+
+    pending_warning = _pending_warning(diag)
+    session = _DIAGNOSTIC_SESSION.get()
+    if session is None:
+        # Warnings emitted outside a diagnostic_report block still surface
+        # immediately; the session machinery is only needed for batching and
+        # deduplicating within top-level compiler operations.
+        _emit_pending_warning(pending_warning)
+        return
+
+    if pending_warning.key in session.seen_warnings:
+        # Re-emitting the same warning from nested passes or revisited CFG nodes should
+        # not duplicate the user-facing Python warning within one operation.
+        return
+
+    session.seen_warnings.add(pending_warning.key)
+    session.pending_warnings.append(pending_warning)
+
+
+def _pending_warning(diag: "Diagnostic") -> PendingWarning:
+    from guppylang_internals.diagnostic import DiagnosticLevel
+    from guppylang_internals.span import to_span
+
+    if diag.level is not DiagnosticLevel.WARNING:
+        raise InternalGuppyError("emit_warning expects a warning-level diagnostic")
+
+    filename = None
+    lineno = None
+    column = None
+    if diag.span is not None:
+        # Python's warning machinery wants file/line information separately rather
+        # than Guppy's richer span object.
+        span = to_span(diag.span)
+        filename = span.start.file
+        lineno = span.start.line
+        column = span.start.column
+
+    message = _warning_message(diag)
+    return PendingWarning(
+        diagnostic=diag,
+        message=message,
+        filename=filename,
+        lineno=lineno,
+        # Deduplicate on source location plus rendered message so repeated reports from
+        # the same site collapse, while distinct warnings on one line still survive.
+        key=WarningKey(filename, lineno, column, message),
+    )
+
+
+def _emit_pending_warning(pending_warning: PendingWarning) -> None:
+    """Emit one queued warning via Python's warning machinery and rich stderr output."""
+
+    if pending_warning.filename is not None and pending_warning.lineno is not None:
+        warnings.warn_explicit(
+            pending_warning.message,
+            GuppyWarning,
+            pending_warning.filename,
+            pending_warning.lineno,
+        )
+    else:
+        warnings.warn(
+            pending_warning.message,
+            GuppyWarning,
+            stacklevel=2,
+        )
+
+    session = _DIAGNOSTIC_SESSION.get()
+    if session is not None and session.rich_warnings:
+        sys.stderr.write(_render_warning(pending_warning))
+        sys.stderr.write("\n")
+
+
+def _render_warning(pending_warning: PendingWarning) -> str:
+    from guppylang_internals.diagnostic import DiagnosticsRenderer
+    from guppylang_internals.engine import DEF_STORE
+
+    renderer = DiagnosticsRenderer(DEF_STORE.sources)
+    try:
+        renderer.render_diagnostic(pending_warning.diagnostic)
+    except KeyError:
+        return pending_warning.message
+    return "\n".join(renderer.buffer)
+
+
+def _warning_message(diag: "Diagnostic") -> str:
+    lines = [diag.rendered_title]
+    if diag.rendered_span_label:
+        lines[0] += f": {diag.rendered_span_label}"
+    if diag.rendered_message:
+        lines.append(diag.rendered_message)
+    lines.extend(
+        [
+            f"{child.level.name.lower().capitalize()}: {child.rendered_message}"
+            for child in diag.children
+            if child.rendered_message
+        ]
+    )
+    return "\n".join(lines)

guppylang/src/guppylang/__init__.py

@@ -1,4 +1,5 @@
 from guppylang_internals.experimental import enable_experimental_features
+from guppylang_internals.warning import GuppyWarning, rich_warnings
 
 from guppylang.decorator import guppy
 from guppylang.module import GuppyModule
@@ -8,6 +9,7 @@
 
 __all__ = (
     "GuppyModule",
+    "GuppyWarning",
     "array",
     "builtins",
     "comptime",
@@ -17,6 +19,7 @@
     "py",
     "quantum",
     "qubit",
+    "rich_warnings",
 )
 
 # This is updated by our release-please workflow, triggered by this

guppylang/src/guppylang/defs.py

@@ -22,6 +22,7 @@
     TracingDefMixin,
 )
 from guppylang_internals.tracing.util import hide_trace
+from guppylang_internals.warning import diagnostic_report
 from hugr.envelope import GeneratorDesc
 from hugr.hugr import Hugr
 from hugr.metadata import HugrGenerator
@@ -85,13 +86,17 @@ class GuppyDefinition(TracingDefMixin):
 
     def compile(self) -> Package:
         """Compile a Guppy definition to HUGR."""
+        # Single-definition entrypoints rely on the wrapped engine helpers for warning
+        # collection. `ENGINE.compile_single()` already establishes the top-level
+        # diagnostic session via `@pretty_errors`.
         package: Package = ENGINE.compile_single(self.id).package
         for mod in package.modules:
             _update_generator_metadata(mod)
         return package
 
     def check(self) -> None:
         """Type-check a Guppy definition."""
+        # As above, warning collection is handled by the wrapped engine entrypoint.
         return ENGINE.check_single(self.id)
 
 
@@ -243,7 +248,6 @@ def compile_entrypoint(self) -> Package:
     def compile_function(self) -> Package:
         """Compile a Guppy function definition to HUGR.
 
-
         Returns:
             Package: The compiled package object.
         """
@@ -276,18 +280,26 @@ def _type_members(self) -> list[DefId]:
 
     def compile(self) -> Package:
         """Compile this collection of definitions into a HUGR package."""
-        ENGINE.check(self.members)
-        # Check fills _type_members with additional members only available after
-        # checking, so we have to call it before compiling (without an engine reset).
-        pointer = ENGINE.compile(self.members + self._type_members(), reset=False)
+        # Unlike the single-definition helpers, a library compile spans multiple
+        # top-level engine calls. Keep one outer diagnostic session here so warnings
+        # flush once for the whole user operation rather than once per engine call.
+        with diagnostic_report():
+            ENGINE.check(self.members)
+            # Check fills _type_members with additional members only available after
+            # checking, so we have to call it before compiling (without an engine
+            # reset).
+            pointer = ENGINE.compile(self.members + self._type_members(), reset=False)
         for mod in pointer.package.modules:
             _update_generator_metadata(mod)
         return pointer.package
 
     def check(self) -> None:
         """Type-check all contained definitions."""
-        ENGINE.check(self.members)
-        ENGINE.check(self._type_members(), reset=False)
+        # Library checks can trigger more than one top-level engine check, so they need
+        # their own outer diagnostic session.
+        with diagnostic_report():
+            ENGINE.check(self.members)
+            ENGINE.check(self._type_members(), reset=False)
 
 
 @dataclass(frozen=True)