diff --git a/zephyr/static/js/narrow.js b/zephyr/static/js/narrow.js
index 7ac2f8b65f..c8fcf58e31 100644
--- a/zephyr/static/js/narrow.js
+++ b/zephyr/static/js/narrow.js
@@ -8,7 +8,8 @@ var persistent_message_id = 0;
 // For narrowing based on a particular message
 var target_id = 0;
 
-var filter_function = false;
+var filter_function   = false;
+var current_operators = false;
 
 exports.active = function () {
     // Cast to bool
@@ -23,6 +24,10 @@ exports.predicate = function () {
     }
 };
 
+exports.operators = function () {
+    return current_operators;
+};
+
 var show_floating_recipient;
 var allow_collapse;
 
@@ -176,7 +181,8 @@ exports.activate = function (operators, opts) {
 
     var was_narrowed = exports.active();
 
-    filter_function = build_filter(operators);
+    filter_function   = build_filter(operators);
+    current_operators = operators;
 
     show_floating_recipient = opts.show_floating_recipient;
     allow_collapse          = opts.allow_collapse;
@@ -274,7 +280,8 @@ exports.deactivate = function () {
     if (!filter_function) {
         return;
     }
-    filter_function = false;
+    filter_function   = false;
+    current_operators = false;
 
     $("#zfilt").removeClass('focused_table');
     $("#zhome").addClass('focused_table');
diff --git a/zephyr/static/js/zephyr.js b/zephyr/static/js/zephyr.js
index 15eb26bd8e..7bc6f295c5 100644
--- a/zephyr/static/js/zephyr.js
+++ b/zephyr/static/js/zephyr.js
@@ -624,11 +624,15 @@ function load_old_messages(anchor, num_before, num_after, cont, for_narrow,
         cont_will_add_messages = false;
     }
 
+    var data = {anchor: anchor, num_before: num_before, num_after: num_after};
+
+    if (for_narrow && narrow.active())
+        data.narrow = JSON.stringify(narrow.operators());
+
     $.ajax({
         type:     'POST',
         url:      '/json/get_old_messages',
-        data:     {anchor: anchor, num_before: num_before, num_after: num_after,
-                   narrow: '{}'}, // FIXME
+        data:     data,
         dataType: 'json',
         success: function (data) {
             if (! data) {
diff --git a/zephyr/tests.py b/zephyr/tests.py
index 78d98562f2..6d3fd6542d 100644
--- a/zephyr/tests.py
+++ b/zephyr/tests.py
@@ -536,8 +536,7 @@ class GetOldMessagesTest(AuthedTestCase):
     fixtures = ['messages.json']
 
     def post_with_params(self, modified_params):
-        post_params = {"anchor": 1, "num_before": 1, "num_after": 1,
-                  "narrow": simplejson.dumps({})}
+        post_params = {"anchor": 1, "num_before": 1, "num_after": 1}
         post_params.update(modified_params)
         result = self.client.post("/json/get_old_messages", dict(post_params))
         self.assert_json_success(result)
@@ -560,26 +559,36 @@ class GetOldMessagesTest(AuthedTestCase):
         self.login("hamlet@humbughq.com")
         self.check_well_formed_messages_response(self.post_with_params({}))
 
-    def test_get_old_messages_with_narrow_recipient_id(self):
+    def test_get_old_messages_with_narrow_pm_with(self):
         """
-        A request for old messages with a narrow recipient_id only returns
-        messages for that id.
+        A request for old messages with a narrow by pm-with only returns
+        conversations with that user.
         """
-        self.login("hamlet@humbughq.com")
-        messages = self.message_stream(User.objects.get(email="hamlet@humbughq.com"))
-        recipient_id = messages[0].recipient.id
+        me = 'hamlet@humbughq.com'
+        def dr_emails(dr):
+            return ','.join(sorted(set([r['email'] for r in dr] + [me])))
 
+        personals = [m for m in self.message_stream(User.objects.get(email=me))
+            if m.recipient.type == Recipient.PERSONAL
+            or m.recipient.type == Recipient.HUDDLE]
+        if not personals:
+            # FIXME: This is bad.  We should use test data that is guaranteed
+            # to contain some personals for every user.  See #617.
+            return
+        emails = dr_emails(get_display_recipient(personals[0].recipient))
+
+        self.login(me)
         result = self.post_with_params({"narrow": simplejson.dumps(
-                    {"recipient_id": recipient_id})})
+                    [['pm-with', emails]])})
         self.check_well_formed_messages_response(result)
 
         for message in result["messages"]:
-            self.assertEquals(message["recipient_id"], recipient_id)
+            self.assertEquals(dr_emails(message['display_recipient']), emails)
 
     def test_get_old_messages_with_narrow_stream(self):
         """
-        A request for old messages with a narrow stream only returns messages
-        for that stream.
+        A request for old messages with a narrow by stream only returns
+        messages for that stream.
         """
         self.login("hamlet@humbughq.com")
         messages = self.message_stream(User.objects.get(email="hamlet@humbughq.com"))
@@ -589,7 +598,7 @@ class GetOldMessagesTest(AuthedTestCase):
         stream_id = stream_messages[0].recipient.id
 
         result = self.post_with_params({"narrow": simplejson.dumps(
-                    {"stream": stream_name})})
+                    [['stream', stream_name]])})
         self.check_well_formed_messages_response(result)
 
         for message in result["messages"]:
