feat(dom): implement Observable subscription lifecycle

moli-benchmark/wpt-cross-current/failed-cases.txt

@@ -2490,7 +2490,6 @@ dom/nodes/moveBefore/moveBefore-from-light-to-shadow.html
 dom/nodes/moveBefore/moveBefore-size-query.html
 dom/nodes/moveBefore/preserve-render-blocking-script.html
 dom/nodes/moveBefore/preserve-render-blocking-style.html
-dom/observable/tentative/idlharness.html
 dom/ranges/tentative/OpaqueRange-auto-disconnect.html
 dom/ranges/tentative/OpaqueRange-basic.html
 dom/ranges/tentative/OpaqueRange-disconnect.html

moli-benchmark/wpt-cross-current/passed-cases.txt

@@ -4532,6 +4532,9 @@ dom/nodes/remove-unscopable.html
 dom/nodes/replaceWith-document-element-crash.html
 dom/nodes/rootNode.html
 dom/nodes/svg-template-querySelector.html
+dom/observable/tentative/idlharness.html
+dom/observable/tentative/observable-constructor.any.js?moli-wpt-any=dedicatedworker
+dom/observable/tentative/observable-constructor.any.js?moli-wpt-any=window
 dom/ranges/Range-adopt-test.html
 dom/ranges/Range-attributes.html
 dom/ranges/Range-cloneContents.html

moli-renderer-v8/src/abort_signal_route.rs

@@ -28,6 +28,36 @@ pub(crate) struct ResolvedAbortSignal<'s> {
 }
 
 impl<'s> ResolvedAbortSignal<'s> {
+    /// Creates a signal in the current realm without consulting author-visible
+    /// constructors or maintaining another store for native algorithms.
+    pub(crate) fn new(scope: &mut v8::PinScope<'s, '_>) -> Option<Self> {
+        let signal = if let Some(host_ptr) = context_host_ptr_from_global_bridge(scope) {
+            crate::native_bridge::abort::create_signal(
+                scope,
+                unsafe { &mut *host_ptr },
+                false,
+                None,
+            )?
+        } else {
+            crate::worker::abort::new_worker_abort_signal(scope)?
+        };
+        Self::resolve(scope, signal)
+    }
+
+    pub(crate) fn reason(self, scope: &mut v8::PinScope<'s, '_>) -> v8::Local<'s, v8::Value> {
+        match self.owner {
+            AbortSignalOwner::Window => {
+                context_host_ptr_from_global_bridge(scope).and_then(|host_ptr| {
+                    unsafe { &mut *host_ptr }.abort_signal_reason(scope, self.signal)
+                })
+            }
+            AbortSignalOwner::Worker => {
+                crate::worker::abort::worker_abort_signal_reason(scope, self.signal)
+            }
+        }
+        .unwrap_or_else(|| v8::undefined(scope).into())
+    }
+
     pub(crate) fn resolve(
         scope: &mut v8::PinScope<'s, '_>,
         signal: v8::Local<'s, v8::Object>,

moli-renderer-v8/src/context_bootstrap/assets/constructor_templates.rs

@@ -296,6 +296,14 @@ pub(in crate::context_bootstrap) fn build_constructor_template_for_profile<'s>(
             .length(1)
             .build(scope)
         }
