diff --git a/api_docs/changelog.md b/api_docs/changelog.md
index 52c8e7e619..4ae4fa6c64 100644
--- a/api_docs/changelog.md
+++ b/api_docs/changelog.md
@@ -20,6 +20,16 @@ format used by the Zulip server that they are interacting with.
 
 ## Changes in Zulip 9.0
 
+**Feature level 271**
+
+* [`GET /messages`](/api/get-messages),
+  [`GET /messages/matches_narrow`](/api/check-messages-match-narrow),
+  [`POST /messages/flags/narrow`](/api/update-message-flags-for-narrow),
+  [`POST /register`](/api/register-queue):
+  Added support for a new [search/narrow filter](/api/construct-narrow),
+  `with`, which returns messages in the same channel and topic as that
+  of the operand of filter, with the first unread message as anchor.
+
 **Feature level 270**
 
 * `PATCH /realm`, [`POST /register`](/api/register-queue),
diff --git a/api_docs/construct-narrow.md b/api_docs/construct-narrow.md
index d63c768cd1..fc03b0eaf2 100644
--- a/api_docs/construct-narrow.md
+++ b/api_docs/construct-narrow.md
@@ -51,7 +51,10 @@ important optimization when fetching messages in certain cases (e.g.,
 when [adding the `read` flag to a user's personal
 messages](/api/update-message-flags-for-narrow)).
 
-**Changes**: In Zulip 9.0 (feature level 265), support was added for a
+**Changes**: In Zulip 9.0 (feature level 271), narrows gained support
+for a new `with` operator for linking to topics.
+
+In Zulip 9.0 (feature level 265), support was added for a
 new `is:followed` filter, matching messages in topics that the current
 user is [following](/help/follow-a-topic).
 
@@ -88,15 +91,24 @@ filters did.
 
 ### Message IDs
 
-The `near` and `id` operators, documented in the help center, use message
-IDs for their operands.
+The `with` operator is designed to be used in links that should follow
+a topic across being moved. Its operand is a message ID.
 
+The `near` and `id` operators, documented in the help center,
+also use message IDs for their operands.
+
+* `with:12345`: Search for the conversation (stream/topic pair) which
+  contains the message with ID `12345`. If such a message exists, and
+  can be accessed by the user, then the search will be treated as having
+  the `channel`/`topic`/`dm` operators corresponding to the
+  conversation containing that message, replacing any such operators
+  in the original request.
 * `near:12345`: Search messages around the message with ID `12345`.
 * `id:12345`: Search for only message with ID `12345`.
 
-The message ID operand for the `id` operator may be encoded as either a
-number or a string. The message ID operand for the `near` operator must
-be encoded as a string.
+The message ID operand for the `with` and `id` operators may be encoded
+as either a number or a string. The message ID operand for the `near`
+operator must be encoded as a string.
 
 **Changes**: Prior to Zulip 8.0 (feature level 194), the message ID
 operand for the `id` operator needed to be encoded as a string.
diff --git a/version.py b/version.py
index 48f33a7f37..6256378b44 100644
--- a/version.py
+++ b/version.py
@@ -34,7 +34,7 @@ DESKTOP_WARNING_VERSION = "5.9.3"
 # new level means in api_docs/changelog.md, as well as "**Changes**"
 # entries in the endpoint's documentation in `zulip.yaml`.
 
-API_FEATURE_LEVEL = 270  # Last bumped for direct_message_permission_group
+API_FEATURE_LEVEL = 271  # Last bumped for `with` operator.
 
 
 # Bump the minor PROVISION_VERSION to indicate that folks should provision
