diff --git a/api_docs/changelog.md b/api_docs/changelog.md
index 776f221683..926b83eb3d 100644
--- a/api_docs/changelog.md
+++ b/api_docs/changelog.md
@@ -20,6 +20,14 @@ format used by the Zulip server that they are interacting with.
 
 ## Changes in Zulip 8.0
 
+**Feature level 236**
+
+* [`POST /messages`](/api/send-message), [`POST
+  /scheduled_messages`](/api/create-scheduled-message): The new
+  `read_by_sender` parameter lets the client override the heuristic
+  that determines whether the new message will be initially marked
+  read by its sender.
+
 **Feature level 235**
 
 * [`PATCH /realm/user_settings_defaults`](/api/update-realm-user-settings-defaults),
diff --git a/version.py b/version.py
index 519dd98865..f3d4e7227a 100644
--- a/version.py
+++ b/version.py
@@ -33,7 +33,7 @@ DESKTOP_WARNING_VERSION = "5.9.3"
 # Changes should be accompanied by documentation explaining what the
 # new level means in api_docs/changelog.md, as well as "**Changes**"
 # entries in the endpoint's documentation in `zulip.yaml`.
-API_FEATURE_LEVEL = 235
+API_FEATURE_LEVEL = 236
 
 # Bump the minor PROVISION_VERSION to indicate that folks should provision
 # only when going from an old version of the code to a newer version. Bump
diff --git a/zerver/actions/message_send.py b/zerver/actions/message_send.py
index d16eedc791..6a3ccb08a0 100644
--- a/zerver/actions/message_send.py
+++ b/zerver/actions/message_send.py
@@ -732,13 +732,11 @@ def create_user_messages(
     followed_topic_email_user_ids: AbstractSet[int],
     mark_as_read_user_ids: Set[int],
     limit_unread_user_ids: Optional[Set[int]],
-    scheduled_message_to_self: bool,
     topic_participant_user_ids: Set[int],
 ) -> List[UserMessageLite]:
     # These properties on the Message are set via
     # render_markdown by code in the Markdown inline patterns
     ids_with_alert_words = rendering_result.user_ids_with_alert_words
-    sender_id = message.sender.id
     is_stream_message = message.is_stream_message()
 
     base_flags = 0
