docs: def ReadCompletedRequests(self, session_id, timestamp=None, limit=None):
      """Fetches all the requests with a status message queued for them."""
      subject = session_id.Add("state")
      requests = {}
      status = {}
  
      for predicate, serialized, _ in self.ResolvePrefix(
          subject, [self.FLOW_REQUEST_PREFIX, self.FLOW_STATUS_PREFIX],
          limit=limit,
          timestamp=timestamp):
  
        parts = predicate.split(":", 3)
        request_id = parts[2]
        if parts[1] == "status":
          status[request_id] = serialized
        else:
          requests[request_id] = serialized
  
      for request_id, serialized in sorted(iteritems(requests)):
        if request_id in status:
          yield (rdf_flow_runner.RequestState.FromSerializedString(serialized),
                 rdf_flows.GrrMessage.FromSerializedString(status[request_id]))