diff --git a/newsfragments/552.bugfix.rst b/newsfragments/552.bugfix.rst
new file mode 100644
index 0000000000..8239edcf25
--- /dev/null
+++ b/newsfragments/552.bugfix.rst
@@ -0,0 +1 @@
+When a Trio task makes improper use of a non-Trio async library, Trio now causes an exception to be raised within the task at the point of the error, rather than abandoning the task and raising an error in its parent. This improves debuggability and resolves the `TrioInternalError` that would sometimes result from the latter strategy.
diff --git a/trio/_core/_run.py b/trio/_core/_run.py
index 96da9262bf..d662d0eb52 100644
--- a/trio/_core/_run.py
+++ b/trio/_core/_run.py
@@ -980,10 +980,17 @@ class Task:
     _counter = attr.ib(init=False, factory=itertools.count().__next__)
 
     # Invariant:
-    # - for unscheduled tasks, _next_send is None
-    # - for scheduled tasks, _next_send is an Outcome object,
-    #   and custom_sleep_data is None
+    # - for unscheduled tasks, _next_send_fn and _next_send are both None
+    # - for scheduled tasks, _next_send_fn(_next_send) resumes the task;
+    #   usually _next_send_fn is self.coro.send and _next_send is an
+    #   Outcome. When recovering from a foreign await, _next_send_fn is
+    #   self.coro.throw and _next_send is an exception. _next_send_fn
+    #   will effectively be at the top of every task's call stack, so
+    #   it should be written in C if you don't want to pollute Trio
+    #   tracebacks with extraneous frames.
+    # - for scheduled tasks, custom_sleep_data is None
     # Tasks start out unscheduled.
+    _next_send_fn = attr.ib(default=None)
     _next_send = attr.ib(default=None)
     _abort_func = attr.ib(default=None)
     custom_sleep_data = attr.ib(default=None)
@@ -1215,7 +1222,8 @@ def reschedule(self, task, next_send=_NO_SEND):
             next_send = Value(None)
 
         assert task._runner is self
-        assert task._next_send is None
+        assert task._next_send_fn is None
+        task._next_send_fn = task.coro.send
         task._next_send = next_send
         task._abort_func = None
         task.custom_sleep_data = None
@@ -1887,8 +1895,9 @@ def run_impl(runner, async_fn, args):
             if runner.instruments:
                 runner.instrument("before_task_step", task)
 
+            next_send_fn = task._next_send_fn
             next_send = task._next_send
-            task._next_send = None
+            task._next_send_fn = task._next_send = None
             final_outcome = None
             try:
                 # We used to unwrap the Outcome object here and send/throw its
@@ -1898,7 +1907,7 @@ def run_impl(runner, async_fn, args):
                 #   https://bugs.python.org/issue29590
                 # So now we send in the Outcome object and unwrap it on the
                 # other side.
-                msg = task.context.run(task.coro.send, next_send)
+                msg = task.context.run(next_send_fn, next_send)
             except StopIteration as stop_iteration:
                 final_outcome = Value(stop_iteration.value)
             except BaseException as task_exc:
@@ -1938,16 +1947,12 @@ def run_impl(runner, async_fn, args):
                         "other framework like asyncio? That won't work "
                         "without some kind of compatibility shim.".format(msg)
                     )
-                    # How can we resume this task? It's blocked in code we
-                    # don't control, waiting for some message that we know
-                    # nothing about. We *could* try using coro.throw(...) to
-                    # blast an exception in and hope that it propagates out,
-                    # but (a) that's complicated because we aren't set up to
-                    # resume a task via .throw(), and (b) even if we did,
-                    # there's no guarantee that the foreign code will respond
-                    # the way we're hoping. So instead we abandon this task
-                    # and propagate the exception into the task's spawner.
-                    runner.task_exited(task, Error(exc))
+                    # The foreign library probably doesn't adhere to our
+                    # protocol of unwrapping whatever outcome gets sent in.
+                    # Instead, we'll arrange to throw `exc` in directly,
+                    # which works for at least asyncio and curio.
+                    runner.reschedule(task, exc)
+                    task._next_send_fn = task.coro.throw
 
             if runner.instruments:
                 runner.instrument("after_task_step", task)
diff --git a/trio/_core/tests/test_run.py b/trio/_core/tests/test_run.py
index 07798be935..f1ca40f4d4 100644
--- a/trio/_core/tests/test_run.py
+++ b/trio/_core/tests/test_run.py
@@ -1782,14 +1782,31 @@ async def async_gen(arg):  # pragma: no cover
 
 
 def test_calling_asyncio_function_gives_nice_error():
-    async def misguided():
+    async def child_xyzzy():
         import asyncio
         await asyncio.Future()
 
+    async def misguided():
+        await child_xyzzy()
+
     with pytest.raises(TypeError) as excinfo:
         _core.run(misguided)
 
     assert "asyncio" in str(excinfo.value)
+    # The traceback should point to the location of the foreign await
+    assert any(  # pragma: no branch
+        entry.name == "child_xyzzy" for entry in excinfo.traceback
+    )
+
+
+async def test_asyncio_function_inside_nursery_does_not_explode():
+    # Regression test for https://github.com/python-trio/trio/issues/552
+    with pytest.raises(TypeError) as excinfo:
+        async with _core.open_nursery() as nursery:
+            import asyncio
+            nursery.start_soon(sleep_forever)
+            await asyncio.Future()
+    assert "asyncio" in str(excinfo.value)
 
 
 async def test_trivial_yields():