@@ -598,13 +607,12 @@ class GetOldMessagesTest(AuthedTestCase):
 
     def test_missing_params(self):
         """
-        anchor, num_before, num_after, and narrow are all required
+        anchor, num_before, and num_after are all required
         POST parameters for get_old_messages.
         """
         self.login("hamlet@humbughq.com")
 
-        required_args = (("anchor", 1), ("num_before", 1), ("num_after", 1),
-                         ("narrow", {}))
+        required_args = (("anchor", 1), ("num_before", 1), ("num_after", 1))
 
         for i in range(len(required_args)):
             post_params = dict(required_args[:i] + required_args[i + 1:])
@@ -637,27 +645,39 @@ class GetOldMessagesTest(AuthedTestCase):
 
     def test_bad_narrow_type(self):
         """
-        narrow must be a dictionary.
+        narrow must be a list of string pairs.
         """
         self.login("hamlet@humbughq.com")
 
         other_params = [("anchor", 0), ("num_before", 0), ("num_after", 0)]
 
-        bad_types = (False, 0, "", "{malformed json,")
+        bad_types = (False, 0, '', '{malformed json,',
+            '{}', '{foo: 3}', '[1,2]', '[["x","y","z"]]')
         for type in bad_types:
             post_params = dict(other_params + [("narrow", type)])
             result = self.client.post("/json/get_old_messages", post_params)
             self.assert_json_error(result,
                                    "Bad value for 'narrow': %s" % (type,))
 
-    def exercise_bad_narrow_content(self, narrow_key, bad_content):
+    def test_bad_narrow_operator(self):
+        """
+        Unrecognized narrow operators are rejected.
+        """
+        self.login("hamlet@humbughq.com")
+        for operator in ['', 'foo', 'stream:verona', '__init__']:
+            params = dict(anchor=0, num_before=0, num_after=0,
+                narrow=simplejson.dumps([[operator, '']]))
+            result = self.client.post("/json/get_old_messages", params)
+            self.assert_json_error_contains(result,
+                "Invalid narrow operator: unknown operator")
+
+    def exercise_bad_narrow_operand(self, operator, operands, error_msg):
         other_params = [("anchor", 0), ("num_before", 0), ("num_after", 0)]
-        for content in bad_content:
-            post_params = dict(other_params + [("narrow",
-                                                simplejson.dumps({narrow_key: content}))])
+        for operand in operands:
+            post_params = dict(other_params + [
+                ("narrow", simplejson.dumps([[operator, operand]]))])
             result = self.client.post("/json/get_old_messages", post_params)
-            self.assert_json_error(result,
-                                   "Invalid %s %s" % (narrow_key, content,))
+            self.assert_json_error_contains(result, error_msg)
 
     def test_bad_narrow_stream_content(self):
         """
@@ -665,17 +685,30 @@ class GetOldMessagesTest(AuthedTestCase):
         returned.
         """
         self.login("hamlet@humbughq.com")