+        ConstructorKind::Observable => {
+            v8::FunctionTemplate::builder(moli_webapi_declare::web_api_constructor!(
+                web_api_interfaces::Observable,
+                crate::observable::constructor
+            ))
+            .length(1)
+            .build(scope)
+        }
         ConstructorKind::DomParser => {
             v8::FunctionTemplate::builder(moli_webapi_declare::web_api_constructor!(
                 web_api_interfaces::DOMParser,

moli-renderer-v8/src/context_bootstrap/assets/prototype_bindings.rs

@@ -412,6 +412,7 @@ pub(super) fn install_constructor_template_bindings<'s>(
     install_constructor_constant_template_bindings(scope, template, spec.interface.name());
     install_css_style_declaration_template_accessors(scope, template, spec.interface.name());
     install_abort_template_bindings(scope, template, spec.interface.name());
+    crate::observable::install_template_bindings(scope, template, spec.interface.name());
     install_attr_template_bindings(scope, template, spec.interface.name());
     install_dom_quad_template_bindings(scope, template, spec.interface.name());
     install_dom_rect_template_bindings(scope, template, spec.interface.name());

moli-renderer-v8/src/context_bootstrap/exposed_interfaces/metadata.rs

@@ -31,6 +31,8 @@ pub(in crate::context_bootstrap) const WORKER_SHARED_INTERFACE_NAMES: &[&str] =
     "WebSocket",
     "AbortSignal",
     "AbortController",
+    "Observable",
+    "Subscriber",
     "Headers",
     "Request",
     "Response",

moli-renderer-v8/src/context_bootstrap/runtime_state.rs

@@ -1895,6 +1895,8 @@ pub(crate) fn finish_context_bootstrap(
         ("AudioBuffer", "AudioBuffer"),
         ("AbortSignal", "AbortSignal"),
         ("AbortController", "AbortController"),
+        ("Observable", "Observable"),
+        ("Subscriber", "Subscriber"),
         ("TextEncoder", "TextEncoder"),
         ("TextDecoder", "TextDecoder"),
         ("ReadableStream", "ReadableStream"),

moli-renderer-v8/src/context_bootstrap/specs/registry.rs

@@ -388,6 +388,14 @@ const CONSTRUCTOR_SPECS_BEFORE_STREAMS: &[ConstructorSpec] = &[
         interface: web_api_interfaces::AbortController::DESCRIPTOR,
         kind: ConstructorKind::AbortController,
     },
+    ConstructorSpec {
+        interface: web_api_interfaces::Observable::DESCRIPTOR,
+        kind: ConstructorKind::Observable,
+    },
+    ConstructorSpec {
+        interface: web_api_interfaces::Subscriber::DESCRIPTOR,
+        kind: ConstructorKind::Illegal,
+    },
     ConstructorSpec {
         interface: web_api_interfaces::BroadcastChannel::DESCRIPTOR,
         kind: ConstructorKind::BroadcastChannel,

moli-renderer-v8/src/context_bootstrap/specs/types.rs

@@ -77,6 +77,7 @@ pub(in crate::context_bootstrap) enum ConstructorKind {
     FileReaderSync,
     XmlSerializer,
     AbortController,
+    Observable,
     BroadcastChannel,
     EventSource,
     IdleDetector,

moli-renderer-v8/src/lib.rs

@@ -74,6 +74,7 @@ mod mutation_coordinator;
 pub(crate) mod native_bridge;
 pub mod network;
 mod network_host;
+mod observable;
 mod observer_runtime;
 mod opfs_owner_tasks;
 mod opfs_task_result;

moli-renderer-v8/src/native_bridge/abort.rs

@@ -377,7 +377,7 @@ pub(super) fn timeout_error_value<'s>(
     dom_exception_value(scope, "signal timed out", "TimeoutError")
 }
 
-fn create_signal<'s>(
+pub(crate) fn create_signal<'s>(
     scope: &mut v8::PinScope<'s, '_>,
     host: &mut super::JsContextHost,
     aborted: bool,

moli-renderer-v8/src/observable.rs

@@ -0,0 +1,423 @@
+//! Observable subscriptions share one producer until their last observer leaves.
+//! Callback residence is V8-traced; the Observable's link to its Subscriber is
+//! weak. AbortSignal state and callback invocation remain with their existing
+//! Window/worker owners.
+
+mod callbacks;
+mod state;
+
+use moli_webapi_declare::WebApiFunctionTemplate;
+
+use crate::{
+    abort_signal_route::ResolvedAbortSignal, util::set_private_value, web_api_interfaces, webidl,
+};
+use callbacks::{invoke, is_current, report};
+use state::*;
+
+#[derive(WebApiFunctionTemplate)]
+#[webapi(interface = web_api_interfaces::Observable, enumerable, receiver)]
+struct ObservablePrototype {
+    #[webapi(method, length = 0, callback = subscribe)]
+    subscribe: (),
+}
+
+#[derive(WebApiFunctionTemplate)]
+#[webapi(interface = web_api_interfaces::Subscriber, enumerable, receiver)]
+struct SubscriberPrototype {
+    #[webapi(method, length = 1, callback = next)]
+    next: (),
+    #[webapi(method, length = 1, callback = error)]
+    error: (),
+    #[webapi(method, length = 0, callback = complete)]
+    complete: (),
+    #[webapi(method, length = 1, callback = add_teardown)]
+    add_teardown: (),
+    #[webapi(accessor_property, getter = active_getter)]
+    active: (),
+    #[webapi(accessor_property, getter = signal_getter)]
+    signal: (),
+}
+
+pub(crate) fn install_template_bindings<'s>(
+    scope: &mut v8::PinScope<'s, '_, ()>,
+    template: v8::Local<'s, v8::FunctionTemplate>,
+    name: &str,
+) {
+    let prototype = template.prototype_template(scope);
+    match name {
+        "Observable" => ObservablePrototype::initialize_prototype_template(scope, prototype),
+        "Subscriber" => SubscriberPrototype::initialize_prototype_template(scope, prototype),
+        _ => {}
+    }
+}
+
+#[derive(webidl::WebIdlArgs)]
+#[webidl(prefix = "Observable")]
+struct ConstructorArgs {
+    #[webidl(required, converter = "callback_function")]
+    callback: webidl::WebIdlCallbackFunction,
+}
+
+#[derive(webidl::WebIdlArgs)]
+#[webidl(prefix = "Observable.subscribe")]
+struct SubscribeArgs<'scope> {
+    #[webidl(with = observer_arg)]
+    observer: v8::Local<'scope, v8::Object>,
+    #[webidl(with = signal_arg)]
+    signal: Option<ResolvedAbortSignal<'scope>>,
+}
+
+#[derive(webidl::WebIdlArgs)]
+#[webidl(prefix = "Subscriber")]
+struct ValueArgs<'scope> {
+    #[webidl(required, converter = "raw")]
+    value: v8::Local<'scope, v8::Value>,
+}
+
+#[derive(webidl::WebIdlArgs)]
+#[webidl(prefix = "Subscriber.addTeardown")]
+struct TeardownArgs {
+    #[webidl(required, converter = "callback_function")]
+    callback: webidl::WebIdlCallbackFunction,
+}
+
+fn dictionary<'s>(
+    value: v8::Local<'s, v8::Value>,
+    message: &'static str,
+) -> Result<Option<v8::Local<'s, v8::Object>>, webidl::WebIdlError> {
+    if value.is_null_or_undefined() {
+        return Ok(None);
+    }
+    v8::Local::<v8::Object>::try_from(value)
+        .map(Some)
+        .map_err(|_| webidl::WebIdlError::custom_message(message))
+}
+
+fn observer_arg<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: &v8::FunctionCallbackArguments<'s>,
+    index: i32,
+) -> Result<v8::Local<'s, v8::Object>, webidl::WebIdlError> {
+    let observer = v8::Object::new(scope);
+    let Some(input) = dictionary(args.get(index), "SubscriptionObserver must be an object")? else {
+        return Ok(observer);
+    };
+    if input.is_callable() {
+        let callback = webidl::convert::<webidl::WebIdlCallbackFunction>(
+            scope,
+            input.into(),
+            webidl::Context::argument("Observable.subscribe", 1),
+        )?;
+        set_callback(scope, observer, NEXT, callback);
+        return Ok(observer);
+    }
+    // Web IDL dictionary member conversion is lexicographic, even though the
+    // subscription itself delivers next/error/complete notifications.
+    for (name, slot) in [("complete", COMPLETE), ("error", ERROR), ("next", NEXT)] {
+        let context = webidl::Context::member("SubscriptionObserver", name);
+        if let Some(value) = webidl::property_result(scope, input, name, context)?
+            && !value.is_undefined()
+        {
+            let callback =
+                webidl::convert::<webidl::WebIdlCallbackFunction>(scope, value, context)?;
+            set_callback(scope, observer, slot, callback);
+        }
+    }
+    Ok(observer)
+}
+
+fn signal_arg<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: &v8::FunctionCallbackArguments<'s>,
+    index: i32,
+) -> Result<Option<ResolvedAbortSignal<'s>>, webidl::WebIdlError> {
+    let Some(options) = dictionary(args.get(index), "SubscribeOptions must be an object")? else {
+        return Ok(None);
+    };
+    let context = webidl::Context::member("SubscribeOptions", "signal");
+    let Some(value) = webidl::property_result(scope, options, "signal", context)? else {
+        return Ok(None);
+    };
+    if value.is_undefined() {
+        return Ok(None);
+    }
+    v8::Local::<v8::Object>::try_from(value)
+        .ok()
+        .filter(|signal| web_api_interfaces::AbortSignal::is_instance(scope, *signal))
+        .and_then(|signal| ResolvedAbortSignal::resolve(scope, signal))
+        .map(Some)
+        .ok_or_else(|| {
+            webidl::WebIdlError::custom_message("SubscribeOptions.signal must be an AbortSignal")
+        })
+}
+
+pub(crate) fn constructor<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    mut rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    if !args.is_construct_call() {
+        webidl::throw_type_error(scope, "Observable must be constructed with new");
+        return;
+    }
+    let Some(parsed) = webidl::parse_args::<ConstructorArgs>(scope, &args) else {
+        return;
+    };
+    initialize_observable(scope, args.this(), parsed.callback);
+    rv.set(args.this().into());
+}
+
+fn subscribe<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    _rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    let Some(parsed) = webidl::parse_args::<SubscribeArgs<'s>>(scope, &args) else {
+        return;
+    };
+    let observable = args.this();
+    if !is_current(scope, observable) {
+        return;
+    }
+    let (subscriber, fresh) = match current_subscriber(scope, observable)
+        .filter(|subscriber| active(scope, *subscriber))
+    {
+        Some(subscriber) => (subscriber, false),
+        None => {
+            let Some(subscriber) = new_subscriber(scope) else {
+                return;
+            };
+            set_subscriber(scope, observable, subscriber);
+            (subscriber, true)
+        }
+    };
+    let mut observers = list(scope, subscriber, OBSERVERS);
+    observers.push(parsed.observer);
+    set_list(scope, subscriber, OBSERVERS, &observers);
+    if let Some(signal) = parsed.signal {
+        if signal.is_aborted(scope) {
+            if fresh {
+                let reason = signal.reason(scope);
+                close(scope, subscriber, Some(reason));
+            } else {
+                observers.pop();
+                set_list(scope, subscriber, OBSERVERS, &observers);
+            }
+        } else {
+            let data =
+                v8::Array::new_with_elements(scope, &[subscriber.into(), parsed.observer.into()]);
+            let algorithm = v8::Function::builder(cancel_observer)
+                .data(data.into())
+                .build(scope)
+                .expect("Observable abort algorithm should allocate");
+            set_private_value(scope, parsed.observer, INPUT_SIGNAL, signal.value().into());
+            set_private_value(scope, parsed.observer, ABORT_ALGORITHM, algorithm.into());
+            signal.register_algorithm(scope, algorithm);
+        }
+    }
+    if fresh
+        && let Some(callback) = object_slot(scope, observable, INITIALIZER)
+        && let Some(exception) = invoke(scope, callback, &[subscriber.into()])
+    {
+        subscriber_error(scope, subscriber, exception);
+    }
+}
+
+fn cancel_observer<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    _rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    let data = v8::Local::<v8::Array>::try_from(args.data()).expect("Observable abort data");
+    let subscriber = v8::Local::<v8::Object>::try_from(data.get_index(scope, 0).unwrap()).unwrap();
+    let observer = v8::Local::<v8::Object>::try_from(data.get_index(scope, 1).unwrap()).unwrap();
+    if !active(scope, subscriber) {
+        return;
+    }
+    let mut observers = list(scope, subscriber, OBSERVERS);
+    observers.retain(|entry| *entry != observer);
+    set_list(scope, subscriber, OBSERVERS, &observers);
+    release_abort_algorithm(scope, observer);
+    if observers.is_empty() {
+        close(scope, subscriber, Some(args.get(0)));
+    }
+}
+
+fn close<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    subscriber: v8::Local<'s, v8::Object>,
+    reason: Option<v8::Local<'s, v8::Value>>,
+) {
+    if !active(scope, subscriber) {
+        return;
+    }
+    set_private_value(
+        scope,
+        subscriber,
+        ACTIVE,
+        v8::Boolean::new(scope, false).into(),
+    );
+    for observer in list(scope, subscriber, OBSERVERS) {
+        release_abort_algorithm(scope, observer);
+    }
+    let signal = object_slot(scope, subscriber, SIGNAL)
+        .and_then(|signal| ResolvedAbortSignal::resolve(scope, signal));
+    let reason = reason
+        .filter(|reason| !reason.is_undefined())
+        .unwrap_or_else(|| crate::native_bridge::abort::abort_error_value(scope));
+    if let Some(signal) = signal {
+        signal.abort(scope, reason);
+    }
+    let teardowns = list(scope, subscriber, TEARDOWNS);
+    set_list(scope, subscriber, TEARDOWNS, &[]);
+    for callback in teardowns.into_iter().rev() {
+        if !is_current(scope, subscriber) {
+            break;
+        }
+        invoke_and_report(scope, callback, &[]);
+    }
+}
+
+fn release_abort_algorithm<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    observer: v8::Local<'s, v8::Object>,
+) {
+    if let Some(signal) = object_slot(scope, observer, INPUT_SIGNAL)
+        .and_then(|signal| ResolvedAbortSignal::resolve(scope, signal))
+        && let Some(algorithm) = object_slot(scope, observer, ABORT_ALGORITHM)
+            .and_then(|value| v8::Local::<v8::Function>::try_from(value).ok())
+    {
+        signal.unregister_algorithm(scope, algorithm);
+    }
+    set_private_value(scope, observer, INPUT_SIGNAL, v8::undefined(scope).into());
+    set_private_value(
+        scope,
+        observer,
+        ABORT_ALGORITHM,
+        v8::undefined(scope).into(),
+    );
+}
+
+fn invoke_and_report<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    callback: v8::Local<'s, v8::Object>,
+    values: &[v8::Local<'s, v8::Value>],
+) {
+    if let Some(error) = invoke(scope, callback, values) {
+        callbacks::report_callback_exception(scope, callback, error);
+    }
+}
+
+fn next<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    _rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    let Some(parsed) = webidl::parse_args::<ValueArgs<'s>>(scope, &args) else {
+        return;
+    };
+    let subscriber = args.this();
+    if !active(scope, subscriber) || !is_current(scope, subscriber) {
+        return;
+    }
+    // Reentrant subscribe/cancel must not change this notification's snapshot.
+    for observer in list(scope, subscriber, OBSERVERS) {
+        if let Some(callback) = object_slot(scope, observer, NEXT) {
+            invoke_and_report(scope, callback, &[parsed.value]);
+        }
+    }
+}
+
+fn subscriber_error<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    subscriber: v8::Local<'s, v8::Object>,
+    error: v8::Local<'s, v8::Value>,
+) {
+    if !active(scope, subscriber) {
+        report(scope, subscriber, error);
+        return;
+    }
+    if !is_current(scope, subscriber) {
+        return;
+    }
+    close(scope, subscriber, Some(error));
+    let observers = list(scope, subscriber, OBSERVERS);
+    set_list(scope, subscriber, OBSERVERS, &[]);
+    for observer in observers {
+        if let Some(callback) = object_slot(scope, observer, ERROR) {
+            invoke_and_report(scope, callback, &[error]);
+        } else {
+            callbacks::report_default_error(scope, error);
+        }
+    }
+}
+
+fn error<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    _rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    let Some(parsed) = webidl::parse_args::<ValueArgs<'s>>(scope, &args) else {
+        return;
+    };
+    subscriber_error(scope, args.this(), parsed.value);
+}
+
+fn complete<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    _rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    let subscriber = args.this();
+    if !active(scope, subscriber) || !is_current(scope, subscriber) {
+        return;
+    }
+    close(scope, subscriber, None);
+    let observers = list(scope, subscriber, OBSERVERS);
+    set_list(scope, subscriber, OBSERVERS, &[]);
+    for observer in observers {
+        if let Some(callback) = object_slot(scope, observer, COMPLETE) {
+            invoke_and_report(scope, callback, &[]);
+        }
+    }
+}
+
+fn add_teardown<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    _rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    let Some(parsed) = webidl::parse_args::<TeardownArgs>(scope, &args) else {
+        return;
+    };
+    let subscriber = args.this();
+    if !is_current(scope, subscriber) {
+        return;
+    }
+    let callback = callbacks::trace(scope, parsed.callback);
+    if active(scope, subscriber) {
+        let mut teardowns = list(scope, subscriber, TEARDOWNS);
+        teardowns.push(callback);
+        set_list(scope, subscriber, TEARDOWNS, &teardowns);
+    } else {
+        invoke_and_report(scope, callback, &[]);
+    }
+}
+
+fn active_getter<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    mut rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    rv.set_bool(active(scope, args.this()));
+}
+
+fn signal_getter<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    args: v8::FunctionCallbackArguments<'s>,
+    mut rv: v8::ReturnValue<'_, v8::Value>,
+) {
+    if let Some(signal) = object_slot(scope, args.this(), SIGNAL) {
+        rv.set(signal.into());
+    }
+}