@@ -769,17 +767,8 @@ def create_user_messages(
     user_messages = []
     for user_profile_id in um_eligible_user_ids:
         flags = base_flags
-        if (
-            (
-                # Messages you sent from a non-API client are
-                # automatically marked as read for yourself; scheduled
-                # messages to yourself only are not.
-                user_profile_id == sender_id
-                and message.sending_client.default_read_by_sender()
-                and not scheduled_message_to_self
-            )
-            or user_profile_id in mark_as_read_user_ids
-            or (limit_unread_user_ids is not None and user_profile_id not in limit_unread_user_ids)
+        if user_profile_id in mark_as_read_user_ids or (
+            limit_unread_user_ids is not None and user_profile_id not in limit_unread_user_ids
         ):
             flags |= UserMessage.flags.read
         if user_profile_id in mentioned_user_ids:
@@ -865,7 +854,6 @@ def do_send_messages(
     send_message_requests_maybe_none: Sequence[Optional[SendMessageRequest]],
     *,
     email_gateway: bool = False,
-    scheduled_message_to_self: bool = False,
     mark_as_read: Sequence[int] = [],
 ) -> List[SentMessageResult]:
     """See
@@ -915,7 +903,6 @@ def do_send_messages(
                 followed_topic_email_user_ids=send_request.followed_topic_email_user_ids,
                 mark_as_read_user_ids=mark_as_read_user_ids,
                 limit_unread_user_ids=send_request.limit_unread_user_ids,
-                scheduled_message_to_self=scheduled_message_to_self,
                 topic_participant_user_ids=send_request.topic_participant_user_ids,
             )
 
@@ -1345,11 +1332,15 @@ def check_send_stream_message(
     stream_name: str,
     topic: str,
     body: str,
+    *,
     realm: Optional[Realm] = None,
+    read_by_sender: bool = False,
 ) -> int:
     addressee = Addressee.for_stream_name(stream_name, topic)
     message = check_message(sender, client, addressee, body, realm)
-    sent_message_result = do_send_messages([message])[0]
+    sent_message_result = do_send_messages(
+        [message], mark_as_read=[sender.id] if read_by_sender else []
+    )[0]
     return sent_message_result.message_id
 
 
@@ -1394,6 +1385,7 @@ def check_send_message(
     widget_content: Optional[str] = None,
     *,
     skip_stream_access_check: bool = False,
+    read_by_sender: bool = False,
 ) -> SentMessageResult:
     addressee = Addressee.legacy_build(sender, recipient_type_name, message_to, topic_name)
     try:
@@ -1413,7 +1405,7 @@ def check_send_message(
         )
     except ZephyrMessageAlreadySentError as e:
         return SentMessageResult(message_id=e.message_id)
-    return do_send_messages([message])[0]
+    return do_send_messages([message], mark_as_read=[sender.id] if read_by_sender else [])[0]
 
 
 def send_rate_limited_pm_notification_to_bot_owner(
diff --git a/zerver/actions/scheduled_messages.py b/zerver/actions/scheduled_messages.py
index cc5a9adbe4..f999331182 100644
--- a/zerver/actions/scheduled_messages.py
+++ b/zerver/actions/scheduled_messages.py
@@ -43,7 +43,9 @@ def check_schedule_message(
     message_content: str,
     deliver_at: datetime,
     realm: Optional[Realm] = None,
+    *,
     forwarder_user_profile: Optional[UserProfile] = None,
+    read_by_sender: Optional[bool] = None,
 ) -> int:
     addressee = Addressee.legacy_build(sender, recipient_type_name, message_to, topic_name)
     send_request = check_message(
@@ -56,11 +58,22 @@ def check_schedule_message(
     )
     send_request.deliver_at = deliver_at
 
-    return do_schedule_messages([send_request], sender)[0]
+    if read_by_sender is None:
+        # Legacy default: a scheduled message you sent from a non-API client is
+        # automatically marked as read for yourself, unless it was sent to
+        # yourself only.
+        read_by_sender = (
+            client.default_read_by_sender() and send_request.message.recipient != sender.recipient
+        )
+
+    return do_schedule_messages([send_request], sender, read_by_sender=read_by_sender)[0]
 
 
 def do_schedule_messages(
-    send_message_requests: Sequence[SendMessageRequest], sender: UserProfile
+    send_message_requests: Sequence[SendMessageRequest],
+    sender: UserProfile,
+    *,
+    read_by_sender: bool = False,
 ) -> List[int]:
     scheduled_messages: List[Tuple[ScheduledMessage, SendMessageRequest]] = []
 
@@ -80,6 +93,7 @@ def do_schedule_messages(
         scheduled_message.realm = send_request.realm
         assert send_request.deliver_at is not None
         scheduled_message.scheduled_timestamp = send_request.deliver_at
+        scheduled_message.read_by_sender = read_by_sender
         scheduled_message.delivery_type = ScheduledMessage.SEND_LATER
 
         scheduled_messages.append((scheduled_message, send_request))
@@ -301,9 +315,19 @@ def send_scheduled_message(scheduled_message: ScheduledMessage) -> None:
         scheduled_message.realm,
     )
 
-    scheduled_message_to_self = scheduled_message.recipient == scheduled_message.sender.recipient
+    read_by_sender = scheduled_message.read_by_sender
+    if read_by_sender is None:  # nocoverage
+        # Legacy default: a scheduled message you sent from a non-API client is
+        # automatically marked as read for yourself, unless it was sent to
+        # yourself only.
+        read_by_sender = (
+            scheduled_message.sending_client.default_read_by_sender()
+            and scheduled_message.recipient != scheduled_message.sender.recipient
+        )
+
     sent_message_result = do_send_messages(
-        [send_request], scheduled_message_to_self=scheduled_message_to_self
+        [send_request],
+        mark_as_read=[scheduled_message.sender_id] if read_by_sender else [],
     )[0]
     scheduled_message.delivered_message_id = sent_message_result.message_id
     scheduled_message.delivered = True
diff --git a/zerver/lib/test_classes.py b/zerver/lib/test_classes.py
index 9cf8d2acac..9dc6663f94 100644
--- a/zerver/lib/test_classes.py
+++ b/zerver/lib/test_classes.py
@@ -1075,10 +1075,11 @@ Output:
         from_user: UserProfile,
         to_user: UserProfile,
         content: str = "test content",
-        sending_client_name: str = "test suite",
+        *,
+        read_by_sender: bool = True,
     ) -> int:
         recipient_list = [to_user.id]
-        (sending_client, _) = Client.objects.get_or_create(name=sending_client_name)
+        (sending_client, _) = Client.objects.get_or_create(name="test suite")
 
         sent_message_result = check_send_message(
             from_user,
@@ -1087,6 +1088,7 @@ Output:
             recipient_list,
             None,
             content,
+            read_by_sender=read_by_sender,
         )
         return sent_message_result.message_id
 
@@ -1095,12 +1097,13 @@ Output:
         from_user: UserProfile,
         to_users: List[UserProfile],
         content: str = "test content",
-        sending_client_name: str = "test suite",
+        *,
+        read_by_sender: bool = True,
     ) -> int:
         to_user_ids = [u.id for u in to_users]
         assert len(to_user_ids) >= 2
 
-        (sending_client, _) = Client.objects.get_or_create(name=sending_client_name)
+        (sending_client, _) = Client.objects.get_or_create(name="test suite")
 
         sent_message_result = check_send_message(
             from_user,
@@ -1109,6 +1112,7 @@ Output:
             to_user_ids,
             None,
             content,
+            read_by_sender=read_by_sender,
         )
         return sent_message_result.message_id
 
@@ -1119,10 +1123,11 @@ Output:
         content: str = "test content",
         topic_name: str = "test",
         recipient_realm: Optional[Realm] = None,
-        sending_client_name: str = "test suite",
+        *,
         allow_unsubscribed_sender: bool = False,
+        read_by_sender: bool = True,
     ) -> int:
-        (sending_client, _) = Client.objects.get_or_create(name=sending_client_name)
+        (sending_client, _) = Client.objects.get_or_create(name="test suite")
 
         message_id = check_send_stream_message(
             sender=sender,
@@ -1131,6 +1136,7 @@ Output:
             topic=topic_name,
             body=content,
             realm=recipient_realm,
+            read_by_sender=read_by_sender,
         )
         if (
             not UserMessage.objects.filter(user_profile=sender, message_id=message_id).exists()
diff --git a/zerver/migrations/0495_scheduledmessage_read_by_sender.py b/zerver/migrations/0495_scheduledmessage_read_by_sender.py
new file mode 100644
index 0000000000..6563cde63d
--- /dev/null
+++ b/zerver/migrations/0495_scheduledmessage_read_by_sender.py
@@ -0,0 +1,17 @@
+# Generated by Django 4.2.8 on 2023-12-14 00:09
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("zerver", "0494_realmuserdefault_automatically_follow_topics_where_mentioned_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="scheduledmessage",
+            name="read_by_sender",
+            field=models.BooleanField(null=True),
+        ),
+    ]
diff --git a/zerver/models.py b/zerver/models.py
index f3d8d31d7b..0ac574088d 100644
--- a/zerver/models.py
+++ b/zerver/models.py
@@ -4629,6 +4629,7 @@ class ScheduledMessage(models.Model):
     stream = models.ForeignKey(Stream, null=True, on_delete=CASCADE)
     realm = models.ForeignKey(Realm, on_delete=CASCADE)
     scheduled_timestamp = models.DateTimeField(db_index=True)
+    read_by_sender = models.BooleanField(null=True)
     delivered = models.BooleanField(default=False)
     delivered_message = models.ForeignKey(Message, null=True, on_delete=CASCADE)
     has_attachment = models.BooleanField(default=False, db_index=True)
diff --git a/zerver/openapi/zulip.yaml b/zerver/openapi/zulip.yaml
index c926766603..caee67bd0c 100644
--- a/zerver/openapi/zulip.yaml
+++ b/zerver/openapi/zulip.yaml
@@ -5606,6 +5606,17 @@ paths:
             in UTC seconds.
           required: true
           example: 3165826990
+        - name: read_by_sender
+          in: query
+          schema:
+            type: boolean
+          description: |
+            Whether the message should be initially marked read by its
+            sender. If unspecified, the server uses a heuristic based
+            on the client name and the recipient.
+
+            **Changes**: New in Zulip 8.0 (feature level 236).
+          example: true
       responses:
         "200":
           description: Success.
@@ -6373,6 +6384,17 @@ paths:
             chosen freely by the client; the server will pass it back to the client without
             inspecting it, as described in the `queue_id` description.
           example: "100.01"
+        - name: read_by_sender
+          in: query
+          schema:
+            type: boolean
+          description: |
+            Whether the message should be initially marked read by its
+            sender. If unspecified, the server uses a heuristic based
+            on the client name.
+
+            **Changes**: New in Zulip 8.0 (feature level 236).
+          example: true
       responses:
         "200":
           description: Success.
diff --git a/zerver/tests/test_digest.py b/zerver/tests/test_digest.py
index f09f907e61..5c2eb3e498 100644
--- a/zerver/tests/test_digest.py
+++ b/zerver/tests/test_digest.py
@@ -33,7 +33,6 @@ from zerver.models import (
     Stream,
     UserActivityInterval,
     UserProfile,
-    get_client,
     get_realm,
     get_stream,
 )
@@ -547,8 +546,6 @@ class TestDigestEmailMessages(ZulipTestCase):
         self.assertEqual(stream_info["html"], [])
 
     def simulate_stream_conversation(self, stream: str, senders: List[str]) -> List[int]:
-        client = "website"  # this makes `default_read_by_sender` return True
-        sending_client = get_client(client)
         message_ids = []  # List[int]
         for sender_name in senders:
             sender = self.example_user(sender_name)
@@ -556,7 +553,6 @@ class TestDigestEmailMessages(ZulipTestCase):
             content = f"some content for {stream} from {sender_name}"
             message_id = self.send_stream_message(sender, stream, content)
             message_ids.append(message_id)
-        Message.objects.filter(id__in=message_ids).update(sending_client=sending_client)
         return message_ids
 
 
diff --git a/zerver/tests/test_message_flags.py b/zerver/tests/test_message_flags.py
index ed06849974..abb5c04436 100644
--- a/zerver/tests/test_message_flags.py
+++ b/zerver/tests/test_message_flags.py
@@ -874,16 +874,11 @@ class GetUnreadMsgsTest(ZulipTestCase):
             message_id = self.send_personal_message(
                 from_user=hamlet,
                 to_user=other_user,
-                sending_client_name="some_api_program",
+                read_by_sender=False,
             )
-
-            # Check our test setup is correct--the message should
-            # not have looked like it was sent by a human.
             message = Message.objects.get(id=message_id)
-            self.assertFalse(message.sending_client.default_read_by_sender())
 
-            # And since it was not sent by a human, it should not
-            # be read, not even by the sender (Hamlet).
+            # This message should not be read, not even by the sender (Hamlet).
             um = UserMessage.objects.get(
                 user_profile_id=hamlet.id,
                 message_id=message_id,
diff --git a/zerver/views/message_send.py b/zerver/views/message_send.py
index e70a6aeda4..ca25345158 100644
--- a/zerver/views/message_send.py
+++ b/zerver/views/message_send.py
@@ -19,7 +19,7 @@ from zerver.lib.message import render_markdown
 from zerver.lib.request import REQ, RequestNotes, has_request_variables
 from zerver.lib.response import json_success
 from zerver.lib.topic import REQ_topic
-from zerver.lib.validator import check_string_in, to_float
+from zerver.lib.validator import check_bool, check_string_in, to_float
 from zerver.lib.zcommand import process_zcommands
 from zerver.lib.zephyr import compute_mit_user_fullname
 from zerver.models import Client, Message, RealmDomain, UserProfile, get_user_including_cross_realm
@@ -136,6 +136,7 @@ def send_message_backend(
     local_id: Optional[str] = REQ(default=None),
     queue_id: Optional[str] = REQ(default=None),
     time: Optional[float] = REQ(default=None, converter=to_float, documentation_pending=True),
+    read_by_sender: Optional[bool] = REQ(json_validator=check_bool, default=None),
 ) -> HttpResponse:
     recipient_type_name = req_type
     if recipient_type_name == "direct":
@@ -221,6 +222,11 @@ def send_message_backend(
             raise JsonableError(_("Invalid mirrored message"))
         sender = user_profile
 
+    if read_by_sender is None:
+        # Legacy default: a message you sent from a non-API client is
+        # automatically marked as read for yourself.
+        read_by_sender = client.default_read_by_sender()
+
     data: Dict[str, int] = {}
     sent_message_result = check_send_message(
         sender,
@@ -236,6 +242,7 @@ def send_message_backend(
         local_id=local_id,
         sender_queue_id=queue_id,
         widget_content=widget_content,
+        read_by_sender=read_by_sender,
     )
     data["id"] = sent_message_result.message_id
     if sent_message_result.automatic_new_visibility_policy:
diff --git a/zerver/views/scheduled_messages.py b/zerver/views/scheduled_messages.py
index b192799baa..e4d1a669a1 100644
--- a/zerver/views/scheduled_messages.py
+++ b/zerver/views/scheduled_messages.py
@@ -16,7 +16,7 @@ from zerver.lib.response import json_success
 from zerver.lib.scheduled_messages import get_undelivered_scheduled_messages
 from zerver.lib.timestamp import timestamp_to_datetime
 from zerver.lib.topic import REQ_topic
-from zerver.lib.validator import check_int, check_string_in, to_non_negative_int
+from zerver.lib.validator import check_bool, check_int, check_string_in, to_non_negative_int
 from zerver.models import Message, UserProfile
 
 
@@ -115,6 +115,7 @@ def create_scheduled_message_backend(
     topic_name: Optional[str] = REQ_topic(),
     message_content: str = REQ("content"),
     scheduled_delivery_timestamp: int = REQ(json_validator=check_int),
+    read_by_sender: Optional[bool] = REQ(json_validator=check_bool, default=None),
 ) -> HttpResponse:
     recipient_type_name = req_type
     if recipient_type_name == "direct":
@@ -147,5 +148,6 @@ def create_scheduled_message_backend(
         deliver_at,
         realm=user_profile.realm,
         forwarder_user_profile=user_profile,
+        read_by_sender=read_by_sender,
     )
     return json_success(request, data={"scheduled_message_id": scheduled_message_id})