-        bad_stream_content = ("non-existent stream", 0, [])
-        self.exercise_bad_narrow_content("stream", bad_stream_content)
+        bad_stream_content = (0, [], ["x", "y"])
+        self.exercise_bad_narrow_operand("stream", bad_stream_content,
+            "Bad value for 'narrow'")
 
     def test_bad_narrow_one_on_one_email_content(self):
         """
-        If an invalid 'emails' is requested in get_old_messages, an
+        If an invalid 'pm-with' is requested in get_old_messages, an
         error is returned.
         """
         self.login("hamlet@humbughq.com")
-        bad_stream_content = (["non-existent email"], "non-existent email", 0, [])
-        self.exercise_bad_narrow_content("emails", bad_stream_content)
+        bad_stream_content = (0, [], ["x","y"])
+        self.exercise_bad_narrow_operand("pm-with", bad_stream_content,
+            "Bad value for 'narrow'")
+
+    def test_bad_narrow_nonexistent_stream(self):
+        self.login("hamlet@humbughq.com")
+        self.exercise_bad_narrow_operand("stream", ['non-existent stream'],
+            "Invalid narrow operator: unknown stream")
+
+    def test_bad_narrow_nonexistent_email(self):
+        self.login("hamlet@humbughq.com")
+        self.exercise_bad_narrow_operand("pm-with", ['non-existent-user@humbughq.com'],
+            "Invalid narrow operator: unknown user")
+
 
 
 class ChangeSettingsTest(AuthedTestCase):
diff --git a/zephyr/views.py b/zephyr/views.py
index 0f2af93b2e..9f89e42764 100644
--- a/zephyr/views.py
+++ b/zephyr/views.py
@@ -63,6 +63,17 @@ def json_to_list(json):
         raise ValueError("argument is not a list")
     return data
 
+def json_to_list_of_string_pairs(json):
+    data = json_to_list(json)
+    for elem in data:
+        if not isinstance(elem, list):
+            raise ValueError("element is not a list")
+        if (len(elem) != 2
+            or any(not isinstance(x, str) and not isinstance(x, unicode)
+                   for x in elem)):
+            raise ValueError("element is not a string pair")
+    return data
+
 def get_stream(stream_name, realm):
     try:
         return Stream.objects.get(name__iexact=stream_name, realm=realm)