diff --git a/web/src/filter.ts b/web/src/filter.ts
index 00a29378ca..e226dbeb55 100644
--- a/web/src/filter.ts
+++ b/web/src/filter.ts
@@ -295,12 +295,12 @@ export class Filter {
     _sorted_term_types?: string[] = undefined;
     _predicate?: (message: Message) => boolean;
     _can_mark_messages_read?: boolean;
+    requires_adjustment_for_moved_with_target?: boolean;
 
     constructor(terms: NarrowTerm[]) {
-        this._terms = this.fix_terms(terms);
-        if (this.has_operator("channel")) {
-            this._sub = stream_data.get_sub_by_name(this.operands("channel")[0]!);
-        }
+        this._terms = terms;
+        this.setup_filter(terms);
+        this.requires_adjustment_for_moved_with_target = this.has_operator("with");
     }
 
     static canonicalize_operator(operator: string): string {
@@ -389,6 +389,38 @@ export class Filter {
         };
     }
 
+    static ensure_channel_topic_terms(orig_terms: NarrowTerm[]): NarrowTerm[] {
+        // In presence of `with` term without channel or topic terms in the narrow, the
+        // narrow is populated with the channel and toipic terms through this operation,
+        // so that `with` can be used as a standalone operator to target conversation.
+        const with_term = orig_terms.find((term: NarrowTerm) => term.operator === "with");
+
+        if (!with_term) {
+            return orig_terms;
+        }
+
+        const updated_terms = [...orig_terms];
+        let channel_term = updated_terms.find(
+            (term: NarrowTerm) => Filter.canonicalize_operator(term.operator) === "channel",
+        );
+        let topic_term = updated_terms.find(
+            (term: NarrowTerm) => Filter.canonicalize_operator(term.operator) === "topic",
+        );
+
+        if (!topic_term) {
+            topic_term = {operator: "topic", operand: ""};
+            const with_index = updated_terms.indexOf(with_term);
+            updated_terms.splice(with_index, 0, topic_term);
+        }
+
+        if (!channel_term) {
+            channel_term = {operator: "channel", operand: ""};
+            const topic_index = updated_terms.indexOf(topic_term);
+            updated_terms.splice(topic_index, 0, channel_term);
+        }
+        return updated_terms;
+    }
+
     /* We use a variant of URI encoding which looks reasonably
        nice and still handles unambiguously cases such as
        spaces in operands.
@@ -576,6 +608,7 @@ export class Filter {
             "channels-public",
             "channel",
             "topic",
+            "with",
             "dm",
             "dm-including",
             "sender",
@@ -775,6 +808,11 @@ export class Filter {
 
     static adjusted_terms_if_moved(raw_terms: NarrowTerm[], message: Message): NarrowTerm[] | null {
         if (message.type !== "stream") {
+            // BUG: We should be replacing the channel/topic/dm
+            // operators with the singular dm operator for the
+            // conversation in question. This isn't very important,
+            // because direct messages cannot be moved, but it should
+            // be fixed, since it is triggerable incorrect behavior.
             return null;
         }
 
@@ -784,6 +822,8 @@ export class Filter {
         const adjusted_terms = [];
         let terms_changed = false;
 
+        raw_terms = Filter.ensure_channel_topic_terms(raw_terms);
+
         for (const term of raw_terms) {
             const adjusted_term = {...term};
             if (
@@ -812,6 +852,13 @@ export class Filter {
         return adjusted_terms;
     }
 
+    setup_filter(terms: NarrowTerm[]): void {
+        this._terms = this.fix_terms(terms);
+        if (this.has_operator("channel")) {
+            this._sub = stream_data.get_sub_by_name(this.operands("channel")[0]!);
+        }
+    }
+
     equals(filter: Filter, excluded_operators?: string[]): boolean {
         return _.isEqual(
             filter.sorted_terms(excluded_operators),
@@ -929,6 +976,7 @@ export class Filter {
             "channels-web-public",
             "not-channels-web-public",
             "near",
+            "with",
         ]);
 
         for (const term of term_types) {
@@ -963,6 +1011,10 @@ export class Filter {
         // it is limited by the user's message history. Therefore, we check "channel"
         // and "topic" together to ensure that the current filter will return all the
         // messages of a conversation.
+        if (_.isEqual(term_types, ["channel", "topic", "with"])) {
+            return true;
+        }
+
         if (_.isEqual(term_types, ["channel", "topic"])) {
             return true;
         }
@@ -1180,6 +1232,7 @@ export class Filter {
         const term_types = this.sorted_term_types();
         if (
             (term_types.length === 3 && _.isEqual(term_types, ["channel", "topic", "near"])) ||
+            (term_types.length === 3 && _.isEqual(term_types, ["channel", "topic", "with"])) ||
             (term_types.length === 2 && _.isEqual(term_types, ["channel", "topic"])) ||
             (term_types.length === 1 && _.isEqual(term_types, ["channel"]))
         ) {
@@ -1362,10 +1415,12 @@ export class Filter {
     fix_terms(terms: NarrowTerm[]): NarrowTerm[] {
         terms = this._canonicalize_terms(terms);
         terms = this._fix_redundant_is_private(terms);
+        terms = this._fix_redundant_with_dm(terms);
         return terms;
     }
 
     _fix_redundant_is_private(terms: NarrowTerm[]): NarrowTerm[] {
+        // Every DM is a DM, so drop `is:dm` if on a DM conversation.
         if (!terms.some((term) => Filter.term_type(term) === "dm")) {
             return terms;
         }
@@ -1373,6 +1428,16 @@ export class Filter {
         return terms.filter((term) => Filter.term_type(term) !== "is-dm");
     }
 
+    _fix_redundant_with_dm(terms: NarrowTerm[]): NarrowTerm[] {
+        // Because DMs can't move, the `with` operator is a noop on a
+        // DM conversation.
+        if (terms.some((term) => Filter.term_type(term) === "dm")) {
+            return terms.filter((term) => Filter.term_type(term) !== "with");
+        }
+
+        return terms;
+    }
+
     _canonicalize_terms(terms_mixed_case: NarrowTerm[]): NarrowTerm[] {
         return terms_mixed_case.map((term: NarrowTerm) => Filter.canonicalize_term(term));
     }
@@ -1512,4 +1577,27 @@ export class Filter {
             !this.has_operand("is", "starred")
         );
     }
+
+    try_adjusting_for_moved_with_target(message?: Message): void {
+        // If we have the message named in a `with` operator
+        // available, either via parameter or message_store,
+        if (!this.requires_adjustment_for_moved_with_target) {
+            return;
+        }
+
+        if (!message) {
+            const message_id = Number.parseInt(this.operands("with")[0]!, 10);
+            message = message_store.get(message_id);
+        }
+
+        if (!message) {
+            return;
+        }
+
+        const adjusted_terms = Filter.adjusted_terms_if_moved(this._terms, message);
+        if (adjusted_terms) {
+            this.setup_filter(adjusted_terms);
+        }
+        this.requires_adjustment_for_moved_with_target = false;
+    }
 }
diff --git a/web/src/hash_parser.ts b/web/src/hash_parser.ts
index b78a24349b..7b75f19f63 100644
--- a/web/src/hash_parser.ts
+++ b/web/src/hash_parser.ts
@@ -127,6 +127,7 @@ export const allowed_web_public_narrows = [
     "search",
     "near",
     "id",
+    "with",
 ];
 
 export function is_spectator_compatible(hash: string): boolean {
diff --git a/web/src/message_fetch.js b/web/src/message_fetch.js
index 2d9b7e55c5..6bf8f0b303 100644
--- a/web/src/message_fetch.js
+++ b/web/src/message_fetch.js
@@ -80,6 +80,9 @@ function process_result(data, opts) {
 
     if (messages.length !== 0) {
         if (opts.msg_list) {
+            if (opts.validate_filter_topic_post_fetch) {
+                opts.msg_list.data.filter.try_adjusting_for_moved_with_target(messages[0]);
+            }
             // Since this adds messages to the MessageList and renders MessageListView,
             // we don't need to call it if msg_list was not defined by the caller.
             message_util.add_old_messages(messages, opts.msg_list);
@@ -394,6 +397,7 @@ export function load_messages_for_narrow(opts) {
         num_after: consts.narrow_after,
         msg_list: opts.msg_list,
         cont: opts.cont,
+        validate_filter_topic_post_fetch: opts.validate_filter_topic_post_fetch,
     });
 }
 
diff --git a/web/src/message_view.js b/web/src/message_view.js
index c20e27dedb..7a9cd4ead3 100644
--- a/web/src/message_view.js
+++ b/web/src/message_view.js
@@ -87,7 +87,19 @@ export function changehash(newhash, trigger) {
         return;
     }
     message_viewport.stop_auto_scrolling();
-    browser_history.set_hash(newhash);
+
+    if (trigger === "retarget topic location") {
+        // It is important to use `replaceState` rather than `replace`
+        // here for the `back` button to work; we don't want to use
+        // any metadata potentially stored by
+        // update_current_history_state_data associated with an old
+        // URL for the target conversation, and conceptually we want
+        // to replace the inaccurate/old URL for the conversation with
+        // the current/corrected value.
+        window.history.replaceState(null, "", newhash);
+    } else {
+        browser_history.set_hash(newhash);
+    }
 }
 
 export function update_hash_to_match_filter(filter, trigger) {
@@ -132,12 +144,14 @@ function create_and_update_message_list(filter, id_info, opts) {
         });
 
         // Populate the message list if we can apply our filter locally (i.e.
-        // with no backend help) and we have the message we want to select.
+        // with no server help) and we have the message we want to select.
         // Also update id_info accordingly.
-        maybe_add_local_messages({
-            id_info,
-            msg_data,
-        });
+        if (!filter.requires_adjustment_for_moved_with_target) {
+            maybe_add_local_messages({
+                id_info,
+                msg_data,
+            });
+        }
 
         if (!id_info.local_select_id) {
             // If we're not actually ready to select an ID, we need to
@@ -306,6 +320,7 @@ export function show(raw_terms, opts) {
         raw_terms = [{operator: "is", operand: "home"}];
     }
     const filter = new Filter(raw_terms);
+    filter.try_adjusting_for_moved_with_target();
 
     if (try_rendering_locally_for_same_narrow(filter, opts)) {
         return;
@@ -642,7 +657,18 @@ export function show(raw_terms, opts) {
                 }
                 message_fetch.load_messages_for_narrow({
                     anchor,
+                    validate_filter_topic_post_fetch:
+                        filter.requires_adjustment_for_moved_with_target,
                     cont() {
+                        if (
+                            !filter.requires_adjustment_for_moved_with_target &&
+                            filter.has_operator("with")
+                        ) {
+                            // We've already adjusted our filter via
+                            // filter.try_adjusting_for_moved_with_target, and
+                            // should update the URL hash accordingly.
+                            update_hash_to_match_filter(filter, "retarget topic location");
+                        }
                         if (!select_immediately) {
                             render_message_list_with_selected_message({
                                 id_info,
diff --git a/web/src/narrow_state.ts b/web/src/narrow_state.ts
index fec4157a15..e2c9023e21 100644
--- a/web/src/narrow_state.ts
+++ b/web/src/narrow_state.ts
@@ -1,6 +1,9 @@
+import assert from "minimalistic-assert";
+
 import * as blueslip from "./blueslip";
 import {Filter} from "./filter";
 import * as message_lists from "./message_lists";
+import * as message_store from "./message_store";
 import {page_params} from "./page_params";
 import * as people from "./people";
 import type {NarrowTerm} from "./state_data";
@@ -265,6 +268,42 @@ export function _possible_unread_message_ids(
     let topic_name;
     let current_filter_pm_string;
 
+    // For the `with` operator, we can only correctly compute the
+    // correct channel/topic for lookup unreads in if we either
+    // have the message in our local cache, or we know the filter
+    // has already been updated for potentially moved messages.
+    //
+    // The code path that needs this function is never called in
+    // the `with` code path, but for safety, we assert that
+    // assumption is not violated.
+    //
+    // If we need to change that assumption, we can try looking up the
+    // target message in message_store, but would need to return
+    // undefined if the target message is not available.
+    assert(!current_filter.requires_adjustment_for_moved_with_target);
+
+    if (current_filter.can_bucket_by("channel", "topic", "with")) {
+        sub = stream_sub(current_filter);
+        topic_name = topic(current_filter);
+
+        const with_operand = current_filter.operands("with")[0]!;
+        const target_id = Number.parseInt(with_operand, 10);
+        const target_message = message_store.get(target_id)!;
+
+        if (target_message?.type === "private") {
+            // BUG: In theory, the fact that we've asserted
+            // !current_filter.requires_adjustment_for_moved_with_target
+            // should mean this is not possible; but
+            // filter.adjusted_terms_if_moved incorrectly does not
+            // ensure this. Once that bug is fixed, we can delete this case.
+            return [];
+        }
+        if (sub === undefined || topic_name === undefined) {
+            return [];
+        }
+        return unread.get_msg_ids_for_topic(sub.stream_id, topic_name);
+    }
+
     if (current_filter.can_bucket_by("channel", "topic")) {
         sub = stream_sub(current_filter);
         topic_name = topic(current_filter);
diff --git a/web/tests/filter.test.js b/web/tests/filter.test.js
index edfa1e412e..46b828be50 100644
--- a/web/tests/filter.test.js
+++ b/web/tests/filter.test.js
@@ -397,6 +397,25 @@ test("basics", () => {
     assert.ok(filter.is_conversation_view());
     assert.ok(!filter.is_conversation_view_with_near());
 
+    terms = [
+        {operator: "channel", operand: "foo"},
+        {operator: "topic", operand: "bar"},
+        {operator: "with", operand: 17},
+    ];
+    filter = new Filter(terms);
+
+    assert.ok(!filter.is_keyword_search());
+    assert.ok(filter.can_mark_messages_read());
+    assert.ok(filter.supports_collapsing_recipients());
+    assert.ok(!filter.contains_only_private_messages());
+    assert.ok(filter.allow_use_first_unread_when_narrowing());
+    assert.ok(filter.includes_full_stream_history());
+    assert.ok(filter.can_apply_locally());
+    assert.ok(!filter.is_personal_filter());
+    assert.ok(!filter.is_conversation_view());
+    assert.ok(filter.can_bucket_by("channel", "topic", "with"));
+    assert.ok(!filter.is_conversation_view_with_near());
+
     // "stream" was renamed to "channel"
     terms = [
         {operator: "stream", operand: "foo"},
@@ -723,6 +742,13 @@ test("redundancies", () => {
     ];
     filter = new Filter(terms);
     assert.ok(filter.can_bucket_by("is-dm", "not-dm"));
+
+    terms = [
+        {operator: "dm", operand: "joe@example.com,"},
+        {operator: "with", operand: "12"},
+    ];
+    filter = new Filter(terms);
+    assert.ok(filter.can_bucket_by("dm"));
 });
 
 test("canonicalization", () => {
@@ -787,6 +813,23 @@ test("canonicalization", () => {
     assert.equal(term.operand, "reaction");
 });
 
+test("ensure_channel_topic_terms", () => {
+    const channel_term = {operator: "channel", operand: ""};
+    const topic_term = {operator: "topic", operand: ""};
+
+    const term_1 = Filter.ensure_channel_topic_terms([{operator: "with", operand: 12}]);
+    const term_2 = Filter.ensure_channel_topic_terms([topic_term, {operator: "with", operand: 12}]);
+    const term_3 = Filter.ensure_channel_topic_terms([
+        channel_term,
+        {operator: "with", operand: 12},
+    ]);
+    const terms = [term_1, term_2, term_3];
+
+    for (const term of terms) {
+        assert.deepEqual(term, [channel_term, topic_term, {operator: "with", operand: 12}]);
+    }
+});
+
 test("predicate_basics", ({override}) => {
     // Predicates are functions that accept a message object with the message
     // attributes (not content), and return true if the message belongs in a
@@ -1598,6 +1641,8 @@ test("term_type", () => {
         ["dm", "near", "is-unread", "has-link"],
     );
 
+    assert_term_sort(["topic", "channel", "with"], ["channel", "topic", "with"]);
+
     assert_term_sort(["bogus", "channel", "topic"], ["channel", "topic", "bogus"]);
     assert_term_sort(["channel", "topic", "channel"], ["channel", "channel", "topic"]);
 
@@ -1713,6 +1758,69 @@ test("update_email", () => {
     assert.deepEqual(filter.operands("channel"), ["steve@foo.com"]);
 });
 
+test("try_adjusting_for_moved_with_target", ({override}) => {
+    const messages = {
+        12: {type: "stream", display_recipient: "Scotland", topic: "Test 1", id: 12},
+        17: {type: "stream", display_recipient: "Verona", topic: "Test 2", id: 17},
+        2: {type: "direct", id: 2},
+    };
+
+    override(message_store, "get", (msg_id) => messages[msg_id]);
+
+    // When the narrow terms are correct, it returns the same terms
+    let terms = [
+        {operator: "channel", operand: "Scotland", negated: false},
+        {operator: "topic", operand: "Test 1", negated: false},
+        {operator: "with", operand: "12", negated: false},
+    ];
+
+    let filter = new Filter(terms);
+    assert.deepEqual(filter.requires_adjustment_for_moved_with_target, true);
+    filter.try_adjusting_for_moved_with_target();
+    assert.deepEqual(filter.requires_adjustment_for_moved_with_target, false);
+    assert.deepEqual(filter.terms(), terms);
+
+    // When the narrow terms are incorrect, the narrow is corrected
+    // to the narrow of the `with` operand.
+    const incorrect_terms = [
+        {operator: "channel", operand: "Verona", negated: false},
+        {operator: "topic", operand: "Test 2", negated: false},
+        {operator: "with", operand: "12", negated: false},
+    ];
+
+    filter = new Filter(incorrect_terms);
+    assert.deepEqual(filter.requires_adjustment_for_moved_with_target, true);
+    filter.try_adjusting_for_moved_with_target();
+    assert.deepEqual(filter.requires_adjustment_for_moved_with_target, false);
+    assert.deepEqual(filter.terms(), terms);
+
+    // when message specified in `with` operator does not exist in
+    // message_store, we rather go to the server, without any updates.
+    terms = [
+        {operator: "channel", operand: "Scotland", negated: false},
+        {operator: "topic", operand: "Test 1", negated: false},
+        {operator: "with", operand: "11", negated: false},
+    ];
+
+    filter = new Filter(terms);
+    filter.try_adjusting_for_moved_with_target();
+    assert.deepEqual(filter.requires_adjustment_for_moved_with_target, true);
+    assert.deepEqual(filter.terms(), terms);
+
+    // Since only "stream" recipient type can be moved, if the message
+    // is of any type other than "stream", same narrow terms are
+    // returned without the `with` operator.
+    terms = [
+        {operator: "channel", operand: "Scotland", negated: false},
+        {operator: "topic", operand: "Test 1", negated: false},
+        {operator: "with", operand: "2", negated: false},
+    ];
+    filter = new Filter(terms);
+    filter.try_adjusting_for_moved_with_target();
+    assert.deepEqual(filter.requires_adjustment_for_moved_with_target, false);
+    assert.deepEqual(filter.terms(), terms);
+});
+
 function make_private_sub(name, stream_id) {
     const sub = {
         name,
@@ -1732,6 +1840,12 @@ function make_web_public_sub(name, stream_id) {
 }
 
 test("navbar_helpers", () => {
+    const sub = {
+        name: "Foo",
+        stream_id: 12,
+    };
+    stream_data.add_sub(sub);
+
     const stream_id = 43;
     make_sub("Foo", stream_id);
 
@@ -1841,6 +1955,11 @@ test("navbar_helpers", () => {
         {operator: "dm", operand: "joe@example.com"},
         {operator: "near", operand: "12"},
     ];
+    const channel_with = [
+        {operator: "channel", operand: "foo"},
+        {operator: "topic", operand: "bar"},
+        {operator: "with", operand: "12"},
+    ];
 
     const test_cases = [
         {
@@ -2039,6 +2158,13 @@ test("navbar_helpers", () => {
             title: properly_separated_names([joe.full_name]),
             redirect_url_with_search: "#",
         },
+        {
+            terms: channel_with,
+            is_common_narrow: true,
+            zulip_icon: "hashtag",
+            title: "Foo",
+            redirect_url_with_search: "#",
+        },
     ];
 
     realm.realm_enable_guest_user_indicator = true;
diff --git a/web/tests/narrow_activate.test.js b/web/tests/narrow_activate.test.js
index 3c182f8597..73bd4796c6 100644
--- a/web/tests/narrow_activate.test.js
+++ b/web/tests/narrow_activate.test.js
@@ -206,6 +206,7 @@ run_test("basics", ({override}) => {
             cont: opts.cont,
             msg_list: opts.msg_list,
             anchor: 1000,
+            validate_filter_topic_post_fetch: false,
         });
 
         opts.cont();
diff --git a/web/tests/narrow_unread.test.js b/web/tests/narrow_unread.test.js
index 3ba09b2889..8ecf07fb0c 100644
--- a/web/tests/narrow_unread.test.js
+++ b/web/tests/narrow_unread.test.js
@@ -30,6 +30,7 @@ people.add_active_user(alice);
 
 function set_filter(terms) {
     const filter = new Filter(terms);
+    filter.try_adjusting_for_moved_with_target();
     message_lists.set_current({
         data: {
             filter,
@@ -64,6 +65,7 @@ run_test("get_unread_ids", () => {
         id: 101,
         type: "stream",
         stream_id: sub.stream_id,
+        display_recipient: sub.name,
         topic: "my topic",
         unread: true,
         mentioned: true,
@@ -77,8 +79,20 @@ run_test("get_unread_ids", () => {
         display_recipient: [{id: alice.user_id}],
     };
 
+    const other_topic_message = {
+        id: 103,
+        type: "stream",
+        stream_id: sub.stream_id,
+        display_recipient: sub.name,
+        topic: "another topic",
+        unread: true,
+        mentioned: false,
+        mentioned_me_directly: false,
+    };
+
     message_store.update_message_cache(stream_msg);
     message_store.update_message_cache(private_msg);
+    message_store.update_message_cache(other_topic_message);
 
     stream_data.add_sub(sub);
 
@@ -202,6 +216,41 @@ run_test("get_unread_ids", () => {
         flavor: "cannot_compute",
     });
 
+    // For a search using `with` operator, our candidate ids
+    // will be the messages present in the channel/topic
+    // containing the message for which the `with` operand
+    // is id to.
+    //
+    // Here we use an empty topic for the operators, and show that
+    // adding the with operator causes us to see unreads in the
+    // destination topic.
+    unread.process_loaded_messages([other_topic_message]);
+    terms = [
+        {operator: "channel", operand: sub.name},
+        {operator: "topic", operand: "another topic"},
+    ];
+    set_filter(terms);
+    unread_ids = candidate_ids();
+    assert.deepEqual(unread_ids, [other_topic_message.id]);
+
+    terms = [
+        {operator: "channel", operand: sub.name},
+        {operator: "topic", operand: "another topic"},
+        {operator: "with", operand: stream_msg.id},
+    ];
+    set_filter(terms);
+    unread_ids = candidate_ids();
+    assert.deepEqual(unread_ids, [stream_msg.id]);
+
+    terms = [
+        {operator: "channel", operand: sub.name},
+        {operator: "topic", operand: "another topic"},
+        {operator: "with", operand: private_msg.id},
+    ];
+    set_filter(terms);
+    unread_ids = candidate_ids();
+    assert.deepEqual(unread_ids, []);
+
     message_lists.set_current(undefined);
     blueslip.expect("error", "unexpected call to get_first_unread_info");
     assert_unread_info({
diff --git a/zerver/lib/narrow.py b/zerver/lib/narrow.py
index 8ce03ebe0d..24bf9c6b92 100644
--- a/zerver/lib/narrow.py
+++ b/zerver/lib/narrow.py
@@ -16,6 +16,7 @@ from typing import (
 )
 
 from django.conf import settings
+from django.contrib.auth.models import AnonymousUser
 from django.core.exceptions import ValidationError
 from django.db import connection
 from django.utils.translation import gettext as _
@@ -44,8 +45,12 @@ from sqlalchemy.types import ARRAY, Boolean, Integer, Text
 from typing_extensions import TypeAlias, override
 
 from zerver.lib.addressee import get_user_profiles, get_user_profiles_by_ids
-from zerver.lib.exceptions import ErrorCode, JsonableError
-from zerver.lib.message import get_first_visible_message_id
+from zerver.lib.exceptions import ErrorCode, JsonableError, MissingAuthenticationError
+from zerver.lib.message import (
+    access_message,
+    access_web_public_message,
+    get_first_visible_message_id,
+)
 from zerver.lib.narrow_predicate import channel_operators, channels_operators
 from zerver.lib.recipient_users import recipient_for_user_profiles
 from zerver.lib.sqlalchemy_utils import get_sqlalchemy_connection
@@ -81,6 +86,7 @@ from zerver.models import (
     UserMessage,
     UserProfile,
 )
+from zerver.models.recipients import get_direct_message_group_user_ids
 from zerver.models.streams import get_active_streams
 from zerver.models.users import (
     get_user_by_id_in_realm_including_cross_realm,
@@ -124,6 +130,7 @@ class NarrowParameter(BaseModel):
             "sender",
             "group-pm-with",
             "dm-including",
+            "with",
         ]
         operators_supporting_ids = ["pm-with", "dm"]
         operators_non_empty_operand = {"search"}
@@ -161,6 +168,7 @@ def is_spectator_compatible(narrow: Iterable[NarrowParameter]) -> bool:
         "search",
         "near",
         "id",
+        "with",
     ]
     for element in narrow:
         operator = element.operator
@@ -200,6 +208,19 @@ class BadNarrowOperatorError(JsonableError):
         return _("Invalid narrow operator: {desc}")
 
 
+class InvalidOperatorCombinationError(JsonableError):
+    code = ErrorCode.BAD_NARROW
+    data_fields = ["desc"]
+
+    def __init__(self, desc: str) -> None:
+        self.desc: str = desc
+
+    @staticmethod
+    @override
+    def msg_format() -> str:
+        return _("Invalid narrow operator combination: {desc}")
+
+
 ConditionTransform: TypeAlias = Callable[[ClauseElement], ClauseElement]
 
 # These delimiters will not appear in rendered messages or HTML-escaped topics.
@@ -863,6 +884,82 @@ def get_channel_from_narrow_access_unchecked(
     return None
 
 
+# This function implements the core logic of the `with` operator,
+# which is designed to support permanent links to a topic that
+# robustly function if the topic is moved.
+#
+# The with operator accepts a message ID as an operand. If the
+# message ID does not exist or is otherwise not accessible to the
+# current user, then it has no effect.
+#
+# Otherwise, the narrow terms are mutated to remove any
+# channel/topic/dm operators, replacing them with the appropriate
+# operators for the conversation view containing the targeted message.
+def update_narrow_terms_containing_with_operator(
+    realm: Realm,
+    maybe_user_profile: Union[UserProfile, AnonymousUser],
+    narrow: Optional[List[NarrowParameter]],
+) -> Optional[List[NarrowParameter]]:
+    if narrow is None:
+        return narrow
+
+    with_operator_terms = list(filter(lambda term: term.operator == "with", narrow))
+
+    if len(with_operator_terms) > 1:
+        raise InvalidOperatorCombinationError(_("Duplicate 'with' operators."))
+    elif len(with_operator_terms) == 0:
+        return narrow
+
+    with_term = with_operator_terms[0]
+    narrow.remove(with_term)
+    try:
+        message_id = int(with_term.operand)
+    except ValueError:
+        # TODO: This probably should be handled earlier.
+        raise BadNarrowOperatorError(_("Invalid 'with' operator"))
+
+    if maybe_user_profile.is_authenticated:
+        try:
+            message = access_message(maybe_user_profile, message_id)
+        except JsonableError:
+            return narrow
+    else:
+        try:
+            message = access_web_public_message(realm, message_id)
+        except MissingAuthenticationError:
+            return narrow
+
+    # TODO: It would be better if the legacy names here are canonicalized
+    # while building a NarrowParameter.
+    filtered_terms = [
+        term
+        for term in narrow
+        if term.operator not in ["stream", "channel", "topic", "dm", "pm-with"]
+    ]
+
+    if message.recipient.type == Recipient.STREAM:
+        channel_id = message.recipient.type_id
+        topic = message.topic_name()
+        channel_conversation_terms = [
+            NarrowParameter(operator="channel", operand=channel_id),
+            NarrowParameter(operator="topic", operand=topic),
+        ]
+        return channel_conversation_terms + filtered_terms
+
+    elif message.recipient.type == Recipient.PERSONAL:
+        dm_conversation_terms = [
+            NarrowParameter(operator="dm", operand=[message.recipient.type_id])
+        ]
+        return dm_conversation_terms + filtered_terms
+
+    elif message.recipient.type == Recipient.DIRECT_MESSAGE_GROUP:
+        huddle_user_ids = list(get_direct_message_group_user_ids(message.recipient))
+        dm_conversation_terms = [NarrowParameter(operator="dm", operand=huddle_user_ids)]
+        return dm_conversation_terms + filtered_terms
+
+    raise AssertionError("Invalid recipient type")
+
+
 def exclude_muting_conditions(
     user_profile: UserProfile, narrow: Optional[List[NarrowParameter]]
 ) -> List[ClauseElement]:
diff --git a/zerver/tests/test_message_fetch.py b/zerver/tests/test_message_fetch.py
index f2c9967ad3..2010a0d0ed 100644
--- a/zerver/tests/test_message_fetch.py
+++ b/zerver/tests/test_message_fetch.py
@@ -47,7 +47,7 @@ from zerver.lib.sqlalchemy_utils import get_sqlalchemy_connection
 from zerver.lib.streams import StreamDict, create_streams_if_needed, get_public_streams_queryset
 from zerver.lib.test_classes import ZulipTestCase
 from zerver.lib.test_helpers import HostRequestMock, get_user_messages, queries_captured
-from zerver.lib.topic import MATCH_TOPIC, RESOLVED_TOPIC_PREFIX, TOPIC_NAME
+from zerver.lib.topic import MATCH_TOPIC, RESOLVED_TOPIC_PREFIX, TOPIC_NAME, messages_for_topic
 from zerver.lib.types import UserDisplayRecipient
 from zerver.lib.upload import create_attachment
 from zerver.lib.url_encoding import near_message_url
@@ -2675,6 +2675,187 @@ class GetOldMessagesTest(ZulipTestCase):
             """
             )
 
+    def test_get_visible_messages_using_narrow_with(self) -> None:
+        hamlet = self.example_user("hamlet")
+        othello = self.example_user("othello")
+        iago = self.example_user("iago")
+        realm = hamlet.realm
+        self.login("iago")
+
+        self.make_stream("dev team", invite_only=True, history_public_to_subscribers=False)
+        self.subscribe(iago, "dev team")
+
+        # Test `with` operator effective when targeting a topic with
+        # message which can be accessed by the user.
+        msg_id = self.send_stream_message(iago, "dev team", topic_name="test")
+
+        narrow = [
+            dict(operator="channel", operand="dev team"),
+            dict(operator="topic", operand="other_topic"),
+            dict(operator="with", operand=msg_id),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(narrow).decode()))
+        self.assertEqual(results["messages"][0]["id"], msg_id)
+        # Notably we returned the message with its actual topic.
+        self.assertEqual(results["messages"][0]["subject"], "test")
+
+        # Test `with` operator without channel/topic operators.
+        narrow = [
+            dict(operator="with", operand=msg_id),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(narrow).decode()))
+        self.assertEqual(results["messages"][0]["id"], msg_id)
+
+        # Test `with` operator ineffective when targeting a topic with
+        # message that can not be accessed by the user.
+        # Since !history_public_to_subscribers, hamlet cannot view.
+        self.subscribe(hamlet, "dev team")
+        self.login("hamlet")
+
+        narrow = [
+            dict(operator="channel", operand="dev team"),
+            dict(operator="topic", operand="test"),
+            dict(operator="with", operand=msg_id),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(narrow).decode()))
+        self.assert_length(results["messages"], 0)
+
+        # Same result with topic specified incorrectly
+        narrow = [
+            dict(operator="channel", operand="dev team"),
+            dict(operator="topic", operand="wrong_guess"),
+            dict(operator="with", operand=msg_id),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(narrow).decode()))
+        self.assert_length(results["messages"], 0)
+
+        # If just with is specified, we get messages a la combined feed,
+        # but not the target message.
+        narrow = [
+            dict(operator="with", operand=msg_id),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(narrow).decode()))
+        self.assertNotIn(msg_id, [message["id"] for message in results["messages"]])
+
+        # Test `with` operator is effective when targeting personal
+        # messages with message id, and returns messages of that narrow.
+        #
+        # This will be relevant if we allow moving DMs in the future.
+        #
+        # First, attempt to view a message ID we can't access.
+        msg_ids = [self.send_personal_message(iago, othello) for _ in range(2)]
+        with_narrow = [
+            # Important: We pass the wrong conversation.
+            dict(operator="dm", operand=[hamlet.id]),
+            dict(operator="with", operand=msg_ids[0]),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(with_narrow).decode()))
+        self.assertNotIn(msg_id, [message["id"] for message in results["messages"]])
+
+        # Now switch to a user how does have access.
+        self.login("iago")
+        with_narrow = [
+            # Important: We pass the wrong conversation.
+            dict(operator="dm", operand=[hamlet.id]),
+            dict(operator="with", operand=msg_ids[0]),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(with_narrow).decode()))
+        for msg in results["messages"]:
+            self.assertIn(msg["id"], msg_ids)
+
+        # Test `with` operator is effective when targeting direct
+        # messages group with message id.
+        iago = self.example_user("iago")
+        cordelia = self.example_user("cordelia")
+        hamlet = self.example_user("othello")
+
+        msg_ids = [self.send_group_direct_message(iago, [cordelia, hamlet]) for _ in range(2)]
+
+        with_narrow = [
+            # Again, query the wrong conversation.
+            dict(operator="dm", operand=[hamlet.id]),
+            dict(operator="with", operand=msg_ids[0]),
+        ]
+        results = self.get_and_check_messages(dict(narrow=orjson.dumps(with_narrow).decode()))
+
+        for msg in results["messages"]:
+            self.assertIn(msg["id"], msg_ids)
+
+        # Test `with` operator effective with spectator access when
+        # spectator has access to message.
+        self.logout()
+        self.setup_web_public_test(5)
+        channel = get_stream("web-public-channel", realm)
+        assert channel.recipient_id is not None
+        message_ids = messages_for_topic(realm.id, channel.recipient_id, "test").values_list(
+            "id", flat=True
+        )
+
+        web_public_narrow = [
+            dict(operator="channels", operand="web-public", negated=False),
+            dict(operator="channel", operand="web-public-channel"),
+            # Important: Pass a topic that doesn't contain the target message
+            dict(operator="topic", operand="wrong topic"),
+            dict(operator="with", operand=message_ids[0]),
+        ]
+        post_params = {
+            "anchor": 0,
+            "num_before": 0,
+            "num_after": 5,
+            "narrow": orjson.dumps(web_public_narrow).decode(),
+        }
+
+        result = self.client_get("/json/messages", dict(post_params))
+        self.verify_web_public_query_result_success(result, 5)
+
+        # Test `with` operator ineffective when spectator does not have
+        # access to message, by trying to access the same set of messages
+        # but when the spectator access is not allowed.
+        do_set_realm_property(hamlet.realm, "enable_spectator_access", False, acting_user=hamlet)
+
+        result = self.client_get("/json/messages", dict(post_params))
+        self.check_unauthenticated_response(result)
+
+        # Test request with multiple `with` operators raises
+        # InvalidOperatorCombinationError
+        self.login("iago")
+        iago = self.example_user("iago")
+        msg_id_1 = self.send_stream_message(iago, "Verona")
+        msg_id_2 = self.send_stream_message(iago, "Scotland")
+
+        narrow = [
+            dict(operator="channel", operand="Verona"),
+            dict(operator="with", operand=msg_id_1),
+            dict(operator="topic", operand="test"),
+            dict(operator="with", operand=msg_id_2),
+        ]
+        post_params = {
+            "anchor": msg_id_1,
+            "num_before": 0,
+            "num_after": 5,
+            "narrow": orjson.dumps(narrow).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.assert_json_error(
+            result, "Invalid narrow operator combination: Duplicate 'with' operators."
+        )
+
+        # Test request with an invalid message id for `with` operator fails.
+        msg_id = self.send_stream_message(iago, "Verona", topic_name="Invalid id")
+        narrow = [
+            dict(operator="channel", operand="Verona"),
+            dict(operator="topic", operand="Invalid id"),
+            dict(operator="with", operand="3.2"),
+        ]
+        post_params = {
+            "anchor": msg_id,
+            "num_before": 0,
+            "num_after": 5,
+            "narrow": orjson.dumps(narrow).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.assert_json_error(result, "Invalid narrow operator: Invalid 'with' operator")
+
     @override_settings(USING_PGROONGA=False)
     def test_messages_in_narrow(self) -> None:
         user = self.example_user("cordelia")
@@ -3615,7 +3796,7 @@ class GetOldMessagesTest(ZulipTestCase):
             ),
         ]
 
-        for operand in ["id", "sender", "channel", "dm-including", "group-pm-with"]:
+        for operand in ["id", "sender", "channel", "dm-including", "group-pm-with", "with"]:
             self.exercise_bad_narrow_operand_using_dict_api(operand, invalid_operands)
 
         # str or int list is required for "dm" and "pm-with" operator
diff --git a/zerver/views/message_fetch.py b/zerver/views/message_fetch.py
index a1e81ec416..7d85ef0566 100644
--- a/zerver/views/message_fetch.py
+++ b/zerver/views/message_fetch.py
@@ -21,6 +21,7 @@ from zerver.lib.narrow import (
     is_spectator_compatible,
     is_web_public_narrow,
     parse_anchor_value,
+    update_narrow_terms_containing_with_operator,
 )
 from zerver.lib.request import RequestNotes
 from zerver.lib.response import json_success
@@ -114,7 +115,9 @@ def get_messages_backend(
     client_gravatar: Json[bool] = True,
     apply_markdown: Json[bool] = True,
 ) -> HttpResponse:
+    realm = get_valid_realm_from_request(request)
     anchor = parse_anchor_value(anchor_val, use_first_unread_anchor_val)
+    narrow = update_narrow_terms_containing_with_operator(realm, maybe_user_profile, narrow)
     if num_before + num_after > MAX_MESSAGES_PER_FETCH:
         raise JsonableError(
             _("Too many messages requested (maximum {max_messages}).").format(
@@ -124,7 +127,6 @@ def get_messages_backend(
     if num_before > 0 and num_after > 0 and not include_anchor:
         raise JsonableError(_("The anchor can only be excluded at an end of the range"))
 
-    realm = get_valid_realm_from_request(request)
     if not maybe_user_profile.is_authenticated:
         # If user is not authenticated, clients must include
         # `streams:web-public` in their narrow query to indicate this