moli-renderer-v8/src/observable/callbacks.rs

@@ -0,0 +1,113 @@
+use crate::{
+    callback_invocation::invoke_synchronous_webidl_callback_function,
+    exception_reporting::build_exception_report_without_stack,
+    util::context_host_ptr_from_global_bridge,
+    v8_traced_webidl_callback::V8TracedWebIdlCallbackFunction, webidl,
+};
+
+pub(super) fn trace<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    callback: webidl::WebIdlCallbackFunction,
+) -> v8::Local<'s, v8::Object> {
+    V8TracedWebIdlCallbackFunction::new(scope, callback).into_object()
+}
+
+fn context_is_current<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    context: v8::Local<'s, v8::Context>,
+) -> bool {
+    if let Some(host_ptr) = context_host_ptr_from_global_bridge(scope) {
+        let host = unsafe { &*host_ptr };
+        host.window_execution_context_identity_for_v8_context(scope, context)
+            .is_some_and(|identity| host.window_execution_context_identity_is_current(identity))
+    } else {
+        crate::worker::get_worker_state(scope).is_some()
+    }
+}
+
+pub(super) fn is_current<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    object: v8::Local<'s, v8::Object>,
+) -> bool {
+    object
+        .get_creation_context(scope)
+        .is_some_and(|context| context_is_current(scope, context))
+}
+
+/// The initializer rethrows into Subscriber.error; observer and teardown
+/// callbacks report errors in their own relevant realm. Neither path runs an
+/// extra microtask checkpoint.
+pub(super) fn invoke<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    carrier: v8::Local<'s, v8::Object>,
+    arguments: &[v8::Local<'s, v8::Value>],
+) -> Option<v8::Local<'s, v8::Value>> {
+    let callback = V8TracedWebIdlCallbackFunction::from_object(carrier).prepare(scope);
+    let context = callback.relevant_context(scope);
+    if !context_is_current(scope, context) {
+        return None;
+    }
+    v8::tc_scope!(let scope, scope);
+    let receiver = v8::undefined(scope).into();
+    invoke_synchronous_webidl_callback_function(scope, &callback, receiver, arguments);
+    let exception = scope.exception();
+    scope.reset();
+    exception
+}
+
+pub(super) fn report_callback_exception<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    carrier: v8::Local<'s, v8::Object>,
+    exception: v8::Local<'s, v8::Value>,
+) {
+    let callback = V8TracedWebIdlCallbackFunction::from_object(carrier).prepare(scope);
+    let context = callback.relevant_context(scope);
+    report_in_context(scope, context, exception);
+}
+
+pub(super) fn report<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    subscriber: v8::Local<'s, v8::Object>,
+    exception: v8::Local<'s, v8::Value>,
+) {
+    if let Some(context) = subscriber.get_creation_context(scope) {
+        report_in_context(scope, context, exception);
+    }
+}
+
+pub(super) fn report_default_error<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    exception: v8::Local<'s, v8::Value>,
+) {
+    // An internal observer's default error algorithm reports in the current
+    // realm. An error pushed after closure instead uses Subscriber's realm.
+    let context = scope.get_current_context();
+    report_in_context(scope, context, exception);
+}
+
+fn report_in_context<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    context: v8::Local<'s, v8::Context>,
+    exception: v8::Local<'s, v8::Value>,
+) {
+    if !context_is_current(scope, context) {
+        return;
+    }
+    let scope = &mut v8::ContextScope::new(scope, context);
+    let message = v8::Exception::create_message(scope, exception);
+    let report = build_exception_report_without_stack(scope, Some(exception), Some(message));
+    if let Some(host_ptr) = context_host_ptr_from_global_bridge(scope) {
+        let identity =
+            unsafe { &*host_ptr }.window_execution_context_identity_for_v8_context(scope, context);
+        crate::host::report_event_callback_exception(
+            scope,
+            host_ptr,
+            "Observable",
+            identity,
+            None,
+            &report,
+        );
+    } else {
+        crate::worker::dispatch_current_worker_callback_exception(scope, report);
+    }
+}

