diff --git a/mypyc/irbuild/statement.py b/mypyc/irbuild/statement.py
index d60de0d90e1c..b20022c1ce8b 100644
--- a/mypyc/irbuild/statement.py
+++ b/mypyc/irbuild/statement.py
@@ -49,6 +49,7 @@
     YieldExpr,
     YieldFromExpr,
 )
+from mypy.types import Instance, get_proper_type
 from mypyc.common import (
     GENERATOR_HELPER_NAME,
     KEEP_ALIVE_SHORT_LIVED,
@@ -114,6 +115,10 @@
     AssignmentTargetTuple,
 )
 from mypyc.primitives.exc_ops import (
+    contextmanager_bind_op,
+    contextmanager_enter_op,
+    contextmanager_exit_exception_op,
+    contextmanager_exit_normal_op,
     error_catch_op,
     exc_matches_op,
     get_exc_info_op,
@@ -1125,10 +1130,36 @@ def transform_with(
     is_async: bool,
     line: int,
 ) -> None:
+
+    if isinstance(expr, mypy.nodes.CallExpr) and isinstance(expr.callee, mypy.nodes.RefExpr):
+        callee = expr.callee
+        ctm_def_node: mypy.nodes.SymbolNode | None = callee.node
+        self_expr: Expression | None = None
+        if isinstance(callee, mypy.nodes.MemberExpr) and not builder.is_module_member_expr(callee):
+            ctm_def_node = None
+            # Since we construct the call to the method ourselves in _transform_with_contextmanager,
+            # we're a little paranoid about doing things in the wrong order in weird cases
+            if not find_walrus_targets(expr) and not expr_has_suspend(expr):
+                self_type = get_proper_type(builder.types[callee.expr])
+                if isinstance(self_type, Instance):
+                    symbol = self_type.type.get(callee.name)
+                    if symbol is not None:
+                        ctm_def_node = symbol.node
+                        self_expr = callee.expr
+
+        if (
+            not is_async
+            and isinstance(dec := ctm_def_node, mypy.nodes.Decorator)
+            and not (self_expr is not None and (dec.func.is_static or dec.func.is_class))
+            and len(dec.decorators) == 1
+            and isinstance(dec1 := dec.decorators[0], mypy.nodes.RefExpr)
+            and dec1.node
+            and dec1.node.fullname == "contextlib.contextmanager"
+        ):
+            return _transform_with_contextmanager(builder, expr, target, body, line, self_expr)
+
     # This is basically a straight transcription of the Python code in PEP 343.
     # I don't actually understand why a bunch of it is the way it is.
-    # We could probably optimize the case where the manager is compiled by us,
-    # but that is not our common case at all, so.
 
     al = "a" if is_async else ""
 
@@ -1136,7 +1167,7 @@ def transform_with(
 
     if not is_async and mgr_v.type == lock_rprimitive:
         transform_with_lock(builder, mgr_v, target, body, line)
-        return
+        return None
     # The methods may be inherited from a non-native class, and then they can't be
     # called natively.
     is_native = (
@@ -1240,6 +1271,87 @@ def finally_body() -> None:
     transform_try_finally_stmt(builder, try_body, finally_body, line)
 
 
+def _transform_with_contextmanager(
+    builder: IRBuilder,
+    expr: mypy.nodes.CallExpr,
+    target: Lvalue | None,
+    with_body: GenFunc,
+    line: int,
+    receiver_expr: Expression | None,
+) -> None:
+
+    # wrapped_call = {expr.callee}.__wrapped__(*args, **kwargs)
+    # where {expr.callee} is either a function or a method
+
+    assert isinstance(expr.callee, mypy.nodes.RefExpr)
+    if receiver_expr is None:
+        wrapped_call = mypy.nodes.CallExpr(
+            mypy.nodes.MemberExpr(expr.callee, "__wrapped__"),
+            expr.args,
+            expr.arg_kinds,
+            expr.arg_names,
+        )
+        wrapped_call.line = line
+        gen = builder.accept(wrapped_call)
+    else:
+        assert isinstance(expr.callee, mypy.nodes.MemberExpr)
+        receiver = builder.accept(receiver_expr)
+        method = builder.py_get_attr(receiver, expr.callee.name, line)
+        wrapped = builder.py_get_attr(method, "__wrapped__", line)
+        wrapped = builder.call_c(contextmanager_bind_op, [method, wrapped], line)
+        args = [builder.accept(arg) for arg in expr.args]
+        gen = builder.py_call(wrapped, args, line, expr.arg_kinds, expr.arg_names)
+
+    # try:
+    #     mgr_target = next(gen)
+    # except StopIteration:
+    #     raise RuntimeError("generator didn't yield") from None
+    mgr_target = builder.call_c(contextmanager_enter_op, [gen], line)
+
+    # try:
+    #     {body}
+
+    def try_body() -> None:
+        if target:
+            builder.assign(builder.get_assignment_target(target), mgr_target, line)
+        with_body()
+
+    # except BaseException as e:
+    #     {contextmanager_exit_exception_op}
+    # finally:
+    #     {contextmanager_exit_normal_op}
+
+    exc = builder.ensure_register(builder.true())
+
+    def except_body() -> None:
+        builder.assign(exc, builder.false(), line)
+        suppressed = builder.call_c(
+            contextmanager_exit_exception_op, [builder.read(gen), *get_sys_exc_info(builder)], line
+        )
+        out_block, reraise_block = BasicBlock(), BasicBlock()
+        builder.add_bool_branch(suppressed, out_block, reraise_block)
+        builder.activate_block(reraise_block)
+        builder.call_c(reraise_exception_op, [], NO_TRACEBACK_LINE_NO)
+        builder.add(Unreachable())
+        builder.activate_block(out_block)
+
+    handlers = [(None, None, except_body)]
+
+    def finally_body() -> None:
+        out_block, exit_block = BasicBlock(), BasicBlock()
+        builder.add(Branch(builder.read(exc), exit_block, out_block, Branch.BOOL))
+        builder.activate_block(exit_block)
+        builder.call_c(contextmanager_exit_normal_op, [builder.read(gen)], line)
+        builder.goto_and_activate(out_block)
+
+    transform_try_finally_stmt(
+        builder,
+        lambda: transform_try_except(builder, try_body, handlers, None, line),
+        finally_body,
+        line,
+    )
+
+
 def transform_with_stmt(builder: IRBuilder, o: WithStmt) -> None:
     # Generate separate logic for each expr in it, left to right
     def generate(i: int) -> None:
diff --git a/mypyc/lib-rt/CPy.h b/mypyc/lib-rt/CPy.h
index dfb7e43ed769..94cc0960050e 100644
--- a/mypyc/lib-rt/CPy.h
+++ b/mypyc/lib-rt/CPy.h
@@ -1071,6 +1071,11 @@ bool CPy_ExceptionMatches(PyObject *type);
 PyObject *CPy_GetExcValue(void);
 tuple_T3OOO CPy_GetExcInfo(void);
 void _CPy_GetExcInfo(PyObject **p_type, PyObject **p_value, PyObject **p_traceback);
+PyObject *CPyContextManager_Bind(PyObject *callable, PyObject *wrapped);
+PyObject *CPyContextManager_Enter(PyObject *gen);
+bool CPyContextManager_ExitNormal(PyObject *gen);
+int CPyContextManager_ExitException(PyObject *gen, PyObject *type,
+                                  PyObject *value, PyObject *traceback);
 void CPyError_OutOfMemory(void);
 void CPy_TypeError(const char *expected, PyObject *value);
 void CPy_AddTraceback(const char *filename, const char *funcname, int line, PyObject *globals);
diff --git a/mypyc/lib-rt/exc_ops.c b/mypyc/lib-rt/exc_ops.c
index f6a79fa8197a..e401f2be8f98 100644
--- a/mypyc/lib-rt/exc_ops.c
+++ b/mypyc/lib-rt/exc_ops.c
@@ -163,6 +163,206 @@ tuple_T3OOO CPy_GetExcInfo(void) {
     return ret;
 }
 
+// Give a contextmanager's generator factory the same binding as its wrapper.
+// The returned callable is owned; static methods remain unbound.
+PyObject *CPyContextManager_Bind(PyObject *callable, PyObject *wrapped) {
+    if (PyMethod_Check(callable)) {
+        // Use the actual bound instance or class, not the attribute lookup's receiver.
+        return PyMethod_New(wrapped, PyMethod_GET_SELF(callable));
+    }
+    Py_INCREF(wrapped);
+    return wrapped;
+}
+
+// Implement contextlib's generator entry protocol. The returned value is owned.
+PyObject *CPyContextManager_Enter(PyObject *gen) {
+    if (!PyIter_Check(gen)) {
+        PyErr_Format(PyExc_TypeError, "'%.200s' object is not an iterator", Py_TYPE(gen)->tp_name);
+        return NULL;
+    }
+    PyObject *value = Py_TYPE(gen)->tp_iternext(gen);
+    if (value != NULL) {
+        return value;
+    }
+    if (!PyErr_Occurred()) {
+        // Iterators may signal exhaustion without setting StopIteration.
+        PyErr_SetNone(PyExc_StopIteration);
+    }
+    if (PyErr_ExceptionMatches(PyExc_StopIteration)) {
+        // raise RuntimeError(...) from None, retaining the suppressed context.
+        tuple_T3OOO old_exc = CPy_CatchError();
+        PyObject *error = PyObject_CallFunction(PyExc_RuntimeError, "s", "generator didn't yield");
+        if (error != NULL) {
+            CPy_RaiseFrom(error, Py_None);
+            Py_DECREF(error);
+        }
+        CPy_RestoreExcInfo(old_exc);
+        Py_DECREF(old_exc.f0);
+        Py_DECREF(old_exc.f1);
+        Py_DECREF(old_exc.f2);
+    }
+    return NULL;
+}
+
+// An extra yield is a protocol error. Always leave an exception pending.
+static void contextmanager_exit_error(PyObject *gen, const char *message) {
+    PyErr_SetString(PyExc_RuntimeError, message);
+#if PY_VERSION_HEX >= 0x030B0000
+    // Since Python 3.11, contextlib closes the generator in a finally block
+    tuple_T3OOO old_exc = CPy_CatchError();
+    PyObject *type, *value, *traceback;
+    PyErr_GetExcInfo(&type, &value, &traceback);
+    PyObject *closed = PyObject_CallMethod(gen, "close", NULL);
+    if (closed != NULL) {
+        Py_DECREF(closed);
+        // A native generator can change the handled exception during close()
+        // Reraise the saved protocol error, as a Python finally block does
+        PyErr_Restore(type, value, traceback);
+    } else {
+        Py_XDECREF(type);
+        Py_XDECREF(value);
+        Py_XDECREF(traceback);
+    }
+    CPy_RestoreExcInfo(old_exc);
+    Py_DECREF(old_exc.f0);
+    Py_DECREF(old_exc.f1);
+    Py_DECREF(old_exc.f2);
+#endif
+}
+
+// Finish a contextmanager whose body succeeded. Return false on error.
+bool CPyContextManager_ExitNormal(PyObject *gen) {
+    // try:
+    //     next(self.gen)
+    if (!PyIter_Check(gen)) {
+        PyErr_Format(PyExc_TypeError, "'%.200s' object is not an iterator", Py_TYPE(gen)->tp_name);
+        return false;
+    }
+    PyObject *value = Py_TYPE(gen)->tp_iternext(gen);
+
+    // except StopIteration:
+    //     return False
+    if (value == NULL) {
+        if (!PyErr_Occurred() || PyErr_ExceptionMatches(PyExc_StopIteration)) {
+            PyErr_Clear();
+            return true;
+        }
+        return false;
+    }
+
+    // else:
+    //     try:
+    //         raise RuntimeError("generator didn't stop")
+    //     finally:
+    //         self.gen.close()
+    Py_DECREF(value);
+    contextmanager_exit_error(gen, "generator didn't stop");
+    return false;
+}
+
+// Match contextlib._GeneratorContextManager.__exit__ for a failed with body.
+int CPyContextManager_ExitException(PyObject *gen, PyObject *type,
+                                    PyObject *value, PyObject *traceback) {
+
+    // Native generators can change the handled exception when throw() unwinds
+    // an except block. Preserve the caller's state before interpreting the result
+    PyObject *saved_type, *saved_value, *saved_traceback;
+    PyErr_GetExcInfo(&saved_type, &saved_value, &saved_traceback);
+
+    // try:
+    //     self.gen.throw(typ, value, traceback)
+#if PY_VERSION_HEX >= 0x030C0000
+    PyObject *yielded = PyObject_CallMethod(gen, "throw", "(O)", value);
+#else
+    PyObject *yielded = PyObject_CallMethod(gen, "throw", "(OOO)", type, value, traceback);
+#endif
+    PyErr_SetExcInfo(saved_type, saved_value, saved_traceback);
+
+    // ... if we didn't get an exception from throw, in particular, StopIteration ...
+    if (yielded != NULL) {
+        // try:
+        //     raise RuntimeError("generator didn't stop after throw()")
+        // finally:
+        //     self.gen.close()
+        Py_DECREF(yielded);
+        contextmanager_exit_error(gen, "generator didn't stop after throw()");
+        return -1;
+    }
+
+    tuple_T3OOO old_exc = CPy_CatchError();
+    PyObject *error = CPy_GetExcValue();
+    int result = -1;
+
+    // # Note the `return False` in the code comments is how contextlib tells Python to
+    // # propagate the exception being handled by __exit__
+
+    // except StopIteration as exc:
+    //     return exc is not value
+    if (CPy_ExceptionMatches(PyExc_StopIteration)) {
+        // A body StopIteration must not be mistaken for generator completion.
+        result = error != value;
+        goto done;
+    }
+    // except RuntimeError as exc:
+    //     if exc is value:
+    //         exc.__traceback__ = traceback
+    //         return False
+    //     if (
+    //         isinstance(value, StopIteration)
+    //         and exc.__cause__ is value
+    //     ):
+    //         value.__traceback__ = traceback
+    //         return False
+    //     raise
+    if (CPy_ExceptionMatches(PyExc_RuntimeError)) {
+        if (error == value) {
+            goto original;
+        }
+        // PEP 479 can wrap the original StopIteration (including subclasses).
+        int is_stop = PyObject_IsInstance(value, PyExc_StopIteration);
+        if (is_stop < 0) {
+            goto done;
+        }
+        if (is_stop) {
+            PyObject *cause = PyObject_GetAttrString(error, "__cause__");
+            if (cause == NULL) {
+                goto done;
+            }
+            bool is_original = cause == value;
+            Py_DECREF(cause);
+            if (is_original) {
+                goto original;
+            }
+        }
+    }
+    // except BaseException as exc:
+    //     if exc is not value:
+    //         raise
+    //     exc.__traceback__ = traceback
+    //     return False
+    else if (error == value) {
+        goto original;
+    }
+    CPy_Reraise();
+    goto done;
+
+original:
+#if PY_VERSION_HEX >= 0x030B0000
+    if (PyObject_SetAttrString(value, "__traceback__", traceback) < 0) {
+        goto done;
+    }
+#endif
+    result = 0;
+
+done:
+    Py_DECREF(error);
+    CPy_RestoreExcInfo(old_exc);
+    Py_DECREF(old_exc.f0);
+    Py_DECREF(old_exc.f1);
+    Py_DECREF(old_exc.f2);
+    return result;
+}
+
 void CPyError_OutOfMemory(void) {
     fprintf(stderr, "fatal: out of memory\n");
     fflush(stderr);
diff --git a/mypyc/primitives/exc_ops.py b/mypyc/primitives/exc_ops.py
index 528a89ef2331..39b9c382345b 100644
--- a/mypyc/primitives/exc_ops.py
+++ b/mypyc/primitives/exc_ops.py
@@ -2,9 +2,50 @@
 
 from __future__ import annotations
 
-from mypyc.ir.ops import ERR_ALWAYS, ERR_FALSE, ERR_NEVER
-from mypyc.ir.rtypes import bit_rprimitive, exc_rtuple, object_rprimitive, void_rtype
-from mypyc.primitives.registry import custom_op, custom_primitive_op
+from mypyc.ir.ops import ERR_ALWAYS, ERR_FALSE, ERR_MAGIC, ERR_NEVER
+from mypyc.ir.rtypes import (
+    bit_rprimitive,
+    c_int_rprimitive,
+    exc_rtuple,
+    object_rprimitive,
+    void_rtype,
+)
+from mypyc.primitives.registry import ERR_NEG_INT, custom_op, custom_primitive_op
+
+# Give a contextmanager's generator factory the same runtime method binding.
+contextmanager_bind_op = custom_op(
+    arg_types=[object_rprimitive, object_rprimitive],
+    return_type=object_rprimitive,
+    c_function_name="CPyContextManager_Bind",
+    error_kind=ERR_MAGIC,
+)
+
+# Enter a @contextmanager generator, requiring it to yield a value.
+contextmanager_enter_op = custom_op(
+    arg_types=[object_rprimitive],
+    return_type=object_rprimitive,
+    c_function_name="CPyContextManager_Enter",
+    error_kind=ERR_MAGIC,
+)
+
+# Finish a @contextmanager generator after a successful with body.
+# Returns true on success; generator/protocol errors propagate normally.
+contextmanager_exit_normal_op = custom_op(
+    arg_types=[object_rprimitive],
+    return_type=bit_rprimitive,
+    c_function_name="CPyContextManager_ExitNormal",
+    error_kind=ERR_FALSE,
+)
+
+# Finish a contextmanager after a body exception. True suppresses the exception;
+# false leaves the original handled exception for the caller to reraise.
+contextmanager_exit_exception_op = custom_op(
+    arg_types=[object_rprimitive] * 4,
+    return_type=c_int_rprimitive,
+    c_function_name="CPyContextManager_ExitException",
+    error_kind=ERR_NEG_INT,
+    truncated_type=bit_rprimitive,
+)
 
 # If the argument is a class, raise an instance of the class. Otherwise, assume
 # that the argument is an exception object, and raise it.
@@ -88,6 +129,11 @@
     arg_types=[], return_type=exc_rtuple, c_function_name="CPy_CatchError", error_kind=ERR_NEVER
 )
 
+# Clear the current exception.
+error_clear_op = custom_op(
+    arg_types=[], return_type=void_rtype, c_function_name="PyErr_Clear", error_kind=ERR_NEVER
+)
+
 # Restore an old "currently handled exception" returned from.
 # error_catch (by sticking it into sys.exc_info())
 restore_exc_info_op = custom_op(
diff --git a/mypyc/test-data/run-functions.test b/mypyc/test-data/run-functions.test
index bb52eeb6d192..3e3801b2e8d7 100644
--- a/mypyc/test-data/run-functions.test
+++ b/mypyc/test-data/run-functions.test
@@ -1309,6 +1309,672 @@ def test_special_case() -> None:
     with f():
         a.pop()
 
+[case testContextManagerSpecialCaseSemantics]
+from contextlib import contextmanager
+from typing import Any, Generator, Iterator, cast
+import traceback
+
+@contextmanager
+def cm_no_yield() -> Iterator[None]:
+    print("enter cm_no_yield")
+    if False:
+        print("UNREACHABLE")
+        yield
+    return
+
+def test_no_yield() -> None:
+    try:
+        with cm_no_yield():
+            print("UNREACHABLE")
+        print("UNREACHABLE")
+    except RuntimeError as e:
+        e_any = cast(Any, e)
+        print(
+            type(e).__name__,
+            str(e),
+            e_any.__cause__ is None,
+            e_any.__suppress_context__,
+        )
+
+@contextmanager
+def cm_no_stop() -> Iterator[str]:
+    print("enter cm_no_stop")
+    yield "first"
+    print("after first yield")
+    yield "second"
+    print("UNREACHABLE")
+
+def test_no_stop() -> None:
+    try:
+        with cm_no_stop() as v:
+            print("entered body", v)
+        print("UNREACHABLE")
+    except RuntimeError as e:
+        print(type(e).__name__, str(e))
+
+@contextmanager
+def cm_value_error_suppress() -> Iterator[None]:
+    print("enter cm_value_error_suppress")
+    try:
+        yield
+        print("UNREACHABLE")
+    except ValueError as e:
+        print(type(e).__name__, str(e))
+
+def test_suppress_value_error() -> None:
+    with cm_value_error_suppress():
+        raise ValueError("boom")
+    print("after suppressing the ValueError")
+
+@contextmanager
+def cm_throw_twice() -> Iterator[None]:
+    print("enter cm_throw_twice")
+    try:
+        yield
+        print("UNREACHABLE")
+    except ValueError:
+        print("throw yield")
+        yield
+        print("UNREACHABLE")
+
+def test_throw_twice() -> None:
+    try:
+        with cm_throw_twice():
+            raise ValueError("oops")
+        print("UNREACHABLE")
+    except RuntimeError as e:
+        print(type(e).__name__, str(e))
+
+@contextmanager
+def cm_stop_passthrough() -> Iterator[None]:
+    print("enter cm_stop_passthrough")
+    yield
+    print("UNREACHABLE")
+
+def test_stop_iteration_passthrough() -> None:
+    try:
+        with cm_stop_passthrough():
+            raise StopIteration("stop")
+        print("UNREACHABLE")
+    except Exception as e:
+        print("stop passthrough", type(e).__name__, str(e))
+
+@contextmanager
+def cm_stop_suppress() -> Iterator[None]:
+    print("enter cm_stop_suppress")
+    try:
+        yield
+        print("UNREACHABLE")
+    except StopIteration as e:
+        print(type(e).__name__, str(e))
+
+def test_stop_iteration_suppress() -> None:
+    with cm_stop_suppress():
+        raise StopIteration("manual stop")
+    print("after suppressing the StopIteration")
+
+@contextmanager
+def cm_value() -> Iterator[int]:
+    print("enter cm_value")
+    try:
+        yield 1
+        print("UNREACHABLE")
+    finally:
+        print("exit cm_value")
+
+def test_traceback() -> None:
+    try:
+        with cm_value():
+            raise ValueError("trace")
+        print("UNREACHABLE")
+    except Exception as e:
+        e_any = cast(Any, e)
+        tb = e_any.__traceback__
+        assert tb is not None
+        tb_mod = cast(Any, traceback)
+        names = [frame.name for frame in tb_mod.extract_tb(tb)]
+        print("traceback_has_cm_value", "test_traceback" in names, "cm_value" in names)
+
+@contextmanager
+def cm_gen() -> Iterator[int]:
+    print("enter cm_gen")
+    try:
+        yield 1
+    finally:
+        print("exit cm_gen")
+
+def test_generator_with_yield() -> None:
+    def gen() -> Generator[int, None, None]:
+        with cm_gen() as v:
+            yield v
+            print("after yield in with")
+
+    g = gen()
+    print(next(g))
+    try:
+        next(g)
+        print("UNREACHABLE")
+    except StopIteration:
+        print("gen done")
+
+@contextmanager
+def cm_log(tag: str) -> Iterator[None]:
+    print("enter cm_log", tag)
+    try:
+        yield
+    finally:
+        print("exit cm_log", tag)
+
+def test_return_in_with() -> None:
+    def inner() -> int:
+        with cm_log("ret"):
+            print("ret body")
+            return 5
+        print("UNREACHABLE")
+        return 0
+
+    print("ret result", inner())
+
+[file driver.py]
+from native import (
+    test_no_yield,
+    test_no_stop,
+    test_suppress_value_error,
+    test_throw_twice,
+    test_stop_iteration_passthrough,
+    test_stop_iteration_suppress,
+    test_traceback,
+    test_generator_with_yield,
+    test_return_in_with,
+)
+
+fns = [
+    test_no_yield,
+    test_no_stop,
+    test_suppress_value_error,
+    test_throw_twice,
+    test_stop_iteration_passthrough,
+    test_stop_iteration_suppress,
+    test_traceback,
+    test_generator_with_yield,
+    test_return_in_with,
+]
+for i, fn in enumerate(fns):
+    if i:
+        print()
+    print(f"Running {fn.__name__}")
+    fn()
+[out]
+Running test_no_yield
+enter cm_no_yield
+RuntimeError generator didn't yield True True
+
+Running test_no_stop
+enter cm_no_stop
+entered body first
+after first yield
+RuntimeError generator didn't stop
+
+Running test_suppress_value_error
+enter cm_value_error_suppress
+ValueError boom
+after suppressing the ValueError
+
+Running test_throw_twice
+enter cm_throw_twice
+throw yield
+RuntimeError generator didn't stop after throw()
+
+Running test_stop_iteration_passthrough
+enter cm_stop_passthrough
+stop passthrough StopIteration stop
+
+Running test_stop_iteration_suppress
+enter cm_stop_suppress
+StopIteration manual stop
+after suppressing the StopIteration
+
+Running test_traceback
+enter cm_value
+exit cm_value
+traceback_has_cm_value True False
+
+Running test_generator_with_yield
+enter cm_gen
+1
+after yield in with
+exit cm_gen
+gen done
+
+Running test_return_in_with
+enter cm_log ret
+ret body
+exit cm_log ret
+ret result 5
+
+[case testContextManagerRuntimePrimitives]
+from contextlib import contextmanager
+from typing import Any, Iterator, cast
+import helper
+
+# C iterators can exhaust without setting StopIteration.
+@contextmanager
+def implicit(values: list[int]) -> Iterator[int]:
+    return iter(values)
+
+def test_empty_iterator() -> None:
+    try:
+        with implicit([]):
+            print("UNREACHABLE")
+        print("UNREACHABLE")
+    except RuntimeError as error:
+        print(str(error), type(error.__context__).__name__)
+
+@contextmanager
+def noniterator() -> Iterator[None]:
+    return cast(Any, 42)
+
+def test_noniterator() -> None:
+    try:
+        with noniterator():
+            print("UNREACHABLE")
+        print("UNREACHABLE")
+    except TypeError as error:
+        print(str(error))
+
+# Other exceptions from next() must propagate unchanged at either boundary.
+@contextmanager
+def fail_entry(error: ValueError) -> Iterator[None]:
+    raise error
+    print("UNREACHABLE")
+    yield
+
+def test_entry_failure_identity() -> None:
+    original = ValueError('original')
+    try:
+        with fail_entry(original):
+            print("UNREACHABLE")
+        print("UNREACHABLE")
+    except ValueError as error:
+        assert error is original
+        print('entry error')
+
+@contextmanager
+def fail_exit(error: ValueError) -> Iterator[None]:
+    yield
+    raise error
+
+def test_exit_failure_identity() -> None:
+    original = ValueError('original')
+    try:
+        with fail_exit(original):
+            print('exit body')
+        print("UNREACHABLE")
+    except ValueError as error:
+        assert error is original
+        print('exit error')
+
+# Protocol errors temporarily replace the handled exception. They must restore
+# it and release the saved references, even after repeated failures.
+@contextmanager
+def missing() -> Iterator[None]:
+    if False:
+        print("UNREACHABLE")
+        yield
+
+def failed_entry() -> None:
+    try:
+        with missing():
+            print("UNREACHABLE")
+            assert False
+        print("UNREACHABLE")
+    except RuntimeError as error:
+        assert isinstance(error.__context__, StopIteration)
+
+@contextmanager
+def twice() -> Iterator[int]:
+    yield 1
+    yield 2
+    print("UNREACHABLE")
+
+def failed_exit() -> None:
+    try:
+        with twice():
+            pass
+    except RuntimeError:
+        pass
+    else:
+        print("UNREACHABLE")
+        assert False
+
+def test_outer_exception() -> None:
+    outer = ValueError('outer')
+    try:
+        raise outer
+    except ValueError:
+        # Also exercise normal exit when a C iterator exhausts without an error.
+        with implicit([7]) as value:
+            assert value == 7
+            assert helper.exc_info()[1] is outer
+        assert helper.exc_info()[1] is outer
+        # Collect exception/traceback cycles before measuring retained references.
+        helper.collect()
+        before = helper.refcount(outer)
+        for i in range(9):
+            failed_entry()
+        helper.collect()
+        assert helper.refcount(outer) == before
+        assert helper.exc_info()[1] is outer
+        for i in range(9):
+            failed_exit()
+        helper.collect()
+        assert helper.refcount(outer) == before
+        assert helper.exc_info()[1] is outer
+    assert helper.exc_info()[1] is None
+    print('outer exception restored; saved references released')
+
+[file helper.py]
+from typing import Any
+import gc
+import sys
+
+def exc_info() -> Any:
+    return sys.exc_info()  # type: ignore
+
+def collect() -> None:
+    gc.collect()
+
+def refcount(value: Any) -> int:
+    return sys.getrefcount(value)  # type: ignore
+
+[file driver.py]
+from native import (
+    test_empty_iterator,
+    test_noniterator,
+    test_entry_failure_identity,
+    test_exit_failure_identity,
+    test_outer_exception,
+)
+
+fns = [
+    test_empty_iterator,
+    test_noniterator,
+    test_entry_failure_identity,
+    test_exit_failure_identity,
+    test_outer_exception,
+]
+for i, fn in enumerate(fns):
+    if i:
+        print()
+    print(f"Running {fn.__name__}")
+    fn()
+[out]
+Running test_empty_iterator
+generator didn't yield StopIteration
+
+Running test_noniterator
+'int' object is not an iterator
+
+Running test_entry_failure_identity
+entry error
+
+Running test_exit_failure_identity
+exit body
+exit error
+
+Running test_outer_exception
+outer exception restored; saved references released
+
+
+[case testContextManagerExceptionalExit]
+from contextlib import contextmanager
+from typing import Any, Iterator
+import helper
+
+@contextmanager
+def scope(generator: Iterator[None]) -> Iterator[None]:
+    return generator
+
+def invoke(generator: Iterator[None], error: BaseException,
+           captured: list[Any]) -> None:
+    with scope(generator):
+        helper.raise_body(error, captured)
+        print("UNREACHABLE")
+
+# Unwinding this handler during throw() can change the native caller's exception state.
+def native_handler() -> Iterator[None]:
+    try:
+        raise TypeError('earlier')
+    except BaseException:
+        yield
+        print("UNREACHABLE")
+
+[file helper.py]
+from typing import Any, Iterator, cast
+
+class Other(BaseException):
+    pass
+
+class StopSubclass(StopIteration):
+    pass
+
+def raise_body(error: BaseException, captured: list[Any]) -> None:
+    try:
+        raise error
+    except BaseException:
+        captured.append(cast(Any, error).__traceback__)
+        raise
+
+def passthrough() -> Iterator[None]:
+    yield
+    print("UNREACHABLE")
+
+def replace(error: BaseException) -> Iterator[None]:
+    try:
+        yield
+        print("UNREACHABLE")
+    except BaseException:
+        raise error
+
+def suppress() -> Iterator[None]:
+    try:
+        yield
+        print("UNREACHABLE")
+    except ValueError:
+        pass
+
+[file driver.py]
+import gc
+import sys
+import helper
+from native import invoke, native_handler
+
+
+def test_exception_identity_and_traceback():
+    # These take different branches in contextlib; all must propagate the same object.
+    for error_type in [RuntimeError, helper.Other, helper.StopSubclass]:
+        original = error_type('body')
+        captured = []
+        try:
+            invoke(helper.passthrough(), original, captured)
+        except BaseException as caught:
+            assert caught is original
+            tb = caught.__traceback__
+            while tb is not None and tb is not captured[0]:
+                tb = tb.tb_next
+            assert tb is captured[0]
+            print(error_type.__name__, 'identity and original traceback preserved')
+        else:
+            print("UNREACHABLE")
+            assert False, 'unexpected suppression'
+
+# The interpreted generator isolates the caller's handled-exception state.
+def propagate_from_handler():
+    original = ValueError('body')
+    try:
+        invoke(native_handler(), original, [])
+    except BaseException as caught:
+        yield caught is original
+    else:
+        print("UNREACHABLE")
+        yield False
+
+
+def test_native_handler_exception_state():
+    assert list(propagate_from_handler()) == [True]
+    assert sys.exc_info()[1] is None
+    try:
+        raise KeyError('outer')
+    except KeyError as outer:
+        assert list(propagate_from_handler()) == [True]
+        assert sys.exc_info()[1] is outer
+    assert sys.exc_info()[1] is None
+    print('native handler preserves body exception with and without an outer handler')
+
+
+def test_replacement_exception():
+    # A different exception from throw() must replace the body's exception.
+    for error_type in [RuntimeError, helper.Other]:
+        original = ValueError('body')
+        replacement = error_type('replacement')
+        try:
+            invoke(helper.replace(replacement), original, [])
+        except BaseException as caught:
+            assert caught is replacement
+            assert caught.__context__ is original
+            assert original.__context__ is None
+            print(f'{type(caught).__name__}({caught})')
+            print(f'{type(original).__name__}({original})')
+        else:
+            print("UNREACHABLE")
+            assert False, 'unexpected suppression'
+
+
+def test_suppression_exception_state_and_references():
+    # Suppression must restore the outer handled exception without retaining its refs.
+    outer = ValueError('outer')
+    try:
+        raise outer
+    except ValueError:
+        gc.collect()
+        before = sys.getrefcount(outer)
+        for i in range(9):
+            invoke(helper.suppress(), ValueError('body'), [])
+            assert sys.exc_info()[1] is outer
+        gc.collect()
+        assert sys.getrefcount(outer) == before
+        print('suppression restores outer exception and releases saved references')
+    assert sys.exc_info()[1] is None
+
+
+fns = [
+    test_exception_identity_and_traceback,
+    test_native_handler_exception_state,
+    test_replacement_exception,
+    test_suppression_exception_state_and_references,
+]
+for i, fn in enumerate(fns):
+    if i:
+        print()
+    print(f"Running {fn.__name__}")
+    fn()
+[out]
+Running test_exception_identity_and_traceback
+RuntimeError identity and original traceback preserved
+Other identity and original traceback preserved
+StopSubclass identity and original traceback preserved
+
+Running test_native_handler_exception_state
+native handler preserves body exception with and without an outer handler
+
+Running test_replacement_exception
+RuntimeError(replacement)
+ValueError(body)
+Other(replacement)
+ValueError(body)
+
+Running test_suppression_exception_state_and_references
+suppression restores outer exception and releases saved references
+
+[case testContextManagerRuntimeMethodBinding]
+from contextlib import contextmanager
+from typing import Callable, Iterator
+from mypy_extensions import mypyc_attr
+
+@mypyc_attr(allow_interpreted_subclasses=True)
+class Base:
+    @contextmanager
+    def cm(self, value: str) -> Iterator[str]:
+        yield 'base:' + value
+
+# The same Base-typed call must preserve the callable's actual binding.
+def invoke(receiver: Callable[[], Base], argument: Callable[[], str]) -> None:
+    with receiver().cm(value=argument()) as value:
+        print('body', value)
+
+[file driver.py]
+from contextlib import contextmanager
+from native import Base, invoke
+
+# A classmethod binds to the runtime class; a staticmethod has no implicit receiver.
+
+class ClassOverride(Base):
+    @classmethod
+    @contextmanager
+    def cm(cls, value):
+        yield cls.__name__ + ':' + value
+
+invoke(ClassOverride, lambda: 'value')
+
+class StaticOverride(Base):
+    @staticmethod
+    @contextmanager
+    def cm(value):
+        yield 'static:' + value
+
+invoke(StaticOverride, lambda: 'value')
+
+# A descriptor can bind to another object. Lookup and arguments run once, in order.
+class Target:
+    tag = 'target'
+
+    @contextmanager
+    def cm(self, value):
+        print('enter', self.tag)
+        yield value
+        print('exit')
+
+target = Target()
+
+class Descriptor:
+    def __get__(self, instance, owner):
+        print('lookup')
+        return target.cm
+
+class Child(Base):
+    cm = Descriptor()
+
+obj = Child()
+
+def receiver():
+    print('receiver')
+    return obj
+
+def argument():
+    print('argument')
+    return 'keyword'
+
+invoke(receiver, argument)
+[out]
+body ClassOverride:value
+body static:value
+receiver
+lookup
+argument
+enter target
+body keyword
+exit
+
 [case testUnpackKwargsCompiled]
 from typing import TypedDict
 from typing_extensions import Unpack