@@ -263,55 +274,91 @@ def api_get_old_messages(request, user_profile,
     return get_old_messages_backend(request, user_profile=user_profile,
                                     apply_markdown=apply_markdown)
 
+class BadNarrowOperator(Exception):
+    def __init__(self, desc):
+        self.desc = desc
+
+    def to_json_error_msg(self):
+        return 'Invalid narrow operator: ' + self.desc
+
+class NarrowBuilder(object):
+    def __init__(self, user_profile):
+        self.user_profile = user_profile
+
+    def __call__(self, operator, operand):
+        # We have to be careful here because we're letting users call a method
+        # by name! The prefix 'by_' prevents it from colliding with builtin
+        # Python __magic__ stuff.
+        method_name = 'by_' + operator.replace('-', '_')
+        method = getattr(self, method_name, None)
+        if method is None:
+            raise BadNarrowOperator('unknown operator ' + operator)
+        return method(operand)
+
+    def by_is(self, operand):
+        if operand == 'private-message':
+            return (Q(recipient__type=Recipient.PERSONAL) |
+                    Q(recipient__type=Recipient.HUDDLE))
+        raise BadNarrowOperator("unknown 'is' operand " + operand)
+
+    def by_stream(self, operand):
+        try:
+            stream = Stream.objects.get(realm=self.user_profile.realm,
+                                        name__iexact=operand)
+        except Stream.DoesNotExist:
+            raise BadNarrowOperator('unknown stream ' + operand)
+        recipient = Recipient.objects.get(type=Recipient.STREAM, type_id=stream.id)
+        return Q(recipient=recipient)
+
+    def by_subject(self, operand):
+        return Q(subject=operand)
+
+    def by_pm_with(self, operand):
+        if ',' in operand:
+            # Huddle
+            try:
+                emails = [e.strip() for e in operand.split(',')]
+                recipient = recipient_for_emails(emails, False,
+                    self.user_profile, self.user_profile)
+            except ValidationError:
+                raise BadNarrowOperator('unknown recipient ' + operand)
+            return Q(recipient=recipient)
+        else:
+            # Personal message
+            try:
+                recipient_profile = UserProfile.objects.get(user__email=operand)
+            except UserProfile.DoesNotExist:
+                raise BadNarrowOperator('unknown user ' + operand)
+
+            recipient = Recipient.objects.get(type=Recipient.PERSONAL,
+                type_id=recipient_profile.id)
+
+            if operand == self.user_profile.user.email:
+                # Personals with self
+                return Q(recipient__type=Recipient.PERSONAL,
+                    sender__user__email=operand,
+                    recipient=recipient)
+            else:
+                # Personals with other user; include both directions.
+                return (Q(recipient__type=Recipient.PERSONAL) &
+                    (Q(sender__user__email=operand) | Q(recipient=recipient)))
+
+    def by_search(self, operand):
+        return (Q(content__icontains=operand) |
+                Q(subject__icontains=operand))
+
 @has_request_variables
 def get_old_messages_backend(request, anchor = POST(converter=to_non_negative_int),
                              num_before = POST(converter=to_non_negative_int),
                              num_after = POST(converter=to_non_negative_int),
-                             narrow = POST('narrow', converter=json_to_dict),
+                             narrow = POST('narrow', converter=json_to_list_of_string_pairs, default=None),
                              user_profile=None, apply_markdown=True):
     query = Message.objects.select_related().filter(usermessage__user_profile = user_profile).order_by('id')
 
-    if 'recipient_id' in narrow:
-        query = query.filter(recipient_id = narrow['recipient_id'])
-    if 'stream' in narrow:
-        try:
-            stream = Stream.objects.get(realm=user_profile.realm, name__iexact=narrow['stream'])
-        except Stream.DoesNotExist:
-            return json_error("Invalid stream %s" % (narrow['stream'],))
-        recipient = Recipient.objects.get(type=Recipient.STREAM, type_id=stream.id)
-        query = query.filter(recipient_id = recipient.id)
-
-    if 'emails' in narrow and (type(narrow['emails']) != type([]) or len(narrow['emails']) == 0):
-        return json_error("Invalid emails %s" % (narrow['emails'],))
-    elif 'emails' in narrow and len(narrow['emails']) == 1:
-        query = query.filter(recipient__type=Recipient.PERSONAL)
-        try:
-            recipient_user = UserProfile.objects.get(user__email = narrow['emails'][0])
-        except UserProfile.DoesNotExist:
-            return json_error("Invalid emails ['" + narrow['emails'][0] + "']")
-        recipient = Recipient.objects.get(type=Recipient.PERSONAL, type_id=recipient_user.id)
-        # If we are narrowed to personals with ourself, we want to search for personals where the user
-        # with the desired e-mail address is the sender *and* the recipient, not personals where the
-        # user with the desired e-mail address is the sender *or* the recipient.
-        if narrow['emails'][0] == user_profile.user.email:
-            query = query.filter(Q(sender__user__email=narrow['emails'][0]) & Q(recipient=recipient))
-        else:
-            query = query.filter(Q(sender__user__email=narrow['emails'][0]) | Q(recipient=recipient))
-    elif 'emails' in narrow:
-        try:
-            recipient = recipient_for_emails(narrow['emails'], False, user_profile, user_profile)
-        except ValidationError, e:
-            return json_error(e.messages[0])
-        query = query.filter(recipient=recipient)
-    elif 'type' in narrow and (narrow['type'] == "private" or narrow['type'] == "all_private_messages"):
-        query = query.filter(Q(recipient__type=Recipient.PERSONAL) | Q(recipient__type=Recipient.HUDDLE))
-
-    if 'subject' in narrow:
-        query = query.filter(subject = narrow['subject'])
-
-    if 'searchterm' in narrow:
-        query = query.filter(Q(content__icontains=narrow['searchterm']) |
-                             Q(subject__icontains=narrow['searchterm']))
+    if narrow is not None:
+        build = NarrowBuilder(user_profile)
+        for operator, operand in narrow:
+            query = query.filter(build(operator, operand))
 
     # We add 1 to the number of messages requested to ensure that the
     # resulting list always contains the anchor message