moli-renderer-v8/src/observable/state.rs

@@ -0,0 +1,204 @@
+use std::{cell::RefCell, collections::HashMap, rc::Rc};
+
+use moli_webapi_declare::WebApiObject;
+
+use super::callbacks;
+use crate::{
+    abort_signal_route::ResolvedAbortSignal,
+    context_bootstrap::ensure_intrinsic_interface_prototype,
+    util::{get_private_value, set_private_value},
+    web_api_interfaces, webidl,
+};
+
+const ID: &str = "__moliObservableId";
+pub(super) const INITIALIZER: &str = "__moliObservableInitializer";
+pub(super) const ACTIVE: &str = "__moliSubscriberActive";
+pub(super) const SIGNAL: &str = "__moliSubscriberSignal";
+pub(super) const OBSERVERS: &str = "__moliSubscriberObservers";
+pub(super) const TEARDOWNS: &str = "__moliSubscriberTeardowns";
+pub(super) const NEXT: &str = "__moliObserverNext";
+pub(super) const ERROR: &str = "__moliObserverError";
+pub(super) const COMPLETE: &str = "__moliObserverComplete";
+pub(super) const INPUT_SIGNAL: &str = "__moliObserverInputSignal";
+pub(super) const ABORT_ALGORITHM: &str = "__moliObserverAbortAlgorithm";
+
+type Store = Rc<RefCell<WeakSubscribers>>;
+
+#[derive(Default)]
+struct WeakSubscribers {
+    next_id: u64,
+    entries: HashMap<u64, Entry>,
+}
+
+struct Entry {
+    _observable: v8::Weak<v8::Object>,
+    subscriber: Option<v8::Weak<v8::Object>>,
+}
+
+pub(super) fn initialize_observable<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    object: v8::Local<'s, v8::Object>,
+    callback: webidl::WebIdlCallbackFunction,
+) {
+    set_callback(scope, object, INITIALIZER, callback);
+    let store = if let Some(store) = scope.get_slot::<Store>() {
+        store.clone()
+    } else {
+        let store = Store::default();
+        scope.set_slot(store.clone());
+        store
+    };
+    let id = {
+        let mut store = store.borrow_mut();
+        store.next_id = store
+            .next_id
+            .checked_add(1)
+            .expect("Observable identity exhausted");
+        store.next_id
+    };
+    let value = v8::BigInt::new_from_u64(scope, id);
+    set_private_value(scope, object, ID, value.into());
+    let weak_store = Rc::downgrade(&store);
+    let weak = v8::Weak::with_finalizer(
+        scope,
+        object,
+        Box::new(move |_| {
+            if let Some(store) = weak_store.upgrade() {
+                store.borrow_mut().entries.remove(&id);
+            }
+        }),
+    );
+    store.borrow_mut().entries.insert(
+        id,
+        Entry {
+            _observable: weak,
+            subscriber: None,
+        },
+    );
+}
+
+fn id<'s>(scope: &mut v8::PinScope<'s, '_>, object: v8::Local<'s, v8::Object>) -> Option<u64> {
+    get_private_value(scope, object, ID)
+        .and_then(|value| v8::Local::<v8::BigInt>::try_from(value).ok())
+        .map(|value| value.u64_value().0)
+}
+
+pub(super) fn current_subscriber<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    observable: v8::Local<'s, v8::Object>,
+) -> Option<v8::Local<'s, v8::Object>> {
+    let id = id(scope, observable)?;
+    scope
+        .get_slot::<Store>()?
+        .borrow()
+        .entries
+        .get(&id)?
+        .subscriber
+        .as_ref()?
+        .to_local(scope)
+}
+
+pub(super) fn set_subscriber<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    observable: v8::Local<'s, v8::Object>,
+    subscriber: v8::Local<'s, v8::Object>,
+) {
+    let id = id(scope, observable).expect("Observable identity");
+    let weak = v8::Weak::new(scope, subscriber);
+    scope
+        .get_slot::<Store>()
+        .unwrap()
+        .borrow_mut()
+        .entries
+        .get_mut(&id)
+        .unwrap()
+        .subscriber = Some(weak);
+}
+
+#[derive(WebApiObject)]
+#[webapi(prototype = "Object", interface = web_api_interfaces::Subscriber)]
+struct SubscriberInstance<'scope> {
+    #[webapi(prototype)]
+    prototype: v8::Local<'scope, v8::Object>,
+    #[webapi(slot = ACTIVE)]
+    active: bool,
+    #[webapi(slot = SIGNAL)]
+    signal: v8::Local<'scope, v8::Object>,
+    #[webapi(slot = OBSERVERS)]
+    observers: v8::Local<'scope, v8::Array>,
+    #[webapi(slot = TEARDOWNS)]
+    teardowns: v8::Local<'scope, v8::Array>,
+}
+
+pub(super) fn new_subscriber<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+) -> Option<v8::Local<'s, v8::Object>> {
+    let prototype = ensure_intrinsic_interface_prototype(scope, "Subscriber").ok()?;
+    let signal = ResolvedAbortSignal::new(scope)?.value();
+    SubscriberInstance::new(
+        prototype,
+        true,
+        signal,
+        v8::Array::new(scope, 0),
+        v8::Array::new(scope, 0),
+    )
+    .bind(scope)
+    .ok()
+}
+
+pub(super) fn object_slot<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    object: v8::Local<'s, v8::Object>,
+    slot: &str,
+) -> Option<v8::Local<'s, v8::Object>> {
+    get_private_value(scope, object, slot)
+        .and_then(|value| v8::Local::<v8::Object>::try_from(value).ok())
+}
+
+pub(super) fn active<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    subscriber: v8::Local<'s, v8::Object>,
+) -> bool {
+    get_private_value(scope, subscriber, ACTIVE).is_some_and(|value| value.is_true())
+}
+
+pub(super) fn list<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    object: v8::Local<'s, v8::Object>,
+    slot: &str,
+) -> Vec<v8::Local<'s, v8::Object>> {
+    let Some(array) = object_slot(scope, object, slot)
+        .and_then(|value| v8::Local::<v8::Array>::try_from(value).ok())
+    else {
+        return Vec::new();
+    };
+    (0..array.length())
+        .filter_map(|i| {
+            array
+                .get_index(scope, i)
+                .and_then(|value| v8::Local::<v8::Object>::try_from(value).ok())
+        })
+        .collect()
+}
+
+pub(super) fn set_list<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    object: v8::Local<'s, v8::Object>,
+    slot: &str,
+    values: &[v8::Local<'s, v8::Object>],
+) {
+    let values: Vec<v8::Local<'s, v8::Value>> =
+        values.iter().map(|value| (*value).into()).collect();
+    let array = v8::Array::new_with_elements(scope, &values);
+    set_private_value(scope, object, slot, array.into());
+}
+
+pub(super) fn set_callback<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+    object: v8::Local<'s, v8::Object>,
+    slot: &str,
+    callback: webidl::WebIdlCallbackFunction,
+) {
+    let callback = callbacks::trace(scope, callback);
+    set_private_value(scope, object, slot, callback.into());
+}

moli-renderer-v8/src/script_vm/tests/mod.rs

@@ -15711,6 +15711,7 @@ mod inspector_unwrap;
 mod lazy_storage;
 mod lazy_window_surfaces;
 mod no_cors_header_fill;
+mod observable;
 mod observer_callbacks;
 mod post_parse;
 mod queue_microtask;

moli-renderer-v8/src/script_vm/tests/observable.rs

@@ -0,0 +1,116 @@
+use super::*;
+
+#[test]
+fn observable_core_lifecycle_conversion_brands_and_exceptions() {
+    let mut vm = new_storage_test_vm("https://observable.test/");
+    let result = vm
+        .eval(&format!(
+            "JSON.stringify({})",
+            include_str!("../../../tests/fixtures/observable-core.js")
+        ))
+        .expect("Observable fixture should evaluate");
+    let result: serde_json::Value = serde_json::from_str(&result).unwrap();
+    assert_eq!(result["failures"], serde_json::json!([]), "{result}");
+    assert!(result["checks"].as_u64().unwrap() >= 65, "{result}");
+}
+
+#[test]
+fn observable_realm_guards_use_the_receiver_and_teardown_owner() {
+    let mut vm = new_storage_test_vm("https://observable-realms.test/");
+    vm.eval(
+        r#"
+(() => {
+  const frame = document.createElement('iframe');
+  frame.id = 'observable-child';
+  (document.body || document.documentElement || document).appendChild(frame);
+})()
+"#,
+    )
+    .unwrap();
+    materialize_single_child_default_realm_for_test(&mut vm, "Observable child realm");
+    let result = vm
+        .eval(
+            r#"
+JSON.stringify((() => {
+  const frame = document.getElementById('observable-child');
+  const child = frame.contentWindow;
+  const events = [];
+  let subscriber, initializers = 0, conversions = 0;
+  const source = new child.Observable(s => { subscriber = s; initializers++; });
+  child.Observable.prototype.subscribe.call(source, value => events.push(value));
+  const checks = [subscriber instanceof child.Subscriber,
+    subscriber.signal instanceof child.AbortSignal];
+  try {
+    child.Observable.prototype.subscribe.call({}, {get next() { conversions++; }});
+    checks.push(false);
+  } catch (error) {
+    checks.push(error instanceof child.TypeError && !(error instanceof TypeError));
+  }
+  checks.push(conversions === 0);
+  Subscriber.prototype.next.call(subscriber, 'live');
+  const pending = new child.Observable(() => events.push('retired initializer'));
+  subscriber.addTeardown(() => events.push('retired teardown'));
+  subscriber.addTeardown(() => { events.push('remove'); frame.remove(); });
+  Subscriber.prototype.complete.call(subscriber);
+  Observable.prototype.subscribe.call(pending);
+  Subscriber.prototype.next.call(subscriber, 'retired next');
+  Subscriber.prototype.addTeardown.call(subscriber, () => events.push('retired added teardown'));
+  checks.push(initializers === 1, !subscriber.active, subscriber.signal.aborted);
+  return {checks, events};
+})())
+"#,
+        )
+        .unwrap();
+    assert_eq!(
+        result,
+        r#"{"checks":[true,true,true,true,true,true,true],"events":["live","remove"]}"#
+    );
+}
+
+#[test]
+fn observable_weak_subscription_and_callback_cycles_are_collectible() {
+    let mut vm = new_storage_test_vm("https://observable-gc.test/");
+    vm.eval(
+        r#"
+(() => {
+  globalThis.__initializers = 0;
+  globalThis.__source = new Observable(s => {
+    __initializers++;
+    globalThis.__weakSubscriber = new WeakRef(s);
+  });
+  __source.subscribe();
+  // Isolate this lexical environment from the live source initializer above.
+  (() => {
+    let cycle;
+    cycle = new Observable(() => cycle);
+    globalThis.__weakCycle = new WeakRef(cycle);
+  })();
+  (() => {
+    const captured = {};
+    globalThis.__weakObserver = new WeakRef(captured);
+    globalThis.__controller = new AbortController();
+    new Observable(s => s.complete()).subscribe(() => captured, {signal: __controller.signal});
+  })();
+})()
+"#,
+    )
+    .unwrap();
+    vm.renderer_document_isolate
+        .clone()
+        .with_entered_renderer_document_isolate(|isolate| {
+            isolate.clear_kept_objects();
+            isolate.low_memory_notification();
+            Ok(())
+        })
+        .unwrap();
+    assert_eq!(
+        vm.eval(
+            r#"
+JSON.stringify([__weakSubscriber.deref() === undefined, __weakCycle.deref() === undefined,
+  __weakObserver.deref() === undefined, (__source.subscribe(), __initializers)])
+"#
+        )
+        .unwrap(),
+        "[true,true,true,2]"
+    );
+}

moli-renderer-v8/src/web_api_interfaces.rs

@@ -316,6 +316,7 @@ interfaces! {
     NotificationEvent: ExtendableEvent;
     OfflineAudioCompletionEvent: Event;
     OfflineAudioContext: BaseAudioContext;
+    Observable;
     OffscreenCanvas;
     OffscreenCanvasRenderingContext2D;
     Option;
@@ -480,6 +481,7 @@ interfaces! {
     StyleSheet;
     StyleSheetList;
     SubmitEvent: Event;
+    Subscriber;
     SubtleCrypto;
     SyncEvent: ExtendableEvent;
     SyncManager;

moli-renderer-v8/src/worker/abort.rs

@@ -332,6 +332,13 @@ fn worker_abort_store(scope: &mut v8::PinScope<'_, '_>) -> Option<Rc<RefCell<Wor
     get_worker_state(scope).map(|state| state.borrow().abort.clone())
 }
 
+pub(crate) fn new_worker_abort_signal<'s>(
+    scope: &mut v8::PinScope<'s, '_>,
+) -> Option<v8::Local<'s, v8::Object>> {
+    let store = worker_abort_store(scope)?;
+    create_signal(scope, &mut store.borrow_mut(), false, None)
+}
+
 pub(crate) fn worker_abort_signal_aborted<'s>(
     scope: &mut v8::PinScope<'s, '_>,
     signal: v8::Local<'s, v8::Object>,

moli-renderer-v8/src/worker/thread/tests/postmessage.rs

@@ -1,5 +1,21 @@
 use super::*;
 
+#[tokio::test]
+async fn worker_observable_core_lifecycle_conversion_brands_and_exceptions() {
+    ensure_v8();
+    let mut handle = spawn_worker(
+        format!(
+            "postMessage({}); close();",
+            include_str!("../../../../tests/fixtures/observable-core.js")
+        ),
+        "https://observable.test/worker.js".into(),
+    );
+    let message = timeout(TIMEOUT, handle.recv()).await.unwrap().unwrap();
+    let result: serde_json::Value = serde_json::from_str(&expect_post_json(message)).unwrap();
+    assert_eq!(result["failures"], serde_json::json!([]), "{result}");
+    assert!(result["checks"].as_u64().unwrap() >= 65, "{result}");
+}
+
 #[tokio::test]
 async fn worker_trusted_types_webidl_surface_checks_descriptors_arguments_and_brands() {
     ensure_v8();

moli-renderer-v8/tests/fixtures/observable-core.js

@@ -0,0 +1,146 @@
+(() => {
+  'use strict';
+  const failures = [];
+  let checks = 0;
+  const check = (ok, label) => { checks++; if (!ok) failures.push(label); };
+  const same = (actual, expected, label) => check(JSON.stringify(actual) === JSON.stringify(expected), label);
+  const throws = (fn, label, expected = TypeError) => {
+    try { fn(); check(false, label); }
+    catch (error) { check(error instanceof expected, label); }
+  };
+  const reported = [];
+  const onError = event => { reported.push(event.error); event.preventDefault(); };
+  addEventListener('error', onError);
+  try {
+    for (const make of [() => Observable(() => {}), () => new Observable(),
+      () => new Observable({}), () => new Subscriber()]) throws(make, 'constructor validation');
+    let calls = 0, subscriber;
+    const source = new Observable(function (value) {
+      check(this === undefined && arguments.length === 1, 'initializer callback this/arguments');
+      calls++;
+      subscriber = value;
+    });
+    check(calls === 0, 'construction is lazy');
+    const order = [];
+    const controller1 = new AbortController(), controller2 = new AbortController();
+    source.subscribe({ next(value) { check(this === undefined, 'observer this'); order.push('a:' + value); } }, {signal: controller1.signal});
+    const signal = subscriber.signal;
+    check(signal !== controller1.signal && signal === subscriber.signal, 'stable independent signal');
+    check(subscriber.active && !signal.aborted, 'active subscription');
+    source.subscribe(value => order.push('b:' + value), {signal: controller2.signal});
+    check(calls === 1, 'shared producer');
+    subscriber.addTeardown(() => { order.push('first teardown'); check(!subscriber.active && signal.aborted, 'teardown sees closed state'); });
+    subscriber.addTeardown(() => order.push('second teardown'));
+    signal.addEventListener('abort', () => order.push('abort'));
+    subscriber.next(1);
+    controller1.abort('one');
+    check(subscriber.active && !signal.aborted, 'first cancellation keeps producer active');
+    subscriber.next(2);
+    controller2.abort('two');
+    check(!subscriber.active && signal.reason === 'two', 'last cancellation closes with reason');
+    subscriber.next(3);
+    subscriber.complete();
+    same(order, ['a:1', 'b:1', 'b:2', 'abort', 'second teardown', 'first teardown'], 'cancellation/teardown order');
+    source.subscribe();
+    check(calls === 2 && subscriber.signal !== signal, 'restart after cancellation');
+    subscriber.complete();
+
+    const converted = [];
+    const observer = {};
+    for (const name of ['next', 'error', 'complete']) Object.defineProperty(observer, name, {get() { converted.push(name); return undefined; }});
+    source.subscribe(observer, {get signal() { converted.push('signal'); return undefined; }});
+    same(converted, ['complete', 'error', 'next', 'signal'], 'dictionary conversion order');
+    subscriber.complete();
+    for (const observer of [1, true, 'text', {next: null}, {error: 1}, {complete: {}}]) throws(() => source.subscribe(observer), 'invalid observer');
+    for (const signal of [null, {}, new Proxy(new AbortController().signal, {})]) throws(() => source.subscribe({}, {signal}), 'signal brand');
+    for (const options of [1, 'text', true]) throws(() => source.subscribe({}, options), 'invalid options');
+    const marker = {marker: true};
+    try { source.subscribe({get next() { throw marker; }}); check(false, 'getter throws'); }
+    catch (error) { check(error === marker, 'getter preserves exception'); }
+    let conversions = 0, traps = 0;
+    const revoked = Proxy.revocable(source, {}); revoked.revoke();
+    for (const fake of [{}, Object.create(source), new Proxy(source, {get() { traps++; }}), revoked.proxy]) {
+      throws(() => Observable.prototype.subscribe.call(fake, {get next() { conversions++; }}), 'Observable receiver');
+    }
+    check(conversions === 0 && traps === 0, 'receiver precedes conversion and proxy traps');
+    for (const fake of [{}, Object.create(subscriber), new Proxy(subscriber, {})]) {
+      for (const name of ['next', 'error', 'complete', 'addTeardown']) throws(() => Subscriber.prototype[name].call(fake, () => {}), 'Subscriber method receiver');
+      for (const name of ['active', 'signal']) throws(() => Object.getOwnPropertyDescriptor(Subscriber.prototype, name).get.call(fake), 'Subscriber getter receiver');
+    }
+    throws(() => subscriber.next(), 'next requires value');
+    throws(() => subscriber.error(), 'error requires value');
+    throws(() => subscriber.addTeardown(null), 'teardown requires callback');
+
+    const preAborted = AbortSignal.abort(marker), closed = [];
+    new Observable(s => {
+      check(!s.active && s.signal.aborted && s.signal.reason === marker, 'pre-aborted initializer runs inactive');
+      s.addTeardown(() => closed.push('a'));
+      s.addTeardown(() => closed.push('b'));
+      s.next(1); s.complete();
+    }).subscribe(() => closed.push('unexpected'), {signal: preAborted});
+    same(closed, ['a', 'b'], 'inactive teardown runs synchronously');
+
+    const notifications = [];
+    let shared;
+    const multicast = new Observable(s => { shared = s; });
+    const cancelled = new AbortController();
+    multicast.subscribe(value => {
+      notifications.push('a' + value);
+      if (value === 1) {
+        multicast.subscribe(value => notifications.push('c' + value));
+        cancelled.abort();
+      }
+    });
+    multicast.subscribe(value => notifications.push('b' + value), {signal: cancelled.signal});
+    shared.next(1); shared.next(2);
+    same(notifications, ['a1', 'b1', 'a2', 'c2'], 'notification uses observer snapshot');
+    shared.complete();
+
+    const exceptions = [new Error('initializer'), new Error('late'), new Error('observer'), new Error('teardown')];
+    let handled;
+    new Observable(() => { throw exceptions[0]; }).subscribe({error(error) { handled = error; }});
+    check(handled === exceptions[0] && reported.length === 0, 'initializer exception goes to observer');
+    new Observable(s => { s.complete(); s.error(exceptions[1]); }).subscribe();
+    check(reported.pop() === exceptions[1], 'late error is reported');
+    const lifecycle = [];
+    new Observable(s => {
+      s.addTeardown(() => lifecycle.push('first'));
+      s.addTeardown(() => { lifecycle.push('throwing'); throw exceptions[3]; });
+      s.addTeardown(() => { lifecycle.push('last'); s.addTeardown(() => lifecycle.push('nested')); s.complete(); });
+      s.next(1);
+      s.complete();
+    }).subscribe({next() { throw exceptions[2]; }, complete() { lifecycle.push('complete'); }});
+    same(lifecycle, ['last', 'nested', 'throwing', 'first', 'complete'], 'reentrant teardown and exceptions');
+    check(reported.shift() === exceptions[2] && reported.shift() === exceptions[3], 'callback exceptions reported in order');
+
+    // Author changes to the public constructors/methods must not redirect the
+    // native close algorithm or creation of its signal.
+    const savedAbortController = globalThis.AbortController;
+    const savedAbort = AbortController.prototype.abort;
+    const savedComplete = Subscriber.prototype.complete;
+    const savedError = Subscriber.prototype.error;
+    const savedSignal = Object.getOwnPropertyDescriptor(AbortController.prototype, 'signal');
+    try {
+      globalThis.AbortController = () => { throw marker; };
+      savedAbortController.prototype.abort = () => { throw marker; };
+      Object.defineProperty(savedAbortController.prototype, 'signal', {configurable: true, get() { throw marker; }});
+      Subscriber.prototype.complete = Subscriber.prototype.error = () => { throw marker; };
+      let completed = false, seen;
+      new Observable(s => { savedComplete.call(s); }).subscribe({complete() { completed = true; }});
+      new Observable(() => { throw marker; }).subscribe({error(value) { seen = value; }});
+      check(completed && seen === marker, 'internal algorithms bypass public methods');
+    } finally {
+      globalThis.AbortController = savedAbortController;
+      savedAbortController.prototype.abort = savedAbort;
+      Object.defineProperty(savedAbortController.prototype, 'signal', savedSignal);
+      Subscriber.prototype.complete = savedComplete;
+      Subscriber.prototype.error = savedError;
+    }
+    check(reported.length === 0, 'no unexpected callback errors');
+  } catch (error) {
+    failures.push('uncaught: ' + error);
+  } finally {
+    removeEventListener('error', onError);
+  }
+  return {checks, failures};
+})()